From 5b931c367f3a78682c9dd825af93f1a0f87684a3 Mon Sep 17 00:00:00 2001 From: Renn F Date: Sat, 27 Jun 2026 07:19:35 +0200 Subject: [PATCH] Fix different project same PR number collision problem MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fix (two layers): 1. Root cause — pr_merge and rebase_pr_for_task now take a required project_id and scope the lookup where(pr_number == X AND project_id == Y). Required so no caller can forget — the bug class can't recur. All 4 call sites updated (choreographer cell_pm_complete, the rebase-retry, the superseded close_pull_request now passes project_id, and _verb_runner._do_pr_merge). 2. Crash guard — _finalize_cell_complete None-checks the complete() return and returns a clean invalid_state envelope (with a remediate hint) instead of dereffing None → 500 → respawn loop. --- .../services/gateway/choreographer/_impl.py | 39 +++++++++-- .../gateway/choreographer/_verb_runner.py | 5 +- roboco/services/git.py | 24 ++++++- tests/unit/gateway/test_choreographer_pm.py | 54 +++++++++++++++- .../gateway/test_merge_conflict_resolver.py | 16 +++-- tests/unit/services/test_git.py | 11 +++- .../services/test_pr_merge_concurrency.py | 64 +++++++++++++++++-- 7 files changed, 193 insertions(+), 20 deletions(-) diff --git a/roboco/services/gateway/choreographer/_impl.py b/roboco/services/gateway/choreographer/_impl.py index 81470810..86748024 100644 --- a/roboco/services/gateway/choreographer/_impl.py +++ b/roboco/services/gateway/choreographer/_impl.py @@ -14,7 +14,7 @@ from __future__ import annotations import contextlib from dataclasses import dataclass from datetime import UTC, datetime -from typing import Any, ClassVar +from typing import Any, ClassVar, cast from uuid import UUID import structlog @@ -5426,7 +5426,10 @@ class Choreographer: target = await resolve_parent_branch(t, self.task) try: merge_result = await self.git.pr_merge( - t.pr_number, target=target, actor_agent_id=pm_agent_id + t.pr_number, + target=target, + project_id=cast("UUID", t.project_id), + actor_agent_id=pm_agent_id, ) except MergeConflictError as exc: # A sibling landed overlapping work first, so this PR can't merge. @@ -5456,6 +5459,28 @@ class Choreographer: notes, merge_commit=merge_commit, ) + # `complete()` returns None when its prerequisites fail — most + # notably the PR-merged guard (`work_session.pr_status == "merged"`), + # which is exactly what breaks when the merge recorded against the + # WRONG task's work session (the cross-repo pr_number collision that + # `pr_merge`'s project_id scoping now prevents). Fail closed into a + # clean invalid_state envelope instead of dereferencing None and + # 500-ing, which left the be-pm thrashing escalate<->blocked. + if t is None: + return Envelope.invalid_state( + message=( + "task could not be completed: its PR is not recorded as " + "merged against this task's work session, or it has " + "incomplete subtasks / an invalid completion status" + ), + remediate=( + "re-issue pr_merge for this task's PR, then complete; if " + "the PR is genuinely merged on GitHub but the work session " + "still shows open, escalate to the CEO to reconcile the " + "session and complete manually" + ), + context_briefing=await self._briefing_for(pm_agent_id, task_id), + ) # Now that the leaf is completed, propagate the completion up to the # parent task: if the parent's subtasks are all terminal, hand the # parent off to the cell_pm for that team so it gets respawned for @@ -5490,12 +5515,17 @@ class Choreographer: CEO (``awaiting_ceo_approval``) so the task leaves the agent loop. """ rebase = await self.git.rebase_pr_for_task( - t.pr_number, actor_agent_id=pm_agent_id + t.pr_number, + project_id=cast("UUID", t.project_id), + actor_agent_id=pm_agent_id, ) status = rebase.get("status") if status == "rebased": merge_result = await self.git.pr_merge( - t.pr_number, target=target, actor_agent_id=pm_agent_id + t.pr_number, + target=target, + project_id=cast("UUID", t.project_id), + actor_agent_id=pm_agent_id, ) return await self._finalize_cell_complete( pm_agent_id, task_id, t, notes, merge_result.get("merge_commit_sha") @@ -5509,6 +5539,7 @@ class Choreographer: "first. Completing the task without a redundant merge." ), actor_agent_id=pm_agent_id, + project_id=cast("UUID", t.project_id), ) return await self._finalize_cell_complete( pm_agent_id, task_id, t, notes, None diff --git a/roboco/services/gateway/choreographer/_verb_runner.py b/roboco/services/gateway/choreographer/_verb_runner.py index e824710d..49d8bf24 100644 --- a/roboco/services/gateway/choreographer/_verb_runner.py +++ b/roboco/services/gateway/choreographer/_verb_runner.py @@ -259,7 +259,10 @@ class VerbRunner: target = await resolve_parent_branch(task, self.task_service) return await self.git_service.pr_merge( - task.pr_number, target=target, actor_agent_id=agent.id + task.pr_number, + target=target, + project_id=task.project_id, + actor_agent_id=agent.id, ) @classmethod diff --git a/roboco/services/git.py b/roboco/services/git.py index 6b12f395..f6c091a4 100644 --- a/roboco/services/git.py +++ b/roboco/services/git.py @@ -3418,6 +3418,7 @@ class GitService(BaseService): pr_number: int, *, target: str, + project_id: UUID, actor_agent_id: UUID | None = None, ) -> dict[str, Any]: """Merge PR `pr_number` into `target`. @@ -3425,6 +3426,14 @@ class GitService(BaseService): Returns: ``{"merge_commit_sha": str | None}``. Looks up the task/project that owns the PR to resolve workspace + token. + ``pr_number`` alone is ambiguous across projects — GitHub numbers + PRs per-repo, but ``tasks.pr_number`` stores the bare integer with + no repo scoping, so two tasks on different repos can share a number. + The caller MUST pass the ``project_id`` the PR belongs to so the + task lookup is scoped to it: a same-numbered PR in another + project's repo is never resolved (and merged, or its work session + marked merged) by accident. Mirrors :meth:`close_pull_request`. + Concurrency: takes a row-level lock on the parent task before invoking the GitHub merge API so that two PMs completing sibling subtasks of the same parent are serialized. On a 409 @@ -3436,7 +3445,10 @@ class GitService(BaseService): from roboco.db.tables import TaskTable as _TaskTable result = await self.session.execute( - select(_TaskTable).where(_TaskTable.pr_number == pr_number).limit(1) + select(_TaskTable) + .where(_TaskTable.pr_number == pr_number) + .where(_TaskTable.project_id == project_id) + .limit(1) ) task = result.scalar_one_or_none() if task is None: @@ -3581,6 +3593,7 @@ class GitService(BaseService): self, pr_number: int, *, + project_id: UUID, actor_agent_id: UUID | None = None, ) -> dict[str, Any]: """Resolve workspace/refs for a PR and rebase its branch onto the base. @@ -3589,13 +3602,20 @@ class GitService(BaseService): that owns the PR (mirrors :meth:`pr_merge`) and reads the PR's head/base refs from GitHub. Returns the same classification dict, or ``{"status": "unknown"}`` when refs can't be resolved. + + ``pr_number`` is ambiguous across projects (GitHub numbers PRs per-repo) + so the caller MUST pass ``project_id`` to scope the task lookup — the + same cross-repo guard as :meth:`pr_merge` / :meth:`close_pull_request`. """ from sqlalchemy import select from roboco.db.tables import TaskTable as _TaskTable result = await self.session.execute( - select(_TaskTable).where(_TaskTable.pr_number == pr_number).limit(1) + select(_TaskTable) + .where(_TaskTable.pr_number == pr_number) + .where(_TaskTable.project_id == project_id) + .limit(1) ) task = result.scalar_one_or_none() if task is None: diff --git a/tests/unit/gateway/test_choreographer_pm.py b/tests/unit/gateway/test_choreographer_pm.py index 74eed570..9520c34c 100644 --- a/tests/unit/gateway/test_choreographer_pm.py +++ b/tests/unit/gateway/test_choreographer_pm.py @@ -318,10 +318,62 @@ async def test_cell_pm_complete_merges_then_completes() -> None: assert env.error is None assert env.status == "completed" git_svc.pr_merge.assert_awaited_once_with( - 8, target="feature/main_pm/abc", actor_agent_id=pm_id + 8, + target="feature/main_pm/abc", + project_id=t.project_id, + actor_agent_id=pm_id, ) +@pytest.mark.asyncio +async def test_cell_pm_complete_does_not_500_when_complete_returns_none() -> None: + """`complete()` returns None when its PR-merged guard fails — which is + exactly what happened live when the merge recorded against the WRONG + task's work session (the cross-repo pr_number collision) left this + task's `pr_status="open"`. `_finalize_cell_complete` used to deref + `t.status` on the None and 500, leaving the cell PM thrashing + escalate<->blocked until the CEO intervened. It must fail closed into a + clean invalid_state envelope, never a 500. + """ + pm_id = uuid4() + task_id = uuid4() + parent_id = uuid4() + t = MagicMock( + id=task_id, + status="awaiting_pm_review", + assigned_to=pm_id, + pr_number=8, + branch_name="feature/backend/abc--def", + parent_task_id=parent_id, + team="backend", + ) + parent = MagicMock( + id=parent_id, branch_name="feature/main_pm/abc", parent_task_id=None + ) + task_svc = AsyncMock() + task_svc.get.side_effect = lambda tid: parent if tid == parent_id else t + task_svc.all_subtasks_terminal.return_value = True + task_svc.cell_pm_complete.return_value = None # complete() rejected + git_svc = AsyncMock() + git_svc.pr_merge.return_value = {"merge_commit_sha": "merge-abc"} + journal_svc = AsyncMock() + journal_svc.has_decision_for_task.return_value = True + journal_svc.latest_decision_at.return_value = datetime.now(UTC) + journal_svc.has_reflect_for_task.return_value = True + deps = _make_deps(task=task_svc, git=git_svc, journal=journal_svc) + c = Choreographer(deps) + + env = await c.cell_pm_complete(pm_id, task_id, notes="reviewed and approved") + body = env.as_dict() + assert body["error"] == "invalid_state" + assert ( + "merged" in body.get("message", "").lower() + or "complete" in body.get("message", "").lower() + ) + # No AttributeError 500 — the verb completed cleanly with a remediation. + assert body.get("remediate") + + @pytest.mark.asyncio async def test_cell_pm_complete_blocks_if_subtasks_unfinished() -> None: pm_id = uuid4() diff --git a/tests/unit/gateway/test_merge_conflict_resolver.py b/tests/unit/gateway/test_merge_conflict_resolver.py index e530d5dd..bfada29c 100644 --- a/tests/unit/gateway/test_merge_conflict_resolver.py +++ b/tests/unit/gateway/test_merge_conflict_resolver.py @@ -62,7 +62,9 @@ async def test_superseded_closes_pr_and_completes_without_merge( return_value=MagicMock(status="completed", parent_task_id=None, team="frontend") ) choreo = _choreo(task, git, monkeypatch) - t = MagicMock(pr_number=159, parent_task_id=None, team="frontend") + t = MagicMock( + pr_number=159, project_id=uuid4(), parent_task_id=None, team="frontend" + ) env = await choreo._resolve_merge_conflict_on_complete( uuid4(), uuid4(), t, "feature/frontend/root--cell", "notes", _EXC @@ -93,7 +95,9 @@ async def test_rebased_retries_merge_and_completes( return_value=MagicMock(status="completed", parent_task_id=None, team="frontend") ) choreo = _choreo(task, git, monkeypatch) - t = MagicMock(pr_number=160, parent_task_id=None, team="backend") + t = MagicMock( + pr_number=160, project_id=uuid4(), parent_task_id=None, team="backend" + ) await choreo._resolve_merge_conflict_on_complete( uuid4(), uuid4(), t, "feature/backend/root--cell", "notes", _EXC @@ -124,7 +128,9 @@ async def test_genuine_conflict_escalates_to_ceo_and_does_not_loop( notify = AsyncMock() monkeypatch.setattr(choreo, "_notify_ceo_merge_conflict", notify) tid = uuid4() - t = MagicMock(pr_number=160, parent_task_id=None, team="backend") + t = MagicMock( + pr_number=160, project_id=uuid4(), parent_task_id=None, team="backend" + ) env = await choreo._resolve_merge_conflict_on_complete( uuid4(), tid, t, "feature/backend/root--cell", "notes", _EXC @@ -155,7 +161,9 @@ async def test_unknown_rebase_outcome_escalates_rather_than_completing( task.admin_set_status = AsyncMock() task.get = AsyncMock(return_value=MagicMock(status="awaiting_ceo_approval")) choreo = _choreo(task, git, monkeypatch) - t = MagicMock(pr_number=160, parent_task_id=None, team="backend") + t = MagicMock( + pr_number=160, project_id=uuid4(), parent_task_id=None, team="backend" + ) await choreo._resolve_merge_conflict_on_complete( uuid4(), uuid4(), t, "feature/backend/root--cell", "notes", _EXC diff --git a/tests/unit/services/test_git.py b/tests/unit/services/test_git.py index ada03f27..ca9b95e4 100644 --- a/tests/unit/services/test_git.py +++ b/tests/unit/services/test_git.py @@ -549,14 +549,19 @@ async def test_pr_merge_returns_merge_commit_dict() -> None: # Merges flow UP the chain (cell -> Main-PM branch), never into master via # this agent path — target is the integration branch, not the default branch. with _patch_project_service(fake_project): - out = await svc.pr_merge(11, target="feature/main_pm/root1234") + out = await svc.pr_merge( + 11, target="feature/main_pm/root1234", project_id=project_id + ) assert out == {"merge_commit_sha": "abc123sha"} @pytest.mark.asyncio async def test_pr_merge_into_default_branch_is_ceo_only() -> None: """The agent merge path refuses to merge into a repo's default branch.""" - fake_task = MagicMock(project_id=uuid4(), parent_task_id=None, assigned_to=uuid4()) + project_id = uuid4() + fake_task = MagicMock( + project_id=project_id, parent_task_id=None, assigned_to=uuid4() + ) fake_project = MagicMock(slug="roboco") result = MagicMock() result.scalar_one_or_none.return_value = fake_task @@ -573,7 +578,7 @@ async def test_pr_merge_into_default_branch_is_ceo_only() -> None: _patch_project_service(fake_project), pytest.raises(UnauthorizedError, match="CEO_ONLY"), ): - await svc.pr_merge(11, target="master") + await svc.pr_merge(11, target="master", project_id=project_id) # Guard fires before any GitHub merge call. merge_api.assert_not_called() diff --git a/tests/unit/services/test_pr_merge_concurrency.py b/tests/unit/services/test_pr_merge_concurrency.py index 39e2f69e..6d6d3eac 100644 --- a/tests/unit/services/test_pr_merge_concurrency.py +++ b/tests/unit/services/test_pr_merge_concurrency.py @@ -110,7 +110,9 @@ async def test_pr_merge_retries_once_on_409_conflict() -> None: _bind(svc, "_sync_target_branch", sync_branch) with _patch_project_service(fake_project): - out = await svc.pr_merge(11, target="feature/backend/parent") + out = await svc.pr_merge( + 11, target="feature/backend/parent", project_id=project_id + ) assert out == {"merge_commit_sha": "merged-sha"} # _call_merge_api invoked twice — once 409, once 200. @@ -151,7 +153,7 @@ async def test_pr_merge_raises_after_second_409() -> None: _bind(svc, "_sync_target_branch", AsyncMock(return_value="abc")) with _patch_project_service(fake_project), pytest.raises(GitError) as exc_info: - await svc.pr_merge(11, target="feature/backend/parent") + await svc.pr_merge(11, target="feature/backend/parent", project_id=project_id) assert "409" in str(exc_info.value) # No infinite retries — exactly two attempts. @@ -188,7 +190,7 @@ async def test_pr_merge_does_not_retry_on_non_409_error() -> None: _bind(svc, "_sync_target_branch", AsyncMock(return_value="abc")) with _patch_project_service(fake_project), pytest.raises(GitError): - await svc.pr_merge(11, target="feature/backend/parent") + await svc.pr_merge(11, target="feature/backend/parent", project_id=project_id) # Only one merge attempt — non-409 error path skips retry. assert call_seq.await_count == 1 @@ -224,7 +226,7 @@ async def test_pr_merge_locks_parent_task_with_for_update() -> None: _bind(svc, "_sync_target_branch", AsyncMock(return_value="abc")) with _patch_project_service(fake_project): - await svc.pr_merge(11, target="feature/backend/parent") + await svc.pr_merge(11, target="feature/backend/parent", project_id=project_id) # Two SELECTs: PR lookup (no lock) + parent lock (FOR UPDATE). assert session.execute.await_count == _EXPECTED_SELECT_CALLS @@ -272,8 +274,60 @@ async def test_pr_merge_skips_parent_lock_for_root_task() -> None: _bind(svc, "_sync_target_branch", AsyncMock(return_value="abc")) with _patch_project_service(fake_project): - out = await svc.pr_merge(11, target="master") + out = await svc.pr_merge(11, target="master", project_id=project_id) assert out == {"merge_commit_sha": "abc"} # Only the PR-lookup SELECT runs — no second SELECT for parent lock. assert session.execute.await_count == 1 + + +# --------------------------------------------------------------------------- +# Regression: PR numbers are per-repo on GitHub but stored globally in +# tasks.pr_number with no uniqueness/repo scoping. Two tasks on different +# projects/repos can share a pr_number, so a bare `where(pr_number == +# X).limit(1)` resolves non-deterministically — it merged the wrong repo's +# PR and marked the WRONG task's work session merged, leaving the real +# task's session `pr_status="open"` so `complete()` rejected (returned +# None) and the cell PM 500'd on `t.status` and thrashed. The task lookup +# MUST be scoped by the caller's project_id (mirrors close_pull_request). +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_pr_merge_scopes_task_lookup_by_project_id() -> None: + project_id = uuid4() + fake_task = MagicMock( + id=uuid4(), + project_id=project_id, + parent_task_id=None, # root — no parent lock SELECT + assigned_to=uuid4(), + work_session_id=None, + ) + fake_project = MagicMock(slug="roboco") + + session = MagicMock() + pr_result = MagicMock() + pr_result.scalar_one_or_none.return_value = fake_task + session.execute = AsyncMock(return_value=pr_result) + session.commit = AsyncMock() + session.rollback = AsyncMock() + session.flush = AsyncMock() + + svc = GitService(session) + _bind(svc, "get_workspace", AsyncMock(return_value=Path("/tmp/ws"))) + _bind(svc, "_get_project_token_or_raise", AsyncMock(return_value="tok")) + _bind(svc, "_parse_github_remote", MagicMock(return_value=("acme", "repo"))) + _bind(svc, "_call_merge_api", AsyncMock(return_value=_fake_response(200))) + _bind(svc, "_delete_pr_branch_best_effort", AsyncMock()) + _bind(svc, "_sync_target_branch", AsyncMock(return_value="sha")) + + with _patch_project_service(fake_project): + await svc.pr_merge(11, target="feature/x", project_id=project_id) + + lookup_stmt = session.execute.await_args_list[0].args[0] + sql = str(lookup_stmt.compile(compile_kwargs={"literal_binds": True})) + assert "pr_number" in sql + assert "project_id" in sql + # The scoped project_id value is bound into the WHERE (Postgres renders + # UUID literals without hyphens), not just the column name. + assert project_id.hex in sql