symfonic.core.callbacks

callbacks

Callback system -- protocol, base class, manager, and NoOp default.

BaseCallbackHandler

Public no-op base for CallbackHandler implementations.

Subclass and override only the hooks you need. Satisfies the :class:~symfonic.core.callbacks.protocol.CallbackHandler protocol via structural typing (every required method is present and async).

on_before_tool_call async

on_before_tool_call(event: BeforeToolCallEvent, state: dict[str, Any]) -> dict[str, Any] | None

Tool-call guard / steering hook. Returns None (proceed).

Returning None executes the tool normally. Override and return {"action": "skip", "content": str, "is_error": bool} to refuse the call and hand the model a synthetic result instead (corrective feedback or a benign cancellation).

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

async def on_before_tool_call(
    self,
    event: BeforeToolCallEvent,
    state: dict[str, Any],
) -> dict[str, Any] | None:
    """Tool-call guard / steering hook.  Returns ``None`` (proceed).

    Returning ``None`` executes the tool normally. Override and return
    ``{"action": "skip", "content": str, "is_error": bool}`` to refuse
    the call and hand the model a synthetic result instead (corrective
    feedback or a benign cancellation).
    """
    return None

on_response_render async

on_response_render(content: str, state: dict[str, Any], metadata: ResponseRenderEvent) -> str

Content-rewriter hook (v7.18.0). Returns content unchanged.

Unlike the fire-and-forget hooks, on_response_render RETURNS the (possibly rewritten) content string. The no-op default returns content verbatim so a subclass that does not override it is a pass-through in the rewriter chain.

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

async def on_response_render(
    self,
    content: str,
    state: dict[str, Any],
    metadata: ResponseRenderEvent,
) -> str:
    """Content-rewriter hook (v7.18.0).  Returns content unchanged.

    Unlike the fire-and-forget hooks, ``on_response_render`` RETURNS
    the (possibly rewritten) content string.  The no-op default
    returns ``content`` verbatim so a subclass that does not override
    it is a pass-through in the rewriter chain.
    """
    return content

on_tool_call_dispatch async

on_tool_call_dispatch(event: ToolCallDispatchEvent, state: dict[str, Any]) -> dict[str, Any] | None

Tool-call rewriter hook (v7.19.0). Returns None (no rewrite).

Returning None is the pass-through contract -- the no-op default never rewrites a dispatched tool call.

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

async def on_tool_call_dispatch(
    self,
    event: ToolCallDispatchEvent,
    state: dict[str, Any],
) -> dict[str, Any] | None:
    """Tool-call rewriter hook (v7.19.0).  Returns ``None`` (no rewrite).

    Returning ``None`` is the pass-through contract -- the no-op
    default never rewrites a dispatched tool call.
    """
    return None

on_tool_result async

on_tool_result(event: ToolResultCallbackEvent) -> None

Fired AFTER a tool returns (v8.6.4). No-op by default.

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

async def on_tool_result(self, event: ToolResultCallbackEvent) -> None:
    """Fired AFTER a tool returns (v8.6.4).  No-op by default."""
    return None

CallbackHandler

Bases: Protocol

Async callback protocol for agent graph lifecycle events.

Implement this protocol to receive notifications about agent start/end, node start/end/error, LLM invocation events, and token composition breakdowns during graph execution.

CallbackManager

CallbackManager(handlers: Sequence[Any] = ())

Fan-out dispatcher for multiple CallbackHandler instances.

Satisfies the CallbackHandler protocol (composite pattern). When the handler list is empty, all methods return immediately (zero-cost).

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

def __init__(self, handlers: Sequence[Any] = ()) -> None:
    self._handlers: list[Any] = list(handlers)
    # TA8.29: which lifecycle pairs this manager has seen opened and not
    # yet closed, per run. Composed rather than inherited -- remembering
    # scopes and fanning events out are two responsibilities, and the
    # ledger has no opinion about handlers.
    #
    # Imported here rather than at module scope so ``import symfonic`` does
    # not pay for a module no import-time code path constructs. The
    # repository's import-footprint gate measures exactly that, and one
    # more eagerly imported module put it over its ceiling.
    self._scopes = _ledger_class()()

is_noop property

is_noop: bool

True when no handlers are registered (zero-cost path).

add

add(handler: Any) -> None

Add a handler to the dispatch list.

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

def add(self, handler: Any) -> None:
    """Add a handler to the dispatch list."""
    self._handlers.append(handler)

close_open_scopes async

close_open_scopes(*, run_id: str, outcome: TerminationOutcome) -> int

Close every lifecycle pair run_id left open. Returns the count.

TA8.29. Exactly one termination per open scope: the ledger removes an entry before its close is dispatched, so a second call closes nothing and a close-then-cancel race cannot double-count a span or a token bill. Zero is the other failure -- a scope left open is the unbalanced-pair defect this method exists to remove.

Every close goes through :meth:_dispatch, which is where the two error families stay separate. A handler that raises here is logged and isolated, and the remaining handlers still get their close: one raising observer silently cancelling everyone else's teardown would be a new defect in the shape of the old one.

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

async def close_open_scopes(
    self, *, run_id: str, outcome: TerminationOutcome
) -> int:
    """Close every lifecycle pair ``run_id`` left open. Returns the count.

    TA8.29. Exactly one termination per open scope: the ledger removes an
    entry *before* its close is dispatched, so a second call closes
    nothing and a close-then-cancel race cannot double-count a span or a
    token bill. Zero is the other failure -- a scope left open is the
    unbalanced-pair defect this method exists to remove.

    Every close goes through :meth:`_dispatch`, which is where the two
    error families stay separate. A handler that raises here is logged and
    isolated, **and the remaining handlers still get their close**: one
    raising observer silently cancelling everyone else's teardown would be
    a new defect in the shape of the old one.
    """
    closes = self._scopes.drain(run_id, outcome)
    for hook, event in closes:
        await self._dispatch(hook, event)
    return len(closes)

describe_hook_handlers

describe_hook_handlers(hook_name: str) -> list[tuple[str, bool]]

Introspection helper for adopter triage (v7.18.1).

Returns [(handler_qualname, implements_hook), ...] for every registered handler. Adopters debugging "which of my handlers is responsible for has_hook returning True?" call this to verify the registration without diffing internal state.

Example::

>>> mgr.describe_hook_handlers("on_response_render")
[('myadopter.SlackRenderCallback', True),
 ('symfonic.telemetry.MetricsHandler', False)]

Zero-cost when called outside a hot path -- intended for diagnostic logs and ad-hoc REPL introspection, not steady-state dispatch. The handler descriptor includes module + qualname so a forked CallbackManager handler shows up immediately as a non-symfonic module path.

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

def describe_hook_handlers(
    self, hook_name: str,
) -> list[tuple[str, bool]]:
    """Introspection helper for adopter triage (v7.18.1).

    Returns ``[(handler_qualname, implements_hook), ...]`` for every
    registered handler.  Adopters debugging "which of my handlers is
    responsible for ``has_hook`` returning ``True``?" call this to
    verify the registration without diffing internal state.

    Example::

        >>> mgr.describe_hook_handlers("on_response_render")
        [('myadopter.SlackRenderCallback', True),
         ('symfonic.telemetry.MetricsHandler', False)]

    Zero-cost when called outside a hot path -- intended for
    diagnostic logs and ad-hoc REPL introspection, not steady-state
    dispatch.  The handler descriptor includes module + qualname
    so a forked CallbackManager handler shows up immediately as a
    non-symfonic module path.
    """
    return [
        (
            f"{type(h).__module__}.{type(h).__qualname__}",
            hasattr(h, hook_name),
        )
        for h in self._handlers
    ]

finish_turn async

finish_turn(*, run_id: str, duration_ms: float, outcome: TerminationOutcome = 'completed', final_response: str | None = None, node_count: int | None = None) -> bool

Terminate run_id once: close open scopes, then end the turn.

Returns True for the caller that actually terminated the run and False for every later one. Two exit paths can reach a turn's teardown -- a consumer's aclose and a task cancellation can arrive in either order -- and both must produce one termination.

The outcome is never collapsed. "completed" is the only value that means the turn answered; "failed", "cancelled" and "abandoned" each say why it did not, and an incomplete turn never gets the terminal a completed one gets.

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

async def finish_turn(
    self,
    *,
    run_id: str,
    duration_ms: float,
    outcome: TerminationOutcome = "completed",
    final_response: str | None = None,
    node_count: int | None = None,
) -> bool:
    """Terminate ``run_id`` once: close open scopes, then end the turn.

    Returns True for the caller that actually terminated the run and False
    for every later one. Two exit paths can reach a turn's teardown -- a
    consumer's ``aclose`` and a task cancellation can arrive in either
    order -- and both must produce **one** termination.

    The ``outcome`` is never collapsed. ``"completed"`` is the only value
    that means the turn answered; ``"failed"``, ``"cancelled"`` and
    ``"abandoned"`` each say why it did not, and an incomplete turn never
    gets the terminal a completed one gets.
    """
    if not self._scopes.claim_termination(run_id):
        return False
    await self.close_open_scopes(run_id=run_id, outcome=outcome)
    await self._dispatch(
        "on_agent_end",
        AgentEndEvent(
            run_id=run_id,
            # Not invented for an incomplete turn: an abandoned or
            # cancelled run produced no final response and counted no
            # nodes, and writing a plausible value here is exactly the
            # fabrication that survives into a metric.
            final_response=final_response if outcome == "completed" else None,
            duration_ms=duration_ms,
            node_count=node_count if outcome == "completed" else None,
            outcome=outcome,
        ),
    )
    return True

has_hook

has_hook(hook_name: str) -> bool

True when at least one handler implements the given hook.

Used at emission sites to skip event construction entirely for OPTIONAL hooks (e.g. on_llm_pre_call, on_user_correction) when no registered handler implements them. Preserves the zero-cost guarantee for optional extensions.

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

def has_hook(self, hook_name: str) -> bool:
    """True when at least one handler implements the given hook.

    Used at emission sites to skip event construction entirely for
    OPTIONAL hooks (e.g. ``on_llm_pre_call``, ``on_user_correction``)
    when no registered handler implements them.  Preserves the
    zero-cost guarantee for optional extensions.
    """
    if not self._handlers:
        return False
    return any(hasattr(h, hook_name) for h in self._handlers)

merge

merge(other: CallbackManager) -> CallbackManager

Return a new CallbackManager combining both handler lists.

The merged manager gets a fresh scope ledger, which is correct because the merged object is what the run's emission sites are handed: every node_start and llm_start of that run is recorded on it, and every close is dispatched from it.

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

def merge(self, other: CallbackManager) -> CallbackManager:
    """Return a new CallbackManager combining both handler lists.

    The merged manager gets a **fresh** scope ledger, which is correct
    because the merged object is what the run's emission sites are handed:
    every ``node_start`` and ``llm_start`` of that run is recorded on it,
    and every close is dispatched from it.
    """
    return CallbackManager(self._handlers + other._handlers)

on_agent_end async

on_agent_end(event: AgentEndEvent) -> None

Dispatch AgentEndEvent to all handlers.

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

async def on_agent_end(self, event: AgentEndEvent) -> None:
    """Dispatch AgentEndEvent to all handlers."""
    self._scopes.claim_termination(event.run_id)
    await self._dispatch("on_agent_end", event)

on_agent_start async

on_agent_start(event: AgentStartEvent) -> None

Dispatch AgentStartEvent to all handlers, and arm the turn.

TA8.29. The termination latch is per TURN, not per run id for this manager's lifetime. SymfonicAgent.resume and resume_interrupt re-enter stream_events with the PAUSED run's id, and a deps-registered CallbackManager is long-lived, so without this the resumed turn would dispatch on_agent_start and then nothing at all -- no AgentEndEvent, and no close for the node and LLM scopes that turn opened.

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

async def on_agent_start(self, event: AgentStartEvent) -> None:
    """Dispatch AgentStartEvent to all handlers, and arm the turn.

    TA8.29. The termination latch is per TURN, not per run id for this
    manager's lifetime. ``SymfonicAgent.resume`` and ``resume_interrupt``
    re-enter ``stream_events`` with the PAUSED run's id, and a
    deps-registered ``CallbackManager`` is long-lived, so without this the
    resumed turn would dispatch ``on_agent_start`` and then nothing at all
    -- no ``AgentEndEvent``, and no close for the node and LLM scopes that
    turn opened.
    """
    self._scopes.begin_turn(event.run_id)
    await self._dispatch("on_agent_start", event)

on_before_tool_call async

on_before_tool_call(event: Any, state: dict[str, Any]) -> dict[str, Any] | None

Consult guard handlers before a tool executes; return a verdict.

The steering / guard seam (companion to the on_tool_call_dispatch rewriter). Handlers are consulted in registration order and the FIRST handler that returns a skip verdict wins — a guard veto short-circuits the chain. Handlers that return None / {"action": "proceed"} defer to the next handler.

Return value delivered to :class:InstrumentedToolNode:

• None -> no guard skipped the call; execute normally.
• {"action": "skip", "content": str, "is_error": bool} -> the tool node synthesizes a ToolMessage from content instead of executing. is_error defaults to True.

Malformed verdicts (missing/empty content on a skip) are dropped with a WARNING and treated as proceed so a buggy guard cannot silently swallow a call. A handler that raises is logged and skipped (the chain proceeds) — error isolation matches the other optional hooks.

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

async def on_before_tool_call(
    self,
    event: Any,
    state: dict[str, Any],
) -> dict[str, Any] | None:
    """Consult guard handlers before a tool executes; return a verdict.

    The steering / guard seam (companion to the ``on_tool_call_dispatch``
    rewriter). Handlers are consulted in registration order and the FIRST
    handler that returns a ``skip`` verdict wins — a guard veto
    short-circuits the chain. Handlers that return ``None`` /
    ``{"action": "proceed"}`` defer to the next handler.

    Return value delivered to :class:`InstrumentedToolNode`:

    * ``None`` -> no guard skipped the call; execute normally.
    * ``{"action": "skip", "content": str, "is_error": bool}`` -> the
      tool node synthesizes a ``ToolMessage`` from ``content`` instead of
      executing. ``is_error`` defaults to ``True``.

    Malformed verdicts (missing/empty ``content`` on a ``skip``) are
    dropped with a WARNING and treated as ``proceed`` so a buggy guard
    cannot silently swallow a call. A handler that raises is logged and
    skipped (the chain proceeds) — error isolation matches the other
    optional hooks.
    """
    if not self._handlers:
        return None
    for handler in self._handlers:
        method = getattr(handler, "on_before_tool_call", None)
        if method is None:
            continue
        try:
            result = await method(event, state)
        except Exception:
            logger.exception(
                "Callback handler %r raised in on_before_tool_call",
                handler,
            )
            continue
        if result is None:
            continue
        if not isinstance(result, dict):
            logger.warning(
                "Callback handler %r returned non-dict from "
                "on_before_tool_call (got %s); treating as proceed",
                handler,
                type(result).__name__,
            )
            continue
        action = result.get("action", "proceed")
        if action == "proceed":
            continue
        if action != "skip":
            logger.warning(
                "Callback handler %r returned unknown on_before_tool_call "
                "action %r; treating as proceed",
                handler,
                action,
            )
            continue
        content = result.get("content")
        if not isinstance(content, str) or not content:
            logger.warning(
                "Callback handler %r returned a 'skip' verdict without a "
                "non-empty string 'content'; treating as proceed",
                handler,
            )
            continue
        # First skip wins — a guard veto short-circuits the chain.
        return {
            "action": "skip",
            "content": content,
            "is_error": bool(result.get("is_error", True)),
        }
    return None

on_fabrication_detected async

on_fabrication_detected(event: FabricationDetectedEvent) -> None

Dispatch FabricationDetectedEvent to handlers that implement it.

on_fabrication_detected is an OPTIONAL hook (the dispatch was implicit in v7.0.6 via raw getattr; v7.0.7 promotes it to a typed event with proper manager-level fan-out). Handlers that do not implement the method are silently skipped. Preserves the zero-cost invariant: callers gate construction of :class:~symfonic.core.contracts.callbacks.FabricationDetectedEvent on :meth:has_hook ("on_fabrication_detected") so no event objects are built when no handler listens.

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

async def on_fabrication_detected(
    self, event: FabricationDetectedEvent,
) -> None:
    """Dispatch FabricationDetectedEvent to handlers that implement it.

    ``on_fabrication_detected`` is an OPTIONAL hook (the dispatch
    was implicit in v7.0.6 via raw ``getattr``; v7.0.7 promotes it
    to a typed event with proper manager-level fan-out).  Handlers
    that do not implement the method are silently skipped.
    Preserves the zero-cost invariant: callers gate construction
    of :class:`~symfonic.core.contracts.callbacks.FabricationDetectedEvent`
    on :meth:`has_hook` ``("on_fabrication_detected")`` so no
    event objects are built when no handler listens.
    """
    await self._dispatch_optional("on_fabrication_detected", event)

on_llm_end async

on_llm_end(event: LLMEndEvent) -> None

Dispatch LLMEndEvent to all handlers, and close its scope.

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

async def on_llm_end(self, event: LLMEndEvent) -> None:
    """Dispatch LLMEndEvent to all handlers, and close its scope."""
    self._scopes.llm_closed(event.run_id, event.node_name)
    await self._dispatch("on_llm_end", event)

on_llm_pre_call async

on_llm_pre_call(event: LLMPreCallEvent) -> None

Dispatch LLMPreCallEvent to handlers that implement it.

on_llm_pre_call is an OPTIONAL hook (added in v6.1.5); not every handler implements it, so we fan out only to handlers that define the attribute. Preserves the zero-cost invariant: if no handler implements the hook the loop is a no-op.

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

async def on_llm_pre_call(self, event: LLMPreCallEvent) -> None:
    """Dispatch LLMPreCallEvent to handlers that implement it.

    ``on_llm_pre_call`` is an OPTIONAL hook (added in v6.1.5); not
    every handler implements it, so we fan out only to handlers that
    define the attribute.  Preserves the zero-cost invariant: if no
    handler implements the hook the loop is a no-op.
    """
    await self._dispatch_optional("on_llm_pre_call", event)

on_llm_start async

on_llm_start(event: LLMStartEvent) -> None

Dispatch LLMStartEvent to all handlers, and open its scope.

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

async def on_llm_start(self, event: LLMStartEvent) -> None:
    """Dispatch LLMStartEvent to all handlers, and open its scope."""
    self._scopes.llm_opened(event)
    await self._dispatch("on_llm_start", event)

on_node_end async

on_node_end(event: NodeEndEvent) -> None

Dispatch NodeEndEvent to all handlers, and close its scope.

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

async def on_node_end(self, event: NodeEndEvent) -> None:
    """Dispatch NodeEndEvent to all handlers, and close its scope."""
    self._scopes.node_closed(event.run_id, event.node_name)
    await self._dispatch("on_node_end", event)

on_node_error async

on_node_error(event: NodeErrorEvent) -> None

Dispatch NodeErrorEvent to all handlers, and close its scope.

An error closes the pair exactly as an end does: the node is no longer running, so a later teardown must not close it a second time.

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

async def on_node_error(self, event: NodeErrorEvent) -> None:
    """Dispatch NodeErrorEvent to all handlers, and close its scope.

    An error closes the pair exactly as an end does: the node is no longer
    running, so a later teardown must not close it a second time.
    """
    self._scopes.node_closed(event.run_id, event.node_name)
    await self._dispatch("on_node_error", event)

on_node_start async

on_node_start(event: NodeStartEvent) -> None

Dispatch NodeStartEvent to all handlers, and open its scope.

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

async def on_node_start(self, event: NodeStartEvent) -> None:
    """Dispatch NodeStartEvent to all handlers, and open its scope."""
    self._scopes.node_opened(event)
    await self._dispatch("on_node_start", event)

on_procedure_selection async

on_procedure_selection(event: ProcedureSelectionEvent) -> None

Dispatch ProcedureSelectionEvent to handlers that implement it (v7.20.0 T-7.20.0.9).

on_procedure_selection is an OPTIONAL hook; fans out only to handlers that define the attribute. Preserves the zero- cost invariant: the engine's force-resolver gates event construction on has_hook("on_procedure_selection") so no event objects are built when no handler listens.

Adopters subscribe to capture per-procedure billing attribution, A/B-test bucket assignment, audit trail entries, and production routing telemetry. See :class:~symfonic.core.contracts.callbacks.ProcedureSelectionEvent.

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

async def on_procedure_selection(
    self, event: ProcedureSelectionEvent,
) -> None:
    """Dispatch ProcedureSelectionEvent to handlers that implement
    it (v7.20.0 T-7.20.0.9).

    ``on_procedure_selection`` is an OPTIONAL hook; fans out only
    to handlers that define the attribute.  Preserves the zero-
    cost invariant: the engine's force-resolver gates event
    construction on ``has_hook("on_procedure_selection")`` so no
    event objects are built when no handler listens.

    Adopters subscribe to capture per-procedure billing
    attribution, A/B-test bucket assignment, audit trail entries,
    and production routing telemetry.  See
    :class:`~symfonic.core.contracts.callbacks.ProcedureSelectionEvent`.
    """
    await self._dispatch_optional("on_procedure_selection", event)

on_react_loop_end async

on_react_loop_end(event: ReactLoopEndEvent) -> None

Dispatch ReactLoopEndEvent to handlers that implement it (v7.17.0).

on_react_loop_end is an OPTIONAL hook fired by the React node when the loop terminates with an empty final_response AFTER the v7.17.0 content-block salvage attempt. Handlers that do not implement it are silently skipped. Preserves the zero-cost invariant: the react node gates event construction on :meth:has_hook ("on_react_loop_end") so no event objects are built when no handler listens.

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

async def on_react_loop_end(self, event: ReactLoopEndEvent) -> None:
    """Dispatch ReactLoopEndEvent to handlers that implement it (v7.17.0).

    ``on_react_loop_end`` is an OPTIONAL hook fired by the React node
    when the loop terminates with an empty ``final_response`` AFTER
    the v7.17.0 content-block salvage attempt.  Handlers that do not
    implement it are silently skipped.  Preserves the zero-cost
    invariant: the react node gates event construction on
    :meth:`has_hook` ``("on_react_loop_end")`` so no event objects
    are built when no handler listens.
    """
    await self._dispatch_optional("on_react_loop_end", event)

on_response_render async

on_response_render(content: str, state: dict[str, Any], metadata: ResponseRenderEvent) -> str

Dispatch ResponseRenderEvent to handlers that implement it.

Unique among optional hooks: on_response_render RETURNS the rewritten string. The runtime substitutes the returned value for the original final_response BEFORE on_agent_end dispatch.

Chaining semantics for multiple handlers (registration order):

• First handler receives content=original_content.
• Each subsequent handler receives the previous handler's return value.
• The final return value is delivered to the runtime.

Error isolation: if a handler raises, the exception is logged and that handler's contribution is SKIPPED (the chain proceeds with the previous content). This matches the error-isolation contract of :meth:_dispatch_optional -- a broken rewriter must not stall the runtime.

Handlers that do not implement the method are silently skipped. Preserves the zero-cost invariant: emission sites gate construction of :class:~symfonic.core.contracts.callbacks. ResponseRenderEvent on :meth:has_hook ("on_response_render") so no event objects are built when no handler listens.

v7.18.0 (Option B; adopter Slack-formatting ask).

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

async def on_response_render(
    self,
    content: str,
    state: dict[str, Any],
    metadata: ResponseRenderEvent,
) -> str:
    """Dispatch ResponseRenderEvent to handlers that implement it.

    Unique among optional hooks: ``on_response_render`` RETURNS
    the rewritten string.  The runtime substitutes the returned
    value for the original ``final_response`` BEFORE ``on_agent_end``
    dispatch.

    Chaining semantics for multiple handlers (registration order):

    * First handler receives ``content=original_content``.
    * Each subsequent handler receives the previous handler's
      return value.
    * The final return value is delivered to the runtime.

    Error isolation: if a handler raises, the exception is logged
    and that handler's contribution is SKIPPED (the chain proceeds
    with the previous content).  This matches the error-isolation
    contract of :meth:`_dispatch_optional` -- a broken rewriter
    must not stall the runtime.

    Handlers that do not implement the method are silently skipped.
    Preserves the zero-cost invariant: emission sites gate
    construction of :class:`~symfonic.core.contracts.callbacks.
    ResponseRenderEvent` on :meth:`has_hook`
    ``("on_response_render")`` so no event objects are built when
    no handler listens.

    v7.18.0 (Option B; adopter Slack-formatting ask).
    """
    if not self._handlers:
        return content
    current = content
    for handler in self._handlers:
        method = getattr(handler, "on_response_render", None)
        if method is None:
            continue
        try:
            result = await method(current, state, metadata)
        except Exception:
            logger.exception(
                "Callback handler %r raised in on_response_render",
                handler,
            )
            continue
        # Defensive type contract: handlers MUST return a string.
        # A non-string return drops the rewriter's contribution
        # silently and proceeds with the previous content -- the
        # runtime never delivers a non-string final_response to
        # adopters.
        if isinstance(result, str):
            current = result
        else:
            logger.warning(
                "Callback handler %r returned non-string from "
                "on_response_render (got %s); skipping rewrite",
                handler,
                type(result).__name__,
            )
    return current

on_token_composition async

on_token_composition(event: TokenCompositionEvent) -> None

Dispatch TokenCompositionEvent to all handlers.

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

async def on_token_composition(self, event: TokenCompositionEvent) -> None:
    """Dispatch TokenCompositionEvent to all handlers."""
    await self._dispatch("on_token_composition", event)

on_tool_call_dispatch async

on_tool_call_dispatch(event: ToolCallDispatchEvent, state: dict[str, Any]) -> dict[str, Any] | None

Dispatch ToolCallDispatchEvent to handlers that implement it.

Unique among optional hooks (alongside on_response_render): on_tool_call_dispatch RETURNS the rewritten {"tool_name", "args"} dict. The React node mutates ai_message.tool_calls in place before the AIMessage returns to LangGraph for dispatch.

Return-value semantics:

• None -> pass-through (no rewrite contribution).
• dict[str, Any] with keys {"tool_name": str, "args": dict} -> rewrite. The dict MUST NOT contain call_id -- the engine owns it. tool_name MUST be a string; args MUST be a dict. Malformed returns are dropped with a structured WARNING (see the type-validation block below).

Chaining semantics for multiple handlers (registration order):

• First handler receives the original event (original tool_name / args).
• Each handler that returns a dict re-emits a SHALLOW-COPIED event with the rewritten tool_name / args so the next handler sees the chain's current state.
• Handlers that return None are pass-throughs (no contribution; the next handler sees the previous state).
• The final return value is delivered to the React node.

Error isolation: a handler that raises is logged and SKIPPED (the chain proceeds with the previous state). Matches the error-isolation contract of _dispatch_optional -- a broken rewriter must not stall the React loop.

Validation gate: the React node verifies the final tool_name against event.available_tools and DROPS rewrites that target unregistered tools. This method does NOT enforce the gate -- it lives at the call site so the dropped-rewrite WARNING carries the React node context.

Handlers that do not implement the method are silently skipped. Preserves the zero-cost invariant: emission sites gate construction of :class:~symfonic.core.contracts.callbacks. ToolCallDispatchEvent on :meth:has_hook ("on_tool_call_dispatch") so no event objects are built when no handler listens.

v7.19.0 (an adopter CSV-export ask -> slack_upload_file + a per-user memory title shape).

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

async def on_tool_call_dispatch(
    self,
    event: ToolCallDispatchEvent,
    state: dict[str, Any],
) -> dict[str, Any] | None:
    """Dispatch ToolCallDispatchEvent to handlers that implement it.

    Unique among optional hooks (alongside ``on_response_render``):
    ``on_tool_call_dispatch`` RETURNS the rewritten ``{"tool_name",
    "args"}`` dict.  The React node mutates ``ai_message.tool_calls``
    in place before the AIMessage returns to LangGraph for dispatch.

    Return-value semantics:

    * ``None`` -> pass-through (no rewrite contribution).
    * ``dict[str, Any]`` with keys ``{"tool_name": str, "args": dict}``
      -> rewrite.  The dict MUST NOT contain ``call_id`` -- the
      engine owns it.  ``tool_name`` MUST be a string; ``args``
      MUST be a dict.  Malformed returns are dropped with a
      structured WARNING (see the type-validation block below).

    Chaining semantics for multiple handlers (registration order):

    * First handler receives the original event (original
      ``tool_name`` / ``args``).
    * Each handler that returns a dict re-emits a SHALLOW-COPIED
      event with the rewritten ``tool_name`` / ``args`` so the
      next handler sees the chain's current state.
    * Handlers that return ``None`` are pass-throughs (no
      contribution; the next handler sees the previous state).
    * The final return value is delivered to the React node.

    Error isolation: a handler that raises is logged and SKIPPED
    (the chain proceeds with the previous state).  Matches the
    error-isolation contract of ``_dispatch_optional`` -- a broken
    rewriter must not stall the React loop.

    Validation gate: the React node verifies the final
    ``tool_name`` against ``event.available_tools`` and DROPS
    rewrites that target unregistered tools.  This method does
    NOT enforce the gate -- it lives at the call site so the
    dropped-rewrite WARNING carries the React node context.

    Handlers that do not implement the method are silently
    skipped.  Preserves the zero-cost invariant: emission sites
    gate construction of :class:`~symfonic.core.contracts.callbacks.
    ToolCallDispatchEvent` on :meth:`has_hook`
    ``("on_tool_call_dispatch")`` so no event objects are built
    when no handler listens.

    v7.19.0 (an adopter CSV-export ask -> slack_upload_file +
    a per-user memory title shape).
    """
    if not self._handlers:
        return None
    from dataclasses import replace as _dc_replace

    current_event = event
    current_rewrite: dict[str, Any] | None = None
    for handler in self._handlers:
        method = getattr(handler, "on_tool_call_dispatch", None)
        if method is None:
            continue
        try:
            result = await method(current_event, state)
        except Exception:
            logger.exception(
                "Callback handler %r raised in on_tool_call_dispatch",
                handler,
            )
            continue
        if result is None:
            # Pass-through -- no rewrite contribution.  The next
            # handler sees the previous state (which may itself
            # be the original event if no prior handler rewrote).
            continue
        # Defensive type validation -- a malformed return drops
        # the contribution silently so the chain can proceed.
        if not isinstance(result, dict):
            logger.warning(
                "Callback handler %r returned non-dict from "
                "on_tool_call_dispatch (got %s); skipping rewrite",
                handler,
                type(result).__name__,
            )
            continue
        _new_name = result.get("tool_name")
        _new_args = result.get("args")
        if not isinstance(_new_name, str) or not isinstance(
            _new_args, dict
        ):
            logger.warning(
                "Callback handler %r returned malformed "
                "on_tool_call_dispatch rewrite (tool_name=%r, "
                "args=%r); skipping",
                handler,
                type(_new_name).__name__,
                type(_new_args).__name__,
            )
            continue
        if "call_id" in result:
            logger.warning(
                "Callback handler %r included 'call_id' in "
                "on_tool_call_dispatch rewrite -- the engine owns "
                "call_id; dropping the key (rewrite proceeds with "
                "the original call_id)",
                handler,
            )
        current_rewrite = {
            "tool_name": _new_name,
            "args": _new_args,
        }
        # Re-emit the event with the rewritten fields so chained
        # handlers see the in-progress state, not the stale
        # original.  ``replace`` clones the frozen dataclass.
        current_event = _dc_replace(
            current_event,
            tool_name=_new_name,
            args=_new_args,
        )
    return current_rewrite

on_tool_result async

on_tool_result(event: ToolResultCallbackEvent) -> None

Dispatch ToolResultCallbackEvent to handlers that implement it (v8.6.4).

on_tool_result is an OPTIONAL hook fired by the tool node (InstrumentedToolNode) AFTER a tool returns, carrying the produced result, call_id, and a deferred flag. Handlers that do not implement it are silently skipped via :meth:_dispatch_optional. Preserves the zero-cost invariant: the emit site gates event construction on :meth:has_hook ("on_tool_result") so no event objects are built when no handler listens.

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

async def on_tool_result(self, event: ToolResultCallbackEvent) -> None:
    """Dispatch ToolResultCallbackEvent to handlers that implement it (v8.6.4).

    ``on_tool_result`` is an OPTIONAL hook fired by the tool node
    (``InstrumentedToolNode``) AFTER a tool returns, carrying the
    produced result, call_id, and a ``deferred`` flag.  Handlers that
    do not implement it are silently skipped via
    :meth:`_dispatch_optional`.  Preserves the zero-cost invariant:
    the emit site gates event construction on
    :meth:`has_hook` ``("on_tool_result")`` so no event objects are
    built when no handler listens.
    """
    await self._dispatch_optional("on_tool_result", event)

on_tool_routing_decision async

on_tool_routing_decision(event: ToolRoutingDecisionEvent) -> None

Dispatch ToolRoutingDecisionEvent to handlers that implement it.

on_tool_routing_decision is an OPTIONAL hook (added in v7.0.1); fans out only to handlers that define the attribute. Preserves the zero-cost invariant: when tool_routing_mode is off the engine never constructs the event OR calls this method.

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

async def on_tool_routing_decision(
    self, event: ToolRoutingDecisionEvent,
) -> None:
    """Dispatch ToolRoutingDecisionEvent to handlers that implement it.

    ``on_tool_routing_decision`` is an OPTIONAL hook (added in
    v7.0.1); fans out only to handlers that define the attribute.
    Preserves the zero-cost invariant: when ``tool_routing_mode``
    is ``off`` the engine never constructs the event OR calls
    this method.
    """
    await self._dispatch_optional("on_tool_routing_decision", event)

open_scope_count

open_scope_count(run_id: str) -> int

How many lifecycle pairs run_id still holds open (diagnostics).

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

def open_scope_count(self, run_id: str) -> int:
    """How many lifecycle pairs ``run_id`` still holds open (diagnostics)."""
    return self._scopes.open_scope_count(run_id)

NoOpCallbackHandler

Zero-overhead default callback handler. All methods are no-ops.

Satisfies CallbackHandler protocol via structural typing.