symfonic.capabilities.memory.operations

operations

What one extraction operation asks for, and what it is allowed to become.

The extraction wire format is the shipped one: a JSON object with an ops list, each op naming an action, a layer, a label, properties, and an importance on the 1–10 grid. Keeping it identical is what makes the migration a service swap rather than a behaviour change.

What differs is the refusals. Every op that does not become a memory says why:

• an action that carries no node (create_edge) would have been written by the shipped commit_pending as an entry with content="" — a blank memory that costs a retrieval slot and renders as nothing;
• an unrecognised layer is skipped by the shipped parser with a bare continue;
• an importance under the write threshold is dropped during the commit, after the memory has already reached a store.

All three are silent today, and all three answer the same operator question: "why is that not remembered?"

ExtractedMemory dataclass

ExtractedMemory(record: MemoryRecord, importance: float = 5.0, action: str = 'upsert_node', properties: dict[str, Any] = dict(), redactions: tuple[str, ...] = ())

One candidate memory: the record, and what legacy needs to store it.

ExtractionRequest dataclass

ExtractionRequest(scope: MemoryScope, user_message: str = '', assistant_message: str = '', turn: int = 0, max_records: int = 50, min_importance: float = 3.0)

One turn, and the ceilings the extraction of it must respect.

ExtractionResult dataclass

ExtractionResult(scope: MemoryScope, turn: int = 0, memories: tuple[ExtractedMemory, ...] = (), dropped: tuple[tuple[str, str], ...] = (), family: ProviderFamily = ProviderFamily.UNKNOWN, redactions: tuple[str, ...] = (), degraded: bool = False, reason: str = '')

What one extraction produced, and a reason for everything it did not.

write_request

write_request() -> WriteRequest

The post-response write this extraction implies.

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

def write_request(self) -> WriteRequest:
    """The post-response write this extraction implies."""
    return WriteRequest(scope=self.scope, records=self.records, turn=self.turn)

classification_identity

classification_identity(properties: Mapping[str, Any], *, layer: Any = None) -> tuple[tuple[str, str] | None, str]

Read the storage-owned atomic classification contract, lazily.

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

def classification_identity(
    properties: Mapping[str, Any], *, layer: Any = None,
) -> tuple[tuple[str, str] | None, str]:
    """Read the storage-owned atomic classification contract, lazily."""
    from symfonic.memory.classification import classification_identity as read
    return read(properties, layer=layer)

mint_operation

mint_operation(op: Mapping[str, Any], index: int, request: ExtractionRequest, family: ProviderFamily, scrubber: CredentialScrubber) -> ExtractedMemory | tuple[str, str]

Mint one memory from one operation, or report why it produced none.

Returns an :class:ExtractedMemory, or (identifier, reason) where an empty reason means "this operation was a legitimate no-op" — a noop action is the model saying there is nothing to remember, which is an answer rather than a refusal.

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

def mint_operation(
    op: Mapping[str, Any],
    index: int,
    request: ExtractionRequest,
    family: ProviderFamily,
    scrubber: CredentialScrubber,
) -> ExtractedMemory | tuple[str, str]:
    """Mint one memory from one operation, or report why it produced none.

    Returns an :class:`ExtractedMemory`, or ``(identifier, reason)`` where an
    empty reason means "this operation was a legitimate no-op" — a ``noop``
    action is the model saying there is nothing to remember, which is an answer
    rather than a refusal.
    """
    identifier = f"op[{index}]"
    action = str(op.get("action", "noop"))
    if action == "noop":
        return identifier, ""
    if action not in NODE_ACTIONS:
        return identifier, (
            f"action {action!r} carries no memory text; only "
            f"{sorted(NODE_ACTIONS)} produce a record"
        )
    try:
        layer = resolve_layer(str(op.get("layer", MemoryLayer.SEMANTIC.value)))
    except MemoryCapabilityError as exc:
        return identifier, str(exc)

    raw_text = str(op.get("label", op.get("id", ""))).strip()
    if not raw_text:
        return identifier, "the operation carries no memory text"
    importance = _as_float(op.get("importance"), default=5.0)
    if importance < request.min_importance:
        return identifier, (
            f"importance {importance} is below the {request.min_importance} "
            "write threshold"
        )

    raw_properties = op.get("properties")
    properties, dropped_keys = scrubber.scrub_properties(
        raw_properties if isinstance(raw_properties, Mapping) else {}
    )
    identity, classification_error = classification_identity(properties, layer=layer)
    if classification_error:
        return identifier, classification_error
    if identity is None:
        return identifier, "classification_missing"
    properties["memory_category"], properties["subject"] = identity
    # A label such as ``preferred_name`` identifies a field but does not carry
    # the fact. Persist the scrubbed values in the retrievable text as well as
    # metadata, otherwise recall tells the model which fact existed and drops
    # its value (the live scaffold stored ``deployment_codename`` but not
    # ``ATLAS-8642``).
    rendered = raw_text
    # Include every property value the label does not already carry.  Testing
    # ``any`` value made ``person:Amiel`` suppress *all* properties because the
    # name was present, silently dropping ``occupation: software engineer``
    # from the only text retrieval renders.  Shared with the compatibility
    # reader so already-persisted rows gain the same complete representation.
    rendered = retrievable_text(
        raw_text,
        {key: value for key, value in properties.items() if key not in _CLASSIFICATION_KEYS},
    )
    scrubbed = scrubber.scrub_text(rendered)
    return ExtractedMemory(
        record=MemoryRecord(
            record_id=record_id(request, index, scrubbed.text),
            layer=layer,
            text=scrubbed.text,
            scope_path=request.scope.path,
            salience=importance_to_salience(importance),
            origin=f"extraction:{family.value}",
        ),
        importance=importance,
        action=action,
        properties=properties,
        redactions=scrubbed.redactions + tuple(f"KEY:{key}" for key in dropped_keys),
    )

record_id

record_id(request: ExtractionRequest, index: int, text: str) -> str

A deterministic, charset-safe id for one extracted memory.

Derived rather than taken from the label, because the label is free-form model output and the id is an upsert key inside a rendered charset — a label containing / would forge one. Derived from the turn and the text, so a retried post-response stage upserts its own memory instead of writing a second copy of it (the idempotency the write port promises).

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

def record_id(request: ExtractionRequest, index: int, text: str) -> str:
    """A deterministic, charset-safe id for one extracted memory.

    Derived rather than taken from the label, because the label is free-form
    model output and the id is an upsert key inside a rendered charset — a
    label containing ``/`` would forge one. Derived from the *turn* and the
    *text*, so a retried post-response stage upserts its own memory instead of
    writing a second copy of it (the idempotency the write port promises).
    """
    digest = hashlib.sha256(text.encode("utf-8")).hexdigest()[:12]
    return f"x{request.turn}.{index}.{digest}"