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.
Classes
dataclass
Fact(text: str, source: str = '', created_at: float = 0.0, trust: str = '')
A single durable memory entry.
Attributes
property
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.
Bases: ABC
Abstract persistent store for extracted memory facts.
Functions
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(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(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(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(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(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)
|
abstractmethod
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."""
|
abstractmethod
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."""
|
abstractmethod
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(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
property
Filesystem location of the JSONL store.
Functions
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
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(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]
|