[F125] serialize same-parent delegate via per-parent advisory lock

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).

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)