Skip to content

symfonic.capabilities.governance.stages.objections

objections

The one evaluation preconditions and policy steering share.

Both stages ask every objector about every call and collect the answers. They differ in authority, not in mechanism: a precondition is a fact about the call the tool itself declared, a policy guard is a rule the deployment imposed. Sharing the loop keeps the containment rule — an objector that raises admits, and says which one — written once.

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.

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