mirror of
https://github.com/rennf93/roboco.git
synced 2026-08-03 07:23:24 +02:00
fix(orchestrator): respect tracing-gap as forward progress
_pm_respawn_should_gate counted PARENT_NOT_CLAIMED rejections as no-progress and killed PMs after 3 strikes — even when the new prompts told them to call i_will_plan first. Reset counter when last response was a tracing_gap (rule-following retry, not stuck).
This commit is contained in:
Renn F
@@ -3688,13 +3688,33 @@ Start now: evidence(task_id="{task_id}")
|
||||
|
||||
_PM_RESPAWN_MAX_UNPRODUCTIVE = 3
|
||||
|
||||
def _pm_respawn_should_gate(self, agent_slug: str, task: dict[str, Any]) -> bool:
|
||||
async def _pm_respawn_should_gate(
|
||||
self, agent_slug: str, task: dict[str, Any]
|
||||
) -> bool:
|
||||
"""Return True when the respawn should be skipped (loop detected).
|
||||
|
||||
Tracks (agent_slug, task_id) -> count of consecutive spawns where
|
||||
the task's status did not advance. When the task status changes,
|
||||
the counter resets. Once the count hits the threshold, the spawn
|
||||
is skipped and a warning logged; operators must intervene.
|
||||
|
||||
Tracing-gap reset (Task 13)
|
||||
---------------------------
|
||||
With the gateway claim-time gates installed, a rule-following PM
|
||||
will hit ``PARENT_NOT_CLAIMED`` (a ``tracing_gap`` envelope) and
|
||||
the prompt will tell it to call the prerequisite verb first.
|
||||
Each retry leaves the task status unchanged but the agent IS
|
||||
making progress through the verb chain. Counting that as a
|
||||
strike kills rule-followers.
|
||||
|
||||
Solution: before incrementing on a same-status spawn, check
|
||||
``audit_log`` for a ``gateway.rejected`` row tagged
|
||||
``reason == "tracing_gap"`` from this (agent, task) since the
|
||||
last check. If found, reset the counter — the agent followed
|
||||
the rules, not stuck.
|
||||
|
||||
Audit lookup is best-effort: any failure falls through to the
|
||||
legacy strike behavior so audit problems don't break the gate.
|
||||
"""
|
||||
task_id = task.get("id")
|
||||
if not task_id:
|
||||
@@ -3702,13 +3722,22 @@ Start now: evidence(task_id="{task_id}")
|
||||
key = (agent_slug, task_id)
|
||||
current_status = task.get("status")
|
||||
record = self._pm_respawn_tracker.get(key)
|
||||
now = datetime.now(UTC)
|
||||
if record is None or record.get("last_status") != current_status:
|
||||
self._pm_respawn_tracker[key] = {
|
||||
"count": 1,
|
||||
"last_status": current_status,
|
||||
"last_check": now,
|
||||
}
|
||||
return False
|
||||
# Same status as last spawn — could be a stuck loop OR a
|
||||
# rule-following retry. Consult audit before counting.
|
||||
if await self._pm_made_rule_following_retry(agent_slug, task_id, record):
|
||||
record["count"] = 1
|
||||
record["last_check"] = now
|
||||
return False
|
||||
record["count"] += 1
|
||||
record["last_check"] = now
|
||||
if record["count"] > self._PM_RESPAWN_MAX_UNPRODUCTIVE:
|
||||
logger.warning(
|
||||
"PM respawn loop detected — skipping spawn",
|
||||
@@ -3725,6 +3754,49 @@ Start now: evidence(task_id="{task_id}")
|
||||
return True
|
||||
return False
|
||||
|
||||
async def _pm_made_rule_following_retry(
|
||||
self,
|
||||
agent_slug: str,
|
||||
task_id: str,
|
||||
record: dict[str, Any],
|
||||
) -> bool:
|
||||
"""Did the agent emit a ``tracing_gap`` envelope since the last check?
|
||||
|
||||
Returns ``False`` for unknown slugs (defensive — the audit query
|
||||
needs an agent UUID, and we'd rather fall through to the legacy
|
||||
strike behavior than crash). Returns ``False`` if the audit
|
||||
lookup raises — observability must never block the gate.
|
||||
"""
|
||||
agent_uuid_str = AGENT_UUIDS.get(agent_slug)
|
||||
if not agent_uuid_str:
|
||||
return False
|
||||
from uuid import UUID
|
||||
|
||||
try:
|
||||
agent_uuid = UUID(agent_uuid_str)
|
||||
task_uuid = UUID(task_id)
|
||||
except (ValueError, TypeError):
|
||||
return False
|
||||
since = record.get("last_check") or datetime.now(UTC)
|
||||
|
||||
from roboco.services.audit import get_audit_service
|
||||
|
||||
audit = get_audit_service()
|
||||
try:
|
||||
return await audit.has_recent_tracing_gap(
|
||||
agent_id=agent_uuid,
|
||||
task_id=task_uuid,
|
||||
since=since,
|
||||
)
|
||||
except Exception as exc:
|
||||
logger.debug(
|
||||
"audit.has_recent_tracing_gap failed; falling back to strike count",
|
||||
agent_slug=agent_slug,
|
||||
task_id=task_id,
|
||||
error=str(exc),
|
||||
)
|
||||
return False
|
||||
|
||||
async def _handle_pm_assigned_task(
|
||||
self, task: dict[str, Any], assigned_to: str
|
||||
) -> None:
|
||||
@@ -3732,7 +3804,7 @@ Start now: evidence(task_id="{task_id}")
|
||||
agent_slug = self._resolve_agent_slug(assigned_to)
|
||||
if agent_slug not in self._PM_AGENTS or self._is_agent_active(agent_slug):
|
||||
return
|
||||
if self._pm_respawn_should_gate(agent_slug, task):
|
||||
if await self._pm_respawn_should_gate(agent_slug, task):
|
||||
return
|
||||
logger.info(
|
||||
"Spawning assigned PM agent",
|
||||
|
||||
@@ -446,6 +446,52 @@ class AuditService(SingletonService):
|
||||
# QUERY METHODS
|
||||
# =========================================================================
|
||||
|
||||
async def has_recent_tracing_gap(
|
||||
self,
|
||||
*,
|
||||
agent_id: UUID,
|
||||
task_id: UUID,
|
||||
since: datetime,
|
||||
) -> bool:
|
||||
"""Has this (agent, task) emitted a ``gateway.rejected`` ``tracing_gap``?
|
||||
|
||||
Used by the orchestrator's PM respawn circuit breaker to tell
|
||||
rule-following retries (agent hit a claim-time gate, returned a
|
||||
``tracing_gap`` envelope, and is being re-spawned to call the
|
||||
prerequisite verb) apart from genuine no-progress hangs. The
|
||||
former must reset the strike count; the latter must increment it.
|
||||
|
||||
Returns ``True`` if at least one ``audit_log`` row exists where:
|
||||
|
||||
* ``event_type == "gateway.rejected"``
|
||||
* ``agent_id`` matches
|
||||
* ``target_id`` matches the task UUID
|
||||
* ``timestamp >= since``
|
||||
* ``details->>'reason' == 'tracing_gap'``
|
||||
|
||||
Best-effort: if the underlying query raises, the caller is expected
|
||||
to fall back to the legacy strike behavior — observability must
|
||||
never block the orchestrator.
|
||||
"""
|
||||
from sqlalchemy import select
|
||||
|
||||
from roboco.db.base import get_session_factory
|
||||
from roboco.db.tables import AuditLogTable
|
||||
|
||||
session_factory = get_session_factory()
|
||||
async with session_factory() as db:
|
||||
query = (
|
||||
select(AuditLogTable.id)
|
||||
.where(AuditLogTable.event_type == "gateway.rejected")
|
||||
.where(AuditLogTable.agent_id == agent_id)
|
||||
.where(AuditLogTable.target_id == task_id)
|
||||
.where(AuditLogTable.timestamp >= since)
|
||||
.where(AuditLogTable.details["reason"].astext == "tracing_gap")
|
||||
.limit(1)
|
||||
)
|
||||
result = await db.execute(query)
|
||||
return result.scalar_one_or_none() is not None
|
||||
|
||||
async def get_recent_events(
|
||||
self,
|
||||
limit: int = 50,
|
||||
|
||||
@@ -0,0 +1,181 @@
|
||||
"""PM respawn loop guard counts no-progress only, not rule-following retries.
|
||||
|
||||
Background
|
||||
----------
|
||||
``_pm_respawn_should_gate`` is the orchestrator's circuit breaker against
|
||||
respawning the same PM on the same task forever. Before Task 13 it
|
||||
counted any spawn whose task status didn't advance as a "strike", and
|
||||
killed respawns after three of them.
|
||||
|
||||
That works for an agent that's hung — but with the gateway claim-time
|
||||
gates installed in Phase 3, a rule-following PM that hits
|
||||
``PARENT_NOT_CLAIMED`` (a ``tracing_gap`` envelope) and is told by the
|
||||
prompt to call ``i_will_plan`` first will keep returning to the same
|
||||
status. Each retry would trip a strike even though the agent did
|
||||
exactly what the gateway told it to do.
|
||||
|
||||
The fix: when audit_log shows the agent emitted a ``gateway.rejected``
|
||||
envelope with ``reason == "tracing_gap"`` since the last spawn, treat
|
||||
that as forward progress (rule-following) and reset the strike count
|
||||
instead of incrementing.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import UTC, datetime
|
||||
from unittest.mock import AsyncMock, patch
|
||||
from uuid import uuid4
|
||||
|
||||
import pytest
|
||||
from roboco.runtime.orchestrator import AgentOrchestrator
|
||||
from roboco.seeds.initial_data import AGENT_UUIDS
|
||||
|
||||
|
||||
def _new_orchestrator() -> AgentOrchestrator:
|
||||
"""Bypass __init__ so tests don't need a full DI graph."""
|
||||
orch = AgentOrchestrator.__new__(AgentOrchestrator)
|
||||
orch._pm_respawn_tracker = {} # type: ignore[attr-defined]
|
||||
return orch
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_three_tracing_gap_responses_do_not_trip_kill() -> None:
|
||||
"""A PM rule-followed three times must NOT be killed.
|
||||
|
||||
Each spawn the audit_log shows a fresh ``gateway.rejected`` row with
|
||||
``reason == "tracing_gap"`` since the last check. That's the
|
||||
rule-following retry pattern — counter must reset every time.
|
||||
"""
|
||||
orch = _new_orchestrator()
|
||||
task_id = str(uuid4())
|
||||
task = {"id": task_id, "status": "pending"}
|
||||
|
||||
fake_audit = AsyncMock()
|
||||
fake_audit.has_recent_tracing_gap = AsyncMock(return_value=True)
|
||||
|
||||
spawn_attempts = 5
|
||||
with patch("roboco.services.audit.get_audit_service", return_value=fake_audit):
|
||||
for _ in range(spawn_attempts):
|
||||
should_gate = await orch._pm_respawn_should_gate("be-pm", task)
|
||||
assert should_gate is False
|
||||
|
||||
# Audit was consulted on every call past the first record-creation.
|
||||
expected_audit_calls = spawn_attempts - 1
|
||||
assert fake_audit.has_recent_tracing_gap.await_count >= expected_audit_calls
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_three_no_progress_spawns_still_trip_kill() -> None:
|
||||
"""When there is NO tracing_gap envelope, the strike logic still bites.
|
||||
|
||||
The classic stuck-loop case must still be detected: agent silently
|
||||
hung, no envelopes emitted, status doesn't change. Strike count
|
||||
increments each spawn and the kill fires after the threshold.
|
||||
"""
|
||||
orch = _new_orchestrator()
|
||||
task_id = str(uuid4())
|
||||
task = {"id": task_id, "status": "pending"}
|
||||
|
||||
fake_audit = AsyncMock()
|
||||
fake_audit.has_recent_tracing_gap = AsyncMock(return_value=False)
|
||||
|
||||
with patch("roboco.services.audit.get_audit_service", return_value=fake_audit):
|
||||
# Spawns 1, 2, 3 — under threshold, all allowed.
|
||||
for _ in range(3):
|
||||
assert await orch._pm_respawn_should_gate("be-pm", task) is False
|
||||
# Spawn 4 — count now exceeds _PM_RESPAWN_MAX_UNPRODUCTIVE = 3.
|
||||
assert await orch._pm_respawn_should_gate("be-pm", task) is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_status_change_resets_strike_count() -> None:
|
||||
"""Pre-existing reset path on real status change must keep working."""
|
||||
orch = _new_orchestrator()
|
||||
task_id = str(uuid4())
|
||||
|
||||
fake_audit = AsyncMock()
|
||||
fake_audit.has_recent_tracing_gap = AsyncMock(return_value=False)
|
||||
|
||||
with patch("roboco.services.audit.get_audit_service", return_value=fake_audit):
|
||||
# Two strikes on pending.
|
||||
await orch._pm_respawn_should_gate(
|
||||
"be-pm", {"id": task_id, "status": "pending"}
|
||||
)
|
||||
await orch._pm_respawn_should_gate(
|
||||
"be-pm", {"id": task_id, "status": "pending"}
|
||||
)
|
||||
# Status advances — counter must drop back to 1.
|
||||
await orch._pm_respawn_should_gate(
|
||||
"be-pm", {"id": task_id, "status": "in_progress"}
|
||||
)
|
||||
record = orch._pm_respawn_tracker[("be-pm", task_id)]
|
||||
assert record["count"] == 1
|
||||
assert record["last_status"] == "in_progress"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_audit_query_uses_correct_agent_uuid_and_task_id() -> None:
|
||||
"""The audit query must scope to (agent UUID, task UUID, since)."""
|
||||
orch = _new_orchestrator()
|
||||
task_id = str(uuid4())
|
||||
task = {"id": task_id, "status": "pending"}
|
||||
|
||||
fake_audit = AsyncMock()
|
||||
fake_audit.has_recent_tracing_gap = AsyncMock(return_value=True)
|
||||
|
||||
with patch("roboco.services.audit.get_audit_service", return_value=fake_audit):
|
||||
# First call seeds the record (no audit query yet).
|
||||
await orch._pm_respawn_should_gate("be-pm", task)
|
||||
# Second call should consult audit with be-pm's UUID + task UUID + since.
|
||||
await orch._pm_respawn_should_gate("be-pm", task)
|
||||
|
||||
fake_audit.has_recent_tracing_gap.assert_awaited()
|
||||
kwargs = fake_audit.has_recent_tracing_gap.call_args.kwargs
|
||||
assert str(kwargs["agent_id"]) == AGENT_UUIDS["be-pm"]
|
||||
assert str(kwargs["task_id"]) == task_id
|
||||
assert isinstance(kwargs["since"], datetime)
|
||||
assert kwargs["since"].tzinfo is UTC
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_unknown_slug_falls_back_to_status_only() -> None:
|
||||
"""A slug not in AGENT_UUIDS must NOT crash; just skip the audit query.
|
||||
|
||||
Defensive: if the slug map drifts, the kill loop guard still works,
|
||||
only it won't get the tracing_gap reset.
|
||||
"""
|
||||
orch = _new_orchestrator()
|
||||
task_id = str(uuid4())
|
||||
task = {"id": task_id, "status": "pending"}
|
||||
|
||||
fake_audit = AsyncMock()
|
||||
fake_audit.has_recent_tracing_gap = AsyncMock(return_value=False)
|
||||
|
||||
with patch("roboco.services.audit.get_audit_service", return_value=fake_audit):
|
||||
# Slug not present in AGENT_UUIDS — should not raise.
|
||||
for _ in range(3):
|
||||
assert await orch._pm_respawn_should_gate("not-a-real-slug", task) is False
|
||||
# Threshold trip path still reachable.
|
||||
assert await orch._pm_respawn_should_gate("not-a-real-slug", task) is True
|
||||
# Audit is never consulted because slug didn't resolve.
|
||||
fake_audit.has_recent_tracing_gap.assert_not_awaited()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_audit_query_failure_does_not_crash_gate() -> None:
|
||||
"""If audit lookup raises, fall back to the legacy strike behavior.
|
||||
|
||||
Audit is observability — it must never block orchestrator decisions.
|
||||
"""
|
||||
orch = _new_orchestrator()
|
||||
task_id = str(uuid4())
|
||||
task = {"id": task_id, "status": "pending"}
|
||||
|
||||
fake_audit = AsyncMock()
|
||||
fake_audit.has_recent_tracing_gap = AsyncMock(side_effect=RuntimeError("db down"))
|
||||
|
||||
with patch("roboco.services.audit.get_audit_service", return_value=fake_audit):
|
||||
# Strikes 1-3: allowed. 4th: gated. Same as the no-tracing-gap case.
|
||||
for _ in range(3):
|
||||
assert await orch._pm_respawn_should_gate("be-pm", task) is False
|
||||
assert await orch._pm_respawn_should_gate("be-pm", task) is True
|
||||
Reference in New Issue
Block a user