symfonic.capabilities.governance.pipeline

pipeline

The pipeline: order, containment, and the evidence trail (T3.4.4).

The pipeline owns the four things no single stage can own:

• order — checked against the rulebook at construction, not at the first governed turn;
• declaration integrity — a stage whose phase or failure mode differs from the rulebook's is refused, so a disarmed guard cannot masquerade as a working one;
• containment — a stage that raises is handled by its declared failure mode, never by a blanket try/except that treats a broken scrubber like a broken critic;
• the trace — every stage's answer, including the ones that abstained, the ones that broke, and the ones observe mode downgraded.

GovernancePipeline

GovernancePipeline(stages: Sequence[GovernanceStage], *, rulebook: RuleBook = CANONICAL_RULEBOOK)

Run governance stages in the declared order and report what they decided.

Source code in src/symfonic/capabilities/governance/pipeline.py

def __init__(
    self,
    stages: Sequence[GovernanceStage],
    *,
    rulebook: RuleBook = CANONICAL_RULEBOOK,
) -> None:
    self._stages = tuple(stages)
    self._rulebook = rulebook
    names = [stage.name for stage in self._stages]
    rulebook.validate(names)
    for stage in self._stages:
        self._assert_declares_what_it_claims(stage)

for_phase

for_phase(phase: str, *, carry: tuple[str, ...] = ()) -> GovernancePipeline | None

The stages this pipeline runs at phase, or None for none.

The kernel runs a turn as a ladder of phases and governance spans three of them -- a credential scrub belongs before the model reads the query, a budget ceiling before a tool is admitted, a reflection pass after the draft exists. Running the whole pipeline at one rung would put the scrubber after the text it was meant to scrub.

None rather than an empty pipeline, because the caller's decision differs: a phase with no stages must contribute no kernel stage at all. A declared stage that examines nothing is exactly the "in-but-inert" shape compose refuses one layer down -- indistinguishable in a trace from a stage that looked and found nothing.

carry names stages that run on every rung, ahead of that rung's own. The rulebook's first rule is why: credential hygiene is ordered first "because every later stage observes the payload", and that is an argument about relative order within a pass, not about which rung a pass happens on. A scrubber that only ran before the model never sees a tool argument or a tool result -- which is where a secret actually travels -- so on a ladder with several passes it has to lead each one.

Carried stages keep their rulebook rank, so the pipeline still refuses an order the rulebook contradicts. A stage already selected for this phase is not added twice.

Source code in src/symfonic/capabilities/governance/pipeline.py

def for_phase(
    self, phase: str, *, carry: tuple[str, ...] = ()
) -> GovernancePipeline | None:
    """The stages this pipeline runs at ``phase``, or ``None`` for none.

    The kernel runs a turn as a ladder of phases and governance spans three
    of them -- a credential scrub belongs before the model reads the query,
    a budget ceiling before a tool is admitted, a reflection pass after the
    draft exists. Running the whole pipeline at one rung would put the
    scrubber after the text it was meant to scrub.

    ``None`` rather than an empty pipeline, because the caller's decision
    differs: a phase with no stages must contribute no kernel stage at all.
    A declared stage that examines nothing is exactly the "in-but-inert"
    shape ``compose`` refuses one layer down -- indistinguishable in a trace
    from a stage that looked and found nothing.

    ``carry`` names stages that run on *every* rung, ahead of that rung's
    own. The rulebook's first rule is why: credential hygiene is ordered
    first "because every later stage observes the payload", and that is an
    argument about relative order within a pass, not about which rung a
    pass happens on. A scrubber that only ran before the model never sees a
    tool argument or a tool result -- which is where a secret actually
    travels -- so on a ladder with several passes it has to lead each one.

    Carried stages keep their rulebook rank, so the pipeline still refuses
    an order the rulebook contradicts. A stage already selected for this
    phase is not added twice.
    """
    selected = tuple(
        stage
        for stage in self._stages
        if str(self._rulebook.rule_for(stage.name).phase) == phase
    )
    carried = tuple(
        stage
        for stage in self._stages
        if stage.name in carry and stage not in selected
    )
    # A rung with only carried stages is still mounted, and that is the
    # case this exists for: a deployment composing no effect stage at all
    # still wants its tool arguments scrubbed before the call is admitted.
    # The "no in-but-inert stage" rule is not in tension with it -- a
    # carried scrubber at that rung has a subject, which is the whole
    # difference between examining nothing and finding nothing.
    if not selected and not carried:
        return None
    return GovernancePipeline(carried + selected, rulebook=self._rulebook)

select

select(names: Sequence[str]) -> GovernancePipeline | None

Select a subsequence without changing its canonical order or rules.

Integration layers choose placement; classification and containment remain properties of the original pipeline.

Source code in src/symfonic/capabilities/governance/pipeline.py

def select(self, names: Sequence[str]) -> GovernancePipeline | None:
    """Select a subsequence without changing its canonical order or rules.

    Integration layers choose placement; classification and containment
    remain properties of the original pipeline.
    """
    for name in names:
        self._rulebook.rule_for(name)
    selected = tuple(stage for stage in self._stages if stage.name in names)
    return GovernancePipeline(selected, rulebook=self._rulebook) if selected else None