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,
)

Emitted when agent execution completes.

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

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

Fires when the janitor sweeps an expired checkpoint.

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

Emitted when the janitor sweeps an expired generic-interrupt token.

Analogous to AskUserExpiredEvent.

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,
)

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.

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.