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.

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

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.

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.

    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)
    unresolved = tuple(row for row in requested if row.status == "gap")
    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",
    }
    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,
    )