symfonic.capabilities.memory.in_memory

in_memory

The in-process reference implementation of all three memory ports.

Not a test double. CON-S-4 requires every port to ship a working in-process implementation co-owned with the port, so the framework composes with zero optional dependencies — an adopter gets a real HMS from pip install symfonic-core with no graph database in sight, and only reaches for Postgres or Mongo when they outgrow a process. The discriminating question from CADR-12 §5 is "would a production configuration ever select this by default?", and for the default single-process deployment the answer is yes.

.. note:: CADR-12 also rules that the in-memory reference stores behind the HMS GraphBackend/VectorBackend ports are integration reference backends. That ruling is about the storage ports (registry row 9) and stands; this class implements the three bridge-facing ports defined in :mod:.ports, which have no backend to be a reference for. See the T3.3.1 evidence for the full placement argument.

Two behaviours are contract, not convenience, and every adapter of these ports owes them:

• Write, then flush. A write lands in a pending buffer; only :meth:flush makes it retrievable. A memory that became visible the instant the post-response stage wrote it would be read back by the next turn before the invocation that produced it had finished — and a cancelled invocation could not take it back.
• Visibility runs one way. A memory is returned only when its scope covers the query's. That is enforced here, in the adapter, because SEC-TEN-5 requires a backend to isolate tenants without importing whatever derived the tenant.

Concurrency: every method mutates only inside a single synchronous span with no await in it, so two concurrent invocations on one event loop cannot interleave a half-applied write. There is no lock because there is nothing for one to protect.

InMemoryHms

InMemoryHms(*, records: Iterable[MemoryRecord] = (), capacity: int = 1024)

A complete HMS held in one process's memory.

Source code in src/symfonic/capabilities/memory/in_memory.py

def __init__(
    self,
    *,
    records: Iterable[MemoryRecord] = (),
    capacity: int = 1024,
) -> None:
    #: scope path -> record id -> record, for retrievable memories.
    self._committed: dict[str, dict[str, MemoryRecord]] = {}
    #: scope path -> record id -> record, for written-but-unflushed memories.
    self._pending: dict[str, dict[str, MemoryRecord]] = {}
    self._capacity = capacity
    for record in records:
        self._committed.setdefault(record.scope_path, {})[record.record_id] = record

discard async

discard(scope: MemoryScope) -> LifecycleReceipt

Drop the pending buffer for scope and below, committing nothing.

The optional rollback capability (MemoryDiscardPort). forget cannot serve as one: it erases the committed memories too, so a turn taking back its own writes would take every earlier turn's with them.

Source code in src/symfonic/capabilities/memory/in_memory.py

async def discard(self, scope: MemoryScope) -> LifecycleReceipt:
    """Drop the *pending* buffer for ``scope`` and below, committing nothing.

    The optional rollback capability (``MemoryDiscardPort``). ``forget``
    cannot serve as one: it erases the committed memories too, so a turn
    taking back its own writes would take every earlier turn's with them.
    """
    discarded: list[str] = []
    for scope_path in self._covered(self._pending, scope):
        discarded.extend(self._pending.pop(scope_path))
    return LifecycleReceipt(scope_path=scope.path, discarded=tuple(sorted(discarded)))

scan_candidates async

scan_candidates(query: MemoryQuery) -> RetrievalResult

Rank eligible rows before capping, retaining O(candidate_limit) rows.

The in-process reference scans its existing store, not a copied list. The ceiling bounds retained candidates, not CPU spent scoring rows.

Source code in src/symfonic/capabilities/memory/in_memory.py

async def scan_candidates(self, query: MemoryQuery) -> RetrievalResult:
    """Rank eligible rows before capping, retaining O(candidate_limit) rows.

    The in-process reference scans its existing store, not a copied list.
    The ceiling bounds retained candidates, not CPU spent scoring rows.
    """
    from .queries import rank_key
    from .record_admission import conversation_local, is_profile

    def eligible():
        for item in self._candidates(query):
            if (query.session_id and conversation_local(item.record)
                    and item.record.metadata.get("session_id") != query.session_id):
                continue
            yield item

    # Reserve a human profile before ordinary cue ranking, as the
    # coordinator does. Otherwise a crowded store can starve that gate.
    profiles = nsmallest(
        1, (item for item in eligible() if is_profile(item.record)), key=rank_key,
    )
    reserved = {item.record.record_id for item in profiles}
    count = 0

    def ordinary():
        nonlocal count
        for item in eligible():
            count += 1
            if item.record.record_id not in reserved:
                yield item

    candidates = nsmallest(query.candidate_limit, ordinary(), key=rank_key)
    candidates = profiles + candidates[:query.candidate_limit - len(profiles)]
    return RetrievalResult(
        memories=rank(candidates),
        unavailable=("memory_scan_incomplete",) if count > query.candidate_limit else (),
    )