fix(gateway): resolve adversarial-review findings on the pr_reviewer flow

An adversarial review of the feature found two blocking defects (both would
surface the moment external_pr_enabled is turned on) plus hardening gaps:

- HIGH: the enforcement legacy role-gate overlay OVERWROTE spec-derived roles,
  so pr_reviewer was erased from the (in_progress->completed) edge it shares
  with the PM self-complete gate — the review task could never complete. Fix:
  UNION legacy + spec roles instead of overwriting (also preserves the legacy
  'add roles' intent on every shared edge).
- HIGH: claim_pr_review routed claim+start through the verb runner, which hit
  start()'s plan gate (planless review task -> None -> crash/respawn loop) and
  auto-created+pushed a stray branch (violating the read-only/branchless
  invariant). Fix: mirror QA's claim_review — a verb-body TaskService.pr_review_claim
  does pending->in_progress with no plan and no branch.
- MED: add the pr_reviewer Write(*)/Edit(*) deny at the permission layer (it
  ingests untrusted PR diffs — make read-only explicit, not implicit).
- MED: regenerate the verb-table artifacts (the schemas existed but the
  generator had not been re-run; the agent prompt showed 'unknown' signatures).

ruff + mypy clean (279 files); foundation + gateway suites green (5205 passed).
This commit is contained in:
Renn F
2026-06-16 11:48:24 +02:00
parent 831321f436
commit 5bea82dbbc
6 changed files with 72 additions and 20 deletions
+2 -2
View File
@@ -7,10 +7,10 @@
| Verb | Body schema |
|------|-------------|
| `claim_pr_review` | `claim_pr_review(unknown — no Pydantic schema)` |
| `claim_pr_review` | `claim_pr_review(task_id: UUID)` |
| `give_me_work` | `give_me_work()` |
| `i_am_idle` | `i_am_idle()` |
| `post_pr_review` | `post_pr_review(unknown — no Pydantic schema)` |
| `post_pr_review` | `post_pr_review(task_id: UUID, body: str, event: str = 'REQUEST_CHANGES')` |
### Content (do) tools
+2 -2
View File
@@ -255,10 +255,10 @@ real tools live in their agent_sdk drivers, not role_config.
| Verb | Body schema |
|------|-------------|
| `claim_pr_review` | `claim_pr_review(unknown — no Pydantic schema)` |
| `claim_pr_review` | `claim_pr_review(task_id: UUID)` |
| `give_me_work` | `give_me_work()` |
| `i_am_idle` | `i_am_idle()` |
| `post_pr_review` | `post_pr_review(unknown — no Pydantic schema)` |
| `post_pr_review` | `post_pr_review(task_id: UUID, body: str, event: str = 'REQUEST_CHANGES')` |
### Content (do) tools
+7 -1
View File
@@ -162,7 +162,13 @@ def _build_role_restricted_transitions() -> dict[tuple[str, str], tuple[str, ...
if t.role_constraint is not None
}
for (src, tgt), roles in _LEGACY_ROLE_GATES.items():
out[(src.value, tgt.value)] = roles
# UNION (not overwrite): legacy gates ADD operational roles to an edge;
# overwriting silently dropped any spec-derived roles that share the
# same edge. pr_review_done puts pr_reviewer on (in_progress, completed)
# — the same edge the legacy PM-self-complete gate pins — so an
# overwrite erased pr_reviewer and the review task could never complete.
existing = out.get((src.value, tgt.value), ())
out[(src.value, tgt.value)] = tuple(sorted(set(existing) | set(roles)))
return out
+12
View File
@@ -909,6 +909,18 @@ class AgentOrchestrator:
"Edit(*)",
],
},
"pr_reviewer": {
# PR reviewer reads untrusted external/fork PR diffs and posts a
# change-request via the gateway — it never writes files. Make the
# read-only invariant explicit at the permission layer (it is the
# highest-value prompt-injection target), not just implicit in the
# absence of a writable mount.
"allow": [],
"deny": [
"Write(*)",
"Edit(*)",
],
},
}
if role not in configs:
@@ -88,19 +88,24 @@ class PRReviewerMixin(_Base):
task_id=task_id,
verb="claim_pr_review",
)
runner = self._verb_runner()
try:
t = await runner.run_intent("claim_pr_review", t, agent, spec_ctx)
except Exception as exc:
return await self._runner_failure(
exc,
t,
role_str,
briefing,
reviewer_agent_id,
task_id,
"claim_pr_review",
# Verb body owns the claim (mirrors QA's claim_review): a specialized
# pending->in_progress claim with NO plan and NO branch. The spec's
# composes=("claim","start") is for the gate above only — routing it
# through the verb runner would hit start()'s plan gate and auto-create
# a branch, neither of which a read-only review task wants.
claimed = await self.task.pr_review_claim(reviewer_agent_id, task_id)
if claimed is None:
return await self._emit_rejection(
Envelope.invalid_state(
message="this external-PR review task is no longer claimable",
remediate="it may already be claimed; give_me_work for the next",
context_briefing=briefing,
).with_introspection(task=t, role=role_str),
agent_id=reviewer_agent_id,
task_id=task_id,
verb="claim_pr_review",
)
t = claimed
evidence = await self._build_pr_review_evidence(t)
return Envelope.ok(
status=str(t.status),
@@ -133,9 +138,7 @@ class PRReviewerMixin(_Base):
task_id=task_id,
verb="post_pr_review",
)
pre = await self._post_pr_review_preflight(
t, reviewer_agent_id, task_id, body
)
pre = await self._post_pr_review_preflight(t, reviewer_agent_id, task_id, body)
if isinstance(pre, Envelope):
return pre
agent, role_str, briefing, spec_ctx = pre
+31
View File
@@ -675,6 +675,37 @@ class TaskService(BaseService):
await self.session.flush()
return task
async def pr_review_claim(
self, reviewer_agent_id: UUID, task_id: UUID
) -> TaskTable | None:
"""Claim an external-PR review task: pending -> in_progress, no plan, no branch.
The review is read-only and does no git of its own, so it must NOT route
through claim()/start() those would require a plan (start() returns
None for a planless task) and auto-create + push a branch. Mirrors
qa_claim's specialized-claim pattern (the verb body owns dispatch instead
of the generic verb runner). The ``is_external_review`` branch-gate
exemption (from source='external_pr') keeps claimed->in_progress valid
with no branch. Returns None if the task is not PENDING (already taken).
"""
task = await self.get(task_id)
if task is None or task.status != TaskStatus.PENDING:
return None
task.assigned_to = cast("Any", reviewer_agent_id)
task.claimed_by = cast("Any", reviewer_agent_id)
self._validate_and_set_status(
task, TaskStatus.CLAIMED, "pr_reviewer", audit_agent_id=reviewer_agent_id
)
self._validate_and_set_status(
task,
TaskStatus.IN_PROGRESS,
"pr_reviewer",
audit_agent_id=reviewer_agent_id,
)
await self.session.flush()
self.log.info("External PR review claimed", task_id=str(task_id))
return task
async def complete_review(
self, reviewer_agent_id: UUID, task_id: UUID, notes: str | None = None
) -> TaskTable | None: