External/roboco
1
0
Fork 0
mirror of https://github.com/rennf93/roboco.git synced 2026-08-03 07:23:24 +02:00
Code Issues Packages Projects Releases Wiki Activity
Files
master
roboco/tests/integration/test_revision_findings_real_db.py
T
Renzo FGitHubRenn F
 879afc14a4 Board Program LEARN context, ruff 0.16, and verb-rejection observability (#700) 
* fix(board): LEARN decisions name the item, not its per-cycle index

A cycle's reject reasons are rendered into the NEXT cycle's exploration
prompt, but the ref recorded alongside each reason was the item's stored
id (item-0/item-1) — a per-cycle index that means something different
every cycle and appears nowhere the explorer can resolve. The reason
survived the loop; what it was about did not.

Record the item's title instead, via a shared learn_ref() helper (falls
back to the id when title-less, and reads target_task_title for Scales,
whose items name the live task they mutate).

* chore(lint): satisfy ruff 0.16 — keyword-only signatures and markdown formatting

The dev toolchain resolved ruff 0.16.0, which stabilises PLR0917 (too many
positional arguments) and formats python code blocks inside markdown. Both
fired repo-wide and neither had anything to do with the code they flagged.

- 36 signatures gain a `*` so their tail arguments are keyword-only, and
  the 104 call sites that passed them positionally are converted. mypy was
  the safety net for the static ones; the full suite caught nine more that
  only bind at runtime (the MCP tool functions, whose real callers already
  pass named JSON arguments).
- 28 markdown files reformatted by 0.16's code-block formatter.
- One RUF036 (`None` mid-union) autofixed in the GitLab provider.

* fix(gateway): log the reason when a verb rejects

A rejected envelope rides an HTTP 200, its body is never logged, and there
is no trace table — so in the access log a verb an agent could not satisfy
looks identical to one that worked. On 2026-07-25 four Board Programs
(Periscope, Sentinel, Scales, Barfly) each POSTed their propose verb three
or four times, persisted nothing, and left their exploration tasks PENDING;
the reason was unrecoverable afterwards, from the logs or from the agents'
own transcripts.

Log error/message/remediate/missing plus the calling agent at
envelope_to_response — the one chokepoint every v1 flow and do route
returns through. Success envelopes stay silent.

---------

Co-authored-by: Renn F <rennf93@users.noreply.github.com>
2026-07-26 15:07:28 +02:00

1376 lines
46 KiB
Python
Raw Permalink Blame History

"""Tier 3 real-DB tests — the revision-findings ledger + i_am_done gate.
Mirrors ``test_lifecycle_real_db.py``: only git is stubbed; the spec,
choreographer, VerbRunner, TaskService, and the ledger repository are all
real. Covers the four producers (fail_review / pr_fail / request_changes /
ceo_reject) persisting structured findings, the dev_notes data-loss-bug fix,
and the i_am_done FINDINGS_ADDRESSED resolution gate end to end.
"""
from __future__ import annotations
from datetime import UTC, datetime
from typing import TYPE_CHECKING, Any
from unittest.mock import AsyncMock
from uuid import UUID, uuid4
import pytest
import pytest_asyncio
from roboco.db.tables import AgentTable, AuditLogTable, ProjectTable, TaskTable
from roboco.foundation.policy.content import Finding, Severity
from roboco.foundation.policy.lifecycle import Status
from roboco.models.base import (
AgentRole,
AgentStatus,
TaskNature,
TaskStatus,
TaskType,
Team,
)
from roboco.seeds.initial_data import AGENT_UUIDS
from roboco.services.base import ValidationError
from roboco.services.gateway.choreographer import Choreographer, ChoreographerDeps
from roboco.services.repositories.review_findings import (
STATUS_ADDRESSED,
STATUS_OPEN,
STATUS_VERIFIED,
ReviewFindingsRepository,
)
from roboco.services.task import TaskService
from sqlalchemy import select
if TYPE_CHECKING:
from collections.abc import AsyncIterator
from sqlalchemy.ext.asyncio import AsyncSession
_BRANCH = "feature/backend/healthz"
_PR_NUMBER = 8
_PR_URL = "https://github.com/example/findings/pull/8"
_EXPECTED_FINDING_COUNT = 2
_EXPECTED_ROUND_COUNT = 2
_GOOD_PLAN = (
"Implement the task on its feature branch: edit the target module, add or "
"update unit tests covering the change, run the suite locally, then commit "
"on the branch and open a PR. Keep the diff focused on the acceptance "
"criteria and verify it before submitting for QA."
)
_STEPS = [
{
"title": "Implement the change",
"description": (
"edit the target file, add tests, run them, and stage the change "
"for commit on the task branch"
),
}
]
_GOOD_TC = ["Follow the existing module's patterns; keep the change minimal."]
_GOOD_RISKS = [
{
"risk": "Scope creep balloons the diff and slows review.",
"mitigation": "Touch only the files the acceptance criteria require.",
}
]
class _StubGit:
"""Deterministic GitService stub — mirrors test_lifecycle_real_db.py."""
def __init__(self, session: Any, task: TaskTable) -> None:
self._session = session
self._task = task
async def commit(
self,
*,
branch_name: str,
message: str,
task_id: UUID,
files: list[str] | None = None,
actor_agent_id: Any = None,
) -> dict[str, Any]:
del branch_name, files, actor_agent_id
sha = uuid4().hex[:40]
commits = list(self._task.commits or [])
commits.append({"sha": sha, "message": message, "task_id": str(task_id)})
self._task.commits = commits
await self._session.flush()
return {"sha": sha, "message": message, "files_changed": 1}
async def push_branch(
self, branch_name: str, *, actor_agent_id: Any = None
) -> tuple[str, int]:
del branch_name, actor_agent_id
return ("ok", 0)
async def push_task_branch(self, agent_id: UUID, task_id: UUID) -> int:
del agent_id, task_id
return 0
async def create_pr(
self,
branch_name: str,
*,
parent: str,
is_root_pr: bool,
actor_agent_id: Any = None,
) -> dict[str, Any]:
del branch_name, parent, actor_agent_id
self._task.pr_number = _PR_NUMBER
self._task.pr_url = _PR_URL
self._task.pr_created = True
await self._session.flush()
return {"pr_number": _PR_NUMBER, "pr_url": _PR_URL, "is_root_pr": is_root_pr}
async def diff(
self,
*,
branch_name: str,
base: Any = None,
actor_agent_id: Any = None,
preferred_parent: Any = None,
) -> str:
del branch_name, base, actor_agent_id, preferred_parent
return "stub diff"
async def list_changed_files(
self, *, branch_name: str, base: Any = None, actor_agent_id: Any = None
) -> list[str]:
del branch_name, base, actor_agent_id
return ["roboco/api/routes/health.py"]
async def pr_target(self, pr_number: int, *, actor_agent_id: Any = None) -> str:
del pr_number, actor_agent_id
return "main"
async def pr_merge(self, *args: Any, **kwargs: Any) -> dict[str, Any]:
del args, kwargs
return {
"merged": True,
"sha": uuid4().hex[:40],
"merge_commit_sha": uuid4().hex[:40],
}
async def is_pr_merged_for_task(self, task_id: UUID) -> bool:
del task_id
return False
async def get_pr_head_sha(self, slug: str, pr_number: int) -> str:
del slug, pr_number
return uuid4().hex[:40]
async def post_pr_review(
self, slug: str, pr_number: int, body: str, *, event: str
) -> None:
del slug, pr_number, body, event
def _mock_evidence_repo() -> Any:
repo = AsyncMock()
for method in (
"list_unread_a2a",
"list_unread_mentions",
"list_pending_notifications",
"task_metadata_gaps",
"recent_team_activity",
"blockers_in_lane",
"journal_highlights_for_task",
):
getattr(repo, method).return_value = []
return repo
def _mock_journal_with_reflect() -> Any:
journal = AsyncMock()
journal.has_reflect_for_task.return_value = True
journal.has_learning_for_task.return_value = True
journal.has_decision_for_task.return_value = True
journal.has_struggle_for_task.return_value = False
journal.has_note_for_task.return_value = True
journal.latest_decision_at.return_value = datetime.now(UTC)
return journal
def _mock_work_session() -> Any:
ws = AsyncMock()
ws.files_changed.return_value = ["roboco/api/routes/health.py"]
ws.has_unpushed_commits.return_value = False
return ws
async def _seed_agents_and_project(db_session: AsyncSession) -> dict[str, Any]:
system_agent = AgentTable(
id=uuid4(),
name="System",
slug=f"system-{uuid4().hex[:8]}",
role=AgentRole.SYSTEM,
team=None,
status=AgentStatus.ACTIVE,
model_config={},
system_prompt="system",
capabilities=[],
permissions={},
metrics={},
)
db_session.add(system_agent)
await db_session.flush()
project = ProjectTable(
id=uuid4(),
name="Findings Ledger Test Project",
slug=f"findings-{uuid4().hex[:8]}",
git_url="https://github.com/example/findings.git",
default_branch="main",
protected_branches=["main"],
assigned_cell=Team.BACKEND,
created_by=system_agent.id,
is_active=True,
)
db_session.add(project)
await db_session.flush()
dev_agent = AgentTable(
id=uuid4(),
name="BE Dev 1",
slug="be-dev-1",
role=AgentRole.DEVELOPER,
team=Team.BACKEND,
status=AgentStatus.ACTIVE,
model_config={},
system_prompt="dev",
capabilities=["python"],
permissions={},
metrics={},
)
qa_agent = AgentTable(
id=uuid4(),
name="BE QA",
slug="be-qa",
role=AgentRole.QA,
team=Team.BACKEND,
status=AgentStatus.ACTIVE,
model_config={},
system_prompt="qa",
capabilities=["review"],
permissions={},
metrics={},
)
cell_pm_agent = AgentTable(
id=uuid4(),
name="BE Cell PM",
slug="be-pm",
role=AgentRole.CELL_PM,
team=Team.BACKEND,
status=AgentStatus.ACTIVE,
model_config={},
system_prompt="cell_pm",
capabilities=["coord"],
permissions={},
metrics={},
)
reviewer_agent = AgentTable(
id=uuid4(),
name="BE PR Reviewer",
slug="be-pr-reviewer",
role=AgentRole.PR_REVIEWER,
team=Team.BACKEND,
status=AgentStatus.ACTIVE,
model_config={},
system_prompt="reviewer",
capabilities=["review"],
permissions={},
metrics={},
)
main_pm_agent = AgentTable(
id=uuid4(),
name="Main PM",
slug=f"main-pm-{uuid4().hex[:8]}",
role=AgentRole.MAIN_PM,
team=Team.MAIN_PM,
status=AgentStatus.ACTIVE,
model_config={},
system_prompt="main_pm",
capabilities=["coord"],
permissions={},
metrics={},
)
db_session.add_all(
[dev_agent, qa_agent, cell_pm_agent, reviewer_agent, main_pm_agent]
)
await db_session.flush()
return {
"system_agent": system_agent,
"project": project,
"dev_agent": dev_agent,
"qa_agent": qa_agent,
"cell_pm_agent": cell_pm_agent,
"reviewer_agent": reviewer_agent,
"main_pm_agent": main_pm_agent,
}
def _build_task(
*, project_id: UUID, creator_id: UUID, assignee_id: UUID | None, status: TaskStatus
) -> TaskTable:
return TaskTable(
id=uuid4(),
title="Add /healthz endpoint",
description="Return 200 OK from /healthz",
status=status,
priority=2,
task_type=TaskType.CODE,
nature=TaskNature.TECHNICAL,
team=Team.BACKEND,
project_id=project_id,
created_by=creator_id,
assigned_to=assignee_id,
branch_name=_BRANCH,
acceptance_criteria=["Returns 200", "Includes timestamp"],
)
@pytest_asyncio.fixture
async def setup(db_session: AsyncSession) -> AsyncIterator[dict[str, Any]]:
seeded = await _seed_agents_and_project(db_session)
task = _build_task(
project_id=seeded["project"].id,
creator_id=seeded["system_agent"].id,
assignee_id=seeded["dev_agent"].id,
status=TaskStatus.PENDING,
)
db_session.add(task)
await db_session.flush()
seeded["task"] = task
yield seeded
def _choreographer(db_session: AsyncSession, task: TaskTable) -> Choreographer:
deps = ChoreographerDeps(
task=TaskService(db_session),
work_session=_mock_work_session(),
git=_StubGit(db_session, task),
a2a=AsyncMock(),
journal=_mock_journal_with_reflect(),
audit=AsyncMock(),
evidence_repo=_mock_evidence_repo(),
)
return Choreographer(deps)
# ---------------------------------------------------------------------------
# fail_review persists structured findings
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_fail_review_persists_findings_round1_and_qa_notes(
db_session: AsyncSession, setup: dict[str, Any]
) -> None:
task = setup["task"]
qa_agent = setup["qa_agent"]
dev_agent = setup["dev_agent"]
task.status = TaskStatus.AWAITING_QA
task.pr_number = _PR_NUMBER
task.pr_url = _PR_URL
task.commits = [
{"sha": uuid4().hex[:40], "message": "feat: x", "task_id": str(task.id)}
]
task.self_verified = True
task.orchestration_markers = {"original_developer": str(dev_agent.id)}
await db_session.flush()
task_service = TaskService(db_session)
c = _choreographer(db_session, task)
await c.claim_review(qa_agent.id, task.id)
env = await c.fail_review(
qa_agent.id,
task.id,
findings=[
{
"file": "roboco/api/routes/health.py",
"line": 12,
"severity": "major",
"expected": "returns 200",
"actual": "returns 500 on the timestamp branch",
}
],
)
assert env.error is None, env.as_dict()
assert env.status == Status.NEEDS_REVISION.value
findings = await ReviewFindingsRepository(db_session).list_for_task(task.id)
assert len(findings) == 1
assert findings[0].origin == "qa"
assert findings[0].round == 1
assert findings[0].status == STATUS_OPEN
# GatewayAgentView (self.task.agent_for) carries no slug field, so
# author_slug falls back to the role string — mirrors the same
# established fallback _post_gate_review uses for reviewer_slug.
assert findings[0].author_slug == "qa"
final = await task_service.get(task.id)
assert final is not None
assert "returns 500 on the timestamp branch" in (final.qa_notes or "")
assert f"[F-{str(findings[0].id)[:8]}]" in (final.qa_notes or "")
@pytest.mark.asyncio
async def test_fail_review_issues_shim_still_works(
db_session: AsyncSession, setup: dict[str, Any]
) -> None:
task = setup["task"]
qa_agent = setup["qa_agent"]
dev_agent = setup["dev_agent"]
task.status = TaskStatus.AWAITING_QA
task.pr_number = _PR_NUMBER
task.commits = [
{"sha": uuid4().hex[:40], "message": "feat: x", "task_id": str(task.id)}
]
task.self_verified = True
task.orchestration_markers = {"original_developer": str(dev_agent.id)}
await db_session.flush()
c = _choreographer(db_session, task)
await c.claim_review(qa_agent.id, task.id)
env = await c.fail_review(
qa_agent.id,
task.id,
issues=["Missing test coverage for the timestamp branch of /healthz"],
)
assert env.error is None, env.as_dict()
findings = await ReviewFindingsRepository(db_session).list_for_task(task.id)
assert len(findings) == 1
assert findings[0].file is None
assert findings[0].severity == "major"
assert (
findings[0].actual
== "Missing test coverage for the timestamp branch of /healthz"
)
@pytest.mark.asyncio
async def test_fail_review_rejects_over_hard_cap(
db_session: AsyncSession, setup: dict[str, Any]
) -> None:
task = setup["task"]
qa_agent = setup["qa_agent"]
task.status = TaskStatus.AWAITING_QA
task.pr_number = _PR_NUMBER
task.commits = [
{"sha": uuid4().hex[:40], "message": "feat: x", "task_id": str(task.id)}
]
task.self_verified = True
await db_session.flush()
c = _choreographer(db_session, task)
await c.claim_review(qa_agent.id, task.id)
findings = [
{
"expected": f"expected {i}",
"actual": f"actual finding number {i}",
"severity": "minor",
}
for i in range(11)
]
env = await c.fail_review(qa_agent.id, task.id, findings=findings)
body = env.as_dict()
assert body["error"] == "invalid_state"
assert "11" in body["message"]
assert await ReviewFindingsRepository(db_session).list_for_task(task.id) == []
@pytest.mark.asyncio
async def test_fail_review_nudge_warns_over_five(
db_session: AsyncSession, setup: dict[str, Any]
) -> None:
task = setup["task"]
qa_agent = setup["qa_agent"]
dev_agent = setup["dev_agent"]
task.status = TaskStatus.AWAITING_QA
task.pr_number = _PR_NUMBER
task.commits = [
{"sha": uuid4().hex[:40], "message": "feat: x", "task_id": str(task.id)}
]
task.self_verified = True
task.orchestration_markers = {"original_developer": str(dev_agent.id)}
await db_session.flush()
c = _choreographer(db_session, task)
await c.claim_review(qa_agent.id, task.id)
findings = [
{
"expected": f"expected {i}",
"actual": f"actual finding number {i}",
"severity": "minor",
}
for i in range(6)
]
env = await c.fail_review(qa_agent.id, task.id, findings=findings)
assert env.error is None, env.as_dict()
assert env.warning is not None
assert "6" in env.warning
# ---------------------------------------------------------------------------
# pr_fail persists structured findings into pr_reviewer_notes
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_pr_fail_persists_findings_and_pr_reviewer_notes(
db_session: AsyncSession, setup: dict[str, Any]
) -> None:
task = setup["task"]
reviewer = setup["reviewer_agent"]
task.status = TaskStatus.AWAITING_PR_REVIEW
task.assigned_to = reviewer.id
task.claimed_by = reviewer.id
task.pr_number = _PR_NUMBER
await db_session.flush()
task_service = TaskService(db_session)
c = _choreographer(db_session, task)
env = await c.pr_fail(
UUID(str(reviewer.id)),
task.id,
findings=[
{
"file": "roboco/api/routes/health.py",
"severity": "blocker",
"expected": "task_id is a UUID",
"actual": "FE sends task_id as a string",
}
],
)
assert env.error is None, env.as_dict()
assert env.status == Status.NEEDS_REVISION.value
findings = await ReviewFindingsRepository(db_session).list_for_task(task.id)
assert len(findings) == 1
assert findings[0].origin == "pr_gate"
# GatewayAgentView carries no slug field — falls back to the role string.
assert findings[0].author_slug == "pr_reviewer"
final = await task_service.get(task.id)
assert final is not None
assert "FE sends task_id as a string" in (final.pr_reviewer_notes or "")
assert (final.notes_structured or {}).get("pr_review", {}).get(
"verdict"
) == "failed"
assert (
len((final.notes_structured or {}).get("pr_review", {}).get("findings", []))
== 1
)
# ---------------------------------------------------------------------------
# request_changes persists structured findings into the new pm_notes slot
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_request_changes_persists_findings_and_pm_notes(
db_session: AsyncSession, setup: dict[str, Any]
) -> None:
task = setup["task"]
cell_pm_agent = setup["cell_pm_agent"]
task.status = TaskStatus.AWAITING_PM_REVIEW
task.assigned_to = cell_pm_agent.id
task.pr_number = _PR_NUMBER
await db_session.flush()
task_service = TaskService(db_session)
c = _choreographer(db_session, task)
env = await c.request_changes(
cell_pm_agent.id,
task.id,
findings=[
{
"file": "frontend/CLAUDE.md",
"severity": "major",
"expected": "scope limited to the endpoint change",
"actual": "CLAUDE.md modified out of scope",
}
],
)
assert env.error is None, env.as_dict()
assert env.status == Status.NEEDS_REVISION.value
findings = await ReviewFindingsRepository(db_session).list_for_task(task.id)
assert len(findings) == 1
assert findings[0].origin == "pm"
final = await task_service.get(task.id)
assert final is not None
assert "CLAUDE.md modified out of scope" in (final.pm_notes or "")
assert (final.notes_structured or {}).get("pm_review") is not None
@pytest.mark.asyncio
async def test_request_changes_dev_notes_survives_a_later_handoff_note(
db_session: AsyncSession, setup: dict[str, Any]
) -> None:
"""Regression: request_changes must NOT raw-append onto dev_notes — the
data-loss bug the ledger fix retires. A later developer handoff note
(apply_structured_note's real overwrite semantics) must show only its OWN
content; the PM's reject lives in pm_notes + the ledger instead."""
task = setup["task"]
cell_pm_agent = setup["cell_pm_agent"]
task.status = TaskStatus.AWAITING_PM_REVIEW
task.assigned_to = cell_pm_agent.id
task.pr_number = _PR_NUMBER
task.dev_notes = "original developer summary"
await db_session.flush()
task_service = TaskService(db_session)
c = _choreographer(db_session, task)
env = await c.request_changes(
cell_pm_agent.id,
task.id,
issues=["revert the out-of-scope doc change"],
)
assert env.error is None, env.as_dict()
mid = await task_service.get(task.id)
assert mid is not None
# dev_notes untouched by request_changes.
assert mid.dev_notes == "original developer summary"
# A later developer handoff note overwrites dev_notes (its normal,
# correct behavior) — this must not have destroyed anything, because
# request_changes never wrote to dev_notes in the first place.
await task_service.record_section_note(
task.id, "developer", {"summary": "Reverted the doc change; back on scope."}
)
final = await task_service.get(task.id)
assert final is not None
assert "Reverted the doc change" in (final.dev_notes or "")
# The PM's reject is still fully intact in its own channels.
assert "revert the out-of-scope doc change" in (final.pm_notes or "")
ledger = await ReviewFindingsRepository(db_session).list_for_task(task.id)
assert len(ledger) == 1
# ---------------------------------------------------------------------------
# ceo_reject persists one origin=ceo finding (TaskService-level, no gateway)
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_ceo_reject_persists_one_finding(
db_session: AsyncSession, setup: dict[str, Any]
) -> None:
task = setup["task"]
task.status = TaskStatus.AWAITING_CEO_APPROVAL
task.pr_number = _PR_NUMBER
await db_session.flush()
task_service = TaskService(db_session)
result = await task_service.ceo_reject(
task.id, "The AC2 timestamp format does not match the API contract."
)
assert result is not None
assert result.status == TaskStatus.NEEDS_REVISION
findings = await ReviewFindingsRepository(db_session).list_for_task(task.id)
assert len(findings) == 1
assert findings[0].origin == "ceo"
assert findings[0].author_slug == "ceo"
assert "AC2 timestamp format" in findings[0].actual
# ---------------------------------------------------------------------------
# i_am_done — the FINDINGS_ADDRESSED resolution gate, end to end
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_i_am_done_full_chain_blocks_then_resolves(
db_session: AsyncSession, setup: dict[str, Any]
) -> None:
"""pending -> in_progress -> awaiting_qa (round 1, no findings, passes
untouched) -> QA fails with 2 findings -> dev recovers -> i_am_done
blocked (both ids named) -> i_am_done(resolved_findings=[...]) -> awaiting_qa."""
task = setup["task"]
dev_agent = setup["dev_agent"]
qa_agent = setup["qa_agent"]
task_service = TaskService(db_session)
stub_git = _StubGit(db_session, task)
c = Choreographer(
ChoreographerDeps(
task=task_service,
work_session=_mock_work_session(),
git=stub_git,
a2a=AsyncMock(),
journal=_mock_journal_with_reflect(),
audit=AsyncMock(),
evidence_repo=_mock_evidence_repo(),
)
)
# --- Round 1: no findings exist yet — i_am_done must pass untouched. ---
env = await c.i_will_work_on(
agent_id=dev_agent.id,
task_id=task.id,
plan=_GOOD_PLAN,
steps=_STEPS,
technical_considerations=_GOOD_TC,
risks=_GOOD_RISKS,
)
assert env.error is None, env.as_dict()
await stub_git.commit(
branch_name=_BRANCH, message="feat(api): add /healthz", task_id=task.id
)
await task_service.add_progress(task.id, dev_agent.id, "implemented /healthz")
env = await c.open_pr(dev_agent.id, task.id)
assert env.error is None, env.as_dict()
await task_service.record_section_note(
task.id,
"developer",
{"summary": "Implemented /healthz with a happy-path test."},
)
env = await c.i_am_done(dev_agent.id, task.id, "tests pass")
assert env.error is None, (
f"round 1 i_am_done should pass untouched: {env.as_dict()}"
)
assert env.status == Status.AWAITING_QA.value
# --- QA fails with two structured findings. ---
env = await c.claim_review(qa_agent.id, task.id)
assert env.error is None, env.as_dict()
env = await c.fail_review(
qa_agent.id,
task.id,
findings=[
{
"expected": "200 on the timestamp branch",
"actual": "500 on the timestamp branch",
"severity": "blocker",
},
{
"expected": "test covers both branches",
"actual": "only the happy path is tested",
"severity": "major",
},
],
)
assert env.error is None, env.as_dict()
assert env.status == Status.NEEDS_REVISION.value
open_findings = await ReviewFindingsRepository(db_session).list_for_task(
task.id, status=STATUS_OPEN
)
assert len(open_findings) == _EXPECTED_FINDING_COUNT
open_ids = [str(f.id)[:8] for f in open_findings]
# --- Dev recovers into in_progress (needs_revision -> in_progress).
# `_dev_reentry` only short-circuits in_progress/claimed same-agent
# re-entry; needs_revision falls through to `_fresh_dev_claim`, which
# re-enforces the rich-plan gate for a developer — so the full plan is
# supplied again here, same as the very first claim. ---
env = await c.i_will_work_on(
agent_id=dev_agent.id,
task_id=task.id,
plan=_GOOD_PLAN,
steps=_STEPS,
technical_considerations=_GOOD_TC,
risks=_GOOD_RISKS,
)
assert env.error is None, env.as_dict()
assert env.status == Status.IN_PROGRESS.value
# --- i_am_done WITHOUT resolving the findings is blocked, naming both ids. ---
env = await c.i_am_done(dev_agent.id, task.id, "fixed the bug")
body = env.as_dict()
assert body["error"] == "tracing_gap", body
for fid in open_ids:
assert f"finding:{fid}" in body["missing"], body["missing"]
assert all(fid in body["remediate"] for fid in open_ids)
# --- i_am_done WITH resolved_findings for both now succeeds. ---
env = await c.i_am_done(
dev_agent.id,
task.id,
"fixed the bug and added the missing test",
resolved_findings=[
{"finding_id": open_ids[0], "commit": "abc123", "note": "fixed the 500"},
{"finding_id": open_ids[1], "commit": "abc123", "note": "added the test"},
],
)
assert env.error is None, f"resolved i_am_done should succeed: {env.as_dict()}"
assert env.status == Status.AWAITING_QA.value
remaining_open = await ReviewFindingsRepository(db_session).list_for_task(
task.id, status=STATUS_OPEN
)
assert remaining_open == []
addressed = await ReviewFindingsRepository(db_session).list_for_task(
task.id, status=STATUS_ADDRESSED
)
assert len(addressed) == _EXPECTED_FINDING_COUNT
# ---------------------------------------------------------------------------
# Mixed findings + issues — both must land, not silently drop one (fail_review)
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_fail_review_merges_findings_and_issues_instead_of_dropping_one(
db_session: AsyncSession, setup: dict[str, Any]
) -> None:
"""Live repro: ``findings if findings else issues_to_findings(issues)``
silently dropped ``issues`` whenever ``findings`` was ALSO supplied. Both
sources must land in the ledger."""
task = setup["task"]
qa_agent = setup["qa_agent"]
dev_agent = setup["dev_agent"]
task.status = TaskStatus.AWAITING_QA
task.pr_number = _PR_NUMBER
task.commits = [
{"sha": uuid4().hex[:40], "message": "feat: x", "task_id": str(task.id)}
]
task.self_verified = True
task.orchestration_markers = {"original_developer": str(dev_agent.id)}
await db_session.flush()
c = _choreographer(db_session, task)
await c.claim_review(qa_agent.id, task.id)
env = await c.fail_review(
qa_agent.id,
task.id,
findings=[
{
"file": "roboco/api/routes/health.py",
"severity": "major",
"expected": "returns 200",
"actual": "returns 500 on the timestamp branch",
}
],
issues=["also missing test coverage for the timestamp branch"],
)
assert env.error is None, env.as_dict()
findings = await ReviewFindingsRepository(db_session).list_for_task(task.id)
assert len(findings) == _EXPECTED_FINDING_COUNT
actuals = {f.actual for f in findings}
assert "returns 500 on the timestamp branch" in actuals
assert "also missing test coverage for the timestamp branch" in actuals
@pytest.mark.asyncio
async def test_fail_review_combined_findings_and_issues_hit_hard_cap(
db_session: AsyncSession, setup: dict[str, Any]
) -> None:
"""The 10-item hard cap applies to the MERGED findings+issues total, not
just whichever source the old ``if findings else`` picked."""
task = setup["task"]
qa_agent = setup["qa_agent"]
task.status = TaskStatus.AWAITING_QA
task.pr_number = _PR_NUMBER
task.commits = [
{"sha": uuid4().hex[:40], "message": "feat: x", "task_id": str(task.id)}
]
task.self_verified = True
await db_session.flush()
c = _choreographer(db_session, task)
await c.claim_review(qa_agent.id, task.id)
findings = [
{"expected": f"expected {i}", "actual": f"actual {i}", "severity": "minor"}
for i in range(6)
]
issues = [f"issue number {i}" for i in range(5)]
env = await c.fail_review(qa_agent.id, task.id, findings=findings, issues=issues)
body = env.as_dict()
assert body["error"] == "invalid_state"
assert "11" in body["message"]
assert await ReviewFindingsRepository(db_session).list_for_task(task.id) == []
# ---------------------------------------------------------------------------
# claim_gate_review evidence: open findings ride alongside the full ledger
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_claim_gate_review_evidence_surfaces_open_findings_separately(
db_session: AsyncSession, setup: dict[str, Any]
) -> None:
"""Live repro: ``_build_gate_review_evidence`` returned only
``prior_findings`` (the full ledger, capped at 10) — 3 OPEN rows got
crowded out by the cap. ``revision_findings`` (open-only) must ride
alongside, mirroring QA's ``claim_review`` evidence."""
task = setup["task"]
reviewer = setup["reviewer_agent"]
task.status = TaskStatus.AWAITING_PR_REVIEW
task.assigned_to = reviewer.id
task.claimed_by = reviewer.id
task.pr_number = _PR_NUMBER
await db_session.flush()
repo = ReviewFindingsRepository(db_session)
await repo.insert_many(
task_id=task.id,
origin="pr_gate",
round=1,
author_slug="be-pr-reviewer",
findings=[
Finding(
severity=Severity.MAJOR,
expected="task_id is a UUID",
actual="FE sends task_id as a string",
)
],
)
c = _choreographer(db_session, task)
env = await c.claim_gate_review(reviewer.id, task.id)
assert env.error is None, env.as_dict()
evidence = env.evidence
assert evidence is not None
assert len(evidence["revision_findings"]) == 1
assert evidence["revision_findings"][0]["status"] == STATUS_OPEN
assert len(evidence["prior_findings"]) == 1
# ---------------------------------------------------------------------------
# submit_up / submit_root — FINDINGS_ADDRESSED gate + resolution
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_submit_up_blocked_by_open_finding_then_resolved(
db_session: AsyncSession, setup: dict[str, Any]
) -> None:
"""Live gap: submit_up had no equivalent of i_am_done's FINDINGS_ADDRESSED
gate — a PM could re-submit an assembled cell root with open pr_gate/pm/
ceo-origin findings still unaddressed."""
cell_pm = setup["cell_pm_agent"]
task: Any = TaskTable(
id=uuid4(),
title="Assemble the healthz cell work",
description="Bubble the cell's healthz work up to the Main PM",
status=TaskStatus.IN_PROGRESS,
priority=2,
task_type=TaskType.CODE,
nature=TaskNature.TECHNICAL,
team=Team.BACKEND,
project_id=setup["project"].id,
created_by=setup["system_agent"].id,
assigned_to=cell_pm.id,
claimed_by=cell_pm.id,
branch_name="feature/backend/healthz-cell",
acceptance_criteria=["Returns 200"],
)
db_session.add(task)
await db_session.flush()
repo = ReviewFindingsRepository(db_session)
rows = await repo.insert_many(
task_id=task.id,
origin="pr_gate",
round=1,
author_slug="be-pr-reviewer",
findings=[
Finding(
severity=Severity.MAJOR,
expected="all checks green",
actual="lint failing on health.py",
)
],
)
finding_id8 = str(rows[0].id)[:8]
c = _choreographer(db_session, task)
blocked = await c.submit_up(cell_pm.id, task.id, "bubbling up the cell's work")
body = blocked.as_dict()
assert body["error"] == "tracing_gap", body
assert f"finding:{finding_id8}" in body["missing"]
ok = await c.submit_up(
cell_pm.id,
task.id,
"bubbling up the cell's work now that lint is fixed",
resolved_findings=[
{"finding_id": finding_id8, "commit": "abc123", "note": "fixed lint"}
],
)
assert ok.error is None, ok.as_dict()
assert ok.status == Status.AWAITING_PR_REVIEW.value
addressed = await repo.list_for_task(task.id, status=STATUS_ADDRESSED)
assert len(addressed) == 1
@pytest.mark.asyncio
async def test_non_owner_resolved_findings_never_mutate_the_ledger(
db_session: AsyncSession, setup: dict[str, Any]
) -> None:
"""A stale non-owner PM's resolved_findings must not be applied: the
ownership rejection commits the session, so a pre-guard apply would
persist — zeroing the FINDINGS_ADDRESSED gate for the real owner."""
cell_pm = setup["cell_pm_agent"]
main_pm = setup["main_pm_agent"]
task: Any = TaskTable(
id=uuid4(),
title="Assemble the healthz cell work",
description="Bubble the cell's healthz work up to the Main PM",
status=TaskStatus.IN_PROGRESS,
priority=2,
task_type=TaskType.CODE,
nature=TaskNature.TECHNICAL,
team=Team.BACKEND,
project_id=setup["project"].id,
created_by=setup["system_agent"].id,
assigned_to=cell_pm.id,
claimed_by=cell_pm.id,
branch_name="feature/backend/healthz-cell",
acceptance_criteria=["Returns 200"],
)
db_session.add(task)
await db_session.flush()
repo = ReviewFindingsRepository(db_session)
rows = await repo.insert_many(
task_id=task.id,
origin="pr_gate",
round=1,
author_slug="be-pr-reviewer",
findings=[
Finding(
severity=Severity.MAJOR,
expected="all checks green",
actual="lint failing on health.py",
)
],
)
finding_id8 = str(rows[0].id)[:8]
c = _choreographer(db_session, task)
rejected = await c.submit_up(
main_pm.id,
task.id,
"sneaky resubmit from a PM that does not own this task",
resolved_findings=[
{"finding_id": finding_id8, "commit": "abc123", "note": "not mine"}
],
)
assert rejected.error == "not_authorized", rejected.as_dict()
still_open = await repo.list_for_task(task.id, status=STATUS_OPEN)
assert len(still_open) == 1, "non-owner resolutions must not touch the ledger"
@pytest.mark.asyncio
async def test_submit_root_blocked_by_open_finding_then_resolved(
db_session: AsyncSession, setup: dict[str, Any]
) -> None:
"""submit_root's root analogue of the submit_up gap above."""
main_pm = setup["main_pm_agent"]
task: Any = TaskTable(
id=uuid4(),
title="Assemble the healthz root work",
description="Open the root->master PR for the healthz rollout",
status=TaskStatus.IN_PROGRESS,
priority=2,
task_type=TaskType.PLANNING,
nature=TaskNature.TECHNICAL,
team=Team.MAIN_PM,
project_id=setup["project"].id,
created_by=setup["system_agent"].id,
assigned_to=main_pm.id,
claimed_by=main_pm.id,
branch_name="feature/main_pm/healthz-root",
acceptance_criteria=["Returns 200"],
)
db_session.add(task)
await db_session.flush()
repo = ReviewFindingsRepository(db_session)
rows = await repo.insert_many(
task_id=task.id,
origin="pm",
round=1,
author_slug="main-pm",
findings=[
Finding(
severity=Severity.MAJOR,
expected="scope limited to the endpoint change",
actual="CLAUDE.md modified out of scope",
)
],
)
finding_id8 = str(rows[0].id)[:8]
c = _choreographer(db_session, task)
blocked = await c.submit_root(main_pm.id, task.id, "opening the root->master PR")
body = blocked.as_dict()
assert body["error"] == "tracing_gap", body
assert f"finding:{finding_id8}" in body["missing"]
ok = await c.submit_root(
main_pm.id,
task.id,
"opening the root->master PR now that scope is reverted",
resolved_findings=[
{"finding_id": finding_id8, "commit": "def456", "note": "reverted"}
],
)
assert ok.error is None, ok.as_dict()
assert ok.status == Status.AWAITING_PR_REVIEW.value
addressed = await repo.list_for_task(task.id, status=STATUS_ADDRESSED)
assert len(addressed) == 1
# ---------------------------------------------------------------------------
# complete() (PM merge) stamps pm-origin addressed findings verified
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_cell_pm_complete_stamps_pm_origin_addressed_findings_verified(
db_session: AsyncSession, setup: dict[str, Any]
) -> None:
"""Live gap: pr_pass's stamp_addressed_verified(origin='pr_gate') was the
ONLY caller of the addressed->verified lifecycle other than i_am_done — a
pm-origin finding (request_changes) never got confirmed anywhere. The PM's
own merge (complete) IS that confirmation."""
cell_pm = setup["cell_pm_agent"]
task: Any = TaskTable(
id=uuid4(),
title="Cell task ready to merge",
description="Merge the cell->root PR",
status=TaskStatus.AWAITING_PM_REVIEW,
priority=2,
task_type=TaskType.CODE,
nature=TaskNature.TECHNICAL,
team=Team.BACKEND,
project_id=setup["project"].id,
created_by=setup["system_agent"].id,
assigned_to=cell_pm.id,
pr_number=_PR_NUMBER,
branch_name=_BRANCH,
acceptance_criteria=["Returns 200"],
)
db_session.add(task)
await db_session.flush()
repo = ReviewFindingsRepository(db_session)
rows = await repo.insert_many(
task_id=task.id,
origin="pm",
round=1,
author_slug="be-pm",
findings=[
Finding(
severity=Severity.MAJOR,
expected="scope limited to the endpoint change",
actual="CLAUDE.md modified out of scope",
)
],
)
await repo.mark_addressed(task.id, str(rows[0].id), commit="abc", note="reverted")
c = _choreographer(db_session, task)
env = await c.cell_pm_complete(
cell_pm.id, task.id, "reviewed and approved; scope reverted"
)
assert env.error is None, env.as_dict()
verified = await repo.list_for_task(task.id, status=STATUS_VERIFIED)
assert len(verified) == 1
assert verified[0].id == rows[0].id
# ---------------------------------------------------------------------------
# ceo_approve stamps ceo-origin addressed findings verified
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_ceo_approve_stamps_ceo_origin_addressed_findings_verified(
db_session: AsyncSession, setup: dict[str, Any]
) -> None:
task = setup["task"]
task.status = TaskStatus.AWAITING_CEO_APPROVAL
await db_session.flush()
repo = ReviewFindingsRepository(db_session)
rows = await repo.insert_many(
task_id=task.id,
origin="ceo",
round=1,
author_slug="ceo",
findings=[
Finding(
severity=Severity.BLOCKER,
expected="CEO sign-off on this task",
actual="the timestamp format did not match the API contract",
)
],
)
await repo.mark_addressed(task.id, str(rows[0].id), commit="abc", note="fixed")
task_service = TaskService(db_session)
result = await task_service.ceo_approve(task.id, "looks good")
assert result is not None
assert result.status == TaskStatus.COMPLETED
verified = await repo.list_for_task(task.id, status=STATUS_VERIFIED)
assert len(verified) == 1
assert verified[0].id == rows[0].id
# ---------------------------------------------------------------------------
# ceo_reject — reason validation (never a 500) + coordination-root rounds
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
@pytest.mark.parametrize("bad_reason", ["", "wip", "n/a", "-", " "])
async def test_ceo_reject_rejects_trivial_reason_cleanly(
db_session: AsyncSession, setup: dict[str, Any], bad_reason: str
) -> None:
"""Live repro: an empty/'wip'/'n/a'/'-'/whitespace reason built a Finding
directly and let pydantic's uncaught ValidationError surface as a 500.
The reason must be validated BEFORE the Finding is constructed."""
task = setup["task"]
task.status = TaskStatus.AWAITING_CEO_APPROVAL
await db_session.flush()
task_service = TaskService(db_session)
with pytest.raises(ValidationError):
await task_service.ceo_reject(task.id, bad_reason)
assert await ReviewFindingsRepository(db_session).list_for_task(task.id) == []
refreshed = await task_service.get(task.id)
assert refreshed is not None
assert refreshed.status == TaskStatus.AWAITING_CEO_APPROVAL
async def _seed_main_pm_and_ceo_fixed_ids(
db_session: AsyncSession,
) -> tuple[UUID, UUID]:
"""Ensure the fixed-UUID main-pm/ceo agents exist — FK targets for the
coordination-root reject path's reassignment + CEO-attributed audit row."""
main_pm_id = UUID(AGENT_UUIDS["main-pm"])
if await db_session.get(AgentTable, main_pm_id) is None:
db_session.add(
AgentTable(
id=main_pm_id,
name="Main PM",
slug="main-pm",
role=AgentRole.MAIN_PM,
team=Team.MAIN_PM,
status=AgentStatus.ACTIVE,
model_config={},
system_prompt="pm",
capabilities=[],
permissions={},
metrics={},
)
)
ceo_id = UUID(AGENT_UUIDS["ceo"])
if await db_session.get(AgentTable, ceo_id) is None:
db_session.add(
AgentTable(
id=ceo_id,
name="CEO",
slug="ceo",
role=AgentRole.CEO,
team=Team.MAIN_PM,
status=AgentStatus.ACTIVE,
model_config={},
system_prompt="ceo",
capabilities=[],
permissions={},
metrics={},
)
)
await db_session.flush()
return main_pm_id, ceo_id
@pytest.mark.asyncio
async def test_ceo_reject_coordination_root_bumps_round_and_audit_each_time(
db_session: AsyncSession, setup: dict[str, Any]
) -> None:
"""Live repro: two consecutive coordination-root rejects both wrote ledger
round=1 — the PENDING-routing path never bumped revision_count (only a
to_status==NEEDS_REVISION transition did) and never emitted the NAMED
task.ceo_reject audit event the rework scorecard attributes rejections by."""
await _seed_main_pm_and_ceo_fixed_ids(db_session)
task: Any = TaskTable(
id=uuid4(),
title="Ship the healthz feature across both repos",
description="Coordinate the healthz rollout across the app + engine repos",
status=TaskStatus.AWAITING_CEO_APPROVAL,
priority=2,
task_type=TaskType.PLANNING,
nature=TaskNature.TECHNICAL,
team=Team.MAIN_PM,
project_id=None,
product_id=None,
batch_id=uuid4(),
parent_task_id=None,
created_by=setup["system_agent"].id,
acceptance_criteria=["Both repos ship the endpoint"],
)
db_session.add(task)
await db_session.flush()
task_service = TaskService(db_session)
first = await task_service.ceo_reject(task.id, "The rollout plan misses repo B.")
assert first is not None
assert first.status == TaskStatus.PENDING
assert first.revision_count == 1
findings = await ReviewFindingsRepository(db_session).list_for_task(task.id)
assert [f.round for f in findings] == [1]
# Re-escalate for a second round (mirrors the Main PM re-submitting and
# the CEO rejecting again).
task.status = TaskStatus.AWAITING_CEO_APPROVAL
await db_session.flush()
second = await task_service.ceo_reject(task.id, "Still missing repo B's rollout.")
assert second is not None
assert second.revision_count == _EXPECTED_ROUND_COUNT
findings = await ReviewFindingsRepository(db_session).list_for_task(task.id)
assert sorted(f.round for f in findings) == [1, _EXPECTED_ROUND_COUNT]
ceo_reject_rows = (
(
await db_session.execute(
select(AuditLogTable).where(
AuditLogTable.target_id == task.id,
AuditLogTable.event_type == "task.ceo_reject",
)
)
)
.scalars()
.all()
)
assert len(ceo_reject_rows) == _EXPECTED_ROUND_COUNT
Reference in New Issue View Git Blame Copy Permalink