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())
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.
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 == []