feat(gateway): audit-log every gate rejection

Choreographer takes an audit dep but never used it. Now every rejection
Envelope (invalid_state, not_authorized, tracing_gap, not_found) writes a
gateway.rejected row with verb + reason + missing fields. Forensic
signal for stuck flows; no behaviour change for happy path.

Adds AuditService.log_event() — generic free-form event_type write so
the gateway doesn't have to extend AuditEventType for every new surface.
Audit writes are best-effort: a failure logs a warning but never
propagates, since the agent's response Envelope is the contract.

roboco/services/audit.py

@@ -370,6 +370,41 @@ class AuditService(SingletonService):
             )
         )
 
+    async def log_event(
+        self,
+        *,
+        event_type: str,
+        agent_id: str | UUID | None = None,
+        task_id: str | UUID | None = None,
+        details: dict[str, Any] | None = None,
+        severity: str = "warning",
+    ) -> None:
+        """Log a generic audit event.
+
+        Free-form ``event_type`` (e.g. ``"gateway.rejected"``) lets callers
+        emit categorized signals without extending ``AuditEventType`` for
+        every new surface. The Choreographer uses this for gate-rejection
+        forensics; other layers can adopt the same shape.
+        """
+        self.log.warning(
+            "Audit event",
+            event_type=event_type,
+            agent_id=str(agent_id) if agent_id else None,
+            task_id=str(task_id) if task_id else None,
+            details=details or {},
+            timestamp=datetime.now(UTC).isoformat(),
+        )
+        await self._persist(
+            _AuditEvent(
+                event_type=event_type,
+                agent_id=agent_id,
+                target_type="task" if task_id else None,
+                target_id=task_id,
+                severity=severity,
+                details=details or {},
+            )
+        )
+
     async def log_agent_event(
         self,
         *,

roboco/services/gateway/choreographer.py

@@ -15,6 +15,8 @@ from dataclasses import dataclass
 from typing import Any
 from uuid import UUID
 
+import structlog
+
 from roboco.config import settings
 from roboco.services.gateway.claim_guards import (
     already_active_guard,
@@ -41,6 +43,8 @@ from roboco.services.gateway.tracing_gate import (
     check_requirements,
 )
 
+logger = structlog.get_logger()
+
 
 @dataclass(frozen=True)
 class ChoreographerDeps:
@@ -128,6 +132,41 @@ class Choreographer:
         if task_id is not None:
             await self.task.heartbeat(task_id)
 
+    async def _emit_rejection(
+        self,
+        env: Envelope,
+        *,
+        agent_id: UUID,
+        task_id: UUID | None,
+        verb: str,
+    ) -> Envelope:
+        """Audit-log a rejection envelope; pass through unchanged on success.
+
+        Idempotent on success envelopes: the early `env.error is None`
+        return is the only fast path. Audit writes are best-effort —
+        failures must NEVER block the verb (the agent's response is the
+        contract; the audit row is observability-only).
+        """
+        if env.error is None:
+            return env
+        try:
+            await self.audit.log_event(
+                event_type="gateway.rejected",
+                agent_id=agent_id,
+                task_id=task_id,
+                details={
+                    "verb": verb,
+                    "reason": env.error,
+                    "message": env.message,
+                    "missing": env.missing or [],
+                },
+            )
+        except Exception as exc:
+            # Audit is best-effort: it must NEVER block the verb. The agent's
+            # response is the contract; the audit row is observability-only.
+            logger.warning("audit.log_event failed", error=str(exc), verb=verb)
+        return env
+
     # --- Phase 1 (developer) verbs ---
 
     async def give_me_work(self, agent_id: UUID) -> Envelope:
@@ -250,7 +289,12 @@ class Choreographer:
         """Claim/start/recover any actionable state of agent_id's task_id."""
         t = await self.task.get(task_id)
         if t is None:
-            return Envelope.not_found(message=f"task {task_id} not found")
+            return await self._emit_rejection(
+                Envelope.not_found(message=f"task {task_id} not found"),
+                agent_id=agent_id,
+                task_id=task_id,
+                verb="i_will_work_on",
+            )
         status = str(t.status)
         briefing = await self._briefing_for(agent_id, task_id)
 
@@ -263,7 +307,12 @@ class Choreographer:
         elif status == "pending":
             # Fresh claim — run all claim-time gates BEFORE mutating state.
             if guard := await self._run_claim_guards(agent_id=agent_id, task=t):
-                return self._with_briefing(guard, briefing)
+                return await self._emit_rejection(
+                    self._with_briefing(guard, briefing),
+                    agent_id=agent_id,
+                    task_id=task_id,
+                    verb="i_will_work_on",
+                )
             if t.assigned_to is None or t.assigned_to != agent_id:
                 t = await self.task.claim(task_id, agent_id)
             if not t.plan and not plan:
@@ -271,10 +320,15 @@ class Choreographer:
                     f"call i_will_work_on(task_id='{task_id}',"
                     f" plan='<one-paragraph plan describing what you will do>')"
                 )
-                return Envelope.tracing_gap(
+                return await self._emit_rejection(
-                    missing=["plan"],
+                    Envelope.tracing_gap(
-                    remediate=remediate,
+                        missing=["plan"],
-                    context_briefing=briefing,
+                        remediate=remediate,
+                        context_briefing=briefing,
+                    ),
+                    agent_id=agent_id,
+                    task_id=task_id,
+                    verb="i_will_work_on",
                 )
             if plan:
                 t = await self.task.set_plan(task_id, plan)
@@ -286,13 +340,23 @@ class Choreographer:
                 agent_id=agent_id, task=t, skip_sequence=True
             )
             if guard:
-                return self._with_briefing(guard, briefing)
+                return await self._emit_rejection(
+                    self._with_briefing(guard, briefing),
+                    agent_id=agent_id,
+                    task_id=task_id,
+                    verb="i_will_work_on",
+                )
             t = await self.task.start(task_id, agent_id)
         else:
-            return Envelope.invalid_state(
+            return await self._emit_rejection(
-                message=f"task {task_id} is in {status}; cannot start work",
+                Envelope.invalid_state(
-                remediate="call give_me_work() to find an actionable task",
+                    message=f"task {task_id} is in {status}; cannot start work",
-                context_briefing=briefing,
+                    remediate="call give_me_work() to find an actionable task",
+                    context_briefing=briefing,
+                ),
+                agent_id=agent_id,
+                task_id=task_id,
+                verb="i_will_work_on",
             )
 
         await self._touch(task_id)
@@ -316,20 +380,32 @@ class Choreographer:
         """Record that the dev made a commit; auto-creates progress entry."""
         t = await self.task.get_active_task_for_agent(agent_id)
         if t is None:
-            return Envelope.invalid_state(
+            return await self._emit_rejection(
-                message="no active task for this agent",
+                Envelope.invalid_state(
-                remediate="call give_me_work() then i_will_work_on(task_id, plan)",
+                    message="no active task for this agent",
-                context_briefing=await self._briefing_for(agent_id, None),
+                    remediate=(
+                        "call give_me_work() then i_will_work_on(task_id, plan)"
+                    ),
+                    context_briefing=await self._briefing_for(agent_id, None),
+                ),
+                agent_id=agent_id,
+                task_id=None,
+                verb="i_have_committed",
             )
         if not t.plan:
             no_plan_remediate = (
                 f"plan must be set first;"
                 f" call i_will_work_on(task_id='{t.id}', plan='...')"
             )
-            return Envelope.tracing_gap(
+            return await self._emit_rejection(
-                missing=["plan"],
+                Envelope.tracing_gap(
-                remediate=no_plan_remediate,
+                    missing=["plan"],
-                context_briefing=await self._briefing_for(agent_id, t.id),
+                    remediate=no_plan_remediate,
+                    context_briefing=await self._briefing_for(agent_id, t.id),
+                ),
+                agent_id=agent_id,
+                task_id=t.id,
+                verb="i_have_committed",
             )
         await self.task.add_progress(t.id, agent_id, message)
         await self._touch(t.id)
@@ -356,22 +432,37 @@ class Choreographer:
         """
         t = await self.task.get(task_id)
         if t is None:
-            return Envelope.not_found(message=f"task {task_id} not found")
+            return await self._emit_rejection(
+                Envelope.not_found(message=f"task {task_id} not found"),
+                agent_id=agent_id,
+                task_id=task_id,
+                verb="submit_for_qa",
+            )
         briefing = await self._briefing_for(agent_id, task_id)
         if t.assigned_to != agent_id:
-            return Envelope.not_authorized(
+            return await self._emit_rejection(
-                message=f"task {task_id} is not assigned to you",
+                Envelope.not_authorized(
-                remediate="call give_me_work() to find your work",
+                    message=f"task {task_id} is not assigned to you",
-                context_briefing=briefing,
+                    remediate="call give_me_work() to find your work",
+                    context_briefing=briefing,
+                ),
+                agent_id=agent_id,
+                task_id=task_id,
+                verb="submit_for_qa",
             )
         if not t.commits:
-            return Envelope.invalid_state(
+            return await self._emit_rejection(
-                message="no commits on this task yet",
+                Envelope.invalid_state(
-                remediate=(
+                    message="no commits on this task yet",
-                    "commit at least one change before submitting for QA — "
+                    remediate=(
-                    "call commit(message='<subject>')"
+                        "commit at least one change before submitting for QA — "
+                        "call commit(message='<subject>')"
+                    ),
+                    context_briefing=briefing,
                 ),
-                context_briefing=briefing,
+                agent_id=agent_id,
+                task_id=task_id,
+                verb="submit_for_qa",
             )
         if t.pr_number is not None:
             return Envelope.ok(
@@ -387,9 +478,7 @@ class Choreographer:
         await self._touch(task_id)
         await self.git.push_branch(t.branch_name)
         parent = parent_branch_for(t.branch_name)
-        pr = await self.git.create_pr(
+        pr = await self.git.create_pr(t.branch_name, parent=parent, is_root_pr=False)
-            t.branch_name, parent=parent, is_root_pr=False
-        )
 
         return Envelope.ok(
             status=str(t.status),
@@ -416,21 +505,35 @@ class Choreographer:
         """
         t = await self.task.get(task_id)
         if t is None:
-            return Envelope.not_found(message=f"task {task_id} not found")
+            return await self._emit_rejection(
+                Envelope.not_found(message=f"task {task_id} not found"),
+                agent_id=agent_id,
+                task_id=task_id,
+                verb="i_am_done",
+            )
         if t.assigned_to != agent_id:
-            return Envelope.not_authorized(
+            return await self._emit_rejection(
-                message="not assigned to you",
+                Envelope.not_authorized(
-                remediate="claim it via i_will_work_on(task_id) first",
+                    message="not assigned to you",
-                context_briefing=await self._briefing_for(agent_id, task_id),
+                    remediate="claim it via i_will_work_on(task_id) first",
+                    context_briefing=await self._briefing_for(agent_id, task_id),
+                ),
+                agent_id=agent_id,
+                task_id=task_id,
+                verb="i_am_done",
             )
 
         # 1. Tracing-gate preconditions (progress / reflect / acceptance)
         if rejection := await self._check_tracing_gates(agent_id, task_id, t):
-            return rejection
+            return await self._emit_rejection(
+                rejection, agent_id=agent_id, task_id=task_id, verb="i_am_done"
+            )
 
         # 2. Field-level gates (Gate Set E) — strict.
         if rejection := await self._check_submit_qa_field_gates(agent_id, task_id, t):
-            return rejection
+            return await self._emit_rejection(
+                rejection, agent_id=agent_id, task_id=task_id, verb="i_am_done"
+            )
 
         # 3. Submit (no catch-up).
         submitted = await self.task.submit_qa(agent_id, task_id, notes)
@@ -456,15 +559,30 @@ class Choreographer:
         """
         t = await self.task.get(task_id)
         if t is None:
-            return Envelope.not_found(message=f"task {task_id} not found")
+            return await self._emit_rejection(
+                Envelope.not_found(message=f"task {task_id} not found"),
+                agent_id=agent_id,
+                task_id=task_id,
+                verb="i_am_done_with_catchup",
+            )
         if t.assigned_to != agent_id:
-            return Envelope.not_authorized(
+            return await self._emit_rejection(
-                message="not assigned to you",
+                Envelope.not_authorized(
-                remediate="claim it via i_will_work_on(task_id) first",
+                    message="not assigned to you",
-                context_briefing=await self._briefing_for(agent_id, task_id),
+                    remediate="claim it via i_will_work_on(task_id) first",
+                    context_briefing=await self._briefing_for(agent_id, task_id),
+                ),
+                agent_id=agent_id,
+                task_id=task_id,
+                verb="i_am_done_with_catchup",
             )
         if rejection := await self._check_tracing_gates(agent_id, task_id, t):
-            return rejection
+            return await self._emit_rejection(
+                rejection,
+                agent_id=agent_id,
+                task_id=task_id,
+                verb="i_am_done_with_catchup",
+            )
         t = await self._run_catch_up(agent_id, task_id, t, notes)
         await self._notify_qa(agent_id, task_id, t)
         return await self._build_i_am_done_ok(agent_id, task_id, t)
@@ -635,7 +753,12 @@ class Choreographer:
         """Escalate task_id and write a struggle journal entry; idle the agent."""
         t = await self.task.get(task_id)
         if t is None:
-            return Envelope.not_found(message=f"task {task_id} not found")
+            return await self._emit_rejection(
+                Envelope.not_found(message=f"task {task_id} not found"),
+                agent_id=agent_id,
+                task_id=task_id,
+                verb="i_am_blocked",
+            )
         await self.journal.write_struggle(
             agent_id=agent_id, task_id=task_id, content=reason
         )
@@ -676,7 +799,9 @@ class Choreographer:
                 context_briefing=briefing,
             )
         if guard := await self._pending_assignment_guard(agent_id, briefing):
-            return guard
+            return await self._emit_rejection(
+                guard, agent_id=agent_id, task_id=None, verb="i_am_idle"
+            )
         await self._auto_pause_in_progress_tasks(agent_id)
         await self.task.mark_agent_idle(agent_id)
         return Envelope.ok(
@@ -741,14 +866,25 @@ class Choreographer:
         """
         t = await self.task.get(task_id)
         if t is None:
-            return Envelope.not_found(message=f"task {task_id} not found")
+            return await self._emit_rejection(
+                Envelope.not_found(message=f"task {task_id} not found"),
+                agent_id=qa_agent_id,
+                task_id=task_id,
+                verb="claim_review",
+            )
         if str(t.status) != "awaiting_qa":
-            return Envelope.invalid_state(
+            return await self._emit_rejection(
-                message=(
+                Envelope.invalid_state(
-                    f"task {task_id} is in {t.status}, expected awaiting_qa for review"
+                    message=(
+                        f"task {task_id} is in {t.status}, "
+                        "expected awaiting_qa for review"
+                    ),
+                    remediate="call give_me_work() to find an actionable QA task",
+                    context_briefing=await self._briefing_for(qa_agent_id, task_id),
                 ),
-                remediate="call give_me_work() to find an actionable QA task",
+                agent_id=qa_agent_id,
-                context_briefing=await self._briefing_for(qa_agent_id, task_id),
+                task_id=task_id,
+                verb="claim_review",
             )
 
         # Gate Set A: ALREADY_ACTIVE / PAUSED_TASKS_EXIST guard QA from
@@ -763,8 +899,13 @@ class Choreographer:
             skip_sequence=True,
         )
         if guard:
-            return self._with_briefing(
+            return await self._emit_rejection(
-                guard, await self._briefing_for(qa_agent_id, task_id)
+                self._with_briefing(
+                    guard, await self._briefing_for(qa_agent_id, task_id)
+                ),
+                agent_id=qa_agent_id,
+                task_id=task_id,
+                verb="claim_review",
             )
 
         t = await self.task.qa_claim(qa_agent_id, task_id)
@@ -809,12 +950,22 @@ class Choreographer:
         """
         t = await self.task.get(task_id)
         if t is None:
-            return Envelope.not_found(message=f"task {task_id} not found")
+            return await self._emit_rejection(
+                Envelope.not_found(message=f"task {task_id} not found"),
+                agent_id=qa_agent_id,
+                task_id=task_id,
+                verb="pass_review",
+            )
         if t.assigned_to != qa_agent_id:
-            return Envelope.not_authorized(
+            return await self._emit_rejection(
-                message="not assigned to you",
+                Envelope.not_authorized(
-                remediate="claim it via claim_review(task_id) first",
+                    message="not assigned to you",
-                context_briefing=await self._briefing_for(qa_agent_id, task_id),
+                    remediate="claim it via claim_review(task_id) first",
+                    context_briefing=await self._briefing_for(qa_agent_id, task_id),
+                ),
+                agent_id=qa_agent_id,
+                task_id=task_id,
+                verb="pass_review",
             )
 
         has_learning = await self.journal.has_learning_for_task(qa_agent_id, task_id)
@@ -824,8 +975,13 @@ class Choreographer:
             evidence_inspected=t.qa_evidence_inspected,
         )
         if missing:
-            return self._qa_tracing_gap(
+            return await self._emit_rejection(
-                missing, task_id, await self._briefing_for(qa_agent_id, task_id)
+                self._qa_tracing_gap(
+                    missing, task_id, await self._briefing_for(qa_agent_id, task_id)
+                ),
+                agent_id=qa_agent_id,
+                task_id=task_id,
+                verb="pass_review",
             )
 
         t = await self.task.qa_pass(qa_agent_id, task_id, notes)
@@ -892,18 +1048,33 @@ class Choreographer:
         """QA fails the task with concrete issues; transitions to needs_revision."""
         t = await self.task.get(task_id)
         if t is None:
-            return Envelope.not_found(message=f"task {task_id} not found")
+            return await self._emit_rejection(
+                Envelope.not_found(message=f"task {task_id} not found"),
+                agent_id=qa_agent_id,
+                task_id=task_id,
+                verb="fail_review",
+            )
         if t.assigned_to != qa_agent_id:
-            return Envelope.not_authorized(
+            return await self._emit_rejection(
-                message="not assigned to you",
+                Envelope.not_authorized(
-                remediate="claim it via claim_review(task_id) first",
+                    message="not assigned to you",
-                context_briefing=await self._briefing_for(qa_agent_id, task_id),
+                    remediate="claim it via claim_review(task_id) first",
+                    context_briefing=await self._briefing_for(qa_agent_id, task_id),
+                ),
+                agent_id=qa_agent_id,
+                task_id=task_id,
+                verb="fail_review",
             )
         if not issues:
-            return Envelope.invalid_state(
+            return await self._emit_rejection(
-                message="fail_review requires at least one issue",
+                Envelope.invalid_state(
-                remediate="pass issues=['<concrete actionable issue>', ...]",
+                    message="fail_review requires at least one issue",
-                context_briefing=await self._briefing_for(qa_agent_id, task_id),
+                    remediate="pass issues=['<concrete actionable issue>', ...]",
+                    context_briefing=await self._briefing_for(qa_agent_id, task_id),
+                ),
+                agent_id=qa_agent_id,
+                task_id=task_id,
+                verb="fail_review",
             )
 
         has_learning = await self.journal.has_learning_for_task(qa_agent_id, task_id)
@@ -914,8 +1085,13 @@ class Choreographer:
             evidence_inspected=t.qa_evidence_inspected,
         )
         if missing:
-            return self._qa_tracing_gap(
+            return await self._emit_rejection(
-                missing, task_id, await self._briefing_for(qa_agent_id, task_id)
+                self._qa_tracing_gap(
+                    missing, task_id, await self._briefing_for(qa_agent_id, task_id)
+                ),
+                agent_id=qa_agent_id,
+                task_id=task_id,
+                verb="fail_review",
             )
 
         t = await self.task.qa_fail(qa_agent_id, task_id, notes, issues)
@@ -941,14 +1117,25 @@ class Choreographer:
         """Documenter claims task in awaiting_documentation; returns evidence inline."""
         t = await self.task.get(task_id)
         if t is None:
-            return Envelope.not_found(message=f"task {task_id} not found")
+            return await self._emit_rejection(
+                Envelope.not_found(message=f"task {task_id} not found"),
+                agent_id=doc_agent_id,
+                task_id=task_id,
+                verb="claim_doc_task",
+            )
         if str(t.status) != "awaiting_documentation":
-            return Envelope.invalid_state(
+            return await self._emit_rejection(
-                message=(
+                Envelope.invalid_state(
-                    f"task {task_id} is in {t.status}, expected awaiting_documentation"
+                    message=(
+                        f"task {task_id} is in {t.status}, "
+                        "expected awaiting_documentation"
+                    ),
+                    remediate="call give_me_work() to find an actionable doc task",
+                    context_briefing=await self._briefing_for(doc_agent_id, task_id),
                 ),
-                remediate="call give_me_work() to find an actionable doc task",
+                agent_id=doc_agent_id,
-                context_briefing=await self._briefing_for(doc_agent_id, task_id),
+                task_id=task_id,
+                verb="claim_doc_task",
             )
 
         # Gate Set A: ALREADY_ACTIVE / PAUSED_TASKS_EXIST. The doc verb only
@@ -962,8 +1149,13 @@ class Choreographer:
             skip_sequence=True,
         )
         if guard:
-            return self._with_briefing(
+            return await self._emit_rejection(
-                guard, await self._briefing_for(doc_agent_id, task_id)
+                self._with_briefing(
+                    guard, await self._briefing_for(doc_agent_id, task_id)
+                ),
+                agent_id=doc_agent_id,
+                task_id=task_id,
+                verb="claim_doc_task",
             )
 
         t = await self.task.doc_claim(doc_agent_id, task_id)
@@ -1006,31 +1198,51 @@ class Choreographer:
         """
         t = await self.task.get(task_id)
         if t is None:
-            return Envelope.not_found(message=f"task {task_id} not found")
+            return await self._emit_rejection(
+                Envelope.not_found(message=f"task {task_id} not found"),
+                agent_id=doc_agent_id,
+                task_id=task_id,
+                verb="i_documented",
+            )
         if t.assigned_to != doc_agent_id:
-            return Envelope.not_authorized(
+            return await self._emit_rejection(
-                message="not assigned to you",
+                Envelope.not_authorized(
-                remediate="claim it via claim_doc_task(task_id) first",
+                    message="not assigned to you",
-                context_briefing=await self._briefing_for(doc_agent_id, task_id),
+                    remediate="claim it via claim_doc_task(task_id) first",
+                    context_briefing=await self._briefing_for(doc_agent_id, task_id),
+                ),
+                agent_id=doc_agent_id,
+                task_id=task_id,
+                verb="i_documented",
             )
         if not notes or len(notes) < settings.docs_notes_min_chars:
-            return Envelope.tracing_gap(
+            return await self._emit_rejection(
-                missing=["docs_notes>=20"],
+                Envelope.tracing_gap(
-                remediate=(
+                    missing=["docs_notes>=20"],
-                    "i_documented requires notes>=20 chars summarizing what you "
+                    remediate=(
-                    "documented and where (file paths)."
+                        "i_documented requires notes>=20 chars summarizing what you "
-                    " Include each file in `files=...`."
+                        "documented and where (file paths)."
+                        " Include each file in `files=...`."
+                    ),
+                    context_briefing=await self._briefing_for(doc_agent_id, task_id),
                 ),
-                context_briefing=await self._briefing_for(doc_agent_id, task_id),
+                agent_id=doc_agent_id,
+                task_id=task_id,
+                verb="i_documented",
             )
         if not files:
-            return Envelope.tracing_gap(
+            return await self._emit_rejection(
-                missing=["files"],
+                Envelope.tracing_gap(
-                remediate=(
+                    missing=["files"],
-                    "i_documented requires files=['<path>', ...]"
+                    remediate=(
-                    " listing the doc files written."
+                        "i_documented requires files=['<path>', ...]"
+                        " listing the doc files written."
+                    ),
+                    context_briefing=await self._briefing_for(doc_agent_id, task_id),
                 ),
-                context_briefing=await self._briefing_for(doc_agent_id, task_id),
+                agent_id=doc_agent_id,
+                task_id=task_id,
+                verb="i_documented",
             )
         t = await self.task.docs_complete(
             doc_agent_id, task_id, notes=notes, files=files
@@ -1107,18 +1319,33 @@ class Choreographer:
         """
         t = await self.task.get(task_id)
         if t is None:
-            return Envelope.not_found(message=f"task {task_id} not found")
+            return await self._emit_rejection(
+                Envelope.not_found(message=f"task {task_id} not found"),
+                agent_id=pm_agent_id,
+                task_id=task_id,
+                verb="i_will_plan",
+            )
         rejection = await self._i_will_plan_preflight(pm_agent_id, task_id, t, plan)
         if rejection is not None:
-            return rejection
+            return await self._emit_rejection(
+                rejection,
+                agent_id=pm_agent_id,
+                task_id=task_id,
+                verb="i_will_plan",
+            )
 
         if t.assigned_to is None or t.assigned_to != pm_agent_id:
             t = await self.task.claim(task_id, pm_agent_id)
             if t is None:
-                return Envelope.invalid_state(
+                return await self._emit_rejection(
-                    message="claim failed",
+                    Envelope.invalid_state(
-                    remediate="task may already be claimed by another agent",
+                        message="claim failed",
-                    context_briefing=await self._briefing_for(pm_agent_id, task_id),
+                        remediate="task may already be claimed by another agent",
+                        context_briefing=await self._briefing_for(pm_agent_id, task_id),
+                    ),
+                    agent_id=pm_agent_id,
+                    task_id=task_id,
+                    verb="i_will_plan",
                 )
         await self.task.set_plan(task_id, plan)
         t = await self.task.start(task_id, pm_agent_id)
@@ -1147,13 +1374,23 @@ class Choreographer:
         """
         parent = await self.task.get(parent_task_id)
         if parent is None:
-            return Envelope.not_found(message=f"task {parent_task_id} not found")
+            return await self._emit_rejection(
+                Envelope.not_found(message=f"task {parent_task_id} not found"),
+                agent_id=pm_agent_id,
+                task_id=parent_task_id,
+                verb="delegate",
+            )
         agent = await self.task.agent_for(pm_agent_id)
         guard = await self._delegate_guard(
             pm_agent_id, parent_task_id, parent, agent, inputs
         )
         if guard is not None:
-            return guard
+            return await self._emit_rejection(
+                guard,
+                agent_id=pm_agent_id,
+                task_id=parent_task_id,
+                verb="delegate",
+            )
 
         new_task = await self._create_subtask_from_inputs(
             pm_agent_id, parent_task_id, parent, inputs
@@ -1387,19 +1624,34 @@ class Choreographer:
         """
         t = await self.task.get(task_id)
         if t is None:
-            return Envelope.not_found(message=f"task {task_id} not found")
+            return await self._emit_rejection(
+                Envelope.not_found(message=f"task {task_id} not found"),
+                agent_id=pm_agent_id,
+                task_id=task_id,
+                verb="submit_up",
+            )
         guard = await self._submit_up_guard(pm_agent_id, task_id, t, notes)
         if guard is not None:
-            return guard
+            return await self._emit_rejection(
+                guard,
+                agent_id=pm_agent_id,
+                task_id=task_id,
+                verb="submit_up",
+            )
 
         parent_branch = parent_branch_for(t.branch_name)
         await self.git.create_pr(t.branch_name, parent=parent_branch, is_root_pr=False)
         t = await self.task.submit_pm_review(pm_agent_id, task_id, notes)
         if t is None:
-            return Envelope.invalid_state(
+            return await self._emit_rejection(
-                message="could not transition to awaiting_pm_review",
+                Envelope.invalid_state(
-                remediate="check task state — must be in_progress with PR ready",
+                    message="could not transition to awaiting_pm_review",
-                context_briefing=await self._briefing_for(pm_agent_id, task_id),
+                    remediate="check task state — must be in_progress with PR ready",
+                    context_briefing=await self._briefing_for(pm_agent_id, task_id),
+                ),
+                agent_id=pm_agent_id,
+                task_id=task_id,
+                verb="submit_up",
             )
         await self._handoff_to_main_pm(pm_agent_id, task_id)
         return Envelope.ok(
@@ -1602,14 +1854,24 @@ class Choreographer:
         """PM unblocks task; restore=True (default) returns to pre_block_state."""
         t = await self.task.get(task_id)
         if t is None:
-            return Envelope.not_found(message=f"task {task_id} not found")
+            return await self._emit_rejection(
+                Envelope.not_found(message=f"task {task_id} not found"),
+                agent_id=pm_agent_id,
+                task_id=task_id,
+                verb="unblock",
+            )
         if str(t.status) != "blocked":
-            return Envelope.invalid_state(
+            return await self._emit_rejection(
-                message=f"task {task_id} is in {t.status}, expected blocked",
+                Envelope.invalid_state(
-                remediate=(
+                    message=f"task {task_id} is in {t.status}, expected blocked",
-                    "this task is not blocked; call triage() to find blocked tasks"
+                    remediate=(
+                        "this task is not blocked; call triage() to find blocked tasks"
+                    ),
+                    context_briefing=await self._briefing_for(pm_agent_id, task_id),
                 ),
-                context_briefing=await self._briefing_for(pm_agent_id, task_id),
+                agent_id=pm_agent_id,
+                task_id=task_id,
+                verb="unblock",
             )
 
         has_decision = await self.journal.has_decision_for_task(pm_agent_id, task_id)
@@ -1618,10 +1880,15 @@ class Choreographer:
                 hint_for_missing_journal_decision,
             )
 
-            return Envelope.tracing_gap(
+            return await self._emit_rejection(
-                missing=["journal:decision"],
+                Envelope.tracing_gap(
-                remediate=hint_for_missing_journal_decision(),
+                    missing=["journal:decision"],
-                context_briefing=await self._briefing_for(pm_agent_id, task_id),
+                    remediate=hint_for_missing_journal_decision(),
+                    context_briefing=await self._briefing_for(pm_agent_id, task_id),
+                ),
+                agent_id=pm_agent_id,
+                task_id=task_id,
+                verb="unblock",
             )
 
         t = await self.task.unblock_with_restore(pm_agent_id, task_id, restore=restore)
@@ -1694,10 +1961,20 @@ class Choreographer:
         """Cell PM completes a task — auto-merges leaf PR into parent branch."""
         t = await self.task.get(task_id)
         if t is None:
-            return Envelope.not_found(message=f"task {task_id} not found")
+            return await self._emit_rejection(
+                Envelope.not_found(message=f"task {task_id} not found"),
+                agent_id=pm_agent_id,
+                task_id=task_id,
+                verb="cell_pm_complete",
+            )
         guard = await self._cell_pm_complete_guard(pm_agent_id, task_id, t)
         if guard is not None:
-            return guard
+            return await self._emit_rejection(
+                guard,
+                agent_id=pm_agent_id,
+                task_id=task_id,
+                verb="cell_pm_complete",
+            )
         target = parent_branch_for(t.branch_name)
         merge_result = await self.git.pr_merge(t.pr_number, target=target)
         leaf_parent_id = t.parent_task_id
@@ -1817,10 +2094,20 @@ class Choreographer:
         """Main PM completes a root task; opens master PR + escalates to CEO."""
         t = await self.task.get(root_task_id)
         if t is None:
-            return Envelope.not_found(message=f"task {root_task_id} not found")
+            return await self._emit_rejection(
+                Envelope.not_found(message=f"task {root_task_id} not found"),
+                agent_id=main_pm_agent_id,
+                task_id=root_task_id,
+                verb="main_pm_complete",
+            )
         guard = await self._main_pm_complete_guard(main_pm_agent_id, root_task_id, t)
         if guard is not None:
-            return guard
+            return await self._emit_rejection(
+                guard,
+                agent_id=main_pm_agent_id,
+                task_id=root_task_id,
+                verb="main_pm_complete",
+            )
 
         needs_pr = t.pr_number is None
         if not needs_pr:
@@ -1848,10 +2135,15 @@ class Choreographer:
             return await self.cell_pm_complete(agent_id, task_id, notes)
         if agent.role == "main_pm":
             return await self.main_pm_complete(agent_id, task_id, notes)
-        return Envelope.not_authorized(
+        return await self._emit_rejection(
-            message=f"role {agent.role} cannot complete tasks via this verb",
+            Envelope.not_authorized(
-            remediate="only cell_pm and main_pm can call complete",
+                message=f"role {agent.role} cannot complete tasks via this verb",
-            context_briefing=await self._briefing_for(agent_id, task_id),
+                remediate="only cell_pm and main_pm can call complete",
+                context_briefing=await self._briefing_for(agent_id, task_id),
+            ),
+            agent_id=agent_id,
+            task_id=task_id,
+            verb="complete",
         )
 
     async def escalate_up(
@@ -1860,7 +2152,12 @@ class Choreographer:
         """Escalate a task to the agent's escalation_target role."""
         t = await self.task.get(task_id)
         if t is None:
-            return Envelope.not_found(message=f"task {task_id} not found")
+            return await self._emit_rejection(
+                Envelope.not_found(message=f"task {task_id} not found"),
+                agent_id=pm_agent_id,
+                task_id=task_id,
+                verb="escalate_up",
+            )
 
         has_decision = await self.journal.has_decision_for_task(pm_agent_id, task_id)
         if not has_decision:
@@ -1868,33 +2165,48 @@ class Choreographer:
                 hint_for_missing_journal_decision,
             )
 
-            return Envelope.tracing_gap(
+            return await self._emit_rejection(
-                missing=["journal:decision"],
+                Envelope.tracing_gap(
-                remediate=hint_for_missing_journal_decision(),
+                    missing=["journal:decision"],
-                context_briefing=await self._briefing_for(pm_agent_id, task_id),
+                    remediate=hint_for_missing_journal_decision(),
+                    context_briefing=await self._briefing_for(pm_agent_id, task_id),
+                ),
+                agent_id=pm_agent_id,
+                task_id=task_id,
+                verb="escalate_up",
             )
 
         me = await self.task.agent_for(pm_agent_id)
         target_slug = me.escalation_target if me else None
         if not target_slug:
-            return Envelope.invalid_state(
+            return await self._emit_rejection(
-                message="no escalation target configured for your role",
+                Envelope.invalid_state(
-                remediate="check agents_config.py ESCALATION_CHAIN for your slug",
+                    message="no escalation target configured for your role",
-                context_briefing=await self._briefing_for(pm_agent_id, task_id),
+                    remediate="check agents_config.py ESCALATION_CHAIN for your slug",
+                    context_briefing=await self._briefing_for(pm_agent_id, task_id),
+                ),
+                agent_id=pm_agent_id,
+                task_id=task_id,
+                verb="escalate_up",
             )
 
         t = await self.task.escalate(pm_agent_id, task_id, reason)
         if t is None:
-            return Envelope.invalid_state(
+            return await self._emit_rejection(
-                message=(
+                Envelope.invalid_state(
-                    f"could not escalate task {task_id} to {target_slug}: "
+                    message=(
-                    "target agent not found or task missing"
+                        f"could not escalate task {task_id} to {target_slug}: "
+                        "target agent not found or task missing"
+                    ),
+                    remediate=(
+                        f"verify {target_slug} exists in agents table and that the "
+                        "task is still present"
+                    ),
+                    context_briefing=await self._briefing_for(pm_agent_id, task_id),
                 ),
-                remediate=(
+                agent_id=pm_agent_id,
-                    f"verify {target_slug} exists in agents table and that the "
+                task_id=task_id,
-                    "task is still present"
+                verb="escalate_up",
-                ),
-                context_briefing=await self._briefing_for(pm_agent_id, task_id),
             )
         return Envelope.ok(
             status=str(t.status),
@@ -1911,21 +2223,36 @@ class Choreographer:
         """Board/Main PM escalates task_id to CEO with reason."""
         t = await self.task.get(task_id)
         if t is None:
-            return Envelope.not_found(message=f"task {task_id} not found")
+            return await self._emit_rejection(
+                Envelope.not_found(message=f"task {task_id} not found"),
+                agent_id=agent_id,
+                task_id=task_id,
+                verb="escalate_to_ceo",
+            )
         me = await self.task.agent_for(agent_id)
         if me.role not in ("main_pm", "product_owner", "head_marketing"):
-            return Envelope.not_authorized(
+            return await self._emit_rejection(
-                message=f"role {me.role} cannot escalate to CEO directly",
+                Envelope.not_authorized(
-                remediate="use escalate_up() to go through your escalation chain",
+                    message=f"role {me.role} cannot escalate to CEO directly",
-                context_briefing=await self._briefing_for(agent_id, task_id),
+                    remediate="use escalate_up() to go through your escalation chain",
+                    context_briefing=await self._briefing_for(agent_id, task_id),
+                ),
+                agent_id=agent_id,
+                task_id=task_id,
+                verb="escalate_to_ceo",
             )
         if str(t.status) != "awaiting_pm_review":
-            return Envelope.invalid_state(
+            return await self._emit_rejection(
-                message=(
+                Envelope.invalid_state(
-                    f"task {task_id} is in {t.status}, expected awaiting_pm_review"
+                    message=(
+                        f"task {task_id} is in {t.status}, expected awaiting_pm_review"
+                    ),
+                    remediate="this task is not at the gate for CEO approval",
+                    context_briefing=await self._briefing_for(agent_id, task_id),
                 ),
-                remediate="this task is not at the gate for CEO approval",
+                agent_id=agent_id,
-                context_briefing=await self._briefing_for(agent_id, task_id),
+                task_id=task_id,
+                verb="escalate_to_ceo",
             )
         has_decision = await self.journal.has_decision_for_task(agent_id, task_id)
         if not has_decision:
@@ -1933,10 +2260,15 @@ class Choreographer:
                 hint_for_missing_journal_decision,
             )
 
-            return Envelope.tracing_gap(
+            return await self._emit_rejection(
-                missing=["journal:decision"],
+                Envelope.tracing_gap(
-                remediate=hint_for_missing_journal_decision(),
+                    missing=["journal:decision"],
-                context_briefing=await self._briefing_for(agent_id, task_id),
+                    remediate=hint_for_missing_journal_decision(),
+                    context_briefing=await self._briefing_for(agent_id, task_id),
+                ),
+                agent_id=agent_id,
+                task_id=task_id,
+                verb="escalate_to_ceo",
             )
         t = await self.task.escalate_to_ceo(task_id, agent_role=me.role, notes=reason)
         # Same as main_pm_complete: CEO acts via UI, not as a spawnable agent.

tests/unit/gateway/test_audit_on_rejection.py

@@ -0,0 +1,226 @@
+"""Every Envelope rejection from a Choreographer verb writes an audit row.
+
+Choreographer takes an ``audit`` dependency but historically never invoked
+it. The result: every rejection envelope (invalid_state, not_authorized,
+tracing_gap, not_found) silently disappeared. With no forensic trail, a
+stuck flow had no breadcrumbs.
+
+These tests pin the ``gateway.rejected`` audit-write behavior across the
+range of rejection-returning verbs, including:
+
+- not_authorized rejections (PM cannot execute code, role-typed claim)
+- invalid_state rejections (no active task, expected status mismatch)
+- tracing_gap rejections (missing notes, missing journal entries)
+- not_found rejections (unknown task id)
+
+The audit call is fire-and-forget; an exception inside ``log_event`` must
+not propagate or alter the envelope returned to the agent.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock
+from uuid import uuid4
+
+import pytest
+from roboco.services.gateway.choreographer import Choreographer, ChoreographerDeps
+
+
+def _make_deps(**overrides: Any) -> ChoreographerDeps:
+    base: dict[str, Any] = {
+        "task": AsyncMock(),
+        "work_session": AsyncMock(),
+        "git": AsyncMock(),
+        "a2a": AsyncMock(),
+        "journal": AsyncMock(),
+        "audit": AsyncMock(),
+        "evidence_repo": AsyncMock(),
+    }
+    base.update(overrides)
+    repo = base["evidence_repo"]
+    for method in (
+        "list_unread_a2a",
+        "list_unread_mentions",
+        "list_pending_notifications",
+        "task_metadata_gaps",
+        "recent_team_activity",
+        "blockers_in_lane",
+        "journal_highlights_for_task",
+    ):
+        getattr(repo, method).return_value = []
+    return ChoreographerDeps(**base)
+
+
+# ---------------------------------------------------------------------------
+# Primary acceptance test: PM cannot claim a code task — not_authorized path
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_pm_cannot_execute_code_writes_audit_row() -> None:
+    """A cell_pm calling i_will_work_on on a code task must:
+
+    1. Return an Envelope with error == 'not_authorized'
+    2. Write a gateway.rejected audit event with verb + reason details
+    """
+    aid = uuid4()
+    tid = uuid4()
+    code_task = MagicMock(
+        id=tid,
+        status="pending",
+        assigned_to=aid,
+        task_type="code",
+        priority=1,
+        parent_task_id=None,
+        sequence=0,
+        team="backend",
+    )
+    task_svc = AsyncMock()
+    task_svc.get.return_value = code_task
+    task_svc.agent_for.return_value = MagicMock(role="cell_pm", team="backend")
+    task_svc.list_in_progress_for_agent.return_value = []
+    task_svc.list_paused_for_agent.return_value = []
+    task_svc.get_subtasks.return_value = []
+    audit_svc = AsyncMock()
+    deps = _make_deps(task=task_svc, audit=audit_svc)
+    c = Choreographer(deps)
+
+    env = await c.i_will_work_on(aid, tid, plan="x")
+
+    assert env.error == "not_authorized"
+    audit_svc.log_event.assert_awaited()
+    args = audit_svc.log_event.await_args
+    assert args.kwargs["event_type"] == "gateway.rejected"
+    assert args.kwargs["details"]["verb"] == "i_will_work_on"
+    assert args.kwargs["details"]["reason"] == "not_authorized"
+
+
+# ---------------------------------------------------------------------------
+# tracing_gap path: i_have_committed with no plan
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_i_have_committed_missing_plan_writes_audit_row() -> None:
+    """Tracing-gap rejection (missing plan) should also be audited."""
+    aid = uuid4()
+    tid = uuid4()
+    task_with_no_plan = MagicMock(
+        id=tid,
+        status="in_progress",
+        assigned_to=aid,
+        plan=None,
+    )
+    task_svc = AsyncMock()
+    task_svc.get_active_task_for_agent.return_value = task_with_no_plan
+    audit_svc = AsyncMock()
+    deps = _make_deps(task=task_svc, audit=audit_svc)
+    c = Choreographer(deps)
+
+    env = await c.i_have_committed(aid, "wip")
+
+    assert env.error == "tracing_gap"
+    audit_svc.log_event.assert_awaited()
+    args = audit_svc.log_event.await_args
+    assert args.kwargs["event_type"] == "gateway.rejected"
+    assert args.kwargs["details"]["verb"] == "i_have_committed"
+    assert args.kwargs["details"]["reason"] == "tracing_gap"
+    assert "plan" in args.kwargs["details"]["missing"]
+
+
+# ---------------------------------------------------------------------------
+# invalid_state path: i_have_committed with no active task
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_i_have_committed_no_active_task_writes_audit_row() -> None:
+    """invalid_state rejection (no active task) is audited."""
+    aid = uuid4()
+    task_svc = AsyncMock()
+    task_svc.get_active_task_for_agent.return_value = None
+    audit_svc = AsyncMock()
+    deps = _make_deps(task=task_svc, audit=audit_svc)
+    c = Choreographer(deps)
+
+    env = await c.i_have_committed(aid, "wip")
+
+    assert env.error == "invalid_state"
+    audit_svc.log_event.assert_awaited()
+    args = audit_svc.log_event.await_args
+    assert args.kwargs["event_type"] == "gateway.rejected"
+    assert args.kwargs["details"]["verb"] == "i_have_committed"
+    assert args.kwargs["details"]["reason"] == "invalid_state"
+
+
+# ---------------------------------------------------------------------------
+# not_found path: unknown task id
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_unknown_task_writes_audit_row() -> None:
+    """not_found rejection (unknown task id) is audited."""
+    aid = uuid4()
+    tid = uuid4()
+    task_svc = AsyncMock()
+    task_svc.get.return_value = None
+    audit_svc = AsyncMock()
+    deps = _make_deps(task=task_svc, audit=audit_svc)
+    c = Choreographer(deps)
+
+    env = await c.i_am_done(aid, tid, notes="something")
+
+    assert env.error == "not_found"
+    audit_svc.log_event.assert_awaited()
+    args = audit_svc.log_event.await_args
+    assert args.kwargs["event_type"] == "gateway.rejected"
+    assert args.kwargs["details"]["verb"] == "i_am_done"
+    assert args.kwargs["details"]["reason"] == "not_found"
+
+
+# ---------------------------------------------------------------------------
+# Happy path must NOT write an audit row.
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_successful_verb_does_not_write_audit_row() -> None:
+    """Successful (non-error) Envelope must not emit gateway.rejected audit."""
+    aid = uuid4()
+    task_svc = AsyncMock()
+    task_svc.list_assigned_for_agent.return_value = []
+    task_svc.list_paused_for_agent.return_value = []
+    audit_svc = AsyncMock()
+    deps = _make_deps(task=task_svc, audit=audit_svc)
+    c = Choreographer(deps)
+
+    env = await c.give_me_work(aid)
+
+    # No rejection, so no audit row.
+    assert env.error is None
+    audit_svc.log_event.assert_not_awaited()
+
+
+# ---------------------------------------------------------------------------
+# Audit failure must NOT block the verb (best-effort rule).
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_audit_log_event_failure_does_not_propagate() -> None:
+    """If log_event raises, the verb still returns the rejection envelope."""
+    aid = uuid4()
+    task_svc = AsyncMock()
+    task_svc.get_active_task_for_agent.return_value = None
+    audit_svc = AsyncMock()
+    audit_svc.log_event.side_effect = RuntimeError("audit DB down")
+    deps = _make_deps(task=task_svc, audit=audit_svc)
+    c = Choreographer(deps)
+
+    # Must not raise; the rejection envelope should still come back.
+    env = await c.i_have_committed(aid, "wip")
+
+    assert env.error == "invalid_state"
+    audit_svc.log_event.assert_awaited()