async def run_phase(
pipeline: GovernancePipeline,
gov_phase: str,
context: Any,
decisions: Any = None,
) -> StageResult[Any]:
"""Run one rung's stages and map the outcome onto a kernel verdict.
Three outcomes, and ``decisions`` records all three under those names:
``applied`` when a rule repaired a call that then ran, ``refused`` when a
rule stopped one, ``discarded`` when a rule asked for a change this rung
cannot make and its failure mode let the turn continue anyway. The third
is the one worth a word of its own -- reporting it as either of the others
would tell an operator a change happened that did not, or that a turn
stopped that did not.
"""
turn_request = getattr(context, "request", None)
if turn_request is None: # pragma: no cover - defensive
return no_change("no turn request on the stage context")
governance_context = context_for(turn_request, getattr(context, "resolved", None))
outcome = await pipeline.run(subject_for(gov_phase, turn_request, context), governance_context)
calls = tuple(getattr(context, "requests", ()) or ())
if outcome.refusal is not None:
rules = rules_in(outcome.refusal)
report(
decisions,
state="refused",
phase=gov_phase,
records=(outcome.refusal,),
reason=outcome.refusal.reason or outcome.refusal.objection,
calls=calls,
)
raise GovernanceRefused(
stage=outcome.refusal.stage,
phase=gov_phase,
disposition=outcome.refusal.disposition,
reason=outcome.refusal.reason,
rule_ids=rules,
)
# A refusal is evaluated before an amendment, and the order is the whole
# correction. It shipped the other way round: "did anything amend the
# call?" came first, an amendment returns ``applied(...)``, and the
# refusal branch below was never reached. A deployment composing a
# defaults-injector alongside a tenant guard therefore had the guard
# silently disabled on every call the injector touched -- which is every
# call, because that is what a defaults-injector is for. Nothing raised,
# nothing was logged, and the decision read ``applied``.
#
# Deliberately stricter than the other phases: a dropped egress annotation
# loses a diagnostic; a dropped effect decision performs the harm.
if gov_phase == "effect":
objected = objection_in(outcome)
if objected is not None:
reason = objected.reason or objected.objection
report(
decisions,
state="refused",
phase=gov_phase,
records=(objected,),
reason=reason,
calls=calls,
)
raise GovernanceRefused(
stage=objected.stage,
phase=gov_phase,
disposition=objected.disposition,
reason=reason,
rule_ids=rules_in(objected, kind=OBJECTION),
)
# Nothing objected, so an amendment is a repair to deliver. A stage that
# rewrote the calls -- scrubbed a credential out of an argument, injected
# a default -- has fixed them, and PRE_TOOL has a contract for handing
# that back.
amendment = amendment_for(gov_phase, outcome, context)
if amendment is not None:
amended, _stages = amendment
acting = acting_records(outcome)
rules = tuple(dict.fromkeys(r for a in acting for r in rules_in(a)))
report(
decisions,
state="applied",
phase=gov_phase,
records=acting,
reason="the call's arguments were amended before dispatch",
calls=amended,
)
return applied(
ResolvedInput(
capability=GOVERNANCE_CAPABILITY,
value=amended,
# Rule names, never a value: provenance travels with the turn
# and would carry a scrubbed credential into every trace that
# reads it.
provenance=f"amended by {', '.join(rules) or 'governance'}",
)
)
blocked = _undeliverable(outcome)
# Egress has no amendment carrier. Returning the original draft after a
# working critic requested revision tells the caller it was reviewed while
# leaking the exact draft the critic rejected. This is deliberately not
# the reflector-outage policy: exceptions still become the metacognition
# stage's declared fail-open outcome above. A normal STEER is evidence a
# configured policy worked, so the only truthful public result is a typed
# refusal until an amendment payload exists.
if gov_phase == "egress" and not blocked:
steering = next(
(
record
for record in outcome.trace
if record.disposition in (Disposition.ANNOTATE, Disposition.STEER)
),
None,
)
if steering is not None:
report(
decisions,
state="refused",
phase=gov_phase,
records=(steering,),
reason=steering.reason or steering.objection,
calls=calls,
)
raise GovernanceRefused(
stage=steering.stage,
phase=gov_phase,
disposition=steering.disposition,
reason=steering.reason or steering.objection,
rule_ids=rules_in(steering, kind=OBJECTION),
)
if blocked:
stranded = tuple(r for r in acting_records(outcome) if r.stage == blocked)
report(
decisions,
state="refused",
phase=gov_phase,
records=stranded,
reason=("the rule amended the turn and this rung has no seam to carry the amendment"),
calls=calls,
)
raise GovernanceRefused(
stage=blocked,
phase=gov_phase,
disposition=Disposition.ANNOTATE,
rule_ids=tuple(r for s in stranded for r in rules_in(s)),
reason=(
"the stage amended the turn and the kernel has no seam to carry "
"the amendment, so continuing would use the value it was meant "
"to change"
),
)
ran = ", ".join(_stage_trace_name(record) for record in outcome.trace) or "none"
# A fail-open stage that asked for a change it cannot make is allowed
# through -- its declared failure mode says proceeding without it is
# acceptable -- but it is never allowed through *quietly*. A reflection
# that requested a revision and produced no revision is the difference
# between "the critic approved this" and "the critic objected and nobody
# was listening", and only a diagnostic keeps those apart.
dropped = tuple(
f"{record.stage} asked to {record.disposition.value} and the kernel has no seam to apply it"
for record in outcome.trace
if record.disposition in (Disposition.ANNOTATE, Disposition.STEER)
)
# Stage records retain their reasons for private decision tracing, but
# kernel stage events cross the public Agent boundary. In particular,
# fail-open exceptions can include backend messages or deployment data.
# Expose only a fixed category here.
degraded = tuple(
_public_degradation_name(record) for record in outcome.trace if record.degraded
)
report(
decisions,
state="discarded",
phase=gov_phase,
records=acting_records(outcome),
reason=(
"the rule asked for a change this rung has no seam to apply, and "
"its declared failure mode allowed the turn to continue"
),
calls=calls,
)
if gov_phase == "ingress" and governance_context.intent is not None:
return applied(
ResolvedInput(
capability=GOVERNANCE_CAPABILITY,
value=governance_context.intent,
provenance="governance.intent_filter",
),
diagnostics=dropped,
)
suffix = f"; {'; '.join(degraded)}" if degraded else ""
return no_change(
f"governance allowed this turn ({gov_phase}: {ran}){suffix}",
diagnostics=dropped + degraded,
)