Skip to content

symfonic.capabilities.governance

governance

Safety and governance stages (T3.4.4).

Seven safety concerns used to live in six places: a scrubber in agent/hygiene.py, an intent filter threaded through three engine entry points, a fabrication detector and a metacognitive critic in agent/middleware/, a precondition gate in core/nodes/, steering inside the tool span seam, and budgeting in services/budget/. Each one worked. Together they had no order anyone had written down, no shared account of what happens when one of them breaks, and no single record of what they decided about a turn.

This package composes them:

  • The rulebook — :mod:.rulebook. The canonical order, and for each stage a declared failure mode with the argument for it. Ordering is the one property no individual stage can enforce, so it lives here and the pipeline refuses a composition that contradicts it.
  • The pipeline — :mod:.pipeline. Order, declaration integrity, containment by declared failure mode, and the trace.
  • The seven stages — :mod:.stages. One module each, each one drivable on its own with no graph, no engine, and no provider.
  • The ports — :mod:.contracts. Classifier, detector, critic, meter, and objector, all structural. The capability implements none of them.

Failure is a declaration, not an accident. Every stage says in advance what its own breakage means. Five fail open, because a broken preference must not become an outage. Two fail closed — credential hygiene and budget — because a scrubber that did not run has not proved anything, and a spend that cannot be measured cannot be bounded.

Objecting is not failing. A stage that finds an unmet precondition, a policy violation, or an unsupported claim is a stage that worked. The legacy gates conflated the two by catching everything and returning "allowed", which made a crashed checker and a clean turn produce the same trace.

This package imports nothing outside itself; a contract test asserts it.

BudgetLimits dataclass 

BudgetLimits(draft_tokens: int | None = None)

None measures without capping; a number caps.

Zero is refused rather than read as "unlimited": a deployment that typed 0 meant "nothing", and silently reading that as "everything" is the most expensive possible misreading.

BudgetStage 

BudgetStage(meter: TokenMeter, limits: BudgetLimits | None = None)

Measure the payload, trim what may be trimmed, refuse what may not.

Source code in src/symfonic/capabilities/governance/stages/budget.py 
58
59
60
def __init__(self, meter: TokenMeter, limits: BudgetLimits | None = None) -> None:
    self._meter = meter
    self._limits = limits or BudgetLimits()

ConfidenceOnlyMetacognitionPolicy 

ConfidenceOnlyMetacognitionPolicy(*, confidence_floor: float = 0.6, sensitive_terms: Sequence[str] = ())

The pre-FP-1 capability policy, retained as an explicit cost opt-out.

It intentionally does not preserve the legacy selective hard floors. A deployment choosing it accepts that numbers, tool actions, recalled tags, findings, and intent do not by themselves request reflection.

Source code in src/symfonic/capabilities/governance/stages/metacognition_policy.py 
141
142
143
144
145
def __init__(
    self, *, confidence_floor: float = 0.6, sensitive_terms: Sequence[str] = ()
) -> None:
    self._floor = validate_floor(confidence_floor)
    self._sensitive_terms = frozenset(term.lower() for term in sensitive_terms if term)

ConfidenceReporter

Bases: Protocol

How sure the caller is of this draft, when nothing else knows.

Synchronous and offline, like :class:TokenMeter and unlike :class:Reflector: the gate exists to decide whether the one expensive port is worth calling, so the signal it decides on must not cost a model call of its own.

Returning None withholds the claim and leaves the gate unfired, which is the same tri-state rule GovernanceSubject.confidence follows. The framework never computes this: a confidence synthesised from whatever happened to be in reach is how the legacy gate came to fire on every turn.

CredentialHygieneStage 

CredentialHygieneStage(patterns: Sequence[str] | None | object = USE_DEFAULT_PATTERNS)

Drop credential-shaped keys before any other stage observes them.

Source code in src/symfonic/capabilities/governance/stages/credentials.py 
116
117
118
119
120
121
def __init__(
    self, patterns: Sequence[str] | None | object = USE_DEFAULT_PATTERNS,
) -> None:
    supplied = None if patterns is USE_DEFAULT_PATTERNS else patterns
    self._pattern = compile_credential_pattern(supplied)  # type: ignore[arg-type]
    self._disabled = self._pattern is None

Disposition

Bases: StrEnum

What a working stage decided about this turn.

ALLOW nothing to say. ANNOTATE the subject was amended; the amendment travels onward. STEER an objection the model (or the caller) must act on; the turn continues, because a governance stage that halts on every objection cannot express "fix this and carry on". REFUSE terminal. No later stage runs.

Enforcement

Bases: StrEnum

The three-position knob every legacy governance flag already had.

OFF is a zero-cost guarantee, not a quiet allow: the pipeline does not run, so no classifier is called and no meter is read.

FabricationDetector

Bases: Protocol

Scans a draft against what actually grounded it.

Synchronous by contract: the legacy detector is regex over text, and an async signature would invite someone to put a network call behind the one stage that has to be cheap enough to run on every turn.

FabricationStage 

FabricationStage(detector: FabricationDetector | None, *, min_confidence: float = 0.6, refuse_min_confidence: float | None = 0.9)

Scan the draft, cross-check the intent, and report what is unsupported.

Source code in src/symfonic/capabilities/governance/stages/fabrication.py 
53
54
55
56
57
58
59
60
61
62
def __init__(
    self,
    detector: FabricationDetector | None,
    *,
    min_confidence: float = 0.6,
    refuse_min_confidence: float | None = 0.9,
) -> None:
    self._detector = detector
    self._min_confidence = min_confidence
    self._refuse_min_confidence = refuse_min_confidence

FailureMode

Bases: StrEnum

What the pipeline does when a stage fails, as opposed to objects.

An objection is the stage working. A raise is the stage broken. The two are separate axes and conflating them is how a crashed scrubber turns into a clean bill of health.

Finding dataclass 

Finding(kind: str, detail: str = '', confidence: float = 1.0, stage: str = '', rule: str = '')

One piece of evidence a stage produced.

confidence is the number thresholds are read against. A stage that cannot estimate one states 1.0 and says so in detail rather than inventing a hedge.

with_stage 

with_stage(stage: str) -> Finding

Stamp the stage, keeping everything else.

replace rather than a field-by-field rebuild: the rebuild dropped whichever field was added last, silently, and a finding that reaches an operator with its rule erased is worse than one that never travelled.

Source code in src/symfonic/capabilities/governance/values.py 
103
104
105
106
107
108
109
110
111
def with_stage(self, stage: str) -> Finding:
    """Stamp the stage, keeping everything else.

    ``replace`` rather than a field-by-field rebuild: the rebuild dropped
    whichever field was added last, silently, and a finding that reaches
    an operator with its rule erased is worse than one that never
    travelled.
    """
    return replace(self, stage=stage)

GovernanceCapability 

GovernanceCapability(pipeline: GovernancePipeline)

One object that answers: what does governance do to this turn, and why?

Source code in src/symfonic/capabilities/governance/capability.py 
59
60
def __init__(self, pipeline: GovernancePipeline) -> None:
    self._pipeline = pipeline

compose classmethod 

compose(*, classifier: IntentClassifier | None = None, detector: FabricationDetector | None = None, reflector: Reflector | None = None, meter: TokenMeter | None = None, preconditions: Sequence[Objector] = (), guards: Sequence[Objector] = (), limits: BudgetLimits | None = None, patterns: Sequence[str] | None | object = USE_DEFAULT_PATTERNS, confidence_floor: float = 0.6, sensitive_terms: Sequence[str] = (), sensitive_tags: Sequence[str] = (), read_only_tools: Sequence[str] = (), trivial_ack_patterns: Sequence[str] = (), confidence: ConfidenceReporter | None = None, trigger_policy: MetacognitionTriggerPolicy | None = None, min_confidence: float = 0.6, refuse_min_confidence: float | None = 0.9, rulebook: RuleBook = CANONICAL_RULEBOOK) -> GovernanceCapability

Build the pipeline in canonical order from the available ports.

Source code in src/symfonic/capabilities/governance/capability.py 
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
@classmethod
def compose(
    cls,
    *,
    classifier: IntentClassifier | None = None,
    detector: FabricationDetector | None = None,
    reflector: Reflector | None = None,
    meter: TokenMeter | None = None,
    preconditions: Sequence[Objector] = (),
    guards: Sequence[Objector] = (),
    limits: BudgetLimits | None = None,
    patterns: Sequence[str] | None | object = USE_DEFAULT_PATTERNS,
    confidence_floor: float = 0.6,
    sensitive_terms: Sequence[str] = (),
    sensitive_tags: Sequence[str] = (),
    read_only_tools: Sequence[str] = (),
    trivial_ack_patterns: Sequence[str] = (),
    confidence: ConfidenceReporter | None = None,
    trigger_policy: MetacognitionTriggerPolicy | None = None,
    min_confidence: float = 0.6,
    refuse_min_confidence: float | None = 0.9,
    rulebook: RuleBook = CANONICAL_RULEBOOK,
) -> GovernanceCapability:
    """Build the pipeline in canonical order from the available ports."""
    # Refused here too, because the stage that checks it is only built
    # when a reflector is wired -- and a reporter that can never work is
    # a misconfiguration whether or not anything would have asked it.
    check_reporter(confidence)
    stages: list[GovernanceStage] = [CredentialHygieneStage(patterns)]
    if classifier is not None:
        stages.append(IntentFilterStage(classifier))
    if preconditions:
        stages.append(ToolPreconditionStage(preconditions))
    if guards:
        stages.append(PolicySteeringStage(guards))
    if meter is not None:
        stages.append(BudgetStage(meter, limits))
    if detector is not None:
        stages.append(
            FabricationStage(
                detector,
                min_confidence=min_confidence,
                refuse_min_confidence=refuse_min_confidence,
            )
        )
    if confidence is not None and reflector is None:
        raise ValueError(
            "a confidence reporter requires a reflector; otherwise its "
            "signal has no governed consumer"
        )
    if reflector is not None:
        stages.append(
            MetacognitionStage(
                reflector,
                confidence_floor=confidence_floor,
                sensitive_terms=sensitive_terms,
                sensitive_tags=sensitive_tags,
                read_only_tools=read_only_tools,
                trivial_ack_patterns=trivial_ack_patterns,
                confidence=confidence,
                trigger_policy=trigger_policy,
            )
        )
    # The list is built in canonical order above; the pipeline still
    # validates it, so an edit that reorders these branches fails a
    # test rather than shipping a reordered safety pipeline.
    return cls(GovernancePipeline(stages, rulebook=rulebook))

declarations 

declarations() -> list[dict[str, str]]

The declaration table for the rulebook this capability composes.

Source code in src/symfonic/capabilities/governance/capability.py 
143
144
145
def declarations(self) -> list[dict[str, str]]:
    """The declaration table for the rulebook this capability composes."""
    return self.describe(self._pipeline.rulebook)

describe staticmethod 

describe(rulebook: RuleBook = CANONICAL_RULEBOOK) -> list[dict[str, str]]

Render the ordered stage/phase/failure-mode/rationale table.

This is the reviewable artefact: the same table a release note or a security review reads, generated from the objects the pipeline actually enforces rather than from a document beside them.

Source code in src/symfonic/capabilities/governance/capability.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
@staticmethod
def describe(rulebook: RuleBook = CANONICAL_RULEBOOK) -> list[dict[str, str]]:
    """Render the ordered stage/phase/failure-mode/rationale table.

    This is the reviewable artefact: the same table a release note or
    a security review reads, generated from the objects the pipeline
    actually enforces rather than from a document beside them.
    """
    return [
        {
            "stage": rule.name,
            "phase": str(rule.phase),
            "failure_mode": str(rule.failure_mode),
            "rationale": rule.rationale,
        }
        for rule in rulebook.rules
    ]

GovernanceContext dataclass 

GovernanceContext(run_id: str = '', scope_path: tuple[str, ...] = (), enforcement: Enforcement = Enforcement.ENFORCE, intent: IntentReading | None = None, metadata: dict[str, Any] = dict())

Ambient facts for one governed turn.

intent starts empty and is filled by the intent stage. A stage reading it before that stage has run reads None — which is the withheld claim, not "no action intended".

GovernanceError

Bases: Exception

Base class for every governance composition failure.

GovernanceOutcome dataclass 

GovernanceOutcome(subject: GovernanceSubject, trace: tuple[StageRecord, ...] = (), refusal: StageRecord | None = None)

The governed turn: the surviving subject and the whole trail.

GovernancePhase

Bases: StrEnum

When a stage runs, relative to the model and to the effects.

The phase is not decoration: it is what makes the canonical order checkable. A stage that claims INGRESS cannot be sequenced after one that claims EGRESS, because "before the model reads the turn" and "after the model has drafted" are not orderings a composition may reverse locally.

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 
51
52
53
54
55
56
57
58
59
60
61
62
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 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
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 
85
86
87
88
89
90
91
92
93
94
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

GovernanceStage

Bases: Protocol

One governed concern.

The three class-level declarations are not metadata: the pipeline checks them against the rulebook at composition time and refuses a stage whose claims differ. apply may raise — that is precisely what failure_mode is a declaration about.

GovernanceSubject dataclass 

GovernanceSubject(query: str = '', draft: str = '', pinned: str = '', grounding: str = '', properties: Mapping[str, Any] = dict(), tool_calls: tuple[ToolCall, ...] = (), confidence: float | None = None)

The turn, as governance sees it.

pinned is the portion of the payload that may never be dropped to make room — boundaries, operator instructions, the safety preamble. Budgeting refuses rather than trims it (T3.2.1's rule: a prompt missing its boundaries is worse than a prompt that refused to build).

with_properties 

with_properties(properties: Mapping[str, Any]) -> GovernanceSubject

A subject carrying these properties, which the result cannot edit.

A proxy rather than dict(...). Several stages read one subject's properties in a turn, and a stage handed a mutable bag could rewrite what a later stage sees -- a side channel around the pipeline, whose whole shape is "a stage says what it decided by returning a verdict".

Source code in src/symfonic/capabilities/governance/context.py 
45
46
47
48
49
50
51
52
53
def with_properties(self, properties: Mapping[str, Any]) -> GovernanceSubject:
    """A subject carrying these properties, which the result cannot edit.

    A proxy rather than ``dict(...)``. Several stages read one subject's
    properties in a turn, and a stage handed a mutable bag could rewrite
    what a later stage sees -- a side channel around the pipeline, whose
    whole shape is "a stage says what it decided by *returning* a verdict".
    """
    return replace(self, properties=MappingProxyType(dict(properties)))

IntentClassifier

Bases: Protocol

Reads the user's turn. Async because the real one calls a model.

IntentFilterStage 

IntentFilterStage(classifier: IntentClassifier | None)

Classify the user's turn and publish the verdict onto the context.

Source code in src/symfonic/capabilities/governance/stages/intent.py 
37
38
def __init__(self, classifier: IntentClassifier | None) -> None:
    self._classifier = classifier

IntentReading dataclass 

IntentReading(label: str, confidence: float = 0.0)

The classifier's answer about the user's turn.

Published once by the intent stage and read by later stages; the canonical order exists so that "later" is a fact rather than a hope.

MetacognitionStage 

MetacognitionStage(reflector: Reflector | None, *, confidence_floor: float = 0.6, sensitive_terms: Sequence[str] = (), sensitive_tags: Sequence[str] = (), read_only_tools: Sequence[str] = (), trivial_ack_patterns: Sequence[str] = (), confidence: ConfidenceReporter | None = None, trigger_policy: MetacognitionTriggerPolicy | None = None)

Reflect on the draft when, and only when, the gate fires.

Source code in src/symfonic/capabilities/governance/stages/metacognition.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def __init__(
    self,
    reflector: Reflector | None,
    *,
    confidence_floor: float = 0.6,
    sensitive_terms: Sequence[str] = (),
    sensitive_tags: Sequence[str] = (),
    read_only_tools: Sequence[str] = (),
    trivial_ack_patterns: Sequence[str] = (),
    confidence: ConfidenceReporter | None = None,
    trigger_policy: MetacognitionTriggerPolicy | None = None,
) -> None:
    self._reflector = reflector
    self._floor = validate_floor(confidence_floor)
    self._confidence = check_reporter(confidence)
    self._policy = _usable_policy(trigger_policy) or SelectiveMetacognitionPolicy(
        confidence_floor=confidence_floor,
        sensitive_terms=sensitive_terms,
        sensitive_tags=sensitive_tags,
        read_only_tools=read_only_tools,
        trivial_ack_patterns=trivial_ack_patterns,
    )

MetacognitionTriggerPolicy

Bases: Protocol

Names the evidence that requires reflection for one draft.

None means the policy deliberately skipped a trivial acknowledgment; a non-empty string is copied into the trace as the trigger. The policy is synchronous and uses only the subject/context evidence already present at the governance boundary.

Objector

Bases: Protocol

A precondition or a policy guard: same shape, different authority.

Returns the objection text, or None to admit. tool_name scopes a precondition to one tool; a guard that omits it sees every call.

PolicySteeringStage 

PolicySteeringStage(guards: Sequence[Objector])

Ask every policy guard about every admitted call.

Source code in src/symfonic/capabilities/governance/stages/steering.py 
37
38
def __init__(self, guards: Sequence[Objector]) -> None:
    self._guards = tuple(guards)

Reflection dataclass 

Reflection(revise: bool, reason: str = '', confidence: float | None = None)

The critic's answer about the draft.

Reflector

Bases: Protocol

The critic. Async, and the only port that costs a model call.

RuleBook 

RuleBook(rules: Iterable[StageRule])

An ordered, immutable set of stage declarations.

Source code in src/symfonic/capabilities/governance/ordering.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def __init__(self, rules: Iterable[StageRule]) -> None:
    ordered = tuple(rules)
    seen: dict[str, StageRule] = {}
    previous = -1
    for rule in ordered:
        if rule.name in seen:
            raise StageOrderError(f"stage {rule.name!r} is declared twice")
        rank = PHASE_RANK[rule.phase]
        if rank < previous:
            raise StageOrderError(
                f"stage {rule.name!r} declares phase {rule.phase} after a later "
                "phase; phases may not run backwards",
            )
        previous = rank
        seen[rule.name] = rule
    self._rules = ordered
    self._by_name = seen

extend 

extend(rule: StageRule, *, after: str | None = None) -> RuleBook

Return a new rulebook with rule inserted after after.

after=None appends. The result is validated by the constructor, so a phase inversion is refused here rather than at the first turn.

Source code in src/symfonic/capabilities/governance/ordering.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def extend(self, rule: StageRule, *, after: str | None = None) -> RuleBook:
    """Return a new rulebook with ``rule`` inserted after ``after``.

    ``after=None`` appends. The result is validated by the constructor,
    so a phase inversion is refused here rather than at the first turn.
    """
    if rule.name in self._by_name:
        raise StageOrderError(f"stage {rule.name!r} is already declared")
    if after is None:
        return RuleBook((*self._rules, rule))
    index = self.rank_of(after)
    return RuleBook(
        (*self._rules[: index + 1], rule, *self._rules[index + 1 :]),
    )

validate 

validate(stages: Sequence[str]) -> None

Refuse a composition that is not a subsequence of this order.

Source code in src/symfonic/capabilities/governance/ordering.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
def validate(self, stages: Sequence[str]) -> None:
    """Refuse a composition that is not a subsequence of this order."""
    seen: set[str] = set()
    previous = -1
    for name in stages:
        rank = self.rank_of(name)
        if name in seen:
            raise StageOrderError(f"stage {name!r} appears twice in the pipeline")
        if rank <= previous:
            raise StageOrderError(
                f"stage {name!r} is composed after {self.names[previous]!r}; the "
                f"declared order is {' -> '.join(self.names)}",
            )
        seen.add(name)
        previous = rank

SelectiveMetacognitionPolicy 

SelectiveMetacognitionPolicy(*, confidence_floor: float = 0.6, sensitive_terms: Sequence[str] = (), sensitive_tags: Sequence[str] = (), read_only_tools: Sequence[str] = (), trivial_ack_patterns: Sequence[str] = ())

Preserve the legacy selective gate's hard floors on the capability path.

A scalar confidence is supplemental evidence. It can never stand in for claims, actions, detector findings, or recall-time sensitivity.

Source code in src/symfonic/capabilities/governance/stages/metacognition_policy.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def __init__(
    self,
    *,
    confidence_floor: float = 0.6,
    sensitive_terms: Sequence[str] = (),
    sensitive_tags: Sequence[str] = (),
    read_only_tools: Sequence[str] = (),
    trivial_ack_patterns: Sequence[str] = (),
) -> None:
    self._floor = validate_floor(confidence_floor)
    self._sensitive_terms = frozenset(term.lower() for term in sensitive_terms if term)
    self._sensitive_tags = frozenset(tag.lower() for tag in sensitive_tags if tag)
    self._read_only = frozenset(read_only_tools)
    self._ack_patterns = tuple(trivial_ack_patterns) or _ACK_PATTERNS

trigger 

trigger(subject: GovernanceSubject, context: GovernanceContext, confidence: float | None) -> str | None

Return a stable trigger name, or None for a trivial ack.

Source code in src/symfonic/capabilities/governance/stages/metacognition_policy.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def trigger(
    self,
    subject: GovernanceSubject,
    context: GovernanceContext,
    confidence: float | None,
) -> str | None:
    """Return a stable trigger name, or ``None`` for a trivial ack."""
    draft = subject.draft
    lowered = draft.lower()
    term = next((term for term in self._sensitive_terms if term in lowered), None)
    if term is not None:
        return f"sensitive_term:{term}"
    tag = self._recalled_tag(subject)
    if tag is not None:
        return f"recalled_sensitive_tag:{tag}"
    if self._findings(context):
        return "fabrication_finding"
    if any(call.name not in self._read_only for call in subject.tool_calls):
        return "mutating_tool"
    if getattr(context.intent, "label", None) in {"action", "ambiguous"}:
        return f"intent:{context.intent.label}"
    if any(pattern.search(draft) for pattern in _CLAIM_PATTERNS):
        return "draft_claim"
    if self._is_trivial_ack(draft):
        return None
    if confidence is not None:
        return "confidence_below_floor" if confidence < self._floor else "substantive_draft"
    return "substantive_draft"

StageContractError

Bases: GovernanceError

A stage object declares something other than what the rulebook does.

The likeliest instance is a stage quietly declaring FAIL_OPEN for a concern the rulebook declares fail-closed — a disarmed guard that would otherwise look identical to a working one.

StageOrderError

Bases: GovernanceError

The composition contradicts the rulebook's order.

Duplicates, inversions, and phase reversals all land here. Ordering is the one property no individual stage can enforce.

StageRecord dataclass 

StageRecord(stage: str, phase: GovernancePhase, failure_mode: FailureMode, disposition: Disposition, reason: str = '', objection: str = '', findings: tuple[Finding, ...] = (), evidence: Mapping[str, Any] = dict(), degraded: bool = False, downgraded: bool = False)

What one stage did, including the stages that did nothing.

degraded means the stage failed and the declared failure mode decided the rest. downgraded means the stage worked and observe mode suppressed its terminal effect. Two flags because they are two different post-mortems.

StageRule dataclass 

StageRule(name: str, phase: GovernancePhase, failure_mode: FailureMode, rationale: str)

One stage's declaration: where it runs, and what a failure means.

rationale is required and is checked for being a sentence rather than a label. A failure mode without a stated reason is a coin flip somebody will later "optimise" in the wrong direction.

StageVerdict dataclass 

StageVerdict(disposition: Disposition, reason: str = '', objection: str = '', subject: GovernanceSubject | None = None, findings: tuple[Finding, ...] = (), evidence: Mapping[str, Any] = dict())

One stage's answer. Built through the four constructors, not by hand.

allow classmethod 

allow(reason: str = '', *, evidence: Mapping[str, Any] | None = None, findings: Sequence[Finding] = ()) -> StageVerdict

Admit the subject, optionally saying what was noticed on the way.

findings on an allow is not a contradiction. A rule that repaired a call and raised no objection admitted it -- and still decided something an operator needs attributed. Without this the repair travelled anonymously, because only objections were recorded.

Source code in src/symfonic/capabilities/governance/records.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
@classmethod
def allow(
    cls,
    reason: str = "",
    *,
    evidence: Mapping[str, Any] | None = None,
    findings: Sequence[Finding] = (),
) -> StageVerdict:
    """Admit the subject, optionally saying what was noticed on the way.

    ``findings`` on an *allow* is not a contradiction. A rule that
    repaired a call and raised no objection admitted it -- and still
    decided something an operator needs attributed. Without this the
    repair travelled anonymously, because only objections were recorded.
    """
    return cls(
        Disposition.ALLOW,
        reason=reason,
        evidence=dict(evidence or {}),
        findings=tuple(findings),
    )

TokenMeter

Bases: Protocol

Counts a payload. Offline by construction — see T3.2.1's budgeting rule.

ToolCall dataclass 

ToolCall(name: str, args: Mapping[str, Any] = dict(), result: str = '')

A reading of one call: enough to govern it, not enough to run it.

with_result 

with_result(result: str) -> ToolCall

Replace the result, keeping the call it belongs to.

The twin of :meth:with_args, and it exists because that method preserves result — deliberately, since amending a call's inputs should not silently rewrite what it returned. Scrubbing the output is a separate decision and therefore a separate method.

Source code in src/symfonic/capabilities/governance/values.py 
146
147
148
149
150
151
152
153
154
def with_result(self, result: str) -> ToolCall:
    """Replace the result, keeping the call it belongs to.

    The twin of :meth:`with_args`, and it exists because that method
    *preserves* ``result`` — deliberately, since amending a call's inputs
    should not silently rewrite what it returned. Scrubbing the output is a
    separate decision and therefore a separate method.
    """
    return ToolCall(name=self.name, args=dict(self.args), result=result)

ToolPreconditionStage 

ToolPreconditionStage(preconditions: Sequence[Objector])

Check every declared precondition against every requested call.

Source code in src/symfonic/capabilities/governance/stages/preconditions.py 
39
40
def __init__(self, preconditions: Sequence[Objector]) -> None:
    self._preconditions = tuple(preconditions)

UnknownStageError

Bases: GovernanceError

A stage name the rulebook does not declare.

Raised rather than defaulted: a stage nobody declared has no declared phase and no declared failure mode, so the pipeline cannot say what it would do when the stage breaks.

compile_credential_pattern 

compile_credential_pattern(patterns: Sequence[str] | None) -> re.Pattern[str] | None

None -> the default set; [] -> disabled; otherwise a replacement.

An invalid fragment raises re.error here, at construction, rather than at scrub time: a misconfigured pattern list must fail while the deployment is being built, not while a secret is passing through it.

Source code in src/symfonic/capabilities/governance/stages/credentials.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
def compile_credential_pattern(
    patterns: Sequence[str] | None,
) -> re.Pattern[str] | None:
    """``None`` -> the default set; ``[]`` -> disabled; otherwise a replacement.

    An invalid fragment raises ``re.error`` here, at construction, rather
    than at scrub time: a misconfigured pattern list must fail while the
    deployment is being built, not while a secret is passing through it.
    """
    if patterns is None:
        return _DEFAULT_PATTERN
    if len(patterns) == 0:
        return None
    joined = "|".join(f"(?:{p})" for p in patterns)
    return re.compile(r"(?i)(" + joined + r")")