fix(gateway): covers_parent_criteria hint that teaches the shape; CEO pause/resume (#686)

* fix(gateway): teach the delegate remediate + PM prompt the covers_parent_criteria shape; allow CEO through the plain pause route

- A child draft rejected for missing covers_parent_criteria now gets a
  copy-pasteable corrected skeleton with the parent's real criteria
  inlined, and the PM delegation guidance shows the field as part of
  every child draft — a PM no longer loops on a rejection that named
  the field but never showed the shape.
- The plain pause route now authorizes the CEO tier like its sibling
  lifecycle routes; agent-side pause restrictions are unchanged.

* fix(gateway): delegate-coverage hint heals and degrades on legacy parents

- The coverage-reject path self-heals a criteria-bearing parent whose
  ids are empty or out of length before rendering the hint, so the
  skeleton always shows real references; the renderer itself also
  falls back to quoted criterion texts for any criterion without an
  id instead of emitting a placeholder or truncating the listing.
- The remediate names both legal reference forms (id or exact text)
  again.
- Route comments state the pause/resume check as deliberately
  CEO-only instead of claiming a precedent whose role set is wider.

* test(gateway): real TaskTable rows in the remediation hint round-trips

mypy over tests/ rejects a SimpleNamespace where unknown_ac_refs takes a
TaskTable; instantiating the ORM row directly needs no session and types
cleanly.

---------

Co-authored-by: Renn F <rennf93@users.noreply.github.com>
This commit is contained in:
Renzo F
2026-07-24 17:10:20 +02:00
committed by GitHub
co-authored by Renn F
parent 4b2546ae19
commit 23ae0ca217
9 changed files with 267 additions and 21 deletions
@@ -12,16 +12,31 @@ before any subtask is created.
from __future__ import annotations
from datetime import UTC, datetime
from typing import Any
from typing import TYPE_CHECKING, Any
from unittest.mock import AsyncMock, MagicMock
from uuid import uuid4
import pytest
from roboco.db.tables import AgentTable, ProjectTable, TaskTable
from roboco.models.base import (
AgentRole,
AgentStatus,
Complexity,
TaskNature,
TaskStatus,
TaskType,
Team,
)
from roboco.services.gateway.choreographer import (
Choreographer,
ChoreographerDeps,
DelegateInputs,
)
from roboco.services.task import get_task_service
from sqlalchemy import select
if TYPE_CHECKING:
from sqlalchemy.ext.asyncio import AsyncSession
def _make_deps(**overrides: Any) -> ChoreographerDeps:
@@ -211,3 +226,82 @@ async def test_delegate_wave_leaving_acs_uncovered_still_succeeds() -> None:
assert coverage["covered"] == ["Criterion A"]
assert coverage["uncovered"] == ["Criterion B", "Criterion C"]
task_svc.create_subtask.assert_awaited_once()
@pytest.mark.asyncio
async def test_ac_coverage_guard_heals_empty_ids_parent_in_place(
db_session: AsyncSession,
) -> None:
"""A criteria-bearing parent whose ``acceptance_criteria_ids`` is empty
(a legacy row from before every AC rewrite reconciled ids) is
self-healed to 1:1 by the reject path itself, against a real DB row —
not just papered over in the rendered hint. Regression coverage for the
adversarial finding on commit d259476b: the pre-fix guard rendered a
literal ``'<id>'`` placeholder and an empty criteria listing on exactly
this row shape, re-rejecting a PM who copy-pasted it verbatim."""
agent = AgentTable(
id=uuid4(),
name="PM",
slug=f"pm-{uuid4().hex[:8]}",
role=AgentRole.CELL_PM,
team=Team.BACKEND,
status=AgentStatus.ACTIVE,
model_config={},
system_prompt="pm",
capabilities=[],
permissions={},
metrics={},
)
db_session.add(agent)
await db_session.flush()
project = ProjectTable(
id=uuid4(),
name="P",
slug=f"p-{uuid4().hex[:6]}",
git_url="https://example.com/r.git",
assigned_cell=Team.BACKEND,
created_by=agent.id,
)
db_session.add(project)
await db_session.flush()
tid = uuid4()
db_session.add(
TaskTable(
id=tid,
title="parent",
description="d",
acceptance_criteria=["Criterion A", "Criterion B"],
acceptance_criteria_ids=[],
status=TaskStatus.IN_PROGRESS,
priority=2,
task_type=TaskType.CODE,
nature=TaskNature.TECHNICAL,
estimated_complexity=Complexity.LOW,
team=Team.BACKEND,
confirmed_by_human=True,
project_id=project.id,
created_by=agent.id,
branch_name="feature/x",
assigned_to=agent.id,
)
)
await db_session.flush()
task_svc = get_task_service(db_session)
parent = await task_svc.get(tid)
assert parent is not None
deps = _make_deps(task=task_svc)
c = Choreographer(deps)
env = await c._delegate_ac_coverage_guard(parent, _inputs(title="Orphan slice"))
assert env is not None
body = env.as_dict()
assert "'<id>'" not in body["remediate"]
assert "Criterion A" in body["remediate"]
assert "Criterion B" in body["remediate"]
row = (
await db_session.execute(select(TaskTable).where(TaskTable.id == tid))
).scalar_one()
assert len(row.acceptance_criteria_ids) == len(row.acceptance_criteria)
assert len(set(row.acceptance_criteria_ids)) == len(row.acceptance_criteria_ids)
+61
View File
@@ -2,11 +2,14 @@
from __future__ import annotations
from roboco.db.tables import TaskTable
from roboco.services.gateway.remediation import (
hint_for_missing_ac_coverage,
hint_for_missing_progress,
hint_for_missing_reflect,
hint_for_unaddressed_acceptance_criteria,
)
from roboco.services.task import TaskService
def test_missing_progress_hint() -> None:
@@ -27,3 +30,61 @@ def test_unaddressed_criteria_hint() -> None:
assert "criterion 1" in h
assert "criterion 3" in h
assert "t-1" in h
def test_missing_ac_coverage_hint_shows_call_shape_and_real_ids() -> None:
h = hint_for_missing_ac_coverage(
ids=["id-a", "id-b"],
texts=["Criterion A", "Criterion B"],
title="Orphan slice",
)
assert "delegate(title='Orphan slice'" in h
assert "covers_parent_criteria=['id-a']" in h
assert 'id-a="Criterion A"' in h
assert 'id-b="Criterion B"' in h
def test_missing_ac_coverage_hint_names_both_legal_reference_forms() -> None:
"""The remediate names both legal ``covers_parent_criteria`` forms —
a criterion's id or its exact text — not just id."""
h = hint_for_missing_ac_coverage(ids=["id-a"], texts=["Criterion A"], title="X")
assert "id(s) or exact text(s)" in h
def test_missing_ac_coverage_hint_handles_no_criteria() -> None:
h = hint_for_missing_ac_coverage(ids=[], texts=[], title="X")
assert "<id>" in h
def test_missing_ac_coverage_hint_empty_ids_uses_quoted_text() -> None:
"""A legacy/unhealed parent whose ids are empty must still get a real,
copy-pasteable reference for every criterion — never a `'<id>'`
placeholder, and never a truncated listing."""
texts = ["Criterion A", "Criterion B", "Criterion C"]
h = hint_for_missing_ac_coverage(ids=[], texts=texts, title="Orphan slice")
assert "'<id>'" not in h
for text in texts:
assert text in h
parent = TaskTable(acceptance_criteria_ids=[], acceptance_criteria=texts)
for text in texts:
assert TaskService.unknown_ac_refs(parent, [text]) == []
def test_missing_ac_coverage_hint_drifted_ids_shorter_than_criteria() -> None:
"""1 id against 3 criteria: every criterion is listed (id for the
first, quoted text for the rest) — zip's `strict=False` used to drop
the tail criteria silently."""
texts = ["Criterion A", "Criterion B", "Criterion C"]
h = hint_for_missing_ac_coverage(ids=["id-a"], texts=texts, title="Orphan slice")
assert "'<id>'" not in h
assert 'id-a="Criterion A"' in h
for text in texts[1:]:
assert text in h
parent = TaskTable(acceptance_criteria_ids=["id-a"], acceptance_criteria=texts)
assert TaskService.unknown_ac_refs(parent, ["id-a"]) == []
for text in texts[1:]:
assert TaskService.unknown_ac_refs(parent, [text]) == []