[feature] wire dev-task collision DAG at cell-PM delegation (sequencing S2)

Pure dev_task_collision_edges in sequencing.py turns a parent's surfaced
siblings into (depends_on_id, task_id) pairs via SequencingService. TaskService.
wire_sibling_collision_dag wires them through add_dependency (idempotent). The
choreographer calls it after each dev-task delegate so the sibling collision DAG
is built incrementally as the cell PM decomposes — file-overlap serializes,
migration chains, shared-last; stable (priority, sequence) ordering keeps edges
from flipping into reverse cycles on re-runs.

tests/integration/test_task_service_transitions.py

@@ -720,6 +720,60 @@ async def test_create_subtask_round_trips_collision_surfaces_and_deps(
     assert sub.dependency_ids == [dep_id]
 
 
+@pytest.mark.asyncio
+async def test_wire_sibling_collision_dag_serializes_overlapping_dev_tasks(
+    task_setup: dict, db_session: AsyncSession
+) -> None:
+    """wire_sibling_collision_dag runs the collision analyzer over a parent's
+    surfaced dev-task siblings and wires dependency_ids so a later dev task
+    whose surface overlaps an earlier one stays PENDING until it completes.
+
+    T1 (a.py) and T2 (b.py) are disjoint -> parallel (no edge). T3 (a.py)
+    overlaps T1 -> T3 depends-on T1. The explicit `depends_on` override on T3
+    (forwarded through create_subtask in S1) is also present."""
+    svc = task_setup["svc"]
+    parent = await svc.create(_req(task_setup))
+    await db_session.flush()
+
+    async def _dev(seq: int, surface: list[str]) -> Any:
+        t = await svc.create_subtask(
+            TaskCreateRequest(
+                title=f"dev-{seq}",
+                description=f"dev task {seq} description long enough",
+                acceptance_criteria=["ac"],
+                team=Team.BACKEND,
+                created_by=task_setup["agent_id"],
+                project_id=task_setup["project_id"],
+                parent_task_id=parent.id,
+                task_type=TaskType.CODE,
+                nature=TaskNature.TECHNICAL,
+                sequence=seq,
+                intends_to_touch=surface,
+            )
+        )
+        await svc.set_sequence(t.id, seq)
+        return t
+
+    t1 = await _dev(0, ["roboco/api/a.py"])
+    t2 = await _dev(1, ["roboco/api/b.py"])
+    t3 = await _dev(2, ["roboco/api/a.py"])
+
+    await svc.wire_sibling_collision_dag(parent.id)
+
+    # Reload to read the wired dependency_ids.
+    r1 = await svc.get(t1.id)
+    r2 = await svc.get(t2.id)
+    r3 = await svc.get(t3.id)
+    assert r1 is not None and r2 is not None and r3 is not None
+    # T1 and T2 are disjoint -> parallel (no collision edge between them).
+    assert r1.dependency_ids == []
+    assert r2.dependency_ids == []
+    # T3 overlaps T1 (same file, same repo) -> T3 depends-on T1.
+    assert t1.id in r3.dependency_ids
+    # T2 does NOT collide with T3 (disjoint file) -> no T2->T3 edge.
+    assert t2.id not in r3.dependency_ids
+
+
 @pytest.mark.asyncio
 async def test_fail_qa_work_session_fallback_excludes_qa_session(
     task_setup: dict, db_session: AsyncSession

tests/unit/services/test_sequencing.py

@@ -9,12 +9,15 @@ the shared threat service, and S1/S2/S7 in one parallel wave.
 
 from __future__ import annotations
 
+from dataclasses import dataclass, field
+from uuid import uuid4
+
 import pytest
 from roboco.foundation.policy.sequencing.models import (
     DraftSurface,
     SequencingError,
 )
-from roboco.services.sequencing import SequencingService
+from roboco.services.sequencing import SequencingService, dev_task_collision_edges
 
 
 def _backend(_i: int) -> str:
@@ -181,3 +184,102 @@ def test_golden_reproduces_ceo_waves() -> None:
     # The page-isolated frontend work (S1/S2/S7) lands in one parallel wave.
     assert _wave_of(plan.waves, S1) == _wave_of(plan.waves, S2)
     assert _wave_of(plan.waves, S2) == _wave_of(plan.waves, S7)
+
+
+# ---------------------------------------------------------------------------
+# dev_task_collision_edges — the dev-task collision DAG (edge kind 3).
+# Pure glue: a parent's surfaced siblings -> (depends_on_id, task_id) pairs.
+# Wraps SequencingService so the choreographer can wire the DAG via add_dependency
+# at cell-PM dev-delegation time (incremental, idempotent). See the multi-level
+# sequencing design doc.
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class _Sib:
+    """Minimal sibling shape — the attributes dev_task_collision_edges reads."""
+
+    id: object
+    priority: int = 2
+    sequence: int = 0
+    intends_to_touch: list[str] = field(default_factory=list)
+    adds_migration: bool = False
+    touches_shared: bool = False
+    project_id: str | None = "proj-backend"
+
+
+def _edge_set(pairs: list[tuple[object, object]]) -> set[tuple[object, object]]:
+    return set(pairs)
+
+
+def test_dev_collision_disjoint_surfaces_are_parallel() -> None:
+    # Same project, disjoint files → no edge (the two dev tasks run together).
+    a, b = (
+        _Sib(uuid4(), sequence=0, intends_to_touch=["a.py"]),
+        _Sib(uuid4(), sequence=1, intends_to_touch=["b.py"]),
+    )
+    assert dev_task_collision_edges([a, b]) == []
+
+
+def test_dev_collision_overlap_serializes_more_important_first() -> None:
+    # Both touch a.py → serialized; lower priority NUMBER runs first.
+    first = _Sib(uuid4(), priority=1, sequence=0, intends_to_touch=["a.py"])
+    second = _Sib(uuid4(), priority=2, sequence=1, intends_to_touch=["a.py"])
+    edges = dev_task_collision_edges([second, first])  # passed out of order
+    assert edges == [
+        (first.id, second.id)
+    ]  # first depends-on nothing; second depends-on first
+
+
+def test_dev_collision_overlap_equal_priority_uses_sequence() -> None:
+    # Equal priority → lower sequence runs first (stable across incremental re-runs).
+    t1 = _Sib(uuid4(), sequence=0, intends_to_touch=["a.py"])
+    t3 = _Sib(uuid4(), sequence=1, intends_to_touch=["a.py"])
+    assert dev_task_collision_edges([t1, t3]) == [(t1.id, t3.id)]
+
+
+def test_dev_collision_skips_unsurfaced_siblings() -> None:
+    # A sibling with no surface is parallel to everything (no edges to/from it).
+    surfaced = _Sib(uuid4(), sequence=0, intends_to_touch=["a.py"])
+    bare = _Sib(uuid4(), sequence=1)  # no intends_to_touch / migration / shared
+    other = _Sib(uuid4(), sequence=2, intends_to_touch=["a.py"])
+    edges = _edge_set(dev_task_collision_edges([surfaced, bare, other]))
+    assert edges == {(surfaced.id, other.id)}
+    assert bare.id not in {e[0] for e in edges} and bare.id not in {e[1] for e in edges}
+
+
+def test_dev_collision_skips_different_project() -> None:
+    # Same path, different repo → no collision (different codebase).
+    a = _Sib(uuid4(), sequence=0, intends_to_touch=["a.py"], project_id="proj-be")
+    b = _Sib(uuid4(), sequence=1, intends_to_touch=["a.py"], project_id="proj-fe")
+    assert dev_task_collision_edges([a, b]) == []
+
+
+def test_dev_collision_migration_chain_serializes() -> None:
+    # Two migration-adders in the same repo chain serially (alembic single-head).
+    m1 = _Sib(uuid4(), sequence=0, adds_migration=True, intends_to_touch=["m1.py"])
+    m2 = _Sib(uuid4(), sequence=1, adds_migration=True, intends_to_touch=["m2.py"])
+    assert dev_task_collision_edges([m1, m2]) == [(m1.id, m2.id)]
+
+
+def test_dev_collision_shared_last_after_non_shared_overlap() -> None:
+    # A touches_shared edit runs after a non-shared task that overlaps it.
+    base = _Sib(uuid4(), sequence=0, intends_to_touch=["svc/shared.py"])
+    shared = _Sib(
+        uuid4(), sequence=1, touches_shared=True, intends_to_touch=["svc/shared.py"]
+    )
+    assert dev_task_collision_edges([base, shared]) == [(base.id, shared.id)]
+
+
+def test_dev_collision_single_surfaced_sibling_no_edge() -> None:
+    solo = _Sib(uuid4(), sequence=0, intends_to_touch=["a.py"])
+    assert dev_task_collision_edges([solo]) == []
+
+
+def test_dev_collision_returns_depends_on_first_pairs() -> None:
+    # Contract: each pair is (depends_on_id, task_id) — task depends-on depends_on.
+    first = _Sib(uuid4(), sequence=0, intends_to_touch=["a.py"])
+    second = _Sib(uuid4(), sequence=1, intends_to_touch=["a.py"])
+    [(dep, task)] = dev_task_collision_edges([first, second])
+    assert dep == first.id
+    assert task == second.id