mirror of
https://github.com/rennf93/roboco.git
synced 2026-08-03 07:23:24 +02:00
1c87a4e4e4
* test: align phase1 smoke mock with the armed team-match gate The8e5f84c4sweep fixed 13 test files' inconsistent-team mocks but ran only the gateway/foundation/runtime subsets; the full gate caught this integration mock whose parent task carried an auto-generated MagicMock team and died on not_authorized before the incomplete_input assertion. * fix(runtime): attribute every agent.spawned audit to its dispatcher A rogue spawner could not be identified live (2026-07-02): agent.spawned rows carry container/model but not which dispatch loop launched them. spawn_agent now takes spawned_by, stamps it into the spawned/spawn_failed audit details, every call site passes its loop name, and an AST sweep test holds future callers to it. * fix(api): admin-complete refuses when the task's PR is still open PATCH status=completed on a task with an OPEN PR stranded its commits unmerged (bit the CEO twice live 2026-07-02). The override now refuses with the PR number/URL and the consequence before the generic hatch text; force:true stays the deliberate, audited escape. * fix(panel): awaiting_ceo_approval offers the working ceo-approve path The header's only approve action was Approve & Merge (POST /approve-and-merge, no notes) which 400s NO_PR on a branchless MegaTask umbrella — the CEO's approve button just failed. Primary action is now Approve & Complete via the CeoApproveDialog (POST /ceo-approve, notes >=20 chars, proven live); Approve & Merge stays for PR-bearing tasks. * test: stop leaking self-heal + rate-limit state into live Redis Two test files wrote real keys into a developer's localhost Redis: self-heal originate tests left self_heal:notified:* (2h TTL) and the i_am_blocked rate-limited tests left a NO-TTL 'anthropic rate-limited' tracker blob — order/state-dependent poison for anything reading the real tracker, and the prime suspect class for the one-off test_self_heal_engine full-run failure (not reproduced in 5x dir runs, adversarial orders, and a green full gate). Both files now point the computed redis_url at an unreachable port; the engines' fail-open paths keep every assertion intact. Leaked keys scrubbed live. * docs: changelog + map delta for the leak-fix batch; mypy-clean attribution test The attribution test's direct method assignments tripped the full gate's mypy (method-assign) — switched to the house monkeypatch idiom, no suppressions. * fix(gate): clear the ten xenon C-ranks; isolate all tests from live Redis Master CI has been red at the phase1 smoke test, so neither CI nor a local full gate had reached the xenon step since the team-match sweep — whose inline 'agent_team=str(agent.team) if ...' kwarg pushed nine verb bodies from B(10) to C(11-12) unseen. A shared actor_context_fields() (_protocol.py) computes (actor_slug, agent_team) once per verb, restoring all nine to B with zero behavior change; the new admin-complete override helper extraction does the same for routes/tasks.py. tests/conftest.py gains an autouse fixture pointing the computed redis_url at an unreachable port for every test — the root fix for the three families caught writing live-Redis keys (self-heal dedupe, rate-limit tracker, notification purpose-dedupe); no test uses a real Redis, and every production path is fail-open by design. * refactor(runtime): delete the never-wired dispatch-time spawn cooldown _safe_spawn / gateway_pre_spawn_check / trigger_filter had no caller in the repo's entire history (87ef42bfonly flipped the flag). Its five rules are superseded: provider parking runs inside spawn_agent, claim freshness is the guards+reaper, runaway respawns are the progress-aware breaker + notification cooldown; the per-task cooldown rule would queue-stall every normal stage handoff if wired today. gateway_triggers table kept inert. Ratified by the CEO over wiring it. * build: serialize uv — gate recipes never implicitly sync the venv Every uv run re-syncs implicitly, so a background make quality plus any foreground uv run raced two writers on one .venv and tore site-packages apart (the recurring rich/pip/bandit ImportError corruption; bit twice today, four times on 2026-07-02's first session). UV_NO_SYNC=1 is now exported Makefile-wide and quality/quality-fast/gate depend on one explicit up-front sync step. * fix(git): PR/merge/branch REST calls honor github_api_base_url Fifteen sites hardcoded https://api.github.com while the CI-run and open-PR-list calls already read settings.github_api_base_url — a GHE or test override silently applied to half the surface. One _api_base() helper keeps them uniform; default behavior unchanged. * ci: split the monolith — backend CI, Panel CI, E2E Smoke ci.yml keeps its file name and the backend quality job only (self-heal / ci-watch / release-readiness default to the ci.yml workflow); the panel job moves to panel-ci.yml scoped to panel/**, and the new scripted-agent lifecycle smoke gets e2e-smoke.yml + a make e2e-smoke target (env-gated out of the default pytest run). Trade: a panel-only red now lands on Panel CI, which the ci.yml-pinned watch engines don't see. * feat(tests): e2e lifecycle smoke harness — scripted agents, real gates tests/e2e_smoke stands up the real API (flow/do routers + middleware on uvicorn) over the ephemeral test Postgres, a local bare origin standing in for GitHub, and a fake GitHub REST layer whose merges are real git merges. A deterministic driver reloads the real MCP flow/do modules per agent and walks claim (real clone + worktree) -> tracing-gap -> note -> plan gate -> commit -> PR -> the full i_am_done ladder -> QA verdicts -> documenter -> awaiting_pm_review in ~5s. Runs via make e2e-smoke + its own CI workflow; skipped (env-gated) in the default suite. The freeze-lift condition's first half: scenario 1 green. --------- Co-authored-by: Renn F <rennf93@users.noreply.github.com>
552 lines
21 KiB
Python
552 lines
21 KiB
Python
"""Unit tests for the rate-limited path in Choreographer.i_am_blocked.
|
|
|
|
Behaviours verified here:
|
|
- i_am_blocked(reason='rate_limited') calls RateLimitStateTracker.activate()
|
|
and stores affected agent IDs; all active agents on the rate-limited
|
|
provider are subsequently marked waiting-long.
|
|
- POST /v1/i_am_blocked with reason='rate_limited' does NOT transition the
|
|
task to 'blocked'; the task remains in its current status (in_progress) and
|
|
the calling agent is parked via mark_waiting_long(waiting_for='rate_limit_lifted').
|
|
- mark_waiting_long is called for every orchestrator-tracked active agent
|
|
sharing the affected provider — call count equals active agent count.
|
|
- A RATE_LIMIT_HIT event is published to the StreamEventBus with fields
|
|
provider, affectedAgents, retryAfterSeconds, and timestamp.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
from roboco.config import settings as cfg
|
|
from roboco.models.events import EventType
|
|
from roboco.services.gateway.choreographer import Choreographer, ChoreographerDeps
|
|
from structlog.testing import capture_logs
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _unreachable_redis(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Point the tracker at an unreachable Redis for every test here.
|
|
|
|
The real RateLimitStateTracker otherwise wrote a NO-TTL "anthropic
|
|
rate-limited" state blob into a developer's live localhost Redis on
|
|
every run — order/state-dependent poison for anything reading the real
|
|
tracker. activate() failing is fine: the parking handler catches and
|
|
logs it, and these tests assert orchestrator parking, not the write.
|
|
(redis_url is a computed property — patch its inputs.)
|
|
"""
|
|
monkeypatch.setattr(cfg, "redis_host", "127.0.0.1")
|
|
monkeypatch.setattr(cfg, "redis_port", 1)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_ACTIVE_AGENTS = ["be-dev-1", "be-dev-2", "be-qa"]
|
|
_PROVIDER = "anthropic"
|
|
|
|
|
|
def _make_evidence_repo() -> AsyncMock:
|
|
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 _make_task_svc(agent_id: object, task_id: object) -> AsyncMock:
|
|
t = MagicMock(
|
|
id=task_id,
|
|
status="in_progress",
|
|
assigned_to=agent_id,
|
|
pre_block_state=None,
|
|
task_type="code",
|
|
team="backend",
|
|
# Avoid issues with spec iteration in claim guards
|
|
dependency_ids=[],
|
|
# acceptance_criteria needed by some paths
|
|
acceptance_criteria=[],
|
|
quick_context=None,
|
|
)
|
|
task_svc = AsyncMock()
|
|
task_svc.session = MagicMock()
|
|
task_svc.session.begin_nested = MagicMock(
|
|
return_value=MagicMock(
|
|
__aenter__=AsyncMock(return_value=None),
|
|
__aexit__=AsyncMock(return_value=False),
|
|
)
|
|
)
|
|
task_svc.get.return_value = t
|
|
task_svc.agent_for.return_value = MagicMock(
|
|
id=agent_id,
|
|
role="developer",
|
|
team="backend",
|
|
slug="be-dev-1", # calling agent's slug
|
|
)
|
|
return task_svc
|
|
|
|
|
|
def _make_orchestrator(
|
|
active_agents: list[str] | None = None,
|
|
provider: str = _PROVIDER,
|
|
) -> MagicMock:
|
|
"""Build a synchronous/async orchestrator mock."""
|
|
agents = active_agents if active_agents is not None else _ACTIVE_AGENTS
|
|
orch = MagicMock()
|
|
orch.get_provider_for_agent = MagicMock(return_value=provider)
|
|
orch.get_active_agent_slugs_for_provider = MagicMock(return_value=agents)
|
|
orch.mark_waiting_long = AsyncMock(return_value=None)
|
|
return orch
|
|
|
|
|
|
def _make_stream_bus() -> AsyncMock:
|
|
bus = AsyncMock()
|
|
bus.publish = AsyncMock(return_value="msg-id-1")
|
|
return bus
|
|
|
|
|
|
def _make_deps(
|
|
agent_id: object,
|
|
task_id: object,
|
|
orchestrator: MagicMock | None = None,
|
|
stream_bus: AsyncMock | None = None,
|
|
) -> ChoreographerDeps:
|
|
return ChoreographerDeps(
|
|
task=_make_task_svc(agent_id, task_id),
|
|
work_session=AsyncMock(),
|
|
git=AsyncMock(),
|
|
a2a=AsyncMock(),
|
|
journal=AsyncMock(),
|
|
audit=AsyncMock(),
|
|
evidence_repo=_make_evidence_repo(),
|
|
orchestrator=orchestrator,
|
|
stream_bus=stream_bus,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Task stays in in_progress, agent parked via mark_waiting_long
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestRateLimitedDoesNotBlockTask:
|
|
async def test_task_status_remains_in_progress(self) -> None:
|
|
"""reason='rate_limited' must NOT transition the task to 'blocked'."""
|
|
agent_id = uuid4()
|
|
task_id = uuid4()
|
|
orch = _make_orchestrator()
|
|
deps = _make_deps(agent_id, task_id, orchestrator=orch)
|
|
c = Choreographer(deps)
|
|
|
|
env = await c.i_am_blocked(agent_id, task_id, "rate_limited")
|
|
|
|
assert env.error is None
|
|
assert env.status == "in_progress"
|
|
|
|
async def test_verb_runner_block_action_not_called(self) -> None:
|
|
"""The `block` action (task.escalate) must NOT run on rate_limited path."""
|
|
agent_id = uuid4()
|
|
task_id = uuid4()
|
|
deps = _make_deps(agent_id, task_id)
|
|
c = Choreographer(deps)
|
|
|
|
await c.i_am_blocked(agent_id, task_id, "rate_limited")
|
|
|
|
# The VerbRunner calls task.escalate for the normal block path.
|
|
# In the rate-limited path this must NOT happen.
|
|
deps.task.escalate.assert_not_awaited()
|
|
|
|
async def test_calling_agent_parked_via_mark_waiting_long(self) -> None:
|
|
"""mark_waiting_long must be called with waiting_for='rate_limit_lifted'."""
|
|
agent_id = uuid4()
|
|
task_id = uuid4()
|
|
orch = _make_orchestrator(active_agents=["be-dev-1"])
|
|
deps = _make_deps(agent_id, task_id, orchestrator=orch)
|
|
c = Choreographer(deps)
|
|
|
|
await c.i_am_blocked(agent_id, task_id, "rate_limited")
|
|
|
|
# Verify that at least one mark_waiting_long call uses the right reason.
|
|
# The implementation calls mark_waiting_long(slug, waiting_for=..., ...)
|
|
# so waiting_for is always a keyword argument.
|
|
waiting_for_values = [
|
|
c.kwargs.get("waiting_for") for c in orch.mark_waiting_long.call_args_list
|
|
]
|
|
assert "rate_limit_lifted" in waiting_for_values
|
|
|
|
async def test_case_insensitive_reason_match(self) -> None:
|
|
"""reason='Rate_Limited' (any case) should trigger the special path."""
|
|
agent_id = uuid4()
|
|
task_id = uuid4()
|
|
orch = _make_orchestrator(active_agents=["be-dev-1"])
|
|
deps = _make_deps(agent_id, task_id, orchestrator=orch)
|
|
c = Choreographer(deps)
|
|
|
|
env = await c.i_am_blocked(agent_id, task_id, "Rate_Limited")
|
|
|
|
assert env.error is None
|
|
assert env.status == "in_progress"
|
|
|
|
async def test_struggle_journal_still_written(self) -> None:
|
|
"""journal.write_struggle must still be written on the rate_limited path."""
|
|
agent_id = uuid4()
|
|
task_id = uuid4()
|
|
deps = _make_deps(agent_id, task_id)
|
|
c = Choreographer(deps)
|
|
|
|
await c.i_am_blocked(agent_id, task_id, "rate_limited")
|
|
|
|
deps.journal.write_struggle.assert_awaited_once()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# mark_waiting_long called for every active agent on affected provider
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestMarkWaitingLongCallCount:
|
|
async def test_call_count_equals_active_agent_count(self) -> None:
|
|
"""mark_waiting_long must be called once per active agent."""
|
|
agent_id = uuid4()
|
|
task_id = uuid4()
|
|
active = ["be-dev-1", "be-dev-2", "be-dev-3"]
|
|
orch = _make_orchestrator(active_agents=active)
|
|
deps = _make_deps(agent_id, task_id, orchestrator=orch)
|
|
c = Choreographer(deps)
|
|
|
|
await c.i_am_blocked(agent_id, task_id, "rate_limited")
|
|
|
|
assert orch.mark_waiting_long.call_count == len(active)
|
|
|
|
async def test_call_count_with_single_active_agent(self) -> None:
|
|
agent_id = uuid4()
|
|
task_id = uuid4()
|
|
orch = _make_orchestrator(active_agents=["be-dev-1"])
|
|
deps = _make_deps(agent_id, task_id, orchestrator=orch)
|
|
c = Choreographer(deps)
|
|
|
|
await c.i_am_blocked(agent_id, task_id, "rate_limited")
|
|
|
|
assert orch.mark_waiting_long.call_count == 1
|
|
|
|
async def test_no_calls_when_no_active_agents(self) -> None:
|
|
agent_id = uuid4()
|
|
task_id = uuid4()
|
|
orch = _make_orchestrator(active_agents=[])
|
|
deps = _make_deps(agent_id, task_id, orchestrator=orch)
|
|
c = Choreographer(deps)
|
|
|
|
await c.i_am_blocked(agent_id, task_id, "rate_limited")
|
|
|
|
assert orch.mark_waiting_long.call_count == 0
|
|
|
|
async def test_no_calls_when_orchestrator_is_none(self) -> None:
|
|
"""When orchestrator is not wired in, no parking happens but no crash."""
|
|
agent_id = uuid4()
|
|
task_id = uuid4()
|
|
deps = _make_deps(agent_id, task_id, orchestrator=None)
|
|
c = Choreographer(deps)
|
|
|
|
env = await c.i_am_blocked(agent_id, task_id, "rate_limited")
|
|
|
|
# Should still succeed; no orchestrator = no parking
|
|
assert env.error is None
|
|
assert env.status == "in_progress"
|
|
|
|
async def test_mark_waiting_long_receives_waiting_for_arg(self) -> None:
|
|
"""Every mark_waiting_long call must carry waiting_for='rate_limit_lifted'."""
|
|
agent_id = uuid4()
|
|
task_id = uuid4()
|
|
active = ["be-dev-1", "be-qa"]
|
|
orch = _make_orchestrator(active_agents=active)
|
|
deps = _make_deps(agent_id, task_id, orchestrator=orch)
|
|
c = Choreographer(deps)
|
|
|
|
await c.i_am_blocked(agent_id, task_id, "rate_limited")
|
|
|
|
for c_args in orch.mark_waiting_long.call_args_list:
|
|
# mark_waiting_long(slug, waiting_for=..., ...) — waiting_for is a kwarg
|
|
assert c_args.kwargs.get("waiting_for") == "rate_limit_lifted"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# RATE_LIMIT_HIT event published with correct payload structure
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestRateLimitHitEventPublished:
|
|
async def test_stream_bus_publish_called_once(self) -> None:
|
|
agent_id = uuid4()
|
|
task_id = uuid4()
|
|
orch = _make_orchestrator()
|
|
bus = _make_stream_bus()
|
|
deps = _make_deps(agent_id, task_id, orchestrator=orch, stream_bus=bus)
|
|
c = Choreographer(deps)
|
|
|
|
await c.i_am_blocked(agent_id, task_id, "rate_limited")
|
|
|
|
bus.publish.assert_awaited_once()
|
|
|
|
async def test_event_type_is_rate_limit_hit(self) -> None:
|
|
agent_id = uuid4()
|
|
task_id = uuid4()
|
|
orch = _make_orchestrator()
|
|
bus = _make_stream_bus()
|
|
deps = _make_deps(agent_id, task_id, orchestrator=orch, stream_bus=bus)
|
|
c = Choreographer(deps)
|
|
|
|
await c.i_am_blocked(agent_id, task_id, "rate_limited")
|
|
|
|
event = bus.publish.call_args.args[0]
|
|
assert event.type == EventType.RATE_LIMIT_HIT
|
|
|
|
async def test_event_data_has_provider_field(self) -> None:
|
|
agent_id = uuid4()
|
|
task_id = uuid4()
|
|
orch = _make_orchestrator(provider="anthropic")
|
|
bus = _make_stream_bus()
|
|
deps = _make_deps(agent_id, task_id, orchestrator=orch, stream_bus=bus)
|
|
c = Choreographer(deps)
|
|
|
|
await c.i_am_blocked(agent_id, task_id, "rate_limited")
|
|
|
|
event = bus.publish.call_args.args[0]
|
|
assert "provider" in event.data
|
|
assert event.data["provider"] == "anthropic"
|
|
|
|
async def test_event_data_has_affected_agents_list(self) -> None:
|
|
agent_id = uuid4()
|
|
task_id = uuid4()
|
|
active = ["be-dev-1", "be-dev-2"]
|
|
orch = _make_orchestrator(active_agents=active)
|
|
bus = _make_stream_bus()
|
|
deps = _make_deps(agent_id, task_id, orchestrator=orch, stream_bus=bus)
|
|
c = Choreographer(deps)
|
|
|
|
await c.i_am_blocked(agent_id, task_id, "rate_limited")
|
|
|
|
event = bus.publish.call_args.args[0]
|
|
assert "affectedAgents" in event.data
|
|
assert isinstance(event.data["affectedAgents"], list)
|
|
assert event.data["affectedAgents"] == active
|
|
|
|
async def test_event_data_has_retry_after_seconds_null_by_default(self) -> None:
|
|
agent_id = uuid4()
|
|
task_id = uuid4()
|
|
orch = _make_orchestrator()
|
|
bus = _make_stream_bus()
|
|
deps = _make_deps(agent_id, task_id, orchestrator=orch, stream_bus=bus)
|
|
c = Choreographer(deps)
|
|
|
|
await c.i_am_blocked(agent_id, task_id, "rate_limited")
|
|
|
|
event = bus.publish.call_args.args[0]
|
|
assert "retryAfterSeconds" in event.data
|
|
assert event.data["retryAfterSeconds"] is None
|
|
|
|
async def test_event_data_retry_after_parsed_from_what_needed(self) -> None:
|
|
"""If what_needed is a numeric string, it becomes retryAfterSeconds."""
|
|
agent_id = uuid4()
|
|
task_id = uuid4()
|
|
orch = _make_orchestrator()
|
|
bus = _make_stream_bus()
|
|
deps = _make_deps(agent_id, task_id, orchestrator=orch, stream_bus=bus)
|
|
c = Choreographer(deps)
|
|
|
|
await c.i_am_blocked(agent_id, task_id, "rate_limited", what_needed="30")
|
|
|
|
event = bus.publish.call_args.args[0]
|
|
assert event.data["retryAfterSeconds"] == float("30")
|
|
|
|
async def test_event_data_has_timestamp_iso_string(self) -> None:
|
|
agent_id = uuid4()
|
|
task_id = uuid4()
|
|
orch = _make_orchestrator()
|
|
bus = _make_stream_bus()
|
|
deps = _make_deps(agent_id, task_id, orchestrator=orch, stream_bus=bus)
|
|
c = Choreographer(deps)
|
|
|
|
await c.i_am_blocked(agent_id, task_id, "rate_limited")
|
|
|
|
event = bus.publish.call_args.args[0]
|
|
assert "timestamp" in event.data
|
|
# ISO string: must be a non-empty string
|
|
ts = event.data["timestamp"]
|
|
assert isinstance(ts, str) and len(ts) > 0
|
|
|
|
async def test_no_publish_when_stream_bus_is_none(self) -> None:
|
|
"""When stream_bus is not wired in, no publish is attempted."""
|
|
agent_id = uuid4()
|
|
task_id = uuid4()
|
|
orch = _make_orchestrator()
|
|
# stream_bus=None: no bus
|
|
deps = _make_deps(agent_id, task_id, orchestrator=orch, stream_bus=None)
|
|
c = Choreographer(deps)
|
|
|
|
env = await c.i_am_blocked(agent_id, task_id, "rate_limited")
|
|
|
|
# Should still succeed
|
|
assert env.error is None
|
|
assert env.status == "in_progress"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# RateLimitStateTracker.activate() called on rate_limited path
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_TRACKER_PATCH = "roboco.services.gateway.rate_limit_tracker.RateLimitStateTracker"
|
|
|
|
|
|
class TestRateLimitTrackerActivateOnParking:
|
|
"""Verify that _handle_rate_limited_parking() calls activate()."""
|
|
|
|
async def test_activate_called_when_provider_known(self) -> None:
|
|
"""activate() must be called once when provider != 'unknown'."""
|
|
agent_id = uuid4()
|
|
task_id = uuid4()
|
|
orch = _make_orchestrator(active_agents=["be-dev-1"], provider=_PROVIDER)
|
|
deps = _make_deps(agent_id, task_id, orchestrator=orch)
|
|
c = Choreographer(deps)
|
|
|
|
mock_tracker = AsyncMock()
|
|
mock_tracker.activate = AsyncMock(return_value=None)
|
|
mock_tracker_cls = MagicMock(return_value=mock_tracker)
|
|
|
|
with patch(_TRACKER_PATCH, mock_tracker_cls):
|
|
await c.i_am_blocked(agent_id, task_id, "rate_limited")
|
|
|
|
mock_tracker_cls.assert_called_once_with(_PROVIDER)
|
|
mock_tracker.activate.assert_awaited_once()
|
|
|
|
async def test_activate_receives_affected_agents(self) -> None:
|
|
"""activate() must be called with the affected_agents list."""
|
|
agent_id = uuid4()
|
|
task_id = uuid4()
|
|
active = ["be-dev-1", "be-dev-2"]
|
|
orch = _make_orchestrator(active_agents=active, provider=_PROVIDER)
|
|
deps = _make_deps(agent_id, task_id, orchestrator=orch)
|
|
c = Choreographer(deps)
|
|
|
|
mock_tracker = AsyncMock()
|
|
mock_tracker.activate = AsyncMock(return_value=None)
|
|
mock_tracker_cls = MagicMock(return_value=mock_tracker)
|
|
|
|
with patch(_TRACKER_PATCH, mock_tracker_cls):
|
|
await c.i_am_blocked(agent_id, task_id, "rate_limited")
|
|
|
|
call_kwargs = mock_tracker.activate.call_args.kwargs
|
|
assert call_kwargs.get("affected_agents") == active
|
|
|
|
async def test_activate_receives_retry_after_from_what_needed(self) -> None:
|
|
"""activate() must receive retry_after parsed from what_needed."""
|
|
agent_id = uuid4()
|
|
task_id = uuid4()
|
|
orch = _make_orchestrator(active_agents=["be-dev-1"], provider=_PROVIDER)
|
|
deps = _make_deps(agent_id, task_id, orchestrator=orch)
|
|
c = Choreographer(deps)
|
|
|
|
mock_tracker = AsyncMock()
|
|
mock_tracker.activate = AsyncMock(return_value=None)
|
|
mock_tracker_cls = MagicMock(return_value=mock_tracker)
|
|
|
|
with patch(_TRACKER_PATCH, mock_tracker_cls):
|
|
await c.i_am_blocked(agent_id, task_id, "rate_limited", what_needed="45")
|
|
|
|
call_kwargs = mock_tracker.activate.call_args.kwargs
|
|
assert call_kwargs.get("retry_after") == float("45")
|
|
|
|
async def test_activate_retry_after_none_when_what_needed_not_numeric(self) -> None:
|
|
"""activate() must receive retry_after=None when what_needed is not a number."""
|
|
agent_id = uuid4()
|
|
task_id = uuid4()
|
|
orch = _make_orchestrator(active_agents=["be-dev-1"], provider=_PROVIDER)
|
|
deps = _make_deps(agent_id, task_id, orchestrator=orch)
|
|
c = Choreographer(deps)
|
|
|
|
mock_tracker = AsyncMock()
|
|
mock_tracker.activate = AsyncMock(return_value=None)
|
|
mock_tracker_cls = MagicMock(return_value=mock_tracker)
|
|
|
|
with patch(_TRACKER_PATCH, mock_tracker_cls):
|
|
await c.i_am_blocked(
|
|
agent_id, task_id, "rate_limited", what_needed="retry soon"
|
|
)
|
|
|
|
call_kwargs = mock_tracker.activate.call_args.kwargs
|
|
assert call_kwargs.get("retry_after") is None
|
|
|
|
async def test_activate_skipped_when_provider_unknown(self) -> None:
|
|
"""activate() must NOT be called when provider resolves to 'unknown'."""
|
|
agent_id = uuid4()
|
|
task_id = uuid4()
|
|
# get_provider_for_agent returns None → provider stays 'unknown'
|
|
orch = MagicMock()
|
|
orch.get_provider_for_agent = MagicMock(return_value=None)
|
|
orch.get_active_agent_slugs_for_provider = MagicMock(return_value=[])
|
|
orch.mark_waiting_long = AsyncMock(return_value=None)
|
|
deps = _make_deps(agent_id, task_id, orchestrator=orch)
|
|
c = Choreographer(deps)
|
|
|
|
mock_tracker = AsyncMock()
|
|
mock_tracker.activate = AsyncMock(return_value=None)
|
|
mock_tracker_cls = MagicMock(return_value=mock_tracker)
|
|
|
|
with patch(_TRACKER_PATCH, mock_tracker_cls):
|
|
env = await c.i_am_blocked(agent_id, task_id, "rate_limited")
|
|
|
|
# No crash, no activate call
|
|
assert env.error is None
|
|
mock_tracker.activate.assert_not_awaited()
|
|
|
|
async def test_activate_failure_does_not_crash_path(self) -> None:
|
|
"""If activate() raises, _handle_rate_limited_parking must still succeed."""
|
|
agent_id = uuid4()
|
|
task_id = uuid4()
|
|
orch = _make_orchestrator(active_agents=["be-dev-1"], provider=_PROVIDER)
|
|
deps = _make_deps(agent_id, task_id, orchestrator=orch)
|
|
c = Choreographer(deps)
|
|
|
|
mock_tracker = AsyncMock()
|
|
mock_tracker.activate = AsyncMock(side_effect=RuntimeError("redis down"))
|
|
mock_tracker_cls = MagicMock(return_value=mock_tracker)
|
|
|
|
with patch(_TRACKER_PATCH, mock_tracker_cls):
|
|
env = await c.i_am_blocked(agent_id, task_id, "rate_limited")
|
|
|
|
assert env.error is None
|
|
assert env.status == "in_progress"
|
|
|
|
async def test_activate_failure_is_logged_not_silent(self) -> None:
|
|
"""An activate() failure must be logged loudly, not bare-suppressed —
|
|
the probe-resume loop is tracker-driven, so a silent failure strands
|
|
every parked agent in WAITING_LONG with no probe ever running.
|
|
"""
|
|
agent_id = uuid4()
|
|
task_id = uuid4()
|
|
orch = _make_orchestrator(active_agents=["be-dev-1"], provider=_PROVIDER)
|
|
deps = _make_deps(agent_id, task_id, orchestrator=orch)
|
|
c = Choreographer(deps)
|
|
|
|
mock_tracker = AsyncMock()
|
|
mock_tracker.activate = AsyncMock(side_effect=RuntimeError("redis down"))
|
|
mock_tracker_cls = MagicMock(return_value=mock_tracker)
|
|
|
|
with patch(_TRACKER_PATCH, mock_tracker_cls), capture_logs() as logs:
|
|
env = await c.i_am_blocked(agent_id, task_id, "rate_limited")
|
|
|
|
assert env.error is None
|
|
assert any(
|
|
"activate" in str(e.get("event", "")).lower()
|
|
and e.get("log_level") == "error"
|
|
for e in logs
|
|
), f"expected an error log about activate failure; got {logs!r}"
|