Skip to content

Index

memory

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.

Classes

FactExtractor 

FactExtractor(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)

Extract memory-worthy facts from a conversation turn via an engine.

Source code in src/openjarvis/memory/extractor.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
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
Functions
extract 
extract(user_text: str, assistant_text: str = '') -> List[str]

Return durable facts from the exchange. Never raises.

Source code in src/openjarvis/memory/extractor.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
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)

MemoryService 

MemoryService(store: FactStore, extractor: FactExtractor, *, max_queue: int = 256)

Background long-term-memory extraction and persistence service.

Source code in src/openjarvis/memory/service.py 
35
36
37
38
39
40
41
42
43
44
45
46
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()
Functions
start 
start() -> None

Start the background worker thread (idempotent).

Source code in src/openjarvis/memory/service.py 
50
51
52
53
54
55
56
57
58
59
60
61
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")
stop 
stop(timeout: float = 2.0) -> None

Signal the worker to drain and stop, then join it (idempotent).

Source code in src/openjarvis/memory/service.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
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")
submit 
submit(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).

Source code in src/openjarvis/memory/service.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
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

Fact dataclass 

Fact(text: str, source: str = '', created_at: float = 0.0)

A single durable memory entry.

FactStore

Bases: ABC

Abstract persistent store for extracted memory facts.

Functions
add abstractmethod 
add(text: str, source: str = '') -> bool

Store text as a fact. Returns True if a new fact was stored.

Source code in src/openjarvis/memory/store.py 
34
35
36
@abstractmethod
def add(self, text: str, source: str = "") -> bool:
    """Store *text* as a fact. Returns True if a new fact was stored."""
add_many 
add_many(texts: Iterable[str], source: str = '') -> int

Store several facts, returning the count of newly stored ones.

Source code in src/openjarvis/memory/store.py 
38
39
40
41
42
43
44
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
list abstractmethod 
list() -> List[Fact]

Return all stored facts, oldest first.

Source code in src/openjarvis/memory/store.py 
46
47
48
@abstractmethod
def list(self) -> List[Fact]:
    """Return all stored facts, oldest first."""
clear abstractmethod 
clear() -> int

Remove all stored facts, returning the number removed.

Source code in src/openjarvis/memory/store.py 
50
51
52
@abstractmethod
def clear(self) -> int:
    """Remove all stored facts, returning the number removed."""
count abstractmethod 
count() -> int

Return the number of stored facts.

Source code in src/openjarvis/memory/store.py 
54
55
56
@abstractmethod
def count(self) -> int:
    """Return the number of stored facts."""

LocalFactStore 

LocalFactStore(path: str | Path = '~/.openjarvis/memory_facts.jsonl', *, max_facts: int = 1000)

Bases: 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.

Source code in src/openjarvis/memory/store.py 
68
69
70
71
72
73
74
75
76
77
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()
Attributes
path property 
path: Path

Filesystem location of the JSONL store.

Functions

build_memory_service 

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()
Source code in src/openjarvis/memory/service.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
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)

create_fact_store 

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.

Source code in src/openjarvis/memory/store.py 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
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")