symfonic.capabilities.memory.admin

admin

Memory administration, scoped, as operations rather than objects.

Generated projects administer memory through operations, not private layers.

Operations, never the store. Handing out a layer makes every consumer invent its own idea of what a scope means, and three consumers doing that is how one tenant ends up reading another. Every method here takes a scope and honours it; there is no accessor that returns the thing underneath.

Staged rows are invisible here, and that is the point. An operator acting on a pending row would be acting on something the turn may still discard. The admin view shows what is committed; discard and publish are how staging is resolved, and they are named as the two different things they are.

MemoryAdminService

MemoryAdminService(store: Any, *, procedural: Any = None, conversation: Any = None, operation_journal: Any = None)

Read, resolve and erase one scope's memories.

One service per process, not per scope: the scope is an argument to every method, because an administrative caller acts on scopes it is authorised for rather than on the one it was built with. That is the opposite of :class:MemoryCapability, which IS a scope -- and the difference is real: a capability serves one agent's turns, this serves an operator's request.

Parameters:

Name	Type	Description	Default
store	Any	an HMS satisfying the retrieval, write and lifecycle ports. Held privately and never returned.	required
operation_journal	Any	only a native Mongo v1 store's forget records in it.	None

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

def __init__(self, store: Any, *, procedural: Any = None,
             conversation: Any = None, operation_journal: Any = None) -> None:
    """
    Args:
        store: an HMS satisfying the retrieval, write and lifecycle ports.
            Held privately and never returned.
        operation_journal: only a native Mongo v1 store's forget records in it.
    """
    self._store, self._procedural = store, procedural
    self._conversation, self._journal = conversation, operation_journal

procedural_review_available property

procedural_review_available: bool

Whether this service was composed with the procedural review door.

approve_procedure async

approve_procedure(scope: MemoryScope, procedure: Any, *, action_tool: str | None = None, precondition: Any = None) -> Any

Approve one draft, and record what a reviewer decided it governs.

Approval is the review, so this is where a tool and a precondition are named. That is not a convenience: the offline extractor reads what a scope did and has no way to know which registered tool a narrated action corresponds to, or what state must hold before repeating it. A draft it wrote governs nothing until a person says what it governs -- which is what makes "draft" the right status for it and human review the quality gate rather than a formality.

Both arguments are optional, so approving a procedure that only informs the prompt stays one call.

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

async def approve_procedure(
    self,
    scope: MemoryScope,
    procedure: Any,
    *,
    action_tool: str | None = None,
    precondition: Any = None,
) -> Any:
    """Approve one draft, and record what a reviewer decided it governs.

    Approval is the review, so this is where a tool and a precondition are
    named. That is not a convenience: the offline extractor reads *what a
    scope did* and has no way to know which registered tool a narrated
    action corresponds to, or what state must hold before repeating it. A
    draft it wrote governs nothing until a person says what it governs --
    which is what makes "draft" the right status for it and human review
    the quality gate rather than a formality.

    Both arguments are optional, so approving a procedure that only informs
    the prompt stays one call.
    """
    layer = self._require_procedural("approve a procedure")
    try:
        return await layer.approve_skill(
            _legacy(scope), _node_id(procedure),
            action_tool=action_tool, precondition=precondition,
        )
    except KeyError as exc:
        if action_tool is not None or precondition is not None:
            raise MemoryContractError(
                "cannot bind a procedure outside the supplied scope"
            ) from exc
        raise

correction async

correction(scope: MemoryScope, record_id: str, text: str, fields: Mapping[str, Any], *, salience: float = 0.9) -> MemoryRecord

Record that a person corrected their own profile. Staged, not published.

The named door for user_manual_edit. Phase 5 promotes a memory carrying that authority onto the scope's profile on the next nap, so the value is a grant rather than a description -- and a grant an extractor could mint by putting a string in its metadata bag would be no grant at all. MemoryRecord refuses it from metadata and accepts it only on its own field; this is where a deployment sets that field without writing the vocabulary out by hand.

Staged rather than published for the same reason every other write here is: a correction that failed halfway should leave nothing, and publish is the step that makes it retrievable.

Returns the record so a caller can name it in a receipt or a log.

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

async def correction(
    self,
    scope: MemoryScope,
    record_id: str,
    text: str,
    fields: Mapping[str, Any],
    *,
    salience: float = 0.9,
) -> MemoryRecord:
    """Record that a person corrected their own profile. Staged, not published.

    The named door for ``user_manual_edit``. Phase 5 promotes a memory
    carrying that authority onto the scope's profile on the next nap, so
    the value is a grant rather than a description -- and a grant an
    extractor could mint by putting a string in its metadata bag would be
    no grant at all. ``MemoryRecord`` refuses it from ``metadata`` and
    accepts it only on its own field; this is where a deployment sets that
    field without writing the vocabulary out by hand.

    Staged rather than published for the same reason every other write
    here is: a correction that failed halfway should leave nothing, and
    ``publish`` is the step that makes it retrievable.

    Returns the record so a caller can name it in a receipt or a log.
    """
    record = MemoryRecord(
        record_id=record_id,
        layer=MemoryLayer.SEMANTIC,
        text=text,
        scope_path=scope.path,
        salience=salience,
        origin="user-correction",
        # The producer's own vocabulary -- which fields, and what they now
        # say. Nested where every producer's metadata goes; the authority
        # above is the only thing that travels at the top.
        metadata=dict(fields),
        edited_by="user_manual_edit",
    )
    await self.stage(scope, (record,))
    return record

decide_draft_procedure async

decide_draft_procedure(scope: MemoryScope, procedure: Any, *, status: str, action_tool: str | None = None, precondition: Any = None, expected_source: object | None = None, expected_source_present: bool = True, expected_action_tool: str | None = None, expected_action_tool_present: bool = False, actor: str | None = None) -> Any | None

Apply an automated decision only while the scoped row is a draft.

The layer re-reads and serialises the native in-process transition.

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

async def decide_draft_procedure(
    self,
    scope: MemoryScope,
    procedure: Any,
    *,
    status: str,
    action_tool: str | None = None,
    precondition: Any = None,
    expected_source: object | None = None,
    expected_source_present: bool = True,
    expected_action_tool: str | None = None,
    expected_action_tool_present: bool = False,
    actor: str | None = None,
) -> Any | None:
    """Apply an automated decision only while the scoped row is a draft.

    The layer re-reads and serialises the native in-process transition.
    """
    _validate_draft_decision(status, action_tool, precondition)
    layer = self._require_procedural("apply an automated procedure decision")
    return await layer.decide_draft_skill(
        _legacy(scope),
        _node_id(procedure),
        status=status,
        action_tool=action_tool,
        precondition=precondition,
        expected_source=expected_source,
        expected_source_present=expected_source_present,
        expected_action_tool=expected_action_tool,
        expected_action_tool_present=expected_action_tool_present,
        actor=actor,
    )

delete_record async

delete_record(scope: MemoryScope, record_id: str)

Exact-owner storage operation; platform callers must audit first.

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

async def delete_record(self, scope: MemoryScope, record_id: str):
    """Exact-owner storage operation; platform callers must audit first."""
    from symfonic.capabilities.memory.record_access import operation
    return await operation(self._store, "delete_record")(scope, record_id)

discard async

discard(scope: MemoryScope) -> Any

Drop staging without committing any of it.

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

async def discard(self, scope: MemoryScope) -> Any:
    """Drop staging without committing any of it."""
    return await operator_call(self._store, scope, "discard", scope)

forget async

forget(scope: MemoryScope) -> Any

Erase scope and everything below it, staged and committed alike.

Idempotent: an empty scope answers with an empty receipt. When composed with a conversation log, erase that scope's transcript as well. A CORE-199 ForgetInProgress means an erase job remains; the transcript waits for the forget that completes it.

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

async def forget(self, scope: MemoryScope) -> Any:
    """Erase ``scope`` and everything below it, staged and committed alike.

    Idempotent: an empty scope answers with an empty receipt. When composed
    with a conversation log, erase that scope's transcript as well.
    A CORE-199 ``ForgetInProgress`` means an erase job remains; the
    transcript waits for the forget that completes it.
    """
    from symfonic.capabilities.memory.admin_erase import forget
    return await forget(self._store, self._conversation, scope, self._journal)

get_record async

get_record(scope: MemoryScope, record_id: str)

Direct published-record lookup, without a retrieval-page ceiling.

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

async def get_record(self, scope: MemoryScope, record_id: str):
    """Direct published-record lookup, without a retrieval-page ceiling."""
    from symfonic.capabilities.memory.record_access import operation
    return await operation(self._store, "get_record")(scope, record_id)

inventory_page async

inventory_page(scope: MemoryScope, *, layer=None, limit=200, cursor=None)

Published records in ID order; continuation is independent of recall.

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

async def inventory_page(self, scope: MemoryScope, *, layer=None, limit=200, cursor=None):
    """Published records in ID order; continuation is independent of recall."""
    from symfonic.capabilities.memory.inventory import page
    return await page(self._store, scope, layer=layer, limit=limit, cursor=cursor)

procedures async

procedures(scope: MemoryScope, *, drafts: bool = True) -> tuple[Any, ...]

What this scope has learned, for a reviewer to read.

Drafts included by default: this is the review queue, and a queue that hid what was waiting on review would be a queue with nothing in it.

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

async def procedures(
    self, scope: MemoryScope, *, drafts: bool = True
) -> tuple[Any, ...]:
    """What this scope has learned, for a reviewer to read.

    Drafts included by default: this is the review queue, and a queue that
    hid what was waiting on review would be a queue with nothing in it.
    """
    layer = self._require_procedural("list this scope's procedures")
    return tuple(
        await layer.query_skills(
            _legacy(scope), "", top_k=200, include_drafts=drafts
        )
    )

publish async

publish(scope: MemoryScope) -> Any

Commit what scope staged, and what anything below it staged.

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

async def publish(self, scope: MemoryScope) -> Any:
    """Commit what ``scope`` staged, and what anything below it staged."""
    return await operator_call(self._store, scope, "flush", scope)

record_page async

record_page(scope: MemoryScope, *, layer: MemoryLayer | None = None, limit: int = DEFAULT_PAGE) -> RetrievalResult

Bounded admin records with explicit drops and incomplete-scan signals.

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

async def record_page(
    self, scope: MemoryScope, *, layer: MemoryLayer | None = None, limit: int = DEFAULT_PAGE
) -> RetrievalResult:
    """Bounded admin records with explicit drops and incomplete-scan signals."""
    layers = RETRIEVABLE_LAYERS if layer is None else frozenset({layer})
    query = MemoryQuery(
        scope=scope,
        layers=layers,
        limit=limit,
        candidate_limit=limit,
        max_record_chars=_ADMIN_RECORD_CHARS,
        max_total_chars=(_ADMIN_RECORD_CHARS + 32) * limit,
    )
    result: RetrievalResult = await self._store.retrieve(query)
    return result

records async

records(scope: MemoryScope, *, layer: MemoryLayer | None = None, limit: int = DEFAULT_PAGE) -> tuple[MemoryRecord, ...]

Committed memories scope may read: its own and its ANCESTORS'.

Reading widens upward, never downward; erasure walks the opposite direction. layer narrows that visibility and cannot grant access. Use record_page when omission/completeness evidence is required.

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

async def records(
    self,
    scope: MemoryScope,
    *,
    layer: MemoryLayer | None = None,
    limit: int = DEFAULT_PAGE,
) -> tuple[MemoryRecord, ...]:
    """Committed memories ``scope`` may read: its own and its ANCESTORS'.

    Reading widens upward, never downward; erasure walks the opposite
    direction. ``layer`` narrows that visibility and cannot grant access.
    Use ``record_page`` when omission/completeness evidence is required.
    """
    result = await self.record_page(scope, layer=layer, limit=limit)
    return tuple(memory.record for memory in result.memories)

reject_procedure async

reject_procedure(scope: MemoryScope, procedure: Any) -> Any

Reject one draft. It stays readable and stops being active.

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

async def reject_procedure(self, scope: MemoryScope, procedure: Any) -> Any:
    """Reject one draft. It stays readable and stops being active."""
    layer = self._require_procedural("reject a procedure")
    return await layer.reject_skill(_legacy(scope), _node_id(procedure))

stage async

stage(scope: MemoryScope, records: tuple[MemoryRecord, ...]) -> Any

Prepare records under scope. Nothing is durable until publish.

Over raw PostgreSQL v1 memory it presents the running turn's erase admission and, outside a run or in one that captured none, refuses (CORE-199, operator_admission): it never observes a fresh one.

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

async def stage(
    self, scope: MemoryScope, records: tuple[MemoryRecord, ...]
) -> Any:
    """Prepare ``records`` under ``scope``. Nothing is durable until publish.

    Over raw PostgreSQL v1 memory it presents the running turn's erase
    admission and, outside a run or in one that captured none, refuses
    (CORE-199, ``operator_admission``): it never observes a fresh one.
    """
    return await operator_call(self._store, scope, "write", WriteRequest(scope, records))