symfonic.capabilities.governance.capability

capability

The governance capability: seven concerns as one composition (T3.4.4).

compose is the only assembler. It builds the pipeline in the canonical order from whatever ports the deployment actually has, and it leaves a stage out rather than in-but-inert when its dependency is missing — an inert stage in a trace is indistinguishable from a stage that examined the turn and found nothing, which is the difference between "we checked" and "nobody ever wired the checker up".

Credential hygiene is the exception: it needs no injection, so it is always present. A deployment that genuinely wants no scrubbing says so with patterns=[], which is recorded in the trace as a disabled stage rather than as an absent one.

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

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

@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

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

@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
    ]