[feature] delegate carries dev-task collision surface (sequencing S1)

The cell/main PM's delegate verb now carries the dev-task collision
surface (intends_to_touch / adds_migration / touches_shared) and an
explicit depends_on override through DelegateRequest -> DelegateInputs
-> _create_subtask_from_inputs -> create_subtask, and create_subtask
forwards sequence / dependency_ids / batch_id / surfaces into the
prepared TaskCreateRequest instead of dropping them (the base create
already persists them at task.py:878-884).

This is the plumbing for the multi-level sequencing model edge kind 3
(dev-task collision DAG). Previously a dev task delegated with a
collision surface or an explicit dependency lost it before persistence
— dependency_ids was always [], so the only dev-task ordering was the
weak assignee-keyed spawn barrier (the live 2026-06-27 out-of-order
break: 40842957 started before 9b3682b8's PR merged). Phase S2 runs
SequencingService over the surfaced siblings and wires the DAG via
add_dependency.

roboco/api/routes/v1/flow_cell_pm.py

@@ -85,6 +85,10 @@ async def delegate(
         estimated_complexity=body.estimated_complexity,
         project_id=body.project_id,
         covers_parent_criteria=body.covers_parent_criteria,
+        intends_to_touch=body.intends_to_touch,
+        adds_migration=body.adds_migration,
+        touches_shared=body.touches_shared,
+        depends_on=body.depends_on,
     )
     env = await choreographer.delegate(x_agent_id, body.parent_task_id, inputs)
     return envelope_to_response(env, request)

roboco/api/routes/v1/flow_main_pm.py

@@ -85,6 +85,10 @@ async def delegate(
         estimated_complexity=body.estimated_complexity,
         project_id=body.project_id,
         covers_parent_criteria=body.covers_parent_criteria,
+        intends_to_touch=body.intends_to_touch,
+        adds_migration=body.adds_migration,
+        touches_shared=body.touches_shared,
+        depends_on=body.depends_on,
     )
     env = await choreographer.delegate(x_agent_id, body.parent_task_id, inputs)
     return envelope_to_response(env, request)

roboco/api/schemas/v1/flow.py

@@ -275,6 +275,18 @@ class DelegateRequest(BaseModel):
     # Parent acceptance-criterion ids this subtask is responsible for. Lets the
     # coverage + roll-up AC gates verify every parent AC is claimed and satisfied.
     covers_parent_criteria: StrList | None = None
+    # Dev-task collision surface (the multi-level sequencing model — edge kind
+    # 3). The cell PM states what each dev task touches so the choreographer can
+    # run SequencingService and wire the dev-task collision DAG (file-overlap
+    # serializes, migration-adders chain, shared-surface edits run last).
+    # Optional: a delegate without surfaces joins no collision edges (parallel).
+    intends_to_touch: StrList | None = None
+    adds_migration: bool = False
+    touches_shared: bool = False
+    # Explicit dependency override (edge the surface rules would miss, e.g. a
+    # non-collision ordering the PM knows). Optional; wired verbatim as
+    # dependency_ids on the created dev task.
+    depends_on: list[UUID] | None = None
 
     # Pre-gateway parity: cross-field validators that catch the most common
     # LLM-vs-schema confusions. Pre-gateway lived in

roboco/services/gateway/choreographer/_impl.py

@@ -323,6 +323,16 @@ class DelegateInputs:
     # Parent AC ids this subtask is responsible for — the decomposition coverage
     # link. Empty/None means the child covers no specific parent criteria yet.
     covers_parent_criteria: list[str] | None = None
+    # Dev-task collision surface (multi-level sequencing — edge kind 3). The
+    # cell PM states what each dev task touches so the choreographer can run
+    # SequencingService and wire the dev-task collision DAG. Optional: a
+    # delegate without surfaces joins no collision edges (parallel).
+    intends_to_touch: list[str] | None = None
+    adds_migration: bool = False
+    touches_shared: bool = False
+    # Explicit dependency override — wired verbatim as dependency_ids on the
+    # created dev task (an edge the surface rules would miss).
+    depends_on: list[UUID] | None = None
 
 
 class Choreographer:
@@ -4846,6 +4856,14 @@ class Choreographer:
             task_type=type_enum,
             nature=nature_enum,
             estimated_complexity=complexity_enum,
+            # Dev-task collision surface (multi-level sequencing — edge kind 3)
+            # + explicit dependency override. Forwarded so create_subtask can
+            # persist them (Phase S2 runs SequencingService over the surfaced
+            # siblings and wires the collision DAG via add_dependency).
+            intends_to_touch=inputs.intends_to_touch,
+            adds_migration=inputs.adds_migration,
+            touches_shared=inputs.touches_shared,
+            dependency_ids=list(inputs.depends_on) if inputs.depends_on else [],
         )
         new_task = await self.task.create_subtask(req)
         # Assign a distinct ordinal within the parent's siblings so the merge

roboco/services/task.py

@@ -8148,6 +8148,17 @@ class TaskService(BaseService):
             task_type=req.task_type,
             nature=req.nature,
             status=req.status or inferred_status,
+            # Forward ordering + collision surface so a dev task delegated with
+            # surfaces/dependencies keeps them (multi-level sequencing — edge
+            # kinds 3 & 4). Previously dropped here, which is why dev-task
+            # dependency_ids was always [] and the only ordering was the weak
+            # assignee-keyed spawn barrier (live 2026-06-27 out-of-order break).
+            sequence=req.sequence,
+            dependency_ids=list(req.dependency_ids) if req.dependency_ids else [],
+            batch_id=req.batch_id,
+            intends_to_touch=req.intends_to_touch,
+            adds_migration=req.adds_migration,
+            touches_shared=req.touches_shared,
         )
         return await self.create(prepared)
 

tests/integration/test_task_service_transitions.py

@@ -676,6 +676,50 @@ async def test_fail_qa_routes_to_dev_via_work_session_when_marker_missing(
     assert markers.get_original_developer(failed) == str(dev_id)
 
 
+@pytest.mark.asyncio
+async def test_create_subtask_round_trips_collision_surfaces_and_deps(
+    task_setup: dict, db_session: AsyncSession
+) -> None:
+    """create_subtask forwards intends_to_touch / adds_migration /
+    touches_shared / sequence / dependency_ids onto the created subtask.
+
+    Previously these were dropped inside create_subtask's `prepared`
+    TaskCreateRequest, so a dev task delegated with a collision surface or an
+    explicit dependency lost it before persistence — the root cause of
+    dev-task dependency_ids always being [] (the live 2026-06-27 out-of-order
+    break). The base ``create`` persists them (task.py:878-884); this test
+    locks the forwarding through create_subtask.
+    """
+    svc = task_setup["svc"]
+    parent = await svc.create(_req(task_setup))
+    await db_session.flush()
+    dep_id = uuid4()
+    sub = await svc.create_subtask(
+        TaskCreateRequest(
+            title="child",
+            description="child description with enough length",
+            acceptance_criteria=["ac1"],
+            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=3,
+            dependency_ids=[dep_id],
+            intends_to_touch=["roboco/services/foo.py"],
+            adds_migration=True,
+            touches_shared=True,
+        )
+    )
+    assert sub.intends_to_touch == ["roboco/services/foo.py"]
+    assert sub.adds_migration is True
+    assert sub.touches_shared is True
+    expected_sequence = 3
+    assert sub.sequence == expected_sequence
+    assert sub.dependency_ids == [dep_id]
+
+
 @pytest.mark.asyncio
 async def test_fail_qa_work_session_fallback_excludes_qa_session(
     task_setup: dict, db_session: AsyncSession

tests/unit/api/routes/v1/test_flow_cell_pm.py

@@ -7,7 +7,7 @@ No DB required — Choreographer is mocked.
 from __future__ import annotations
 
 from unittest.mock import AsyncMock, MagicMock
-from uuid import uuid4
+from uuid import UUID, uuid4
 
 import pytest
 from fastapi import FastAPI
@@ -296,6 +296,50 @@ async def test_delegate_dispatches_inputs_bundle() -> None:
     assert inputs.task_type == "code"
 
 
+@pytest.mark.asyncio
+async def test_delegate_forwards_collision_surfaces_to_inputs() -> None:
+    """The dev-task collision surface (intends_to_touch / adds_migration /
+    touches_shared) and an explicit depends_on override must round-trip from
+    the HTTP body through DelegateInputs so the cell PM can express the
+    dev-task collision DAG (the missing edge kind 3 — see the multi-level
+    sequencing design). Today these fields do not exist on DelegateRequest /
+    DelegateInputs, so the test would 422 / AttributeError — the red signal.
+    """
+    mock_chore = MagicMock()
+    mock_chore.delegate = AsyncMock(
+        return_value=_make_envelope(status="created", task_id=_TASK_ID)
+    )
+    client = TestClient(_build_app(mock_chore))
+    dep_id = str(uuid4())
+
+    resp = client.post(
+        "/api/v1/flow/cell_pm/delegate",
+        json={
+            "parent_task_id": _TASK_ID,
+            "title": "Implement /v1/foo",
+            "description": "Add the foo endpoint with passing tests.",
+            "assigned_to": "be-dev-1",
+            "team": "backend",
+            "task_type": "code",
+            "nature": "technical",
+            "estimated_complexity": "medium",
+            "acceptance_criteria": ["GET /v1/foo returns 200 with body"],
+            "intends_to_touch": ["roboco/api/routes/v1/foo.py"],
+            "adds_migration": True,
+            "touches_shared": False,
+            "depends_on": [dep_id],
+        },
+        headers=_HEADERS,
+    )
+
+    assert resp.status_code == _HTTP_200, resp.text
+    inputs = mock_chore.delegate.await_args.args[2]
+    assert inputs.intends_to_touch == ["roboco/api/routes/v1/foo.py"]
+    assert inputs.adds_migration is True
+    assert inputs.touches_shared is False
+    assert inputs.depends_on == [UUID(dep_id)]
+
+
 @pytest.mark.asyncio
 async def test_submit_up_dispatches_notes() -> None:
     """POST /api/v1/flow/cell_pm/submit_up forwards task_id and notes."""

tests/unit/api/routes/v1/test_flow_main_pm.py

@@ -7,7 +7,7 @@ No DB required — Choreographer is mocked.
 from __future__ import annotations
 
 from unittest.mock import AsyncMock, MagicMock
-from uuid import uuid4
+from uuid import UUID, uuid4
 
 import pytest
 from fastapi import FastAPI
@@ -273,6 +273,49 @@ async def test_delegate_to_cell_pm_dispatches_inputs_bundle() -> None:
     assert inputs.task_type == "planning"
 
 
+@pytest.mark.asyncio
+async def test_delegate_forwards_collision_surfaces_to_inputs() -> None:
+    """Main-PM delegate forwards the dev-task collision surface + depends_on
+    override through DelegateInputs (parity with the cell-PM route — the
+    main PM delegates cell-tasks/dev-tasks the same way)."""
+    mock_chore = MagicMock()
+    mock_chore.delegate = AsyncMock(
+        return_value=_make_envelope(status="created", task_id=_TASK_ID)
+    )
+    client = TestClient(_build_app(mock_chore))
+    dep_id = str(uuid4())
+
+    resp = client.post(
+        "/api/v1/flow/main_pm/delegate",
+        json={
+            "parent_task_id": _TASK_ID,
+            "title": "Backend slice",
+            "description": "Plan + drive backend work for feature X end to end.",
+            "assigned_to": "be-pm",
+            "team": "backend",
+            "task_type": "planning",
+            "nature": "technical",
+            "estimated_complexity": "high",
+            "acceptance_criteria": [
+                "all subtasks created with acceptance criteria",
+                "branch + PR opened against the slice",
+            ],
+            "intends_to_touch": ["roboco/services/foo.py"],
+            "adds_migration": True,
+            "touches_shared": True,
+            "depends_on": [dep_id],
+        },
+        headers=_HEADERS,
+    )
+
+    assert resp.status_code == _HTTP_200, resp.text
+    inputs = mock_chore.delegate.await_args.args[2]
+    assert inputs.intends_to_touch == ["roboco/services/foo.py"]
+    assert inputs.adds_migration is True
+    assert inputs.touches_shared is True
+    assert inputs.depends_on == [UUID(dep_id)]
+
+
 @pytest.mark.asyncio
 async def test_escalate_to_ceo_dispatches() -> None:
     mock_chore = MagicMock()