symfonic.memory.reconcile

reconcile

Deciding what a crash left behind: publish, discard, or quarantine.

Pending rows survive a process (that is the point of persisting them), so a restarted process finds writes whose run it never saw. Three outcomes, and the third is the one that matters:

• the run finalized successfully -> publish (:func:published_properties);
• the run failed or was rolled back -> discard, the rows are deleted;
• the outcome is unknown -> the rows stay hidden, stay exempt from every clock, and are marked quarantined with a diagnostic.

Nothing is published because time passed and nothing is expired because time passed. Both of those are the same mistake in opposite directions: a timer that publishes invents a turn that may never have completed, and a timer that expires destroys work a delayed finalization was about to claim. A row nobody can vouch for is a row a human has to look at, and quarantine is how it says so without pretending to have decided.

What is not wired here, stated plainly rather than faked. The repo's authoritative run/checkpoint state is :class:CheckpointRegistry, and its _refs live in one process's memory — reconcile_crash_expiry reconciles against state that a crash takes with it. So there is no durable ledger a restarted process can consult to learn that batch B belonged to a finalized run. :class:UnknownOutcomes is therefore the honest default binding: it answers :data:RunOutcome.UNKNOWN for everything, which routes every recovered row to quarantine. An adopter with a durable run ledger binds :class:RunOutcomePort to it and gets the other two branches. Supplying a default that guessed "finalized" would publish, on every restart, exactly the memories the write/flush split exists to withhold.

The other half, also not wired. Reconciliation needs to find the pending rows, and GraphBackend's reads are ancestor-scoped: asking at scope S returns rows at S and above, never below. :func:pending_rows_at is therefore exact-scope — correct for the realistic recovery path, where a resumed session knows its own scope — and a sweep for "every pending row anywhere under this tenant" would need a descendant read, the mirror of delete_subtree, which T3.3.4 finding 5 did not ask for and this module does not invent.

PendingReconciler

PendingReconciler(graph: Any, *, outcomes: RunOutcomePort | None = None)

Resolves recovered pending rows against authoritative run state.

Source code in src/symfonic/memory/reconcile.py

def __init__(self, graph: Any, *, outcomes: RunOutcomePort | None = None) -> None:
    self._graph = graph
    self._outcomes = outcomes if outcomes is not None else UnknownOutcomes()

reconcile async

reconcile(scope: TenantScope) -> ReconciliationOutcome

Resolve every pending row at scope, one batch at a time.

The outcome is asked once per batch, not once per row: a batch is the unit a run either finished or did not, and asking per row invites a ledger that is momentarily inconsistent to publish half of one.

Source code in src/symfonic/memory/reconcile.py

async def reconcile(self, scope: TenantScope) -> ReconciliationOutcome:
    """Resolve every pending row at ``scope``, one batch at a time.

    The outcome is asked once per batch, not once per row: a batch is the
    unit a run either finished or did not, and asking per row invites a
    ledger that is momentarily inconsistent to publish half of one.
    """
    rows = await pending_rows_at(self._graph, scope)
    published: list[str] = []
    discarded: list[str] = []
    quarantined: list[str] = []
    diagnostics: list[str] = []
    decided: dict[str, RunOutcome] = {}

    for node in rows:
        batch = str(node.properties.get(PENDING_BATCH_KEY) or "")
        if batch not in decided:
            decided[batch] = await self._outcomes.outcome_for(batch)
        outcome = decided[batch]
        node_id = NodeId(str(node.id))

        if outcome is RunOutcome.FINALIZED:
            await self._graph.update_node(
                scope,
                node_id,
                {"properties": published_properties(node.properties)},
            )
            published.append(str(node.id))
        elif outcome is RunOutcome.FAILED:
            await self._graph.delete_node(scope, node_id, cascade=True)
            discarded.append(str(node.id))
        else:
            reason = (
                f"no authoritative outcome for write batch {batch!r}; "
                "held hidden and unexpiring pending a decision"
            )
            await self._graph.update_node(
                scope,
                node_id,
                {
                    "properties": quarantined_properties(
                        node.properties, reason=reason
                    )
                },
            )
            quarantined.append(str(node.id))
            diagnostics.append(f"{node.id}: {reason}")

    return ReconciliationOutcome(
        published=tuple(sorted(published)),
        discarded=tuple(sorted(discarded)),
        quarantined=tuple(sorted(quarantined)),
        diagnostics=tuple(sorted(diagnostics)),
    )

ReconciliationOutcome dataclass

ReconciliationOutcome(published: tuple[str, ...] = (), discarded: tuple[str, ...] = (), quarantined: tuple[str, ...] = (), diagnostics: tuple[str, ...] = ())

What one reconciliation pass did, per row.

needs_attention property

needs_attention: bool

Whether a human has to look. Quarantine is never self-clearing.

RunOutcome

Bases: StrEnum

What an authoritative run ledger says about one write batch.

RunOutcomePort

Bases: Protocol

The seam to whatever the adopter treats as authoritative run state.

outcome_for async

outcome_for(batch_id: str) -> RunOutcome

Report the outcome of the run that produced batch_id.

Must answer :data:RunOutcome.UNKNOWN rather than guessing. An implementation that returned FINALIZED for a batch it has no record of would publish a rolled-back turn's memories.

Source code in src/symfonic/memory/reconcile.py

async def outcome_for(self, batch_id: str) -> RunOutcome:
    """Report the outcome of the run that produced ``batch_id``.

    Must answer :data:`RunOutcome.UNKNOWN` rather than guessing. An
    implementation that returned ``FINALIZED`` for a batch it has no record
    of would publish a rolled-back turn's memories.
    """
    ...

UnknownOutcomes

The default binding: knows nothing, and says so.

Not a stub standing in for a real implementation — it is the correct answer for a process with no durable run ledger, which is every process here today (see the module docstring). Every recovered batch quarantines.

pending_rows_at async

pending_rows_at(graph: Any, scope: TenantScope) -> list[Any]

Pending rows stored at exactly scope.

query_nodes returns the query's ancestors too, and an ancestor's pending rows belong to a different run at a wider scope; reconciling them from here would resolve another owner's writes. The subtree predicate does the filtering, which for an ancestor-returning read means exact match.

Source code in src/symfonic/memory/reconcile.py

async def pending_rows_at(graph: Any, scope: TenantScope) -> list[Any]:
    """Pending rows stored at exactly ``scope``.

    ``query_nodes`` returns the query's *ancestors* too, and an ancestor's
    pending rows belong to a different run at a wider scope; reconciling them
    from here would resolve another owner's writes. The subtree predicate does
    the filtering, which for an ancestor-returning read means exact match.
    """
    rows = await graph.query_nodes(scope, {}, limit=None)
    return [
        node
        for node in rows
        if is_pending(node.properties)
        and is_in_subtree(
            stored_scope_path(node.properties, node.tenant_id), scope.scope_path
        )
    ]