Capability layer and public authoring contract (LAY-ADR §2).
A capability owns one coherent piece of agent behaviour behind a narrow
contract. Per the T1.2.1 dependency matrix a capability may import the
kernel's contracts and its own package — nothing else. Collaboration
between capabilities is not an import edge: it goes through ports
declared to the compiler and wired via the compiled plan.
The names exported here are the small public vocabulary an adopter needs to
implement :class:symfonic.CapabilityConfig. Runtime machinery remains under
symfonic.kernel; capability authors should not need that internal package
to declare a stage and return a result.
CapabilityContribution
dataclass
CapabilityContribution(capability: str, stages: tuple[StageDescriptor, ...] = (), handlers: Mapping[str, Any] = (lambda: MappingProxyType({}))(), effect_grants: frozenset[str] = frozenset(), tools: tuple[Any, ...] = (), preconditions: tuple[Any, ...] = ())
One capability's declaration and the code that answers it.
validate
Refuse the shapes that compile and cannot work, or must not.
Raised at fold time so the offending capability is named, rather than
surfacing later as a stage nothing answers or an effect nobody granted.
Source code in src/symfonic/kernel/contracts/contributions.py
| def validate(self) -> None:
"""Refuse the shapes that compile and cannot work, or must not.
Raised at fold time so the offending capability is named, rather than
surfacing later as a stage nothing answers or an effect nobody granted.
"""
self._require_own_stages()
declared = [stage.stage_id for stage in self.stages]
unanswered = sorted(set(declared) - set(self.handlers))
if unanswered:
raise ConfigurationError(
f"capability {self.capability!r} declares stage(s) {unanswered} "
"and supplies no handler for them. A declared stage nothing "
"answers orders correctly, appears in diagnostics, and does "
"nothing — supply a handler, or do not declare the stage."
)
stray = sorted(set(self.handlers) - set(declared))
if stray:
raise ConfigurationError(
f"capability {self.capability!r} supplies handler(s) {stray} for "
"stages it does not declare. A handler for a stage that will "
"never be compiled is dead code the plan cannot reach."
)
needed = frozenset().union(
*(frozenset(stage.effects) for stage in self.stages)
) if self.stages else frozenset()
require_known_families(
self.effect_grants, subject=f"capability {self.capability!r}"
)
ungranted = sorted(needed - self.effect_grants)
if ungranted:
raise ConfigurationError(
f"capability {self.capability!r} declares stage effect(s) "
f"{ungranted} that its own contribution does not grant (STG-8). "
"A stage may exercise only effect families the plan granted; "
"declaring one it did not ask for makes the grant list a "
"description rather than a bound."
)
|
Phase
Bases: StrEnum
The eight phases, in the only order they ever run (STG-1).
StageDescriptor
dataclass
StageDescriptor(stage_id: str, phase: Phase, capability: str = _KERNEL_CAPABILITY, priority: int = 0, after: tuple[str, ...] = (), before: tuple[str, ...] = (), optional_after: tuple[str, ...] = (), optional_before: tuple[str, ...] = (), effects: frozenset[str] = frozenset(), emits: frozenset[str] = frozenset(), kind: StageKind = StageKind.COMPILATION, config: Mapping[str, Any] = (lambda: MappingProxyType({}))())
One capability's declaration of a stage it contributes (CON-C-1).
validate
Shape validation, performed the moment a capability registers (STG-4).
Resolution of the constraints against the full stage set happens later,
at compile time, when the whole set is finally knowable.
Source code in src/symfonic/kernel/contracts/stages.py
| def validate(self) -> None:
"""Shape validation, performed the moment a capability registers (STG-4).
Resolution of the constraints against the full stage set happens later,
at compile time, when the whole set is finally knowable.
"""
if not self.stage_id:
raise ConfigurationError("a stage descriptor must declare a non-empty stage_id.")
if self.phase not in PHASE_LADDER:
raise ConfigurationError(
f"stage {self.stage_id!r} declares unknown phase {self.phase!r}; "
f"the ladder is {[phase.value for phase in PHASE_LADDER]}."
)
if self.phase in KERNEL_OWNED_PHASES and self.capability != _KERNEL_CAPABILITY:
raise ConfigurationError(
f"stage {self.stage_id!r} (capability {self.capability!r}) targets the "
f"kernel-owned phase {self.phase.value!r}; only the kernel registers "
"into 'bind' and 'teardown'."
)
constrained = (
*self.after,
*self.before,
*self.optional_after,
*self.optional_before,
)
if self.stage_id in constrained:
raise ConfigurationError(
f"stage {self.stage_id!r} declares a constraint on itself."
)
# Checked here rather than at dispatch: an unknown family that reaches
# dispatch is indistinguishable from one that was simply not granted,
# so a typo would read as a policy refusal.
require_known_families(self.effects, subject=f"stage {self.stage_id!r}")
# Scoped to PROMPT_ASSEMBLY on purpose. Only that phase is split today,
# because only that phase has a pure consumer that a prior effect must
# not contaminate. A POST_MODEL stage writing to memory is effectful and
# has no compiler downstream to keep pure, so the distinction has
# nothing to say about it -- and applying the rule there would have made
# every effectful stage outside P1 declare a kind that means nothing.
if (
self.phase is Phase.PROMPT_ASSEMBLY
and self.kind is StageKind.COMPILATION
and self.effects
):
raise ConfigurationError(
f"stage {self.stage_id!r} (capability {self.capability!r}) is a "
f"compilation stage and declares effect(s) {sorted(self.effects)}. "
"STG-7: compilation is a pure function of (plan, request, "
"snapshot). A stage that reaches outside declares "
"kind=StageKind.RESOLUTION, runs before every compilation stage "
"in its phase, and contributes to the snapshot rather than to "
"the assembly."
)
|
no_change
no_change(reason: str, *, diagnostics: tuple[str, ...] = (), counts: Mapping[str, int] | None = None) -> StageResult[Any]
The stage ran and had nothing to do. reason is not optional.
Counts are accepted here too, and that is the interesting case: a
retrieval that found nothing reports found=0, which is a measurement.
Reporting no counts at all would leave a reader unable to tell it from a
stage that does not count.
Source code in src/symfonic/kernel/contracts/results.py
| def no_change(
reason: str,
*,
diagnostics: tuple[str, ...] = (),
counts: Mapping[str, int] | None = None,
) -> StageResult[Any]:
"""The stage ran and had nothing to do. ``reason`` is not optional.
Counts are accepted here too, and that is the interesting case: a
retrieval that found nothing reports ``found=0``, which is a measurement.
Reporting no counts at all would leave a reader unable to tell it from a
stage that does not count.
"""
return StageResult(
outcome=StageOutcome.NO_CHANGE,
reason=reason,
diagnostics=diagnostics,
counts=freeze_counts(counts),
)
|