workflow_mcp.apply_workflow re-checks POLICY at apply time but never re-validates the composer's declared capabilities against

workflow_mcp.apply_workflow re-checks POLICY at apply time but never re-validates the composer's declared capabilities against the fresh blast radius (v1.3.1)

Reproduction steps:

1. An AI/agent composes a dynamic workflow over MCP (workflow_mcp.py, cap-off

#16 MCP dynamic workflows) and stages it with a declared capability scope:
wf = {"id":"notify_wf","nodes":[{"id":"send","type":"effect",
"provider":"twilio","verb":"sms_send","args":{...}}],
"capabilities":{"providers":["twilio"],"max_spend_cents":1000,
"egress_hosts":["api.twilio.com"],
"allow_irreversible":False}}
stage_workflow(wf, ws=ws, signing=signing, policy_gate=policy_gate)
At stage time, WE.plan_workflow() estimates the blast radius; if it is
within the declared capabilities, _check_capabilities() passes and the
plan is signed + persisted with a one-time consent_token. Nothing external
has fired yet — this is the documented guarantee ("NOTHING has fired").

1. The consent_token has NO expiry (only an atomic os.rename claim, no

time-based check) — stage and apply can be arbitrarily far apart while the
station's installed integrations / policy config changes underneath (a new
module install, an integration's action_class changing, a policy edit).

1. apply_workflow(token, ...) explicitly re-runs WE.plan_workflow() on the

same staged workflow spec — "Policy may have changed since staging; re-plan
and refuse if anything now blocks" — and inspects the fresh blast radius
(rb) but ONLY for rb.get("requires") == "block" / per-node policy
blocks. It never calls _check_capabilities(wf.get("capabilities"), rb).
Confirmed by grep: _check_capabilities is defined once (line 137) and
called exactly once, inside stage_workflow (line 202) — zero call sites
in apply_workflow.

1. Run repro_mcp_workflow_capability_not_rechecked_at_apply_v131.py against a

clean v1.3.1 extraction. It drives the REAL stage_workflow/apply_workflow
unmodified, injecting a plan_workflow double whose first call (staging)
returns a blast radius within the declared capability and whose second
call (the apply-time recheck) returns one that violates every declared
field (extra provider, 500x the spend ceiling, an extra egress host, a
now-irreversible action) with policy still "auto_approve". Output:
stage ok: True | blast_radius: {providers:[twilio], spend:500, ...}
apply ok: True outcome: COMPLETED
plan_workflow call count: 2
CONFIRMED — apply_workflow executed a workflow whose apply-time recheck
blast radius violates every declared capability.

Expected:
stage_workflow's own comment states the guarantee plainly: "the airlock
computes the REAL blast radius and refuses to stage if the declaration is
exceeded ... the workflow provably ran within a scope the composer committed
to up front — not just whatever its nodes happened to reach." Since
apply_workflow already re-plans for the (weaker) policy guarantee, the same
re-plan result must also be checked against the capability declaration before
the run proceeds — a capability ceiling checked once, at a point that can
predate the run by an unbounded amount of time, is not a ceiling on the run.

Actual:
_check_capabilities() is called exactly once, inside stage_workflow(), against
a plan-time ESTIMATE. apply_workflow() computes a second, fresh blast radius
via the identical WE.plan_workflow() call and uses it ONLY for the policy-block
check, discarding it without ever calling _check_capabilities() again. Of the
four capability fields, egress_hosts is the starkest case: grepping
workflow_engine.py confirms "egress_domains"/"egress_hosts" appears ONLY in
blast-radius computation (line ~335/504) — it is never read anywhere in
run_workflow() as an enforcement check. The ONLY code in the whole platform
that ever validates a workflow's declared egress ceiling is
workflow_mcp._check_capabilities(), and it runs once, at stage time.
(max_spend_cents and providers/allow_irreversible do have a SEPARATE,
independent engine-level backstop inside run_workflow() itself — cumulative-
spend enforcement, and the kind=="effect"-gated _cap_providers/
_cap_no_irreversible check — so those two have a degree of defense-in-depth
the missing re-check alone doesn't fully undo; egress_hosts has none at all.)

Suggested fix:
Re-run the SAME capability check apply_workflow already has the inputs for,
using the fresh recheck blast radius, right alongside the existing policy-block
check:

recheck = WE.plan_workflow(staged["workflow"], signing=signing,
policy_gate=policy_gate)
rb = recheck.get("blast_radius") or {}
if rb.get("requires") == "block" or any(...):
keep_consumed = True
return {"ok": False, "error": "blocked by policy at apply time"}
cap_err = _check_capabilities(staged["workflow"].get("capabilities"), rb)
if cap_err:
keep_consumed = True
return {"ok": False, "error": "capability exceeded at apply time: %s" % cap_err,
"blast_radius": rb}

Separately, give the consent_token a bounded lifetime (staged["created"] is
already recorded) so a capability declaration validated against a stale
registry/policy snapshot cannot be exercised arbitrarily long after the
composer approved it. And thread egress_hosts into run_workflow() itself
(mirroring how max_spend_cents and providers already got an engine-level
backstop) so it has independent enforcement rather than relying solely on this
one MCP-specific staging check.