From 145b2828ba952c46f63355c50a29889a146fa126 Mon Sep 17 00:00:00 2001 From: Renn F Date: Sun, 28 Jun 2026 23:58:00 +0200 Subject: [PATCH] [F125] serialize same-parent delegate via per-parent advisory lock MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The delegate sibling-dedup guard read the parent's existing subtasks via an unlocked get_subtasks SELECT (the dedup read) then created the subtask (the write) with no DB serialization between them. Two concurrent delegate calls for the same parent (PM re-delegating while a reaper re-dispatches, or two orchestrator ticks racing) each read a duplicate-free sibling set, each passed the dedup guard, and each created a subtask — the parent got the duplicate the guard exists to prevent (the smoke-run runaway pattern). Fix: a PostgreSQL transaction-scoped advisory lock keyed by the parent task id (seed 1, disjoint from the per-agent claim lock's seed 0), acquired at the top of the delegate body before the first get_subtasks read (the briefing context read AND the dedup sibling read) and held through create_subtask's flush + the outer request commit. The second concurrent same-parent delegate blocks until the first commits, then its dedup read sees the committed sibling and is rejected. Per-PARENT (not per-agent): a coordinator PM legitimately delegates many subtasks under one parent in quick succession and plans many roots in parallel — a per-agent lock would serialize all of a PM's delegates and regress the PM coordinator concurrency feature. The per-parent lock serializes only same-parent delegates (the dedup invariant is per-parent) and leaves different parents untouched. TDD: red-first ordering test (lock acquired before first get_subtasks read and before create_subtask) + no-regression test (create still runs). --- .../services/gateway/choreographer/_impl.py | 14 ++ roboco/services/task.py | 36 ++++ .../unit/gateway/test_delegate_parent_lock.py | 181 ++++++++++++++++++ 3 files changed, 231 insertions(+) create mode 100644 tests/unit/gateway/test_delegate_parent_lock.py diff --git a/roboco/services/gateway/choreographer/_impl.py b/roboco/services/gateway/choreographer/_impl.py index e15543f0..64462713 100644 --- a/roboco/services/gateway/choreographer/_impl.py +++ b/roboco/services/gateway/choreographer/_impl.py @@ -4165,6 +4165,20 @@ class Choreographer: task_id=parent_task_id, verb="delegate", ) + # Serialize same-parent delegates across the sibling-dedup read -> + # create_subtask write critical section. The dedup guard reads the + # parent's existing subtasks via an unlocked SELECT, then the verb + # body creates the subtask — with no DB serialization two concurrent + # same-parent delegates each read a duplicate-free set and each + # create a subtask (the duplicate the guard exists to prevent). The + # per-parent transaction-scoped advisory lock is held until the outer + # request commits, so the second same-parent delegate blocks until + # the first commits and its dedup read then sees the committed + # sibling. Acquired before the first get_subtasks read (the briefing + # context read AND the dedup sibling read) so it spans the whole + # section. Per-parent (not per-agent) so a coordinator PM's parallel + # root planning is not serialized — only same-parent delegates are. + await self.task.acquire_delegate_parent_lock(parent_task_id) agent = await self.task.agent_for(pm_agent_id) role_str = str(agent.role) if agent is not None else "cell_pm" briefing = await self._briefing_for(pm_agent_id, parent_task_id, task=parent) diff --git a/roboco/services/task.py b/roboco/services/task.py index 731f84c1..7eeb92f7 100644 --- a/roboco/services/task.py +++ b/roboco/services/task.py @@ -2370,6 +2370,42 @@ class TaskService(BaseService): {"aid": str(agent_id)}, ) + async def acquire_delegate_parent_lock(self, parent_task_id: UUID) -> None: + """Take a per-parent transaction-scoped advisory lock for delegate. + + The sibling-dedup guard (``_delegate_sibling_dedup_guard``) reads the + parent's existing subtasks via an unlocked ``get_subtasks`` SELECT, + then the verb body calls ``create_subtask`` (the write) — with no DB + serialization between the two. Two concurrent ``delegate`` calls for + the SAME parent (a PM re-delegating while a reaper re-dispatches, or + two orchestrator ticks racing) each read a duplicate-free sibling set, + each pass the guard, and each create a subtask → the parent gets the + duplicate the guard exists to prevent. A ``pg_advisory_xact_lock`` + keyed by the parent serializes the WHOLE dedup-read -> create critical + section per parent — held until the outer request transaction commits, + the second concurrent same-parent delegate blocks until the first + commits, then its dedup read sees the first's committed sibling and is + rejected. + + Per-PARENT (not per-agent): a coordinator PM legitimately delegates many + subtasks under one parent in quick succession and plans many roots in + parallel — a per-agent lock would serialize all of a PM's delegates and + regress coordinator concurrency. The per-parent lock serializes only + same-parent delegates (the dedup invariant is per-parent) and leaves + different parents untouched. Seed ``1`` keeps this in a disjoint key + space from the per-agent claim lock (seed ``0``). Transaction-scoped so + it auto-releases on commit or rollback and cannot outlive the request. + ``hashtextextended`` maps a UUID to a ``bigint`` key; a hash collision + only causes benign false serialization (two different parents + momentarily serializing), never a false negative. + """ + await self.session.execute( + text( + "SELECT pg_advisory_xact_lock(hashtextextended(CAST(:pid AS text), 1))" + ), + {"pid": str(parent_task_id)}, + ) + async def claim( self, task_id: UUID, agent_id: UUID, allow_reassign: bool = False ) -> TaskTable | None: diff --git a/tests/unit/gateway/test_delegate_parent_lock.py b/tests/unit/gateway/test_delegate_parent_lock.py new file mode 100644 index 00000000..cf2cd74b --- /dev/null +++ b/tests/unit/gateway/test_delegate_parent_lock.py @@ -0,0 +1,181 @@ +"""F125 — the delegate sibling-dedup guard had a read/write TOCTOU. + +``_delegate_sibling_dedup_guard`` reads the parent's existing subtasks via an +unlocked ``get_subtasks`` SELECT (the dedup read), then the verb body calls +``create_subtask`` (the write) — with no DB serialization between the two. Two +concurrent ``delegate`` calls for the SAME parent (a PM re-delegating while a +stale-heartbeat reaper unclaims + re-dispatches, or two orchestrator ticks +racing) each read an empty/duplicate-free sibling set, each pass the dedup +guard, and each create a subtask → the parent gets the duplicate the guard +exists to prevent (the smoke-run runaway pattern the guard was built for). + +The fix: a PostgreSQL transaction-scoped advisory lock keyed by the parent +task id, acquired at the TOP of the delegate body — before the first +``get_subtasks`` read (the briefing's context read AND the dedup guard's +sibling read) and held through ``create_subtask``'s flush + the outer request +commit. The second concurrent same-parent delegate blocks on the lock until +the first commits; its dedup read then sees the first's committed sibling and +is rejected. Different parents hash to different keys (seed ``1``, disjoint +from the per-agent claim lock's seed ``0``) so cross-parent delegates are not +serialized — the PM coordinator concurrency feature (parallel root planning) +is preserved. + +CRITICAL logical-regression guard: the lock is per-PARENT, not per-agent. A +single cell_pm / main_pm legitimately delegates many subtasks under one parent +in quick succession (a per-dev sequenced queue), and a coordinator PM plans +many roots in parallel. A per-agent lock would serialize all of a PM's +delegates and regress coordinator concurrency; a per-parent lock serializes +only same-parent delegates (the actual dedup invariant is per-parent) and +leaves different parents untouched. +""" + +from __future__ import annotations + +from datetime import UTC, datetime +from typing import Any +from unittest.mock import AsyncMock, MagicMock +from uuid import uuid4 + +import pytest +from roboco.services.gateway.choreographer import ( + Choreographer, + ChoreographerDeps, + DelegateInputs, +) + + +def _make_deps(task: AsyncMock) -> ChoreographerDeps: + base: dict[str, Any] = { + "task": task, + "work_session": AsyncMock(), + "git": AsyncMock(), + "a2a": AsyncMock(), + "journal": AsyncMock(), + "audit": AsyncMock(), + "evidence_repo": AsyncMock(), + } + repo = base["evidence_repo"] + for m 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, m).return_value = [] + # A fresh decision within the recency window so the delegate tracing gate + # (journal:decision required) passes without a separate write. + base["journal"].latest_decision_at.return_value = datetime.now(UTC) + return ChoreographerDeps(**base) + + +def _parent(pm_id: object) -> MagicMock: + return MagicMock( + id=uuid4(), + project_id=uuid4(), + product_id=None, + status="in_progress", + assigned_to=pm_id, + # delegate obligates the PM's quick_context resumption section. + quick_context="Decomposition planned; cells implement their slice next.", + ) + + +def _inputs() -> DelegateInputs: + return DelegateInputs( + title="Implement endpoint", + description="Add /v1/foo endpoint with tests", + assigned_to="be-dev-1", + team="backend", + task_type="code", + nature="technical", + acceptance_criteria=["GET /v1/foo returns 200 with body"], + ) + + +@pytest.mark.asyncio +async def test_delegate_acquires_parent_lock_before_sibling_read() -> None: + """The per-parent advisory lock MUST be acquired before the first + ``get_subtasks`` read (the briefing's context read, which precedes the + dedup guard's sibling read) and held through ``create_subtask``. This is + the ordering that closes the TOCTOU: the second concurrent same-parent + delegate blocks on the lock before it can read siblings, so its dedup read + sees the first's committed subtask and is rejected. A lock acquired AFTER + the dedup read but before the create would NOT close the race (the read + already missed the concurrent insert) — so 'lock before create' alone is + insufficient; the lock must precede the read.""" + pm_id = uuid4() + parent = _parent(pm_id) + task_svc = AsyncMock() + task_svc.get.return_value = parent + task_svc.agent_for.return_value = MagicMock(role="cell_pm", team="backend") + task_svc.create_subtask.return_value = MagicMock(id=uuid4()) + + # Shared call-order recorder: the lock must precede every get_subtasks + # read (briefing context + dedup siblings) and the create. + calls: list[str] = [] + + async def _lock(_pid: object) -> None: + calls.append("lock") + + async def _read_subtasks(_pid: object) -> list[Any]: + calls.append("get_subtasks") + return [] + + async def _create_subtask(_req: object) -> Any: + calls.append("create") + return MagicMock(id=uuid4()) + + task_svc.acquire_delegate_parent_lock = _lock + task_svc.get_subtasks.side_effect = _read_subtasks + task_svc.create_subtask.side_effect = _create_subtask + + deps = _make_deps(task_svc) + c = Choreographer(deps) + + env = await c.delegate(pm_id, parent.id, _inputs()) + assert env.error is None, env.as_dict() + + # The flow reached the create (otherwise the lock-ordering assertion would + # pass for the wrong reason — a short-circuit before the create). + assert "create" in calls, calls + + # The lock was acquired exactly once, before the first sibling read, and + # before the create — so it spans the dedup read -> create critical section. + assert calls.count("lock") == 1, calls + first_lock = calls.index("lock") + first_read = calls.index("get_subtasks") + first_create = calls.index("create") + assert first_lock < first_read, ( + f"parent lock must be acquired before the first get_subtasks read; " + f"order was {calls}" + ) + assert first_lock < first_create, ( + f"parent lock must be held through create_subtask; order was {calls}" + ) + + +@pytest.mark.asyncio +async def test_delegate_still_creates_subtask_with_parent_lock() -> None: + """No-regression: acquiring the per-parent lock must not break the normal + delegate path — the subtask is still created (env.error is None, + create_subtask awaited once). The lock is transparent to the happy path.""" + pm_id = uuid4() + parent = _parent(pm_id) + task_svc = AsyncMock() + task_svc.get.return_value = parent + task_svc.agent_for.return_value = MagicMock(role="cell_pm", team="backend") + task_svc.get_subtasks.return_value = [] + task_svc.create_subtask.return_value = MagicMock(id=uuid4()) + # Leave the default AsyncMock for acquire_delegate_parent_lock so we can + # assert it was awaited with the parent id (the lock is transparent to the + # happy path — the create still runs). + deps = _make_deps(task_svc) + c = Choreographer(deps) + + env = await c.delegate(pm_id, parent.id, _inputs()) + assert env.error is None, env.as_dict() + task_svc.create_subtask.assert_awaited_once() + task_svc.acquire_delegate_parent_lock.assert_awaited_once_with(parent.id)