fix(memory): address service follow-ups (#591)

Closes #582. Route fact-store construction through a new FactStoreRegistry (local backend registered by default); align the default facts path with get_config_dir(); wire completed chat exchanges (streamed and non-streamed) through the EventBus so the memory service captures them consistently; reload the local fact store from disk before operations so external clears don't resurrect stale facts; make the affected config/persona/memory/CLI/route tests hermetic; and refresh uv.lock with the current resolver (locks pytest-xdist + transitive deps, drops py3.14 artifacts since the project constrains Python <3.14).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

src/openjarvis/cli/chat_cmd.py

@@ -11,7 +11,9 @@ from rich.markdown import Markdown
 
 from openjarvis.cli._tool_names import resolve_tool_names
 from openjarvis.core.config import load_config
+from openjarvis.core.events import EventBus
 from openjarvis.core.types import Message, Role
+from openjarvis.memory import publish_completed_exchange
 
 
 def _read_input(prompt: str = "You> ") -> Optional[str]:
@@ -57,6 +59,7 @@ def chat(
     console = Console(stderr=True)
 
     config = load_config()
+    bus = EventBus(record_history=False)
 
     import dataclasses as _dc
 
@@ -97,12 +100,11 @@ def chat(
     if agent_key and agent_key != "none":
         try:
             import openjarvis.agents  # noqa: F401 — trigger registration
-            from openjarvis.core.events import EventBus
             from openjarvis.core.registry import AgentRegistry
 
             if AgentRegistry.contains(agent_key):
                 agent_cls = AgentRegistry.get(agent_key)
-                kwargs: dict = {"bus": EventBus()}
+                kwargs: dict = {"bus": bus}
 
                 if getattr(agent_cls, "accepts_tools", False):
                     tool_names_list = resolve_tool_names(
@@ -184,7 +186,7 @@ def chat(
     try:
         from openjarvis.memory import build_memory_service
 
-        memory_service = build_memory_service(config, engine, model)
+        memory_service = build_memory_service(config, engine, model, event_bus=bus)
         if memory_service is not None:
             memory_service.start()
             console.print("[dim]  Memory: active[/dim]")
@@ -280,9 +282,12 @@ def chat(
             console.print(Markdown(content))
             console.print()
 
-            # Hand the exchange to the memory service (non-blocking).
-            if memory_service is not None:
-                memory_service.submit(user_input, content)
+            publish_completed_exchange(
+                bus,
+                user_input,
+                content,
+                source="cli.chat",
+            )
         except KeyboardInterrupt:
             console.print("\n[dim]Generation interrupted.[/dim]")
         except Exception as exc:

src/openjarvis/cli/serve.py

@@ -498,7 +498,12 @@ def serve(
     try:
         from openjarvis.memory import build_memory_service
 
-        memory_service = build_memory_service(config, engine, model_name)
+        memory_service = build_memory_service(
+            config,
+            engine,
+            model_name,
+            event_bus=bus,
+        )
         if memory_service is not None:
             memory_service.start()
             console.print("  Memory svc: [cyan]active[/cyan]")

src/openjarvis/core/config.py

@@ -932,7 +932,9 @@ class StorageConfig:
     backend: str = "local"  # fact-store backend ("local" = on-disk JSONL)
     extraction_model: str = ""  # model for fact extraction ("" = active model)
     max_facts: int = 1000  # cap on stored facts (oldest evicted past the cap)
-    facts_path: str = str(DEFAULT_CONFIG_DIR / "memory_facts.jsonl")
+    facts_path: str = field(
+        default_factory=lambda: str(get_config_dir() / "memory_facts.jsonl")
+    )
 
 
 # Backward-compatibility alias

src/openjarvis/core/events.py

@@ -27,6 +27,7 @@ class EventType(str, Enum):
     TOOL_CALL_END = "tool_call_end"
     MEMORY_STORE = "memory_store"
     MEMORY_RETRIEVE = "memory_retrieve"
+    CHAT_EXCHANGE_COMPLETED = "chat_exchange_completed"
     AGENT_TURN_START = "agent_turn_start"
     AGENT_TURN_END = "agent_turn_end"
     TELEMETRY_RECORD = "telemetry_record"

src/openjarvis/core/registry.py

@@ -11,6 +11,7 @@ from typing import TYPE_CHECKING, Any, Callable, Dict, Generic, Tuple, Type, Typ
 if TYPE_CHECKING:
     from openjarvis.agents._stubs import BaseAgent
     from openjarvis.engine._stubs import InferenceEngine
+    from openjarvis.memory.store import FactStore
     from openjarvis.tools.storage._stubs import MemoryBackend
 
 T = TypeVar("T")
@@ -109,6 +110,10 @@ class MemoryRegistry(RegistryBase[Type["MemoryBackend"]]):
     """Registry for memory / retrieval backends."""
 
 
+class FactStoreRegistry(RegistryBase[Type["FactStore"]]):
+    """Registry for automatic-memory fact store backends."""
+
+
 class AgentRegistry(RegistryBase[Type["BaseAgent"]]):
     """Registry for agent implementations."""
 
@@ -170,6 +175,7 @@ __all__ = [
     "CompressionRegistry",
     "ConnectorRegistry",
     "EngineRegistry",
+    "FactStoreRegistry",
     "LearningRegistry",
     "MemoryRegistry",
     "MinerRegistry",

src/openjarvis/memory/__init__.py

@@ -9,7 +9,11 @@ and configured via the ``[memory]`` section of ``config.toml``.
 from __future__ import annotations
 
 from openjarvis.memory.extractor import FactExtractor
-from openjarvis.memory.service import MemoryService, build_memory_service
+from openjarvis.memory.service import (
+    MemoryService,
+    build_memory_service,
+    publish_completed_exchange,
+)
 from openjarvis.memory.store import (
     Fact,
     FactStore,
@@ -25,4 +29,5 @@ __all__ = [
     "MemoryService",
     "build_memory_service",
     "create_fact_store",
+    "publish_completed_exchange",
 ]

src/openjarvis/memory/service.py

@@ -20,6 +20,7 @@ import queue
 import threading
 from typing import Any, List, Optional
 
+from openjarvis.core.events import Event, EventBus, EventType
 from openjarvis.memory.extractor import FactExtractor
 from openjarvis.memory.store import Fact, FactStore, create_fact_store
 
@@ -37,10 +38,13 @@ class MemoryService:
         store: FactStore,
         extractor: FactExtractor,
         *,
+        event_bus: EventBus | None = None,
         max_queue: int = 256,
     ) -> None:
         self._store = store
         self._extractor = extractor
+        self._event_bus = event_bus
+        self._subscribed = False
         self._queue: "queue.Queue[Any]" = queue.Queue(maxsize=max(1, max_queue))
         self._thread: Optional[threading.Thread] = None
         self._running = threading.Event()
@@ -52,6 +56,7 @@ class MemoryService:
         if self._running.is_set():
             return
         self._running.set()
+        self._subscribe_events()
         self._thread = threading.Thread(
             target=self._loop,
             name="memory-service",
@@ -73,6 +78,7 @@ class MemoryService:
         if thread is not None:
             thread.join(timeout=timeout)
         self._thread = None
+        self._unsubscribe_events()
         logger.debug("Memory service stopped")
 
     @property
@@ -99,6 +105,34 @@ class MemoryService:
             logger.debug("Memory service queue full; dropping exchange")
             return False
 
+    def _subscribe_events(self) -> None:
+        """Subscribe to lifecycle events that feed automatic memory."""
+        if self._event_bus is None or self._subscribed:
+            return
+        self._event_bus.subscribe(
+            EventType.CHAT_EXCHANGE_COMPLETED,
+            self._on_completed_exchange,
+        )
+        self._subscribed = True
+
+    def _unsubscribe_events(self) -> None:
+        """Unsubscribe from lifecycle events (idempotent)."""
+        if self._event_bus is None or not self._subscribed:
+            return
+        self._event_bus.unsubscribe(
+            EventType.CHAT_EXCHANGE_COMPLETED,
+            self._on_completed_exchange,
+        )
+        self._subscribed = False
+
+    def _on_completed_exchange(self, event: Event) -> None:
+        """Queue a completed chat exchange published on the event bus."""
+        data = event.data or {}
+        self.submit(
+            str(data.get("user_text", "") or ""),
+            str(data.get("assistant_text", "") or ""),
+        )
+
     # -- worker -------------------------------------------------------------
 
     def _loop(self) -> None:
@@ -145,6 +179,8 @@ def build_memory_service(
     config: Any,
     engine: Any,
     default_model: str = "",
+    *,
+    event_bus: EventBus | None = None,
 ) -> Optional[MemoryService]:
     """Build a :class:`MemoryService` from config, or ``None`` if disabled.
 
@@ -170,11 +206,32 @@ def build_memory_service(
 
     store = create_fact_store(
         getattr(mem, "backend", "local"),
-        path=getattr(mem, "facts_path", "~/.openjarvis/memory_facts.jsonl"),
+        path=getattr(mem, "facts_path", None),
         max_facts=getattr(mem, "max_facts", 1000),
     )
     extractor = FactExtractor(engine, model)
-    return MemoryService(store, extractor)
+    return MemoryService(store, extractor, event_bus=event_bus)
 
 
-__all__ = ["MemoryService", "build_memory_service"]
+def publish_completed_exchange(
+    bus: EventBus | None,
+    user_text: str,
+    assistant_text: str = "",
+    *,
+    source: str = "",
+) -> bool:
+    """Publish a completed chat exchange for lifecycle subscribers."""
+    if bus is None or not user_text or not user_text.strip():
+        return False
+    bus.publish(
+        EventType.CHAT_EXCHANGE_COMPLETED,
+        {
+            "user_text": user_text,
+            "assistant_text": assistant_text or "",
+            "source": source,
+        },
+    )
+    return True
+
+
+__all__ = ["MemoryService", "build_memory_service", "publish_completed_exchange"]

src/openjarvis/memory/store.py

@@ -18,6 +18,14 @@ from dataclasses import asdict, dataclass
 from pathlib import Path
 from typing import Iterable, List
 
+from openjarvis.core.paths import get_config_dir
+from openjarvis.core.registry import FactStoreRegistry
+
+
+def _default_fact_path() -> Path:
+    """Return the env-aware default JSONL path for automatic memory facts."""
+    return get_config_dir() / "memory_facts.jsonl"
+
 
 @dataclass(slots=True)
 class Fact:
@@ -56,6 +64,7 @@ class FactStore(ABC):
         """Return the number of stored facts."""
 
 
+@FactStoreRegistry.register("local")
 class LocalFactStore(FactStore):
     """Append-only JSONL fact store on the local filesystem.
 
@@ -67,11 +76,13 @@ class LocalFactStore(FactStore):
 
     def __init__(
         self,
-        path: str | Path = "~/.openjarvis/memory_facts.jsonl",
+        path: str | Path | None = None,
         *,
         max_facts: int = 1000,
     ) -> None:
-        self._path = Path(path).expanduser()
+        self._path = (
+            Path(path).expanduser() if path is not None else _default_fact_path()
+        )
         self._max_facts = max(0, int(max_facts))
         self._lock = threading.Lock()
         self._facts: List[Fact] = self._load()
@@ -116,6 +127,10 @@ class LocalFactStore(FactStore):
         tmp.write_text(payload, encoding="utf-8")
         os.replace(tmp, self._path)
 
+    def _sync_from_disk_locked(self) -> None:
+        """Refresh in-memory facts from disk while holding ``self._lock``."""
+        self._facts = self._load()
+
     # -- FactStore API ------------------------------------------------------
 
     def add(self, text: str, source: str = "") -> bool:
@@ -123,6 +138,7 @@ class LocalFactStore(FactStore):
         if not text:
             return False
         with self._lock:
+            self._sync_from_disk_locked()
             lowered = text.lower()
             if any(f.text.lower() == lowered for f in self._facts):
                 return False  # dedupe
@@ -135,10 +151,12 @@ class LocalFactStore(FactStore):
 
     def list(self) -> List[Fact]:
         with self._lock:
+            self._sync_from_disk_locked()
             return list(self._facts)
 
     def clear(self) -> int:
         with self._lock:
+            self._sync_from_disk_locked()
             removed = len(self._facts)
             self._facts = []
             if self._path.exists():
@@ -150,6 +168,7 @@ class LocalFactStore(FactStore):
 
     def count(self) -> int:
         with self._lock:
+            self._sync_from_disk_locked()
             return len(self._facts)
 
     @property
@@ -158,22 +177,32 @@ class LocalFactStore(FactStore):
         return self._path
 
 
+def _ensure_fact_store_backends_registered() -> None:
+    """Restore built-in fact-store registrations if a test cleared registries."""
+    if not FactStoreRegistry.contains("local"):
+        FactStoreRegistry.register_value("local", LocalFactStore)
+
+
 def create_fact_store(
     backend: str = "local",
     *,
-    path: str | Path = "~/.openjarvis/memory_facts.jsonl",
+    path: str | Path | None = None,
     max_facts: int = 1000,
 ) -> FactStore:
     """Construct a fact store for the configured *backend*.
 
     Only the ``"local"`` (on-disk JSONL) backend is supported today; the
-    factory exists so additional backends can be added without changing the
-    service or CLI wiring.
+    registry-backed constructor exists so additional backends can be added
+    without changing the service or CLI wiring.
     """
+    _ensure_fact_store_backends_registered()
     key = (backend or "local").strip().lower()
-    if key == "local":
-        return LocalFactStore(path, max_facts=max_facts)
-    raise ValueError(f"Unknown memory backend '{backend}'. Supported backends: local")
+    if not FactStoreRegistry.contains(key):
+        supported = ", ".join(FactStoreRegistry.keys())
+        raise ValueError(
+            f"Unknown memory backend '{backend}'. Supported backends: {supported}"
+        )
+    return FactStoreRegistry.create(key, path, max_facts=max_facts)
 
 
 __all__ = ["Fact", "FactStore", "LocalFactStore", "create_fact_store"]

src/openjarvis/server/routes.py

@@ -195,7 +195,13 @@ async def chat_completions(request_body: ChatCompletionRequest, request: Request
         # from the engine for true real-time output.
         if request_body.tools:
             return await _handle_stream_tools(
-                engine, model, request_body, complexity_info, app_config=config
+                engine,
+                model,
+                request_body,
+                complexity_info,
+                app_config=config,
+                bus=getattr(request.app.state, "bus", None),
+                memory_service=getattr(request.app.state, "memory_service", None),
             )
         return await _handle_stream(
             engine,
@@ -204,6 +210,8 @@ async def chat_completions(request_body: ChatCompletionRequest, request: Request
             complexity_info,
             trace_store=getattr(request.app.state, "trace_store", None),
             app_config=config,
+            bus=getattr(request.app.state, "bus", None),
+            memory_service=getattr(request.app.state, "memory_service", None),
         )
 
     # Non-streaming: use agent if available, otherwise direct engine call.
@@ -247,20 +255,44 @@ async def chat_completions(request_body: ChatCompletionRequest, request: Request
         getattr(request.app.state, "memory_service", None),
         query_text_for_complexity,
         response,
+        bus=getattr(request.app.state, "bus", None),
+        source="server.chat",
     )
     return response
 
 
-def _remember_exchange(memory_service, user_text: str, response) -> None:
-    """Submit a completed exchange to the memory service (non-blocking)."""
-    if memory_service is None or not user_text:
+def _response_content(response) -> str:
+    """Extract assistant text from an OpenAI-compatible response object."""
+    content = ""
+    choices = getattr(response, "choices", None)
+    if choices:
+        content = getattr(choices[0].message, "content", "") or ""
+    return content
+
+
+def _record_completed_exchange(
+    memory_service,
+    user_text: str,
+    assistant_text: str,
+    *,
+    bus=None,
+    source: str = "server.chat",
+) -> None:
+    """Publish or submit a completed exchange without blocking a reply."""
+    if not user_text:
         return
     try:
-        content = ""
-        choices = getattr(response, "choices", None)
-        if choices:
-            content = getattr(choices[0].message, "content", "") or ""
-        memory_service.submit(user_text, content)
+        if bus is not None:
+            from openjarvis.memory import publish_completed_exchange
+
+            publish_completed_exchange(
+                bus,
+                user_text,
+                assistant_text,
+                source=source,
+            )
+        elif memory_service is not None:
+            memory_service.submit(user_text, assistant_text)
     except Exception:  # noqa: BLE001 — memory is best-effort, never fail a reply
         logging.getLogger("openjarvis.server").debug(
             "Memory submit failed",
@@ -268,6 +300,24 @@ def _remember_exchange(memory_service, user_text: str, response) -> None:
         )
 
 
+def _remember_exchange(
+    memory_service,
+    user_text: str,
+    response,
+    *,
+    bus=None,
+    source: str = "server.chat",
+) -> None:
+    """Record a completed non-streaming exchange."""
+    _record_completed_exchange(
+        memory_service,
+        user_text,
+        _response_content(response),
+        bus=bus,
+        source=source,
+    )
+
+
 def _handle_direct(
     engine,
     model: str,
@@ -457,6 +507,8 @@ async def _handle_stream_tools(
     complexity_info=None,
     *,
     app_config=None,
+    bus=None,
+    memory_service=None,
 ):
     """Stream a raw OpenAI-compat function-calling response via SSE.
 
@@ -477,8 +529,14 @@ async def _handle_stream_tools(
     messages = _ensure_identity_prompt(messages, app_config)
     chunk_id = f"chatcmpl-{uuid.uuid4().hex[:12]}"
     use_cloud = is_cloud_model(model)
+    query_text = ""
+    for _m in reversed(req.messages):
+        if _m.role == "user" and _m.content:
+            query_text = _m.content
+            break
 
     async def generate():
+        full_content = ""
         # Send the role chunk first (OpenAI convention).
         first_chunk = ChatCompletionChunk(
             id=chunk_id,
@@ -497,6 +555,7 @@ async def _handle_stream_tools(
                 tools=req.tools,
             ):
                 if sc.content:
+                    full_content += sc.content
                     content_chunk = ChatCompletionChunk(
                         id=chunk_id,
                         model=model,
@@ -553,6 +612,14 @@ async def _handle_stream_tools(
         if complexity_info is not None:
             finish_dict["complexity"] = complexity_info.model_dump()
         yield f"data: {_json.dumps(finish_dict)}\n\n"
+        if full_content:
+            _record_completed_exchange(
+                memory_service,
+                query_text,
+                full_content,
+                bus=bus,
+                source="server.chat.stream",
+            )
         yield "data: [DONE]\n\n"
 
     return StreamingResponse(
@@ -570,6 +637,8 @@ async def _handle_stream(
     *,
     trace_store=None,
     app_config=None,
+    bus=None,
+    memory_service=None,
 ):
     """Stream response using SSE format.
 
@@ -710,6 +779,15 @@ async def _handle_stream(
                 ended_at=time.time(),
             )
 
+        if full_content:
+            _record_completed_exchange(
+                memory_service,
+                query_text,
+                full_content,
+                bus=bus,
+                source="server.chat.stream",
+            )
+
         # Send finish chunk with usage data if available
         import json as _json
 

tests/cli/test_chat_cmd.py

@@ -15,6 +15,7 @@ from openjarvis.agents._stubs import (
 )
 from openjarvis.cli.chat_cmd import _read_input, chat
 from openjarvis.core.config import JarvisConfig
+from openjarvis.core.events import Event, EventBus, EventType
 from openjarvis.core.registry import AgentRegistry, ToolRegistry
 from openjarvis.core.types import ToolCall, ToolResult
 from openjarvis.tools._stubs import BaseTool, ToolSpec
@@ -121,25 +122,45 @@ class TestChatAgents:
         assert "failed" not in result.output.lower()
 
     def test_memory_service_started_fed_and_stopped(self) -> None:
-        """The REPL starts the memory service, submits each turn, and stops it."""
+        """The REPL starts memory, publishes each turn, and stops it."""
 
         class _SpyMemoryService:
-            def __init__(self) -> None:
+            def __init__(self, bus: EventBus) -> None:
+                self.bus = bus
                 self.started = False
                 self.stopped = False
                 self.submissions: list[tuple[str, str]] = []
 
             def start(self) -> None:
                 self.started = True
+                self.bus.subscribe(
+                    EventType.CHAT_EXCHANGE_COMPLETED,
+                    self._on_completed_exchange,
+                )
 
-            def submit(self, user_text: str, assistant_text: str = "") -> bool:
-                self.submissions.append((user_text, assistant_text))
-                return True
+            def _on_completed_exchange(self, event: Event) -> None:
+                self.submissions.append(
+                    (
+                        event.data["user_text"],
+                        event.data.get("assistant_text", ""),
+                    )
+                )
 
             def stop(self, timeout: float = 2.0) -> None:
                 self.stopped = True
+                self.bus.unsubscribe(
+                    EventType.CHAT_EXCHANGE_COMPLETED,
+                    self._on_completed_exchange,
+                )
+
+        spy: _SpyMemoryService | None = None
+
+        def _build_memory_service(*args, event_bus: EventBus | None = None, **kwargs):
+            nonlocal spy
+            assert event_bus is not None
+            spy = _SpyMemoryService(event_bus)
+            return spy
 
-        spy = _SpyMemoryService()
         engine = MagicMock()
         engine.engine_id = "mock"
         engine.generate.return_value = {"content": "engine fallback"}
@@ -154,7 +175,7 @@ class TestChatAgents:
             patch("openjarvis.intelligence.register_builtin_models"),
             patch(
                 "openjarvis.memory.build_memory_service",
-                return_value=spy,
+                side_effect=_build_memory_service,
             ),
         ):
             result = CliRunner().invoke(
@@ -164,6 +185,7 @@ class TestChatAgents:
             )
 
         assert result.exit_code == 0
+        assert spy is not None
         assert spy.started is True
         assert spy.stopped is True
         assert spy.submissions == [("hello", "simple ok")]

tests/conftest.py

@@ -18,6 +18,7 @@ from openjarvis.core.registry import (
     CompressionRegistry,
     ConnectorRegistry,
     EngineRegistry,
+    FactStoreRegistry,
     MemoryRegistry,
     MinerRegistry,
     ModelRegistry,
@@ -35,6 +36,7 @@ def _clean_registries() -> None:
     ModelRegistry.clear()
     EngineRegistry.clear()
     MemoryRegistry.clear()
+    FactStoreRegistry.clear()
     MinerRegistry.clear()
     AgentRegistry.clear()
     ToolRegistry.clear()

tests/core/test_config_phase5.py

@@ -48,6 +48,7 @@ class TestConfigPhase5:
         with pytest.raises(KeyError):
             ModelRegistry.get("iso-test")
 
-    def test_load_config_default(self):
-        cfg = load_config()
+    def test_load_config_default(self, tmp_path, monkeypatch):
+        monkeypatch.setenv("OPENJARVIS_HOME", str(tmp_path / "home"))
+        cfg = load_config(tmp_path / "missing-config.toml")
         assert isinstance(cfg, JarvisConfig)

tests/memory/test_fact_store.py

@@ -6,6 +6,7 @@ import json
 
 import pytest
 
+from openjarvis.core.registry import FactStoreRegistry
 from openjarvis.memory.store import LocalFactStore, create_fact_store
 
 
@@ -74,6 +75,20 @@ def test_clear(tmp_path):
     assert LocalFactStore(path).count() == 0
 
 
+def test_external_clear_does_not_resurrect_stale_facts(tmp_path):
+    """A running store instance must not re-flush facts cleared elsewhere."""
+    path = tmp_path / "facts.jsonl"
+    running = LocalFactStore(path)
+    cli = LocalFactStore(path)
+
+    running.add("old fact")
+    assert cli.clear() == 1
+
+    running.add("new fact")
+
+    assert [f.text for f in LocalFactStore(path).list()] == ["new fact"]
+
+
 def test_load_skips_malformed_lines(tmp_path):
     path = tmp_path / "facts.jsonl"
     path.write_text(
@@ -107,6 +122,26 @@ def test_create_fact_store_local(tmp_path):
     assert isinstance(store, LocalFactStore)
 
 
+def test_create_fact_store_uses_fact_store_registry(tmp_path):
+    class CustomFactStore(LocalFactStore):
+        pass
+
+    FactStoreRegistry.register_value("custom", CustomFactStore)
+
+    store = create_fact_store("custom", path=tmp_path / "f.jsonl", max_facts=5)
+
+    assert isinstance(store, CustomFactStore)
+
+
+def test_create_fact_store_default_path_uses_openjarvis_home(tmp_path, monkeypatch):
+    monkeypatch.setenv("OPENJARVIS_HOME", str(tmp_path))
+
+    store = create_fact_store("local")
+
+    assert isinstance(store, LocalFactStore)
+    assert store.path == tmp_path / "memory_facts.jsonl"
+
+
 def test_create_fact_store_unknown_backend(tmp_path):
     with pytest.raises(ValueError):
         create_fact_store("cloud", path=tmp_path / "f.jsonl")

tests/memory/test_memory_service.py

@@ -7,7 +7,12 @@ import time
 from types import SimpleNamespace
 
 from openjarvis.core.config import StorageConfig
-from openjarvis.memory.service import MemoryService, build_memory_service
+from openjarvis.core.events import EventBus
+from openjarvis.memory.service import (
+    MemoryService,
+    build_memory_service,
+    publish_completed_exchange,
+)
 from openjarvis.memory.store import LocalFactStore
 
 
@@ -67,6 +72,38 @@ def test_submit_extracts_and_stores(tmp_path):
         svc.stop()
 
 
+def test_completed_exchange_event_extracts_and_stores(tmp_path):
+    bus = EventBus(record_history=True)
+    extractor = FakeExtractor(["User likes jazz"])
+    store = LocalFactStore(tmp_path / "facts.jsonl")
+    svc = MemoryService(store, extractor, event_bus=bus)
+    svc.start()
+    try:
+        assert publish_completed_exchange(
+            bus,
+            "I like jazz",
+            "Noted.",
+            source="test",
+        )
+        assert _wait_until(lambda: svc.fact_count() == 1)
+        assert extractor.calls == [("I like jazz", "Noted.")]
+    finally:
+        svc.stop()
+
+
+def test_completed_exchange_event_unsubscribes_on_stop(tmp_path):
+    bus = EventBus(record_history=True)
+    extractor = FakeExtractor(["User likes jazz"])
+    store = LocalFactStore(tmp_path / "facts.jsonl")
+    svc = MemoryService(store, extractor, event_bus=bus)
+    svc.start()
+    svc.stop()
+
+    publish_completed_exchange(bus, "I like jazz", "Noted.", source="test")
+
+    assert extractor.calls == []
+
+
 def test_submit_when_not_running_is_dropped(tmp_path):
     extractor = FakeExtractor(["x"])
     svc = _service(tmp_path, extractor)

tests/prompt/test_persona_scope.py

@@ -33,7 +33,7 @@ def test_path_traversal_rejected(bad):
         SystemPromptBuilder._resolve_persona(MemoryFilesConfig(persona_name=bad))
 
 
-def test_none_persona_build_does_not_raise():
+def test_none_persona_build_does_not_raise(tmp_path, monkeypatch):
     """Regression (#497): `--persona none` resolves to empty file paths; building
     the prompt must not raise IsADirectoryError when those empty paths are read
     (Path("") is "." — reading a directory raised before the empty-path guard).
@@ -42,7 +42,8 @@ def test_none_persona_build_does_not_raise():
 
     from openjarvis.core.config import load_config
 
-    cfg = load_config()
+    monkeypatch.setenv("OPENJARVIS_HOME", str(tmp_path / "home"))
+    cfg = load_config(tmp_path / "missing-config.toml")
     mf = dataclasses.replace(cfg.memory_files, persona_name="none")
     builder = SystemPromptBuilder(
         agent_template=cfg.agent.default_system_prompt or "",

tests/server/test_routes.py

@@ -10,6 +10,7 @@ import pytest
 fastapi = pytest.importorskip("fastapi")
 from fastapi.testclient import TestClient  # noqa: E402
 
+from openjarvis.core.events import EventBus, EventType  # noqa: E402
 from openjarvis.server.app import create_app  # noqa: E402
 
 # ---------------------------------------------------------------------------
@@ -54,10 +55,19 @@ def _make_agent(content="Hello from agent"):
     return agent
 
 
+def _test_config():
+    from openjarvis.core.config import JarvisConfig
+
+    cfg = JarvisConfig()
+    cfg.analytics.enabled = False
+    cfg.traces.enabled = False
+    return cfg
+
+
 @pytest.fixture
 def client():
     engine = _make_engine()
-    app = create_app(engine, "test-model")
+    app = create_app(engine, "test-model", config=_test_config())
     return TestClient(app)
 
 
@@ -65,7 +75,7 @@ def client():
 def client_with_agent():
     engine = _make_engine()
     agent = _make_agent()
-    app = create_app(engine, "test-model", agent=agent)
+    app = create_app(engine, "test-model", agent=agent, config=_test_config())
     return TestClient(app)
 
 
@@ -92,7 +102,12 @@ class TestMemoryServiceWiring:
     def test_non_streaming_completion_feeds_memory(self):
         engine = _make_engine(content="remembered reply")
         spy = _SpyMemoryService()
-        app = create_app(engine, "test-model", memory_service=spy)
+        app = create_app(
+            engine,
+            "test-model",
+            memory_service=spy,
+            config=_test_config(),
+        )
         client = TestClient(app)
 
         resp = client.post(
@@ -109,7 +124,13 @@ class TestMemoryServiceWiring:
         engine = _make_engine()
         agent = _make_agent(content="agent reply")
         spy = _SpyMemoryService()
-        app = create_app(engine, "test-model", agent=agent, memory_service=spy)
+        app = create_app(
+            engine,
+            "test-model",
+            agent=agent,
+            memory_service=spy,
+            config=_test_config(),
+        )
         client = TestClient(app)
 
         resp = client.post(
@@ -122,9 +143,83 @@ class TestMemoryServiceWiring:
         assert resp.status_code == 200
         assert spy.submissions == [("remember this", "agent reply")]
 
+    def test_non_streaming_completion_publishes_completed_exchange(self):
+        bus = EventBus(record_history=True)
+        engine = _make_engine(content="event reply")
+        app = create_app(engine, "test-model", bus=bus, config=_test_config())
+        client = TestClient(app)
+
+        resp = client.post(
+            "/v1/chat/completions",
+            json={
+                "model": "test-model",
+                "messages": [{"role": "user", "content": "publish this"}],
+            },
+        )
+
+        assert resp.status_code == 200
+        events = [
+            e for e in bus.history if e.event_type == EventType.CHAT_EXCHANGE_COMPLETED
+        ]
+        assert len(events) == 1
+        assert events[0].data["user_text"] == "publish this"
+        assert events[0].data["assistant_text"] == "event reply"
+
+    def test_streaming_completion_feeds_memory_without_bus(self):
+        engine = _make_engine()
+        spy = _SpyMemoryService()
+        app = create_app(
+            engine,
+            "test-model",
+            memory_service=spy,
+            config=_test_config(),
+        )
+        client = TestClient(app)
+
+        resp = client.post(
+            "/v1/chat/completions",
+            json={
+                "model": "test-model",
+                "messages": [{"role": "user", "content": "stream remember"}],
+                "stream": True,
+            },
+        )
+
+        assert resp.status_code == 200
+        assert "data:" in resp.text
+        assert spy.submissions == [("stream remember", "Hello world")]
+
+    def test_streaming_completion_publishes_completed_exchange(self):
+        bus = EventBus(record_history=True)
+        engine = _make_engine()
+        app = create_app(engine, "test-model", bus=bus, config=_test_config())
+        client = TestClient(app)
+
+        resp = client.post(
+            "/v1/chat/completions",
+            json={
+                "model": "test-model",
+                "messages": [{"role": "user", "content": "stream event"}],
+                "stream": True,
+            },
+        )
+
+        assert resp.status_code == 200
+        assert "data:" in resp.text
+        events = [
+            e for e in bus.history if e.event_type == EventType.CHAT_EXCHANGE_COMPLETED
+        ]
+        assert len(events) == 1
+        assert events[0].data["user_text"] == "stream event"
+        assert events[0].data["assistant_text"] == "Hello world"
+
     def test_no_memory_service_is_noop(self):
         engine = _make_engine()
-        app = create_app(engine, "test-model")  # memory_service defaults to None
+        app = create_app(
+            engine,
+            "test-model",
+            config=_test_config(),
+        )  # memory_service defaults to None
         client = TestClient(app)
         resp = client.post(
             "/v1/chat/completions",
@@ -207,7 +302,7 @@ class TestChatCompletions:
             "model": "test-model",
             "finish_reason": "tool_calls",
         }
-        app = create_app(engine, "test-model")
+        app = create_app(engine, "test-model", config=_test_config())
         client = TestClient(app)
         resp = client.post(
             "/v1/chat/completions",
@@ -255,7 +350,7 @@ class TestChatCompletions:
             "finish_reason": "tool_calls",
         }
         agent = _make_agent(content="GENERIC AGENT FILLER")
-        app = create_app(engine, "test-model", agent=agent)
+        app = create_app(engine, "test-model", agent=agent, config=_test_config())
         client = TestClient(app)
 
         resp = client.post(
@@ -342,7 +437,7 @@ class TestChatCompletions:
             lambda data: received_records.append(data),
         )
 
-        app = create_app(wrapped, "test-model")
+        app = create_app(wrapped, "test-model", config=_test_config())
         app.state.bus = bus
         client = TestClient(app)
 
@@ -483,7 +578,13 @@ class TestChatCompletions:
         # bus present + agent registered == the exact live condition under
         # which the pre-fix code routed to the (broken) agent stream bridge.
         agent = _make_agent(content="GENERIC AGENT FILLER")
-        app = create_app(engine, "test-model", agent=agent, bus=EventBus())
+        app = create_app(
+            engine,
+            "test-model",
+            agent=agent,
+            bus=EventBus(),
+            config=_test_config(),
+        )
         client = TestClient(app)
 
         resp = client.post(
@@ -592,6 +693,15 @@ def _make_capturing_engine(captured: list):
     return engine
 
 
+def _identity_config():
+    from openjarvis.core.config import JarvisConfig
+
+    cfg = JarvisConfig()
+    cfg.agent.default_system_prompt = "You are OpenJarvis."
+    cfg.analytics.enabled = False
+    return cfg
+
+
 class TestIdentityPromptInjection:
     """Regression for #540.
 
@@ -607,7 +717,7 @@ class TestIdentityPromptInjection:
     def test_stream_injects_identity_when_absent(self):
         captured: list = []
         engine = _make_capturing_engine(captured)
-        client = TestClient(create_app(engine, "test-model"))
+        client = TestClient(create_app(engine, "test-model", config=_identity_config()))
 
         resp = client.post(
             "/v1/chat/completions",
@@ -628,7 +738,7 @@ class TestIdentityPromptInjection:
     def test_stream_no_double_injection_when_client_supplies_system(self):
         captured: list = []
         engine = _make_capturing_engine(captured)
-        client = TestClient(create_app(engine, "test-model"))
+        client = TestClient(create_app(engine, "test-model", config=_identity_config()))
 
         resp = client.post(
             "/v1/chat/completions",
@@ -652,7 +762,7 @@ class TestIdentityPromptInjection:
         captured: list = []
         engine = _make_capturing_engine(captured)
         # No agent -> non-stream request goes through _handle_direct.
-        client = TestClient(create_app(engine, "test-model"))
+        client = TestClient(create_app(engine, "test-model", config=_identity_config()))
 
         resp = client.post(
             "/v1/chat/completions",
@@ -670,7 +780,7 @@ class TestIdentityPromptInjection:
     def test_direct_no_double_injection_when_client_supplies_system(self):
         captured: list = []
         engine = _make_capturing_engine(captured)
-        client = TestClient(create_app(engine, "test-model"))
+        client = TestClient(create_app(engine, "test-model", config=_identity_config()))
 
         resp = client.post(
             "/v1/chat/completions",
@@ -691,7 +801,7 @@ class TestIdentityPromptInjection:
     def test_stream_tools_injects_identity_when_absent(self):
         captured: list = []
         engine = _make_capturing_engine(captured)
-        client = TestClient(create_app(engine, "test-model"))
+        client = TestClient(create_app(engine, "test-model", config=_identity_config()))
 
         resp = client.post(
             "/v1/chat/completions",
@@ -733,7 +843,7 @@ class TestModelsEndpoint:
 
     def test_multiple_models(self):
         engine = _make_engine(models=["model-a", "model-b", "model-c"])
-        app = create_app(engine, "model-a")
+        app = create_app(engine, "model-a", config=_test_config())
         client = TestClient(app)
         resp = client.get("/v1/models")
         data = resp.json()
@@ -754,7 +864,7 @@ class TestHealthEndpoint:
     def test_unhealthy(self):
         engine = _make_engine()
         engine.health.return_value = False
-        app = create_app(engine, "test-model")
+        app = create_app(engine, "test-model", config=_test_config())
         client = TestClient(app)
         resp = client.get("/health")
         assert resp.status_code == 503
@@ -768,19 +878,19 @@ class TestHealthEndpoint:
 class TestCreateApp:
     def test_app_state(self):
         engine = _make_engine()
-        app = create_app(engine, "test-model")
+        app = create_app(engine, "test-model", config=_test_config())
         assert app.state.engine is engine
         assert app.state.model == "test-model"
 
     def test_app_with_agent(self):
         engine = _make_engine()
         agent = _make_agent()
-        app = create_app(engine, "test-model", agent=agent)
+        app = create_app(engine, "test-model", agent=agent, config=_test_config())
         assert app.state.agent is agent
 
     def test_app_without_agent(self):
         engine = _make_engine()
-        app = create_app(engine, "test-model")
+        app = create_app(engine, "test-model", config=_test_config())
         assert app.state.agent is None
 
 
@@ -804,6 +914,7 @@ def _traces_enabled_config(tmp_path):
     cfg = JarvisConfig()
     cfg.traces.enabled = True
     cfg.traces.db_path = str(tmp_path / "traces.db")
+    cfg.analytics.enabled = False
     return cfg