symfonic.agent.backend.plan

plan

Compiling one Agent call into an InvocationPlan.

This is the facade's side of the compile seam: it constructs the live port objects (a chat model, a tool executor, a response adapter) and hands them to the one compiler in symfonic.kernel. Constructing bindings here rather than inside the compiler is what keeps IPL-8 true — the compiler stays a pure function of values, and everything that has to build something happens on the composition side of the line.

Every ConfigurationError the facade raises is raised here, before the first inference call (FERR-3). Discovering that a provider cannot do structured output after spending tokens is the failure mode RES-9 exists to prevent.

AgentPlanFactory dataclass

AgentPlanFactory(model_provider: Any, instructions: str | None, tools: NormalizedTools, max_model_rounds: int = MAX_TOOL_ITERATIONS, model: ModelConfig | None = None, role_models: Mapping[str, ModelConfig] = dict(), stages: tuple[Any, ...] = (), stage_handlers: Mapping[str, Any] = dict(), tool_preconditions: tuple[Any, ...] = (), authorized_effects: frozenset[str] = frozenset({'model_call'}), capability_names: tuple[str, ...] = ())

Turns the facade's four construction inputs into a compiled plan.

compile

compile(output_type: type[BaseModel] | None, *, sink_factory: Callable[[ModelResolution], Any] | None = None) -> InvocationPlan

Validate eagerly, bind the ports, and compile exactly one plan.

sink_factory is the seam ServiceBindings.event_sink reserved. It is called at most once, here, with the ModelResolution this plan was compiled with, and whatever it returns is bound as the sink; None — the simple facade's answer, which has no callback input — leaves the binding unset, and InvocationRunner then constructs no CallbackEventAdapter at all.

A factory rather than a sink because the two things a sink needs are produced on opposite sides of this call: the caller owns the run's identity (its id, its session, its tenant) and this method owns the model that identity has to name. Handing the resolution out rather than making the caller re-derive it keeps one answer to "which model ran?" — the caller's copy cannot drift from the plan's, because it is the plan's.

Binding it here rather than on the caller's side is also what makes the sink's reference frozen at compile time along with every other port (IPL-4). equality_key excludes bindings, so a bound sink changes neither the config digest nor plan identity (IPL-7): two runs of the same agent still compile the same plan shape with different sinks.

Source code in src/symfonic/agent/backend/plan.py

def compile(
    self,
    output_type: type[BaseModel] | None,
    *,
    sink_factory: Callable[[ModelResolution], Any] | None = None,
) -> InvocationPlan:
    """Validate eagerly, bind the ports, and compile exactly one plan.

    ``sink_factory`` is the seam ``ServiceBindings.event_sink`` reserved.
    It is called at most once, here, with the ``ModelResolution`` this plan
    was compiled with, and whatever it returns is bound as the sink;
    ``None`` — the simple facade's answer, which has no callback input —
    leaves the binding unset, and ``InvocationRunner`` then constructs no
    ``CallbackEventAdapter`` at all.

    A *factory* rather than a sink because the two things a sink needs are
    produced on opposite sides of this call: the caller owns the run's
    identity (its id, its session, its tenant) and this method owns the
    model that identity has to name. Handing the resolution out rather than
    making the caller re-derive it keeps one answer to "which model ran?" —
    the caller's copy cannot drift from the plan's, because it *is* the
    plan's.

    Binding it here rather than on the caller's side is also what makes the
    sink's reference frozen at compile time along with every other port
    (IPL-4). ``equality_key`` excludes bindings, so a bound sink changes
    neither the config digest nor plan identity (IPL-7): two runs of the
    same agent still compile the same plan shape with different sinks.
    """
    validate_output_type(output_type)

    config = self._resolve_config()
    chat_model = self.model_provider.get_chat_model(config)
    response_model = chat_model
    if output_type is not None:
        # Some OpenAI-compatible reasoning models need a different request
        # mode for the terminal schema pass than for the conversational
        # turn.  Keep this an optional provider seam so existing adopter
        # providers continue to work unchanged; the ordinary model still
        # owns every tool/reasoning round.
        structured_model = getattr(
            self.model_provider, "get_structured_chat_model", None
        )
        if callable(structured_model):
            specialized = structured_model(config)
            if specialized is not None:
                response_model = specialized
        require_structured_output(response_model, output_type)

    bound = chat_model
    if self.tools:
        bound = chat_model.bind_tools(list(self.tools.tools))
    from symfonic.agent.backend.binding import palette_projector

    adapted = ModelAdapter(bound)
    projector = palette_projector(
        self.model_provider,
        config,
        chat_model,
        self.tools.tools if self.tools else (),
        adapted,
        ModelAdapter,
    )

    resolution = ModelResolution(
        # T3.1.1: the detected wire family, not the provider class name.
        # ``ModelResolution.provider_family`` is documented as a family and
        # is read as one downstream (cache dialect, content-block encoding,
        # diagnostics); stamping ``"OpenAIProvider"`` here made every one of
        # those reads miss without ever raising.
        provider_family=detect_provider_family(self.model_provider, config),
        model_name=getattr(config, "model_name", None),
        response_format=ResponseFormat(
            mode="structured" if output_type is not None else "text",
            schema_name=output_type.__name__ if output_type else None,
        ),
    )

    return compile_invocation_plan(
        CompileRequest(
            config_digest=self._digest(config, output_type),
            model=resolution,
            instructions=self.instructions,
            tools=tuple(
                ToolDescriptor(name=tool.name, description=tool.description or "")
                for tool in self.tools.tools
            ),
            stages=self.stages,
            capabilities=self.capability_names,
            bindings=ServiceBindings(
                conversation=ConversationAdapter(self.model_provider),
                model=adapted,
                palette=projector,
                tools=ToolAdapter(self.tools),
                response=ResponseAdapter(
                    response_model,
                    output_type,
                    repair_attempts=config.structured_output_repair_attempts,
                ),
                stage_handlers=dict(self.stage_handlers),
                tool_preconditions=tuple(self.tool_preconditions),
                # TA8.20 filled the seam this row reserved. The simple Agent
                # facade still has no callback input and still passes no
                # factory, so its plans bind nothing and pay nothing; the
                # cutover delegate passes
                # ``symfonic.agent.cutover.observability.RunObservability``,
                # which composes the run's ``ObservabilityBridge`` from the
                # resolution above.
                event_sink=(
                    None if sink_factory is None else sink_factory(resolution)
                ),
            ),
            # Nothing a capability declared reaches this union, and there
            # is no field on this factory through which it could: what a
            # capability asked for is a *narrowing* of what the facade
            # already granted, checked at fold time. An earlier version
            # unioned in a ``capability_grants`` field, which let a
            # capability widen the plan by declaring a grant; the field is
            # gone rather than merely unused, so the bypass has no seam to
            # come back through.
            effect_grants=frozenset(self.authorized_effects)
            | ({"tool_call"} if self.tools else frozenset()),
            limits=PlanLimits(max_model_rounds=self.max_model_rounds),
            event_program=EventProgram(
                emitted=_EMITTED,
                adapters=tuple(ADAPTER_POLICIES.values()),
            ),
        )
    )