mirror of
https://github.com/rennf93/roboco.git
synced 2026-08-03 07:23:24 +02:00
feat(tasks): sequence is the bar — strict sibling ordering at the claim chokepoint (#452)
* feat(tasks): enforce sibling sequence order at the claim chokepoint A task with a parent and effective sequence N (COALESCE(sequence, 0)) can no longer be claimed while any sibling with a strictly lower effective sequence is non-terminal — assignee-blind, independent of and stricter than dependency_ids, enforced in _validate_claim_preconditions so both claim paths (gateway verbs and the dispatcher's raw REST claim) cross it. Ties run parallel; cancelled siblings never block; sequence 0 and parentless tasks are unaffected. Live failure this guards: a PM delegated revision subtasks sequenced 0..3 with no dependency edges and seq 2 started alongside seq 0 — sequence was advisory-only. set_sequence's contract updated accordingly. New e2e smoke case drives the refusal and the post-completion claim through the real gateway. * chore(scripts): skip .uv-cache and .claude in the prose scanner Repo-local tool dirs (private uv cache, agent worktrees) carry vendored and generated markdown that tripped make reflow-check. * fix(tasks): wave-derived delegation sequences + claim-gate hardening Three fixes from the adversarial review of the sequence claim gate: Delegation no longer stamps a raw per-sibling ordinal (deterministic merge-order bookkeeping) as sequence — under the strict gate that serialized ALL delegated work, including fully independent cross-dev and cross-cell siblings. Sequences are now wave-derived post-wiring (stamp_wave_sequence: 1 + max same-parent dependency sequence, 0 when independent), so independent siblings tie and run parallel while colliding/ordered work ascends. The cross-cell UX wiring restamps instead of writing relative ux+1 values (a relative write could invert a collision-derived stamp), and the dispatch merge/lane barriers gain a created_at tiebreak for wave-tied siblings so shared-branch merge order stays deterministic. PM-authored sequences are never rewritten. The guard now also fires on reclaims from needs_revision (a lower- sequence sibling delegated after the first claim was invisible), and tasks.parent_task_id gains an index (migration 069) — the guard's sibling probe ran as a Seq Scan on the hottest verb. --------- Co-authored-by: Renn F <rennf93@users.noreply.github.com>
This commit is contained in:
@@ -0,0 +1,29 @@
|
||||
"""Add ix_tasks_parent_task_id — sibling scans on the claim hot path.
|
||||
|
||||
Postgres does not auto-index FK columns, so every sibling lookup
|
||||
(``get_subtasks``, the sequence claim guard's blocking-sibling probe, the
|
||||
dispatch merge/lane barriers) was a Seq Scan over tasks. The sequence guard
|
||||
runs on every PENDING/NEEDS_REVISION claim, so the scan sat on the hottest
|
||||
verb. Plain btree; additive; no data change.
|
||||
|
||||
Revision ID: 069_tasks_parent_task_id_idx
|
||||
Revises: 068_tasks_constraints_column
|
||||
Create Date: 2026-07-10
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from alembic import op
|
||||
|
||||
revision = "069_tasks_parent_task_id_idx"
|
||||
down_revision = "068_tasks_constraints_column"
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
op.create_index("ix_tasks_parent_task_id", "tasks", ["parent_task_id"])
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
op.drop_index("ix_tasks_parent_task_id", table_name="tasks")
|
||||
+4
-1
@@ -272,7 +272,10 @@ class TaskTable(Base):
|
||||
|
||||
# Relationships
|
||||
parent_task_id: Mapped[UUID | None] = mapped_column(
|
||||
UUID(as_uuid=True), ForeignKey("tasks.id", ondelete="SET NULL"), nullable=True
|
||||
UUID(as_uuid=True),
|
||||
ForeignKey("tasks.id", ondelete="SET NULL"),
|
||||
nullable=True,
|
||||
index=True,
|
||||
)
|
||||
dependency_ids: Mapped[list[PyUUID]] = mapped_column(
|
||||
ARRAY(UUID(as_uuid=True)), default=list
|
||||
|
||||
@@ -503,6 +503,25 @@ def _read_project_slug(task: dict[str, Any]) -> str | None:
|
||||
return str(inner) if inner else None
|
||||
|
||||
|
||||
def _created_before(sib_created: Any, task_created: Any) -> bool:
|
||||
"""True if ``sib_created`` (a datetime row value) precedes ``task_created``
|
||||
(an ISO string from the API task payload, or a datetime).
|
||||
|
||||
The equal-sequence tiebreak for the merge / lane dispatch barriers: wave-
|
||||
stamped independent siblings share a sequence, so creation order decides
|
||||
who merges first. Fail-open (False) on any missing/unparseable value —
|
||||
a tie that can't be ordered must not wedge dispatch.
|
||||
"""
|
||||
if sib_created is None or not task_created:
|
||||
return False
|
||||
try:
|
||||
if isinstance(task_created, str):
|
||||
task_created = datetime.fromisoformat(task_created)
|
||||
return bool(sib_created < task_created)
|
||||
except (TypeError, ValueError):
|
||||
return False
|
||||
|
||||
|
||||
def _is_coordination_task(task: dict[str, Any]) -> bool:
|
||||
"""True for a task that does no git of its own.
|
||||
|
||||
@@ -12548,7 +12567,9 @@ Never `commit`, never write code, never run `git`. PMs coordinate.
|
||||
a later sibling before an earlier one diverges the branch and wedges the
|
||||
loser. Hold a higher-sequence sibling's review/merge dispatch until the
|
||||
earlier ones land (or are cancelled). Loop-free: the task simply isn't
|
||||
dispatched this tick — no reject, no respawn churn.
|
||||
dispatched this tick — no reject, no respawn churn. Equal sequences
|
||||
(wave-stamped independent siblings — parallel to CLAIM and build) tie-
|
||||
break by ``created_at`` so the shared-branch merge stays serialized.
|
||||
|
||||
Only same-team siblings block (they target the same branch). Terminal
|
||||
siblings (completed/cancelled) never block, so a cancelled sibling can't
|
||||
@@ -12579,18 +12600,39 @@ Never `commit`, never write code, never run `git`. PMs coordinate.
|
||||
error=str(exc),
|
||||
)
|
||||
return False
|
||||
for sib in siblings:
|
||||
sib_seq = getattr(sib, "sequence", 0) or 0
|
||||
sib_team = getattr(sib, "team", None)
|
||||
sib_status = getattr(sib, "status", None)
|
||||
sib_team_val = getattr(sib_team, "value", sib_team)
|
||||
if (
|
||||
str(sib_team_val) == str(team)
|
||||
and sib_seq < seq
|
||||
and sib_status not in terminal
|
||||
):
|
||||
return True
|
||||
return False
|
||||
task_created = task.get("created_at")
|
||||
return any(
|
||||
self._is_earlier_live_team_sibling(
|
||||
sib,
|
||||
team=str(team),
|
||||
seq=seq,
|
||||
terminal=terminal,
|
||||
task_created=task_created,
|
||||
)
|
||||
for sib in siblings
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _is_earlier_live_team_sibling(
|
||||
sib: Any, *, team: str, seq: int, terminal: set[Any], task_created: Any
|
||||
) -> bool:
|
||||
"""True if ``sib`` is an earlier non-terminal same-team sibling.
|
||||
|
||||
Earlier = lower sequence, or an equal sequence created first (the
|
||||
wave-tie tiebreak that keeps the shared-branch merge serialized).
|
||||
"""
|
||||
sib_seq = getattr(sib, "sequence", 0) or 0
|
||||
sib_team = getattr(sib, "team", None)
|
||||
sib_team_val = getattr(sib_team, "value", sib_team)
|
||||
earlier = sib_seq < seq or (
|
||||
sib_seq == seq
|
||||
and _created_before(getattr(sib, "created_at", None), task_created)
|
||||
)
|
||||
return (
|
||||
str(sib_team_val) == team
|
||||
and earlier
|
||||
and getattr(sib, "status", None) not in terminal
|
||||
)
|
||||
|
||||
async def _blocked_by_earlier_lane_sibling(self, task: dict[str, Any]) -> bool:
|
||||
"""True if the SAME dev has an earlier non-terminal code sibling.
|
||||
@@ -12606,6 +12648,8 @@ Never `commit`, never write code, never run `git`. PMs coordinate.
|
||||
keyed on team): this is keyed on the assignee and only gates ``code``.
|
||||
Loop-free (skip this tick — no reject, no respawn churn) and best-effort
|
||||
(any lookup failure falls through to dispatch so the check never wedges).
|
||||
Equal sequences tie-break by ``created_at``, mirroring the merge
|
||||
barrier, so a dev's wave-tied queue keeps a deterministic order.
|
||||
"""
|
||||
if str(task.get("task_type") or "") != "code":
|
||||
return False
|
||||
@@ -12634,26 +12678,47 @@ Never `commit`, never write code, never run `git`. PMs coordinate.
|
||||
)
|
||||
return False
|
||||
task_id = str(task.get("id"))
|
||||
task_created = task.get("created_at")
|
||||
return any(
|
||||
self._is_earlier_live_lane_sibling(
|
||||
sib, task_id=task_id, owner=str(owner), seq=seq, terminal=terminal
|
||||
sib,
|
||||
task_id=task_id,
|
||||
owner=str(owner),
|
||||
seq=seq,
|
||||
terminal=terminal,
|
||||
task_created=task_created,
|
||||
)
|
||||
for sib in siblings
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _is_earlier_live_lane_sibling(
|
||||
sib: Any, *, task_id: str, owner: str, seq: int, terminal: set[Any]
|
||||
sib: Any,
|
||||
*,
|
||||
task_id: str,
|
||||
owner: str,
|
||||
seq: int,
|
||||
terminal: set[Any],
|
||||
task_created: Any = None,
|
||||
) -> bool:
|
||||
"""True if ``sib`` is a lower-sequence non-terminal code task for ``owner``."""
|
||||
"""True if ``sib`` is an earlier non-terminal code task for ``owner``.
|
||||
|
||||
Earlier = lower sequence, or an equal sequence created first (the
|
||||
wave-tie tiebreak mirroring the merge barrier).
|
||||
"""
|
||||
if str(sib.id) == task_id:
|
||||
return False
|
||||
sib_type = getattr(sib, "task_type", None)
|
||||
sib_type_val = getattr(sib_type, "value", sib_type)
|
||||
sib_seq = getattr(sib, "sequence", 0) or 0
|
||||
earlier = sib_seq < seq or (
|
||||
sib_seq == seq
|
||||
and _created_before(getattr(sib, "created_at", None), task_created)
|
||||
)
|
||||
return (
|
||||
str(getattr(sib, "assigned_to", None)) == owner
|
||||
and str(sib_type_val) == "code"
|
||||
and (getattr(sib, "sequence", 0) or 0) < seq
|
||||
and earlier
|
||||
and getattr(sib, "status", None) not in terminal
|
||||
)
|
||||
|
||||
|
||||
@@ -5627,7 +5627,13 @@ class Choreographer:
|
||||
)
|
||||
|
||||
async def _depend_frontend_on_ux(self, fe_task: Any, parent_id: Any) -> None:
|
||||
"""Make a new FRONTEND cell task wait on its non-terminal UX/UI sibling."""
|
||||
"""Make a new FRONTEND cell task wait on its non-terminal UX/UI sibling.
|
||||
|
||||
The wire is followed by a wave restamp (not a hand-rolled ``ux+1``
|
||||
write) so the sequence reflects EVERY same-parent dependency the task
|
||||
carries — a relative write could undercut a collision-derived stamp
|
||||
and invert the claim gate's order against a wired edge.
|
||||
"""
|
||||
from roboco.foundation.identity import Team
|
||||
from roboco.models.base import TaskStatus
|
||||
|
||||
@@ -5645,12 +5651,13 @@ class Choreographer:
|
||||
)
|
||||
if ux is not None:
|
||||
await self.task.add_dependency(fe_task.id, ux.id)
|
||||
await self.task.set_sequence(
|
||||
fe_task.id, (getattr(ux, "sequence", 0) or 0) + 1
|
||||
)
|
||||
await self.task.stamp_wave_sequence(fe_task.id)
|
||||
|
||||
async def _depend_backend_on_ux(self, be_task: Any, parent_id: Any) -> None:
|
||||
"""Make a new BACKEND cell task wait on its non-terminal UX/UI sibling."""
|
||||
"""Make a new BACKEND cell task wait on its non-terminal UX/UI sibling.
|
||||
|
||||
Wire + wave restamp, mirroring :meth:`_depend_frontend_on_ux`.
|
||||
"""
|
||||
from roboco.foundation.identity import Team
|
||||
from roboco.models.base import TaskStatus
|
||||
|
||||
@@ -5668,19 +5675,22 @@ class Choreographer:
|
||||
)
|
||||
if ux is not None:
|
||||
await self.task.add_dependency(be_task.id, ux.id)
|
||||
await self.task.set_sequence(
|
||||
be_task.id, (getattr(ux, "sequence", 0) or 0) + 1
|
||||
)
|
||||
await self.task.stamp_wave_sequence(be_task.id)
|
||||
|
||||
async def _depend_pending_frontends_on_ux(
|
||||
self, ux_task: Any, parent_id: Any
|
||||
) -> None:
|
||||
"""Retro-wire not-yet-started FRONTEND siblings onto a new UX/UI task."""
|
||||
"""Retro-wire not-yet-started FRONTEND siblings onto a new UX/UI task.
|
||||
|
||||
Each retro-wired sibling is wave-restamped so it lands above the UX
|
||||
task's OWN stamped wave (the new task is stamped before this runs) —
|
||||
a relative ``ux+1`` write here read the pre-stamp sequence and could
|
||||
invert the order.
|
||||
"""
|
||||
from roboco.foundation.identity import Team
|
||||
from roboco.models.base import TaskStatus
|
||||
|
||||
not_started = {TaskStatus.BACKLOG, TaskStatus.PENDING}
|
||||
ux_sequence = (getattr(ux_task, "sequence", 0) or 0) + 1
|
||||
siblings = await self.task.get_subtasks(parent_id)
|
||||
for fe in siblings:
|
||||
if (
|
||||
@@ -5689,17 +5699,20 @@ class Choreographer:
|
||||
and fe.status in not_started
|
||||
):
|
||||
await self.task.add_dependency(fe.id, ux_task.id)
|
||||
await self.task.set_sequence(fe.id, ux_sequence)
|
||||
await self.task.stamp_wave_sequence(fe.id)
|
||||
|
||||
async def _depend_pending_backends_on_ux(
|
||||
self, ux_task: Any, parent_id: Any
|
||||
) -> None:
|
||||
"""Retro-wire not-yet-started BACKEND siblings onto a new UX/UI task."""
|
||||
"""Retro-wire not-yet-started BACKEND siblings onto a new UX/UI task.
|
||||
|
||||
Wire + wave restamp per sibling, mirroring
|
||||
:meth:`_depend_pending_frontends_on_ux`.
|
||||
"""
|
||||
from roboco.foundation.identity import Team
|
||||
from roboco.models.base import TaskStatus
|
||||
|
||||
not_started = {TaskStatus.BACKLOG, TaskStatus.PENDING}
|
||||
ux_sequence = (getattr(ux_task, "sequence", 0) or 0) + 1
|
||||
siblings = await self.task.get_subtasks(parent_id)
|
||||
for be in siblings:
|
||||
if (
|
||||
@@ -5708,7 +5721,7 @@ class Choreographer:
|
||||
and be.status in not_started
|
||||
):
|
||||
await self.task.add_dependency(be.id, ux_task.id)
|
||||
await self.task.set_sequence(be.id, ux_sequence)
|
||||
await self.task.stamp_wave_sequence(be.id)
|
||||
|
||||
async def _resolve_subtask_project(
|
||||
self, parent: Any, inputs: DelegateInputs
|
||||
@@ -5847,14 +5860,6 @@ class Choreographer:
|
||||
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
|
||||
# order is deterministic. Within-cell siblings were all left at the
|
||||
# default sequence 0 — which is why two leaf PRs raced into the same
|
||||
# cell branch and the second wedged. Each new sibling takes the next
|
||||
# ordinal (the count of pre-existing siblings).
|
||||
siblings = await self.task.get_subtasks(parent_task_id)
|
||||
next_seq = len([s for s in siblings if s.id != new_task.id])
|
||||
await self.task.set_sequence(new_task.id, next_seq)
|
||||
# Wire the dev-task collision DAG (multi-level sequencing edge kind 3):
|
||||
# run the deterministic analyzer over the parent's surfaced siblings
|
||||
# and add_dependency each collision edge so a later dev task stays
|
||||
@@ -5884,6 +5889,16 @@ class Choreographer:
|
||||
Team.UX_UI.value,
|
||||
):
|
||||
await self.task.wire_by_osmosis_edge(new_task.id)
|
||||
# Post-wiring wave stamp: sequence = 1 + max(same-parent dependency
|
||||
# sequences), 0 when independent — so independent siblings tie (and
|
||||
# run in parallel under the sequence claim gate) while colliding /
|
||||
# ordered ones ascend. Replaces the raw per-sibling ordinal, which
|
||||
# gave independent siblings distinct sequences and serialized ALL
|
||||
# delegated work fleet-wide. The cross-cell UX wiring (in the caller)
|
||||
# restamps any task it re-wires, so every sequence write in the
|
||||
# delegate flow is this one wave computation; merge order for wave
|
||||
# ties falls back to created_at in the merge/lane dispatch barriers.
|
||||
await self.task.stamp_wave_sequence(new_task.id)
|
||||
return new_task
|
||||
|
||||
@staticmethod
|
||||
|
||||
@@ -298,11 +298,14 @@ def dev_task_collision_edges(siblings: list) -> list[tuple[object, object]]:
|
||||
sibling without a ``project_id`` cannot collide (collisions are scoped to
|
||||
a repo) and is skipped. Siblings are ordered by ``(priority, sequence)``
|
||||
before indexing so the analyzer's edge order is stable across re-runs
|
||||
(a newly-delegated sibling takes a fresh ``sequence``; existing siblings
|
||||
keep theirs), so re-running after each delegate only ADDS edges for the
|
||||
new sibling's collisions — never flips an existing pair's order into a
|
||||
reverse edge (which would cycle). ``add_dependency`` dedupes, so repeated
|
||||
wiring is a no-op on already-wired pairs.
|
||||
(a newly-delegated sibling arrives at the default sequence and is
|
||||
wave-stamped post-wiring; existing siblings keep theirs, and sequence
|
||||
ties break by the stable ``created_at`` list order), so re-running after
|
||||
each delegate only ADDS edges for the new sibling's collisions — never
|
||||
flips an existing pair's order into a reverse edge (which would cycle:
|
||||
waves are edge-consistent, so an edge-wired pair can never invert its
|
||||
sort). ``add_dependency`` dedupes, so repeated wiring is a no-op on
|
||||
already-wired pairs.
|
||||
"""
|
||||
# Collision edges from DECLARED surfaces. Fewer than two surfaced siblings
|
||||
# -> no collision path (edges stays empty); the undeclared-surface fallback
|
||||
@@ -384,9 +387,12 @@ def by_osmosis_tail_dev_tasks(
|
||||
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).
|
||||
highest-``sequence`` dev task under each predecessor cell-task; sequence
|
||||
ties (wave-stamped independent siblings) resolve to the LAST group member
|
||||
(groups arrive in ``created_at`` order — the merge barrier's tiebreak —
|
||||
so the tail is the last-merging 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 []
|
||||
@@ -394,6 +400,9 @@ def by_osmosis_tail_dev_tasks(
|
||||
for group in predecessor_dev_task_groups:
|
||||
if not group:
|
||||
continue
|
||||
tail = max(group, key=lambda t: int(getattr(t, "sequence", 0)))
|
||||
_, tail = max(
|
||||
enumerate(group),
|
||||
key=lambda pair: (int(getattr(pair[1], "sequence", 0)), pair[0]),
|
||||
)
|
||||
tails.append(getattr(tail, "id", tail))
|
||||
return tails
|
||||
|
||||
+105
-10
@@ -2633,6 +2633,65 @@ class TaskService(BaseService):
|
||||
)
|
||||
return True
|
||||
|
||||
async def _claim_blocked_by_sequence(self, task: TaskTable) -> str | None:
|
||||
"""Non-None (naming the blocking sibling) when a same-parent sibling
|
||||
with a strictly lower effective sequence is still non-terminal.
|
||||
|
||||
CEO directive: sequence is the bar, independent of ``dependency_ids``
|
||||
— a PM can delegate siblings with ``sequence`` 0..N and wire no
|
||||
dependency edges between them, and claim order must still hold (the
|
||||
4-revision-subtask live failure this guards). STRICTER than
|
||||
dependency edges where both exist: a wave-N sibling waits for EVERY
|
||||
wave-(N-1) sibling, not just its edge targets — no edges-exist
|
||||
exemption. Effective sequence is ``COALESCE(sequence, 0)`` on both
|
||||
sides; ties run in parallel. Effective sequence 0 or no parent is
|
||||
unaffected. Scoped to PENDING and NEEDS_REVISION claims — wider than
|
||||
``_claim_blocked_by_dependencies`` (PENDING only), deliberately: a
|
||||
lower-sequence sibling delegated AFTER the first claim must still
|
||||
hold a needs_revision reclaim; the dep guard's identical
|
||||
needs_revision gap is pre-existing and left as-is.
|
||||
"""
|
||||
if (
|
||||
task.status not in (TaskStatus.PENDING, TaskStatus.NEEDS_REVISION)
|
||||
or task.parent_task_id is None
|
||||
):
|
||||
return None
|
||||
seq = task.sequence or 0
|
||||
if seq == 0:
|
||||
return None
|
||||
terminal = (TaskStatus.COMPLETED, TaskStatus.CANCELLED)
|
||||
result = await self.session.execute(
|
||||
select(TaskTable.title, TaskTable.sequence)
|
||||
.where(
|
||||
TaskTable.parent_task_id == task.parent_task_id,
|
||||
TaskTable.id != task.id,
|
||||
func.coalesce(TaskTable.sequence, 0) < seq,
|
||||
TaskTable.status.notin_(terminal),
|
||||
)
|
||||
.limit(1)
|
||||
)
|
||||
row = result.first()
|
||||
if row is None:
|
||||
return None
|
||||
blocker = f"{row.title!r} (sequence {row.sequence or 0})"
|
||||
self.log.warning(
|
||||
"Cannot claim task - sequence_held",
|
||||
task_id=str(task.id),
|
||||
blocked_by=blocker,
|
||||
)
|
||||
return blocker
|
||||
|
||||
async def _claim_blocked_by_sequencing(self, task: TaskTable) -> bool:
|
||||
"""True when either sequencing guard blocks this PENDING claim.
|
||||
|
||||
One call site for both `_claim_blocked_by_dependencies` (edges) and
|
||||
`_claim_blocked_by_sequence` (sibling order) — keeps
|
||||
`_validate_claim_preconditions` under the return-statement budget.
|
||||
"""
|
||||
if await self._claim_blocked_by_dependencies(task):
|
||||
return True
|
||||
return await self._claim_blocked_by_sequence(task) is not None
|
||||
|
||||
async def _validate_claim_preconditions(
|
||||
self,
|
||||
task: TaskTable,
|
||||
@@ -2647,7 +2706,7 @@ class TaskService(BaseService):
|
||||
self.log.warning(f"Cannot claim task - {error}", task_id=str(task.id))
|
||||
return False
|
||||
|
||||
if await self._claim_blocked_by_dependencies(task):
|
||||
if await self._claim_blocked_by_sequencing(task):
|
||||
return False
|
||||
|
||||
if error := self._validate_claim_team(task, agent):
|
||||
@@ -7159,12 +7218,13 @@ class TaskService(BaseService):
|
||||
async def set_sequence(self, task_id: UUID, sequence: int) -> None:
|
||||
"""Set a task's sibling-ordering sequence (lower = first).
|
||||
|
||||
`sequence` is a display / dispatch-priority field only — it orders
|
||||
siblings in `list_pending`, `list_for_team`, and the panel and carries
|
||||
no claim-gating semantics (dependencies gate claims). Cross-cell
|
||||
fan-out uses it so an upstream design task sorts ahead of the
|
||||
implementation tasks that depend on it. No-op if the task is gone or
|
||||
already at `sequence`.
|
||||
`sequence` orders siblings in `list_pending`, `list_for_team`, and the
|
||||
panel, AND is claim-gated: `_claim_blocked_by_sequence` refuses to
|
||||
claim a PENDING task while a same-parent sibling with a strictly
|
||||
lower effective sequence is still non-terminal — independent of, and
|
||||
stricter than, `dependency_ids`. Cross-cell fan-out uses it so an
|
||||
upstream design task sorts ahead of the implementation tasks that
|
||||
depend on it. No-op if the task is gone or already at `sequence`.
|
||||
"""
|
||||
task = await self.get(task_id)
|
||||
if task is None:
|
||||
@@ -7173,6 +7233,36 @@ class TaskService(BaseService):
|
||||
task.sequence = sequence
|
||||
await self.session.flush()
|
||||
|
||||
async def stamp_wave_sequence(self, task_id: UUID) -> None:
|
||||
"""Stamp a freshly delegated subtask's sequence as its collision-DAG wave.
|
||||
|
||||
sequence = 1 + max(sequence of each SAME-PARENT dependency target), or
|
||||
0 when the task depends on no sibling — so independent siblings tie
|
||||
(parallel under `_claim_blocked_by_sequence`) and colliding / ordered
|
||||
ones ascend. Replaces the raw per-sibling ordinal, which serialized
|
||||
even fully independent siblings fleet-wide under the claim gate. Must
|
||||
run POST-WIRING (after the collision DAG / explicit depends_on /
|
||||
cross-cell edges land) so every edge source is reflected. Only the new
|
||||
task is stamped — existing siblings' sequences (prompter batch waves,
|
||||
API-authored values) are never rewritten here.
|
||||
"""
|
||||
task = await self.get(task_id)
|
||||
if task is None or task.parent_task_id is None:
|
||||
return
|
||||
wave = 0
|
||||
dep_ids = list(task.dependency_ids or [])
|
||||
if dep_ids:
|
||||
result = await self.session.execute(
|
||||
select(func.max(func.coalesce(TaskTable.sequence, 0))).where(
|
||||
TaskTable.id.in_(dep_ids),
|
||||
TaskTable.parent_task_id == task.parent_task_id,
|
||||
)
|
||||
)
|
||||
max_seq = result.scalar()
|
||||
if max_seq is not None:
|
||||
wave = int(max_seq) + 1
|
||||
await self.set_sequence(task_id, wave)
|
||||
|
||||
async def unmet_dependency_ids(self, dependency_ids: list[UUID]) -> list[UUID]:
|
||||
"""Return the subset of dependency IDs whose status is non-terminal.
|
||||
|
||||
@@ -7253,6 +7343,10 @@ class TaskService(BaseService):
|
||||
lane should NOT pin the dev — the orchestrator spawns it once the lane
|
||||
clears. Without this the dev can neither idle nor proceed without jumping
|
||||
its own queue order. Only ``code`` queues sequence this way.
|
||||
|
||||
Equal sequences (wave ties) tie-break by ``created_at`` so a dev's
|
||||
lane keeps a deterministic order now that independent siblings share
|
||||
a wave instead of taking distinct ordinals.
|
||||
"""
|
||||
if str(getattr(task, "task_type", "")) != TaskType.CODE.value:
|
||||
return False
|
||||
@@ -7262,7 +7356,7 @@ class TaskService(BaseService):
|
||||
if parent_id is None or owner is None or seq is None:
|
||||
return False
|
||||
result = await self.session.execute(
|
||||
select(TaskTable.status, TaskTable.sequence).where(
|
||||
select(TaskTable.status, TaskTable.sequence, TaskTable.created_at).where(
|
||||
TaskTable.parent_task_id == parent_id,
|
||||
TaskTable.assigned_to == owner,
|
||||
TaskTable.task_type == TaskType.CODE,
|
||||
@@ -7271,8 +7365,9 @@ class TaskService(BaseService):
|
||||
)
|
||||
terminal = {TaskStatus.COMPLETED, TaskStatus.CANCELLED}
|
||||
return any(
|
||||
(sib_seq or 0) < seq and status not in terminal
|
||||
for status, sib_seq in result.all()
|
||||
((sib_seq or 0), sib_created) < (seq, task.created_at)
|
||||
and status not in terminal
|
||||
for status, sib_seq, sib_created in result.all()
|
||||
)
|
||||
|
||||
async def get_all_descendants(self, task_id: UUID) -> list[TaskTable]:
|
||||
|
||||
@@ -38,6 +38,8 @@ SKIP_DIRS = {
|
||||
".ruff_cache",
|
||||
".pytest_cache",
|
||||
".superpowers",
|
||||
".uv-cache",
|
||||
".claude",
|
||||
}
|
||||
# Excluded subtrees / files: archive, generated (regenerated by `make lifecycle`
|
||||
# — never hand-edit), gitignored internal notes, untracked working specs.
|
||||
|
||||
@@ -0,0 +1,143 @@
|
||||
"""Scenario: the sibling-sequence claim gate holds independent of edges.
|
||||
|
||||
CEO directive: sequence is the bar, full stop. A same-parent sibling with a
|
||||
strictly lower sequence must hold the later sibling's claim even when NO
|
||||
dependency edge was ever wired between them — the live bug this guardrails:
|
||||
a PM delegated a batch of revision subtasks by sequence alone (0..3, no
|
||||
depends_on edges), and a later sequence claimed in parallel with an earlier,
|
||||
still-open one.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
from tests.e2e_smoke.arcs import (
|
||||
origin_branch,
|
||||
seed_company,
|
||||
seed_project,
|
||||
seed_task,
|
||||
set_branch_name,
|
||||
task_state,
|
||||
)
|
||||
from tests.e2e_smoke.harness import ScriptedAgent, expect_error
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from tests.e2e_smoke.harness import E2EStack
|
||||
|
||||
# Pydantic's IWillPlanRequest.approach enforces >= 150 chars, and the PM
|
||||
# sub_tasks gate requires a real (>= 60 char) description, so both blocked
|
||||
# and unblocked calls need compliant payloads — the sequence gate must be
|
||||
# the thing that fires, not an earlier content gate.
|
||||
_APPROACH = (
|
||||
"Plan revision 1 once revision 0 reaches a terminal state. This batch "
|
||||
"wires no dependency edges between its siblings, only sequence, so the "
|
||||
"claim gate alone must hold the delegation order end to end."
|
||||
)
|
||||
_SUB_TASKS = [
|
||||
{
|
||||
"title": "Land revision 1",
|
||||
"description": (
|
||||
"Apply the second sequenced revision once revision 0 is terminal "
|
||||
"and open its leaf PR against the cell branch."
|
||||
),
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
def _cancel(stack: E2EStack, task_id: Any) -> None:
|
||||
"""Direct terminal-status write — mirrors dispatcher_assign/set_branch_name's
|
||||
style (arcs.py): stands in for the real dev->QA->doc->PM chain reaching a
|
||||
terminal state, out of scope for this gate-only scenario."""
|
||||
from roboco.db.tables import TaskTable
|
||||
from roboco.models.base import TaskStatus
|
||||
from sqlalchemy import select
|
||||
|
||||
async def _run(session: AsyncSession) -> None:
|
||||
row = (
|
||||
await session.execute(select(TaskTable).where(TaskTable.id == task_id))
|
||||
).scalar_one()
|
||||
row.status = TaskStatus.CANCELLED
|
||||
|
||||
stack.run_db(_run)
|
||||
|
||||
|
||||
def test_sibling_sequence_blocks_claim_until_earlier_sibling_terminal(
|
||||
e2e_stack: E2EStack,
|
||||
) -> None:
|
||||
from roboco.models import Team
|
||||
from roboco.models.base import TaskType
|
||||
|
||||
stack = e2e_stack
|
||||
company = seed_company(stack)
|
||||
project_id, _project_slug = seed_project(stack, company)
|
||||
main_pm = ScriptedAgent(stack, company.main_pm_id, "main-pm", "main_pm")
|
||||
|
||||
parent_id = seed_task(
|
||||
stack,
|
||||
title="Revision batch",
|
||||
description="Coordinates a sequenced batch of revision subtasks.",
|
||||
acceptance_criteria=["every revision lands"],
|
||||
task_type=TaskType.PLANNING,
|
||||
team=Team.MAIN_PM,
|
||||
project_id=project_id,
|
||||
created_by=company.main_pm_id,
|
||||
assigned_to=company.main_pm_id,
|
||||
)
|
||||
seq0_id = seed_task(
|
||||
stack,
|
||||
title="Revision 0",
|
||||
description="First sequenced revision subtask.",
|
||||
acceptance_criteria=["revision 0 lands"],
|
||||
task_type=TaskType.PLANNING,
|
||||
team=Team.MAIN_PM,
|
||||
project_id=project_id,
|
||||
parent_task_id=parent_id,
|
||||
sequence=0,
|
||||
created_by=company.main_pm_id,
|
||||
assigned_to=company.main_pm_id,
|
||||
)
|
||||
seq1_id = seed_task(
|
||||
stack,
|
||||
title="Revision 1",
|
||||
description="Second sequenced revision subtask.",
|
||||
acceptance_criteria=["revision 1 lands"],
|
||||
task_type=TaskType.PLANNING,
|
||||
team=Team.MAIN_PM,
|
||||
project_id=project_id,
|
||||
parent_task_id=parent_id,
|
||||
sequence=1,
|
||||
created_by=company.main_pm_id,
|
||||
assigned_to=company.main_pm_id,
|
||||
)
|
||||
branch = f"feature/main_pm/{str(seq1_id)[:8]}"
|
||||
origin_branch(stack, branch, start="master")
|
||||
set_branch_name(stack, seq1_id, branch)
|
||||
# No wire_dependency() call anywhere in this test — sequence alone must
|
||||
# hold the order; seq0 stays PENDING (open, non-terminal).
|
||||
|
||||
expect_error(
|
||||
main_pm.flow(
|
||||
"i_will_plan",
|
||||
task_id=str(seq1_id),
|
||||
plan="Land revision 1 once revision 0 is terminal.",
|
||||
approach=_APPROACH,
|
||||
sub_tasks=_SUB_TASKS,
|
||||
),
|
||||
"invalid_state",
|
||||
"main_pm i_will_plan seq-1 while seq-0 open (no dependency edge)",
|
||||
)
|
||||
assert task_state(stack, seq1_id)["status"] == "pending"
|
||||
|
||||
_cancel(stack, seq0_id)
|
||||
|
||||
env = main_pm.flow(
|
||||
"i_will_plan",
|
||||
task_id=str(seq1_id),
|
||||
plan="Land revision 1 now that revision 0 is terminal.",
|
||||
approach=_APPROACH,
|
||||
sub_tasks=_SUB_TASKS,
|
||||
)
|
||||
assert env.get("error") != "invalid_state", env
|
||||
assert task_state(stack, seq1_id)["status"] == "in_progress", env
|
||||
@@ -25,7 +25,7 @@ from roboco.models.base import (
|
||||
from roboco.models.task import TaskCreateRequest
|
||||
from roboco.services.base import ConflictError
|
||||
from roboco.services.task import SoftBlockInfo, TaskService, get_task_service
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy import select, text
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import AsyncIterator
|
||||
@@ -1353,6 +1353,340 @@ async def test_claim_pending_with_unmet_dependency_returns_none(
|
||||
assert claimed.status == TaskStatus.CLAIMED
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Sequence claim guardrail (CEO directive: sequence is the bar, independent
|
||||
# of dependency_ids — see _claim_blocked_by_sequence).
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_claim_blocked_by_lower_sequence_sibling_no_dependency_edge(
|
||||
task_setup: dict, db_session: AsyncSession
|
||||
) -> None:
|
||||
"""A same-parent, lower-sequence, non-terminal sibling blocks a claim even
|
||||
with NO dependency edge wired — the live 4-revision-subtask bug: a PM
|
||||
delegated siblings with sequence 0..3 and no depends_on edges between
|
||||
them, so a later sequence claimed alongside an earlier one."""
|
||||
svc = task_setup["svc"]
|
||||
parent = await svc.create(_req(task_setup, title="parent"))
|
||||
seq0 = await svc.create(
|
||||
_req(task_setup, title="seq-0 sibling", parent_task_id=parent.id, sequence=0)
|
||||
)
|
||||
seq2 = await svc.create(
|
||||
_req(task_setup, title="seq-2 sibling", parent_task_id=parent.id, sequence=2)
|
||||
)
|
||||
seq0.status = TaskStatus.IN_PROGRESS
|
||||
await db_session.flush()
|
||||
|
||||
assert await svc.claim(seq2.id, task_setup["agent_id"]) is None
|
||||
|
||||
seq0.status = TaskStatus.COMPLETED
|
||||
await db_session.flush()
|
||||
seq2.branch_name = "feature/backend/abcd1234"
|
||||
await db_session.flush()
|
||||
claimed = await svc.claim(seq2.id, task_setup["agent_id"])
|
||||
assert claimed is not None
|
||||
assert claimed.status == TaskStatus.CLAIMED
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_claim_allows_equal_sequence_siblings_in_parallel(
|
||||
task_setup: dict, db_session: AsyncSession
|
||||
) -> None:
|
||||
"""Ties run in parallel: two siblings at the same sequence both claim."""
|
||||
svc = task_setup["svc"]
|
||||
parent = await svc.create(_req(task_setup, title="parent"))
|
||||
a = await svc.create(
|
||||
_req(task_setup, title="tie-a", parent_task_id=parent.id, sequence=1)
|
||||
)
|
||||
b = await svc.create(
|
||||
_req(task_setup, title="tie-b", parent_task_id=parent.id, sequence=1)
|
||||
)
|
||||
a.branch_name = "feature/backend/aaaa1111"
|
||||
b.branch_name = "feature/backend/bbbb2222"
|
||||
await db_session.flush()
|
||||
|
||||
claimed_a = await svc.claim(a.id, task_setup["agent_id"])
|
||||
claimed_b = await svc.claim(b.id, task_setup["agent_id"])
|
||||
assert claimed_a is not None
|
||||
assert claimed_b is not None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_claim_not_blocked_by_cancelled_lower_sequence_sibling(
|
||||
task_setup: dict, db_session: AsyncSession
|
||||
) -> None:
|
||||
"""A cancelled (terminal) lower-sequence sibling never holds the claim."""
|
||||
svc = task_setup["svc"]
|
||||
parent = await svc.create(_req(task_setup, title="parent"))
|
||||
seq0 = await svc.create(
|
||||
_req(task_setup, title="seq-0 cancelled", parent_task_id=parent.id, sequence=0)
|
||||
)
|
||||
seq1 = await svc.create(
|
||||
_req(task_setup, title="seq-1", parent_task_id=parent.id, sequence=1)
|
||||
)
|
||||
seq0.status = TaskStatus.CANCELLED
|
||||
seq1.branch_name = "feature/backend/cccc3333"
|
||||
await db_session.flush()
|
||||
|
||||
claimed = await svc.claim(seq1.id, task_setup["agent_id"])
|
||||
assert claimed is not None
|
||||
assert claimed.status == TaskStatus.CLAIMED
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_claim_blocked_by_null_sequence_sibling_coalesced_to_zero(
|
||||
task_setup: dict, db_session: AsyncSession
|
||||
) -> None:
|
||||
"""COALESCE(sequence, 0): a NULL-sequence sibling is treated as sequence 0
|
||||
and blocks a seq-1 claim, same as an explicit 0. `sequence` is NOT NULL in
|
||||
the live schema — this defends the query anyway, matching the `sib_seq or
|
||||
0` idiom `has_earlier_incomplete_code_sibling` already uses; the
|
||||
constraint is relaxed transiently (rolled back at teardown) to exercise
|
||||
it for real."""
|
||||
svc = task_setup["svc"]
|
||||
parent = await svc.create(_req(task_setup, title="parent"))
|
||||
null_seq = await svc.create(
|
||||
_req(task_setup, title="null-seq sibling", parent_task_id=parent.id, sequence=0)
|
||||
)
|
||||
seq1 = await svc.create(
|
||||
_req(task_setup, title="seq-1", parent_task_id=parent.id, sequence=1)
|
||||
)
|
||||
await db_session.execute(
|
||||
text("ALTER TABLE tasks ALTER COLUMN sequence DROP NOT NULL")
|
||||
)
|
||||
await db_session.execute(
|
||||
text("UPDATE tasks SET sequence = NULL WHERE id = :id"), {"id": null_seq.id}
|
||||
)
|
||||
await db_session.flush()
|
||||
|
||||
assert await svc.claim(seq1.id, task_setup["agent_id"]) is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_claim_sequence_zero_or_parentless_unaffected(
|
||||
task_setup: dict, db_session: AsyncSession
|
||||
) -> None:
|
||||
"""Effective sequence 0 and a parentless task are never sequence-gated,
|
||||
regardless of other non-terminal same-parent siblings."""
|
||||
svc = task_setup["svc"]
|
||||
parent = await svc.create(_req(task_setup, title="parent"))
|
||||
seq0 = await svc.create(
|
||||
_req(task_setup, title="seq-0", parent_task_id=parent.id, sequence=0)
|
||||
)
|
||||
await svc.create(
|
||||
_req(task_setup, title="seq-1 noise", parent_task_id=parent.id, sequence=1)
|
||||
)
|
||||
lone = await svc.create(_req(task_setup, title="parentless", sequence=5))
|
||||
seq0.branch_name = "feature/backend/dddd4444"
|
||||
lone.branch_name = "feature/backend/eeee5555"
|
||||
await db_session.flush()
|
||||
|
||||
assert (await svc.claim(seq0.id, task_setup["agent_id"])) is not None
|
||||
assert (await svc.claim(lone.id, task_setup["agent_id"])) is not None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_claim_blocked_by_sequence_names_distinct_reason(
|
||||
task_setup: dict, db_session: AsyncSession
|
||||
) -> None:
|
||||
"""`_claim_blocked_by_sequence` names the blocking sibling — a distinct
|
||||
reason from `_claim_blocked_by_dependencies` (audit/logs tell them apart
|
||||
since this task carries no dependency edge at all)."""
|
||||
svc = task_setup["svc"]
|
||||
parent = await svc.create(_req(task_setup, title="parent"))
|
||||
seq0 = await svc.create(
|
||||
_req(task_setup, title="seq-0 blocker", parent_task_id=parent.id, sequence=0)
|
||||
)
|
||||
seq1 = await svc.create(
|
||||
_req(task_setup, title="seq-1", parent_task_id=parent.id, sequence=1)
|
||||
)
|
||||
seq0.status = TaskStatus.IN_PROGRESS
|
||||
await db_session.flush()
|
||||
|
||||
assert await svc._claim_blocked_by_dependencies(seq1) is False
|
||||
reason = await svc._claim_blocked_by_sequence(seq1)
|
||||
assert reason is not None
|
||||
assert "seq-0 blocker" in reason
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_claim_batch_wave_blocked_by_all_wave0_siblings_no_edges(
|
||||
task_setup: dict, db_session: AsyncSession
|
||||
) -> None:
|
||||
"""STRICTER than dependency edges where both exist: a wave-1 root-subtask
|
||||
waits for EVERY wave-0 sibling, not just the one edge target the
|
||||
collision analyzer happened to wire — no edges-exist exemption."""
|
||||
svc = task_setup["svc"]
|
||||
umbrella = await svc.create(_req(task_setup, title="umbrella"))
|
||||
wave0_a = await svc.create(
|
||||
_req(task_setup, title="wave0-a", parent_task_id=umbrella.id, sequence=0)
|
||||
)
|
||||
wave0_b = await svc.create(
|
||||
_req(task_setup, title="wave0-b", parent_task_id=umbrella.id, sequence=0)
|
||||
)
|
||||
wave1 = await svc.create(
|
||||
_req(task_setup, title="wave1", parent_task_id=umbrella.id, sequence=1)
|
||||
)
|
||||
# Only wave0_a gets an edge (the file-overlap-conditioned edge the
|
||||
# analyzer wires) — wave0_b shares no surface with wave1, so production
|
||||
# never wires an edge to it.
|
||||
await svc.add_dependency(wave1.id, wave0_a.id)
|
||||
wave0_a.status = TaskStatus.COMPLETED
|
||||
await db_session.flush()
|
||||
|
||||
# The edge is satisfied, but wave0_b is still open with a lower sequence.
|
||||
assert await svc._claim_blocked_by_dependencies(wave1) is False
|
||||
assert await svc.claim(wave1.id, task_setup["agent_id"]) is None
|
||||
|
||||
wave0_b.status = TaskStatus.CANCELLED
|
||||
wave1.branch_name = "feature/backend/ffff6666"
|
||||
await db_session.flush()
|
||||
claimed = await svc.claim(wave1.id, task_setup["agent_id"])
|
||||
assert claimed is not None
|
||||
assert claimed.status == TaskStatus.CLAIMED
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_needs_revision_reclaim_blocked_by_lower_sequence_sibling(
|
||||
task_setup: dict, db_session: AsyncSession
|
||||
) -> None:
|
||||
"""A needs_revision reclaim is sequence-gated too: a lower-sequence
|
||||
sibling delegated AFTER the first claim must hold the reclaim (the gap a
|
||||
PENDING-only scope left open). The dep guard's identical needs_revision
|
||||
gap is pre-existing and deliberately untouched."""
|
||||
svc = task_setup["svc"]
|
||||
parent = await svc.create(_req(task_setup, title="parent"))
|
||||
low = await svc.create(
|
||||
_req(task_setup, title="late-delegated seq-0", parent_task_id=parent.id)
|
||||
)
|
||||
high = await svc.create(
|
||||
_req(
|
||||
task_setup, title="seq-1 in revision", parent_task_id=parent.id, sequence=1
|
||||
)
|
||||
)
|
||||
high.status = TaskStatus.NEEDS_REVISION
|
||||
high.branch_name = "feature/backend/9999aaaa"
|
||||
low.status = TaskStatus.IN_PROGRESS
|
||||
await db_session.flush()
|
||||
|
||||
assert await svc.claim(high.id, task_setup["agent_id"]) is None
|
||||
|
||||
low.status = TaskStatus.COMPLETED
|
||||
await db_session.flush()
|
||||
assert await svc.claim(high.id, task_setup["agent_id"]) is not None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Wave stamping at delegation (stamp_wave_sequence) — sequences derive from
|
||||
# the wired collision DAG, not a per-sibling ordinal.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_stamp_wave_sequence_independent_siblings_tie_and_claim_parallel(
|
||||
task_setup: dict, db_session: AsyncSession
|
||||
) -> None:
|
||||
"""Independent siblings (no edges) share wave 0 — both claimable in
|
||||
parallel. The raw ordinal gave them 0/1 and the claim gate then
|
||||
serialized ALL delegated work fleet-wide."""
|
||||
svc = task_setup["svc"]
|
||||
parent = await svc.create(_req(task_setup, title="parent"))
|
||||
a = await svc.create(_req(task_setup, title="indep-a", parent_task_id=parent.id))
|
||||
b = await svc.create(_req(task_setup, title="indep-b", parent_task_id=parent.id))
|
||||
await svc.stamp_wave_sequence(a.id)
|
||||
await svc.stamp_wave_sequence(b.id)
|
||||
assert (a.sequence, b.sequence) == (0, 0)
|
||||
|
||||
a.branch_name = "feature/backend/aaaa0001"
|
||||
b.branch_name = "feature/backend/bbbb0002"
|
||||
await db_session.flush()
|
||||
assert await svc.claim(a.id, task_setup["agent_id"]) is not None
|
||||
assert await svc.claim(b.id, task_setup["agent_id"]) is not None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_stamp_wave_sequence_ascends_for_dependent_siblings(
|
||||
task_setup: dict,
|
||||
) -> None:
|
||||
"""Colliding / ordered siblings ascend: each dependent stamps one wave
|
||||
above the max of its same-parent dependency targets, and the claim gate
|
||||
blocks the later wave while the earlier is open."""
|
||||
svc = task_setup["svc"]
|
||||
parent = await svc.create(_req(task_setup, title="parent"))
|
||||
t0 = await svc.create(_req(task_setup, title="wave-0", parent_task_id=parent.id))
|
||||
t1 = await svc.create(_req(task_setup, title="wave-1", parent_task_id=parent.id))
|
||||
t2 = await svc.create(_req(task_setup, title="wave-2", parent_task_id=parent.id))
|
||||
await svc.stamp_wave_sequence(t0.id)
|
||||
await svc.add_dependency(t1.id, t0.id)
|
||||
await svc.stamp_wave_sequence(t1.id)
|
||||
await svc.add_dependency(t2.id, t1.id)
|
||||
await svc.stamp_wave_sequence(t2.id)
|
||||
assert (t0.sequence, t1.sequence, t2.sequence) == (0, 1, 2)
|
||||
|
||||
assert await svc._claim_blocked_by_sequence(t1) is not None
|
||||
assert await svc.claim(t1.id, task_setup["agent_id"]) is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_stamp_wave_sequence_preserves_stamps_and_ignores_nonsibling_deps(
|
||||
task_setup: dict,
|
||||
) -> None:
|
||||
"""Only the NEW task is stamped — an explicitly authored sibling sequence
|
||||
(API create / prompter batch wave) is never rewritten — and a dependency
|
||||
on a task under a DIFFERENT parent contributes no wave."""
|
||||
svc = task_setup["svc"]
|
||||
parent = await svc.create(_req(task_setup, title="parent"))
|
||||
authored = await svc.create(
|
||||
_req(task_setup, title="authored seq-3", parent_task_id=parent.id, sequence=3)
|
||||
)
|
||||
outside = await svc.create(_req(task_setup, title="other-parent task"))
|
||||
new = await svc.create(
|
||||
_req(
|
||||
task_setup,
|
||||
title="new sibling",
|
||||
parent_task_id=parent.id,
|
||||
dependency_ids=[outside.id],
|
||||
)
|
||||
)
|
||||
await svc.stamp_wave_sequence(new.id)
|
||||
authored_stamp = 3
|
||||
assert new.sequence == 0 # non-sibling dep excluded from the wave
|
||||
assert authored.sequence == authored_stamp # PM-authored stamp untouched
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_stamp_wave_sequence_after_collision_wiring_matches_delegate_flow(
|
||||
task_setup: dict,
|
||||
) -> None:
|
||||
"""The delegate-flow composition end to end: create → wire the collision
|
||||
DAG → stamp. Disjoint-surface siblings tie at wave 0; a file-overlap
|
||||
sibling lands one wave above the sibling it collides with."""
|
||||
svc = task_setup["svc"]
|
||||
parent = await svc.create(_req(task_setup, title="parent"))
|
||||
|
||||
async def _delegate(title: str, surface: list[str]) -> Any:
|
||||
t = await svc.create(
|
||||
_req(
|
||||
task_setup,
|
||||
title=title,
|
||||
parent_task_id=parent.id,
|
||||
intends_to_touch=surface,
|
||||
)
|
||||
)
|
||||
await svc.wire_sibling_collision_dag(parent.id)
|
||||
await svc.stamp_wave_sequence(t.id)
|
||||
return t
|
||||
|
||||
a = await _delegate("touches a.py", ["roboco/api/a.py"])
|
||||
b = await _delegate("touches b.py", ["roboco/api/b.py"])
|
||||
c = await _delegate("touches a.py too", ["roboco/api/a.py"])
|
||||
|
||||
assert (a.sequence, b.sequence) == (0, 0) # disjoint → parallel
|
||||
assert c.sequence == 1 # collides with a → next wave
|
||||
assert await svc._claim_blocked_by_sequence(c) is not None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_claim_already_claimed_by_other_returns_none(
|
||||
task_setup: dict, db_session: AsyncSession
|
||||
@@ -1890,7 +2224,6 @@ async def _seed_minimal_task(db_session: AsyncSession, tid: UUID) -> UUID:
|
||||
async def test_add_dependency_rejects_self_reference(
|
||||
db_session: AsyncSession,
|
||||
) -> None:
|
||||
|
||||
tid = uuid4()
|
||||
await _seed_minimal_task(db_session, tid)
|
||||
svc = get_task_service(db_session)
|
||||
@@ -1900,7 +2233,6 @@ async def test_add_dependency_rejects_self_reference(
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_add_dependency_rejects_cycle(db_session: AsyncSession) -> None:
|
||||
|
||||
a, b, c = uuid4(), uuid4(), uuid4()
|
||||
await _seed_minimal_task(db_session, a)
|
||||
await _seed_minimal_task(db_session, b)
|
||||
@@ -2050,7 +2382,6 @@ async def test_pass_qa_refuses_in_progress(db_session: AsyncSession) -> None:
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_pass_qa_accepts_awaiting_qa(db_session: AsyncSession) -> None:
|
||||
|
||||
tid = uuid4()
|
||||
await _seed_minimal_task(db_session, tid)
|
||||
|
||||
|
||||
@@ -10,6 +10,7 @@ Keyed on the assignee (not the team like the merge barrier) and gates only
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import UTC, datetime
|
||||
from typing import Any, cast
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
from uuid import uuid4
|
||||
@@ -127,6 +128,27 @@ async def test_higher_sequence_same_dev_sibling_does_not_block() -> None:
|
||||
assert await orch._blocked_by_earlier_lane_sibling(task) is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_equal_sequence_same_dev_tiebreaks_by_created_at() -> None:
|
||||
"""Wave ties in a dev's own lane order by created_at (mirroring the merge
|
||||
barrier): the earlier-created tied sibling holds the later one; the
|
||||
later-created one does not hold the earlier."""
|
||||
orch = _new_orchestrator()
|
||||
task = _task(0, "be-dev-1")
|
||||
task["created_at"] = "2026-07-10T12:00:00+00:00"
|
||||
earlier = _sibling(0, "be-dev-1", TaskStatus.IN_PROGRESS)
|
||||
earlier.created_at = datetime(2026, 7, 10, 11, 0, tzinfo=UTC)
|
||||
p1, p2 = _patch_siblings([earlier])
|
||||
with p1, p2:
|
||||
assert await orch._blocked_by_earlier_lane_sibling(task) is True
|
||||
|
||||
later = _sibling(0, "be-dev-1", TaskStatus.IN_PROGRESS)
|
||||
later.created_at = datetime(2026, 7, 10, 13, 0, tzinfo=UTC)
|
||||
p1, p2 = _patch_siblings([later])
|
||||
with p1, p2:
|
||||
assert await orch._blocked_by_earlier_lane_sibling(task) is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_non_code_task_is_never_gated_without_db() -> None:
|
||||
orch = _new_orchestrator()
|
||||
|
||||
@@ -9,6 +9,7 @@ cancelled sibling can't deadlock the rest.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import UTC, datetime
|
||||
from typing import Any, cast
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
from uuid import uuid4
|
||||
@@ -111,6 +112,61 @@ async def test_higher_sequence_sibling_does_not_block() -> None:
|
||||
assert await orch._blocked_by_earlier_sibling(task) is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_equal_sequence_created_earlier_sibling_blocks() -> None:
|
||||
"""Wave ties (independent siblings share a sequence) tie-break by
|
||||
created_at: the earlier-created same-team sibling merges first, so the
|
||||
later one's review dispatch is held — the shared-cell-branch merge race
|
||||
the ordinal used to prevent."""
|
||||
orch = _new_orchestrator()
|
||||
task = {
|
||||
"id": str(uuid4()),
|
||||
"parent_task_id": str(uuid4()),
|
||||
"sequence": 0,
|
||||
"team": "frontend",
|
||||
"created_at": "2026-07-10T12:00:00+00:00",
|
||||
}
|
||||
earlier = _sibling(0, "frontend", TaskStatus.IN_PROGRESS)
|
||||
earlier.created_at = datetime(2026, 7, 10, 11, 0, tzinfo=UTC)
|
||||
p1, p2 = _patch_siblings([earlier])
|
||||
with p1, p2:
|
||||
assert await orch._blocked_by_earlier_sibling(task) is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_equal_sequence_created_later_sibling_does_not_block() -> None:
|
||||
orch = _new_orchestrator()
|
||||
task = {
|
||||
"id": str(uuid4()),
|
||||
"parent_task_id": str(uuid4()),
|
||||
"sequence": 0,
|
||||
"team": "frontend",
|
||||
"created_at": "2026-07-10T12:00:00+00:00",
|
||||
}
|
||||
later = _sibling(0, "frontend", TaskStatus.IN_PROGRESS)
|
||||
later.created_at = datetime(2026, 7, 10, 13, 0, tzinfo=UTC)
|
||||
p1, p2 = _patch_siblings([later])
|
||||
with p1, p2:
|
||||
assert await orch._blocked_by_earlier_sibling(task) is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_equal_sequence_unparseable_created_at_fails_open() -> None:
|
||||
"""A tie that can't be ordered (missing/mock created_at) must not wedge
|
||||
dispatch — the tiebreak degrades to not-blocked."""
|
||||
orch = _new_orchestrator()
|
||||
task = {
|
||||
"id": str(uuid4()),
|
||||
"parent_task_id": str(uuid4()),
|
||||
"sequence": 0,
|
||||
"team": "frontend",
|
||||
}
|
||||
tie = _sibling(0, "frontend", TaskStatus.IN_PROGRESS)
|
||||
p1, p2 = _patch_siblings([tie])
|
||||
with p1, p2:
|
||||
assert await orch._blocked_by_earlier_sibling(task) is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_no_parent_returns_false_without_db() -> None:
|
||||
orch = _new_orchestrator()
|
||||
|
||||
@@ -1352,9 +1352,12 @@ async def test_unclaimed_parent_acs_counts_live_children_not_just_completed() ->
|
||||
|
||||
|
||||
def _svc_with_sibling_status_seq(rows: list[tuple]) -> TaskService:
|
||||
"""TaskService whose execute() yields (status, sequence) sibling rows."""
|
||||
"""TaskService whose execute() yields (status, sequence[, created_at])
|
||||
sibling rows; 2-tuples are padded with created_at=None (only compared on
|
||||
a sequence tie)."""
|
||||
row_width = 3 # (status, sequence, created_at)
|
||||
res = MagicMock()
|
||||
res.all.return_value = rows
|
||||
res.all.return_value = [r if len(r) == row_width else (*r, None) for r in rows]
|
||||
return TaskService(MagicMock(execute=AsyncMock(return_value=res)))
|
||||
|
||||
|
||||
@@ -1386,6 +1389,25 @@ async def test_earlier_incomplete_code_sibling_false_when_earlier_terminal() ->
|
||||
assert await svc.has_earlier_incomplete_code_sibling(task) is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_earlier_incomplete_code_sibling_tie_breaks_by_created_at() -> None:
|
||||
# Wave ties (equal sequence) order by created_at: the earlier-created
|
||||
# tied sibling holds the later one; a later-created one does not.
|
||||
task = _build_task(
|
||||
task_type=TaskType.CODE.value,
|
||||
parent_task_id=uuid4(),
|
||||
assigned_to=uuid4(),
|
||||
sequence=1,
|
||||
created_at=datetime(2026, 7, 10, 12, 0, tzinfo=UTC),
|
||||
)
|
||||
earlier = datetime(2026, 7, 10, 11, 0, tzinfo=UTC)
|
||||
later = datetime(2026, 7, 10, 13, 0, tzinfo=UTC)
|
||||
svc = _svc_with_sibling_status_seq([(TaskStatus.IN_PROGRESS, 1, earlier)])
|
||||
assert await svc.has_earlier_incomplete_code_sibling(task) is True
|
||||
svc = _svc_with_sibling_status_seq([(TaskStatus.IN_PROGRESS, 1, later)])
|
||||
assert await svc.has_earlier_incomplete_code_sibling(task) is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_earlier_incomplete_code_sibling_false_for_higher_seq_only() -> None:
|
||||
# A LATER sibling (seq 3) does not hold an earlier leaf (seq 2).
|
||||
|
||||
Reference in New Issue
Block a user