Feature/observability gateway health (#247)

* feat(observability): revision_count + audit_log query index (migration 045)

Adds tasks.revision_count (the O(1) rework counter — forward-only, existing
rows default 0) and the composite index audit_log(target_id, event_type,
timestamp) that powers the cycle-time and rework reconstruction queries.
Verified the real upgrade/downgrade/upgrade chain on a throwaway pgvector PG.
First task of the 0.10.0 observability dashboards.

* feat(observability): count reworks + attribute qa_fail/pr_fail to the rejector

Every transition into needs_revision increments tasks.revision_count at the
single audit chokepoint (exactly once per bounce, across all paths incl. pr_fail
and ceo_reject), so the rework rate is an O(1) read. A QA or PR-review bounce
also emits a named task.qa_fail / task.pr_fail audit event carrying the
rejector's agent_id, so the per-agent rework scorecard charges the rejection to
the reviewer who made it, not the developer who owns the task.

* feat(observability): cycle-time, bottleneck, rework, and scorecard metrics

MetricsService gains four read methods on the audit_log + tasks data: per-stage
cycle time reconstructed from the transition journey (excluding the named
qa_fail/pr_fail events), bottleneck distribution (cumulative dwell + live parked
counts), rework rate (overall/by-team/by-agent with rejector attribution + cost
via spawn-session task_id), and a fused per-agent/per-cell scorecard. Dataclass
models with to_dict(). Verified against a real Postgres journey.

* feat(observability): cycle-time/bottleneck/rework/scorecard read endpoints

Thin read-only routes on the dashboard router delegating to MetricsService:
/metrics/cycle-time, /metrics/bottlenecks, /metrics/rework, and
/metrics/scorecard/{agent,team}. 404 when an agent scorecard target is absent.
5 route tests (200 + shape + the agent-404 case).

* feat(panel): Delivery observability tab (cycle-time, bottlenecks, rework, scorecards)

A third Metrics tab built on the observability endpoints: a per-stage
cycle-time bar chart, a bottleneck panel (worst stage + cumulative dwell +
live parked counts), a rework panel (rate + by-team + by-agent attribution +
cost), and per-cell scorecards. Reuses Recharts + Card/Badge/Skeleton and the
React-Query hook pattern; observabilityApi mirrors usageApi with mock-mode
fallbacks. tsc + eslint clean; 113 panel tests pass.

* docs(observability): changelog + CLAUDE.md for the delivery dashboards

* feat(gateway-health): recover a broken-but-alive agent instead of protecting it

The verb-heartbeat cannot tell a quiet-healthy agent from one whose MCP gateway
is broken (a corrupted /app/.venv firing no verb) yet whose container is up — the
reaper's live-skip would shield it forever. The reaper now probes the gateway
out-of-band (docker exec: does the gateway venv import its deps?) and, once it
has been broken past gateway_health_grace_seconds (tolerating a transient probe
miss), kills + evicts the container so it falls through to release + respawn.
Probe-inconclusive or healthy spares the container. Gated by
gateway_health_enabled (default-on reliability fix; in the panel Feature Flags).
Defers the optional agent-side self-check + full registry re-adoption — the
reaper's docker-liveness fallback already recovers a broken-after-restart agent.

* docs(gateway-health): changelog + CLAUDE.md for broken-but-alive recovery

* docs(observability): user-facing docs for the Delivery dashboards + gateway-health

Documents the new Metrics -> Delivery tab (cycle-time, bottlenecks, rework with
rejector attribution, cell scorecards) in the panel guide and the operations
health-and-metrics guide, and adds the gateway-health env vars + an agent-gateway
recovery note. Published MkDocs site only; settings.md's default-off flag table
intentionally omits the default-on gateway-health flag (same as overload-break).

* chore(release): cut 0.10.0 (changelog section + version refs)

* fix(gateway): exempt PM coordinators from single-task claim guards

A Main/Cell PM plans and delegates many root tasks in parallel; the work
then runs in the delegated cells, not in the PM's own hands. But the
claim-time concurrency guards meant for developers — already_active and
paused (the latter firing after i_am_idle auto-pauses the PM's own
umbrella) — were applied to the PM too, so once it held one root it could
never plan a second: it thrashed between its claimed roots and respawned
forever, burning tokens for zero progress.

_run_claim_guards now skips already_active/paused for the coordinator PM
roles (_COORDINATOR_ROLES = {main_pm, cell_pm}); only unmet_dependency — a
real upstream sequence constraint, which parks the root back to pending —
still gates a PM. paused_tasks_guard also excludes the target task itself,
so a PM re-entering its own paused umbrella never self-blocks.

Tests: a coordinator plans a second root with one in_progress + one paused
sibling (full path + claimed-recovery path), the paused target exclusion,
and the developer guards still fire. Repurposed the pre-fix test that
asserted the now-removed PM block.

* fix(metrics): coerce SQL avg/extract hours aggregates to float (panel toFixed crash)

EXTRACT(epoch ...) returns numeric on PostgreSQL 14+, which asyncpg surfaces
as a Decimal; a Decimal serializes to a quoted JSON string, so the panel's
avg_cycle_hours.toFixed(1) (and the other hours fields) threw 'toFixed is not
a function' and blanked the Delivery tab.

A single _as_hours helper now rounds every SQL-averaged hours field to a real
float — avg_cycle_hours on the new scorecards plus the pre-existing
avg_completion_hours / avg_blocked_hours / longest_blocked_hours. Token and
cost fields were already float()-cast and are unaffected.

Regression test asserts _as_hours coerces Decimal -> float and preserves the
None/zero behavior.

* feat(panel): edit a task's sequence from the details page

A task's sequence (order within siblings, lower runs first) was display-only
with no way to change it from the UI, and TaskUpdate didn't carry the field
so PATCH couldn't set it either. The details page's Dependencies tab now has
an inline sequence editor mirroring the parent / dependency editors, and
PATCH /tasks/{id} accepts a sequence field (owner or privileged role) through
the existing generic update path.

* fix(mypy): green the full make-quality type gate

make quality runs 'mypy roboco/ tests/', which the per-module checks on the
0.10.0 branch never exercised. Two issues surfaced:

- The coordinator-exemption change added role_str to
  Choreographer._run_claim_guards but not to the ChoreographerHelpers
  protocol base, so the composed Choreographer had incompatible base-class
  signatures. Sync the protocol signature.

- The gateway-health / stale-reaper tests stubbed methods by direct
  assignment (orch._m = AsyncMock()) and typed their duck-typed task doubles
  as object, tripping method-assign / assignment / attr-defined. Switch to
  monkeypatch.setattr (keeping a local mock ref for the assertions) and type
  the doubles as Any — no type: ignore.

Full mypy roboco/ tests/ clean (785 files); the 21 runtime tests pass.

* fix(metrics): static cycle-time SQL — clear bandit B608 (CI gate)

The cycle-time query interpolated an optional team clause into the text() SQL
via an f-string, which bandit flags as B608 (hardcoded SQL) and turned the
merge gate red. The team value was always a bound parameter, so it was a false
positive — but the f-string is the trigger. Rebuilt as one static query with
(CAST(:team AS text) IS NULL OR a.details->>'team' = :team) and an always-bound
team param (CAST, not ::text — SQLAlchemy's :param parser collides with
PostgreSQL's :: cast operator, which broke the query as a stray param).

Full make quality green vs a real pgvector PG (all 21 gate steps).

---------

Co-authored-by: Renn F <rennf93@users.noreply.github.com>
This commit is contained in:
Renzo F
2026-06-23 07:26:41 +02:00
committed by GitHub
co-authored by Renn F
parent 07008baaf5
commit c09cf80b40
42 changed files with 2419 additions and 63 deletions
+17
View File
@@ -255,6 +255,23 @@ def test_transform_update_data_handles_null_unassign() -> None:
assert out["assigned_to"] is None
def test_transform_update_data_passes_sequence() -> None:
"""sequence is editable via PATCH (panel task-details sequence editor)."""
new_order = 3
update = TaskUpdate(sequence=new_order)
out = transform_update_data(update)
assert out["sequence"] == new_order
# Unset sequence is omitted (exclude_unset), so a partial PATCH never
# clobbers the existing order.
assert "sequence" not in transform_update_data(TaskUpdate(priority=1))
def test_task_update_sequence_rejects_negative() -> None:
"""sequence has ge=0 — a negative order is a validation error, not stored."""
with pytest.raises(ValueError, match="sequence"):
TaskUpdate(sequence=-1)
# ---------------------------------------------------------------------------
# task_to_response / task_list_to_response
# ---------------------------------------------------------------------------
@@ -19,6 +19,7 @@ from uuid import uuid4
import pytest
from roboco.services.gateway.choreographer import Choreographer, ChoreographerDeps
from roboco.services.gateway.claim_guards import paused_tasks_guard
# #172: a developer fresh claim must carry a substantive step checklist.
# Inert on re-entry/error/non-dev paths, so safe to pass everywhere.
@@ -559,3 +560,137 @@ async def test_claim_doc_task_blocks_when_documenter_has_paused_task() -> None:
assert body["error"] == "invalid_state"
assert "resume" in body["remediate"].lower()
task_svc.doc_claim.assert_not_awaited()
# ---------------------------------------------------------------------------
# Coordinator exemption — a PM plans + delegates many roots in parallel, so the
# single-active-task guards (already_active / paused) must NOT gate it; only a
# real upstream sequence dependency may hold a PM's root back. (Developers stay
# blocked — see the A.2 / A.3 tests above.)
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_main_pm_can_plan_second_root_despite_active_and_paused() -> None:
"""A main_pm may plan a new root while it already holds one in_progress and
one paused root — the developer concurrency guards do not gate a coordinator.
"""
pm_id = uuid4()
task_id = uuid4()
target = MagicMock(
id=task_id,
status="pending",
plan=None,
assigned_to=None,
parent_task_id=None,
sequence=0,
task_type="code",
team="main_pm",
)
claimed = MagicMock(
id=task_id, status="claimed", plan=None, assigned_to=pm_id, task_type="code"
)
started = MagicMock(
id=task_id,
status="in_progress",
plan={"text": "x"},
assigned_to=pm_id,
task_type="code",
)
# The coordinator already holds one in_progress root and one paused root —
# both would trip the guards for a non-PM caller.
other_active = MagicMock(id=uuid4(), status="in_progress")
other_paused = MagicMock(id=uuid4(), status="paused")
task_svc = _task_svc_with(
target,
role="main_pm",
agent_id=pm_id,
lookups={"in_progress": [other_active], "paused": [other_paused]},
)
task_svc.claim.return_value = claimed
task_svc.set_plan.return_value = claimed
task_svc.start.return_value = started
deps = _make_deps(task=task_svc)
c = Choreographer(deps)
env = await c.i_will_plan(
pm_id,
task_id,
plan="route to backend + frontend cells",
rich_plan={
"approach": (
"Route this root to the backend and frontend cells in parallel: "
"be-pm owns the API contract, fe-pm consumes it. No cross-cell "
"dependency for this slice, so both cells start at once."
),
"sub_tasks": [
{
"title": "Backend slice",
"description": (
"be-pm decomposes the API change and assigns be-dev-1, "
"who implements with tests and opens the leaf PR for QA."
),
}
],
},
)
assert env.error is None, env.as_dict()
task_svc.start.assert_awaited()
@pytest.mark.asyncio
async def test_main_pm_recovers_claimed_root_with_paused_sibling() -> None:
"""The live deadlock: a respawned main_pm re-enters i_will_plan on a stuck
`claimed` root while another root is paused (i_am_idle auto-paused it). The
paused guard must NOT block the coordinator's recovery — set_plan + start
runs and the root reaches in_progress.
"""
pm_id = uuid4()
task_id = uuid4()
claimed = MagicMock(
id=task_id,
status="claimed",
plan=None,
assigned_to=pm_id,
parent_task_id=None,
sequence=0,
task_type="code",
team="main_pm",
branch_name="feature/main_pm/abc",
)
started = MagicMock(
id=task_id,
status="in_progress",
plan={"text": "x"},
assigned_to=pm_id,
task_type="code",
)
other_paused = MagicMock(id=uuid4(), status="paused")
task_svc = _task_svc_with(
claimed, role="main_pm", agent_id=pm_id, lookups={"paused": [other_paused]}
)
task_svc.set_plan.return_value = started
task_svc.start.return_value = started
deps = _make_deps(task=task_svc)
c = Choreographer(deps)
env = await c.i_will_plan(
pm_id, task_id, plan="route this root to the backend + frontend cells"
)
assert env.error is None, env.as_dict()
task_svc.start.assert_awaited_once_with(task_id, pm_id)
def test_paused_tasks_guard_excludes_target() -> None:
"""A paused task that IS the claim target must not self-block, mirroring
already_active_guard's target exclusion (the 2026-06-14 self-deadlock)."""
target_id = uuid4()
other_id = uuid4()
# Only the target itself is paused -> no block.
assert paused_tasks_guard([MagicMock(id=target_id)], target_id) is None
# A different paused task -> block, naming that task.
env = paused_tasks_guard([MagicMock(id=other_id)], target_id)
assert env is not None
assert str(other_id) in env.as_dict()["remediate"]
# Back-compat: with no target supplied, any paused task blocks.
assert paused_tasks_guard([MagicMock(id=other_id)]) is not None
@@ -362,10 +362,12 @@ async def test_i_will_work_on_in_progress_assigned_to_self_idempotent() -> None:
@pytest.mark.asyncio
async def test_i_will_plan_pm_with_already_active_task_rejects() -> None:
"""The already_active_guard still fires on i_will_plan even though
pm_cannot_execute_code is skipped. Covers _impl.py:1106-1108
(with-briefing wrap of the guard rejection).
async def test_i_will_plan_pm_exempt_from_already_active_guard() -> None:
"""A PM coordinator is exempt from already_active_guard on i_will_plan: it
plans + delegates many roots in parallel, so holding one in_progress root
must NOT block planning another. (The guard still fires for developers see
test_choreographer_claim_guards.py.) Repurposed from the pre-fix test that
asserted the now-removed PM block.
"""
pm_id = uuid4()
task_id = uuid4()
@@ -380,12 +382,25 @@ async def test_i_will_plan_pm_with_already_active_task_rejects() -> None:
parent_task_id=None,
task_type="planning",
)
started = MagicMock(
id=task_id,
status="in_progress",
plan={"text": "x"},
assigned_to=pm_id,
title="t",
team="backend",
task_type="planning",
)
busy_task = MagicMock(id=other_task_id, status="in_progress")
task_svc = AsyncMock()
task_svc.get.return_value = target
task_svc.agent_for.return_value = MagicMock(role="cell_pm", team="backend")
task_svc.list_in_progress_for_agent.return_value = [busy_task]
task_svc.list_paused_for_agent.return_value = []
task_svc.get_subtasks.return_value = []
task_svc.claim.return_value = target
task_svc.set_plan.return_value = target
task_svc.start.return_value = started
deps = _make_deps(task=task_svc)
c = Choreographer(deps)
env = await c.i_will_plan(
@@ -412,8 +427,8 @@ async def test_i_will_plan_pm_with_already_active_task_rejects() -> None:
},
)
body = env.as_dict()
assert body["error"] == "invalid_state"
assert "in_progress task" in body["message"]
assert body.get("error") is None, body
task_svc.start.assert_awaited()
@pytest.mark.asyncio
+195
View File
@@ -0,0 +1,195 @@
"""Gateway-health recovery: probe a broken-but-alive agent + reap it past grace.
The verb-heartbeat can't tell a quiet-healthy agent from one whose MCP gateway
is broken (corrupted /app/.venv) yet whose container is up. The reaper now probes
out-of-band and, past a grace window, kills + evicts the broken container so it
falls through to release + respawn instead of being protected forever.
"""
from __future__ import annotations
import asyncio
from datetime import UTC, datetime, timedelta
from typing import Any, cast
from unittest.mock import AsyncMock
from uuid import uuid4
import pytest
from roboco.config import settings
from roboco.runtime.orchestrator import AgentOrchestrator
from roboco.services.settings import FEATURE_FLAGS
class _FakeProc:
def __init__(self, rc: int) -> None:
self._rc = rc
async def wait(self) -> int:
return self._rc
def _orch(monkeypatch: pytest.MonkeyPatch) -> AgentOrchestrator:
orch = AgentOrchestrator.__new__(AgentOrchestrator) # bypass __init__
orch._instances = {}
orch._gateway_broken_since = {}
monkeypatch.setattr(orch, "_resolve_agent_slug", lambda _owner: "be-dev-1")
return orch
# ─── probe ──────────────────────────────────────────────────────────────────
@pytest.mark.asyncio
async def test_probe_healthy(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(
asyncio, "create_subprocess_exec", AsyncMock(return_value=_FakeProc(0))
)
assert await AgentOrchestrator._probe_gateway_health("be-dev-1") is True
@pytest.mark.asyncio
async def test_probe_broken(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(
asyncio, "create_subprocess_exec", AsyncMock(return_value=_FakeProc(1))
)
assert await AgentOrchestrator._probe_gateway_health("be-dev-1") is False
@pytest.mark.asyncio
async def test_probe_infra_error_is_none(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(
asyncio, "create_subprocess_exec", AsyncMock(side_effect=OSError("no docker"))
)
assert await AgentOrchestrator._probe_gateway_health("be-dev-1") is None
# ─── recovery decision ──────────────────────────────────────────────────────
def _task() -> Any:
return type("T", (), {"id": uuid4(), "assigned_to": uuid4(), "claimed_by": None})()
@pytest.mark.asyncio
async def test_disabled_never_recovers(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(settings, "gateway_health_enabled", False)
orch = _orch(monkeypatch)
remove = AsyncMock()
monkeypatch.setattr(orch, "_remove_container", remove)
monkeypatch.setattr(orch, "_probe_gateway_health", AsyncMock(return_value=False))
assert await orch._maybe_recover_broken_gateway(_task()) is False
remove.assert_not_awaited()
@pytest.mark.asyncio
async def test_healthy_is_spared(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(settings, "gateway_health_enabled", True)
orch = _orch(monkeypatch)
orch._gateway_broken_since["be-dev-1"] = datetime.now(UTC) # stale mark
remove = AsyncMock()
monkeypatch.setattr(orch, "_remove_container", remove)
monkeypatch.setattr(orch, "_probe_gateway_health", AsyncMock(return_value=True))
assert await orch._maybe_recover_broken_gateway(_task()) is False
remove.assert_not_awaited()
assert "be-dev-1" not in orch._gateway_broken_since # mark cleared
@pytest.mark.asyncio
async def test_first_broken_sighting_waits_for_grace(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(settings, "gateway_health_enabled", True)
orch = _orch(monkeypatch)
remove = AsyncMock()
monkeypatch.setattr(orch, "_remove_container", remove)
monkeypatch.setattr(orch, "_probe_gateway_health", AsyncMock(return_value=False))
assert await orch._maybe_recover_broken_gateway(_task()) is False
remove.assert_not_awaited()
assert "be-dev-1" in orch._gateway_broken_since # grace mark recorded
@pytest.mark.asyncio
async def test_broken_past_grace_is_killed(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(settings, "gateway_health_enabled", True)
orch = _orch(monkeypatch)
# _gateway_health_grace is a test-only injection read via getattr(..., None);
# 0 means "past grace immediately".
monkeypatch.setattr(orch, "_gateway_health_grace", 0, raising=False)
orch._gateway_broken_since["be-dev-1"] = datetime.now(UTC) - timedelta(seconds=5)
orch._instances["be-dev-1"] = cast("Any", object())
remove = AsyncMock()
monkeypatch.setattr(orch, "_remove_container", remove)
monkeypatch.setattr(orch, "_probe_gateway_health", AsyncMock(return_value=False))
assert await orch._maybe_recover_broken_gateway(_task()) is True
remove.assert_awaited_once_with("roboco-agent-be-dev-1")
assert "be-dev-1" not in orch._instances # evicted
@pytest.mark.asyncio
async def test_inconclusive_probe_is_spared(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(settings, "gateway_health_enabled", True)
orch = _orch(monkeypatch)
remove = AsyncMock()
monkeypatch.setattr(orch, "_remove_container", remove)
monkeypatch.setattr(orch, "_probe_gateway_health", AsyncMock(return_value=None))
assert await orch._maybe_recover_broken_gateway(_task()) is False
remove.assert_not_awaited()
# ─── reaper wiring ──────────────────────────────────────────────────────────
def _stale_task() -> Any:
return type(
"T",
(),
{
"id": uuid4(),
"last_heartbeat_at": datetime.now(UTC) - timedelta(seconds=600),
},
)()
@pytest.mark.asyncio
async def test_reaper_reaps_broken_gateway_agent(
monkeypatch: pytest.MonkeyPatch,
) -> None:
orch = AgentOrchestrator.__new__(AgentOrchestrator)
orch._claim_heartbeat_ttl = 300
monkeypatch.setattr(orch, "_assignee_has_active_instance", lambda _t: True)
monkeypatch.setattr(orch, "_maybe_kill_wedged_grok", AsyncMock(return_value=False))
monkeypatch.setattr(
orch, "_maybe_recover_broken_gateway", AsyncMock(return_value=True)
)
task = _stale_task()
svc = AsyncMock()
svc.list_in_progress_or_claimed.return_value = [task]
svc.unclaim_for_reaper = AsyncMock()
await orch._reap_with_service(svc)
svc.unclaim_for_reaper.assert_awaited_once_with(task.id)
@pytest.mark.asyncio
async def test_reaper_spares_healthy_live_agent(
monkeypatch: pytest.MonkeyPatch,
) -> None:
orch = AgentOrchestrator.__new__(AgentOrchestrator)
orch._claim_heartbeat_ttl = 300
monkeypatch.setattr(orch, "_assignee_has_active_instance", lambda _t: True)
monkeypatch.setattr(orch, "_maybe_kill_wedged_grok", AsyncMock(return_value=False))
monkeypatch.setattr(
orch, "_maybe_recover_broken_gateway", AsyncMock(return_value=False)
)
svc = AsyncMock()
svc.list_in_progress_or_claimed.return_value = [_stale_task()]
svc.unclaim_for_reaper = AsyncMock()
await orch._reap_with_service(svc)
svc.unclaim_for_reaper.assert_not_awaited()
# ─── config flag ────────────────────────────────────────────────────────────
def test_flag_defaults_on_and_is_registered() -> None:
assert settings.gateway_health_enabled is True
assert "gateway_health_enabled" in {key for key, _ in FEATURE_FLAGS}
+45 -5
View File
@@ -24,7 +24,9 @@ from roboco.seeds.initial_data import AGENT_UUIDS
@pytest.mark.asyncio
async def test_reap_stale_claims_releases_dead_holders() -> None:
async def test_reap_stale_claims_releases_dead_holders(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""A task past TTL is unclaimed; a fresh one is left alone."""
stale_id = uuid4()
fresh_id = uuid4()
@@ -41,6 +43,9 @@ async def test_reap_stale_claims_releases_dead_holders() -> None:
)()
orch = AgentOrchestrator.__new__(AgentOrchestrator) # bypass __init__
monkeypatch.setattr(
orch, "_maybe_recover_broken_gateway", AsyncMock(return_value=False)
)
orch._claim_heartbeat_ttl = 300
svc = AsyncMock()
svc.list_in_progress_or_claimed.return_value = [stale_task, fresh_task]
@@ -52,12 +57,17 @@ async def test_reap_stale_claims_releases_dead_holders() -> None:
@pytest.mark.asyncio
async def test_reap_stale_claims_releases_holders_with_null_heartbeat() -> None:
async def test_reap_stale_claims_releases_holders_with_null_heartbeat(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""A claimed task that never heartbeated (NULL column) is treated as stale."""
null_id = uuid4()
null_task = type("T", (), {"id": null_id, "last_heartbeat_at": None})()
orch = AgentOrchestrator.__new__(AgentOrchestrator)
monkeypatch.setattr(
orch, "_maybe_recover_broken_gateway", AsyncMock(return_value=False)
)
orch._claim_heartbeat_ttl = 300
svc = AsyncMock()
svc.list_in_progress_or_claimed.return_value = [null_task]
@@ -69,7 +79,9 @@ async def test_reap_stale_claims_releases_holders_with_null_heartbeat() -> None:
@pytest.mark.asyncio
async def test_reap_stale_claims_swallows_unclaim_errors() -> None:
async def test_reap_stale_claims_swallows_unclaim_errors(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""An unclaim_for_reaper failure must not abort the reap loop."""
stale_a = uuid4()
stale_b = uuid4()
@@ -82,6 +94,9 @@ async def test_reap_stale_claims_swallows_unclaim_errors() -> None:
)()
orch = AgentOrchestrator.__new__(AgentOrchestrator)
monkeypatch.setattr(
orch, "_maybe_recover_broken_gateway", AsyncMock(return_value=False)
)
orch._claim_heartbeat_ttl = 300
svc = AsyncMock()
svc.list_in_progress_or_claimed.return_value = [task_a, task_b]
@@ -95,7 +110,9 @@ async def test_reap_stale_claims_swallows_unclaim_errors() -> None:
@pytest.mark.asyncio
async def test_reap_spares_claims_whose_assignee_container_is_alive() -> None:
async def test_reap_spares_claims_whose_assignee_container_is_alive(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""A stale-heartbeat task is NOT reaped while its assignee container lives.
A developer deep in a long edit/test cycle outruns the heartbeat TTL; the
@@ -128,6 +145,9 @@ async def test_reap_spares_claims_whose_assignee_container_is_alive() -> None:
)()
orch = AgentOrchestrator.__new__(AgentOrchestrator)
monkeypatch.setattr(
orch, "_maybe_recover_broken_gateway", AsyncMock(return_value=False)
)
orch._claim_heartbeat_ttl = 300
orch._instances = {
"be-dev-1": AgentInstance(agent_id="be-dev-1", state=AgentState.ACTIVE)
@@ -172,6 +192,9 @@ async def test_reaper_kills_and_releases_wedged_grok_container(
)()
orch = AgentOrchestrator.__new__(AgentOrchestrator)
monkeypatch.setattr(
orch, "_maybe_recover_broken_gateway", AsyncMock(return_value=False)
)
orch._claim_heartbeat_ttl = 300
orch._grok_idle_kill_ttl = 900
orch._instances = {"be-dev-1": _grok_instance()}
@@ -210,6 +233,9 @@ async def test_reaper_spares_grok_container_within_kill_ttl(
)()
orch = AgentOrchestrator.__new__(AgentOrchestrator)
monkeypatch.setattr(
orch, "_maybe_recover_broken_gateway", AsyncMock(return_value=False)
)
orch._claim_heartbeat_ttl = 300
orch._grok_idle_kill_ttl = 900
orch._instances = {"be-dev-1": _grok_instance()}
@@ -249,6 +275,9 @@ async def test_reaper_never_kills_non_grok_container(
claude_cfg = type("C", (), {"provider_type": "anthropic"})()
orch = AgentOrchestrator.__new__(AgentOrchestrator)
monkeypatch.setattr(
orch, "_maybe_recover_broken_gateway", AsyncMock(return_value=False)
)
orch._claim_heartbeat_ttl = 300
orch._grok_idle_kill_ttl = 900
orch._instances = {
@@ -292,6 +321,9 @@ async def test_reap_spares_live_container_on_registry_miss(
)()
orch = AgentOrchestrator.__new__(AgentOrchestrator)
monkeypatch.setattr(
orch, "_maybe_recover_broken_gateway", AsyncMock(return_value=False)
)
orch._claim_heartbeat_ttl = 300
orch._grok_idle_kill_ttl = 900
orch._instances = {} # registry lost; container still up
@@ -326,6 +358,9 @@ async def test_reap_releases_on_registry_miss_when_container_gone(
)()
orch = AgentOrchestrator.__new__(AgentOrchestrator)
monkeypatch.setattr(
orch, "_maybe_recover_broken_gateway", AsyncMock(return_value=False)
)
orch._claim_heartbeat_ttl = 300
orch._grok_idle_kill_ttl = 900
orch._instances = {}
@@ -342,7 +377,9 @@ async def test_reap_releases_on_registry_miss_when_container_gone(
@pytest.mark.asyncio
async def test_registry_uninitialised_skips_docker_fallback() -> None:
async def test_registry_uninitialised_skips_docker_fallback(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""With `_instances` never initialised (None — the __new__ unit harness), the
Docker fallback is skipped and the stale task reaps as before; no accidental
Docker probing where there's no registry to be amnesiac about.
@@ -361,6 +398,9 @@ async def test_registry_uninitialised_skips_docker_fallback() -> None:
)()
orch = AgentOrchestrator.__new__(AgentOrchestrator)
monkeypatch.setattr(
orch, "_maybe_recover_broken_gateway", AsyncMock(return_value=False)
)
orch._claim_heartbeat_ttl = 300
# _instances intentionally NOT set -> getattr yields None -> no fallback.
svc = AsyncMock()
@@ -0,0 +1,33 @@
"""Regression: SQL avg/extract "hours" aggregates must serialize as JSON numbers.
``EXTRACT(epoch ...)`` returns ``numeric`` on PostgreSQL 14+, which asyncpg
surfaces as a ``Decimal``. A ``Decimal`` serializes to a JSON *string*, so the
panel's ``avg_cycle_hours.toFixed(1)`` (and the other hours fields) threw
``toFixed is not a function`` against the live deploy. ``_as_hours`` coerces to a
real ``float`` so the field is always a JSON number.
"""
from __future__ import annotations
from decimal import Decimal
import pytest
from roboco.services.metrics import _as_hours
def test_as_hours_coerces_decimal_to_float() -> None:
result = _as_hours(Decimal("1.21"))
assert result == pytest.approx(1.21)
assert isinstance(result, float) # not Decimal -> serializes as a JSON number
def test_as_hours_rounds_to_two_places() -> None:
assert _as_hours(Decimal("1.236")) == pytest.approx(1.24)
assert _as_hours(3.14159) == pytest.approx(3.14)
assert isinstance(_as_hours(3.14159), float)
def test_as_hours_none_and_zero_yield_none() -> None:
assert _as_hours(None) is None
assert _as_hours(0) is None
assert _as_hours(Decimal("0")) is None
@@ -0,0 +1,44 @@
"""TaskService._audit_events_for — the rejector-attributed audit event selection.
A transition always emits the generic ``task.<status>``; a reviewer bounce to
needs_revision additionally emits ``task.qa_fail`` / ``task.pr_fail`` keyed on
the acting role, so the per-agent rework scorecard can attribute the rejection.
"""
from __future__ import annotations
from roboco.services.task import TaskService
def test_generic_transition_emits_only_status_event() -> None:
assert TaskService._audit_events_for("awaiting_qa", "developer") == [
"task.awaiting_qa"
]
def test_qa_fail_adds_named_event() -> None:
assert TaskService._audit_events_for("needs_revision", "qa") == [
"task.needs_revision",
"task.qa_fail",
]
def test_pr_fail_adds_named_event() -> None:
assert TaskService._audit_events_for("needs_revision", "pr_reviewer") == [
"task.needs_revision",
"task.pr_fail",
]
def test_ceo_reject_to_needs_revision_has_no_named_event() -> None:
# A CEO rejection is a needs_revision bounce but not a QA/PR-review fail.
assert TaskService._audit_events_for("needs_revision", "ceo") == [
"task.needs_revision"
]
def test_named_event_only_on_needs_revision() -> None:
# A reviewer role on a non-needs_revision transition gets no named event.
assert TaskService._audit_events_for("awaiting_pm_review", "pr_reviewer") == [
"task.awaiting_pm_review"
]