diff --git a/roboco/services/audit.py b/roboco/services/audit.py index 959b2c37..3aee901e 100644 --- a/roboco/services/audit.py +++ b/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, *, diff --git a/roboco/services/gateway/choreographer.py b/roboco/services/gateway/choreographer.py index f36eb33e..d1df3e97 100644 --- a/roboco/services/gateway/choreographer.py +++ b/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='')" ) - return Envelope.tracing_gap( - missing=["plan"], - remediate=remediate, - context_briefing=briefing, + return await self._emit_rejection( + Envelope.tracing_gap( + missing=["plan"], + 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( - message=f"task {task_id} is in {status}; cannot start work", - remediate="call give_me_work() to find an actionable task", - context_briefing=briefing, + return await self._emit_rejection( + Envelope.invalid_state( + message=f"task {task_id} is in {status}; cannot start work", + 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( - message="no active task for this agent", - remediate="call give_me_work() then i_will_work_on(task_id, plan)", - context_briefing=await self._briefing_for(agent_id, None), + return await self._emit_rejection( + Envelope.invalid_state( + message="no active task for this agent", + 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( - missing=["plan"], - remediate=no_plan_remediate, - context_briefing=await self._briefing_for(agent_id, t.id), + return await self._emit_rejection( + Envelope.tracing_gap( + missing=["plan"], + 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( - message=f"task {task_id} is not assigned to you", - remediate="call give_me_work() to find your work", - context_briefing=briefing, + return await self._emit_rejection( + Envelope.not_authorized( + message=f"task {task_id} is not assigned to you", + 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( - message="no commits on this task yet", - remediate=( - "commit at least one change before submitting for QA — " - "call commit(message='')" + return await self._emit_rejection( + Envelope.invalid_state( + message="no commits on this task yet", + remediate=( + "commit at least one change before submitting for QA — " + "call commit(message='')" + ), + 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( - t.branch_name, parent=parent, is_root_pr=False - ) + pr = await self.git.create_pr(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( - message="not assigned to you", - remediate="claim it via i_will_work_on(task_id) first", - context_briefing=await self._briefing_for(agent_id, task_id), + return await self._emit_rejection( + Envelope.not_authorized( + message="not assigned to you", + 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( - message="not assigned to you", - remediate="claim it via i_will_work_on(task_id) first", - context_briefing=await self._briefing_for(agent_id, task_id), + return await self._emit_rejection( + Envelope.not_authorized( + message="not assigned to you", + 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( - message=( - f"task {task_id} is in {t.status}, expected awaiting_qa for review" + return await self._emit_rejection( + Envelope.invalid_state( + 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", - context_briefing=await self._briefing_for(qa_agent_id, task_id), + agent_id=qa_agent_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( - guard, await self._briefing_for(qa_agent_id, task_id) + return await self._emit_rejection( + 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( - message="not assigned to you", - remediate="claim it via claim_review(task_id) first", - context_briefing=await self._briefing_for(qa_agent_id, task_id), + return await self._emit_rejection( + Envelope.not_authorized( + message="not assigned to you", + 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( - missing, task_id, await self._briefing_for(qa_agent_id, task_id) + return await self._emit_rejection( + 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( - message="not assigned to you", - remediate="claim it via claim_review(task_id) first", - context_briefing=await self._briefing_for(qa_agent_id, task_id), + return await self._emit_rejection( + Envelope.not_authorized( + message="not assigned to you", + 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( - message="fail_review requires at least one issue", - remediate="pass issues=['', ...]", - context_briefing=await self._briefing_for(qa_agent_id, task_id), + return await self._emit_rejection( + Envelope.invalid_state( + message="fail_review requires at least one issue", + remediate="pass issues=['', ...]", + 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( - missing, task_id, await self._briefing_for(qa_agent_id, task_id) + return await self._emit_rejection( + 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( - message=( - f"task {task_id} is in {t.status}, expected awaiting_documentation" + return await self._emit_rejection( + Envelope.invalid_state( + 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", - context_briefing=await self._briefing_for(doc_agent_id, task_id), + agent_id=doc_agent_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( - guard, await self._briefing_for(doc_agent_id, task_id) + return await self._emit_rejection( + 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( - message="not assigned to you", - remediate="claim it via claim_doc_task(task_id) first", - context_briefing=await self._briefing_for(doc_agent_id, task_id), + return await self._emit_rejection( + Envelope.not_authorized( + message="not assigned to you", + 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( - missing=["docs_notes>=20"], - remediate=( - "i_documented requires notes>=20 chars summarizing what you " - "documented and where (file paths)." - " Include each file in `files=...`." + return await self._emit_rejection( + Envelope.tracing_gap( + missing=["docs_notes>=20"], + remediate=( + "i_documented requires notes>=20 chars summarizing what you " + "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( - missing=["files"], - remediate=( - "i_documented requires files=['', ...]" - " listing the doc files written." + return await self._emit_rejection( + Envelope.tracing_gap( + missing=["files"], + remediate=( + "i_documented requires files=['', ...]" + " 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( - message="claim failed", - remediate="task may already be claimed by another agent", - context_briefing=await self._briefing_for(pm_agent_id, task_id), + return await self._emit_rejection( + Envelope.invalid_state( + message="claim failed", + 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( - message="could not transition to awaiting_pm_review", - remediate="check task state — must be in_progress with PR ready", - context_briefing=await self._briefing_for(pm_agent_id, task_id), + return await self._emit_rejection( + Envelope.invalid_state( + message="could not transition to awaiting_pm_review", + 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( - message=f"task {task_id} is in {t.status}, expected blocked", - remediate=( - "this task is not blocked; call triage() to find blocked tasks" + return await self._emit_rejection( + Envelope.invalid_state( + message=f"task {task_id} is in {t.status}, expected blocked", + 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( - missing=["journal:decision"], - remediate=hint_for_missing_journal_decision(), - context_briefing=await self._briefing_for(pm_agent_id, task_id), + return await self._emit_rejection( + Envelope.tracing_gap( + missing=["journal:decision"], + 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( - message=f"role {agent.role} cannot complete tasks via this verb", - remediate="only cell_pm and main_pm can call complete", - context_briefing=await self._briefing_for(agent_id, task_id), + return await self._emit_rejection( + Envelope.not_authorized( + message=f"role {agent.role} cannot complete tasks via this verb", + 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( - missing=["journal:decision"], - remediate=hint_for_missing_journal_decision(), - context_briefing=await self._briefing_for(pm_agent_id, task_id), + return await self._emit_rejection( + Envelope.tracing_gap( + missing=["journal:decision"], + 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( - message="no escalation target configured for your role", - remediate="check agents_config.py ESCALATION_CHAIN for your slug", - context_briefing=await self._briefing_for(pm_agent_id, task_id), + return await self._emit_rejection( + Envelope.invalid_state( + message="no escalation target configured for your role", + 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( - message=( - f"could not escalate task {task_id} to {target_slug}: " - "target agent not found or task missing" + return await self._emit_rejection( + Envelope.invalid_state( + message=( + 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=( - 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), + agent_id=pm_agent_id, + task_id=task_id, + verb="escalate_up", ) 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( - message=f"role {me.role} cannot escalate to CEO directly", - remediate="use escalate_up() to go through your escalation chain", - context_briefing=await self._briefing_for(agent_id, task_id), + return await self._emit_rejection( + Envelope.not_authorized( + message=f"role {me.role} cannot escalate to CEO directly", + 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( - message=( - f"task {task_id} is in {t.status}, expected awaiting_pm_review" + return await self._emit_rejection( + Envelope.invalid_state( + 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", - context_briefing=await self._briefing_for(agent_id, task_id), + agent_id=agent_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( - missing=["journal:decision"], - remediate=hint_for_missing_journal_decision(), - context_briefing=await self._briefing_for(agent_id, task_id), + return await self._emit_rejection( + Envelope.tracing_gap( + missing=["journal:decision"], + 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. diff --git a/tests/unit/gateway/test_audit_on_rejection.py b/tests/unit/gateway/test_audit_on_rejection.py new file mode 100644 index 00000000..459e316d --- /dev/null +++ b/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()