symfonic.core.callbacks.lifecycle_scopes

lifecycle_scopes

The ledger of lifecycle scopes a turn has opened and not yet closed.

TA8.29. A node_start without its node_end, or an llm_start without its llm_end, is an unbalanced pair: a span left open, a timer never stopped, an accounting scope that never closed. Before this module the typed streaming path produced one on every abandoned or cancelled turn, and then dispatched the normal AgentEndEvent on top of it, so an observer could not tell the abandoned turn from the answered one.

Two rules decide everything here, and they are opposites that are easy to confuse:

• Closing an open scope with a non-successful outcome is cleanup. The scope really was opened, the run really did stop, and saying so balances the pair.
• Fabricating a successful close is not. A NodeEndEvent for a node that never finished, or an LLMEndEvent carrying invented token counts, reports work that did not happen — and a plausible number is worse than a missing one, because it survives into a bill.

So an open node scope closes as :class:NodeErrorEvent (the non-successful close the vocabulary already had) and an open LLM scope closes as an :class:LLMEndEvent whose outcome says it was closed rather than finished and whose usage is left empty — nothing was reported, so nothing is claimed.

Cleanup is idempotent: exactly one termination per open scope. Not zero, which is the unbalanced pair this exists to remove, and not two, which would close a span twice and double-count its cost. Every entry is removed from the ledger before it is dispatched, so a close that races another close finds nothing left to close.

The termination latch is per turn, not per run id for the lifetime of the manager. :meth:LifecycleScopeLedger.begin_turn clears it, and the manager calls that from on_agent_start. A run id really is re-entered in shipped code -- SymfonicAgent.resume and resume_interrupt re-enter stream_events with the PAUSED run's id -- and a latch that never cleared would drop the resumed turn's AgentEndEvent entirely and leave the scopes that turn opened unclosed: the same unbalanced pair, reintroduced on the resume path.

Scope of the exactly-once claim. Scopes are keyed (run_id, node_name), which is the keying the OTel CallbackBridge already uses. That key distinguishes scopes across runs and across nodes, and it distinguishes sequential scopes within one node (start/end/start/end). It does NOT distinguish two scopes open concurrently under the same node_name in one run: the second llm_start replaces the first in the ledger, so a teardown closes one where two were opened. The exactly-once claim is therefore made over sequential lifecycle pairs; concurrent same-name pairs would need a key carrying the call's identity (an iteration or turn index), which the close sites -- on_node_end / on_llm_end -- do not currently supply.

LifecycleScopeLedger

LifecycleScopeLedger()

Remembers open node/LLM scopes per run and closes them exactly once.

Source code in src/symfonic/core/callbacks/lifecycle_scopes.py

def __init__(self) -> None:
    self._nodes: OrderedDict[tuple[str, str], NodeStartEvent] = OrderedDict()
    self._llms: OrderedDict[tuple[str, str], LLMStartEvent] = OrderedDict()
    self._ended: OrderedDict[str, bool] = OrderedDict()

begin_turn

begin_turn(run_id: str) -> None

Arm run_id to terminate again: a new turn has started.

The latch :meth:claim_termination sets is per turn. Without this, a run id that terminated once could never terminate again on the same manager, and the resume path -- which deliberately re-enters stream_events with the paused run's id -- would lose its terminal and leak every scope the resumed turn opened.

Only the latch is cleared. Scopes the previous turn left open are NOT dropped here: dropping them silently would be the zero-terminations half of the defect, so they stay in the ledger and are closed by whichever termination reaches them.

Source code in src/symfonic/core/callbacks/lifecycle_scopes.py

def begin_turn(self, run_id: str) -> None:
    """Arm ``run_id`` to terminate again: a new turn has started.

    The latch :meth:`claim_termination` sets is per turn. Without this,
    a run id that terminated once could never terminate again on the same
    manager, and the resume path -- which deliberately re-enters
    ``stream_events`` with the paused run's id -- would lose its terminal
    and leak every scope the resumed turn opened.

    Only the latch is cleared. Scopes the previous turn left open are NOT
    dropped here: dropping them silently would be the zero-terminations
    half of the defect, so they stay in the ledger and are closed by
    whichever termination reaches them.
    """
    self._ended.pop(run_id, None)

claim_termination

claim_termination(run_id: str) -> bool

True for the first caller to terminate run_id; False after.

This is what makes close-then-cancel and cancel-then-close produce one termination either way rather than two.

Source code in src/symfonic/core/callbacks/lifecycle_scopes.py

def claim_termination(self, run_id: str) -> bool:
    """True for the first caller to terminate ``run_id``; False after.

    This is what makes ``close-then-cancel`` and ``cancel-then-close``
    produce one termination either way rather than two.
    """
    if run_id in self._ended:
        return False
    self._remember(self._ended, run_id, True)
    return True

drain

drain(run_id: str, outcome: TerminationOutcome) -> list[tuple[str, object]]

Remove and describe every scope run_id still holds open.

Returns [(hook_name, event), ...] in innermost-first order — LLM scopes before the node scopes that contain them — so a consumer that nests spans unwinds them in the order it opened them.

Removal happens here, before any dispatch. A handler that re-enters the manager from inside its own teardown therefore finds an empty ledger instead of a second copy of the same close.

Source code in src/symfonic/core/callbacks/lifecycle_scopes.py

def drain(
    self, run_id: str, outcome: TerminationOutcome
) -> list[tuple[str, object]]:
    """Remove and describe every scope ``run_id`` still holds open.

    Returns ``[(hook_name, event), ...]`` in **innermost-first** order —
    LLM scopes before the node scopes that contain them — so a consumer
    that nests spans unwinds them in the order it opened them.

    Removal happens here, before any dispatch. A handler that re-enters
    the manager from inside its own teardown therefore finds an empty
    ledger instead of a second copy of the same close.
    """
    llms = [key for key in self._llms if key[0] == run_id]
    nodes = [key for key in self._nodes if key[0] == run_id]
    opened_llms = [self._llms.pop(key) for key in llms]
    opened_nodes = [self._nodes.pop(key) for key in nodes]
    if outcome == "completed":
        # Nothing is dispatched, but the entries are still dropped: they
        # belong to a run that is over either way.
        return []
    error_type, reason = _CLOSE_REASON[outcome]
    closes: list[tuple[str, object]] = [
        (
            "on_llm_end",
            LLMEndEvent(
                model=start.model,
                output="",
                # Empty, never zeroed-and-claimed: the provider reported
                # no usage for a call that never returned, and an invented
                # count is the fabrication this whole path exists to avoid.
                usage={},
                run_id=start.run_id,
                node_name=start.node_name,
                turn_index=start.turn_index,
                outcome=outcome,
            ),
        )
        for start in reversed(opened_llms)
    ]
    closes.extend(
        (
            "on_node_error",
            NodeErrorEvent(
                node_name=start.node_name,
                run_id=start.run_id,
                error=reason,
                error_type=error_type,
            ),
        )
        for start in reversed(opened_nodes)
    )
    return closes

open_scope_count

open_scope_count(run_id: str) -> int

How many scopes run_id still holds open. Diagnostics only.

Source code in src/symfonic/core/callbacks/lifecycle_scopes.py

def open_scope_count(self, run_id: str) -> int:
    """How many scopes ``run_id`` still holds open. Diagnostics only."""
    return sum(1 for key in self._nodes if key[0] == run_id) + sum(
        1 for key in self._llms if key[0] == run_id
    )