From c74df10435f2d9431bb97c721f793863ac895753 Mon Sep 17 00:00:00 2001 From: Andrii Pasternak Date: Sun, 19 Apr 2026 01:19:14 +0100 Subject: [PATCH 1/3] test(cleanup): failing tests for #378 stale-slot race MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds TestProcessStaleSlotReclaims (8 scenarios) to test_watchdog_unit.py and tests/unit/test_schedule_status_observability.py (4 scenarios). Tests reproduce the race where cleanup service's Phase 3 marks executions FAILED with "Stale execution — slot TTL expired" while the task is still running on the agent, and the subsequent SUCCESS write overwrites FAILED. These tests fail against the current code and pass after the fix + log that follow in subsequent commits. Refs #378 Co-Authored-By: Claude Opus 4.7 (1M context) --- tests/test_watchdog_unit.py | 245 +++++++++++++++++ .../test_schedule_status_observability.py | 246 ++++++++++++++++++ 2 files changed, 491 insertions(+) create mode 100644 tests/unit/test_schedule_status_observability.py diff --git a/tests/test_watchdog_unit.py b/tests/test_watchdog_unit.py index a366530ca..360646834 100644 --- a/tests/test_watchdog_unit.py +++ b/tests/test_watchdog_unit.py @@ -887,3 +887,248 @@ def test_works_without_client(self, mock_queue_fn, mock_slot_fn, mock_db): service._get_execution_error.assert_not_called() error_arg = mock_db.mark_execution_failed_by_watchdog.call_args[0][1] assert error_arg == "cleanup reason only" + + +# --------------------------------------------------------------------------- +# Phase 3 slot reclaim re-verification tests (Issue #378) +# --------------------------------------------------------------------------- + +# Eager import so @patch("services.cleanup_service.*") decorators can resolve +# the module before test setup runs. +import services.cleanup_service # noqa: E402,F401 + + +class TestProcessStaleSlotReclaims: + """Tests for _process_stale_slot_reclaims() — Phase 3 slot cleanup with #378 race fix. + + The bug: cleanup service's Phase 3 sometimes marked executions FAILED with + "Stale execution — slot TTL expired" even though the task was still running + (or had just completed). The fix adds a just-in-time re-verify call to the + agent right before failing. On agent unreachable, we skip this cycle — the + 120-min Phase 1 stale cleanup is the backstop. + """ + + pytestmark = pytest.mark.unit + + PHANTOM_ERROR_PREFIX = "Stale execution — slot TTL expired" + + def _make_service(self): + from services.cleanup_service import CleanupService + return CleanupService() + + def _make_report(self): + from services.cleanup_service import CleanupReport + return CleanupReport() + + @patch("services.cleanup_service.httpx.AsyncClient") + @patch("services.cleanup_service.db") + def test_empty_reclaimed_is_noop(self, mock_db, mock_httpx): + """No reclaimed slots → method returns early without any calls.""" + service = self._make_service() + service._get_agent_running_ids = AsyncMock() + service._terminate_on_agent = AsyncMock() + + report = self._make_report() + asyncio.run(service._process_stale_slot_reclaims({}, set(), report)) + + service._get_agent_running_ids.assert_not_called() + service._terminate_on_agent.assert_not_called() + mock_db.fail_stale_slot_execution.assert_not_called() + # httpx.AsyncClient should not even be constructed + mock_httpx.assert_not_called() + assert report.stale_slot_executions == 0 + + @patch("services.cleanup_service.httpx.AsyncClient") + @patch("services.cleanup_service.db") + def test_skips_when_in_confirmed_running_ids(self, mock_db, mock_httpx): + """Phase 0 confirmed this exec as running → Phase 3 skips without + even calling fail_stale_slot_execution. Regression guard for #226.""" + mock_httpx.return_value.__aenter__ = AsyncMock( + return_value=AsyncMock() + ) + mock_httpx.return_value.__aexit__ = AsyncMock(return_value=False) + + service = self._make_service() + service._get_agent_running_ids = AsyncMock(return_value=set()) + service._terminate_on_agent = AsyncMock() + + reclaimed = {"agent-a": ["exec-1"]} + confirmed = {"exec-1"} + report = self._make_report() + + asyncio.run(service._process_stale_slot_reclaims(reclaimed, confirmed, report)) + + mock_db.fail_stale_slot_execution.assert_not_called() + service._terminate_on_agent.assert_not_called() + assert report.stale_slot_executions == 0 + + @patch("services.cleanup_service.httpx.AsyncClient") + @patch("services.cleanup_service.db") + def test_378_race_skips_when_reverify_shows_still_running(self, mock_db, mock_httpx): + """#378 core scenario: Phase 0 missed this exec (agent had just + finished handing it back), but just-in-time re-verify catches + the agent still has it → Phase 3 skips. No FAILED row written.""" + mock_httpx.return_value.__aenter__ = AsyncMock( + return_value=AsyncMock() + ) + mock_httpx.return_value.__aexit__ = AsyncMock(return_value=False) + + service = self._make_service() + # Re-verify returns a set containing the exec → still running on agent + service._get_agent_running_ids = AsyncMock(return_value={"exec-1"}) + service._terminate_on_agent = AsyncMock() + + reclaimed = {"agent-a": ["exec-1"]} + report = self._make_report() + + asyncio.run(service._process_stale_slot_reclaims(reclaimed, set(), report)) + + mock_db.fail_stale_slot_execution.assert_not_called() + service._terminate_on_agent.assert_not_called() + assert report.stale_slot_executions == 0 + + @patch("services.cleanup_service.httpx.AsyncClient") + @patch("services.cleanup_service.db") + def test_proceeds_to_fail_when_reverify_confirms_inactive(self, mock_db, mock_httpx): + """Re-verify returns set without the exec → agent confirms gone → + proceed to terminate + fail. Phantom-stale error message emitted.""" + mock_httpx.return_value.__aenter__ = AsyncMock( + return_value=AsyncMock() + ) + mock_httpx.return_value.__aexit__ = AsyncMock(return_value=False) + + mock_db.fail_stale_slot_execution.return_value = True + + service = self._make_service() + service._get_agent_running_ids = AsyncMock(return_value=set()) + service._terminate_on_agent = AsyncMock(return_value=True) + + reclaimed = {"agent-a": ["exec-1"]} + report = self._make_report() + + asyncio.run(service._process_stale_slot_reclaims(reclaimed, set(), report)) + + service._terminate_on_agent.assert_called_once_with(ANY, "agent-a", "exec-1") + mock_db.fail_stale_slot_execution.assert_called_once() + call_kwargs = mock_db.fail_stale_slot_execution.call_args.kwargs + assert call_kwargs["execution_id"] == "exec-1" + assert self.PHANTOM_ERROR_PREFIX in call_kwargs["error"] + assert report.stale_slot_executions == 1 + + @patch("services.cleanup_service.httpx.AsyncClient") + @patch("services.cleanup_service.db") + def test_skips_when_agent_unreachable(self, mock_db, mock_httpx): + """Re-verify returns None (agent unreachable) → skip this cycle. + Phase 1's 120-min stale cleanup is the backstop for truly stuck + agents — we do NOT maintain cross-cycle defer state because + cleanup_stale_slots removes reclaimed IDs from Redis permanently.""" + mock_httpx.return_value.__aenter__ = AsyncMock( + return_value=AsyncMock() + ) + mock_httpx.return_value.__aexit__ = AsyncMock(return_value=False) + + service = self._make_service() + service._get_agent_running_ids = AsyncMock(return_value=None) + service._terminate_on_agent = AsyncMock() + + reclaimed = {"agent-a": ["exec-1"]} + report = self._make_report() + + asyncio.run(service._process_stale_slot_reclaims(reclaimed, set(), report)) + + mock_db.fail_stale_slot_execution.assert_not_called() + service._terminate_on_agent.assert_not_called() + assert report.stale_slot_executions == 0 + + @patch("services.cleanup_service.httpx.AsyncClient") + @patch("services.cleanup_service.db") + def test_per_agent_batching_single_reverify_call(self, mock_db, mock_httpx): + """Two stale slots on the same agent → _get_agent_running_ids + called exactly once for that agent (per-agent batching via + asyncio.gather, not per-execution).""" + mock_httpx.return_value.__aenter__ = AsyncMock( + return_value=AsyncMock() + ) + mock_httpx.return_value.__aexit__ = AsyncMock(return_value=False) + + mock_db.fail_stale_slot_execution.return_value = True + + service = self._make_service() + service._get_agent_running_ids = AsyncMock(return_value=set()) # both inactive + service._terminate_on_agent = AsyncMock(return_value=True) + + reclaimed = {"agent-a": ["exec-1", "exec-2"]} + report = self._make_report() + + asyncio.run(service._process_stale_slot_reclaims(reclaimed, set(), report)) + + # One re-verify call for the agent, not two + assert service._get_agent_running_ids.call_count == 1 + # Both executions failed + assert mock_db.fail_stale_slot_execution.call_count == 2 + assert report.stale_slot_executions == 2 + + @patch("services.cleanup_service.httpx.AsyncClient") + @patch("services.cleanup_service.db") + def test_multi_agent_reverify_dispatched_in_parallel(self, mock_db, mock_httpx): + """Two different agents → both re-verify calls dispatched via + asyncio.gather. Verified by call count — both agents queried + regardless of order.""" + mock_httpx.return_value.__aenter__ = AsyncMock( + return_value=AsyncMock() + ) + mock_httpx.return_value.__aexit__ = AsyncMock(return_value=False) + + mock_db.fail_stale_slot_execution.return_value = True + + service = self._make_service() + # Both agents report their execs as inactive — simple fail path + service._get_agent_running_ids = AsyncMock(return_value=set()) + service._terminate_on_agent = AsyncMock(return_value=True) + + reclaimed = {"agent-a": ["exec-a1"], "agent-b": ["exec-b1"]} + report = self._make_report() + + asyncio.run(service._process_stale_slot_reclaims(reclaimed, set(), report)) + + # Both agents queried + assert service._get_agent_running_ids.call_count == 2 + called_agents = { + call.args[1] for call in service._get_agent_running_ids.call_args_list + } + assert called_agents == {"agent-a", "agent-b"} + assert mock_db.fail_stale_slot_execution.call_count == 2 + assert report.stale_slot_executions == 2 + + @patch("services.cleanup_service.httpx.AsyncClient") + @patch("services.cleanup_service.db") + def test_one_agent_raises_others_proceed(self, mock_db, mock_httpx): + """One agent's re-verify raises → asyncio.gather(return_exceptions=True) + captures it → that agent's execs are skipped as unreachable; + other agents proceed normally.""" + mock_httpx.return_value.__aenter__ = AsyncMock( + return_value=AsyncMock() + ) + mock_httpx.return_value.__aexit__ = AsyncMock(return_value=False) + + mock_db.fail_stale_slot_execution.return_value = True + + service = self._make_service() + + async def flaky(_client, name): + if name == "agent-a": + raise RuntimeError("boom") + return set() # agent-b: exec inactive + + service._get_agent_running_ids = AsyncMock(side_effect=flaky) + service._terminate_on_agent = AsyncMock(return_value=True) + + reclaimed = {"agent-a": ["exec-a1"], "agent-b": ["exec-b1"]} + report = self._make_report() + + asyncio.run(service._process_stale_slot_reclaims(reclaimed, set(), report)) + + # Only agent-b's exec was failed; agent-a treated as unreachable + mock_db.fail_stale_slot_execution.assert_called_once() + assert mock_db.fail_stale_slot_execution.call_args.kwargs["execution_id"] == "exec-b1" + assert report.stale_slot_executions == 1 diff --git a/tests/unit/test_schedule_status_observability.py b/tests/unit/test_schedule_status_observability.py new file mode 100644 index 000000000..04f1d392e --- /dev/null +++ b/tests/unit/test_schedule_status_observability.py @@ -0,0 +1,246 @@ +""" +Schedule status observability log (Issue #378) + +Regression test for the narrowly-scoped WARNING log in +`db.schedules.ScheduleOperations.update_execution_status`: when a row whose +error matches the Phase-3 phantom-stale pattern is overwritten by SUCCESS, +we emit a log line tagged "residual race condition (#378)" so we can +observe residual races in production without changing update semantics. + +Scoped to the stale-slot error pattern so other legitimate FAILED→SUCCESS +transitions (startup recovery, Phase 0 auto-terminate, Phase 1 stale +cleanup) do NOT misfire the log. + +Covered scenarios: +1. FAILED row with stale-slot pattern → SUCCESS → log emitted +2. FAILED row with a DIFFERENT error → SUCCESS → log NOT emitted +3. RUNNING row → SUCCESS → log NOT emitted (happy path, no prior failure) +4. FAILED row with stale-slot pattern → FAILED (same status) → log NOT emitted +""" + +from __future__ import annotations + +import logging +import sqlite3 +import sys +from datetime import datetime, timezone +from pathlib import Path +from unittest.mock import MagicMock + +import pytest + +# --------------------------------------------------------------------------- +# Bootstrap: make src/backend importable. Copy of the bootstrap from +# tests/unit/test_backlog.py so the same path-shadow issues are handled. +# --------------------------------------------------------------------------- + +_THIS = Path(__file__).resolve() +_BACKEND = _THIS.parent.parent.parent / "src" / "backend" +_BACKEND_STR = str(_BACKEND) +for _shadow in ("utils", "utils.api_client", "utils.assertions", "utils.cleanup"): + sys.modules.pop(_shadow, None) +while _BACKEND_STR in sys.path: + sys.path.remove(_BACKEND_STR) +sys.path.insert(0, _BACKEND_STR) + + +PHANTOM_ERROR_PATTERN = "Stale execution — slot TTL expired" +RESIDUAL_LOG_MARKER = "residual race condition (#378)" + + +@pytest.fixture +def tmp_db(tmp_path, monkeypatch): + """Minimal schedule_executions schema for update_execution_status.""" + db_path = tmp_path / "trinity.db" + monkeypatch.setenv("TRINITY_DB_PATH", str(db_path)) + + conn = sqlite3.connect(str(db_path)) + conn.row_factory = sqlite3.Row + cur = conn.cursor() + cur.execute( + """ + CREATE TABLE schedule_executions ( + id TEXT PRIMARY KEY, + schedule_id TEXT NOT NULL, + agent_name TEXT NOT NULL, + status TEXT NOT NULL, + started_at TEXT NOT NULL, + completed_at TEXT, + duration_ms INTEGER, + message TEXT NOT NULL, + response TEXT, + error TEXT, + triggered_by TEXT NOT NULL, + context_used INTEGER, + context_max INTEGER, + cost REAL, + tool_calls TEXT, + execution_log TEXT, + claude_session_id TEXT + ) + """ + ) + conn.commit() + conn.close() + + # Re-import modules that read DB_PATH at import time. + for mod in ("db.connection", "db.schedules"): + sys.modules.pop(mod, None) + + yield db_path + + +@pytest.fixture +def schedule_ops(tmp_db): + """Fresh ScheduleOperations bound to tmp_db.""" + from db.schedules import ScheduleOperations + + return ScheduleOperations(user_ops=MagicMock(), agent_ops=MagicMock()) + + +def _insert(tmp_db: Path, *, execution_id: str, status: str, error: str | None): + """Seed a schedule_executions row with a given status + error.""" + conn = sqlite3.connect(str(tmp_db)) + conn.execute( + "INSERT INTO schedule_executions " + "(id, schedule_id, agent_name, status, started_at, message, error, triggered_by) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?)", + ( + execution_id, + "sched-1", + "agent-a", + status, + datetime.now(timezone.utc).isoformat(), + "test message", + error, + "scheduler", + ), + ) + conn.commit() + conn.close() + + +def _get_status(tmp_db: Path, execution_id: str) -> str: + conn = sqlite3.connect(str(tmp_db)) + row = conn.execute( + "SELECT status FROM schedule_executions WHERE id = ?", + (execution_id,), + ).fetchone() + conn.close() + return row[0] if row else "" + + +class TestResidualRaceObservabilityLog: + """Issue #378: warn when SUCCESS overwrites a Phase-3 phantom-stale FAILED.""" + + pytestmark = pytest.mark.unit + + def test_logs_when_success_overwrites_phantom_stale_failed( + self, tmp_db, schedule_ops, caplog + ): + """Row FAILED with stale-slot pattern, then SUCCESS → WARNING log.""" + from models import TaskExecutionStatus + + _insert( + tmp_db, + execution_id="exec-378", + status=TaskExecutionStatus.FAILED, + error=f"{PHANTOM_ERROR_PATTERN} for agent 'agent-a', cleaned by cleanup service", + ) + + with caplog.at_level(logging.WARNING, logger="db.schedules"): + updated = schedule_ops.update_execution_status( + execution_id="exec-378", + status=TaskExecutionStatus.SUCCESS, + response="agent returned result", + ) + + assert updated is True + assert _get_status(tmp_db, "exec-378") == TaskExecutionStatus.SUCCESS + + matching = [r for r in caplog.records if RESIDUAL_LOG_MARKER in r.getMessage()] + assert len(matching) == 1, ( + f"Expected exactly one #378 residual-race log, got " + f"{len(matching)}. Messages: {[r.getMessage() for r in caplog.records]}" + ) + assert "exec-378" in matching[0].getMessage() + + def test_does_not_log_when_failed_error_is_from_other_cleanup_path( + self, tmp_db, schedule_ops, caplog + ): + """Codex Point 6: startup recovery / Phase 0 auto-terminate / Phase 1 + stale cleanup also write FAILED via unguarded update_execution_status. + Those FAILED→SUCCESS transitions must NOT trigger the #378 log.""" + from models import TaskExecutionStatus + + _insert( + tmp_db, + execution_id="exec-other", + status=TaskExecutionStatus.FAILED, + error="Execution auto-terminated after 16 minutes by watchdog " + "(exceeded timeout of 900s)", + ) + + with caplog.at_level(logging.WARNING, logger="db.schedules"): + updated = schedule_ops.update_execution_status( + execution_id="exec-other", + status=TaskExecutionStatus.SUCCESS, + response="late agent response", + ) + + assert updated is True + assert _get_status(tmp_db, "exec-other") == TaskExecutionStatus.SUCCESS + + matching = [r for r in caplog.records if RESIDUAL_LOG_MARKER in r.getMessage()] + assert matching == [], ( + "#378 log misfired on a non-stale-slot FAILED→SUCCESS transition. " + f"Messages: {[r.getMessage() for r in caplog.records]}" + ) + + def test_does_not_log_on_running_to_success_happy_path( + self, tmp_db, schedule_ops, caplog + ): + """The normal happy path (RUNNING → SUCCESS, no prior failure) must + not trigger the log.""" + from models import TaskExecutionStatus + + _insert( + tmp_db, + execution_id="exec-happy", + status=TaskExecutionStatus.RUNNING, + error=None, + ) + + with caplog.at_level(logging.WARNING, logger="db.schedules"): + updated = schedule_ops.update_execution_status( + execution_id="exec-happy", + status=TaskExecutionStatus.SUCCESS, + response="ok", + ) + + assert updated is True + assert _get_status(tmp_db, "exec-happy") == TaskExecutionStatus.SUCCESS + matching = [r for r in caplog.records if RESIDUAL_LOG_MARKER in r.getMessage()] + assert matching == [] + + def test_does_not_log_on_same_status_write(self, tmp_db, schedule_ops, caplog): + """FAILED → FAILED with stale-slot pattern → no log (not an overwrite + of FAILED by SUCCESS, just a re-write).""" + from models import TaskExecutionStatus + + _insert( + tmp_db, + execution_id="exec-same", + status=TaskExecutionStatus.FAILED, + error=f"{PHANTOM_ERROR_PATTERN} for agent 'agent-a', cleaned by cleanup service", + ) + + with caplog.at_level(logging.WARNING, logger="db.schedules"): + schedule_ops.update_execution_status( + execution_id="exec-same", + status=TaskExecutionStatus.FAILED, + error="re-fail", + ) + + matching = [r for r in caplog.records if RESIDUAL_LOG_MARKER in r.getMessage()] + assert matching == [] From e428fcbf32b0ba24068add4c067f507afcd7402c Mon Sep 17 00:00:00 2001 From: Andrii Pasternak Date: Sun, 19 Apr 2026 01:19:27 +0100 Subject: [PATCH 2/3] fix(cleanup): JIT re-verify before failing reclaimed slots (#378) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 3 (slot cleanup) previously marked reclaimed executions FAILED using only Phase 0's `confirmed_running_ids` snapshot. Between Phase 0's batch query and Phase 3's decision, an agent could drop a just-completed execution from its registry while the corresponding SUCCESS response was still in flight to the backend — Phase 3 would then write FAILED and be overwritten by the late SUCCESS, producing a phantom failure flash. This extracts the Phase 3 loop into `_process_stale_slot_reclaims` (mirrors the testability pattern of `_reconcile_orphaned_executions`) and adds a just-in-time re-verify call to each agent immediately before writing FAILED. Per-agent fan-out via `asyncio.gather` keeps worst-case Phase 3 wall-time at O(5s) regardless of fleet size (same pattern as Phase 0 at cleanup_service.py:258-263). On agent unreachable during re-verify: skip this cycle. No cross-cycle deferral state — `slot_service.cleanup_stale_slots` removes reclaimed IDs from Redis permanently (zremrangebyscore), so a deferred ID would never reappear to be re-checked. Transient unreachability is caught by Phase 0's orphan recovery on later cycles; truly stuck agents by Phase 1 (120-min stale cleanup). Refs #378 Co-Authored-By: Claude Opus 4.7 (1M context) --- src/backend/services/cleanup_service.py | 124 +++++++++++++++++++----- 1 file changed, 102 insertions(+), 22 deletions(-) diff --git a/src/backend/services/cleanup_service.py b/src/backend/services/cleanup_service.py index 7b1093681..bbdde9267 100644 --- a/src/backend/services/cleanup_service.py +++ b/src/backend/services/cleanup_service.py @@ -176,7 +176,8 @@ async def _run_cleanup_inner(self) -> CleanupReport: except Exception as e: logger.error(f"[Cleanup] Error marking stale activities: {e}") - # 3. Cleanup stale Redis slots and fail corresponding execution records (#219, #226) + # 3. Cleanup stale Redis slots and fail corresponding execution records + # (#219, #226, #378 — see _process_stale_slot_reclaims docstring). try: slot_service = get_slot_service() @@ -189,24 +190,112 @@ async def _run_cleanup_inner(self) -> CleanupReport: ) report.stale_slots = sum(len(ids) for ids in reclaimed.values()) - # Fail execution records whose slots were reclaimed, - # but skip IDs the watchdog confirmed as still running (#226). + await self._process_stale_slot_reclaims( + reclaimed, confirmed_running_ids, report + ) + except Exception as e: + logger.error(f"[Cleanup] Error cleaning stale slots: {e}") + + self.last_run_at = utc_now_iso() + self.last_report = report + + if report.total > 0: + logger.info(f"[Cleanup] Cycle complete: {report.to_dict()}") + + return report + + async def _process_stale_slot_reclaims( + self, + reclaimed: Dict[str, List[str]], + confirmed_running_ids: set, + report: CleanupReport, + ) -> None: + """Fail execution records whose slots were reclaimed, with just-in-time + re-verify to prevent phantom failures (#378). + + The bug: cleanup service's Phase 3 sometimes marked executions FAILED + with "Stale execution — slot TTL expired" even though the task was + still running (agent had just dropped it from its registry after + completion, so Phase 0's batch query missed it). The SUCCESS response + then arrived after Phase 3 already wrote FAILED — user saw the flip. + + The fix: + - Do a just-in-time re-verify with each agent RIGHT BEFORE writing + FAILED, closing the window between Phase 0 and Phase 3. + - Parallel fan-out via asyncio.gather (mirrors Phase 0 pattern at + _reconcile_orphaned_executions). + - On agent unreachable: skip this cycle. Do NOT accumulate cross-cycle + state — slot_service.cleanup_stale_slots removes reclaimed IDs from + Redis permanently, so the same ID will not reappear in a later + cycle. The 120-min Phase 1 stale cleanup is the backstop for truly + stuck agents. + """ + if not reclaimed: + return + + agent_names = list(reclaimed.keys()) + async with httpx.AsyncClient(timeout=WATCHDOG_HTTP_TIMEOUT) as client: + results = await asyncio.gather( + *(self._get_agent_running_ids(client, name) for name in agent_names), + return_exceptions=True, + ) + per_agent_running: Dict[str, Optional[set]] = {} + for name, result in zip(agent_names, results): + if isinstance(result, BaseException): + logger.warning( + f"[Cleanup] Phase 3 re-verify failed for '{name}': {result}" + ) + per_agent_running[name] = None + else: + # result is Optional[set] here after the BaseException branch + per_agent_running[name] = result + for agent_name, execution_ids in reclaimed.items(): + running_ids = per_agent_running.get(agent_name) + for execution_id in execution_ids: + # #226: Phase 0 already confirmed this exec as running — + # trust it to save an HTTP call. if execution_id in confirmed_running_ids: logger.info( - f"[Cleanup] Skipping execution {execution_id} for agent " - f"'{agent_name}' — watchdog confirmed still running" + f"[Cleanup] Skipping {execution_id} for '{agent_name}' " + f"— watchdog confirmed still running" + ) + continue + + # Just-in-time re-verify interpretation + if running_ids is None: + # Agent unreachable during re-verify. Skip this cycle; + # Phase 1 (120-min stale cleanup) is the backstop. + logger.info( + f"[Cleanup] Skipping {execution_id} for '{agent_name}' " + f"— agent unreachable during re-verify (#378); " + f"Phase 1 stale cleanup is fallback" ) continue + + if execution_id in running_ids: + # #378: agent says this exec is still running — the + # slot TTL fired prematurely relative to the task. + # Skip; the task's own SUCCESS/FAILED write will + # land correctly later. + logger.info( + f"[Cleanup] Skipping {execution_id} for '{agent_name}' " + f"— re-verification shows still running (#378)" + ) + continue + + # Re-verify confirmed inactive → safe to fail. try: - # Issue #61: Attempt to terminate execution on agent before - # marking it failed. Best-effort — may fail if agent unreachable. + # Issue #61: best-effort terminate before marking failed. try: - async with httpx.AsyncClient(timeout=WATCHDOG_HTTP_TIMEOUT) as term_client: - await self._terminate_on_agent(term_client, agent_name, execution_id) + await self._terminate_on_agent( + client, agent_name, execution_id + ) except Exception as term_err: - logger.debug(f"[Cleanup] Could not terminate {execution_id}: {term_err}") + logger.debug( + f"[Cleanup] Could not terminate {execution_id}: {term_err}" + ) updated = db.fail_stale_slot_execution( execution_id=execution_id, @@ -215,22 +304,13 @@ async def _run_cleanup_inner(self) -> CleanupReport: if updated: report.stale_slot_executions += 1 logger.info( - f"[Cleanup] Failed execution {execution_id} for agent '{agent_name}' (slot reclaimed)" + f"[Cleanup] Failed execution {execution_id} for agent " + f"'{agent_name}' (slot reclaimed)" ) except Exception as e: logger.error( - f"[Cleanup] Error failing execution {execution_id} after slot reclaim: {e}" + f"[Cleanup] Error failing {execution_id} after slot reclaim: {e}" ) - except Exception as e: - logger.error(f"[Cleanup] Error cleaning stale slots: {e}") - - self.last_run_at = utc_now_iso() - self.last_report = report - - if report.total > 0: - logger.info(f"[Cleanup] Cycle complete: {report.to_dict()}") - - return report async def _reconcile_orphaned_executions(self) -> tuple[int, int, set]: """Reconcile DB execution state against agent process registries. From 5a006508d37faba197b2ab3716fc81d13411c722 Mon Sep 17 00:00:00 2001 From: Andrii Pasternak Date: Sun, 19 Apr 2026 01:19:36 +0100 Subject: [PATCH 3/3] fix(db): scoped observability log for #378 residual races + docs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Emit a WARNING log in `update_execution_status` when a SUCCESS write overwrites a row whose error matches the Phase-3 phantom-stale marker ("Stale execution — slot TTL expired"). Gives production signal if the primary fix misses a corner case, without changing update semantics — the agent's authoritative SUCCESS still wins. The pattern match is narrow on purpose: other legitimate cleanup paths (Phase 0 auto-terminate, Phase 1 stale cleanup, startup recovery) also write FAILED via unguarded `update_execution_status`. An unscoped log would misattribute those transitions to #378. Grep production logs with: docker logs trinity-backend | grep "residual race condition (#378)" Updates `docs/memory/feature-flows/cleanup-service.md` with a new "Phase 3 Slot Reclaim Re-verification (Issue #378)" section and adds an index row to `docs/memory/feature-flows.md`. Refs #378 Co-Authored-By: Claude Opus 4.7 (1M context) --- docs/memory/feature-flows.md | 1 + docs/memory/feature-flows/cleanup-service.md | 53 +++++++++++++++----- src/backend/db/schedules.py | 33 +++++++++++- 3 files changed, 72 insertions(+), 15 deletions(-) diff --git a/docs/memory/feature-flows.md b/docs/memory/feature-flows.md index 474862996..be5a10a19 100644 --- a/docs/memory/feature-flows.md +++ b/docs/memory/feature-flows.md @@ -12,6 +12,7 @@ | Date | ID | Feature | Flow | |------|-----|---------|------| | 2026-04-19 | #211 | Auto-propagate global GitHub PAT to running agents on update — per-agent PAT holders and agents without `GITHUB_PAT` in `.env` are skipped; delete does NOT propagate | [github-sync.md](feature-flows/github-sync.md), [platform-settings.md](feature-flows/platform-settings.md) | +| 2026-04-19 | #378 | Cleanup service Phase 3 just-in-time re-verify + parallel per-agent fan-out — eliminates phantom stale-slot failures for still-running tasks; adds residual-race observability log | [cleanup-service.md](feature-flows/cleanup-service.md) | | 2026-04-18 | DOCS-QA-001 | Trinity Docs Q&A — public Vertex AI Search endpoint + in-app floating help widget (#391) | [trinity-docs-qa.md](feature-flows/trinity-docs-qa.md) | | 2026-04-17 | #376 | Proactive messaging UI toggle — SharingPanel shows allow_proactive switch per shared user | [proactive-messaging.md](feature-flows/proactive-messaging.md), [agent-sharing.md](feature-flows/agent-sharing.md) | | 2026-04-16 | #321 | Proactive agent messaging — agents send messages to users by verified email via Telegram/Slack/web | [proactive-messaging.md](feature-flows/proactive-messaging.md) | diff --git a/docs/memory/feature-flows/cleanup-service.md b/docs/memory/feature-flows/cleanup-service.md index 1da5c8715..e4d3b95a9 100644 --- a/docs/memory/feature-flows/cleanup-service.md +++ b/docs/memory/feature-flows/cleanup-service.md @@ -96,22 +96,19 @@ Seven sequential operations, each wrapped in individual try/except. Watchdog run ``` Calls `DatabaseManager.mark_stale_activities_failed()` which delegates to `ActivityOperations.mark_stale_activities_failed()`. -5. **Cleanup stale Redis slots and fail execution records** (lines 123-148, Issue #219, #226, #61) +5. **Cleanup stale Redis slots and fail execution records** (Issues #219, #226, #61, #378) ```python slot_service = get_slot_service() agent_timeouts = db.get_all_execution_timeouts() # #226: per-agent TTL reclaimed = await slot_service.cleanup_stale_slots(agent_timeouts=agent_timeouts) report.stale_slots = sum(len(ids) for ids in reclaimed.values()) - # Fail execution records whose slots were reclaimed, skip confirmed running (#226) - for agent_name, execution_ids in reclaimed.items(): - for execution_id in execution_ids: - if execution_id in confirmed_running_ids: - continue # Watchdog verified still running - # Issue #61: Attempt to terminate process before marking failed (best-effort) - await self._terminate_on_agent(client, agent_name, execution_id) - db.fail_stale_slot_execution(execution_id, error=...) + # #378: delegates to _process_stale_slot_reclaims which re-verifies + # each agent just-in-time before writing FAILED + await self._process_stale_slot_reclaims( + reclaimed, confirmed_running_ids, report + ) ``` - Calls `SlotService.cleanup_stale_slots()` with per-agent timeouts (#226). The service scans all `agent:slots:*` keys, computes each agent's TTL as `timeout_seconds + 5 min buffer` (or default 20 min if no timeout configured), removes entries older than that TTL, and returns a dict mapping agent names to reclaimed execution IDs. **Issue #61**: Before failing the execution, the cleanup service attempts to terminate any orphaned Claude process on the agent (best-effort, failures logged). The cleanup service then fails the corresponding `schedule_executions` DB records using a guarded update (`WHERE status = 'running'`), skipping any IDs the watchdog confirmed as still running. + Calls `SlotService.cleanup_stale_slots()` with per-agent timeouts (#226). The service scans all `agent:slots:*` keys, computes each agent's TTL as `timeout_seconds + 5 min buffer` (or default 20 min if no timeout configured), removes entries older than that TTL, and returns a dict mapping agent names to reclaimed execution IDs. Phase 3 is then implemented by `_process_stale_slot_reclaims()` — see [Phase 3 Slot Reclaim Re-verification](#phase-3-slot-reclaim-re-verification-issue-378) below. ### Watchdog Reconciliation (Issue #129) @@ -159,6 +156,35 @@ Shared DRY helper for both orphan recovery and auto-terminate: #### `_terminate_on_agent(client, agent_name, execution_id)` → `bool` `POST http://agent-{name}:8000/api/executions/{id}/terminate`. Returns True if HTTP 2xx (agent confirmed termination), False otherwise. Callers only proceed with DB/resource cleanup on success — failed terminations are deferred to the 120-min stale cleanup safety net. +### Phase 3 Slot Reclaim Re-verification (Issue #378) + +Before this fix, Phase 3 could mark an execution `FAILED` with "Stale execution — slot TTL expired" while the task was actually still running on the agent (agent had just dropped it from its registry before Phase 0's batch query, so `confirmed_running_ids` missed it). The agent's authoritative `SUCCESS` response then arrived seconds later and overwrote `FAILED` → `SUCCESS`, causing a phantom failure flash in the UI. + +#### `_process_stale_slot_reclaims(reclaimed, confirmed_running_ids, report)` → `None` + +Replaces the inline Phase 3 loop. Extracted as its own method for direct unit testing (mirrors `_reconcile_orphaned_executions` testability pattern). Key additions over the old inline loop: + +1. **Parallel per-agent re-verify fan-out** — one `GET /api/executions/running` call per agent (not per-execution), dispatched concurrently via `asyncio.gather(..., return_exceptions=True)`. Mirrors Phase 0's pattern. Worst-case Phase 3 wall-time goes from O(N_agents × 5s) serial to O(5s) parallel when agents are slow. +2. **Just-in-time re-verify** — the agent is re-queried as close as possible to the `fail_stale_slot_execution` write, minimizing the race window that Phase 0's earlier batch query leaves open. +3. **Per-execution decision matrix**: + +| Phase 0 said running? | Re-verify says? | Action | +|---|---|---| +| Yes (`confirmed_running_ids`) | — | **SKIP** (trust Phase 0, save an HTTP call) | +| No | Agent unreachable (None) | **SKIP this cycle** — Phase 1 (120-min stale cleanup) is the backstop | +| No | Agent says still running | **SKIP** — #378 race closed; agent's own SUCCESS write will land correctly | +| No | Agent says not running | **FAIL** — terminate (best-effort, #61) + `fail_stale_slot_execution` with phantom-stale error | + +4. **No cross-cycle state** — `slot_service.cleanup_stale_slots` removes reclaimed IDs from Redis permanently (`zremrangebyscore`), so a deferred ID cannot reappear in a later cycle's `reclaimed` dict. Any "retry on next cycle" state machine would be dead code. Transiently-unreachable agents are caught by Phase 0's orphan recovery on subsequent cycles (when the agent becomes reachable again) and by Phase 1's 120-min stale cleanup as a final backstop. + +#### Residual-race observability + +`db.schedules.update_execution_status` emits a narrowly-scoped `logger.warning` whenever a `SUCCESS` write overwrites a row whose existing error matches the `_STALE_SLOT_ERROR_PATTERN = "Stale execution — slot TTL expired"` marker. Purely observational — update semantics are unchanged (the agent's SUCCESS still wins). The pattern match prevents misattribution of other legitimate FAILED→SUCCESS transitions (Phase 0 auto-terminate, Phase 1 stale cleanup, startup recovery) to #378. Grep with: + +```bash +docker logs trinity-backend | grep "residual race condition (#378)" +``` + ### Startup Loop (`_cleanup_loop`) ``` @@ -475,8 +501,8 @@ This is a purely backend service. The only "UI" is the two admin API endpoints u | File | Role | |------|------| -| `src/backend/services/cleanup_service.py` | Service class, watchdog reconciliation, and global instance | -| `src/backend/db/schedules.py` | `get_running_executions_with_agent_info()` (Issue #129), `mark_execution_failed_by_watchdog()` (Issue #129), `mark_stale_executions_failed()`, `mark_execution_dispatched()`, `mark_no_session_executions_failed()` (Issue #106), `fail_stale_slot_execution()` (Issue #219), `finalize_orphaned_skipped_executions()` (Issue #106) | +| `src/backend/services/cleanup_service.py` | Service class, watchdog reconciliation, Phase 3 re-verification (Issue #378), and global instance | +| `src/backend/db/schedules.py` | `get_running_executions_with_agent_info()` (Issue #129), `mark_execution_failed_by_watchdog()` (Issue #129), `mark_stale_executions_failed()`, `mark_execution_dispatched()`, `mark_no_session_executions_failed()` (Issue #106), `fail_stale_slot_execution()` (Issue #219), `finalize_orphaned_skipped_executions()` (Issue #106), residual-race observability log in `update_execution_status()` (Issue #378) | | `src/backend/db/activities.py` | `mark_stale_activities_failed()` | | `src/backend/database.py` | Delegation methods on DatabaseManager | | `src/backend/services/slot_service.py` | `cleanup_stale_slots()` Redis cleanup, returns reclaimed IDs (Issue #219), `release_slot()` used by watchdog | @@ -487,4 +513,5 @@ This is a purely backend service. The only "UI" is the two admin API endpoints u | `docker/base-image/agent_server/services/process_registry.py` | `get_last_error()` method scans log buffer for errors (Issue #286) | | `tests/test_cleanup_service.py` | API integration tests for cleanup (Issue #106) | | `tests/test_watchdog.py` | API integration tests for watchdog fields (Issue #129) | -| `tests/test_watchdog_unit.py` | Unit tests for watchdog reconciliation logic (Issue #129), error context tests (Issue #286) | +| `tests/test_watchdog_unit.py` | Unit tests for watchdog reconciliation logic (Issue #129), error context tests (Issue #286), Phase 3 re-verify tests (Issue #378) | +| `tests/unit/test_schedule_status_observability.py` | Residual-race observability log tests (Issue #378) | diff --git a/src/backend/db/schedules.py b/src/backend/db/schedules.py index 9f04f9abe..ce8d396c8 100644 --- a/src/backend/db/schedules.py +++ b/src/backend/db/schedules.py @@ -21,6 +21,12 @@ logger = logging.getLogger(__name__) +# #378: Error-message marker written by cleanup_service._process_stale_slot_reclaims +# when Phase 3 fails an execution. Used to scope the residual-race WARNING log +# below so it doesn't misfire on other legitimate FAILED→SUCCESS transitions +# (e.g. Phase 0 auto-terminate, Phase 1 stale cleanup, startup recovery). +_STALE_SLOT_ERROR_PATTERN = "Stale execution — slot TTL expired" + class ScheduleOperations: """Schedule and execution database operations.""" @@ -909,12 +915,35 @@ def update_execution_status( with get_db_connection() as conn: cursor = conn.cursor() - # Get started_at for duration calculation - cursor.execute("SELECT started_at FROM schedule_executions WHERE id = ?", (execution_id,)) + # Get started_at for duration calculation and current status/error + # for #378 residual-race observability (see log below). + cursor.execute( + "SELECT started_at, status, error FROM schedule_executions WHERE id = ?", + (execution_id,), + ) row = cursor.fetchone() if not row: return False + # #378: warn when SUCCESS overwrites a Phase-3 phantom-stale FAILED. + # This lets us observe residual races in production without + # changing update semantics (agent's response still wins). Scoped + # to the stale-slot error pattern so other legitimate FAILED→SUCCESS + # transitions (Phase 0/1 recovery, startup recovery) don't misfire. + current_status = row["status"] if "status" in row.keys() else None + current_error = row["error"] if "error" in row.keys() else None + if ( + status == TaskExecutionStatus.SUCCESS + and current_status == TaskExecutionStatus.FAILED + and current_error + and _STALE_SLOT_ERROR_PATTERN in current_error + ): + logger.warning( + f"[DB] SUCCESS overwrote Phase-3 stale-slot FAILED for execution " + f"{execution_id} — residual race condition (#378). Prior error: " + f"{current_error[:200]}" + ) + # Use parse_iso_timestamp to handle both 'Z' and non-'Z' timestamps started_at = parse_iso_timestamp(row["started_at"]) completed_at = parse_iso_timestamp(utc_now_iso())