Skip to content

symfonic.capabilities.governance.stages

stages

The seven governed concerns, one module each (T3.4.4).

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)

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

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

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

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

ObjectionSweep dataclass 

ObjectionSweep(findings: tuple[Finding, ...] = (), checked: int = 0, degraded: tuple[str, ...] = ())

What every objector said about every call, and which ones broke.

amendments property 

amendments: tuple[Finding, ...]

The findings that name a rule which rewrote a call.

objections property 

objections: tuple[Finding, ...]

The findings that mean this call is not admissible.

Separate from amendments because a stage steers on one and not the other: a rule that fixed the call did not object to it, and a stage that steered on a successful repair would refuse the very call the repair made admissible.

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)

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"

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)

check_reporter 

check_reporter(confidence: ConfidenceReporter | None) -> ConfidenceReporter | None

The reporter, or a refusal raised where nothing will contain it.

This stage is FAIL_OPEN: the pipeline catches whatever apply raises and allows the turn. So a per-turn refusal cannot protect a composed caller -- it produces the disabled gate :func:_validated exists to prevent, recorded in degraded_stages and nowhere a caller is obliged to look. A port that can never work is knowable before any turn runs, and construction happens outside the containment boundary, so that is where it is refused.

An async reporter is the mistake worth naming: :class:Reflector sits beside this port and is async, and runtime_checkable cannot tell them apart because it only checks that the method exists. Left to run, it returns an un-awaited coroutine every turn -- a warning, a leak, and a gate that quietly stopped guarding.

Source code in src/symfonic/capabilities/governance/stages/metacognition_validation.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def check_reporter(confidence: ConfidenceReporter | None) -> ConfidenceReporter | None:
    """The reporter, or a refusal raised where nothing will contain it.

    This stage is ``FAIL_OPEN``: the pipeline catches whatever ``apply``
    raises and allows the turn. So a per-turn refusal cannot protect a
    composed caller -- it *produces* the disabled gate :func:`_validated`
    exists to prevent, recorded in ``degraded_stages`` and nowhere a caller
    is obliged to look. A port that can never work is knowable before any
    turn runs, and construction happens outside the containment boundary, so
    that is where it is refused.

    An async reporter is the mistake worth naming: :class:`Reflector` sits
    beside this port and *is* async, and ``runtime_checkable`` cannot tell
    them apart because it only checks that the method exists. Left to run, it
    returns an un-awaited coroutine every turn -- a warning, a leak, and a
    gate that quietly stopped guarding.
    """
    if confidence is None:
        return None
    method = getattr(confidence, "confidence", None)
    if not callable(method):
        raise ValueError(
            "a confidence reporter must provide a callable confidence(subject, "
            f"context); {type(confidence).__name__} does not"
        )
    if is_async_callable(method):
        raise ValueError(
            "a confidence reporter must be synchronous: the gate decides "
            "whether the one port that costs a model call is worth calling, "
            f"so the signal it decides on cannot cost one. {type(confidence).__name__}"
            ".confidence is async"
        )
    try:
        inspect.signature(method).bind(object(), object())
    except TypeError as exc:
        raise ValueError(
            "a confidence reporter must accept confidence(subject, context); "
            f"{type(confidence).__name__}.confidence does not: {exc}"
        ) from None
    except ValueError:
        # Some C-backed callables have no inspectable signature. They remain
        # valid; their per-turn result still has the conservative value policy.
        pass
    return confidence

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

sweep 

sweep(objectors: Sequence[Objector], calls: Sequence[ToolCall], subject: GovernanceSubject, *, kind: str) -> ObjectionSweep

Ask each applicable objector about each call.

tool_name on the objector scopes it to one tool; an objector that omits the attribute is asked about every call. An objector that raises is skipped for that call and named in degraded — the finding it might have produced is unknowable, and pretending otherwise in either direction (admit silently, or refuse) would be a guess.

Source code in src/symfonic/capabilities/governance/stages/objections.py 
 55
 56
 57
 58
 59
 60
 61
 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
def sweep(
    objectors: Sequence[Objector],
    calls: Sequence[ToolCall],
    subject: GovernanceSubject,
    *,
    kind: str,
) -> ObjectionSweep:
    """Ask each applicable objector about each call.

    ``tool_name`` on the objector scopes it to one tool; an objector that
    omits the attribute is asked about every call. An objector that raises
    is skipped for that call and named in ``degraded`` — the finding it
    might have produced is unknowable, and pretending otherwise in either
    direction (admit silently, or refuse) would be a guess.
    """
    findings: list[Finding] = []
    degraded: list[str] = []
    checked = 0
    for call in calls:
        for objector in objectors:
            scope = getattr(objector, "tool_name", None)
            if scope is not None and scope != call.name:
                continue
            checked += 1
            before = dict(call.args)
            try:
                objection = objector.check(call, subject)
            except Exception:
                degraded.append(getattr(objector, "name", call.name))
                continue
            if dict(call.args) != before:
                # A rule that rewrote the call without objecting has still
                # made a decision, and until now it made it anonymously: the
                # sweep only recorded objections, so a repair reached the tool
                # with nothing naming the rule that performed it. An amendment
                # nobody can attribute is one nobody can turn off.
                #
                # The *keys* it touched are named and their values are not.
                # "which rule changed which field" is what an operator needs;
                # the value is frequently the credential the rule just removed.
                findings.append(
                    Finding(
                        kind=ObjectionSweep.AMENDMENT,
                        detail=(
                            f"{call.name}: amended "
                            f"{', '.join(sorted(set(before) ^ set(call.args)) or ['arguments'])}"
                        ),
                        rule=str(getattr(objector, "name", "") or call.name),
                    ),
                )
            if objection:
                findings.append(
                    Finding(
                        kind=kind,
                        detail=f"{call.name}: {objection}",
                        # The rule's own name, so attribution survives a
                        # reworded objection and a renamed class. Falling
                        # back to the tool name keeps a finding attributable
                        # when an objector declared none -- badly, but a
                        # dashboard grouping by "publish" is still better
                        # than one grouping by "".
                        rule=str(getattr(objector, "name", "") or call.name),
                    ),
                )
    return ObjectionSweep(
        findings=tuple(findings), checked=checked, degraded=tuple(degraded),
    )