admin/OpenJarvis
1
0
Fork 0
mirror of https://github.com/open-jarvis/OpenJarvis.git synced 2026-07-27 21:05:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(memory): native persistent memory service integrated into core (#579)

Browse Source 
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.
This commit is contained in:
Elliot Slusky
2026-06-22 13:59:02 -07:00
committed by GitHub
parent 9b7b3681f6
commit 433d10db5e
16 changed files with 1266 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/openjarvis/cli/chat_cmd.py
+20
View File
@@ -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"]
src/openjarvis/cli/memory_cmd.py
+60
View File
@@ -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",
src/openjarvis/cli/serve.py
+14
View File
@@ -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,
src/openjarvis/core/config.py
+24 -1
View File
@@ -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
src/openjarvis/memory/__init__.py
+28
View File
@@ -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",
]
src/openjarvis/memory/extractor.py
+150
View File
@@ -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"]
src/openjarvis/memory/service.py
+180
View File
@@ -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"]
src/openjarvis/memory/store.py
+179
View File
@@ -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"]
src/openjarvis/server/app.py
+14
View File
@@ -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)
src/openjarvis/server/routes.py
+34 -9
View File
@@ -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(
tests/cli/test_chat_cmd.py
+48
View File
@@ -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"
tests/cli/test_memory_cmd.py
+65
View File
@@ -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
tests/memory/test_fact_extractor.py
+105
View File
@@ -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"]
tests/memory/test_fact_store.py
+112
View File
@@ -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")
tests/memory/test_memory_service.py
+171
View File
@@ -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)
tests/server/test_routes.py
+62
View File
@@ -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(