symfonic.memory.orchestrator.orchestrator_writes

orchestrator_writes

The orchestrator's write path: extract, commit, and upgrade durability.

Split out of :mod:symfonic.memory.orchestrator.orchestrator (307 lines against the 300-line budget). MemoryOrchestrator has two halves that share construction and nothing else: hydrate_context reads memory for a turn, and the three methods here write what the turn produced -- extraction into operations, dispatch of those operations onto their layers, and the after-the-fact durability upgrade.

The class mixes this in, so MemoryOrchestrator.commit_pending and its siblings are reached exactly as before.

OrchestratorWriteMixin

Extraction, commit, and durability upgrade for MemoryOrchestrator.

Supplied by the host orchestrator; declared so the mixin's reads are typed.

commit_pending async

commit_pending(scope: TenantScope, operations: list[MemoryOperation]) -> None

Apply pending operations to the appropriate layers.

Filters operations below the write_policy.min_importance_threshold before committing.

Parameters:

Name	Type	Description	Default
scope	TenantScope	Tenant scope for isolation.	required
operations	list[MemoryOperation]	List of operations to apply.	required

Source code in src/symfonic/memory/orchestrator/orchestrator_writes.py

async def commit_pending(
    self,
    scope: TenantScope,
    operations: list[MemoryOperation],
) -> None:
    """Apply pending operations to the appropriate layers.

    Filters operations below the write_policy.min_importance_threshold
    before committing.

    Args:
        scope: Tenant scope for isolation.
        operations: List of operations to apply.
    """
    threshold = self._config.write_policy.min_importance_threshold

    for op in operations:
        if op.importance < threshold:
            logger.debug(
                "Skipping operation %s (importance %.1f < threshold %.1f)",
                op.action, op.importance, threshold,
            )
            continue

        if op.action == "noop":
            continue

        layer = self._layers.get(op.layer)
        if layer is None:
            logger.warning("Layer %s not enabled, skipping operation", op.layer)
            continue

        entry = MemoryEntry(
            layer=op.layer,
            tenant_id=scope.tenant_id,
            content=op.node.label if op.node else "",
            node_id=op.node.id if op.node else None,
            importance=op.importance,
            metadata=op.node.properties if op.node else {},
        )

        await layer.write(scope, entry)

extract_memories async

extract_memories(scope: TenantScope, interaction: dict[str, Any], llm: Any, *, callback_manager: Any = None, run_id: str = '') -> list[MemoryOperation]

Extract structured memory operations from a conversation turn.

Uses the LLM to analyze the interaction and produce MemoryOperations for each memory layer as appropriate.

Parameters:

Name	Type	Description	Default
scope	TenantScope	Tenant scope for isolation.	required
interaction	dict[str, Any]	Dict with 'user_message' and 'assistant_response'.	required
llm	Any	LLM instance for extraction.	required
callback_manager	Any	v7.4.3 (adopter Ask 5) -- forwarded to the underlying ExtractionService so the consolidation LLM call emits on_llm_end(node_name="consolidation_extractor"). None preserves byte-identical pre-7.4.3 behaviour.	None
run_id	str	Engine run identifier for callback correlation.	''

Returns:

Type	Description
list[MemoryOperation]	List of MemoryOperations to be applied via commit_pending.

Source code in src/symfonic/memory/orchestrator/orchestrator_writes.py

async def extract_memories(
    self,
    scope: TenantScope,
    interaction: dict[str, Any],
    llm: Any,
    *,
    callback_manager: Any = None,
    run_id: str = "",
) -> list[MemoryOperation]:
    """Extract structured memory operations from a conversation turn.

    Uses the LLM to analyze the interaction and produce MemoryOperations
    for each memory layer as appropriate.

    Args:
        scope: Tenant scope for isolation.
        interaction: Dict with 'user_message' and 'assistant_response'.
        llm: LLM instance for extraction.
        callback_manager: v7.4.3 (adopter Ask 5) -- forwarded to the
            underlying ``ExtractionService`` so the consolidation
            LLM call emits ``on_llm_end(node_name="consolidation_extractor")``.
            ``None`` preserves byte-identical pre-7.4.3 behaviour.
        run_id: Engine run identifier for callback correlation.

    Returns:
        List of MemoryOperations to be applied via commit_pending.
    """
    return await self._extraction.extract(
        scope, interaction, llm,
        callback_manager=callback_manager,
        run_id=run_id,
    )

upgrade_durability async

upgrade_durability(scope: TenantScope, node_id: NodeId, new_durability: Durability) -> None

Flip the durability marker on a stored graph node (v7.26.2).

The explicit promotion / retirement signal for the consolidation gate. Typical use: write a transient row ("syncing..."), then once the state stabilises write a durable row ("complete: 42 messages") and call upgrade_durability(transient_id, "expired") so the transient row is pruned on the next Phase 8 cleanup.

Implementation (Contract E, verified per-backend at T0): all three graph backends' update_node either REPLACE the whole properties dict (InMemory, Postgres) or route a top-level properties key to a full $set (Mongo, after the v7.26.2 translation fix). A flat durability key cannot be used because it collides with the read-only MemoryNode.durability computed property. So this is a read-modify-write: fetch the node, mutate the properties dict, write the FULL dict back. This preserves every OTHER property on the node.

Raises:

Type	Description
ValueError	on an unknown durability literal (fail-loud).
KeyError / backend error	when the node does not exist.

Source code in src/symfonic/memory/orchestrator/orchestrator_writes.py

async def upgrade_durability(
    self,
    scope: TenantScope,
    node_id: NodeId,
    new_durability: Durability,
) -> None:
    """Flip the durability marker on a stored graph node (v7.26.2).

    The explicit promotion / retirement signal for the consolidation
    gate.  Typical use: write a transient row ("syncing..."), then once
    the state stabilises write a durable row ("complete: 42 messages")
    and call ``upgrade_durability(transient_id, "expired")`` so the
    transient row is pruned on the next Phase 8 cleanup.

    Implementation (Contract E, verified per-backend at T0): all three
    graph backends' ``update_node`` either REPLACE the whole properties
    dict (InMemory, Postgres) or route a top-level ``properties`` key to
    a full ``$set`` (Mongo, after the v7.26.2 translation fix).  A flat
    ``durability`` key cannot be used because it collides with the
    read-only ``MemoryNode.durability`` computed property.  So this is a
    read-modify-write: fetch the node, mutate the properties dict, write
    the FULL dict back.  This preserves every OTHER property on the node.

    Raises:
        ValueError: on an unknown durability literal (fail-loud).
        KeyError / backend error: when the node does not exist.
    """
    validated = coerce_durability(new_durability)
    if self._graph is None:
        raise RuntimeError(
            "upgrade_durability requires a graph store; "
            "orchestrator was constructed without one.",
        )
    node = await self._graph.get_node(scope, node_id)
    if node is None:
        raise KeyError(
            f"Node {node_id} not found for tenant {scope.tenant_id}",
        )
    merged: dict[str, object] = dict(node.properties or {})
    merged["durability"] = validated
    # Full-dict write -- the backend full-replaces properties.  Because
    # we copied the existing dict first, no sibling property is lost.
    await self._graph.update_node(scope, node_id, {"properties": merged})
    logger.info(
        "Durability upgraded: tenant=%s node=%s -> %s",
        scope.tenant_id, node_id, validated,
    )