Skip to content

symfonic.migration

migration

The executable contract for migrating legacy default behaviour.

compose_legacy_defaults is deliberately a factory, rather than a second agent class. It returns the public :class:symfonic.Agent after composing the behaviour the table says is reproducible. A migration starts with the table: defaults which need a deployment-owned dependency refuse at construction, and defaults with no published replacement name their owning follow-up issue.

agent_from_config answers a different question: which settings of an existing FrameworkConfig it carries onto the v11 Agent. It translates a small named set, inherits stock values, and refuses every other active setting.

LegacyDefaultRow dataclass

LegacyDefaultRow(behavior: str, legacy_default: str, activation: str, replacement: str, status: Literal['supported', 'gap', 'inert'], evidence: str, issue: int | None = None)

One claimed legacy default and the current migration result.

evidence is a test or source location that observes the stated result; issue is required for a gap so absence remains actionable. inert rows are not omitted: they say why a default-valued field never starts a behaviour on its own.

LegacyParityRefusal

LegacyParityRefusal(*, unresolved: Sequence[LegacyDefaultRow] = (), missing_dependencies: Sequence[str] = ())

Bases: ConfigurationError

The requested legacy claim cannot truthfully be composed yet.

Source code in src/symfonic/migration/defaults.py
def __init__(
    self,
    *,
    unresolved: Sequence[LegacyDefaultRow] = (),
    missing_dependencies: Sequence[str] = (),
) -> None:
    self.unresolved = tuple(unresolved)
    self.missing_dependencies = tuple(missing_dependencies)
    parts: list[str] = []
    if self.unresolved:
        parts.append(
            "unresolved legacy defaults: "
            + ", ".join(
                f"{row.behavior} (#{row.issue})" for row in self.unresolved
            )
        )
    if self.missing_dependencies:
        parts.append(
            "missing composition dependencies: " + ", ".join(self.missing_dependencies)
        )
    super().__init__(
        "cannot compose the requested legacy defaults; " + "; ".join(parts)
    )

UnsupportedFrameworkSettingError

UnsupportedFrameworkSettingError(refused: Sequence[tuple[str, str]], *, composed: bool)

Bases: ConfigurationError

Active FrameworkConfig settings outside this boundary's contract.

Raised at composition, before any Agent exists. settings lists every refused path at once, so a migration sees the whole set rather than one field per attempt.

Source code in src/symfonic/migration/framework_config_contract.py
def __init__(self, refused: Sequence[tuple[str, str]], *, composed: bool) -> None:
    self.settings = tuple(path for path, _ in refused)
    self.contract = FRAMEWORK_CONFIG_CONTRACT
    self.line = LEVER_RETIREMENT_LINE
    self.pin = LAST_PUBLISHED_HONOURING_LINE
    details = "; ".join(f"{path}: {why}" for path, why in refused)
    super().__init__(
        f"agent_from_config ({self.contract}) refuses active FrameworkConfig "
        f"settings this boundary does not support: {details}. Remove them, "
        "or compose the native capability on the Agent where one exists; "
        f"this composition does not read them. Settings retired on {self.line} "
        f"are honoured only by symfonic-core {self.pin}."
        + (
            " Composition switches are refused at False because the caller "
            "composed capabilities this boundary does not inspect."
            if composed and any(path in COMPOSITION_SWITCHES for path, _ in refused)
            else ""
        )
    )

agent_from_config

agent_from_config(config: FrameworkConfig, model_provider: Any, *, instructions: str | None = None, tools: Sequence[Any] = (), capabilities: Sequence[Any] = ()) -> Agent

Return an Agent honouring config's translated settings.

The caller must explicitly set auto_consolidate=False on the legacy config: native QUICK requires separately proved exact-case AgentHost composition. An omitted stock True refuses before Agent construction.

instructions, tools and capabilities are the caller's, passed to Agent unchanged: the config never contributes a system prompt, a tool or a memory port. Memory, sessions and history therefore work exactly as they do on any Agent -- through the capabilities composed here and the session_id/history arguments of each call.

agent.model and agent.recursion_limit at their stock values are not requests: the Agent keeps its provider-declared model and native round budget. agent.model is translated field by field: only the fields away from ModelConfig() are applied, over the provider's declared default.

Raises:

Type Description
TypeError

config is not a FrameworkConfig.

UnsupportedFrameworkSettingError

an active setting is outside the contract, or synthesize_on_empty_final is set beside a TerminalSynthesisCapability the caller composed. Raised before the Agent exists, so nothing reaches the provider, memory or a tool.

Source code in src/symfonic/migration/framework_config.py
def agent_from_config(
    config: FrameworkConfig,
    model_provider: Any,
    *,
    instructions: str | None = None,
    tools: Sequence[Any] = (),
    capabilities: Sequence[Any] = (),
) -> Agent:
    """Return an ``Agent`` honouring ``config``'s translated settings.

    The caller must explicitly set ``auto_consolidate=False`` on the legacy
    config: native QUICK requires separately proved exact-case AgentHost
    composition. An omitted stock ``True`` refuses before Agent construction.

    ``instructions``, ``tools`` and ``capabilities`` are the caller's, passed
    to ``Agent`` unchanged: the config never contributes a system prompt, a
    tool or a memory port. Memory, sessions and history therefore work exactly
    as they do on any ``Agent`` -- through the capabilities composed here and
    the ``session_id``/``history`` arguments of each call.

    ``agent.model`` and ``agent.recursion_limit`` at their stock values are not
    requests: the ``Agent`` keeps its provider-declared model and native round
    budget. ``agent.model`` is translated field by field: only the fields away
    from ``ModelConfig()`` are applied, over the provider's declared default.

    Raises:
        TypeError: ``config`` is not a ``FrameworkConfig``.
        UnsupportedFrameworkSettingError: an active setting is outside the
            contract, or ``synthesize_on_empty_final`` is set beside a
            ``TerminalSynthesisCapability`` the caller composed. Raised before
            the ``Agent`` exists, so nothing reaches the provider, memory or a
            tool.
    """
    if not isinstance(config, FrameworkConfig):
        raise TypeError(f"agent_from_config needs a FrameworkConfig, not {type(config).__name__}")
    if config.auto_consolidate is not False or "auto_consolidate" not in config.model_fields_set:
        # A caller-owned Sequence may do work while being inspected. Refuse
        # legacy consolidation before touching it.
        refused = unsupported_settings(config, composed=False)
        raise UnsupportedFrameworkSettingError(refused, composed=False)
    composed = tuple(capabilities)
    refused = unsupported_settings(config, composed=bool(composed))
    if config.synthesize_on_empty_final and any(
        isinstance(cap, TerminalSynthesisCapability) for cap in composed
    ):
        # The Agent would refuse the pair too, but unversioned: a host could
        # not tell it from a failed turn. Refused here it maps like any other.
        refused = (*refused, ("synthesize_on_empty_final", _DUPLICATE_SYNTHESIS))
    if refused:
        raise UnsupportedFrameworkSettingError(refused, composed=bool(composed))
    if config.synthesize_on_empty_final:
        composed = (*composed, TerminalSynthesisCapability())
    stock = FrameworkConfig().agent
    model, limit = config.agent.model, config.agent.recursion_limit
    return Agent(
        model_provider,
        instructions=instructions,
        model=_translated_model(model, model_provider),
        tools=tools,
        capabilities=composed,
        max_model_rounds=(
            None if equivalent(limit, stock.recursion_limit) else rounds_for_recursion_limit(limit)
        ),
    )

compose_legacy_defaults

compose_legacy_defaults(model_provider: Any, *, store: Any | None = None, scope: Any | None = None, instructions: str | None = None, tools: Sequence[Any] = (), behaviors: Sequence[str] | None = None, conversation: Any | None = None, extractor: Any | None = None, activation: Any | None = None, consolidation: Any | None = None, recent_turns: int = 2, max_model_rounds: int | None = None) -> Agent

Compose the reproducible legacy defaults into a public Agent.

With no behaviors argument this requests the whole legacy-default claim. It refuses all unresolved rows and every absent dependency in one error. Pass :data:SUPPORTED_DEFAULT_BEHAVIORS to compose only the verified subset. This is intentional: a successful call is exactly the explicit list it was given, never a silently smaller default set. lazy_tooling with no store or memory row cannot activate, so it composes nothing.

The required deployment dependencies are named before an Agent is constructed: an extractor for automatic consolidation, a conversation source for the two-turn working window, an association source wrapped in SpreadingActivation, and a consolidation coordinator for quick naps.

Source code in src/symfonic/migration/defaults.py
def compose_legacy_defaults(
    model_provider: Any,
    *,
    store: Any | None = None,
    scope: Any | None = None,
    instructions: str | None = None,
    tools: Sequence[Any] = (),
    behaviors: Sequence[str] | None = None,
    conversation: Any | None = None,
    extractor: Any | None = None,
    activation: Any | None = None,
    consolidation: Any | None = None,
    recent_turns: int = 2,
    max_model_rounds: int | None = None,
) -> Agent:
    """Compose the reproducible legacy defaults into a public ``Agent``.

    With no ``behaviors`` argument this requests the whole legacy-default claim. It refuses all
    unresolved rows and every absent dependency in one error. Pass
    :data:`SUPPORTED_DEFAULT_BEHAVIORS` to compose only the verified subset. This is intentional: a
    successful call is exactly the explicit list it was given, never a silently smaller default set.
    ``lazy_tooling`` with no store or memory row cannot activate, so it composes nothing.

    The required deployment dependencies are named before an ``Agent`` is
    constructed: an extractor for automatic consolidation, a conversation
    source for the two-turn working window, an association source wrapped in
    ``SpreadingActivation``, and a consolidation coordinator for quick naps.
    """
    requested = _requested_rows(behaviors)
    requested_names = {row.behavior for row in requested}
    missing: list[str] = []
    memory_requested = requested_names & {
        "auto_hydrate", "auto_consolidate", "spreading_activation", "enabled_layers",
        "working_memory_recent_turns", "quick_nap_interval",
    }
    inert = "lazy_tooling" if store is None and not memory_requested else None
    unresolved = tuple(row for row in requested if row.status == "gap" and row.behavior != inert)
    if memory_requested and store is None:
        missing.append("store for memory composition")
    if memory_requested and scope is None:
        missing.append("scope for memory composition")
    if "auto_consolidate" in requested_names and extractor is None:
        missing.append("extractor for auto_consolidate")
    if "spreading_activation" in requested_names and activation is None:
        missing.append("activation for spreading_activation")
    if "working_memory_recent_turns" in requested_names and conversation is None:
        missing.append("conversation for working_memory_recent_turns")
    if "quick_nap_interval" in requested_names and consolidation is None:
        missing.append("consolidation for quick_nap_interval")
    if unresolved or missing:
        raise LegacyParityRefusal(unresolved=unresolved, missing_dependencies=missing)

    capabilities: list[Any] = []
    if memory_requested:
        assert store is not None and scope is not None
        capabilities.extend(memory_capabilities(
            store,
            scope,
            extractor=extractor if "auto_consolidate" in requested_names else None,
            conversation=(
                conversation if "working_memory_recent_turns" in requested_names else None
            ),
            recent_turns=(
                recent_turns if "working_memory_recent_turns" in requested_names else 0
            ),
            activation=activation if "spreading_activation" in requested_names else None,
            consolidation=consolidation if "quick_nap_interval" in requested_names else None,
        ))
    if "auto_hydrate" in requested_names:
        # Retrieval reaches the model only when prompting compiles its snapshot.
        capabilities.append(PromptingCapability())
    return Agent(
        model_provider,
        instructions=instructions,
        tools=tools,
        max_model_rounds=max_model_rounds,
        capabilities=capabilities,
    )