From 433d10db5e3307b7d56f1b96eb1d279bf2500092 Mon Sep 17 00:00:00 2001 From: Elliot Slusky <44592435+ElliotSlusky@users.noreply.github.com> Date: Mon, 22 Jun 2026 13:59:02 -0700 Subject: [PATCH] feat(memory): native persistent memory service integrated into core (#579) Adds the openjarvis.memory package (LocalFactStore, FactExtractor, background MemoryService), starts/stops it in the jarvis serve and jarvis chat lifecycle, feeds completed non-streaming exchanges to it, adds [memory] config support, and adds jarvis memory list/clear CLI commands. Extraction runs on a background thread and degrades to a no-op on any failure (BrokenPipe, timeouts, unparseable output) so it can never block a reply or crash the host. Disabled by default. Closes #393, #571, #572, #573. --- src/openjarvis/cli/chat_cmd.py | 20 ++++ src/openjarvis/cli/memory_cmd.py | 60 ++++++++++ src/openjarvis/cli/serve.py | 14 +++ src/openjarvis/core/config.py | 25 +++- src/openjarvis/memory/__init__.py | 28 +++++ src/openjarvis/memory/extractor.py | 150 +++++++++++++++++++++++ src/openjarvis/memory/service.py | 180 ++++++++++++++++++++++++++++ src/openjarvis/memory/store.py | 179 +++++++++++++++++++++++++++ src/openjarvis/server/app.py | 14 +++ src/openjarvis/server/routes.py | 43 +++++-- tests/cli/test_chat_cmd.py | 48 ++++++++ tests/cli/test_memory_cmd.py | 65 ++++++++++ tests/memory/test_fact_extractor.py | 105 ++++++++++++++++ tests/memory/test_fact_store.py | 112 +++++++++++++++++ tests/memory/test_memory_service.py | 171 ++++++++++++++++++++++++++ tests/server/test_routes.py | 62 ++++++++++ 16 files changed, 1266 insertions(+), 10 deletions(-) create mode 100644 src/openjarvis/memory/__init__.py create mode 100644 src/openjarvis/memory/extractor.py create mode 100644 src/openjarvis/memory/service.py create mode 100644 src/openjarvis/memory/store.py create mode 100644 tests/memory/test_fact_extractor.py create mode 100644 tests/memory/test_fact_store.py create mode 100644 tests/memory/test_memory_service.py diff --git a/src/openjarvis/cli/chat_cmd.py b/src/openjarvis/cli/chat_cmd.py index 5610c5bd..8a746f6d 100644 --- a/src/openjarvis/cli/chat_cmd.py +++ b/src/openjarvis/cli/chat_cmd.py @@ -179,6 +179,19 @@ def chat( _notifications = NotificationDispatcher(get_status()) + # Automatic long-term memory — extracts durable facts in the background. + memory_service = None + try: + from openjarvis.memory import build_memory_service + + memory_service = build_memory_service(config, engine, model) + if memory_service is not None: + memory_service.start() + console.print("[dim] Memory: active[/dim]") + except Exception as exc: + console.print(f"[yellow]Memory service unavailable: {exc}[/yellow]") + memory_service = None + # Conversation state if not system_prompt: from openjarvis.prompt.builder import SystemPromptBuilder @@ -266,10 +279,17 @@ def chat( console.print() 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) except KeyboardInterrupt: console.print("\n[dim]Generation interrupted.[/dim]") except Exception as exc: console.print(f"\n[red]Error: {exc}[/red]\n") + if memory_service is not None: + memory_service.stop() + __all__ = ["chat"] diff --git a/src/openjarvis/cli/memory_cmd.py b/src/openjarvis/cli/memory_cmd.py index 8042f982..263b1086 100644 --- a/src/openjarvis/cli/memory_cmd.py +++ b/src/openjarvis/cli/memory_cmd.py @@ -167,6 +167,66 @@ def search( console.print(table) +def _get_fact_store(): + """Instantiate the automatic-memory fact store from config.""" + from openjarvis.memory.store import create_fact_store + + config = load_config() + mem = config.memory + return create_fact_store( + getattr(mem, "backend", "local"), + path=getattr(mem, "facts_path", "~/.openjarvis/memory_facts.jsonl"), + max_facts=getattr(mem, "max_facts", 1000), + ) + + +@memory.command(name="list") +def list_facts() -> None: + """List durable facts captured by the automatic memory service.""" + console = Console() + + store = _get_fact_store() + facts = store.list() + if not facts: + console.print("[yellow]No memory facts stored yet.[/yellow]") + return + + table = Table(title=f"Memory Facts ({len(facts)})") + table.add_column("#", style="dim", width=4) + table.add_column("Fact") + table.add_column("Source", style="cyan") + for i, fact in enumerate(facts, 1): + table.add_row(str(i), fact.text, fact.source or "-") + console.print(table) + + +@memory.command() +@click.option( + "--yes", + "-y", + is_flag=True, + default=False, + help="Skip the confirmation prompt.", +) +def clear(yes: bool) -> None: + """Remove all durable facts captured by the automatic memory service.""" + console = Console() + + store = _get_fact_store() + count = store.count() + if count == 0: + console.print("[yellow]No memory facts to clear.[/yellow]") + return + + if not yes: + if not click.confirm(f"Remove all {count} stored memory fact(s)?"): + console.print("[dim]Aborted.[/dim]") + return + + removed = store.clear() + console.print(f"[green]Cleared {removed} memory fact(s).[/green]") + + @memory.command() @click.option( "--backend", diff --git a/src/openjarvis/cli/serve.py b/src/openjarvis/cli/serve.py index 75b3a943..886fc0ad 100644 --- a/src/openjarvis/cli/serve.py +++ b/src/openjarvis/cli/serve.py @@ -493,6 +493,19 @@ def serve( except Exception as exc: logger.debug("Memory backend init failed: %s", exc) + # Automatic long-term memory service (background fact extraction). + memory_service = None + try: + from openjarvis.memory import build_memory_service + + memory_service = build_memory_service(config, engine, model_name) + if memory_service is not None: + memory_service.start() + console.print(" Memory svc: [cyan]active[/cyan]") + except Exception as exc: + logger.debug("Memory service init failed: %s", exc) + memory_service = None + # Set up agent manager agent_manager = None if config.agent_manager.enabled: @@ -667,6 +680,7 @@ def serve( channel_bridge=channel_bridge, config=config, memory_backend=memory_backend, + memory_service=memory_service, speech_backend=speech_backend, agent_manager=agent_manager, agent_scheduler=agent_scheduler, diff --git a/src/openjarvis/core/config.py b/src/openjarvis/core/config.py index 85eecc3c..5a358083 100644 --- a/src/openjarvis/core/config.py +++ b/src/openjarvis/core/config.py @@ -910,7 +910,13 @@ class LearningConfig: @dataclass(slots=True) class StorageConfig: - """Storage (memory) backend settings.""" + """Storage (memory) backend settings. + + Covers both the retrieval/document store (``default_backend``, ``db_path``, + chunking, context injection) and the automatic long-term memory service + (``enabled``, ``backend``, ``extraction_model``, ``max_facts``, + ``facts_path``) configured under ``[memory]`` in ``config.toml``. + """ default_backend: str = "sqlite" db_path: str = field(default_factory=lambda: str(get_config_dir() / "memory.db")) @@ -920,6 +926,14 @@ class StorageConfig: chunk_size: int = 512 chunk_overlap: int = 64 + # Automatic memory service — extracts durable facts from conversations in + # the background and persists them across sessions (see openjarvis.memory). + enabled: bool = False # start the memory service with serve/chat + 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") + # Backward-compatibility alias MemoryConfig = StorageConfig @@ -2003,6 +2017,15 @@ context_from_memory = true [tools.storage] default_backend = "sqlite" +# Automatic long-term memory: extracts durable facts from conversations in the +# background and persists them. Starts/stops with `jarvis serve` and +# `jarvis chat`; manage stored facts with `jarvis memory list` / `clear`. +[memory] +enabled = false # set true to enable the memory service +backend = "local" # fact-store backend (local = on-disk JSONL) +extraction_model = "" # model for fact extraction ("" = active model) +max_facts = 1000 # cap on stored facts + [tools.mcp] enabled = true diff --git a/src/openjarvis/memory/__init__.py b/src/openjarvis/memory/__init__.py new file mode 100644 index 00000000..d0550516 --- /dev/null +++ b/src/openjarvis/memory/__init__.py @@ -0,0 +1,28 @@ +"""Native persistent long-term memory for OpenJarvis. + +This package provides the automatic memory service that extracts durable facts +from conversations in the background and persists them across sessions. It is +started and stopped as part of the ``jarvis serve`` / ``jarvis chat`` lifecycle +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.store import ( + Fact, + FactStore, + LocalFactStore, + create_fact_store, +) + +__all__ = [ + "Fact", + "FactStore", + "FactExtractor", + "LocalFactStore", + "MemoryService", + "build_memory_service", + "create_fact_store", +] diff --git a/src/openjarvis/memory/extractor.py b/src/openjarvis/memory/extractor.py new file mode 100644 index 00000000..8a08ee6d --- /dev/null +++ b/src/openjarvis/memory/extractor.py @@ -0,0 +1,150 @@ +"""LLM-backed extraction of durable facts from a conversation turn. + +The extractor takes a single (user, assistant) exchange and asks a small +local model to distill any long-term, user-specific facts worth remembering. +It is deliberately defensive: extraction runs on a background thread far from +the request path, so *any* failure — a dropped Ollama connection, a timeout, a +``BrokenPipeError`` when the client went away, or simply unparseable output — +must degrade to "no facts" rather than propagate. +""" + +from __future__ import annotations + +import json +import logging +import re +from typing import Any, List, Optional + +from openjarvis.core.types import Message, Role + +logger = logging.getLogger(__name__) + +_DEFAULT_SYSTEM_PROMPT = ( + "You extract durable, long-term facts about the user from a single " + "conversation exchange. A good fact is stable over time and useful in " + "future conversations: preferences, identity, goals, ongoing projects, " + "constraints, or relationships. Ignore one-off task details, small talk, " + "and anything the assistant said about itself.\n\n" + "Respond with ONLY a JSON array of short fact strings (each under 200 " + "characters). If there is nothing worth remembering, respond with []." +) + + +class FactExtractor: + """Extract memory-worthy facts from a conversation turn via an engine.""" + + def __init__( + self, + engine: Any, + model: str, + *, + temperature: float = 0.0, + max_tokens: int = 512, + max_facts_per_turn: int = 10, + max_fact_chars: int = 200, + system_prompt: Optional[str] = None, + ) -> None: + self._engine = engine + self._model = model + self._temperature = temperature + self._max_tokens = max_tokens + self._max_facts_per_turn = max_facts_per_turn + self._max_fact_chars = max_fact_chars + self._system_prompt = system_prompt or _DEFAULT_SYSTEM_PROMPT + + def extract(self, user_text: str, assistant_text: str = "") -> List[str]: + """Return durable facts from the exchange. Never raises.""" + user_text = (user_text or "").strip() + if not user_text: + return [] + + exchange = f"User: {user_text}" + if assistant_text and assistant_text.strip(): + exchange += f"\nAssistant: {assistant_text.strip()}" + + messages = [ + Message(role=Role.SYSTEM, content=self._system_prompt), + Message(role=Role.USER, content=exchange), + ] + + try: + result = self._engine.generate( + messages, + model=self._model, + temperature=self._temperature, + max_tokens=self._max_tokens, + ) + except BrokenPipeError: + # The classic failure mode: the model call's transport died. + # Extraction is best-effort, so swallow it. + logger.debug("Memory extraction aborted: broken pipe", exc_info=True) + return [] + except Exception: # noqa: BLE001 — extraction must never crash the worker + logger.debug("Memory extraction failed", exc_info=True) + return [] + + if isinstance(result, dict): + content = result.get("content", "") or "" + else: + content = str(result) + + return self._parse(content) + + # -- parsing ------------------------------------------------------------ + + def _parse(self, content: str) -> List[str]: + """Parse model output into a clean, deduped, capped list of facts.""" + if not content or not content.strip(): + return [] + + raw = self._coerce_to_list(content) + + facts: List[str] = [] + seen: set[str] = set() + for item in raw: + fact = self._clean_fact(item) + if not fact: + continue + key = fact.lower() + if key in seen: + continue + seen.add(key) + facts.append(fact) + if len(facts) >= self._max_facts_per_turn: + break + return facts + + def _coerce_to_list(self, content: str) -> List[str]: + """Best-effort conversion of model output to a list of strings.""" + # 1. Try to locate and parse a JSON array anywhere in the output + # (models often wrap it in prose or code fences). + match = re.search(r"\[.*\]", content, re.DOTALL) + if match: + try: + parsed = json.loads(match.group(0)) + if isinstance(parsed, list): + return [str(x) for x in parsed] + except (json.JSONDecodeError, ValueError): + pass + + # 2. Fall back to line-based parsing (markdown bullets / numbered). + items: List[str] = [] + for line in content.splitlines(): + line = line.strip() + if not line: + continue + line = re.sub(r"^\s*(?:[-*•]|\d+[.)])\s*", "", line) + items.append(line) + return items + + def _clean_fact(self, item: str) -> str: + fact = str(item).strip().strip("\"'").strip() + # Drop obvious non-facts the model sometimes emits. + if not fact or fact.lower() in ("[]", "none", "n/a", "null"): + return "" + if len(fact) > self._max_fact_chars: + fact = fact[: self._max_fact_chars].rstrip() + return fact + + +__all__ = ["FactExtractor"] diff --git a/src/openjarvis/memory/service.py b/src/openjarvis/memory/service.py new file mode 100644 index 00000000..b576528c --- /dev/null +++ b/src/openjarvis/memory/service.py @@ -0,0 +1,180 @@ +"""Persistent memory service: async fact extraction integrated into core. + +``MemoryService`` runs fact extraction on a dedicated background thread so it +never blocks ``jarvis serve`` request handling or the ``jarvis chat`` REPL. +Callers hand off an exchange via :meth:`submit`, which enqueues the work and +returns immediately — the slow model call and disk write happen out of band. +The worker swallows every per-job error (including ``BrokenPipeError`` when a +client disconnects mid-extraction), so a flaky extraction model can never take +down the host process. + +The service is started and stopped as part of the OpenJarvis lifecycle (see +``cli/serve.py`` and ``cli/chat_cmd.py``) and is configured through the +``[memory]`` section of ``config.toml``. +""" + +from __future__ import annotations + +import logging +import queue +import threading +from typing import Any, List, Optional + +from openjarvis.memory.extractor import FactExtractor +from openjarvis.memory.store import Fact, FactStore, create_fact_store + +logger = logging.getLogger(__name__) + +# Sentinel pushed onto the queue to wake the worker for shutdown. +_STOP = object() + + +class MemoryService: + """Background long-term-memory extraction and persistence service.""" + + def __init__( + self, + store: FactStore, + extractor: FactExtractor, + *, + max_queue: int = 256, + ) -> None: + self._store = store + self._extractor = extractor + self._queue: "queue.Queue[Any]" = queue.Queue(maxsize=max(1, max_queue)) + self._thread: Optional[threading.Thread] = None + self._running = threading.Event() + + # -- lifecycle ---------------------------------------------------------- + + def start(self) -> None: + """Start the background worker thread (idempotent).""" + if self._running.is_set(): + return + self._running.set() + self._thread = threading.Thread( + target=self._loop, + name="memory-service", + daemon=True, + ) + self._thread.start() + logger.debug("Memory service started") + + def stop(self, timeout: float = 2.0) -> None: + """Signal the worker to drain and stop, then join it (idempotent).""" + if not self._running.is_set(): + return + self._running.clear() + try: + self._queue.put_nowait(_STOP) + except queue.Full: + pass # worker will notice the cleared flag on its next loop + thread = self._thread + if thread is not None: + thread.join(timeout=timeout) + self._thread = None + logger.debug("Memory service stopped") + + @property + def is_running(self) -> bool: + return self._running.is_set() + + # -- submission --------------------------------------------------------- + + def submit(self, user_text: str, assistant_text: str = "") -> bool: + """Queue an exchange for extraction. Non-blocking; never raises. + + Returns True if the job was enqueued, False if the service is not + running or the queue is full (in which case the exchange is dropped + rather than blocking the caller — extraction is best-effort). + """ + if not self._running.is_set(): + return False + if not user_text or not user_text.strip(): + return False + try: + self._queue.put_nowait((user_text, assistant_text)) + return True + except queue.Full: + logger.debug("Memory service queue full; dropping exchange") + return False + + # -- worker ------------------------------------------------------------- + + def _loop(self) -> None: + while True: + try: + job = self._queue.get(timeout=0.5) + except queue.Empty: + if not self._running.is_set(): + break + continue + if job is _STOP: + self._queue.task_done() + break + try: + self._process(job) + except Exception: # noqa: BLE001 — a bad job must not kill the worker + logger.debug("Memory extraction job failed", exc_info=True) + finally: + self._queue.task_done() + if not self._running.is_set() and self._queue.empty(): + break + + def _process(self, job: Any) -> None: + user_text, assistant_text = job + facts = self._extractor.extract(user_text, assistant_text) + if facts: + stored = self._store.add_many(facts, source="auto") + if stored: + logger.debug("Memory service stored %d new fact(s)", stored) + + # -- store passthroughs ------------------------------------------------- + + def list_facts(self) -> List[Fact]: + return self._store.list() + + def clear_facts(self) -> int: + return self._store.clear() + + def fact_count(self) -> int: + return self._store.count() + + +def build_memory_service( + config: Any, + engine: Any, + default_model: str = "", +) -> Optional[MemoryService]: + """Build a :class:`MemoryService` from config, or ``None`` if disabled. + + Reads the ``[memory]`` section (``config.memory`` / ``config.tools.storage``) + for ``enabled``, ``backend``, ``extraction_model``, ``max_facts`` and + ``facts_path``. Returns ``None`` when memory is disabled or no engine / + extraction model is available, so callers can simply do:: + + svc = build_memory_service(config, engine, model) + if svc is not None: + svc.start() + """ + mem = getattr(config, "memory", None) + if mem is None or not getattr(mem, "enabled", False): + return None + if engine is None: + return None + + model = getattr(mem, "extraction_model", "") or default_model + if not model: + logger.debug("Memory service disabled: no extraction model available") + return None + + store = create_fact_store( + getattr(mem, "backend", "local"), + path=getattr(mem, "facts_path", "~/.openjarvis/memory_facts.jsonl"), + max_facts=getattr(mem, "max_facts", 1000), + ) + extractor = FactExtractor(engine, model) + return MemoryService(store, extractor) + + +__all__ = ["MemoryService", "build_memory_service"] diff --git a/src/openjarvis/memory/store.py b/src/openjarvis/memory/store.py new file mode 100644 index 00000000..4498eb39 --- /dev/null +++ b/src/openjarvis/memory/store.py @@ -0,0 +1,179 @@ +"""Persistent stores for automatically extracted long-term memory facts. + +A *fact* is a short, durable statement worth remembering about the user +(e.g. ``"Prefers concise answers"``). Facts are produced by the memory +service's background extractor and persisted here so they survive across +sessions. The store is intentionally small and self-contained: it dedupes, +caps the total number of facts, and is safe to call from multiple threads. +""" + +from __future__ import annotations + +import json +import os +import threading +import time +from abc import ABC, abstractmethod +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Iterable, List + + +@dataclass(slots=True) +class Fact: + """A single durable memory entry.""" + + text: str + source: str = "" + created_at: float = 0.0 + + +class FactStore(ABC): + """Abstract persistent store for extracted memory facts.""" + + @abstractmethod + def add(self, text: str, source: str = "") -> bool: + """Store *text* as a fact. Returns True if a new fact was stored.""" + + def add_many(self, texts: Iterable[str], source: str = "") -> int: + """Store several facts, returning the count of newly stored ones.""" + added = 0 + for text in texts: + if self.add(text, source=source): + added += 1 + return added + + @abstractmethod + def list(self) -> List[Fact]: + """Return all stored facts, oldest first.""" + + @abstractmethod + def clear(self) -> int: + """Remove all stored facts, returning the number removed.""" + + @abstractmethod + def count(self) -> int: + """Return the number of stored facts.""" + + +class LocalFactStore(FactStore): + """Append-only JSONL fact store on the local filesystem. + + Facts are kept human-readable (one JSON object per line) so they can be + inspected or edited by hand. Writes are atomic (temp file + rename) and + guarded by a lock, so concurrent ``add`` calls from the extraction worker + and ``list``/``clear`` from the CLI never corrupt the file. + """ + + def __init__( + self, + path: str | Path = "~/.openjarvis/memory_facts.jsonl", + *, + max_facts: int = 1000, + ) -> None: + self._path = Path(path).expanduser() + self._max_facts = max(0, int(max_facts)) + self._lock = threading.Lock() + self._facts: List[Fact] = self._load() + + # -- persistence -------------------------------------------------------- + + def _load(self) -> List[Fact]: + if not self._path.exists(): + return [] + facts: List[Fact] = [] + try: + text = self._path.read_text(encoding="utf-8") + except OSError: + return [] + for line in text.splitlines(): + line = line.strip() + if not line: + continue + try: + obj = json.loads(line) + except (json.JSONDecodeError, ValueError): + continue # skip malformed lines rather than crashing + fact_text = str(obj.get("text", "")).strip() + if not fact_text: + continue + facts.append( + Fact( + text=fact_text, + source=str(obj.get("source", "")), + created_at=float(obj.get("created_at", 0.0) or 0.0), + ) + ) + return facts + + def _flush(self) -> None: + """Atomically rewrite the JSONL file from the in-memory list.""" + self._path.parent.mkdir(parents=True, exist_ok=True) + tmp = self._path.with_suffix(self._path.suffix + ".tmp") + payload = "".join( + json.dumps(asdict(f), ensure_ascii=False) + "\n" for f in self._facts + ) + tmp.write_text(payload, encoding="utf-8") + os.replace(tmp, self._path) + + # -- FactStore API ------------------------------------------------------ + + def add(self, text: str, source: str = "") -> bool: + text = (text or "").strip() + if not text: + return False + with self._lock: + lowered = text.lower() + if any(f.text.lower() == lowered for f in self._facts): + return False # dedupe + self._facts.append(Fact(text=text, source=source, created_at=time.time())) + # Enforce the cap by evicting the oldest entries. + if self._max_facts and len(self._facts) > self._max_facts: + self._facts = self._facts[-self._max_facts :] + self._flush() + return True + + def list(self) -> List[Fact]: + with self._lock: + return list(self._facts) + + def clear(self) -> int: + with self._lock: + removed = len(self._facts) + self._facts = [] + if self._path.exists(): + try: + self._path.unlink() + except OSError: + self._flush() + return removed + + def count(self) -> int: + with self._lock: + return len(self._facts) + + @property + def path(self) -> Path: + """Filesystem location of the JSONL store.""" + return self._path + + +def create_fact_store( + backend: str = "local", + *, + path: str | Path = "~/.openjarvis/memory_facts.jsonl", + 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. + """ + 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") + + +__all__ = ["Fact", "FactStore", "LocalFactStore", "create_fact_store"] diff --git a/src/openjarvis/server/app.py b/src/openjarvis/server/app.py index 170c59e1..35d42844 100644 --- a/src/openjarvis/server/app.py +++ b/src/openjarvis/server/app.py @@ -151,6 +151,7 @@ def create_app( channel_bridge=None, config=None, memory_backend=None, + memory_service=None, speech_backend=None, agent_manager=None, agent_scheduler=None, @@ -221,6 +222,7 @@ def create_app( app.state.channel_bridge = channel_bridge app.state.config = config app.state.memory_backend = memory_backend + app.state.memory_service = memory_service app.state.speech_backend = speech_backend app.state.agent_manager = agent_manager app.state.agent_scheduler = agent_scheduler @@ -292,6 +294,18 @@ def create_app( except Exception as _exc: logger.debug("Analytics init skipped: %s", _exc) + # Stop the background memory service cleanly when the server shuts down. + if memory_service is not None: + + @app.on_event("shutdown") + async def _shutdown_memory_service() -> None: + svc = getattr(app.state, "memory_service", None) + if svc is not None: + try: + svc.stop() + except Exception: + pass + app.include_router(router) app.include_router(dashboard_router) app.include_router(comparison_router) diff --git a/src/openjarvis/server/routes.py b/src/openjarvis/server/routes.py index 320f2488..a1f38f3a 100644 --- a/src/openjarvis/server/routes.py +++ b/src/openjarvis/server/routes.py @@ -223,7 +223,7 @@ async def chat_completions(request_body: ChatCompletionRequest, request: Request # the agent to execute them), add an explicit opt-in header rather # than removing this guard — silent re-routing is what produced #414. if agent is not None and not request_body.tools: - return _handle_agent( + response = _handle_agent( agent, model, request_body, @@ -231,16 +231,41 @@ async def chat_completions(request_body: ChatCompletionRequest, request: Request trace_store=getattr(request.app.state, "trace_store", None), bus=getattr(request.app.state, "bus", None), ) + else: + bus = getattr(request.app.state, "bus", None) + response = _handle_direct( + engine, + model, + request_body, + bus=bus, + complexity_info=complexity_info, + app_config=config, + ) - bus = getattr(request.app.state, "bus", None) - return _handle_direct( - engine, - model, - request_body, - bus=bus, - complexity_info=complexity_info, - app_config=config, + # Hand the completed exchange to the background memory service. + _remember_exchange( + getattr(request.app.state, "memory_service", None), + query_text_for_complexity, + response, ) + 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: + return + try: + content = "" + choices = getattr(response, "choices", None) + if choices: + content = getattr(choices[0].message, "content", "") or "" + memory_service.submit(user_text, content) + except Exception: # noqa: BLE001 — memory is best-effort, never fail a reply + logging.getLogger("openjarvis.server").debug( + "Memory submit failed", + exc_info=True, + ) def _handle_direct( diff --git a/tests/cli/test_chat_cmd.py b/tests/cli/test_chat_cmd.py index 18adc151..ad5cab47 100644 --- a/tests/cli/test_chat_cmd.py +++ b/tests/cli/test_chat_cmd.py @@ -120,6 +120,54 @@ class TestChatAgents: assert "simple ok" in result.output 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.""" + + class _SpyMemoryService: + def __init__(self) -> None: + self.started = False + self.stopped = False + self.submissions: list[tuple[str, str]] = [] + + def start(self) -> None: + self.started = True + + def submit(self, user_text: str, assistant_text: str = "") -> bool: + self.submissions.append((user_text, assistant_text)) + return True + + def stop(self, timeout: float = 2.0) -> None: + self.stopped = True + + spy = _SpyMemoryService() + engine = MagicMock() + engine.engine_id = "mock" + engine.generate.return_value = {"content": "engine fallback"} + config = JarvisConfig() + config.intelligence.default_model = "test-model" + + AgentRegistry.register_value("simple_chat_agent", _SimpleChatAgent) + + with ( + patch("openjarvis.cli.chat_cmd.load_config", return_value=config), + patch("openjarvis.engine.get_engine", return_value=("mock", engine)), + patch("openjarvis.intelligence.register_builtin_models"), + patch( + "openjarvis.memory.build_memory_service", + return_value=spy, + ), + ): + result = CliRunner().invoke( + chat, + ["--agent", "simple_chat_agent", "--model", "test-model"], + input="hello\n/quit\n", + ) + + assert result.exit_code == 0 + assert spy.started is True + assert spy.stopped is True + assert spy.submissions == [("hello", "simple ok")] + def test_tool_agent_uses_legacy_agent_tools_and_prompts_confirmation(self) -> None: engine = MagicMock() engine.engine_id = "mock" diff --git a/tests/cli/test_memory_cmd.py b/tests/cli/test_memory_cmd.py index dd86adaa..64201c16 100644 --- a/tests/cli/test_memory_cmd.py +++ b/tests/cli/test_memory_cmd.py @@ -9,6 +9,7 @@ from click.testing import CliRunner from openjarvis.cli import cli from openjarvis.core.registry import MemoryRegistry +from openjarvis.memory.store import LocalFactStore from openjarvis.tools.storage.sqlite import SQLiteMemory @@ -108,3 +109,67 @@ def test_memory_stats_shows_count(tmp_path: Path, monkeypatch): assert result.exit_code == 0 assert "2" in result.output backend.close() + + +def _patch_fact_store(monkeypatch, tmp_path: Path) -> LocalFactStore: + """Point ``jarvis memory list/clear`` at a temp fact store.""" + mod = importlib.import_module("openjarvis.cli.memory_cmd") + store = LocalFactStore(tmp_path / "facts.jsonl") + monkeypatch.setattr(mod, "_get_fact_store", lambda: store) + return store + + +def test_memory_list_empty(tmp_path: Path, monkeypatch): + _patch_fact_store(monkeypatch, tmp_path) + result = CliRunner().invoke(cli, ["memory", "list"]) + assert result.exit_code == 0 + assert "No memory facts" in result.output + + +def test_memory_list_shows_facts(tmp_path: Path, monkeypatch): + store = _patch_fact_store(monkeypatch, tmp_path) + store.add("User prefers dark mode") + store.add("User lives in Berlin") + + result = CliRunner().invoke(cli, ["memory", "list"]) + assert result.exit_code == 0 + assert "dark mode" in result.output + assert "Berlin" in result.output + + +def test_memory_clear_with_confirmation(tmp_path: Path, monkeypatch): + store = _patch_fact_store(monkeypatch, tmp_path) + store.add("fact one") + store.add("fact two") + + result = CliRunner().invoke(cli, ["memory", "clear"], input="y\n") + assert result.exit_code == 0 + assert "Cleared 2" in result.output + assert store.count() == 0 + + +def test_memory_clear_aborted(tmp_path: Path, monkeypatch): + store = _patch_fact_store(monkeypatch, tmp_path) + store.add("keep me") + + result = CliRunner().invoke(cli, ["memory", "clear"], input="n\n") + assert result.exit_code == 0 + assert "Aborted" in result.output + assert store.count() == 1 + + +def test_memory_clear_yes_flag(tmp_path: Path, monkeypatch): + store = _patch_fact_store(monkeypatch, tmp_path) + store.add("fact") + + result = CliRunner().invoke(cli, ["memory", "clear", "--yes"]) + assert result.exit_code == 0 + assert "Cleared 1" in result.output + assert store.count() == 0 + + +def test_memory_clear_empty(tmp_path: Path, monkeypatch): + _patch_fact_store(monkeypatch, tmp_path) + result = CliRunner().invoke(cli, ["memory", "clear"]) + assert result.exit_code == 0 + assert "No memory facts to clear" in result.output diff --git a/tests/memory/test_fact_extractor.py b/tests/memory/test_fact_extractor.py new file mode 100644 index 00000000..3e98c333 --- /dev/null +++ b/tests/memory/test_fact_extractor.py @@ -0,0 +1,105 @@ +"""Tests for the LLM-backed fact extractor (openjarvis.memory.extractor).""" + +from __future__ import annotations + +from openjarvis.memory.extractor import FactExtractor + + +class FakeEngine: + """Engine stub returning a canned completion (or raising).""" + + def __init__(self, content="", *, raises=None): + self._content = content + self._raises = raises + self.calls = [] + + def generate(self, messages, *, model, temperature=0.7, max_tokens=1024, **kwargs): + self.calls.append((messages, model, temperature, max_tokens)) + if self._raises is not None: + raise self._raises + return {"content": self._content} + + +def test_parses_json_array(): + engine = FakeEngine('["User likes coffee", "User lives in Berlin"]') + extractor = FactExtractor(engine, "qwen3:14b") + facts = extractor.extract("I like coffee and live in Berlin", "Noted.") + assert facts == ["User likes coffee", "User lives in Berlin"] + + +def test_parses_json_array_wrapped_in_prose(): + engine = FakeEngine('Sure! Here are the facts:\n["Fact A", "Fact B"]\nDone.') + extractor = FactExtractor(engine, "m") + assert extractor.extract("hi", "hello") == ["Fact A", "Fact B"] + + +def test_empty_array_returns_no_facts(): + engine = FakeEngine("[]") + extractor = FactExtractor(engine, "m") + assert extractor.extract("just chatting", "ok") == [] + + +def test_line_fallback_for_bullets(): + engine = FakeEngine("- User is a teacher\n- User has two kids\n") + extractor = FactExtractor(engine, "m") + assert extractor.extract("about me", "noted") == [ + "User is a teacher", + "User has two kids", + ] + + +def test_dedupe_within_turn(): + engine = FakeEngine('["likes tea", "Likes Tea", "likes tea"]') + extractor = FactExtractor(engine, "m") + assert extractor.extract("x", "y") == ["likes tea"] + + +def test_cap_facts_per_turn(): + items = [f'"fact {i}"' for i in range(20)] + engine = FakeEngine("[" + ", ".join(items) + "]") + extractor = FactExtractor(engine, "m", max_facts_per_turn=3) + assert len(extractor.extract("x", "y")) == 3 + + +def test_truncates_long_facts(): + long_fact = "z" * 500 + engine = FakeEngine(f'["{long_fact}"]') + extractor = FactExtractor(engine, "m", max_fact_chars=50) + facts = extractor.extract("x", "y") + assert len(facts) == 1 + assert len(facts[0]) == 50 + + +def test_empty_user_text_skips_engine(): + engine = FakeEngine('["should not be called"]') + extractor = FactExtractor(engine, "m") + assert extractor.extract(" ", "y") == [] + assert engine.calls == [] + + +def test_broken_pipe_returns_empty(): + engine = FakeEngine(raises=BrokenPipeError("client gone")) + extractor = FactExtractor(engine, "m") + # Must not raise — extraction is best-effort. + assert extractor.extract("hi", "hello") == [] + + +def test_generic_exception_returns_empty(): + engine = FakeEngine(raises=RuntimeError("ollama exploded")) + extractor = FactExtractor(engine, "m") + assert extractor.extract("hi", "hello") == [] + + +def test_handles_non_dict_result(): + class StrEngine: + def generate(self, *a, **k): + return '["plain string result"]' + + extractor = FactExtractor(StrEngine(), "m") + assert extractor.extract("x", "y") == ["plain string result"] + + +def test_filters_non_fact_tokens(): + engine = FakeEngine('["none", "N/A", "Real fact"]') + extractor = FactExtractor(engine, "m") + assert extractor.extract("x", "y") == ["Real fact"] diff --git a/tests/memory/test_fact_store.py b/tests/memory/test_fact_store.py new file mode 100644 index 00000000..ac269769 --- /dev/null +++ b/tests/memory/test_fact_store.py @@ -0,0 +1,112 @@ +"""Tests for the persistent fact store (openjarvis.memory.store).""" + +from __future__ import annotations + +import json + +import pytest + +from openjarvis.memory.store import LocalFactStore, create_fact_store + + +def test_add_and_list(tmp_path): + store = LocalFactStore(tmp_path / "facts.jsonl") + assert store.add("User prefers concise answers") is True + assert store.add("User lives in Berlin") is True + + facts = store.list() + assert [f.text for f in facts] == [ + "User prefers concise answers", + "User lives in Berlin", + ] + assert store.count() == 2 + + +def test_add_dedupes_case_insensitive(tmp_path): + store = LocalFactStore(tmp_path / "facts.jsonl") + assert store.add("Likes coffee") is True + assert store.add("likes coffee") is False # duplicate + assert store.count() == 1 + + +def test_add_skips_empty(tmp_path): + store = LocalFactStore(tmp_path / "facts.jsonl") + assert store.add("") is False + assert store.add(" ") is False + assert store.count() == 0 + + +def test_add_many(tmp_path): + store = LocalFactStore(tmp_path / "facts.jsonl") + added = store.add_many(["a", "b", "a", "c"]) # one dupe + assert added == 3 + assert store.count() == 3 + + +def test_max_facts_evicts_oldest(tmp_path): + store = LocalFactStore(tmp_path / "facts.jsonl", max_facts=2) + store.add("first") + store.add("second") + store.add("third") + facts = [f.text for f in store.list()] + assert facts == ["second", "third"] # oldest dropped + + +def test_persistence_across_instances(tmp_path): + path = tmp_path / "facts.jsonl" + store = LocalFactStore(path) + store.add("durable fact") + + reloaded = LocalFactStore(path) + assert [f.text for f in reloaded.list()] == ["durable fact"] + + +def test_clear(tmp_path): + path = tmp_path / "facts.jsonl" + store = LocalFactStore(path) + store.add("one") + store.add("two") + + removed = store.clear() + assert removed == 2 + assert store.count() == 0 + # A fresh instance also sees an empty store. + assert LocalFactStore(path).count() == 0 + + +def test_load_skips_malformed_lines(tmp_path): + path = tmp_path / "facts.jsonl" + path.write_text( + '{"text": "good fact"}\n' + "this is not json\n" + '{"text": ""}\n' # empty text ignored + '{"text": "another good"}\n', + encoding="utf-8", + ) + store = LocalFactStore(path) + assert [f.text for f in store.list()] == ["good fact", "another good"] + + +def test_jsonl_round_trip_is_valid_json(tmp_path): + path = tmp_path / "facts.jsonl" + store = LocalFactStore(path) + store.add("fact one", source="auto") + + lines = [ + line for line in path.read_text(encoding="utf-8").splitlines() if line.strip() + ] + assert len(lines) == 1 + obj = json.loads(lines[0]) + assert obj["text"] == "fact one" + assert obj["source"] == "auto" + assert "created_at" in obj + + +def test_create_fact_store_local(tmp_path): + store = create_fact_store("local", path=tmp_path / "f.jsonl", max_facts=5) + assert isinstance(store, LocalFactStore) + + +def test_create_fact_store_unknown_backend(tmp_path): + with pytest.raises(ValueError): + create_fact_store("cloud", path=tmp_path / "f.jsonl") diff --git a/tests/memory/test_memory_service.py b/tests/memory/test_memory_service.py new file mode 100644 index 00000000..117c9394 --- /dev/null +++ b/tests/memory/test_memory_service.py @@ -0,0 +1,171 @@ +"""Tests for the background memory service (openjarvis.memory.service).""" + +from __future__ import annotations + +import threading +import time +from types import SimpleNamespace + +from openjarvis.core.config import StorageConfig +from openjarvis.memory.service import MemoryService, build_memory_service +from openjarvis.memory.store import LocalFactStore + + +def _wait_until(predicate, timeout=2.0, interval=0.01): + deadline = time.time() + timeout + while time.time() < deadline: + if predicate(): + return True + time.sleep(interval) + return predicate() + + +class FakeExtractor: + """Extractor stub with controllable output, blocking and failures.""" + + def __init__(self, facts=None, *, raises=None, gate=None): + self._facts = facts or [] + self._raises = raises + self._gate = gate # optional threading.Event to block on + self.calls = [] + + def extract(self, user_text, assistant_text=""): + self.calls.append((user_text, assistant_text)) + if self._gate is not None: + self._gate.wait(timeout=2.0) + if self._raises is not None: + raise self._raises + return list(self._facts) + + +def _service(tmp_path, extractor, **kwargs): + store = LocalFactStore(tmp_path / "facts.jsonl") + return MemoryService(store, extractor, **kwargs) + + +def test_start_stop_lifecycle(tmp_path): + svc = _service(tmp_path, FakeExtractor()) + assert svc.is_running is False + svc.start() + assert svc.is_running is True + svc.start() # idempotent + assert svc.is_running is True + svc.stop() + assert svc.is_running is False + svc.stop() # idempotent + + +def test_submit_extracts_and_stores(tmp_path): + extractor = FakeExtractor(["User likes hiking"]) + svc = _service(tmp_path, extractor) + svc.start() + try: + assert svc.submit("I love hiking", "Nice!") is True + assert _wait_until(lambda: svc.fact_count() == 1) + assert [f.text for f in svc.list_facts()] == ["User likes hiking"] + finally: + svc.stop() + + +def test_submit_when_not_running_is_dropped(tmp_path): + extractor = FakeExtractor(["x"]) + svc = _service(tmp_path, extractor) + assert svc.submit("hi", "there") is False + assert extractor.calls == [] + + +def test_submit_empty_user_text_dropped(tmp_path): + extractor = FakeExtractor(["x"]) + svc = _service(tmp_path, extractor) + svc.start() + try: + assert svc.submit(" ", "y") is False + finally: + svc.stop() + + +def test_worker_survives_extractor_broken_pipe(tmp_path): + """A BrokenPipeError in one job must not kill the worker.""" + extractor = FakeExtractor(raises=BrokenPipeError("client gone")) + svc = _service(tmp_path, extractor) + svc.start() + try: + svc.submit("first", "a") + assert _wait_until(lambda: len(extractor.calls) == 1) + # Service is still alive and accepting work. + assert svc.is_running is True + assert svc.submit("second", "b") is True + assert _wait_until(lambda: len(extractor.calls) == 2) + finally: + svc.stop() + + +def test_worker_survives_generic_exception(tmp_path): + extractor = FakeExtractor(raises=RuntimeError("boom")) + svc = _service(tmp_path, extractor) + svc.start() + try: + svc.submit("x", "y") + assert _wait_until(lambda: len(extractor.calls) == 1) + assert svc.is_running is True + finally: + svc.stop() + + +def test_submit_returns_false_when_queue_full(tmp_path): + """Backpressure: a full queue drops work instead of blocking the caller.""" + gate = threading.Event() + extractor = FakeExtractor(["fact"], gate=gate) + svc = _service(tmp_path, extractor, max_queue=1) + svc.start() + try: + # First submit is pulled by the worker and blocks on the gate. + assert svc.submit("job1", "a") is True + assert _wait_until(lambda: len(extractor.calls) == 1) + # Fill the (size-1) queue, then the next submit must be dropped. + assert svc.submit("job2", "b") is True + dropped = svc.submit("job3", "c") + assert dropped is False + finally: + gate.set() + svc.stop() + + +def test_build_memory_service_disabled_returns_none(tmp_path): + cfg = SimpleNamespace(memory=StorageConfig(enabled=False)) + assert build_memory_service(cfg, object(), "model") is None + + +def test_build_memory_service_no_engine_returns_none(tmp_path): + cfg = SimpleNamespace(memory=StorageConfig(enabled=True)) + assert build_memory_service(cfg, None, "model") is None + + +def test_build_memory_service_no_model_returns_none(tmp_path): + cfg = SimpleNamespace(memory=StorageConfig(enabled=True, extraction_model="")) + assert build_memory_service(cfg, object(), "") is None + + +def test_build_memory_service_enabled(tmp_path): + cfg = SimpleNamespace( + memory=StorageConfig( + enabled=True, + extraction_model="qwen3:14b", + facts_path=str(tmp_path / "facts.jsonl"), + max_facts=10, + ) + ) + svc = build_memory_service(cfg, object(), "fallback-model") + assert isinstance(svc, MemoryService) + + +def test_build_memory_service_falls_back_to_default_model(tmp_path): + cfg = SimpleNamespace( + memory=StorageConfig( + enabled=True, + extraction_model="", + facts_path=str(tmp_path / "facts.jsonl"), + ) + ) + svc = build_memory_service(cfg, object(), "active-model") + assert isinstance(svc, MemoryService) diff --git a/tests/server/test_routes.py b/tests/server/test_routes.py index 7d70da92..aa6c4812 100644 --- a/tests/server/test_routes.py +++ b/tests/server/test_routes.py @@ -74,6 +74,68 @@ def client_with_agent(): # --------------------------------------------------------------------------- +class _SpyMemoryService: + """Minimal stand-in capturing memory submissions.""" + + def __init__(self) -> None: + self.submissions: list[tuple[str, str]] = [] + + def submit(self, user_text: str, assistant_text: str = "") -> bool: + self.submissions.append((user_text, assistant_text)) + return True + + def stop(self, timeout: float = 2.0) -> None: + pass + + +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) + client = TestClient(app) + + resp = client.post( + "/v1/chat/completions", + json={ + "model": "test-model", + "messages": [{"role": "user", "content": "I like jazz"}], + }, + ) + assert resp.status_code == 200 + assert spy.submissions == [("I like jazz", "remembered reply")] + + def test_agent_completion_feeds_memory(self): + engine = _make_engine() + agent = _make_agent(content="agent reply") + spy = _SpyMemoryService() + app = create_app(engine, "test-model", agent=agent, memory_service=spy) + client = TestClient(app) + + resp = client.post( + "/v1/chat/completions", + json={ + "model": "test-model", + "messages": [{"role": "user", "content": "remember this"}], + }, + ) + assert resp.status_code == 200 + assert spy.submissions == [("remember this", "agent reply")] + + def test_no_memory_service_is_noop(self): + engine = _make_engine() + app = create_app(engine, "test-model") # memory_service defaults to None + client = TestClient(app) + resp = client.post( + "/v1/chat/completions", + json={ + "model": "test-model", + "messages": [{"role": "user", "content": "Hello"}], + }, + ) + assert resp.status_code == 200 + + class TestChatCompletions: def test_basic_completion(self, client): resp = client.post(