From c7acf2ff27c4bd7aaba482e17763efa6fb1c1ca3 Mon Sep 17 00:00:00 2001 From: Johannes du Plessis Date: Tue, 9 Jun 2026 13:19:11 -0700 Subject: [PATCH] Revert out-of-process usage-snapshot builder (#1473) * Revert "fix: schedule usage snapshot runs with explicit empty input (#1470)" This reverts commit 49aa638c5862bbe3e7074a5943584afd14b54e05. * Revert "feat: out-of-process usage-snapshot builder (Phase 1) (#1468)" This reverts commit 798cc6edf18670a8879aa84eef291ef826565a5d. --- agent/dashboard/agent_usage.py | 55 +++---- agent/dashboard/routes.py | 25 +-- agent/dashboard/usage_snapshot_cron.py | 133 --------------- agent/usage_snapshot.py | 75 --------- agent/webapp.py | 18 --- langgraph.json | 3 +- tests/test_agent_usage.py | 43 +---- tests/test_usage_snapshot.py | 214 ------------------------- 8 files changed, 41 insertions(+), 525 deletions(-) delete mode 100644 agent/dashboard/usage_snapshot_cron.py delete mode 100644 agent/usage_snapshot.py delete mode 100644 tests/test_usage_snapshot.py diff --git a/agent/dashboard/agent_usage.py b/agent/dashboard/agent_usage.py index 5bff61f0..a53ddd7e 100644 --- a/agent/dashboard/agent_usage.py +++ b/agent/dashboard/agent_usage.py @@ -5,7 +5,7 @@ from __future__ import annotations import asyncio import logging from collections import Counter -from collections.abc import Callable +from collections.abc import Awaitable, Callable from datetime import UTC, datetime, timedelta from typing import Any, Literal @@ -25,6 +25,7 @@ _AGENT_SOURCES = frozenset({"dashboard", "github", "slack", "linear"}) _PR_REFRESH_INTERVAL_MS = 10 * 60 * 1000 _MAX_PR_REFRESH_PER_REQUEST = 25 _PR_REFRESH_CONCURRENCY = 5 +_CACHE_TTL_MS = 10 * 60 * 1000 _CACHE_SEARCH_LIMIT = 1000 _GITHUB_API = "https://api.github.com" @@ -604,45 +605,24 @@ async def refresh_reviewer_stats_cache(period: str | None = "30d") -> dict[str, return snapshot -def _empty_usage_snapshot(period: Period) -> dict[str, Any]: - return {"period": period, "users": [], "total_members": 0, "computing": True} - - -def _empty_reviewer_snapshot(period: Period) -> dict[str, Any]: - return { - "period": period, - "reviewed_prs": 0, - "prs_with_findings": 0, - "findings_recorded": 0, - "surfaced_findings": 0, - "addressed_findings": 0, - "resolved_after_update": 0, - "dismissed_findings": 0, - "unresolved_surfaced_findings": 0, - "resolution_rate": 0.0, - "human_replies": 0, - "severity_counts": {}, - "top_categories": [], - "computing": True, - } - - async def _cached_snapshot( namespace: list[str], period: Period, - empty: Callable[[Period], dict[str, Any]], + refresh: Callable[[str | None], Awaitable[dict[str, Any]]], + *, + schedule_refresh: Callable[[Period], None] | None = None, ) -> tuple[dict[str, Any], int | None]: - """Pure read: return the precomputed snapshot (any age) or a typed placeholder. - - Never computes inline and never schedules a refresh — the scheduled builder - owns refresh cadence (see ``agent/usage_snapshot.py``). - """ cached = await _get_value(namespace, period) if cached: snapshot = cached.get("snapshot") + generated_at_ms = _coerce_int(cached.get("generated_at_ms")) if isinstance(snapshot, dict): - return snapshot, _coerce_int(cached.get("generated_at_ms")) or None - return empty(period), None + if not generated_at_ms or _now_ms() - generated_at_ms <= _CACHE_TTL_MS: + return snapshot, generated_at_ms or None + if schedule_refresh is not None: + schedule_refresh(period) + return snapshot, generated_at_ms + return await refresh(period), _now_ms() def _usage_payload_from_snapshot( @@ -697,7 +677,6 @@ def _usage_payload_from_snapshot( "total_members": _coerce_int(snapshot.get("total_members")), "current_user_rank": current_user_row["rank"] if current_user_row else None, "generated_at_ms": generated_at_ms, - "computing": bool(snapshot.get("computing")), } @@ -707,18 +686,22 @@ async def list_agent_usage_leaderboard( limit: int, current_login: str | None, current_email: str | None, + schedule_usage_refresh: Callable[[Period], None] | None = None, + schedule_reviewer_refresh: Callable[[Period], None] | None = None, ) -> dict[str, Any]: - """Return cached Open SWE Agent and reviewer usage stats (pure cache read).""" + """Return cached Open SWE Agent and reviewer usage stats.""" normalized_period = _normalize_period(period) usage_snapshot, generated_at_ms = await _cached_snapshot( USAGE_LEADERBOARD_CACHE_NAMESPACE, normalized_period, - _empty_usage_snapshot, + refresh_usage_leaderboard_cache, + schedule_refresh=schedule_usage_refresh, ) reviewer_stats, reviewer_generated_at_ms = await _cached_snapshot( REVIEWER_STATS_CACHE_NAMESPACE, normalized_period, - _empty_reviewer_snapshot, + refresh_reviewer_stats_cache, + schedule_refresh=schedule_reviewer_refresh, ) payload = _usage_payload_from_snapshot( usage_snapshot, diff --git a/agent/dashboard/routes.py b/agent/dashboard/routes.py index 5fc94a4e..f832a71d 100644 --- a/agent/dashboard/routes.py +++ b/agent/dashboard/routes.py @@ -8,12 +8,16 @@ import os from typing import Any import httpx -from fastapi import APIRouter, Depends, HTTPException, Request +from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Request from fastapi.responses import RedirectResponse, Response, StreamingResponse from pydantic import BaseModel from .admin import is_admin -from .agent_usage import list_agent_usage_leaderboard +from .agent_usage import ( + list_agent_usage_leaderboard, + refresh_reviewer_stats_cache, + refresh_usage_leaderboard_cache, +) from .analyzer_cron import remove_continual_cron from .enabled_repos import ( list_enabled_review_repos, @@ -93,7 +97,6 @@ from .thread_api import ( send_dashboard_message, stream_dashboard_thread, ) -from .usage_snapshot_cron import ensure_usage_snapshot_cron, trigger_usage_snapshot_build from .user_mappings import ( delete_mapping, get_mapping, @@ -707,27 +710,25 @@ async def api_delete_review_style( @router.get("/agent-usage-leaderboard") async def api_agent_usage_leaderboard( + background_tasks: BackgroundTasks, period: str | None = "30d", limit: int = 10, session: dict[str, Any] = _SESSION_DEP, ) -> dict[str, Any]: - await ensure_usage_snapshot_cron() return await list_agent_usage_leaderboard( period=period, limit=limit, current_login=session["sub"], current_email=session.get("email"), + schedule_usage_refresh=lambda cache_period: background_tasks.add_task( + refresh_usage_leaderboard_cache, cache_period + ), + schedule_reviewer_refresh=lambda cache_period: background_tasks.add_task( + refresh_reviewer_stats_cache, cache_period + ), ) -@router.post("/admin/usage/rebuild") -async def api_admin_usage_rebuild( - session: dict[str, Any] = _ADMIN_DEP, -) -> dict[str, Any]: - await ensure_usage_snapshot_cron() - return await trigger_usage_snapshot_build() - - @router.get("/schedules") async def api_list_schedules( session: dict[str, Any] = _SESSION_DEP, diff --git a/agent/dashboard/usage_snapshot_cron.py b/agent/dashboard/usage_snapshot_cron.py deleted file mode 100644 index 99684c15..00000000 --- a/agent/dashboard/usage_snapshot_cron.py +++ /dev/null @@ -1,133 +0,0 @@ -"""Global cron + on-demand scheduling for the usage-snapshot builder. - -One global cron rebuilds every period's snapshot ~every 10 min (staggered off -:00). Registration is idempotent and concrete (mirrors ``analyzer_cron``): we -look for an existing cron tagged ``{"kind": "usage_snapshot"}`` before creating -one, and stash the ``cron_id`` under ``["agent_usage", "meta"] / "cron"``. -""" - -from __future__ import annotations - -import logging -from typing import Any - -from .review_style_jobs import _client - -logger = logging.getLogger(__name__) - -_ASSISTANT_ID = "usage_snapshot" -_CRON_KIND = "usage_snapshot" -# A StateGraph run with no input raises EmptyInputError at __start__, so every -# scheduled run must carry an explicit empty input. rev marks crons created -# with the correct payload; older revs are reaped and recreated. -_CRON_INPUT: dict[str, Any] = {} -_CRON_REV = 2 -_SCHEDULE = "7,17,27,37,47,57 * * * *" # ~every 10 min, staggered off :00 -_META_NAMESPACE: list[str] = ["agent_usage", "meta"] -_META_CRON_KEY = "cron" - -# Once we've confirmed a cron is registered, steady-state requests skip the -# loopback check entirely — there is nothing left to re-verify. -_registered_cron_id: str | None = None - - -def _cron_id_of(cron: Any) -> str | None: - cron_id = cron.get("cron_id") if isinstance(cron, dict) else getattr(cron, "cron_id", None) - return cron_id if isinstance(cron_id, str) and cron_id else None - - -def _cron_rev_of(cron: Any) -> int: - metadata = cron.get("metadata") if isinstance(cron, dict) else getattr(cron, "metadata", None) - rev = metadata.get("rev") if isinstance(metadata, dict) else None - return rev if isinstance(rev, int) else 1 - - -async def _existing_cron_id() -> str | None: - item = await _client().store.get_item(_META_NAMESPACE, _META_CRON_KEY) - value = item.get("value") if isinstance(item, dict) else getattr(item, "value", None) - if not isinstance(value, dict) or value.get("rev") != _CRON_REV: - return None - cron_id = value.get("cron_id") - return cron_id if isinstance(cron_id, str) and cron_id else None - - -async def ensure_usage_snapshot_cron() -> str | None: - """Idempotently register the global usage-snapshot cron. Returns its id.""" - global _registered_cron_id - if _registered_cron_id: - return _registered_cron_id - - try: - existing = await _existing_cron_id() - except Exception: - logger.debug("Could not read usage snapshot cron meta", exc_info=True) - existing = None - if existing: - _registered_cron_id = existing - return existing - - try: - crons = await _client().crons.search(metadata={"kind": _CRON_KIND}, limit=10) - current = [ - cid for c in (crons or []) if (cid := _cron_id_of(c)) and _cron_rev_of(c) == _CRON_REV - ] - stale = [ - cid for c in (crons or []) if (cid := _cron_id_of(c)) and _cron_rev_of(c) != _CRON_REV - ] - # Stale revs were created without input and fail at __start__; replace them. - await _delete_crons(stale) - if current: - keep = current[0] - # search-then-create isn't atomic; concurrent replicas can each - # create one. Reap any extras so we don't fire the build N times. - await _delete_crons(current[1:]) - await _store_cron_id(keep) - _registered_cron_id = keep - return keep - except Exception: - logger.debug("Usage snapshot cron search failed", exc_info=True) - - try: - cron = await _client().crons.create( - _ASSISTANT_ID, - schedule=_SCHEDULE, - input=_CRON_INPUT, - metadata={"kind": _CRON_KIND, "rev": _CRON_REV}, - ) - except Exception: - logger.exception("Failed to create usage snapshot cron") - return None - - cron_id = _cron_id_of(cron) - if cron_id: - await _store_cron_id(cron_id) - _registered_cron_id = cron_id - return cron_id - return None - - -async def _delete_crons(cron_ids: list[str]) -> None: - for cron_id in cron_ids: - try: - await _client().crons.delete(cron_id) - except Exception: - logger.debug("Could not delete duplicate usage cron %s", cron_id, exc_info=True) - - -async def _store_cron_id(cron_id: str) -> None: - try: - await _client().store.put_item( - _META_NAMESPACE, _META_CRON_KEY, {"cron_id": cron_id, "rev": _CRON_REV} - ) - except Exception: - logger.debug("Could not persist usage snapshot cron id", exc_info=True) - - -async def trigger_usage_snapshot_build() -> dict[str, Any]: - """Fire-and-forget: schedule (not execute) one immediate build run.""" - try: - await _client().runs.create(None, _ASSISTANT_ID, input=_CRON_INPUT) - except Exception: - logger.debug("Could not schedule immediate usage snapshot build", exc_info=True) - return {"status": "error"} - return {"status": "scheduled"} diff --git a/agent/usage_snapshot.py b/agent/usage_snapshot.py deleted file mode 100644 index a90163f2..00000000 --- a/agent/usage_snapshot.py +++ /dev/null @@ -1,75 +0,0 @@ -"""Out-of-process builder graph for the usage-tab snapshots. - -A single pure-python node (no LLM, no sandbox) that rebuilds every period's -usage leaderboard + reviewer-stats snapshot and writes them to the cache -namespaces. Scheduled by a global cron (see ``dashboard/usage_snapshot_cron.py``) -so heavy/looping work never runs on the run-serving HTTP process (the #1434 bug -class). The whole build is bounded by ``asyncio.timeout`` so a wedged build -self-cancels before the next tick. -""" - -from __future__ import annotations - -import asyncio -import logging -import os -from typing import Any, TypedDict - -from langgraph.graph import END, START, StateGraph -from langgraph.graph.state import RunnableConfig - -from .dashboard.agent_usage import ( - refresh_reviewer_stats_cache, - refresh_usage_leaderboard_cache, -) - -logger = logging.getLogger(__name__) - -_PERIODS = ("7d", "30d", "all") -_BUILD_TIMEOUT_S = 120 - - -class UsageSnapshotState(TypedDict, total=False): - result: dict[str, Any] - - -def _cron_enabled() -> bool: - value = os.environ.get("USAGE_SNAPSHOT_CRON_ENABLED", "true").strip().lower() - return value not in {"0", "false", "no", "off"} - - -async def _build(state: UsageSnapshotState, config: RunnableConfig) -> dict[str, Any]: - if not _cron_enabled(): - logger.info("Usage snapshot build skipped: USAGE_SNAPSHOT_CRON_ENABLED is off") - return {"result": {"status": "disabled"}} - - built: list[str] = [] - failed: list[str] = [] - try: - async with asyncio.timeout(_BUILD_TIMEOUT_S): - for period in _PERIODS: - for label, refresh in ( - ("usage", refresh_usage_leaderboard_cache), - ("reviewer", refresh_reviewer_stats_cache), - ): - try: - await refresh(period) - built.append(f"{label}:{period}") - except Exception: - logger.exception("Usage snapshot build failed for %s:%s", label, period) - failed.append(f"{label}:{period}") - except TimeoutError: - logger.warning("Usage snapshot build exceeded %ss budget; cancelled", _BUILD_TIMEOUT_S) - return {"result": {"status": "timeout", "built": built, "failed": failed}} - - status = "ok" if not failed else "partial" - logger.info("Usage snapshot build %s: built=%s failed=%s", status, built, failed) - return {"result": {"status": status, "built": built, "failed": failed}} - - -def get_usage_snapshot(config: RunnableConfig | None = None): - builder = StateGraph(UsageSnapshotState) - builder.add_node("build", _build) - builder.add_edge(START, "build") - builder.add_edge("build", END) - return builder.compile().with_config(config or {}) diff --git a/agent/webapp.py b/agent/webapp.py index dff127c0..98fa9404 100644 --- a/agent/webapp.py +++ b/agent/webapp.py @@ -1,6 +1,5 @@ """Custom FastAPI routes for LangGraph server.""" -import asyncio import hashlib import hmac import json @@ -111,26 +110,9 @@ logger = logging.getLogger(__name__) @asynccontextmanager async def lifespan(_app: FastAPI) -> AsyncIterator[None]: - from .dashboard.usage_snapshot_cron import ( - ensure_usage_snapshot_cron, - trigger_usage_snapshot_build, - ) from .utils.sandbox import validate_sandbox_startup_config validate_sandbox_startup_config() - - # One bounded scheduling call — never a retained/looping task (the exact - # #1434 bug class). These are loopback HTTP calls into a server that is - # still starting, so cap them hard: a hang must not block startup. Any - # failure/timeout is fine — the cron and lazy per-request registration both - # catch up within the cadence. - try: - async with asyncio.timeout(5): - await ensure_usage_snapshot_cron() - await trigger_usage_snapshot_build() - except Exception: - logger.debug("Usage snapshot bootstrap on startup skipped", exc_info=True) - yield diff --git a/langgraph.json b/langgraph.json index 1a56a8ab..861ffc49 100644 --- a/langgraph.json +++ b/langgraph.json @@ -5,8 +5,7 @@ "agent": "agent.server:get_agent", "reviewer": "agent.reviewer:get_reviewer_agent", "analyzer": "agent.analyzer:get_analyzer", - "scheduler": "agent.scheduler:get_scheduler", - "usage_snapshot": "agent.usage_snapshot:get_usage_snapshot" + "scheduler": "agent.scheduler:get_scheduler" }, "dependencies": ["."], "http": { diff --git a/tests/test_agent_usage.py b/tests/test_agent_usage.py index 6af3efd5..b340558f 100644 --- a/tests/test_agent_usage.py +++ b/tests/test_agent_usage.py @@ -38,7 +38,7 @@ class FakeClient: @pytest.mark.asyncio -async def test_cached_usage_payload_returns_snapshot_without_refresh(monkeypatch): +async def test_cached_usage_payload_returns_stale_snapshot_and_schedules_refresh(monkeypatch): usage_snapshot = { "period": "30d", "total_members": 2, @@ -101,54 +101,27 @@ async def test_cached_usage_payload_returns_snapshot_without_refresh(monkeypatch } ) monkeypatch.setattr(agent_usage, "_client", lambda: FakeClient(store=store)) + monkeypatch.setattr(agent_usage, "_now_ms", lambda: agent_usage._CACHE_TTL_MS + 2) - refreshed: list[str] = [] - monkeypatch.setattr( - agent_usage, - "refresh_usage_leaderboard_cache", - lambda *a, **k: refreshed.append("usage"), - ) - monkeypatch.setattr( - agent_usage, - "refresh_reviewer_stats_cache", - lambda *a, **k: refreshed.append("reviewer"), - ) - + usage_refreshes: list[str] = [] + reviewer_refreshes: list[str] = [] payload = await agent_usage.list_agent_usage_leaderboard( period="30d", limit=1, current_login="octo", current_email="octo@example.com", + schedule_usage_refresh=usage_refreshes.append, + schedule_reviewer_refresh=reviewer_refreshes.append, ) - # Pure read: stale snapshot is served as-is, never refreshed inline. - assert refreshed == [] + assert usage_refreshes == ["30d"] + assert reviewer_refreshes == ["30d"] assert payload["rows"][0]["user"]["email"] == "octo@example.com" assert payload["total_members"] == 2 assert payload["reviewer_stats"]["surfaced_findings"] == 1 assert store.puts == [] -@pytest.mark.asyncio -async def test_cold_cache_returns_computing_placeholder(monkeypatch): - store = FakeStore({}) - monkeypatch.setattr(agent_usage, "_client", lambda: FakeClient(store=store)) - - payload = await agent_usage.list_agent_usage_leaderboard( - period="7d", - limit=10, - current_login="octo", - current_email="octo@example.com", - ) - - assert payload["rows"] == [] - assert payload["total_members"] == 0 - assert payload["generated_at_ms"] is None - assert payload["computing"] is True - assert payload["reviewer_stats"]["computing"] is True - assert store.puts == [] - - @pytest.mark.asyncio async def test_reviewer_stats_snapshot_counts_surfaced_and_resolved_findings(monkeypatch): threads = [ diff --git a/tests/test_usage_snapshot.py b/tests/test_usage_snapshot.py deleted file mode 100644 index 669612a8..00000000 --- a/tests/test_usage_snapshot.py +++ /dev/null @@ -1,214 +0,0 @@ -from __future__ import annotations - -import inspect - -import pytest - -from agent import usage_snapshot -from agent.dashboard import usage_snapshot_cron - - -class FakeStore: - def __init__(self) -> None: - self.values: dict[tuple[tuple[str, ...], str], dict] = {} - - async def get_item(self, namespace, key): - value = self.values.get((tuple(namespace), key)) - return {"value": value} if value is not None else None - - async def put_item(self, namespace, key, value): - self.values[(tuple(namespace), key)] = value - - -class FakeCrons: - def __init__(self, existing: list[dict] | None = None) -> None: - self.existing = existing or [] - self.created: list[dict] = [] - self.deleted: list[str] = [] - - async def search(self, **kwargs): - return self.existing - - async def create(self, assistant_id, **kwargs): - self.created.append({"assistant_id": assistant_id, **kwargs}) - return {"cron_id": "cron-new"} - - async def delete(self, cron_id): - self.deleted.append(cron_id) - - -@pytest.fixture(autouse=True) -def _reset_cron_memo(): - usage_snapshot_cron._registered_cron_id = None - yield - usage_snapshot_cron._registered_cron_id = None - - -class FakeRuns: - def __init__(self) -> None: - self.created: list[tuple] = [] - - async def create(self, thread, assistant_id, **kwargs): - self.created.append((thread, assistant_id, kwargs)) - return {"run_id": "run-1"} - - -class FakeClient: - def __init__(self, store=None, crons=None, runs=None): - self.store = store or FakeStore() - self.crons = crons or FakeCrons() - self.runs = runs or FakeRuns() - - -@pytest.mark.asyncio -async def test_builder_refreshes_all_periods(monkeypatch): - calls: list[tuple[str, str]] = [] - - async def fake_usage(period): - calls.append(("usage", period)) - - async def fake_reviewer(period): - calls.append(("reviewer", period)) - - monkeypatch.setattr(usage_snapshot, "refresh_usage_leaderboard_cache", fake_usage) - monkeypatch.setattr(usage_snapshot, "refresh_reviewer_stats_cache", fake_reviewer) - monkeypatch.setenv("USAGE_SNAPSHOT_CRON_ENABLED", "true") - - result = await usage_snapshot._build({}, {}) - - assert result["result"]["status"] == "ok" - assert ("usage", "7d") in calls - assert ("reviewer", "all") in calls - assert len(calls) == 6 - - -@pytest.mark.asyncio -async def test_builder_kill_switch(monkeypatch): - monkeypatch.setenv("USAGE_SNAPSHOT_CRON_ENABLED", "false") - result = await usage_snapshot._build({}, {}) - assert result["result"]["status"] == "disabled" - - -@pytest.mark.asyncio -async def test_builder_partial_on_failure(monkeypatch): - async def fake_usage(period): - raise RuntimeError("boom") - - async def fake_reviewer(period): - return None - - monkeypatch.setattr(usage_snapshot, "refresh_usage_leaderboard_cache", fake_usage) - monkeypatch.setattr(usage_snapshot, "refresh_reviewer_stats_cache", fake_reviewer) - monkeypatch.setenv("USAGE_SNAPSHOT_CRON_ENABLED", "true") - - result = await usage_snapshot._build({}, {}) - assert result["result"]["status"] == "partial" - assert len(result["result"]["failed"]) == 3 - - -@pytest.mark.asyncio -async def test_cron_registration_idempotent_and_creates_once(monkeypatch): - client = FakeClient() - monkeypatch.setattr(usage_snapshot_cron, "_client", lambda: client) - - first = await usage_snapshot_cron.ensure_usage_snapshot_cron() - assert first == "cron-new" - assert len(client.crons.created) == 1 - created = client.crons.created[0] - assert created["input"] == {} - assert created["metadata"]["rev"] == usage_snapshot_cron._CRON_REV - - # Second call reads the persisted id, creates nothing new. - second = await usage_snapshot_cron.ensure_usage_snapshot_cron() - assert second == "cron-new" - assert len(client.crons.created) == 1 - - -@pytest.mark.asyncio -async def test_cron_reuses_existing_search_hit(monkeypatch): - rev = usage_snapshot_cron._CRON_REV - client = FakeClient( - crons=FakeCrons(existing=[{"cron_id": "cron-existing", "metadata": {"rev": rev}}]) - ) - monkeypatch.setattr(usage_snapshot_cron, "_client", lambda: client) - - cron_id = await usage_snapshot_cron.ensure_usage_snapshot_cron() - assert cron_id == "cron-existing" - assert client.crons.created == [] - - -@pytest.mark.asyncio -async def test_cron_reaps_duplicate_search_hits(monkeypatch): - rev = usage_snapshot_cron._CRON_REV - client = FakeClient( - crons=FakeCrons( - existing=[ - {"cron_id": "cron-a", "metadata": {"rev": rev}}, - {"cron_id": "cron-b", "metadata": {"rev": rev}}, - ] - ), - ) - monkeypatch.setattr(usage_snapshot_cron, "_client", lambda: client) - - cron_id = await usage_snapshot_cron.ensure_usage_snapshot_cron() - assert cron_id == "cron-a" - assert client.crons.deleted == ["cron-b"] - assert client.crons.created == [] - - -@pytest.mark.asyncio -async def test_cron_replaces_stale_rev(monkeypatch): - """Crons created without input (rev 1) fail at __start__ — reap and recreate.""" - client = FakeClient(crons=FakeCrons(existing=[{"cron_id": "cron-old"}])) - monkeypatch.setattr(usage_snapshot_cron, "_client", lambda: client) - - cron_id = await usage_snapshot_cron.ensure_usage_snapshot_cron() - assert cron_id == "cron-new" - assert client.crons.deleted == ["cron-old"] - assert client.crons.created[0]["input"] == {} - - -@pytest.mark.asyncio -async def test_stale_stored_cron_id_is_ignored(monkeypatch): - store = FakeStore() - store.values[(("agent_usage", "meta"), "cron")] = {"cron_id": "cron-old"} - client = FakeClient(store=store) - monkeypatch.setattr(usage_snapshot_cron, "_client", lambda: client) - - cron_id = await usage_snapshot_cron.ensure_usage_snapshot_cron() - assert cron_id == "cron-new" - - -@pytest.mark.asyncio -async def test_cron_memoizes_and_skips_loopback(monkeypatch): - store = FakeStore() - client = FakeClient(store=store) - monkeypatch.setattr(usage_snapshot_cron, "_client", lambda: client) - - await usage_snapshot_cron.ensure_usage_snapshot_cron() - - # After registration, a poisoned client proves no further loopback happens. - def _boom(): - raise AssertionError("ensure_usage_snapshot_cron should be memoized") - - monkeypatch.setattr(usage_snapshot_cron, "_client", _boom) - assert await usage_snapshot_cron.ensure_usage_snapshot_cron() == "cron-new" - - -@pytest.mark.asyncio -async def test_trigger_build_schedules_threadless_run(monkeypatch): - client = FakeClient() - monkeypatch.setattr(usage_snapshot_cron, "_client", lambda: client) - - result = await usage_snapshot_cron.trigger_usage_snapshot_build() - assert result["status"] == "scheduled" - assert client.runs.created == [(None, "usage_snapshot", {"input": {}})] - - -def test_lifespan_retains_no_background_task(): - """Regression guard for the #1434 bug: lifespan must not spawn a looping task.""" - from agent import webapp - - src = inspect.getsource(webapp.lifespan) - assert "create_task" not in src - assert "while True" not in src