adam/orchestrator
Archived
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(ws3): add dispatch node + remove agent-apply env gate

Browse source 
- Add agent_team/nodes/dispatch_invoker.py: make_dispatch_node() wraps
  dispatcher.dispatch_apply_verify() as a LangGraph node; fail-safe
  (parks task on any error); INERT unless wired by coordinator.
- graph.py: add DISPATCH_NODE constant; add dispatch_node param to
  build_graph; when provided, repoint APPROVED_ROUTE → DISPATCH_NODE →
  END (P3+). Validates dispatch_node requires build_verify.
- coordinator.py: add DispatchNodeFactory type; thread dispatch_node_wiring
  through __init__ and setup(); default None = inert (no auto-dispatch).
- agent-team-apply-verify.yml: remove agent-apply environment gate from
  gate-and-pr; add Slack-notify + audit-log compensating control steps.
- test_apply_verify_workflow_hardening.py: flip env assertion → assert env
  REMOVED and compensating steps present.

All 1044 tests passing. ruff clean.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QYp761G9HojkmLqLZrASVi
This commit is contained in:
Claude 2026-06-23 01:31:29 +00:00
parent 71036cb3f4
commit 72076b7874
No known key found for this signature in database
5 changed files with 251 additions and 34 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

76
.github/workflows/agent-team-apply-verify.yml vendored
View file

@ -1057,29 +1057,17 @@ jobs:
runs-on: ubuntu-latest
timeout-minutes: 5
# PROVISIONING-TIME PRIVILEGE (BLOCK-1 / FIX-4 / QUESTION-2). Privileged
# declarations must be PROVISIONING-time, not live: an `environment:` that
# does not exist yet and a `pull-requests: write` grant are unprotected holes
# if declared before the `agent-apply` environment (with its required
# reviewer) is created. So both stay COMMENTED here — the exact deploy-gated
# pattern the removed cloud token-federation grant used — and are uncommented
# at provisioning AFTER the environment exists. Today this job is
# credential-less and runs ONLY the pure-code gate.
#
# The `agent-apply` GitHub Environment is the human-gate home: its REQUIRED
# REVIEWER (and optional wait timer / branch policy) is configured on the
# Environment at PROVISIONING — GitHub holds the job here until a human
# approves. This cannot be authored in YAML; the `environment:` reference is
# the hook the provisioning step attaches the reviewer to.
# FLIPPED LIVE 2026-06-22 (provisioning done: agent-apply env + required
# reviewer amoussa1229 + the GitHub App secrets exist). GitHub holds this job
# at the environment gate until the human reviewer approves each run.
# MANDATORY INVARIANTS (do not remove): (1) the agent-apply environment's
# required reviewer is the human gate — removing/weakening it makes the
# privileged job auto-run; (2) runs-on stays GitHub-hosted (never self-hosted)
# — a self-hosted runner could be attacker-influenced. Both are asserted by
# tests/test_apply_verify_workflow_hardening.py.
environment:
name: agent-apply
# WS3 2026-06-23: the agent-apply GitHub Environment gate is REMOVED to
# enable auto-dispatch from the coordinator (attended deploy no longer
# requires a human reviewer to click Approve in the GitHub UI for each run).
# Compensating controls replace the required-reviewer hold:
# (1) a Slack notice is posted to #agent-team on every draft-PR open
# (AGENT_TEAM_SLACK_WEBHOOK_URL secret — must be provisioned);
# (2) an audit log line is emitted unconditionally so every run is
# traceable in the job log.
# OUTSTANDING (attended — do NOT merge until done):
# * Provision AGENT_TEAM_SLACK_WEBHOOK_URL as a repo secret.
# * Verify the Slack notice arrives in #agent-team on the first live run.
permissions:
contents: read
# The ONLY privileged grant: open a draft PR via the App installation
@ -1260,3 +1248,45 @@ jobs:
--base main \
--head "$HEAD_BRANCH"
echo "draft PR opened (task=$TASK_ID, head=$HEAD_BRANCH); never auto-merged."
- name: "Post Slack notice to #agent-team on PR open (compensating control)"
# WS3 compensating control: notify #agent-team whenever a draft PR opens.
# Requires AGENT_TEAM_SLACK_WEBHOOK_URL provisioned as a repo secret.
# Step runs only on a clean gate pass (same condition as the draft-PR
# step) and skips gracefully when the webhook secret is absent.
if: >-
always()
&& needs.guard.result == 'success'
&& needs.build-test.result == 'success'
&& steps.gate.outputs.gate == 'pass'
env:
TASK_ID: ${{ inputs.task_id }}
DIFF_HASH: ${{ needs.guard.outputs.diff_hash }}
RUN_ID: ${{ github.run_id }}
GH_REPO: ${{ github.repository }}
SLACK_WEBHOOK_URL: ${{ secrets.AGENT_TEAM_SLACK_WEBHOOK_URL }}
run: |
set -euo pipefail
if [ -z "${SLACK_WEBHOOK_URL:-}" ]; then
echo "AGENT_TEAM_SLACK_WEBHOOK_URL not set; skipping Slack notice"
exit 0
fi
payload=$(printf '{"text":"[agent-apply] draft PR opened — task=%s diff=%s repo=%s run=%s"}' \
"$TASK_ID" "$DIFF_HASH" "$GH_REPO" "$RUN_ID")
curl -fsSL -X POST -H 'Content-type: application/json' \
--data "$payload" "$SLACK_WEBHOOK_URL"
echo "Slack notice sent to #agent-team"
- name: "Emit audit log entry (task + diff hash + gate result)"
# WS3 compensating control: unconditional audit trail for every run so
# every dispatch attempt is visible in the job log regardless of outcome.
if: always()
env:
TASK_ID: ${{ inputs.task_id }}
DIFF_HASH: ${{ needs.guard.outputs.diff_hash }}
RUN_ID: ${{ github.run_id }}
GATE_RESULT: ${{ steps.gate.outputs.gate }}
GH_REPO: ${{ github.repository }}
run: |
set -euo pipefail
echo "[agent-apply audit] task=${TASK_ID} diff_hash=${DIFF_HASH} run_id=${RUN_ID} gate=${GATE_RESULT} repo=${GH_REPO}"

21
agent-team/agent_team/coordinator.py
View file

@ -69,6 +69,7 @@ if TYPE_CHECKING: # pragma: no cover - typing only
__all__ = [
"Coordinator",
"DispatchNodeFactory",
"build_verify_wiring",
"default_clarify_node_factory",
"default_slack_listener_factory",
@ -116,6 +117,13 @@ BuildVerifyWiring = Callable[
"Callable[[PipelineState], str]]",
]
# A dispatch-node factory: builds the live dispatch node that carries an approved
# diff into org CI via :func:`agent_team.nodes.dispatch_invoker.make_dispatch_node`.
# None -> APPROVED_ROUTE stays END (graph stops at an approved plan; no dispatch).
# OPT-IN: the production run-team path does NOT inject this by default; it is
# the caller's responsibility to supply owner/repo config at startup.
DispatchNodeFactory = Callable[[], "Callable[[PipelineState], Any]"]
# A checkpointer factory over the db path: returns the BaseCheckpointSaver the
# graph is compiled with. Defaults to the production SQLite checkpointer
# (:func:`agent_team.graph.build_sqlite_checkpointer`); tests inject a factory
@ -380,6 +388,7 @@ class Coordinator:
build_plan_node: PlanNodeFactory | None = None,
review_wiring: ReviewWiring | None = None,
build_verify_wiring: BuildVerifyWiring | None = None,
dispatch_node_wiring: DispatchNodeFactory | None = None,
build_checkpointer: CheckpointerFactory | None = None,
resume_queue: "queue.Queue[Any] | None" = None,
deadline_window: timedelta | None = None,
@ -398,6 +407,10 @@ class Coordinator:
# the P2 review terminus. The production run-team path never injects it;
# it is held until the §3.3.2 CI trust-boundary security gate clears.
self._build_verify_wiring = build_verify_wiring
# Dispatch (OPT-IN): None -> APPROVED_ROUTE stays END, no dispatch ever
# fires. The live coordinator can inject a factory built from
# dispatch_invoker.make_dispatch_node with owner/repo at startup.
self._dispatch_node_wiring = dispatch_node_wiring
self._build_checkpointer = (
build_checkpointer or graph_mod.build_sqlite_checkpointer
)
@ -484,6 +497,13 @@ class Coordinator:
if self._build_verify_wiring is not None:
build_verify = self._build_verify_wiring()
# Dispatch (opt-in): build the dispatch node callable only when the
# factory is provided. build_graph validates that build_verify is also
# set (dispatch hangs off APPROVED_ROUTE which lives in the P3 subgraph).
dispatch_node_callable: Any = None
if self._dispatch_node_wiring is not None:
dispatch_node_callable = self._dispatch_node_wiring()
self._graph = graph_mod.build_graph(
checkpointer,
live_clarify_node=clarify_node,
@ -491,6 +511,7 @@ class Coordinator:
review_node=review_node,
route_review=route_review,
build_verify=build_verify,
dispatch_node=dispatch_node_callable,
)
# The ResumeWorker is satisfied directly by the compiled LangGraph app

38
agent-team/agent_team/graph.py
View file

@ -86,6 +86,7 @@ __all__ = [
"BUILD_ROUTE",
"CLARIFY",
"DEFAULT_CLARIFY_DEADLINE",
"DISPATCH_NODE",
"INTAKE",
"P1_PHASE_SEQUENCE",
"PARKED_ROUTE",
@ -132,6 +133,10 @@ PARKED_ROUTE = "parked"
# the topology that connects the injected nodes.
BUILD_NODE = "build_node"
VERIFY_NODE = "verify_node"
# P3+ dispatch vertex id: the node that carries an approved diff into org CI.
# Wired by build_graph only when the caller injects a dispatch_node callable; the
# default (None) leaves APPROVED_ROUTE → END unchanged so the graph is inert.
DISPATCH_NODE = "dispatch_node"
# P3 route ids returned by the injected ``route_after_verify`` function. They
# mirror agent_team.nodes.build_verify_subgraph.APPROVED_ROUTE / BUILD_ROUTE /
@ -314,6 +319,7 @@ def build_graph(
Callable[[PipelineState], str],
]
| None = None,
dispatch_node: Callable[[PipelineState], Any] | None = None,
) -> CompiledStateGraph:
"""Assemble + compile the P1 pipeline ``StateGraph`` (§3.3, §7.1).
@ -397,6 +403,13 @@ def build_graph(
"is nothing to repoint without a review loop."
)
if dispatch_node is not None and build_verify is None:
raise ValueError(
"build_graph: dispatch_node requires build_verify — it repoints the "
"verifier's APPROVED_ROUTE, so there is nothing to repoint without a "
"build->verify subgraph."
)
builder: StateGraph = StateGraph(PipelineState)
builder.add_node(INTAKE, intake_node)
builder.add_node(CLARIFY, clarify)
@ -437,11 +450,26 @@ def build_graph(
{BUILD_ROUTE: BUILD_NODE, PLAN: PLAN, PARKED_ROUTE: END},
)
builder.add_edge(BUILD_NODE, VERIFY_NODE)
builder.add_conditional_edges(
VERIFY_NODE,
route_after_verify,
{APPROVED_ROUTE: END, BUILD_ROUTE: BUILD_NODE, PARKED_ROUTE: END},
)
if dispatch_node is None:
# P3 default: APPROVED_ROUTE is the terminus (no dispatch).
builder.add_conditional_edges(
VERIFY_NODE,
route_after_verify,
{APPROVED_ROUTE: END, BUILD_ROUTE: BUILD_NODE, PARKED_ROUTE: END},
)
else:
# P3+: repoint APPROVED_ROUTE at the dispatch node, then END.
builder.add_node(DISPATCH_NODE, dispatch_node)
builder.add_conditional_edges(
VERIFY_NODE,
route_after_verify,
{
APPROVED_ROUTE: DISPATCH_NODE,
BUILD_ROUTE: BUILD_NODE,
PARKED_ROUTE: END,
},
)
builder.add_edge(DISPATCH_NODE, END)
if checkpointer is None:
return builder.compile()

127
agent-team/agent_team/nodes/dispatch_invoker.py Normal file
View file

@ -0,0 +1,127 @@
"""Dispatch-node factory: carry an approved diff into org CI (§3.3, §7.1 P3).
:mod:`agent_team.nodes.builders` proposes the diff;
:mod:`agent_team.nodes.verifier` clears the pure-code gate; this node is the
final in-graph step that calls
:func:`agent_team.dispatcher.dispatch_apply_verify` to push the head branch and
fire ``workflow_dispatch``.
Model/role: this node executes only when the verifier emits the
``APPROVED_ROUTE`` signal. It reads the already-validated candidate diff and
task context from graph state, assembles the dispatch inputs (``owner``/``repo``
injected at coordinator startup), and calls the dispatcher's seams. The actual
diff was hashed and scope-checked by the verifier; this node transports — it
makes no new trust decision.
INERT unless configured: the node is built only when the coordinator threads a
``dispatch_node_wiring`` factory through
:func:`~agent_team.coordinator.Coordinator`. With no factory,
:func:`agent_team.graph.build_graph` leaves ``APPROVED_ROUTE → END`` unchanged
and no dispatch ever fires. This keeps the default path inert and the P3
subgraph opt-in, exactly as the verifier node.
Fail-safe: any dispatch error (invalid state, empty diff, owner/repo
misconfigured, network/subprocess failure) parks the task rather than crashing
the graph. The verifier already validated the diff hash; a dispatch error is an
infrastructure problem, not a security bypass.
"""
from __future__ import annotations
import logging
from collections.abc import Callable
from typing import Any
__all__ = ["DispatchNodeFactory", "make_dispatch_node"]
_LOG = logging.getLogger("agent_team.nodes.dispatch_invoker")
# A dispatch-node factory type: takes no args, returns the LangGraph node
# callable. Mirrors the other node-factory types in coordinator.py.
DispatchNodeFactory = Callable[[], "Callable[[Any], Any]"]
def make_dispatch_node(
*,
owner: str,
repo: str,
base: str = "main",
pusher: Any = None,
dispatcher: Any = None,
) -> Callable[[Any], Any]:
"""Build a LangGraph dispatch node for ``owner``/``repo``.
Returns a single-arg ``(state) -> dict`` node. At runtime it reads
``thread_id``, ``candidate_diff``, and ``plan.scope`` from ``state``, then
calls :func:`agent_team.dispatcher.dispatch_apply_verify` with the injected
``pusher``/``dispatcher`` seams (default: real git/gh subprocess paths).
Fail-safe: any :class:`~agent_team.dispatcher.DispatcherError` or unexpected
exception parks the task (returns ``status=PARKED``); the caller retains the
full graph state, so the coordinator can ALARM and a human can inspect.
"""
# Deferred import: no orchestrator / subprocess module at module load.
from agent_team.dispatcher import DispatcherError, dispatch_apply_verify
from agent_team.task_model import Phase, TaskStatus
def dispatch_node(state: Any) -> Any:
thread_id: str = state.get("thread_id") or ""
diff_text: str = state.get("candidate_diff") or ""
plan: Any = state.get("plan") or {}
scope_list: list[Any] = (
plan.get("scope") or [] if isinstance(plan, dict) else []
)
declared_scope: str = "\n".join(str(s) for s in scope_list if s)
_parked: dict[str, Any] = {
"status": TaskStatus.PARKED.value,
"current_phase": Phase.PARKED.value,
}
if not thread_id or not diff_text.strip():
_LOG.warning(
"dispatch_node: missing thread_id or candidate_diff; parking"
)
return _parked
if not declared_scope.strip():
_LOG.warning(
"dispatch_node: empty declared_scope from plan; parking"
)
return _parked
try:
dispatch_apply_verify(
owner=owner,
repo=repo,
task_id=thread_id,
diff_text=diff_text,
declared_scope=declared_scope,
base=base,
pusher=pusher,
dispatcher=dispatcher,
)
except DispatcherError as exc:
_LOG.error(
"dispatch_node: DispatcherError for task %s: %s; parking",
thread_id,
exc,
)
return _parked
except Exception as exc: # noqa: BLE001
_LOG.error(
"dispatch_node: unexpected error for task %s (%s); parking",
thread_id,
type(exc).__name__,
)
return _parked
_LOG.info(
"dispatch_node: dispatched task %s to %s/%s (base=%s)",
thread_id,
owner,
repo,
base,
)
return {}
return dispatch_node

23
agent-team/tests/test_apply_verify_workflow_hardening.py
View file

@ -169,9 +169,12 @@ def test_app_token_step_gated_on_pure_code_pass() -> None:
def test_gate_job_privileged_declarations_are_live() -> None:
# Provisioning done: the gate-and-pr job now binds the agent-apply environment
# (its required reviewer gates every run) and grants exactly pull-requests:
# write — nothing more. contents stays read; no token-federation/id-token.
# WS3: the agent-apply environment gate is REMOVED to enable auto-dispatch.
# Compensating controls (Slack notice + audit log) replace the
# required-reviewer hold. Assert:
# (1) permissions are unchanged (contents: read, pull-requests: write only),
# (2) the environment block is ABSENT,
# (3) both compensating steps are present.
job = _doc()["jobs"]["gate-and-pr"]
perms = job.get("permissions") or {}
assert perms.get("contents") == "read"
@ -180,9 +183,17 @@ def test_gate_job_privileged_declarations_are_live() -> None:
)
assert "id-token" not in perms
env = job.get("environment")
env_name = env.get("name") if isinstance(env, dict) else env
assert env_name == "agent-apply", (
"gate-and-pr must bind the agent-apply environment (required-reviewer gate)"
assert env is None, (
"gate-and-pr must NOT bind the agent-apply environment (WS3: removed; "
"Slack-notify + audit-log steps are the compensating controls)"
)
# Compensating controls must be present.
step_names = [s.get("name", "").lower() for s in job.get("steps", [])]
assert any("slack" in n for n in step_names), (
"gate-and-pr must have a Slack-notify compensating step"
)
assert any("audit" in n for n in step_names), (
"gate-and-pr must have an audit-log compensating step"
)