From 0e7674af3de1e1d91c838c41466bcef5f33ce64b Mon Sep 17 00:00:00 2001 From: Renn F Date: Tue, 30 Jun 2026 13:26:35 +0200 Subject: [PATCH] [chore] logical-gaps: verb_runner trailing-None side-effect guard + actor_agent_id threading (3 gaps) _verb_runner.py: - run_intent skips the side_effects loop when a TRAILING composed action returned None (its source-status check failed under a concurrent transition). Previously the loop ran unconditionally on the None task and _do_push_branch(None)/_do_pr_merge(None) crashed with a NoneType AttributeError, turning the clean INVALID_STATE the entry/intermediate guards give into a 500/respawn loop. The trailing None now flows to the caller's `if task is None` handler. Latent today (no shipped intent has both a None-capable compose and trailing side_effects) but the runner is generic. (gap: runner-side-effects-fire- on-trailing-none-task) - _do_push_branch / _do_create_pr / _do_create_root_pr forward actor_agent_id=agent.id into git_service (push_branch / create_pr), matching _do_pr_merge. Without it, a verb on a task whose assigned_to was cleared before the side effect falls through to created_by and pushes from / opens a PR against the wrong workspace. (gap: side-effect-handlers-drop-actor-agent-id) - _do_escalate_to_ceo forwards actor_agent_id=agent.id so the awaiting_ceo_approval audit row attributes to the specific PM/Board agent. (gap: do-escalate-to-ceo-drops-actor-agent-id) task.py: escalate_to_ceo gains actor_agent_id param, passed as audit_agent_id to _validate_and_set_status and recorded as escalated_by_agent_id in the event payload + log. escalate_to_ceo_for_agent forwards agent.agent_id. _impl.py: the main_pm complete->escalate path forwards actor_agent_id=main_pm_agent_id. TDD: 5 red->green tests (synthetic trailing-None intent, actor forwarding for push_branch/create_pr/create_root_pr/escalate_to_ceo) + real-DB audit test asserting the awaiting_ceo_approval row carries the actor UUID. Updated 3 board escalate_to_ceo tests to assert the forwarded actor. --- .../services/gateway/choreographer/_impl.py | 8 +- .../gateway/choreographer/_verb_runner.py | 52 +++++-- roboco/services/task.py | 24 ++- .../unit/gateway/test_choreographer_board.py | 9 +- tests/unit/gateway/test_verb_runner.py | 146 ++++++++++++++++++ tests/unit/services/test_audit_agent_id.py | 98 ++++++++++++ 6 files changed, 319 insertions(+), 18 deletions(-) diff --git a/roboco/services/gateway/choreographer/_impl.py b/roboco/services/gateway/choreographer/_impl.py index ccb9d387..dfa7aa28 100644 --- a/roboco/services/gateway/choreographer/_impl.py +++ b/roboco/services/gateway/choreographer/_impl.py @@ -6602,9 +6602,13 @@ class Choreographer: # Use kwargs — service signature is (task_id, agent_role="cell_pm", # notes=None). Positional was passing agent_id as task_id and the - # actual task_id as agent_role. + # actual task_id as agent_role. Forward the main PM's UUID so the + # awaiting_ceo_approval audit row attributes to the specific PM. t = await self.task.escalate_to_ceo( - task_id=root_task_id, agent_role="main_pm", notes=notes + task_id=root_task_id, + agent_role="main_pm", + notes=notes, + actor_agent_id=main_pm_agent_id, ) # Defense-in-depth: escalate_to_ceo returns None when it refuses (e.g. a # transition guard rejects). Surface that as a clean rejection instead of diff --git a/roboco/services/gateway/choreographer/_verb_runner.py b/roboco/services/gateway/choreographer/_verb_runner.py index 49d8bf24..3aa16e20 100644 --- a/roboco/services/gateway/choreographer/_verb_runner.py +++ b/roboco/services/gateway/choreographer/_verb_runner.py @@ -95,8 +95,18 @@ class VerbRunner: "Re-fetch with evidence(task_id) and re-issue your verb." ) task = await self._dispatch_atomic(action_name, task, agent, context) - for side_effect_name in intent.side_effects: - await self._dispatch_side_effect(side_effect_name, task, agent) + # A TRAILING None (the last composed action returned None because its + # source-status check failed under a concurrent transition) is the + # verb's own result and flows out as the runner's return value. The + # side_effects loop must NOT run on that None — a side effect + # dereferences task.branch_name / task.pr_number and crashes + # (_do_push_branch(None) -> None.branch_name AttributeError), turning + # the clean INVALID_STATE the entry/intermediate guards give into a + # 500/respawn loop. Skip side effects so the caller's `if task is None` + # handler surfaces the verb-specific message. + if task is not None: + for side_effect_name in intent.side_effects: + await self._dispatch_side_effect(side_effect_name, task, agent) return task async def _dispatch_atomic( @@ -181,10 +191,17 @@ class VerbRunner: ) -> Any: # Use the actor's real role — escalate_to_ceo is allow-listed for # main_pm, product_owner, head_marketing in the spec, and the task - # service stamps the escalator's role into the audit trail. + # service stamps the escalator's role into the audit trail. Forward + # the actor's UUID so the awaiting_ceo_approval audit row attributes + # the escalation to the specific PM/Board agent (every sibling + # transition forwards the actor; this branch was the only one that + # lost it, leaving a role-only record ambiguous across same-role PMs). agent_role = str(agent.role) if agent is not None else "main_pm" return await self.task_service.escalate_to_ceo( - task_id=task.id, agent_role=agent_role, notes=ctx.notes or "" + task_id=task.id, + agent_role=agent_role, + notes=ctx.notes or "", + actor_agent_id=agent.id, ) async def _do_block(self, task: Any, agent: Any, ctx: spec.Context) -> Any: @@ -236,22 +253,37 @@ class VerbRunner: # -- Side-effect handlers --------------------------------------------- - async def _do_push_branch(self, task: Any, _agent: Any) -> Any: - return await self.git_service.push_branch(task.branch_name) + async def _do_push_branch(self, task: Any, agent: Any) -> Any: + # Forward the actor so the workspace resolves from the actor's clone + # (actor_agent_id wins over the assigned_to/created_by fallback) — + # matches _do_pr_merge. Without it, a verb on a task whose + # assigned_to was cleared before the side effect falls through to the + # wrong workspace and pushes from / opens a PR against it. + return await self.git_service.push_branch( + task.branch_name, actor_agent_id=agent.id + ) - async def _do_create_pr(self, task: Any, _agent: Any) -> Any: + async def _do_create_pr(self, task: Any, agent: Any) -> Any: from roboco.services.gateway.merge_chain import resolve_parent_branch parent = await resolve_parent_branch(task, self.task_service) return await self.git_service.create_pr( - task.branch_name, parent=parent, is_root_pr=False + task.branch_name, + parent=parent, + is_root_pr=False, + actor_agent_id=agent.id, ) - async def _do_create_root_pr(self, task: Any, _agent: Any) -> Any: + async def _do_create_root_pr(self, task: Any, agent: Any) -> Any: # Root→master PR for the in-path gate's root level (submit_root). The # base is always master and is_root_pr marks it for the CEO-merge path. + # The PM opening the master PR is the assigned_to-may-be-None case + # create_pr's actor_agent_id exists for. return await self.git_service.create_pr( - task.branch_name, parent="master", is_root_pr=True + task.branch_name, + parent="master", + is_root_pr=True, + actor_agent_id=agent.id, ) async def _do_pr_merge(self, task: Any, agent: Any) -> Any: diff --git a/roboco/services/task.py b/roboco/services/task.py index 4ed8316a..be2bd67d 100644 --- a/roboco/services/task.py +++ b/roboco/services/task.py @@ -5196,6 +5196,7 @@ class TaskService(BaseService): task_id: UUID, agent_role: str = "cell_pm", notes: str | None = None, + actor_agent_id: UUID | None = None, ) -> TaskTable | None: """ Escalate a task to CEO for final approval (PM only). @@ -5209,6 +5210,11 @@ class TaskService(BaseService): task_id: The task to escalate agent_role: Role of the agent escalating (must be PM) notes: Optional notes for the CEO + actor_agent_id: The escalating agent's UUID, stamped on the + audit row so the awaiting_ceo_approval transition attributes + to the specific PM/Board agent (every sibling transition + forwards the actor; without it the record is role-only and + ambiguous across same-role PMs). Returns: The escalated task or None if escalation not allowed @@ -5255,7 +5261,10 @@ class TaskService(BaseService): # Validate transition with PM role requirement self._validate_and_set_status( - task, TaskStatus.AWAITING_CEO_APPROVAL, agent_role + task, + TaskStatus.AWAITING_CEO_APPROVAL, + agent_role, + audit_agent_id=actor_agent_id, ) await self.session.flush() @@ -5263,13 +5272,20 @@ class TaskService(BaseService): await self._emit_task_event( EventType.TASK_AWAITING_CEO_APPROVAL, task_id, - {"escalated_by_role": agent_role, "notes": notes}, + { + "escalated_by_role": agent_role, + "escalated_by_agent_id": str(actor_agent_id) + if actor_agent_id + else None, + "notes": notes, + }, ) self.log.info( "Task escalated to CEO for approval", task_id=str(task_id), escalated_by_role=agent_role, + escalated_by_agent_id=str(actor_agent_id) if actor_agent_id else None, ) return task @@ -7151,7 +7167,9 @@ class TaskService(BaseService): task, task_id, agent, permissions, notes ) - escalated = await self.escalate_to_ceo(task_id, agent.role.value, notes) + escalated = await self.escalate_to_ceo( + task_id, agent.role.value, notes, actor_agent_id=agent.agent_id + ) if not escalated: raise ValidationError( "Cannot escalate to CEO - task must be in awaiting_pm_review status" diff --git a/tests/unit/gateway/test_choreographer_board.py b/tests/unit/gateway/test_choreographer_board.py index 7caa5e6f..1ebbc997 100644 --- a/tests/unit/gateway/test_choreographer_board.py +++ b/tests/unit/gateway/test_choreographer_board.py @@ -72,7 +72,7 @@ async def test_board_escalate_to_ceo_succeeds_for_product_owner() -> None: after = MagicMock(**{**t.__dict__, "status": "awaiting_ceo_approval"}) task_svc = AsyncMock() task_svc.get.return_value = t - task_svc.agent_for.return_value = MagicMock(role="product_owner") + task_svc.agent_for.return_value = MagicMock(id=agent_id, role="product_owner") task_svc.escalate_to_ceo.return_value = after journal_svc = AsyncMock() journal_svc.has_decision_for_task.return_value = True @@ -87,6 +87,7 @@ async def test_board_escalate_to_ceo_succeeds_for_product_owner() -> None: task_id=task_id, agent_role="product_owner", notes="ready for CEO sign-off", + actor_agent_id=agent_id, ) @@ -102,7 +103,7 @@ async def test_board_escalate_to_ceo_succeeds_for_head_marketing() -> None: after = MagicMock(**{**t.__dict__, "status": "awaiting_ceo_approval"}) task_svc = AsyncMock() task_svc.get.return_value = t - task_svc.agent_for.return_value = MagicMock(role="head_marketing") + task_svc.agent_for.return_value = MagicMock(id=agent_id, role="head_marketing") task_svc.escalate_to_ceo.return_value = after journal_svc = AsyncMock() journal_svc.has_decision_for_task.return_value = True @@ -117,6 +118,7 @@ async def test_board_escalate_to_ceo_succeeds_for_head_marketing() -> None: task_id=task_id, agent_role="head_marketing", notes="brand-affecting change", + actor_agent_id=agent_id, ) @@ -236,7 +238,7 @@ async def test_board_escalate_to_ceo_succeeds_for_main_pm() -> None: after = MagicMock(**{**t.__dict__, "status": "awaiting_ceo_approval"}) task_svc = AsyncMock() task_svc.get.return_value = t - task_svc.agent_for.return_value = MagicMock(role="main_pm") + task_svc.agent_for.return_value = MagicMock(id=agent_id, role="main_pm") task_svc.escalate_to_ceo.return_value = after journal_svc = AsyncMock() journal_svc.has_decision_for_task.return_value = True @@ -251,6 +253,7 @@ async def test_board_escalate_to_ceo_succeeds_for_main_pm() -> None: task_id=task_id, agent_role="main_pm", notes="root task done", + actor_agent_id=agent_id, ) diff --git a/tests/unit/gateway/test_verb_runner.py b/tests/unit/gateway/test_verb_runner.py index 292d49f5..56be045b 100644 --- a/tests/unit/gateway/test_verb_runner.py +++ b/tests/unit/gateway/test_verb_runner.py @@ -8,6 +8,7 @@ commits. from __future__ import annotations +import dataclasses from typing import TYPE_CHECKING from unittest.mock import AsyncMock, MagicMock from uuid import uuid4 @@ -260,3 +261,148 @@ async def test_runner_does_not_run_side_effects_if_compose_fails() -> None: await runner.run_intent("i_will_work_on", task, agent, ctx) git_svc.push_branch.assert_not_called() git_svc.create_pr.assert_not_called() + + +@pytest.mark.asyncio +async def test_runner_skips_side_effects_when_trailing_compose_returns_none( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A TRAILING composed action returning None (its source-status check + failed under a concurrent transition) must flow cleanly to the caller's + ``if task is None`` handler. The side_effects loop must NOT run on the + None task — ``_do_push_branch(None)`` dereferences ``task.branch_name`` + and crashes, turning a clean INVALID_STATE into a 500/respawn loop. + + Latent today (no shipped intent has both a None-capable compose and a + trailing side_effect), but the runner is generic and any future intent + inherits the crash-to-500 instead of the clean INVALID_STATE the + entry/intermediate None guards give. + """ + task_svc = AsyncMock() + task_svc.session.begin_nested = MagicMock( + return_value=MagicMock(__aenter__=AsyncMock(), __aexit__=AsyncMock()) + ) + # start() returns None — its source status was invalid (concurrent change). + task_svc.start = AsyncMock(return_value=None) + git_svc = AsyncMock() + git_svc.push_branch = AsyncMock() + runner = VerbRunner(task_service=task_svc, git_service=git_svc) + + # Synthetic intent: a None-capable compose + a trailing side_effect. + synthetic = dataclasses.replace( + spec._INTENT_VERBS["open_pr"], + name="synthetic_push_after_start", + composes=("start",), + side_effects=("push_branch",), + pre_side_effects=(), + extra_preconditions=(), + ) + monkeypatch.setitem(spec._INTENT_VERBS, "synthetic_push_after_start", synthetic) + + task = MagicMock(id=uuid4(), status="claimed", plan=None, commits=[]) + agent = MagicMock(id=uuid4(), role="developer") + ctx = spec.Context() + + result = await runner.run_intent("synthetic_push_after_start", task, agent, ctx) + # The trailing None flows out as the verb result, not a side_effect crash. + assert result is None + git_svc.push_branch.assert_not_called() + + +@pytest.mark.asyncio +async def test_runner_forwards_actor_agent_id_to_push_branch_and_create_pr() -> None: + """push_branch / create_pr side effects must forward the actor's + agent.id as ``actor_agent_id`` — the actor is the authoritative workspace + resolver (``actor_agent_id or assigned_to or created_by``). Without it, a + side_effect-bearing verb on a task whose ``assigned_to`` was cleared (e.g. + after pr_pass) falls through to created_by and pushes from / opens a PR + against the wrong workspace. Mirrors _do_pr_merge, which already forwards.""" + task_svc = AsyncMock() + task_svc.session.begin_nested = MagicMock( + return_value=MagicMock(__aenter__=AsyncMock(), __aexit__=AsyncMock()) + ) + git_svc = AsyncMock() + git_svc.push_branch = AsyncMock() + git_svc.create_pr = AsyncMock(return_value={"pr_number": 42}) + runner = VerbRunner(task_service=task_svc, git_service=git_svc) + + agent = MagicMock(id=uuid4(), role="developer") + task = MagicMock( + id=uuid4(), + status="in_progress", + commits=["abc"], + pr_number=None, + parent_task_id=None, + branch_name="feature/backend/ABC12345", + project_id=uuid4(), + ) + ctx = spec.Context() + + await runner.run_intent("open_pr", task, agent, ctx) + + _, push_kwargs = git_svc.push_branch.call_args + assert push_kwargs.get("actor_agent_id") == agent.id, ( + "push_branch must resolve the workspace from the actor, not the fallback" + ) + _, pr_kwargs = git_svc.create_pr.call_args + assert pr_kwargs.get("actor_agent_id") == agent.id, ( + "create_pr must resolve the workspace from the actor, not the fallback" + ) + + +@pytest.mark.asyncio +async def test_runner_forwards_actor_agent_id_to_create_root_pr() -> None: + """The root→master PR side effect (submit_root's pre_side_effect) must + forward the actor's agent.id too — a PM opening the master PR is exactly + the ``assigned_to may be None at completion time`` case create_pr's + actor_agent_id exists for.""" + task_svc = AsyncMock() + task_svc.session.begin_nested = MagicMock( + return_value=MagicMock(__aenter__=AsyncMock(), __aexit__=AsyncMock()) + ) + task_svc.submit_for_review = AsyncMock( + return_value=MagicMock(status="awaiting_pr_review") + ) + git_svc = AsyncMock() + git_svc.create_pr = AsyncMock(return_value={"pr_number": 7}) + runner = VerbRunner(task_service=task_svc, git_service=git_svc) + + agent = MagicMock(id=uuid4(), role="main_pm") + task = MagicMock( + id=uuid4(), + status="in_progress", + parent_task_id=None, + branch_name="feature/main_pm/ROOT0001", + project_id=uuid4(), + ) + ctx = spec.Context(notes="root scope complete; bubbling to master") + + await runner.run_intent("submit_root", task, agent, ctx) + + assert git_svc.create_pr.call_args.kwargs.get("is_root_pr") is True + assert git_svc.create_pr.call_args.kwargs.get("actor_agent_id") == agent.id + + +@pytest.mark.asyncio +async def test_runner_forwards_actor_agent_id_to_escalate_to_ceo() -> None: + """escalate_to_ceo must thread the actor's agent.id so the awaiting_ceo + approval audit row attributes the escalation to the specific PM/Board + agent, not just a role. Every sibling transition (claim/start/qa_pass/ + pr_pass) forwards the actor UUID; the escalate_to_ceo branch was the only + one that lost it.""" + task_svc = AsyncMock() + task_svc.session.begin_nested = MagicMock( + return_value=MagicMock(__aenter__=AsyncMock(), __aexit__=AsyncMock()) + ) + task_svc.escalate_to_ceo = AsyncMock( + return_value=MagicMock(status="awaiting_ceo_approval") + ) + runner = VerbRunner(task_service=task_svc, git_service=AsyncMock()) + + agent = MagicMock(id=uuid4(), role="main_pm") + task = MagicMock(id=uuid4(), status="awaiting_pm_review", pr_number=99) + ctx = spec.Context(notes="escalating for CEO sign-off") + + await runner.run_intent("escalate_to_ceo", task, agent, ctx) + + assert task_svc.escalate_to_ceo.call_args.kwargs.get("actor_agent_id") == agent.id diff --git a/tests/unit/services/test_audit_agent_id.py b/tests/unit/services/test_audit_agent_id.py index 03bfd8ed..03900aaa 100644 --- a/tests/unit/services/test_audit_agent_id.py +++ b/tests/unit/services/test_audit_agent_id.py @@ -350,3 +350,101 @@ async def test_log_agent_event_unknown_slug_writes_null_agent_id( # and the slug is preserved in details for forensic lookup. assert rows[0].agent_id is None assert rows[0].details.get("agent_slug") == unknown_slug + + +@pytest.mark.asyncio +async def test_escalate_to_ceo_writes_audit_with_actor_agent_id( + patched_session_factory: AsyncSession, +) -> None: + """End-to-end: ``escalate_to_ceo(actor_agent_id=...)`` -> the + ``task.awaiting_ceo_approval`` audit row carries the escalating agent's + UUID, not NULL. + + Every sibling transition (claim/start/qa_pass/pr_pass) forwards the actor + UUID to ``_validate_and_set_status(audit_agent_id=...)``; the + escalate_to_ceo branch lost it, attributing the escalation only to a role + (ambiguous when multiple PMs of the same role could escalate). Mirrors + ``test_submit_for_qa_writes_audit_with_dev_agent_id``. + """ + actor_uuid, _ = await _seed_agent_with_slug(patched_session_factory) + system_uuid, _ = await _seed_agent_with_slug(patched_session_factory) + + project = ProjectTable( + id=uuid4(), + name="Escalate Audit Test Project", + slug=f"escalate-audit-{uuid4().hex[:8]}", + git_url="https://github.com/example/escalate-audit.git", + default_branch="main", + protected_branches=["main"], + assigned_cell=Team.BACKEND, + created_by=system_uuid, + is_active=True, + ) + patched_session_factory.add(project) + await patched_session_factory.flush() + + task = TaskTable( + id=uuid4(), + title="Escalate audit-id test", + description="Verifies escalate_to_ceo stamps the actor UUID on the audit row.", + acceptance_criteria=["audit row has agent_id populated"], + status=TaskStatus.AWAITING_PM_REVIEW, + priority=2, + task_type=TaskType.CODE, + nature=TaskNature.TECHNICAL, + project_id=project.id, + branch_name="feature/main_pm/ROOT0001", + pr_number=99, + pr_url="https://github.com/example/escalate-audit/pull/99", + docs_complete=True, + pr_created=True, + created_by=system_uuid, + assigned_to=actor_uuid, + claimed_by=actor_uuid, + team=Team.BACKEND, + dependency_ids=[], + blocker_ids=[], + sequence=0, + plan={"steps": ["impl"]}, + estimated_complexity=Complexity.MEDIUM, + checkpoints=[], + progress_updates=[], + commits=[{"sha": "abc123", "message": "[ROOT0001] init"}], + documents=[], + dev_notes="root complete", + self_verified=True, + ) + patched_session_factory.add(task) + await patched_session_factory.commit() + + service = TaskService(patched_session_factory) + captured_task_id = UUID(str(task.id)) + result = await service.escalate_to_ceo( + captured_task_id, + agent_role="main_pm", + actor_agent_id=actor_uuid, + ) + assert result is not None + assert result.status == TaskStatus.AWAITING_CEO_APPROVAL + + # Drain any background tasks the transition scheduled. + pending = [bg for bg in service._background_tasks if not bg.done()] + if pending: + await asyncio.gather(*pending, return_exceptions=True) + for _ in range(5): + await asyncio.sleep(0.05) + + result_rows = await patched_session_factory.execute( + select(AuditLogTable) + .where(AuditLogTable.event_type == "task.awaiting_ceo_approval") + .where(AuditLogTable.target_id == captured_task_id) + ) + rows = list(result_rows.scalars().all()) + assert len(rows) == 1, ( + f"Expected exactly one task.awaiting_ceo_approval audit row, got {len(rows)}" + ) + assert rows[0].agent_id == actor_uuid, ( + f"audit_log.agent_id must be the escalating PM's UUID ({actor_uuid}), " + f"got {rows[0].agent_id} — escalate_to_ceo dropped the actor from the " + "audit trail" + )