From 9927d248ea5b1910033aa237fae05be4394dddd9 Mon Sep 17 00:00:00 2001 From: Renn F Date: Sun, 28 Jun 2026 04:22:27 +0200 Subject: [PATCH] [feature] wire cell-task wave chain + by-osmosis edge (sequencing S3) Kind 2 (cell-task wave chain): a new cell-task under root-subtask UT_n depends on every cell-task under every root-subtask in UT_n.dependency_ids (the kind-1 wave-chain edges), so its branch carries the previous wave's merged cell work. Re-derived from the root-subtask's deps, not the cell-task's own dependency_ids (which also carry UX/product-fanout edges the by-osmosis edge must not pick up). A root may fan to several cell-tasks (different cells), so the previous wave's cell-task is a SET. Kind 4 (by-osmosis): the first dev task (sequence 0) under a cell-task depends on each predecessor cell-task's tail (max-sequence) dev task, so the new wave's first branch carries the previous wave's fully-merged tail. Subsequent dev tasks inherit the tail via kind 3 or the merged base. Both wired from _create_subtask_from_inputs, dispatched on parent.team (MAIN_PM -> kind 2; cell team -> kind 4). Pure helpers (cell_task_wave_chain_depends_on, by_osmosis_tail_dev_tasks) unit-tested in test_sequencing.py; TaskService methods integration-tested. Idempotent + best-effort throughout (add_dependency dedupes; missing predecessors are no-ops). Also fixes a latent mypy-tests gap (estimated_complexity required on direct TaskCreateRequest calls in the S2 tests). --- .../services/gateway/choreographer/_impl.py | 19 +++ roboco/services/sequencing.py | 56 +++++++ roboco/services/task.py | 74 +++++++++ .../test_task_service_transitions.py | 144 ++++++++++++++++++ tests/unit/services/test_sequencing.py | 84 +++++++++- 5 files changed, 376 insertions(+), 1 deletion(-) diff --git a/roboco/services/gateway/choreographer/_impl.py b/roboco/services/gateway/choreographer/_impl.py index 71e9a797..ef9ded19 100644 --- a/roboco/services/gateway/choreographer/_impl.py +++ b/roboco/services/gateway/choreographer/_impl.py @@ -4884,6 +4884,25 @@ class Choreographer: # dev_task_collision_edges orders by (priority, sequence) so re-runs # only add edges (never flip an existing pair's order into a cycle). await self.task.wire_sibling_collision_dag(parent_task_id) + # Multi-level sequencing — edge kinds 2 + 4: the cell-task wave chain + # and the by-osmosis dev-task edge. Dispatch on the parent's team: a + # MAIN_PM parent is a root-subtask, so the new task is a CELL-TASK → + # wire kind 2 (it depends on the previous wave's cell-tasks); a cell + # team parent is a cell-task, so the new task is a DEV TASK → wire kind + # 4 (its first dev task carries the previous wave's merged tail). kind 3 + # (wire_sibling_collision_dag, above) runs for both. Idempotent + + # best-effort: missing predecessors / cell-tasks contribute no edge. + from roboco.foundation.identity import Team + + parent_team = getattr(parent, "team", None) + if parent_team == Team.MAIN_PM.value: + await self.task.wire_cell_task_wave_chain(new_task.id) + elif parent_team in ( + Team.BACKEND.value, + Team.FRONTEND.value, + Team.UX_UI.value, + ): + await self.task.wire_by_osmosis_edge(new_task.id) # Thread the parent's existing session links onto the # new subtask so the assigned agent (dev/qa/doc) lands in the # group chat the PM has already been talking in. Pre-gateway diff --git a/roboco/services/sequencing.py b/roboco/services/sequencing.py index 73c48e1c..2cc90acf 100644 --- a/roboco/services/sequencing.py +++ b/roboco/services/sequencing.py @@ -270,3 +270,59 @@ def dev_task_collision_edges(siblings: list) -> list[tuple[object, object]]: # any warning attributable. Empty capacity -> no warnings emitted. plan = SequencingService().analyze(surfaces, lambda _idx: "", {}) return [(surfaced[a].id, surfaced[b].id) for a, b in plan.edges] + + +# --------------------------------------------------------------------------- +# Multi-level sequencing — edge kinds 2 + 4 (cell-task wave chain + by-osmosis). +# Pure glue (no DB): the choreographer's TaskService wrappers walk the tree and +# hand the gathered objects to these helpers, which return the IDs to +# add_dependency. Pure so the edge logic is unit-testable without a database. +# --------------------------------------------------------------------------- + + +def cell_task_wave_chain_depends_on( + predecessor_root_ids: list, + cell_tasks_by_root: dict, +) -> list: + """Kind 2: the cell-task IDs a new cell-task should depend on. + + For each predecessor root-subtask (the kind-1 wave-chain edges on the new + cell-task's root-subtask — i.e. ``root.dependency_ids``), EVERY cell-task + under it. The new cell-task waits for the whole previous wave's cell work so + its branch carries the merged tail; a root-subtask may fan to several + cell-tasks (different cells), so the previous wave's "cell-task" is a SET, + not a single task. Idempotent by construction (``add_dependency`` dedupes); + over-serializes safely (a predecessor cell-task already terminal is a no-op + gate). A predecessor root with no cell tasks contributes nothing. + """ + deps: list = [] + for rid in predecessor_root_ids: + for ct in cell_tasks_by_root.get(rid, []): + deps.append(getattr(ct, "id", ct)) + return deps + + +def by_osmosis_tail_dev_tasks( + is_first_dev_task: bool, + predecessor_dev_task_groups: list, +) -> list: + """Kind 4: the tail dev-task IDs a new dev task should depend on. + + Only the FIRST dev task (``sequence == 0``) under a cell-task carries the + by-osmosis edge — its branch is cut first and must carry the previous wave's + fully-merged tail. Subsequent dev tasks inherit the tail via the kind-3 + collision DAG (they depend on earlier siblings) or share the cell branch's + already-merged base, so they need no explicit edge. "Tail" = the + highest-``sequence`` dev task under each predecessor cell-task; a + predecessor with no dev tasks contributes no edge. Idempotent + best-effort + (a tail already terminal is a no-op gate). + """ + if not is_first_dev_task: + return [] + tails: list = [] + for group in predecessor_dev_task_groups: + if not group: + continue + tail = max(group, key=lambda t: int(getattr(t, "sequence", 0))) + tails.append(getattr(tail, "id", tail)) + return tails diff --git a/roboco/services/task.py b/roboco/services/task.py index 16ef98c8..eb4a45ec 100644 --- a/roboco/services/task.py +++ b/roboco/services/task.py @@ -5961,6 +5961,80 @@ class TaskService(BaseService): for depends_on_id, task_id in edges: await self.add_dependency(UUID(str(task_id)), UUID(str(depends_on_id))) + async def wire_cell_task_wave_chain(self, cell_task_id: UUID) -> None: + """Wire the cell-task wave chain (multi-level sequencing edge kind 2). + + A new cell-task (under root-subtask ``UT_n``) depends on every cell-task + under every root-subtask ``UT_n`` itself depends on — ``UT_n`` 's + ``dependency_ids`` are the kind-1 wave-chain edges, so this chains the + cell-tasks of wave *k* onto the cell-tasks of wave *k-1*. A root-subtask + may fan to several cell-tasks (different cells), so the previous wave's + "cell-task" is a SET, not a single task, and the new cell-task waits for + the whole set so its branch carries the merged tail. + + ``UT_n`` is held PENDING by ``list_pending(filter_by_dependencies=True)`` + until its kind-1 predecessors are terminal, so by the time the Main PM + delegates ``UT_n`` 's cell-tasks the predecessor cell-tasks already exist + and are terminal — the edge is therefore a no-op gate in the common case + but documents the lineage and protects against any re-ordering. + Idempotent (``add_dependency`` dedupes) + best-effort (missing + predecessor / cell-task contributes no edge). + """ + from roboco.services.sequencing import cell_task_wave_chain_depends_on + + cell_task = await self.get(cell_task_id) + if cell_task is None or cell_task.parent_task_id is None: + return + root = await self.get(UUID(str(cell_task.parent_task_id))) + if root is None: + return + predecessor_root_ids = list(root.dependency_ids) + cell_tasks_by_root: dict = {} + for pred_root_id in predecessor_root_ids: + cell_tasks_by_root[pred_root_id] = await self.get_subtasks( + UUID(str(pred_root_id)) + ) + for dep_id in cell_task_wave_chain_depends_on( + predecessor_root_ids, cell_tasks_by_root + ): + await self.add_dependency(cell_task_id, UUID(str(dep_id))) + + async def wire_by_osmosis_edge(self, dev_task_id: UUID) -> None: + """Wire the by-osmosis edge (multi-level sequencing edge kind 4). + + The FIRST dev task (``sequence == 0``) under a cell-task depends on each + predecessor cell-task's tail (highest-``sequence``) dev task, so the new + wave's first branch carries the previous wave's fully-merged tail. The + predecessor cell-tasks are re-derived from the cell-task's root-subtask's + kind-1 ``dependency_ids`` (the same source :meth:`wire_cell_task_wave_chain` + uses), not from the cell-task's own ``dependency_ids`` — which also carry + UX/product-fanout deps the by-osmosis edge must not pick up. + + Subsequent dev tasks (``sequence > 0``) inherit the tail via the kind-3 + collision DAG or share the cell branch's already-merged base, so they + take no explicit edge. Idempotent + best-effort: a predecessor cell-task + with no dev tasks, or a missing predecessor, contributes no edge; a tail + already terminal is a no-op gate. + """ + from roboco.services.sequencing import by_osmosis_tail_dev_tasks + + dev_task = await self.get(dev_task_id) + if dev_task is None or dev_task.parent_task_id is None: + return + is_first = int(getattr(dev_task, "sequence", 0)) == 0 + cell_task = await self.get(UUID(str(dev_task.parent_task_id))) + if cell_task is None or cell_task.parent_task_id is None: + return + root = await self.get(UUID(str(cell_task.parent_task_id))) + if root is None: + return + groups: list = [] + for pred_root_id in list(root.dependency_ids): + for pred_ct in await self.get_subtasks(UUID(str(pred_root_id))): + groups.append(await self.get_subtasks(UUID(str(pred_ct.id)))) + for dep_id in by_osmosis_tail_dev_tasks(is_first, groups): + await self.add_dependency(dev_task_id, UUID(str(dep_id))) + async def set_sequence(self, task_id: UUID, sequence: int) -> None: """Set a task's sibling-ordering sequence (lower = first). diff --git a/tests/integration/test_task_service_transitions.py b/tests/integration/test_task_service_transitions.py index c426493a..27bd1e85 100644 --- a/tests/integration/test_task_service_transitions.py +++ b/tests/integration/test_task_service_transitions.py @@ -705,6 +705,7 @@ async def test_create_subtask_round_trips_collision_surfaces_and_deps( parent_task_id=parent.id, task_type=TaskType.CODE, nature=TaskNature.TECHNICAL, + estimated_complexity=Complexity.MEDIUM, sequence=3, dependency_ids=[dep_id], intends_to_touch=["roboco/services/foo.py"], @@ -747,6 +748,7 @@ async def test_wire_sibling_collision_dag_serializes_overlapping_dev_tasks( parent_task_id=parent.id, task_type=TaskType.CODE, nature=TaskNature.TECHNICAL, + estimated_complexity=Complexity.MEDIUM, sequence=seq, intends_to_touch=surface, ) @@ -774,6 +776,148 @@ async def test_wire_sibling_collision_dag_serializes_overlapping_dev_tasks( assert t2.id not in r3.dependency_ids +@pytest.mark.asyncio +async def test_wire_cell_task_wave_chain_chains_to_predecessor_cell_tasks( + task_setup: dict, db_session: AsyncSession +) -> None: + """Kind 2: a cell-task under root-subtask R1 depends on every cell-task + under R0, where R1 depends-on R0 (the kind-1 wave-chain edge on R1). + + A root-subtask may fan to several cell-tasks (different cells) — both of + R0's cell-tasks are wired onto R1's cell-task so its branch carries the + whole previous wave's merged cell work. Idempotent (add_dependency dedupes). + """ + svc = task_setup["svc"] + # Two root-subtasks; R1 depends-on R0 (the kind-1 edge lives on R1). + r0 = await svc.create( + _req( + task_setup, + title="r0", + team=Team.MAIN_PM, + task_type=TaskType.PLANNING, + nature=TaskNature.TECHNICAL, + ) + ) + r1 = await svc.create( + _req( + task_setup, + title="r1", + team=Team.MAIN_PM, + task_type=TaskType.PLANNING, + nature=TaskNature.TECHNICAL, + dependency_ids=[UUID(str(r0.id))], + ) + ) + await db_session.flush() + + async def _cell(parent_id: UUID, team: Team) -> Any: + return await svc.create_subtask( + TaskCreateRequest( + title=f"ct-{team.value}", + description="cell task description long enough", + acceptance_criteria=["ac"], + team=team, + created_by=task_setup["agent_id"], + project_id=task_setup["project_id"], + parent_task_id=parent_id, + task_type=TaskType.CODE, + nature=TaskNature.TECHNICAL, + estimated_complexity=Complexity.MEDIUM, + ) + ) + + # R0 fans to two cell-tasks (backend + frontend — the cross-cell fanout the + # CEO confirmed a root-subtask may carry). + ct0a = await _cell(UUID(str(r0.id)), Team.BACKEND) + ct0b = await _cell(UUID(str(r0.id)), Team.FRONTEND) + # R1's cell-task. + ct1 = await _cell(UUID(str(r1.id)), Team.BACKEND) + + await svc.wire_cell_task_wave_chain(ct1.id) + # Idempotent: a second run adds no duplicates. + await svc.wire_cell_task_wave_chain(ct1.id) + + r1ct = await svc.get(ct1.id) + assert r1ct is not None + # ct1 depends on BOTH of R0's cell-tasks (the whole previous wave's set). + assert UUID(str(ct0a.id)) in r1ct.dependency_ids + assert UUID(str(ct0b.id)) in r1ct.dependency_ids + # Idempotency: each predecessor appears once. + assert r1ct.dependency_ids.count(UUID(str(ct0a.id))) == 1 + + +@pytest.mark.asyncio +async def test_wire_by_osmosis_edge_first_dev_task_depends_on_prev_wave_tail( + task_setup: dict, db_session: AsyncSession +) -> None: + """Kind 4: the first dev task (sequence 0) of a cell-task under R1 depends + on the tail (highest-sequence) dev task of R0's cell-task. A non-first dev + task (sequence > 0) gets no by-osmosis edge (it inherits the tail via the + kind-3 collision DAG or shares the merged base).""" + svc = task_setup["svc"] + r0 = await svc.create( + _req( + task_setup, + title="r0", + team=Team.MAIN_PM, + task_type=TaskType.PLANNING, + nature=TaskNature.TECHNICAL, + ) + ) + r1 = await svc.create( + _req( + task_setup, + title="r1", + team=Team.MAIN_PM, + task_type=TaskType.PLANNING, + nature=TaskNature.TECHNICAL, + dependency_ids=[UUID(str(r0.id))], + ) + ) + await db_session.flush() + + async def _sub(parent_id: UUID, seq: int) -> Any: + t = await svc.create_subtask( + TaskCreateRequest( + title=f"t{seq}", + description=f"subtask {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, + estimated_complexity=Complexity.MEDIUM, + sequence=seq, + ) + ) + await svc.set_sequence(t.id, seq) + return t + + # R0's cell-task with two dev tasks; tail = sequence 1. + ct0 = await _sub(UUID(str(r0.id)), 0) + d0a = await _sub(UUID(str(ct0.id)), 0) + d0b = await _sub(UUID(str(ct0.id)), 1) # tail + # R1's cell-task with the first dev task (sequence 0) + a later one. + ct1 = await _sub(UUID(str(r1.id)), 0) + first = await _sub(UUID(str(ct1.id)), 0) + second = await _sub(UUID(str(ct1.id)), 1) + + await svc.wire_by_osmosis_edge(first.id) + await svc.wire_by_osmosis_edge(second.id) + + rf = await svc.get(first.id) + rs = await svc.get(second.id) + assert rf is not None and rs is not None + # The first dev task carries the by-osmosis edge to R0's cell-task's TAIL + # (d0b, sequence 1) — not the non-tail d0a. + assert UUID(str(d0b.id)) in rf.dependency_ids + assert UUID(str(d0a.id)) not in rf.dependency_ids + # The second dev task (sequence 1) gets NO by-osmosis edge. + assert UUID(str(d0b.id)) not in rs.dependency_ids + + @pytest.mark.asyncio async def test_fail_qa_work_session_fallback_excludes_qa_session( task_setup: dict, db_session: AsyncSession diff --git a/tests/unit/services/test_sequencing.py b/tests/unit/services/test_sequencing.py index 2da58c7d..09091fed 100644 --- a/tests/unit/services/test_sequencing.py +++ b/tests/unit/services/test_sequencing.py @@ -17,7 +17,12 @@ from roboco.foundation.policy.sequencing.models import ( DraftSurface, SequencingError, ) -from roboco.services.sequencing import SequencingService, dev_task_collision_edges +from roboco.services.sequencing import ( + SequencingService, + by_osmosis_tail_dev_tasks, + cell_task_wave_chain_depends_on, + dev_task_collision_edges, +) def _backend(_i: int) -> str: @@ -283,3 +288,80 @@ def test_dev_collision_returns_depends_on_first_pairs() -> None: [(dep, task)] = dev_task_collision_edges([first, second]) assert dep == first.id assert task == second.id + + +# --------------------------------------------------------------------------- +# cell_task_wave_chain_depends_on — the cell-task wave chain (edge kind 2). +# Pure glue: a new cell-task under root-subtask UT_n depends on every cell-task +# under every root-subtask UT_n itself depends on (the kind-1 wave-chain edges). +# --------------------------------------------------------------------------- + + +def test_wave_chain_collects_all_predecessor_cell_tasks() -> None: + # Two predecessor root-subtasks: one fans to two cell-tasks, the other to one. + ct_a1, ct_a2, ct_b1 = _Sib(uuid4()), _Sib(uuid4()), _Sib(uuid4()) + root_a, root_b = object(), object() + deps = cell_task_wave_chain_depends_on( + [root_a, root_b], {root_a: [ct_a1, ct_a2], root_b: [ct_b1]} + ) + assert set(deps) == {ct_a1.id, ct_a2.id, ct_b1.id} + + +def test_wave_chain_empty_when_no_predecessor_roots() -> None: + assert cell_task_wave_chain_depends_on([], {}) == [] + + +def test_wave_chain_skips_root_with_no_cell_tasks() -> None: + root = object() + assert cell_task_wave_chain_depends_on([root], {root: []}) == [] + # A predecessor root absent from the map contributes nothing (no KeyError). + assert cell_task_wave_chain_depends_on([object()], {}) == [] + + +def test_wave_chain_preserves_predecessor_order() -> None: + # Edges are appended in predecessor-root order then cell-task order — stable + # so add_dependency (which dedupes) sees a deterministic sequence. + ct_a, ct_b = _Sib(uuid4()), _Sib(uuid4()) + root_a, root_b = object(), object() + deps = cell_task_wave_chain_depends_on( + [root_a, root_b], {root_a: [ct_a], root_b: [ct_b]} + ) + assert deps == [ct_a.id, ct_b.id] + + +# --------------------------------------------------------------------------- +# by_osmosis_tail_dev_tasks — the by-osmosis edge (edge kind 4). +# Pure glue: the first dev task of a cell-task depends on each predecessor +# cell-task's tail (highest-sequence) dev task. Only sequence 0 carries it. +# --------------------------------------------------------------------------- + + +def test_by_osmosis_skips_non_first_dev_task() -> None: + tail = _Sib(uuid4(), sequence=2) + # is_first_dev_task=False -> no edges, regardless of predecessor groups. + assert by_osmosis_tail_dev_tasks(False, [[tail]]) == [] + + +def test_by_osmosis_picks_max_sequence_per_group() -> None: + t0 = _Sib(uuid4(), sequence=0) + t1 = _Sib(uuid4(), sequence=1) + t2 = _Sib(uuid4(), sequence=2) + assert by_osmosis_tail_dev_tasks(True, [[t0, t1, t2]]) == [t2.id] + + +def test_by_osmosis_one_tail_per_predecessor_group() -> None: + a_tail = _Sib(uuid4(), sequence=2) + b_tail = _Sib(uuid4(), sequence=4) + a_group = [_Sib(uuid4(), sequence=0), _Sib(uuid4(), sequence=1), a_tail] + b_group = [_Sib(uuid4(), sequence=3), b_tail] + assert by_osmosis_tail_dev_tasks(True, [a_group, b_group]) == [a_tail.id, b_tail.id] + + +def test_by_osmosis_skips_empty_predecessor_group() -> None: + # A predecessor cell-task with no dev tasks contributes no edge. + tail = _Sib(uuid4(), sequence=1) + assert by_osmosis_tail_dev_tasks(True, [[], [tail]]) == [tail.id] + + +def test_by_osmosis_no_edges_when_no_predecessor_groups() -> None: + assert by_osmosis_tail_dev_tasks(True, []) == []