Skip to content

symfonic.platform.governance_amendment

governance_amendment

Reading a governance outcome back as a batch of kernel calls.

The mirror of :mod:symfonic.platform.governance_subject, which translates the other way. Two vocabularies meet at this boundary -- the kernel names a request's payload arguments and governance names it args -- and the inbound half already had a module saying so. This is the outbound half, and it exists because a governance stage can now repair a call rather than only refuse one.

amendment_for

amendment_for(gov_phase: str, outcome: GovernanceOutcome, context: Any) -> tuple[tuple[Any, ...], tuple[str, ...]] | None

The rewritten batch a stage produced, or None if none did.

Only the effect rung: it is the only phase whose subject carries calls, and the only one the kernel folds.

None for a no-op is the load-bearing case. An objector that inspects a call and leaves it alone is a reader, not a writer, and reporting an amendment on every governed turn would make a second amending capability collide with a stage that changed nothing.

Identity is not amendable. Arguments are what a rule may repair; which call this is -- its id, its tool, its place in the batch -- is what the model decided, and a stage that changed it would be substituting a different action with the model's reasoning still attached.

Source code in src/symfonic/platform/governance_amendment.py
def amendment_for(
    gov_phase: str, outcome: GovernanceOutcome, context: Any
) -> tuple[tuple[Any, ...], tuple[str, ...]] | None:
    """The rewritten batch a stage produced, or ``None`` if none did.

    Only the effect rung: it is the only phase whose subject carries calls,
    and the only one the kernel folds.

    ``None`` for a no-op is the load-bearing case. An objector that inspects a
    call and leaves it alone is a reader, not a writer, and reporting an
    amendment on every governed turn would make a second amending capability
    collide with a stage that changed nothing.

    Identity is not amendable. Arguments are what a rule may repair; *which*
    call this is -- its id, its tool, its place in the batch -- is what the
    model decided, and a stage that changed it would be substituting a
    different action with the model's reasoning still attached.
    """
    if gov_phase != "effect":
        return None
    amended = tuple(getattr(outcome.subject, "tool_calls", ()) or ())
    original = tuple(getattr(context, "requests", ()) or ())
    if not amended or not original:
        return None
    if len(amended) != len(original):
        raise _refusal(
            f"governance returned {len(amended)} call(s) for a round of "
            f"{len(original)}. A stage may repair a call's arguments; adding "
            "or dropping one is choosing a different action than the model "
            "asked for."
        )

    rules = _amending_rules(outcome)
    changed = False
    calls: list[Any] = []
    for request, call in zip(original, amended, strict=True):
        name = str(getattr(call, "name", ""))
        if name != getattr(request, "name", ""):
            raise _refusal(
                f"governance renamed a call ({request.name!r} -> {name!r}) "
                f"[{', '.join(rules) or 'unattributed'}]. Arguments are "
                "amendable; which tool runs is not."
            )
        args = dict(getattr(call, "args", None) or {})
        if args != dict(getattr(request, "arguments", None) or {}):
            changed = True
        calls.append(replace(request, arguments=args))
    if not changed:
        return None
    return tuple(calls), rules or ("governance",)