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

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

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, *, event_bus: EventBus | None = None, scanner: Any = None, max_queue: int = 256)

Background long-term-memory extraction and persistence service.

Source code in src/openjarvis/memory/service.py

def __init__(
    self,
    store: FactStore,
    extractor: FactExtractor,
    *,
    event_bus: EventBus | None = None,
    scanner: Any = None,
    max_queue: int = 256,
) -> None:
    self._store = store
    self._extractor = extractor
    self._event_bus = event_bus
    self._scanner = scanner
    self._subscribed = False
    self._queue: "queue.Queue[Any]" = queue.Queue(maxsize=max(1, max_queue))
    self._thread: Optional[threading.Thread] = None
    self._running = threading.Event()

Functions

start

start() -> None

Start the background worker thread (idempotent).

Source code in src/openjarvis/memory/service.py

def start(self) -> None:
    """Start the background worker thread (idempotent)."""
    if self._running.is_set():
        return
    self._running.set()
    self._subscribe_events()
    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

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
    self._unsubscribe_events()
    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

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, trust: str = '')

A single durable memory entry.

Attributes

trusted_for_recall property

trusted_for_recall: bool

Whether this fact may be placed in model-facing context.

Only untrusted — the tier the memory service assigns when the injection scanner flags a fact's own text — is withheld. Unknown future tiers fail closed rather than silently becoming prompt input.

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

@abstractmethod
def add(self, text: str, source: str = "") -> bool:
    """Store *text* as a fact. Returns True if a new fact was stored."""

set_trust

set_trust(index: int, trust: str) -> bool

Set the provenance tier of the index-th fact (0-based) as returned by :meth:list. Returns True if a fact was updated.

Source code in src/openjarvis/memory/store.py

def set_trust(self, index: int, trust: str) -> bool:
    """Set the provenance tier of the *index*-th fact (0-based) as returned
    by :meth:`list`. Returns True if a fact was updated."""
    raise NotImplementedError

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

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

add_with_trust

add_with_trust(text: str, source: str = '', trust: str = '') -> bool

Store a provenance-aware fact without breaking legacy backends.

Third-party stores implementing the original add(text, source) contract inherit this adapter. Recallable facts are stored normally; quarantined or unknown tiers are dropped because a backend that cannot persist provenance cannot safely retain them for model-facing recall. Provenance-aware stores should override this method.

Source code in src/openjarvis/memory/store.py

def add_with_trust(self, text: str, source: str = "", trust: str = "") -> bool:
    """Store a provenance-aware fact without breaking legacy backends.

    Third-party stores implementing the original ``add(text, source)``
    contract inherit this adapter. Recallable facts are stored normally;
    quarantined or unknown tiers are dropped because a backend that cannot
    persist provenance cannot safely retain them for model-facing recall.
    Provenance-aware stores should override this method.
    """
    if (trust or "").strip().lower() not in _RECALLABLE_TIERS:
        return False
    return self.add(text, source=source)

add_many_with_trust

add_many_with_trust(texts: Iterable[str], source: str = '', trust: str = '') -> int

Store several provenance-aware facts.

Source code in src/openjarvis/memory/store.py

def add_many_with_trust(
    self,
    texts: Iterable[str],
    source: str = "",
    trust: str = "",
) -> int:
    """Store several provenance-aware facts."""
    return sum(
        bool(self.add_with_trust(text, source=source, trust=trust))
        for text in texts
    )

promote_reviewed

promote_reviewed(index: int, expected_text: str) -> bool

Promote the reviewed fact at index when it still matches.

The default preserves compatibility with third-party implementations; stores with concurrent writers should override this with an atomic identity check.

Source code in src/openjarvis/memory/store.py

def promote_reviewed(self, index: int, expected_text: str) -> bool:
    """Promote the reviewed fact at *index* when it still matches.

    The default preserves compatibility with third-party implementations;
    stores with concurrent writers should override this with an atomic
    identity check.
    """
    del expected_text
    return self.set_trust(index, TRUST_TRUSTED)

list abstractmethod

list() -> List[Fact]

Return all stored facts, oldest first.

Source code in src/openjarvis/memory/store.py

@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

@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

@abstractmethod
def count(self) -> int:
    """Return the number of stored facts."""

LocalFactStore

LocalFactStore(path: str | Path | None = None, *, 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

def __init__(
    self,
    path: str | Path | None = None,
    *,
    max_facts: int = 1000,
) -> None:
    max_facts = int(max_facts)
    if max_facts <= 0:
        raise ValueError(f"max_facts must be a positive integer, got {max_facts}")
    self._path = (
        Path(path).expanduser() if path is not None else _default_fact_path()
    )
    self._max_facts = max_facts
    self._lock = threading.Lock()
    self._facts: List[Fact] = self._load()

Attributes

path property

path: Path

Filesystem location of the JSONL store.

Functions

promote_reviewed

promote_reviewed(index: int, expected_text: str) -> bool

Atomically promote exactly the fact a user reviewed.

Source code in src/openjarvis/memory/store.py

def promote_reviewed(self, index: int, expected_text: str) -> bool:
    """Atomically promote exactly the fact a user reviewed."""
    with self._lock, _cross_process_lock(self._lock_path()):
        self._sync_from_disk_locked()
        if not 0 <= index < len(self._facts):
            return False
        fact = self._facts[index]
        if fact.text != expected_text or fact.trusted_for_recall:
            return False
        fact.trust = TRUST_TRUSTED
        self._flush()
    return True

Functions

build_memory_service

build_memory_service(config: Any, engine: Any, default_model: str = '', *, event_bus: EventBus | None = None) -> Optional[MemoryService]

Build a :class:MemoryService from config, or None if disabled.

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

def build_memory_service(
    config: Any,
    engine: Any,
    default_model: str = "",
    *,
    event_bus: EventBus | None = None,
) -> Optional[MemoryService]:
    """Build a :class:`MemoryService` from config, or ``None`` if disabled.

    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", None),
        max_facts=getattr(mem, "max_facts", 1000),
    )
    extractor = FactExtractor(engine, model)
    return MemoryService(
        store, extractor, event_bus=event_bus, scanner=_build_scanner()
    )

publish_completed_exchange

publish_completed_exchange(bus: EventBus | None, user_text: str, assistant_text: str = '', *, source: str = '') -> bool

Publish a completed chat exchange for lifecycle subscribers.

Source code in src/openjarvis/memory/service.py

def publish_completed_exchange(
    bus: EventBus | None,
    user_text: str,
    assistant_text: str = "",
    *,
    source: str = "",
) -> bool:
    """Publish a completed chat exchange for lifecycle subscribers."""
    if bus is None or not user_text or not user_text.strip():
        return False
    bus.publish(
        EventType.CHAT_EXCHANGE_COMPLETED,
        {
            "user_text": user_text,
            "assistant_text": assistant_text or "",
            "source": source,
        },
    )
    return True

create_fact_store

create_fact_store(backend: str = 'local', *, path: str | Path | None = None, max_facts: int = 1000) -> FactStore

Construct a fact store for the configured backend.

Only the "local" (on-disk JSONL) backend is supported today; the registry-backed constructor exists so additional backends can be added without changing the service or CLI wiring.

Source code in src/openjarvis/memory/store.py

def create_fact_store(
    backend: str = "local",
    *,
    path: str | Path | None = None,
    max_facts: int = 1000,
) -> FactStore:
    """Construct a fact store for the configured *backend*.

    Only the ``"local"`` (on-disk JSONL) backend is supported today; the
    registry-backed constructor exists so additional backends can be added
    without changing the service or CLI wiring.
    """
    _ensure_fact_store_backends_registered()
    key = (backend or "local").strip().lower()
    if not FactStoreRegistry.contains(key):
        supported = ", ".join(FactStoreRegistry.keys())
        raise ValueError(
            f"Unknown memory backend '{backend}'. Supported backends: {supported}"
        )
    return FactStoreRegistry.create(key, path, max_facts=max_facts)

load_configured_facts

load_configured_facts(config: Any) -> List[Fact]

Load automatic-memory facts from config when the service is enabled.

Context injection is also used by short-lived commands such as jarvis ask, where no :class:MemoryService instance exists. This helper gives those callers the same configured fact-store view without coupling them to the service lifecycle.

Source code in src/openjarvis/memory/store.py

def load_configured_facts(config: Any) -> List[Fact]:
    """Load automatic-memory facts from *config* when the service is enabled.

    Context injection is also used by short-lived commands such as
    ``jarvis ask``, where no :class:`MemoryService` instance exists.  This
    helper gives those callers the same configured fact-store view without
    coupling them to the service lifecycle.
    """
    memory = getattr(config, "memory", None)
    if memory is None or not getattr(memory, "enabled", False):
        return []

    store = create_fact_store(
        getattr(memory, "backend", "local"),
        path=getattr(memory, "facts_path", None),
        max_facts=getattr(memory, "max_facts", 1000),
    )
    # This helper feeds short-lived model paths (ask, SDK, system
    # orchestrator). Keep the full list available through FactStore.list() for
    # auditing/CLI use, but never return quarantined facts for model recall.
    return [fact for fact in store.list() if fact.trusted_for_recall]