symfonic.core.contracts

contracts

Phase 3 contracts — shared types with zero internal dependencies.

ActivationEvent

Bases: BaseModel

A single memory node activated during spreading activation.

AgentEndEvent dataclass

AgentEndEvent(run_id: str, final_response: str | None, duration_ms: float, node_count: int | None, outcome: TerminationOutcome = 'completed')

Emitted when agent execution ends, however it ended.

In the streaming path, final_response and node_count are None because stream() does not accumulate final state.

outcome (TA8.29) is the discriminator that makes an abandoned or cancelled turn distinguishable from a completed one at this seam. It is additive with a default: every existing constructor keeps working and keeps meaning "completed", which is what makes closing the incomplete-turn defect a fix rather than a breaking change. A required field would have broken every adopter who builds this event today.

An observer that only ever saw the normal terminal cannot tell a run that answered from a run that was walked away from -- and the two have opposite meanings for latency, cost and error budgets.

AgentStartEvent dataclass

AgentStartEvent(query: str, run_id: str)

Emitted when agent execution begins.

AskUserAnsweredEvent

Bases: BaseModel

Fires when a resume call succeeds and the graph is unpaused.

AskUserExpiredEvent

Bases: BaseModel

Would fire when a janitor sweeps an expired checkpoint.

.. warning:: Nothing constructs this today. There is no janitor. A handler written for it will never be called. It is kept declared rather than deleted because it is already exported, and because the shape is right if a sweeper is ever built.

The sweeper was deliberately NOT built (decision 2026-08-07). The problem it was meant to solve -- a user who abandons an ask_user question poisoning the thread forever -- is instead fixed where it costs nothing and needs nothing to fire:

• _close_unanswered_tool_calls (core/nodes/react.py) pairs the orphaned call on the wire, leaving the checkpoint untouched;
• ask_user_pending_is_current (core/nodes/elicitation.py) makes every router ignore a marker from an earlier run.

Both act on the next turn regardless of whether anything swept, and critically both keep the checkpoint lossless -- so a user who answers hours later can still resume. A janitor cannot: it writes a NEW checkpoint, which voids the checkpoint_id the pause token pins, destroying exactly that late answer. That trade-off is why sweeping is opt-in work for an operator who wants it, not framework behaviour.

AskUserQuestionEvent

Bases: BaseModel

Emitted when the LLM calls the built-in ask_user tool.

The graph is paused at this point and will only advance when a matching POST to /api/v1/resume/{pause_token} arrives.

BuildResult dataclass

BuildResult(content: str, status: Literal['success', 'partial', 'failed'], warnings: tuple[BuildWarning, ...] = (), degraded_sections: tuple[str, ...] = ())

Result of a prompt build operation.

BuildWarning dataclass

BuildWarning(section: str, type: Literal['degraded', 'critical_error'], message: str)

Warning emitted during prompt build.

CapabilityError

CapabilityError(capability: str, cause: str, code: str, metadata: dict[str, Any] | None = None)

Bases: Exception

Error from an external capability (fetch, search, etc.).

Not a frozen dataclass — exceptions must support traceback assignment. Fields are set as regular attributes for compatibility.

Source code in src/symfonic/core/contracts/errors.py

def __init__(
    self,
    capability: str,
    cause: str,
    code: str,
    metadata: dict[str, Any] | None = None,
) -> None:
    self.capability = capability
    self.cause = cause
    self.code = code
    self.metadata = metadata or {}
    super().__init__(str(self))

ConsolidationScheduledEvent

Bases: BaseModel

Emitted after consolidation is scheduled (background task, not yet complete).

ExtensionEvent

Bases: BaseModel

Vendor-specific or custom event. Never dropped, never raises.

FabricationDetectedEvent dataclass

FabricationDetectedEvent(run_id: str, mode: Literal['off', 'observe', 'revise', 'refuse'], action_taken: Literal['none', 'observed', 'revise_forced', 'refuse_replaced'], findings: list[dict[str, Any]], refuse_threshold: float, max_finding_confidence: float, original_draft_preview: str, replacement_preview: str | None, preview_truncated: bool)

Emitted when the fabrication check finds suspect content (v7.0.7).

Fires in observe / revise / refuse modes whenever :func:~symfonic.agent.middleware.fabrication.detect_fabrication returns at least one finding. The action_taken field discriminates whether the detector merely observed, requested a revision, or replaced the draft with the safe envelope -- so a single typed event covers the full mode matrix.

The companion ExtensionEvent(type='fabrication_detected'| 'fabrication_blocked') payload that v7.0.6 emitted is preserved one release for back-compat: handlers that expose on_fabrication_detected still receive it in v7.0.7 alongside this typed event. The untyped path is deprecated and will be removed in v7.1.

Privacy. original_draft_preview and replacement_preview contain raw LLM output text (truncated at 240 chars; the preview_truncated flag is True when the full draft was longer). PII redaction is the handler's responsibility -- the same stance documented for :class:LLMPreCallEvent (v6.1.7).

Use this event to:

• Count detections per tenant / per mode for SRE dashboards.
• Tune fabrication_refuse_min_confidence from telemetry by bucketing max_finding_confidence against action_taken.
• Feed fabrication rates per mode and action_taken into production observability stacks.

Attributes:

Name	Type	Description
run_id	str	The agent run identifier this turn belongs to.
mode	Literal['off', 'observe', 'revise', 'refuse']	The effective fabrication-check mode that produced the finding ("off" only ever appears here when a future caller forces emission; the standard wiring never fires the event in "off").
action_taken	Literal['none', 'observed', 'revise_forced', 'refuse_replaced']	What the wiring did with the findings. "none" is reserved for callers that bypass the dispatch (currently unused). "observed" -- telemetry only, draft untouched. "revise_forced" -- forced metacog revise verdict. "refuse_replaced" -- draft replaced with the safe envelope.
findings	list[dict[str, Any]]	List of FabricationFinding.to_event_payload() dicts (JSON-serialisable shape). Same payload shape used by the deprecated ExtensionEvent path.
refuse_threshold	float	The fabrication_refuse_min_confidence value in effect for this turn.
max_finding_confidence	float	max(f.confidence for f in findings) -- the strongest signal in the finding list.
original_draft_preview	str	First 240 characters of the draft response that produced the findings. Empty string when the draft was empty (the wiring short-circuits before that, so this is informational only).
replacement_preview	str | None	First 240 characters of the replacement text used in refuse mode. None for any other action_taken value.
preview_truncated	bool	True when len(original_draft) > 240.

GraphEdgeCreatedEvent

Bases: BaseModel

Emitted when a new edge is written during consolidation.

GraphNodeCreatedEvent

Bases: BaseModel

Emitted when a new node is written to the graph during consolidation.

InterruptEvent

Bases: BaseModel

Generic pause event emitted when a registered interrupt fires.

Mirrors :class:AskUserQuestionEvent for non-ask_user interrupts.

Fields

name: Registered interrupt name (must match a prior :meth:SymfonicAgent.register_interrupt call). pause_token: Opaque JWT carrying the same security envelope as ask_user (HMAC-signed, scope-bound, request-hash-bound, expiry-bounded, single-use via the DB-backed interrupt_consumed_tokens table). payload: JSON-serialized payload matching the registered payload_schema. expires_at: Token expiry (UTC). Resume after this raises. interrupt_id: Client-side correlation id, format i-<12hex>.

InterruptExpiredEvent

Bases: BaseModel

Would be emitted when a janitor sweeps an expired interrupt token.

.. warning:: Nothing constructs this today. There is no janitor. A handler written for it will never be called.

Analogous to :class:~symfonic.core.contracts.types.AskUserExpiredEvent, including the reasoning for why no sweeper was built -- see that class's docstring for the decision and what replaced it.

InterruptRegistration dataclass

InterruptRegistration(name: str, payload_schema: type[BaseModel], response_schema: type[BaseModel], validate_response: Callable[[BaseModel, BaseModel], None] | None = None, cross_scope_allowed: bool = False, built_in: bool = False, metadata: dict[str, Any] = dict())

Boot-time registration record for a named interrupt point.

Held in a per-agent registry (SymfonicAgent._registered_interrupts) and consulted by the engine's mint path AND the resume router so the same metadata controls both ends of the pause/resume cycle.

Attributes:

Name	Type	Description
name	str	Stable identifier (e.g. "approval_required"). Must be a valid Python identifier-like string -- the engine echoes it into stream events and the resume URL is name-aware.
payload_schema	type[BaseModel]	Pydantic BaseModel subclass for the agent-side payload (i.e. what the LLM/graph hands to the user).
response_schema	type[BaseModel]	Pydantic BaseModel subclass for the user-side response (i.e. what the user POSTs to /resume).
validate_response	Callable[[BaseModel, BaseModel], None] | None	Optional cross-validator that receives (response, payload) and raises on mismatch. Used by ask_user to verify selections match the question's allowed labels; custom interrupts can skip this if the response schema's own validators are sufficient.
cross_scope_allowed	bool	When True, the resume endpoint accepts a token issued under one scope to be redeemed under another. Logs an audit entry on each cross-scope redemption. When False (default) same-scope binding is enforced.
built_in	bool	True for ask_user; False for caller-registered interrupts. The engine uses this to skip the experimental_interrupt flag check for built-ins.

InterruptResolvedEvent

Bases: BaseModel

Emitted when a generic interrupt is resumed successfully.

Companion to InterruptEvent; analogous to AskUserAnsweredEvent.

LLMEndEvent dataclass

LLMEndEvent(model: str, output: str, usage: dict, run_id: str = '', node_name: str = '', pricing_unknown: bool = False, duration_ms: float = 0.0, started_at_utc: str = '', iteration_index: int = 0, turn_index: int = 0, outcome: TerminationOutcome = 'completed')

Fired when the LLM returns a response.

node_name (v7.4.3) discriminates between LLM call sites so consumers can attribute usage / cost / spans back to the originating subsystem (e.g. "react", "metacognition_critic", "context_window_summariser", "insight_extractor"). The default empty string preserves byte-identical payloads for pre-v7.4.3 handlers that do not inspect the field.

pricing_unknown (v7.4.5) is a programmatic discriminator that fires True when the model name had no entry in the pricing registry at typed-usage materialisation time -- letting downstream consumers tell "known model, $0 turn" apart from "unknown model, cost defaulted to $0". Default False preserves byte-identical payloads for pre-7.4.5 handlers; emitters that build the event from a raw usage dict can leave the flag alone and rely on :attr:typed_usage to surface the discriminator on demand.

duration_ms / started_at_utc (v7.15.0) carry wall-clock latency and the UTC ISO-8601 timestamp of the LLM invocation. Both default to zero / empty so pre-v7.15.0 emission paths that did not capture timing preserve byte-identical payloads. Production emitters use the :func:symfonic.core.callbacks.emit.llm_timing context manager which populates them exception-safely.

iteration_index (v7.15.0) is the 0-indexed React loop iteration that produced this LLM call. 0 is the natural sentinel for non-React sites (each fires once per consolidation cycle) and for the first React iteration. React emission sites compute it as sum(1 for m in state["messages"] if isinstance(m, AIMessage)) BEFORE the current ainvoke.

typed_usage property

typed_usage: TokenUsage

Return a typed TokenUsage built from the raw usage dict.

The returned :class:TokenUsage carries its own pricing_unknown flag derived from the live pricing registry at access time.

LLMPreCallEvent dataclass

LLMPreCallEvent(model: str, messages: list, tools: list, run_id: str, invocation_params: dict = dict(), node_name: str = '')

Fired right before the LLM provider call, after system-message consolidation and tool binding. Represents the exact payload the provider is about to receive.

Use this to log the wire-accurate prompt (e.g. for gateway trace events, compliance capture, replay fixtures). For the engine-view (pre-consolidation) prompt, use :class:LLMStartEvent instead.

messages here is the post-consolidation list: a single leading :class:~langchain_core.messages.SystemMessage (when any system content is present) followed by the non-system conversation messages in order.

tools is a list of {"name": str, "description": str, "schema": dict} records describing the resolved tool definitions the provider will see. Empty when no tools are bound.

invocation_params carries provider-native invocation kwargs (temperature, max_tokens, tool_choice, etc.) when the underlying LangChain chat-model adapter exposes them via _get_invocation_params. Empty {} when the adapter does not surface them.

.. note:: Payload fields may contain sensitive content (PII, credentials embedded in prompts, customer data). PII redaction is the callback author's responsibility -- this event ships the wire-accurate payload verbatim by design.

LLMStartEvent dataclass

LLMStartEvent(model: str, messages: list, run_id: str = '', system_prompt: str = '', node_name: str = '', turn_index: int = 0)

Fired when the LLM is invoked with the full message list.

This is the engine-view event: messages is the conversation history exactly as the react node received it, BEFORE system-message consolidation and BEFORE tool binding. Use this to debug the engine-side prompt assembly (memory hydration, context window composition, etc.).

For the wire-accurate payload the provider is about to receive, use :class:LLMPreCallEvent instead.

MemoryExtractedEvent

Bases: BaseModel

Emitted when an extraction block is parsed from the stream.

MessageEndEvent

Bases: BaseModel

End of message.

MessageStartEvent

Bases: BaseModel

Start of a new message.

NodeEndEvent dataclass

NodeEndEvent(node_name: str, run_id: str, duration_ms: float)

Emitted after a graph node completes successfully.

NodeErrorEvent dataclass

NodeErrorEvent(node_name: str, run_id: str, error: str, error_type: str)

Emitted when a graph node raises an exception.

NodeStartEvent dataclass

NodeStartEvent(node_name: str, run_id: str)

Emitted before a graph node executes.

ResponseCompleteEvent

Bases: BaseModel

Final clean response text — equivalent of AgentResponse.final_response.

SearchResultEvent

Bases: BaseModel

Search result from external capability.

TextDeltaEvent

Bases: BaseModel

Incremental text chunk from LLM.

ThinkingDeltaEvent

Bases: BaseModel

Incremental thinking/reasoning chunk.

ToolCallDeltaEvent

Bases: BaseModel

Incremental tool call arguments.

ToolCallPolicy dataclass

ToolCallPolicy(name: str, match: Callable[[DispatchContext], bool], when_tool: str | frozenset[str] | None = None, redirect_to: str | None = None, model: ModelConfig | None = None, args_transform: Callable[[dict[str, Any], DispatchContext], dict[str, Any]] | None = None, guard: Callable[[DispatchContext], bool] | None = None)

Declarative {intent -> model + tool-redirect} rule (v8.1.0).

Fields:

• name -- required, non-empty. For logging / inspection / test assertions.
• match -- the intent gate, Callable[[DispatchContext], bool]. Runs at BOTH seams against an equivalent DispatchContext (the adopter classifies intent off messages[-1].content / scope / resolved_tool_names). Core ships NO intent classifier -- intent is adopter business logic.
• when_tool -- POST-model emitted-tool filter. str | frozenset[str] | None. None matches any emitted tool. NO meaning at the PRE-model seam (no tool exists yet there). A SEPARATE field from match so match stays a pure intent predicate reusable at both seams.
• redirect_to -- POST-model target tool name. Setting it makes the policy a redirect policy. Subject to the palette gate (an unrouted target is dropped with a WARNING; the original dispatches).
• model -- PRE-model per-intent model override (ModelConfig). Applied ONLY at the PRE-model seam.
• args_transform -- optional POST-model arg rewriter, Callable[[args, ctx], args]. None passes args through.
• guard -- optional deterministic precondition, Callable[[DispatchContext], bool]. Distinct from match: match = "is this the right intent?"; guard = "is the precondition satisfiable?" (e.g. the stored table exists for THIS scope). guard gates the REDIRECT only -- it does NOT gate model-selection (a missing data source must not change which model reasons). The policy redirects only when match(ctx) AND (guard is None or guard(ctx)).

matches_tool

matches_tool(tool_name: str) -> bool

True when tool_name satisfies when_tool (POST-model).

Source code in src/symfonic/core/contracts/tool_policy.py

def matches_tool(self, tool_name: str) -> bool:
    """True when ``tool_name`` satisfies ``when_tool`` (POST-model)."""
    if self.when_tool is None:
        return True
    if isinstance(self.when_tool, str):
        return tool_name == self.when_tool
    return tool_name in self.when_tool

ToolCallStartEvent

Bases: BaseModel

Tool call initiated by LLM.

A single logical tool call can legitimately produce two of these events at different lifecycle points: "announced" when the model's own streamed generation first reveals the tool call, and "executing" when the runtime actually dispatches it (e.g. LangGraph's on_tool_start). They carry different tool_call_id spaces (model tool-call id vs. run id) and are not otherwise correlated -- consumers that want exactly one "announce this to the user" signal per real tool call should filter to a single phase rather than assume one event.

ToolProgressEvent

Bases: BaseModel

Intermediate progress emitted by an async-generator tool.

Yielded between ToolCallStartEvent and the final ToolResultEvent. Carries an opaque payload; the tool author chooses the shape. sequence is monotonically increasing (0-indexed) per tool_call_id so a consumer can detect drops.

The engine emits this when a registered tool is an inspect.isasyncgenfunction: each non-final yield becomes one ToolProgressEvent; the last yield becomes the existing ToolResultEvent. Regular async def tools see zero behaviour change.

ToolResultEvent

Bases: BaseModel

Result returned from tool execution.

UsageEvent

Bases: BaseModel

Token usage report.