symfonic.platform.governance_decisions

governance_decisions

What governance decided, in a shape an operator can group and count.

A refusal already reaches a caller as an exception and an amendment already reaches the tool. Neither answers the question an operator asks the next morning: which rule did that, to which call, and how often?

Three properties shape this module.

Rule identity is declared, never inferred. Not the objection's text, which is prose someone will reword; not the class name, which is a refactor away; not the position in a list, which changes when a deployment inserts a rule above it. An Objector says what it is called, and that is what a dashboard groups by.

A decision carries no values. No arguments, no credential, no scrubbed text. Scrubbing a secret out of a call and then writing it into the line that records the scrub is not a smaller leak -- it is the same leak, in a destination usually kept longer and read by more people.

Attribution survives plurality. Two rules that both changed one call are two rule ids on one decision, not one entry saying "governance".

DecisionLog dataclass

DecisionLog(decisions: list[GovernanceDecision] = list())

The in-process sink, for a deployment that wants one and nothing more.

Not a default: governance() records nothing unless handed a sink, so a deployment that never asked for a decision log does not accumulate one for the life of the process.

by_rule

by_rule(rule_id: str) -> tuple[GovernanceDecision, ...]

Every decision one rule took part in.

The question a dashboard asks, and the reason rule_ids is a tuple rather than a string: a decision two rules made is found by either.

Source code in src/symfonic/platform/governance_decisions.py

def by_rule(self, rule_id: str) -> tuple[GovernanceDecision, ...]:
    """Every decision one rule took part in.

    The question a dashboard asks, and the reason ``rule_ids`` is a tuple
    rather than a string: a decision two rules made is found by either.
    """
    return tuple(d for d in self.decisions if rule_id in d.rule_ids)

DecisionSink

Where decisions go. A protocol in all but name.

Anything with record(decision) is one, so a deployment writes to its own store without inheriting from this package.

GovernanceDecision dataclass

GovernanceDecision(state: str, phase: str, stage: str, rule_ids: tuple[str, ...], call_id: str, tool: str, reason: str, at: datetime = (lambda: datetime.now(UTC))())

One governance outcome for one call.

state is the vocabulary, and it is three words on purpose:

applied A rule rewrote the call and the tool ran with the rewrite. refused A rule objected and the call did not run. discarded A rule asked for a change this phase has no way to make. It is neither of the above and must not be reported as either: an operator reading applied for a change that never happened is worse off than one reading nothing.

acting_records

acting_records(outcome: Any) -> tuple[Any, ...]

The records that changed something, in the order the stages ran.

Two ways a stage can have acted, and the second is easy to miss. It may have steered -- an objection the kernel then has to resolve -- or it may have quietly repaired the subject and admitted it. The second leaves an ALLOW disposition and an amendment finding, so filtering on disposition alone reports a repaired call as though no rule touched it.

Source code in src/symfonic/platform/governance_decisions.py

def acting_records(outcome: Any) -> tuple[Any, ...]:
    """The records that changed something, in the order the stages ran.

    Two ways a stage can have acted, and the second is easy to miss. It may
    have *steered* -- an objection the kernel then has to resolve -- or it may
    have quietly *repaired* the subject and admitted it. The second leaves an
    ALLOW disposition and an amendment finding, so filtering on disposition
    alone reports a repaired call as though no rule touched it.
    """
    from symfonic.capabilities.governance.values import Disposition

    return tuple(
        record
        for record in outcome.trace
        if record.disposition in (Disposition.ANNOTATE, Disposition.STEER)
        or any(f.kind == AMENDMENT for f in getattr(record, "findings", ()) or ())
    )

objection_in

objection_in(outcome: Any) -> Any

The first record where a rule said no, or None.

record.objection is the distinction, and it already existed in the contract. StageVerdict.annotate(subject, ...) is how a stage says "I changed this" -- credential hygiene scrubbing an argument -- and carries no objection. StageVerdict.steer(objection, ...) is how a stage says "I will not have this", and the objection is its first argument.

Disposition cannot tell them apart: a repair and a refusal both leave ANNOTATE or STEER, which is how a repair came to outrank a refusal in the first place. Neither can the finding kinds alone, because a stage that repairs by replacing the subject records findings of its own kind rather than amendment findings -- reading those as objections made credential hygiene refuse every turn it scrubbed.

Source code in src/symfonic/platform/governance_decisions.py

def objection_in(outcome: Any) -> Any:
    """The first record where a rule said no, or ``None``.

    ``record.objection`` is the distinction, and it already existed in the
    contract. ``StageVerdict.annotate(subject, ...)`` is how a stage says "I
    changed this" -- credential hygiene scrubbing an argument -- and carries no
    objection. ``StageVerdict.steer(objection, ...)`` is how a stage says "I
    will not have this", and the objection is its first argument.

    Disposition cannot tell them apart: a repair and a refusal both leave
    ANNOTATE or STEER, which is how a repair came to outrank a refusal in the
    first place. Neither can the finding kinds alone, because a stage that
    repairs by *replacing the subject* records findings of its own kind rather
    than amendment findings -- reading those as objections made credential
    hygiene refuse every turn it scrubbed.
    """
    for record in outcome.trace:
        if str(getattr(record, "objection", "") or "").strip():
            return record
    return None

report

report(sink: Any, *, state: str, phase: str, records: Sequence[Any], reason: str, calls: Sequence[Any] = ()) -> None

Write one decision per call the rules acted on, or one for the turn.

calls are the kernel's requests for this round, and they are the only thing read for identity -- call_id and tool name, never arguments. An ingress or egress rung governs a turn rather than a call and records a single decision with the call fields empty, which is honest: inventing a call id there would let a dashboard join on something that does not exist.

A sink of None is the ordinary case and returns immediately, so a deployment that composed no log pays for none of this.

Source code in src/symfonic/platform/governance_decisions.py

def report(
    sink: Any,
    *,
    state: str,
    phase: str,
    records: Sequence[Any],
    reason: str,
    calls: Sequence[Any] = (),
) -> None:
    """Write one decision per call the rules acted on, or one for the turn.

    ``calls`` are the kernel's requests for this round, and they are the only
    thing read for identity -- ``call_id`` and tool name, never arguments.
    An ingress or egress rung governs a turn rather than a call and records a
    single decision with the call fields empty, which is honest: inventing a
    call id there would let a dashboard join on something that does not exist.

    A ``sink`` of ``None`` is the ordinary case and returns immediately, so a
    deployment that composed no log pays for none of this.
    """
    if sink is None or not records:
        return
    stages = ", ".join(dict.fromkeys(str(r.stage) for r in records))
    rule_ids = tuple(dict.fromkeys(rule for r in records for rule in rules_in(r)))
    subjects = tuple(calls) or (None,)
    for call in subjects:
        sink.record(
            GovernanceDecision(
                state=state,
                phase=phase,
                stage=stages,
                rule_ids=rule_ids,
                call_id=str(getattr(call, "call_id", "") or ""),
                tool=str(getattr(call, "name", "") or ""),
                reason=reason,
            )
        )

rules_in

rules_in(record: Any, kind: str | None = None) -> tuple[str, ...]

The rules that spoke in one stage record, each named individually.

A stage is a place; a rule is a decision. policy_steering is where three tenant rules run, and reporting all three as "policy_steering" tells an operator which file to open rather than which rule to change. Findings carry the declared name, so they are the source; the stage name is the fallback for a stage that objected without one.

Source code in src/symfonic/platform/governance_decisions.py

def rules_in(record: Any, kind: str | None = None) -> tuple[str, ...]:
    """The rules that spoke in one stage record, each named individually.

    A stage is a place; a rule is a decision. ``policy_steering`` is where
    three tenant rules run, and reporting all three as "policy_steering"
    tells an operator which *file* to open rather than which rule to change.
    Findings carry the declared name, so they are the source; the stage name
    is the fallback for a stage that objected without one.
    """
    findings = tuple(getattr(record, "findings", ()) or ())
    if kind == OBJECTION:
        # Only the rules that objected. A record can carry both -- one rule
        # repaired the call and another refused it -- and naming the repairer
        # in a refusal points an operator at the wrong rule.
        findings = tuple(f for f in findings if f.kind != AMENDMENT)
    named = tuple(
        dict.fromkeys(
            finding.rule for finding in findings if getattr(finding, "rule", "")
        )
    )
    return named or (str(getattr(record, "stage", "")),)