symfonic.capabilities.memory.capture

capture

Opt-in, in-process score observations for paired local evaluation.

Source record identities exist only in this collector to join two executions. Only :class:PairedCandidateDecision may leave it. Nothing is persisted or exported by this module, and the default coordinator never constructs one.

LocalCandidateCapture

LocalCandidateCapture(side: Literal['legacy', 'native'])

One route's one-query observations, retained only until local pairing.

A capture is deliberately single-use. Reusing it across turns could join unrelated candidates that happen to have the same source identity.

Activation-only selected rows retain actual traversal provenance and have no candidate gate score.

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

def __init__(self, side: Literal["legacy", "native"]) -> None:
    if side not in ("legacy", "native"):
        raise ValueError("capture side must be legacy or native")
    self.side = side
    self._decisions: dict[str, _LocalDecision] | None = None
    self._query: MemoryQuery | None = None
    self._selected_lines: dict[str, str] = {}
    self._finalized = False
    self._invalid = False
    self._block = ""
    self._provider_visible: bool | None = None
    self._provider_observed = False

finalize

finalize(hydration: CoordinatedHydration) -> None

Resolve hydration inclusion after an eager, verbatim block exists.

Later prompt assembly can still reject this contribution. This call supports the direct declared block and refuses ambiguous line matches; operators must verify downstream assembly separately.

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

def finalize(self, hydration: CoordinatedHydration) -> None:
    """Resolve hydration inclusion after an eager, verbatim block exists.

    Later prompt assembly can still reject this contribution. This call
    supports the direct declared block and refuses ambiguous
    line matches; operators must verify downstream assembly separately.
    """
    if self._invalid or self._decisions is None or self._finalized:
        raise ValueError("capture must be observed once before finalization")
    if hydration.query != self._query:
        raise ValueError("hydration is not from the captured query")
    if hydration.contribution is None and hydration.block:
        raise ValueError("no declared hydration block")
    if (
        hydration.contribution is not None
        and hydration.contribution.source.block != hydration.block
    ):
        raise ValueError("hydration block is transformed or ambiguous")

    result = hydration.retrieval.result
    selected = (*result.memories, *(item.memory for item in result.linked))
    if {item.record.record_id for item in selected} != set(self._selected_lines):
        raise ValueError("hydration is not from this captured retrieval")
    working_lines = [turn.line() for turn in hydration.working.turns]
    candidate_lines = [*working_lines, *self._selected_lines.values()]
    if len(set(candidate_lines)) != len(candidate_lines):
        raise ValueError("duplicate rendered lines cannot establish visibility")
    block_lines = hydration.block.split("\n") if hydration.block else []
    if (
        len(set(block_lines)) != len(block_lines)
        or not set(block_lines) <= set(candidate_lines)
    ):
        raise ValueError("hydration block is transformed or ambiguous")
    self._decisions = {
        source_id: _LocalDecision(
            source_id=item.source_id,
            score=item.score,
            score_is_final=item.score_is_final,
            gate_admitted=item.gate_admitted,
            gate_reason=item.gate_reason,
            rank_selected=item.rank_selected,
            activation_selected=item.activation_selected,
            linked_selected=item.linked_selected,
            hydration_included=(
                (item.rank_selected or item.linked_selected)
                and self._selected_lines[source_id] in block_lines
            ),
        )
        for source_id, item in self._decisions.items()
    }
    self._finalized = True
    self._block = hydration.block

invalidate

invalidate() -> None

Discard an ambiguous capture without affecting the retrieval turn.

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

def invalidate(self) -> None:
    """Discard an ambiguous capture without affecting the retrieval turn."""
    self._invalid = True
    self._decisions = None
    self._selected_lines = {}
    self._finalized = False
    self._query = None
    self._block = ""
    self._provider_visible = None
    self._provider_observed = False

observe

observe(query: MemoryQuery, candidates: tuple[Candidate, ...], kept: tuple[RetrievedMemory, ...], selected: tuple[RetrievedMemory, ...], observed: list[tuple[str, float | None, str]] | None, activated: tuple[RetrievedMemory, ...] = (), activation_nodes: tuple[ActivatedNode, ...] = (), *, linked: tuple[LinkedRecall, ...] = ()) -> None

Record the gate and rank selection before hydration's final bound.

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

def observe(
    self,
    query: MemoryQuery,
    candidates: tuple[Candidate, ...],
    kept: tuple[RetrievedMemory, ...],
    selected: tuple[RetrievedMemory, ...],
    observed: list[tuple[str, float | None, str]] | None,
    activated: tuple[RetrievedMemory, ...] = (),
    activation_nodes: tuple[ActivatedNode, ...] = (),
    *, linked: tuple[LinkedRecall, ...] = (),
) -> None:
    """Record the gate and rank selection before hydration's final bound."""
    ranked_ids = {item.record.record_id for item in selected}
    linked_ids = {item.record.record_id for item in linked}
    if ranked_ids & linked_ids:
        raise ValueError("ranked and linked selection overlap")
    selected = (*selected, *(item.memory for item in linked))
    if self._decisions is not None or self._invalid:
        raise ValueError("capture already observed or invalidated")
    if observed is None or len(observed) != len(candidates):
        raise ValueError("missing gate observations")
    kept_ids = {item.record.record_id for item in kept}
    selected_ids = {item.record.record_id for item in selected}
    if len(kept_ids) != len(kept) or len(selected_ids) != len(selected):
        raise ValueError("duplicate candidate identities cannot be paired safely")
    decisions: dict[str, _LocalDecision] = {}
    for candidate, (observed_id, gate_score, reason) in zip(candidates, observed, strict=True):
        source_id = candidate.record.record_id
        if source_id in decisions:
            raise ValueError("duplicate candidate identities cannot be paired safely")
        if observed_id != source_id or reason not in GATE_REASONS:
            raise ValueError("invalid gate observation")
        if (source_id in kept_ids) != (reason == "admitted"):
            raise ValueError("gate observation disagrees with admitted rows")
        score = candidate.signals.cue if self.side == "legacy" else gate_score
        if self.side == "legacy" and not candidate.score_is_final:
            raise ValueError("legacy capture requires a final legacy score marker")
        if gate_score is not None and (
            isinstance(gate_score, bool)
            or not isinstance(gate_score, (int, float))
            or not math.isfinite(gate_score)
            or not 0.0 <= gate_score <= 1.0
        ):
            raise ValueError("invalid numeric gate score")
        decisions[source_id] = _LocalDecision(
            source_id=source_id,
            score=score,
            score_is_final=candidate.score_is_final,
            gate_admitted=source_id in kept_ids,
            rank_selected=source_id in ranked_ids,
            linked_selected=source_id in linked_ids,
            gate_reason=reason,
        )
    activated_by_id = {item.record.record_id: item for item in activated}
    nodes_by_id = {node.record_id: node for node in activation_nodes if node.hop > 0}
    if len(activated_by_id) != len(activated) or len(nodes_by_id) != sum(
        node.hop > 0 for node in activation_nodes
    ):
        raise ValueError("ambiguous activation provenance")
    for item in selected:
        source_id = item.record.record_id
        if source_id in kept_ids and source_id not in linked_ids:
            continue
        activated_item = activated_by_id.get(source_id)
        node = nodes_by_id.get(source_id)
        if (
            activated_item is None or node is None
            or activated_item != item
            or item.record.origin != activation_origin(node.source_id, node.hop)
            or item.score != node.score
        ):
            raise ValueError("activated row lacks verified traversal provenance")
        prior = decisions.get(source_id)
        decisions[source_id] = _LocalDecision(
            source_id=source_id,
            score=prior.score if prior else None,
            score_is_final=prior.score_is_final if prior else False,
            gate_admitted=prior.gate_admitted if prior else False,
            gate_reason=prior.gate_reason if prior else None,
            rank_selected=source_id in ranked_ids,
            linked_selected=source_id in linked_ids,
            activation_selected=True,
        )
    self._decisions = decisions
    self._selected_lines = {item.record.record_id: item.line() for item in selected}
    self._selected_lines.update({item.record.record_id: item.line() for item in linked})
    self._query = query

observe_provider_messages

observe_provider_messages(messages: object) -> None

Observe the exact message argument at one model call, once.

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

def observe_provider_messages(self, messages: object) -> None:
    """Observe the exact message argument at one model call, once."""
    from symfonic.capabilities.memory.capture_visibility import UnobservableWire, visible_block

    if self._invalid or not self._finalized or self._provider_observed:
        raise ValueError("provider observation requires one finalized capture")
    self._provider_observed = True
    try:
        self._provider_visible = visible_block(
            messages, self._block, tuple(self._selected_lines.values())
        )
    except UnobservableWire:
        # The gate/hydration decisions remain valid; visibility is unknown.
        return
    except ValueError:
        self.invalidate()
        raise

PairedCandidateDecision dataclass

PairedCandidateDecision(query_id: str, candidate_id: str, legacy_raw_score: float | None, native_gate_score: float | None, native_score_is_final: bool, legacy_present: bool, native_present: bool, legacy_coordinator_admitted: bool, native_coordinator_admitted: bool, legacy_gate_reason: str | None, native_gate_reason: str | None, legacy_rank_selected: bool, native_rank_selected: bool, legacy_activation_selected: bool, native_activation_selected: bool, legacy_hydration_included: bool, native_hydration_included: bool, legacy_provider_visible: bool | None = None, native_provider_visible: bool | None = None, legacy_linked_selected: bool = False, native_linked_selected: bool = False)

Numeric-only row with fresh IDs unrelated to source identities.

Legacy coordinator admission is the post-port gate, not the legacy engine's earlier retrieval/admission decision.

pair_captures

pair_captures(legacy: LocalCandidateCapture, native: LocalCandidateCapture) -> tuple[PairedCandidateDecision, ...]

Join within this process, then release only fresh-ID numeric rows.

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

def pair_captures(
    legacy: LocalCandidateCapture, native: LocalCandidateCapture
) -> tuple[PairedCandidateDecision, ...]:
    """Join within this process, then release only fresh-ID numeric rows."""
    from symfonic.capabilities.memory.capture_pairing import pair_captures as pair
    return pair(legacy, native)