fix(deps): gate the pre-assigned dev path on unmet dependencies

A dev subtask is always pre-assigned (assigned_to=<dev>), so it never
flows through the unassigned claim pool's dependency filter
(list_pending(filter_by_dependencies=True)). Every path that acts on a
pre-assigned pending dev subtask previously ignored dependency_ids: the
orchestrator spawned the dev container, give_me_work offered the task,
and the claim verb accepted it — letting a frontend dev code ahead of an
unfinished UX/UI design.

Hold the pre-assigned dev at each path it actually arrives by, until
every dependency reaches a terminal state:

- orchestrator _validate_task_for_spawn now consults dependency_ids via
  _check_dependencies_terminal and skips the spawn while any dependency
  is non-terminal (fail-closed on an unreadable dependency);
- TaskService.list_pending_for_agent excludes a pre-assigned task with
  unmet dependencies so give_me_work does not offer it;
- the Choreographer claim guard set rejects the claim with a clear
  remediate via a new unmet_dependency_guard.

Add TaskService.unmet_dependency_ids as the single source of truth for
"which dependency IDs are not yet terminal" and route the existing
inherit_unmet_dependencies through it.
This commit is contained in:
Renn F
2026-06-03 20:36:35 +02:00
parent 5462fe3ae6
commit b0a596ecac
5 changed files with 436 additions and 27 deletions
+54 -17
View File
@@ -3314,25 +3314,14 @@ Start by:
"""
from roboco.agents_config import get_agent_role
task_id = task.get("id")
if not task_id:
return "Task missing ID"
min_description_len = 10
if shape_err := await self._check_spawn_task_shape(client, task):
return shape_err
description = (task.get("description") or "").strip()
if len(description) < min_description_len:
return (
f"Task {task_id} has inadequate description ({len(description)} chars)"
)
# A coordination task carries a product instead of a repo; only a task
# with neither is genuinely unroutable.
if not task.get("project_id") and not _is_coordination_task(task):
await self._auto_block_task(
client, task_id, "Task needs a project_id or product_id"
)
return f"Task {task_id} needs a project or product"
if dep_err := await self._check_dependencies_terminal(client, task):
return dep_err
# _check_spawn_task_shape guarantees a non-empty id past this point.
task_id = str(task.get("id"))
parent_id = task.get("parent_task_id")
if parent_id:
err = await self._check_parent_branch_ready(client, task_id, parent_id)
@@ -3348,6 +3337,54 @@ Start by:
return None # All validations passed
async def _check_spawn_task_shape(
self, client: httpx.AsyncClient, task: dict[str, Any]
) -> str | None:
"""Reject a task that is structurally unroutable (id/description/repo)."""
task_id = task.get("id")
if not task_id:
return "Task missing ID"
min_description_len = 10
description = (task.get("description") or "").strip()
if len(description) < min_description_len:
return (
f"Task {task_id} has inadequate description ({len(description)} chars)"
)
# A coordination task carries a product instead of a repo; only a task
# with neither is genuinely unroutable.
if not task.get("project_id") and not _is_coordination_task(task):
await self._auto_block_task(
client, task_id, "Task needs a project_id or product_id"
)
return f"Task {task_id} needs a project or product"
return None
async def _check_dependencies_terminal(
self, client: httpx.AsyncClient, task: dict[str, Any]
) -> str | None:
"""Hold a pre-assigned task whose dependencies are not yet terminal.
A dev subtask is always pre-assigned, so it never passes through the
unassigned claim pool's dependency filter. Without this gate the
dispatcher would spawn the dev container while a cross-cell dependency
(e.g. the UX/UI design the frontend dev waits on) is still open. Return
a skip reason while ANY dependency is non-terminal; allow the spawn
once every dependency reaches completed/cancelled.
"""
dependency_ids = task.get("dependency_ids") or []
if not dependency_ids:
return None
terminal = ("completed", "cancelled")
for dep_id in dependency_ids:
dep_resp = await client.get(f"{self._api_url}/tasks/{dep_id}")
# A dependency we cannot read is treated as unmet — fail closed
# rather than spawn ahead of work whose state is unknown.
if not dep_resp.is_success or dep_resp.json().get("status") not in terminal:
return (
f"Task {task.get('id')} waiting on non-terminal dependency {dep_id}"
)
return None
async def _auto_block_task(
self, client: httpx.AsyncClient, task_id: str, reason: str
) -> None:
@@ -25,6 +25,7 @@ from roboco.services.gateway.claim_guards import (
already_active_guard,
paused_tasks_guard,
sibling_sequence_guard,
unmet_dependency_guard,
)
from roboco.services.gateway.envelope import Envelope
from roboco.services.gateway.evidence_builder import (
@@ -700,6 +701,11 @@ class Choreographer:
paused = await self.task.list_paused_for_agent(agent_id)
if guard := paused_tasks_guard(paused):
return guard
dep_ids = list(task.dependency_ids)
if dep_ids:
unmet = await self.task.unmet_dependency_ids(dep_ids)
if guard := unmet_dependency_guard(task, unmet):
return guard
if not skip_sequence:
siblings = await self._fetch_siblings(task)
if guard := sibling_sequence_guard(task, siblings):
+28
View File
@@ -81,6 +81,34 @@ def paused_tasks_guard(paused_tasks: list[Any]) -> Envelope | None:
)
def unmet_dependency_guard(
target_task: Any, unmet_dependency_ids: list[UUID]
) -> Envelope | None:
"""Refuse claim while the task has non-terminal dependencies.
A task may not be claimed until every task it ``depends_on`` reaches a
terminal state (completed/cancelled). This holds the pre-assigned dev
that arrives via the claim verb directly the dependency filter on the
unassigned claim pool (``list_pending(filter_by_dependencies=True)``)
never sees a pre-assigned task. ``unmet_dependency_ids`` is resolved by
the caller (it requires a DB read) so this predicate stays pure.
"""
if not unmet_dependency_ids:
return None
blockers = ", ".join(str(dep_id) for dep_id in unmet_dependency_ids)
return Envelope.invalid_state(
message=(
f"task {target_task.id} depends on unfinished work; "
f"{len(unmet_dependency_ids)} dependency(ies) not yet "
"completed/cancelled."
),
remediate=(
f"wait for dependency task(s) {blockers} to reach "
"completed/cancelled before claiming this task"
),
)
def _earlier_blocking_sibling(
target_task: Any, siblings: list[Any], my_sequence: int
) -> Any | None:
+40 -10
View File
@@ -4192,6 +4192,30 @@ class TaskService(BaseService):
task.dependency_ids = [*task.dependency_ids, depends_on_id]
await self.session.flush()
async def unmet_dependency_ids(self, dependency_ids: list[UUID]) -> list[UUID]:
"""Return the subset of dependency IDs whose status is non-terminal.
A dependency is "met" only once it reaches a terminal state
(completed/cancelled). This is the single source of truth for the
"can a task that depends on these proceed?" question reused by
`list_pending`, `list_pending_for_agent`, `inherit_unmet_dependencies`,
and the claim-time dependency guard. An empty input returns an empty
list (no dependencies = nothing unmet).
"""
if not dependency_ids:
return []
terminal = {TaskStatus.COMPLETED, TaskStatus.CANCELLED}
dep_result = await self.session.execute(
select(TaskTable.id, TaskTable.status).where(
TaskTable.id.in_(dependency_ids)
)
)
return [
dep_id
for dep_id, dep_status in dep_result.all()
if dep_status not in terminal
]
async def inherit_unmet_dependencies(
self, subtask_id: UUID, parent_id: UUID
) -> None:
@@ -4208,15 +4232,8 @@ class TaskService(BaseService):
parent = await self.get(parent_id)
if parent is None or not parent.dependency_ids:
return
dep_result = await self.session.execute(
select(TaskTable.id, TaskTable.status).where(
TaskTable.id.in_(parent.dependency_ids)
)
)
terminal = {TaskStatus.COMPLETED, TaskStatus.CANCELLED}
for dep_id, dep_status in dep_result.all():
if dep_status not in terminal:
await self.add_dependency(subtask_id, dep_id)
for dep_id in await self.unmet_dependency_ids(list(parent.dependency_ids)):
await self.add_dependency(subtask_id, dep_id)
async def get_subtasks(self, parent_task_id: UUID) -> list[TaskTable]:
"""Get all subtasks of a parent task."""
@@ -4990,6 +5007,13 @@ class TaskService(BaseService):
assigned_to=<them> + status=pending got 'no work' until they
triage()'d explicitly.
A pre-assigned task with unmet (non-terminal) dependencies is held
back: offering it would let the agent claim and work ahead of a
dependency that has not resolved (e.g. a frontend dev coding before
the UX/UI design lands). The pre-assigned path bypasses
`list_pending(filter_by_dependencies=True)`, so the dependency gate
must be applied here too.
Ordered by sequence asc, then priority asc, then created_at asc so
earlier-sequence tasks win.
"""
@@ -5006,7 +5030,13 @@ class TaskService(BaseService):
)
)
result = await self.session.execute(query)
return list(result.scalars().all())
tasks = list(result.scalars().all())
available: list[TaskTable] = []
for task in tasks:
if await self.unmet_dependency_ids(list(task.dependency_ids)):
continue
available.append(task)
return available
async def list_paused_for_agent(self, agent_id: UUID) -> list[TaskTable]:
"""Paused tasks assigned to the agent."""
@@ -0,0 +1,308 @@
"""A pre-assigned dev subtask with an unmet (non-terminal) dependency must be
held at EVERY path the developer actually arrives by not only by the
unassigned claim pool's dependency filter.
A dev subtask is always pre-assigned (``assigned_to=<dev>``), so it never flows
through ``list_pending(filter_by_dependencies=True)``. The three real arrival
paths are exercised here against a real database:
(a) orchestrator spawn dispatch ``_validate_task_for_spawn`` (the HTTP path
the dev container is spawned by) must return a skip reason and not spawn;
(b) ``give_me_work`` ``TaskService.list_pending_for_agent`` must exclude it;
(c) claim the Choreographer's ``_run_claim_guards`` (invoked by the
``i_will_work_on`` verb) must reject it.
Once the UX/UI dependency reaches a terminal state, all three allow the dev to
proceed. This intentionally does NOT assert via
``list_pending(filter_by_dependencies=True)`` that gate serves the
unassigned claim pool, not the pre-assigned dev.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
from unittest.mock import PropertyMock, patch
from uuid import uuid4
import pytest
import pytest_asyncio
from fastapi import FastAPI
from httpx import ASGITransport, AsyncClient
from roboco.api.deps import get_db
from roboco.api.routes.tasks import router as tasks_router
from roboco.db.tables import AgentTable, ProductTable, ProjectTable, TaskTable
from roboco.models import AgentRole, AgentStatus, Team
from roboco.models.base import Complexity, TaskNature, TaskStatus, TaskType
from roboco.models.task import TaskCreateRequest
from roboco.runtime.orchestrator import AgentOrchestrator
from roboco.services.gateway.choreographer._impl import (
Choreographer,
ChoreographerDeps,
)
from roboco.services.task import TaskService
if TYPE_CHECKING:
from collections.abc import AsyncIterator
from sqlalchemy.ext.asyncio import AsyncSession
# A canonical developer slug so the orchestrator's get_agent_role() classifies
# the assignee as "developer" and the real dev-spawn validation runs.
_DEV_SLUG = "fe-dev-1"
_API_BASE = "http://test/api"
@pytest_asyncio.fixture
async def dep_gate_setup(db_session: AsyncSession) -> AsyncIterator[dict]:
system = AgentTable(
id=uuid4(),
name="System",
slug=f"system-{uuid4().hex[:8]}",
role=AgentRole.SYSTEM,
team=None,
status=AgentStatus.ACTIVE,
model_config={},
system_prompt="s",
capabilities=[],
permissions={},
metrics={},
)
fe_dev = AgentTable(
id=uuid4(),
name="FE Dev",
slug=f"fe-dev-{uuid4().hex[:8]}",
role=AgentRole.DEVELOPER,
team=Team.FRONTEND,
status=AgentStatus.ACTIVE,
model_config={},
system_prompt="s",
capabilities=[],
permissions={},
metrics={},
)
db_session.add_all([system, fe_dev])
await db_session.flush()
fe_project = ProjectTable(
id=uuid4(),
name="FE",
slug=f"fe-{uuid4().hex[:6]}",
git_url="https://example.com/fe.git",
assigned_cell=Team.FRONTEND,
created_by=system.id,
)
ux_project = ProjectTable(
id=uuid4(),
name="UX",
slug=f"ux-{uuid4().hex[:6]}",
git_url="https://example.com/ux.git",
assigned_cell=Team.UX_UI,
created_by=system.id,
)
product = ProductTable(
id=uuid4(),
name="Prod",
slug=f"prod-{uuid4().hex[:6]}",
created_by=system.id,
)
db_session.add_all([fe_project, ux_project, product])
await db_session.flush()
svc = TaskService(db_session)
choreo = Choreographer(
ChoreographerDeps(
task=svc,
work_session=None,
git=None,
a2a=None,
journal=None,
audit=None,
evidence_repo=None,
)
)
app = FastAPI()
app.include_router(tasks_router, prefix="/api/tasks")
async def _override_db() -> AsyncIterator[AsyncSession]:
yield db_session
app.dependency_overrides[get_db] = _override_db
# Bare orchestrator (skip __init__/settings I/O); _api_url is patched to the
# in-process ASGI app so its HTTP dispatch hits the real DB.
orch = AgentOrchestrator.__new__(AgentOrchestrator)
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
with patch.object(
AgentOrchestrator,
"_api_url",
new_callable=PropertyMock,
return_value=_API_BASE,
):
yield {
"svc": svc,
"choreo": choreo,
"orch": orch,
"client": client,
"creator": system.id,
"fe_dev_db_id": fe_dev.id,
"fe_project_id": fe_project.id,
"ux_project_id": ux_project.id,
"product_id": product.id,
}
app.dependency_overrides.clear()
async def _seed_dev_subtask_with_unmet_dep(setup: dict) -> dict:
"""A frontend dev subtask pre-assigned to the dev, depending on a UX task.
Mirrors the product fan-out: a UX/UI cell task and a frontend cell task
under a board root, with a pre-assigned dev subtask under the frontend cell
whose dependency is the (still pending) UX task.
"""
svc: TaskService = setup["svc"]
root = await svc.create(
TaskCreateRequest(
title="Build the feature (board fan-out)",
description="a real coordination task description over twenty chars",
acceptance_criteria=["delegated to frontend + ux_ui cells"],
team=Team.BOARD,
created_by=setup["creator"],
project_id=None,
product_id=setup["product_id"],
task_type=TaskType.CODE,
nature=TaskNature.NON_TECHNICAL,
estimated_complexity=Complexity.HIGH,
)
)
ux_cell = await svc.create_subtask(
TaskCreateRequest(
title="UX/UI design for the feature",
description="a real ux design task description over twenty chars",
acceptance_criteria=["wireframes approved"],
team=Team.UX_UI,
created_by=setup["creator"],
project_id=setup["ux_project_id"],
product_id=setup["product_id"],
parent_task_id=root.id,
task_type=TaskType.DESIGN,
nature=TaskNature.TECHNICAL,
estimated_complexity=Complexity.MEDIUM,
)
)
fe_cell = await svc.create_subtask(
TaskCreateRequest(
title="Frontend implementation for the feature",
description="a real frontend cell task description over twenty chars",
acceptance_criteria=["UI matches the design"],
team=Team.FRONTEND,
created_by=setup["creator"],
project_id=setup["fe_project_id"],
product_id=setup["product_id"],
parent_task_id=root.id,
task_type=TaskType.CODE,
nature=TaskNature.TECHNICAL,
estimated_complexity=Complexity.MEDIUM,
)
)
# Give the frontend cell (the dev subtask's parent) a branch so the
# orchestrator's parent-branch gate is satisfied and the dependency gate is
# the only thing that can hold the dev.
fe_cell.branch_name = "feature/frontend/FECELL01"
await svc.session.flush()
dev_subtask = await svc.create_subtask(
TaskCreateRequest(
title="Implement the login form component",
description="a real dev subtask description over twenty chars long",
acceptance_criteria=["form renders and submits"],
team=Team.FRONTEND,
created_by=setup["creator"],
project_id=setup["fe_project_id"],
product_id=setup["product_id"],
parent_task_id=fe_cell.id,
assigned_to=setup["fe_dev_db_id"],
task_type=TaskType.CODE,
nature=TaskNature.TECHNICAL,
estimated_complexity=Complexity.MEDIUM,
)
)
assert dev_subtask.status == TaskStatus.PENDING
# The dev subtask depends on the UX cell task (cross-cell sequencing).
await svc.add_dependency(dev_subtask.id, ux_cell.id)
await svc.session.flush()
refreshed = await svc.get(dev_subtask.id)
assert refreshed is not None
assert ux_cell.id in refreshed.dependency_ids, (
"precondition: dev subtask must depend on the UX cell task"
)
return {"ux_cell": ux_cell, "fe_cell": fe_cell, "dev_subtask": dev_subtask}
def _as_dict(task: TaskTable) -> dict:
"""The task dict the orchestrator dispatcher operates on."""
return {
"id": str(task.id),
"description": task.description,
"project_id": str(task.project_id) if task.project_id else None,
"product_id": str(task.product_id) if task.product_id else None,
"parent_task_id": str(task.parent_task_id) if task.parent_task_id else None,
"dependency_ids": [str(d) for d in task.dependency_ids],
"estimated_complexity": task.estimated_complexity.value,
"task_type": task.task_type.value,
}
@pytest.mark.asyncio
async def test_all_three_dev_paths_gate_then_release(dep_gate_setup: dict) -> None:
svc: TaskService = dep_gate_setup["svc"]
choreo: Choreographer = dep_gate_setup["choreo"]
orch: AgentOrchestrator = dep_gate_setup["orch"]
client: AsyncClient = dep_gate_setup["client"]
fe_dev_db_id = dep_gate_setup["fe_dev_db_id"]
tree = await _seed_dev_subtask_with_unmet_dep(dep_gate_setup)
ux_cell = tree["ux_cell"]
dev_subtask = tree["dev_subtask"]
dev_dict = _as_dict(dev_subtask)
# --- (a) orchestrator spawn dispatch: _validate_task_for_spawn ---
# REAL boundary: the dev container is spawned by _spawn_pending_dev ->
# _validate_task_for_spawn. With UX pending, it must return a skip reason.
issue = await orch._validate_task_for_spawn(client, dev_dict, _DEV_SLUG)
assert issue is not None, "spawn validation must hold the dev while UX is unmet"
assert "dependency" in issue
# --- (b) give_me_work: list_pending_for_agent ---
offered = await svc.list_pending_for_agent(fe_dev_db_id)
assert dev_subtask.id not in {t.id for t in offered}, (
"give_me_work must not offer the dev subtask while UX is unmet"
)
# --- (c) claim: _run_claim_guards (the i_will_work_on guard set) ---
held = await svc.get(dev_subtask.id)
guard = await choreo._run_claim_guards(agent_id=fe_dev_db_id, task=held)
assert guard is not None, "claim guard must reject while UX is unmet"
assert guard.error == "invalid_state"
assert guard.remediate is not None
assert str(ux_cell.id) in guard.remediate
# --- UX dependency reaches a terminal state ---
ux_row = await svc.get(ux_cell.id)
assert ux_row is not None
ux_row.status = TaskStatus.COMPLETED
await svc.session.flush()
# All three now allow the dev to proceed.
assert await orch._validate_task_for_spawn(client, dev_dict, _DEV_SLUG) is None, (
"spawn validation must allow the dev once UX is terminal"
)
offered_after = await svc.list_pending_for_agent(fe_dev_db_id)
assert dev_subtask.id in {t.id for t in offered_after}, (
"give_me_work must offer the dev subtask once UX is terminal"
)
released = await svc.get(dev_subtask.id)
assert (
await choreo._run_claim_guards(agent_id=fe_dev_db_id, task=released) is None
), "claim guard must allow once UX is terminal"