From 94f061f10c393167749600b948370bcff9a242c2 Mon Sep 17 00:00:00 2001 From: Oranje AI <293843428+oranjeai@users.noreply.github.com> Date: Tue, 4 Aug 2026 02:29:53 +1200 Subject: [PATCH] plans: reconcile 'implementing' plans orphaned by a daemon restart A plan in `implementing` asserts a live in-process implementation run, but that run is a bare `asyncio.create_task`. A SIGKILL/OOM/daemon restart destroys it and nothing reconciles plans at startup, so the row keeps asserting an obligation nobody holds. The plan is then terminally wedged: re-approval is refused (both surfaces gate on `pending`), `plan_propose` is refused forever for that task (`get_pending_plan_task_ids` counts `implementing`), and no agent tool or UI flow returns it to `pending` -- the only escape is a hand-crafted PATCH. Workflow runs made the same RAM-only trade and paired it with a startup recovery pass (`WorkflowRunService.start`), whose module docstring states the doctrine: runs do not survive a restart, so the pass marks orphans failed and notifies. Sessions have one too (`recover_orphaned_sessions`). Plans were the holdout. This adds the same guarantee: - `get_implementing_plans` / `fail_orphaned_plan` (a status-only CAS) in `db/plans.py`; - `recover_orphaned_plans` in `plan_service.py`, sending one aggregate notification keyed on what it actually flipped; - one call in the gateway lifespan. `failed`, not back to `pending`: it is already what both approval surfaces write when `engine.run` raises, and it unblocks `plan_propose` without blindly re-running a plan whose partial effects may already exist. No migration, no new status, no frontend diff (`STATUS_STYLES` already carries `failed`). Plans whose implementation session is enrolled for resume and still eligible are left `implementing`. That exclusion is correctness, not caution: `nerve restart --resume ` is a supported action, and a resumed session completes through `task_done`, which closes only an `implementing` plan -- sweeping it would leave the resumed session unable to ever close its own plan, and would immediately allow a duplicate plan while it still works. Eligibility is re-checked against the engine's own four skip predicates, since a session the engine will skip never resumes. The pass runs before `cron.start()`: cron's catch-up can dispatch a planner run whose `plan_propose` would read a stale `implementing` row and permanently skip the task. Two error cases go opposite ways. A missing resume queue is a definite answer from its sole writer (the CLI appends before triggering the restart), so sweep. An OSError is not an answer -- our read and the engine's happen at different instants, so a transient error here can succeed there -- so skip the sweep entirely and let the next restart retry. The queue is only ever read, never drained; the engine's resume task is the sole drainer and runs later. Out of scope, deliberately: a plan reconciled to `failed` is not re-adopted if its session is later resumed (the task still completes; only the plan's label differs -- re-adopting needs durable recovery provenance, i.e. a schema change); settling a plan whose resumed turn ends without `task_done`; the task/plan status mismatch on a late approval failure; the `PlansPage` filter list lacking `failed` and `done` (pre-existing). Tests: `tests/test_plan_restart_recovery.py` (18). Two are behavioural witnesses -- against the pre-change tree the seeded plan is still `implementing` after startup, and still `implementing` when cron starts. --- docs/plans.md | 10 + nerve/agent/plan_service.py | 166 ++++++- nerve/db/plans.py | 32 ++ nerve/gateway/server.py | 14 + tests/test_plan_restart_recovery.py | 706 ++++++++++++++++++++++++++++ 5 files changed, 927 insertions(+), 1 deletion(-) create mode 100644 tests/test_plan_restart_recovery.py diff --git a/docs/plans.md b/docs/plans.md index 46497ded..e3b0c7f3 100644 --- a/docs/plans.md +++ b/docs/plans.md @@ -101,6 +101,16 @@ User reviews (via /plans UI or chat tools) | `implementing` | Implementation session is running | | `declined` | Rejected by user | | `superseded` | Replaced by a newer version | +| `failed` | Implementation did not complete (the run raised, or a daemon restart interrupted it) | + +**Restarts.** Plans do **not** survive a daemon restart. `implementing` asserts a +live in-process implementation run, so on startup a recovery pass marks any +orphaned `implementing` plan `failed` ("interrupted by nerve restart") and sends +a high-priority notification. Nothing resumes on its own: the work may be +partially done, so review the task and propose a fresh version if it is still +needed. Plans whose implementation session was enrolled via +`nerve restart --resume` are left `implementing`, because that session resumes +and closes its own plan. ## Cron Job diff --git a/nerve/agent/plan_service.py b/nerve/agent/plan_service.py index a00ec6db..4801e678 100644 --- a/nerve/agent/plan_service.py +++ b/nerve/agent/plan_service.py @@ -1,4 +1,4 @@ -"""Shared plan-revision dispatch logic. +"""Shared plan-revision dispatch and restart-recovery logic. Both the HTTP route ``/api/plans/{plan_id}/revise`` and the MCP tool ``plan_revise`` need to do the same thing: validate the plan, persist @@ -11,6 +11,13 @@ Keeping this in one place prevents the two surfaces from drifting apart again. The HTTP route translates the exceptions raised here into HTTP status codes; the MCP tool translates them into user-facing text. + +Plans do not survive a daemon restart: ``status='implementing'`` asserts +an in-process implementation obligation, but that obligation is a bare +``asyncio`` task, so a SIGKILL/OOM/restart destroys it and leaves the row +claiming a run that no longer exists. :func:`recover_orphaned_plans` is +the startup reconciliation pass for exactly that, mirroring the one +workflow runs already have. """ from __future__ import annotations @@ -19,6 +26,8 @@ import logging from typing import TYPE_CHECKING +from nerve.config import RESUME_QUEUE_FILE + if TYPE_CHECKING: from nerve.agent.engine import AgentEngine from nerve.db import Database @@ -28,6 +37,11 @@ _FALLBACK_PLANNER_SESSION = "cron:task-planner" +# Reason a restart-reconciled plan reads ``failed``. ``plans`` has no error +# column (and this change deliberately adds no migration), so the reason lives +# in the log line and the notification body. +_PLAN_RESTART_ERROR = "interrupted by nerve restart" + # The prompt template lives here so future tweaks (model selection, # wording, calling conventions) happen in exactly one place. @@ -162,3 +176,153 @@ async def request_plan_revision( "session_id": session_id, "status": "revision_requested", } + + +# --------------------------------------------------------------------------- # +# Restart recovery # +# --------------------------------------------------------------------------- # + + +def _enrolled_resume_session_ids() -> set[str] | None: + """Session ids ``nerve restart --resume`` enrolled, or None if unreadable. + + Parsed exactly as :meth:`AgentEngine.resume_enrolled_sessions` parses the + same file, so the two readers can never disagree about what is enrolled. + + Never drains the file: the engine's resume task is the sole drainer and it + runs later in startup, so consuming it here would silently cancel every + enrolled resume. + + The two error cases go opposite ways. A missing file is a definite answer + from the sole writer (the CLI appends *before* triggering the restart): + nothing was enrolled, so return an empty set. An ``OSError`` is not an + answer, so return None and let the caller skip the sweep -- our read and + the engine's happen at different instants, so a transient error here can + succeed there and resume a session whose plan we would have failed. + """ + try: + raw = RESUME_QUEUE_FILE.read_text() + except FileNotFoundError: + return set() + except OSError as e: + logger.error( + "Plan recovery: could not read resume queue %s: %s", RESUME_QUEUE_FILE, e, + ) + return None + return {sid for line in raw.splitlines() if (sid := line.strip())} + + +async def _resume_eligible(db: "Database", session_id: str) -> bool: + """Whether the engine would actually resume ``session_id``. + + Re-evaluates the four skip predicates in + :meth:`AgentEngine.resume_enrolled_sessions` (missing / archived / + satellite / no SDK session to resume). Enrollment alone is not evidence of + a live obligation: a session the engine will skip never resumes, so + sparing its plan would wedge the plan for nothing. + + All four are pure reads of one ``sessions`` row, so this costs one lookup + per enrolled id and adds no ordering constraint. + """ + from nerve.agent.sessions import SessionStatus + + session = await db.get_session(session_id) + if not session: + return False + if session.get("status") == SessionStatus.ARCHIVED.value: + return False + if session.get("source") == "external": + return False + if not session.get("sdk_session_id"): + return False + return True + + +async def recover_orphaned_plans(db: "Database", notification_service=None) -> int: + """Reconcile plans orphaned by a daemon restart; return how many flipped. + + A plan in ``implementing`` asserts a live in-process implementation run. + The only things that can move it off that status are the ``asyncio`` task + approval spawned and a later ``task_done``, so a restart leaves the row + asserting an obligation nobody holds: re-approval is refused (both + surfaces gate on ``pending``) and ``plan_propose`` is refused forever for + that task, with no agent tool or UI flow able to recover it. + + Each orphan is CASed to ``failed`` -- the status both approval surfaces + already write when the implementation run does not complete -- which + unblocks ``plan_propose`` so the recovery route is "propose v+1" rather + than blindly replaying a stale plan over partial effects. + + Plans whose implementation session is enrolled for resume and still + eligible are left ``implementing``. That exclusion is correctness, not + caution: a resumed session completes through ``task_done``, which closes + only an ``implementing`` plan, so sweeping it would leave the resumed + session unable to ever close its own plan -- and would immediately let a + duplicate plan be proposed while it is still working. + + Must run before the cron service starts: cron's catch-up can dispatch a + planner run whose ``plan_propose`` would read a stale ``implementing`` row + and permanently skip the task. + + A plan reconciled to ``failed`` is not re-adopted if its session is later + resumed (the task still completes; only the plan's label reads ``failed``). + Distinguishing a restart-swept plan from a legitimately failed one needs + durable recovery provenance, i.e. a schema change this change avoids. + """ + enrolled = _enrolled_resume_session_ids() + if enrolled is None: + # Fail closed: leave every plan alone rather than risk failing one + # whose session the engine is about to resume. Costs exactly the + # status quo -- plans stay ``implementing`` and the next restart + # retries. + logger.error( + "Plan recovery skipped: resume queue unreadable. Plans stay " + "'implementing'; the next restart retries.", + ) + return 0 + + orphaned = await db.get_implementing_plans() + flipped: list[str] = [] + for plan in orphaned: + sid = plan.get("impl_session_id") + # Cheap membership test first, so the common case (nothing enrolled) + # never pays for a session lookup. A NULL owner can never match. + if sid and sid in enrolled and await _resume_eligible(db, sid): + logger.info( + "Plan recovery: plan %s left 'implementing' -- its session %s is " + "enrolled for resume", plan["id"], sid, + ) + continue + if await db.fail_orphaned_plan(plan["id"]): + flipped.append(plan["id"]) + + if flipped: + logger.info( + "Plan recovery: %d plan(s) marked failed (%s): %s", + len(flipped), _PLAN_RESTART_ERROR, ", ".join(flipped), + ) + if notification_service is not None: + # Keyed on what we actually flipped, not on what we read: a plan + # that legitimately left ``implementing`` between the read and the + # write (a concurrent ``task_done``) must not be reported as + # interrupted. + try: + await notification_service.send_notification( + session_id="system", + title="Plans interrupted by restart", + body=( + f"{len(flipped)} plan(s) were marked failed after a Nerve " + f"restart ({_PLAN_RESTART_ERROR}): {', '.join(flipped)}. " + "Their implementation sessions did not survive the " + "restart, so the work may be partially done. Review the " + "task, then propose a fresh plan version if it is still " + "needed." + ), + priority="high", + ) + except Exception: + # A notification failure must not abort recovery -- the rows are + # already reconciled, which is the part that matters. + logger.exception("Plan recovery: notification failed") + + return len(flipped) diff --git a/nerve/db/plans.py b/nerve/db/plans.py index a2150d09..a22a6977 100644 --- a/nerve/db/plans.py +++ b/nerve/db/plans.py @@ -82,3 +82,35 @@ async def get_pending_plan_task_ids(self) -> list[str]: "SELECT DISTINCT task_id FROM plans WHERE status IN ('pending', 'implementing')" ) as cursor: return [row[0] async for row in cursor] + + async def get_implementing_plans(self) -> list[dict]: + """All ``implementing`` plans, oldest first -- the restart-recovery input. + + Deliberately not ``list_plans(status="implementing")``: that helper caps + at ``limit=100``, so a larger backlog would silently under-recover. A + recovery pass must see every row, hence a dedicated unlimited query (the + same reason ``get_active_workflow_runs`` exists alongside + ``list_workflow_runs``). + """ + async with self.db.execute( + """SELECT * FROM plans WHERE status = 'implementing' + ORDER BY created_at ASC, id ASC""" + ) as cursor: + return [dict(row) async for row in cursor] + + async def fail_orphaned_plan(self, plan_id: str) -> bool: + """CAS a plan ``implementing -> failed``; True only if this call flipped it. + + The predicate is status-only on purpose. Keying it on ``impl_session_id`` + would never match the rows ``PATCH /api/plans/{id}`` can create, which + take an arbitrary status with no whitelist and so can be ``implementing`` + with a NULL owner -- SQL ``= ?`` bound to NULL matches nothing, so exactly + those rows would stay wedged forever. Status-only is sound because no + path moves a plan ``implementing -> implementing``: the writers of + ``implementing`` require ``pending`` first. + """ + result = await self._write( + "UPDATE plans SET status = 'failed' WHERE id = ? AND status = 'implementing'", + (plan_id,), + ) + return (result.rowcount or 0) == 1 diff --git a/nerve/gateway/server.py b/nerve/gateway/server.py index 8aea902e..3062836f 100644 --- a/nerve/gateway/server.py +++ b/nerve/gateway/server.py @@ -344,6 +344,20 @@ async def lifespan(app: FastAPI): await telegram_channel.start() logger.info("Telegram bot started") + # Reconcile plans orphaned by a restart. Like workflow runs, a plan hands + # its liveness to an in-process task, so 'implementing' rows survive a + # restart with no obligation behind them. Placed after notification wiring + # and the Telegram channel (so the alert can actually be delivered) and + # BEFORE cron starts: cron's catch-up pass can dispatch a planner run whose + # plan_propose would read a stale 'implementing' row and permanently skip + # the task. + try: + from nerve.agent.plan_service import recover_orphaned_plans + + await recover_orphaned_plans(db, notification_service) + except Exception as e: + logger.error("Plan recovery failed: %s", e) + # Start cron service global _cron_service cron_task = None diff --git a/tests/test_plan_restart_recovery.py b/tests/test_plan_restart_recovery.py new file mode 100644 index 00000000..dfe9a330 --- /dev/null +++ b/tests/test_plan_restart_recovery.py @@ -0,0 +1,706 @@ +"""Tests for the plan restart-recovery pass (``recover_orphaned_plans``). + +``plans.status = 'implementing'`` asserts a live in-process implementation run, +but that run is a bare ``asyncio`` task, so a SIGKILL/OOM/daemon restart leaves +the row asserting an obligation nobody holds: re-approval is refused (both +surfaces gate on ``pending``) and ``plan_propose`` is refused forever for that +task. These tests pin the startup pass that reconciles such rows to ``failed``, +mirroring the recovery pass workflow runs already have. + +The one piece of apparent extra machinery -- the resume-queue exclusion -- is +correctness, not caution: ``nerve restart --resume `` is a supported +operator action, and a resumed implementation session completes through +``task_done``, which closes only an ``implementing`` plan. Sweeping such a plan +would leave the resumed session unable to close it. T10-T13 and T15 pin that. +""" + +from __future__ import annotations + +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + + +# --------------------------------------------------------------------------- # +# Helpers # +# --------------------------------------------------------------------------- # + + +async def _recover(db, notification_service=None) -> int: + """Call the pass under test. + + IMPORTANT: The import is inside the function on purpose. A module-level import of a + symbol this change ADDS would make the whole module a collection error when + the file is run against the pre-change tree, which would hide the two + behavioural witnesses (``test_lifespan_recovers_an_orphaned_plan`` and + ``test_recovery_runs_before_cron_starts``): those must fail on plan state, + not on an ImportError. + """ + from nerve.agent.plan_service import recover_orphaned_plans + + return await recover_orphaned_plans(db, notification_service) + + +async def _seed_plan( + db, + plan_id: str, + status: str = "implementing", + impl_session_id: str | None = None, + task_id: str = "task-1", +) -> None: + """Create a plan row in an arbitrary status, with an optional owner session. + + ``create_plan`` always writes ``pending`` and never sets ``impl_session_id``, + so both are applied afterwards via ``update_plan`` (which is exactly how the + approval surfaces reach ``implementing`` too). + """ + await db.create_plan(plan_id, task_id, f"content of {plan_id}") + fields: dict = {"status": status} + if impl_session_id is not None: + fields["impl_session_id"] = impl_session_id + await db.update_plan(plan_id, **fields) + + +async def _seed_resumable_session(db, session_id: str) -> None: + """A session the engine's resume pass would accept: present, not archived, + not a satellite, with an SDK session to resume.""" + await db.create_session(session_id, source="web", status="running") + await db.update_session_metadata(session_id, {"sdk_session_id": f"sdk-{session_id}"}) + + +def _notif_stub() -> MagicMock: + return MagicMock(send_notification=AsyncMock(return_value="notif-1")) + + +def _queue(monkeypatch, tmp_path, *session_ids: str, raw: str | None = None): + """Point ``plan_service.RESUME_QUEUE_FILE`` at a temp file and fill it. + + IMPORTANT: Patching the module global is what matters, not ``NERVE_HOME``: + ``RESUME_QUEUE_FILE`` is evaluated at ``nerve.config`` import time, so an + env var set inside a test lands too late and the test would read the live + daemon's queue. + """ + q = tmp_path / "resume-queue" + if raw is not None: + q.write_text(raw) + elif session_ids: + q.write_text("".join(f"{sid}\n" for sid in session_ids)) + monkeypatch.setattr("nerve.agent.plan_service.RESUME_QUEUE_FILE", q) + return q + + +def _no_queue(monkeypatch, tmp_path): + """Point the queue at a path that does not exist (nothing was enrolled).""" + q = tmp_path / "absent-queue" + monkeypatch.setattr("nerve.agent.plan_service.RESUME_QUEUE_FILE", q) + return q + + +# --------------------------------------------------------------------------- # +# Store level # +# --------------------------------------------------------------------------- # + + +@pytest.mark.asyncio +async def test_get_implementing_plans_is_unlimited_and_ordered(db): + """T1: every ``implementing`` row, oldest first, with no limit. + + The 150-row fixture is the discriminator against + ``list_plans(status="implementing")``, whose ``limit=100`` would silently + under-recover a larger backlog. + """ + for i in range(150): + await _seed_plan(db, f"plan-impl-{i:03d}") + for other in ("pending", "done", "declined", "superseded", "failed"): + await _seed_plan(db, f"plan-{other}", status=other) + + rows = await db.get_implementing_plans() + + assert len(rows) == 150 + assert {r["status"] for r in rows} == {"implementing"} + assert [r["id"] for r in rows] == sorted(r["id"] for r in rows) + + +@pytest.mark.asyncio +async def test_fail_orphaned_plan_cas_only_flips_implementing(db): + """T2: True + flip for ``implementing``; False + byte-unchanged otherwise.""" + await _seed_plan(db, "plan-impl") + assert await db.fail_orphaned_plan("plan-impl") is True + assert (await db.get_plan("plan-impl"))["status"] == "failed" + + for other in ("pending", "done", "declined", "superseded", "failed"): + pid = f"plan-{other}" + await _seed_plan(db, pid, status=other) + before = await db.get_plan(pid) + assert await db.fail_orphaned_plan(pid) is False + assert await db.get_plan(pid) == before + + +@pytest.mark.asyncio +async def test_fail_orphaned_plan_flips_a_null_owner_row(db): + """T3: a NULL ``impl_session_id`` row still flips. + + This is the test that discriminates the status-only CAS from an + owner-keyed one: ``PATCH /api/plans/{id}`` takes an arbitrary status with + no whitelist, so it can create ``implementing`` with no owner, and + ``AND impl_session_id = ?`` bound to NULL matches nothing. + """ + await _seed_plan(db, "plan-noowner") + assert (await db.get_plan("plan-noowner"))["impl_session_id"] is None + + assert await db.fail_orphaned_plan("plan-noowner") is True + assert (await db.get_plan("plan-noowner"))["status"] == "failed" + + +# --------------------------------------------------------------------------- # +# Helper level # +# --------------------------------------------------------------------------- # + + +@pytest.mark.asyncio +async def test_recovery_flips_implementing_to_failed_only(db, monkeypatch, tmp_path): + """T4: exactly ``failed`` (the literal), other statuses untouched, count returned.""" + _no_queue(monkeypatch, tmp_path) + await _seed_plan(db, "plan-a") + await _seed_plan(db, "plan-b") + survivors = {} + for other in ("pending", "done", "declined", "superseded", "failed"): + pid = f"plan-{other}" + await _seed_plan(db, pid, status=other) + survivors[pid] = other + + flipped = await _recover(db, _notif_stub()) + + assert flipped == 2 + assert (await db.get_plan("plan-a"))["status"] == "failed" + assert (await db.get_plan("plan-b"))["status"] == "failed" + for pid, status in survivors.items(): + assert (await db.get_plan(pid))["status"] == status + + +@pytest.mark.asyncio +async def test_recovery_sends_exactly_one_notification(db, monkeypatch, tmp_path): + """T5: one high-priority system notification naming every flipped id.""" + _no_queue(monkeypatch, tmp_path) + await _seed_plan(db, "plan-a") + await _seed_plan(db, "plan-b") + notif = _notif_stub() + + await _recover(db, notif) + + assert notif.send_notification.await_count == 1 + kwargs = notif.send_notification.await_args.kwargs + assert kwargs["session_id"] == "system" + assert kwargs["priority"] == "high" + assert "plan-a" in kwargs["body"] and "plan-b" in kwargs["body"] + + +@pytest.mark.asyncio +async def test_recovery_is_silent_with_nothing_to_do(db, monkeypatch, tmp_path): + """T5b: no ``implementing`` rows means no notification at all.""" + _no_queue(monkeypatch, tmp_path) + await _seed_plan(db, "plan-pending", status="pending") + notif = _notif_stub() + + assert await _recover(db, notif) == 0 + assert notif.send_notification.await_count == 0 + + +@pytest.mark.asyncio +async def test_recovery_ignores_a_plan_that_left_implementing_mid_sweep( + db, monkeypatch, tmp_path, +): + """T6: a row leaving ``implementing`` between read and write is not claimed. + + Pins notify-on-flipped rather than notify-on-orphaned: with only this plan + orphaned, a pass that keyed its notification off the pre-CAS list would + alert about a plan that legitimately completed. + """ + _no_queue(monkeypatch, tmp_path) + await _seed_plan(db, "plan-racer") + notif = _notif_stub() + + real_cas = db.fail_orphaned_plan + + async def racing_cas(plan_id: str) -> bool: + # A concurrent ``task_done`` lands after the read, before our write. + await db.update_plan(plan_id, status="done") + return await real_cas(plan_id) + + with patch.object(db, "fail_orphaned_plan", side_effect=racing_cas): + flipped = await _recover(db, notif) + + assert flipped == 0 + assert (await db.get_plan("plan-racer"))["status"] == "done" + assert notif.send_notification.await_count == 0 + + +@pytest.mark.asyncio +async def test_recovery_survives_a_failing_notification(db, monkeypatch, tmp_path): + """T7: rows are already reconciled, so a notification error cannot abort.""" + _no_queue(monkeypatch, tmp_path) + await _seed_plan(db, "plan-a") + await _seed_plan(db, "plan-b") + notif = MagicMock(send_notification=AsyncMock(side_effect=RuntimeError("boom"))) + + assert await _recover(db, notif) == 2 + assert (await db.get_plan("plan-a"))["status"] == "failed" + assert (await db.get_plan("plan-b"))["status"] == "failed" + + +# --------------------------------------------------------------------------- # +# Resume-queue exclusion # +# --------------------------------------------------------------------------- # + + +@pytest.mark.asyncio +async def test_enrolled_and_eligible_plan_is_spared(db, monkeypatch, tmp_path): + """T10: the enrolled plan stays ``implementing``, an unenrolled one fails. + + Both in ONE sweep, so this discriminates per-plan exclusion from an + all-or-nothing skip. + """ + await _seed_resumable_session(db, "impl-live") + _queue(monkeypatch, tmp_path, "impl-live") + await _seed_plan(db, "plan-resumed", impl_session_id="impl-live") + await _seed_plan(db, "plan-orphan", impl_session_id="impl-dead") + notif = _notif_stub() + + flipped = await _recover(db, notif) + + assert flipped == 1 + assert (await db.get_plan("plan-resumed"))["status"] == "implementing" + assert (await db.get_plan("plan-orphan"))["status"] == "failed" + body = notif.send_notification.await_args.kwargs["body"] + assert "plan-orphan" in body and "plan-resumed" not in body + # The COUNT in the body must also be what was flipped, not what was read: + # this sweep reads 2 orphans and flips 1, so a body built from the pre-CAS + # list would announce "2 plan(s)" while naming only one. + assert body.startswith("1 plan(s)"), body + + +@pytest.mark.asyncio +async def test_eligibility_preflight_covers_every_skip_class(db, monkeypatch, tmp_path): + """T15: one enrolled plan per engine skip class is still swept. + + ``resume_enrolled_sessions`` skips missing / archived / satellite / + no-SDK-session ids, and a skipped session never resumes -- so sparing its + plan would wedge the plan for nothing. All four classes plus an eligible + control in ONE sweep: a single-class fixture could not tell a full + preflight from one checking only that predicate. + """ + # 1. session row absent entirely (no create_session at all) + await _seed_plan(db, "plan-missing", impl_session_id="impl-missing") + # 2. archived + await _seed_resumable_session(db, "impl-archived") + await db.update_session_fields("impl-archived", {"status": "archived"}) + await _seed_plan(db, "plan-archived", impl_session_id="impl-archived") + # 3. satellite (source="external") + await db.create_session("impl-external", source="external", status="running") + await db.update_session_metadata("impl-external", {"sdk_session_id": "sdk-ext"}) + await _seed_plan(db, "plan-external", impl_session_id="impl-external") + # 4. no SDK session to resume + await db.create_session("impl-nosdk", source="web", status="running") + await _seed_plan(db, "plan-nosdk", impl_session_id="impl-nosdk") + # control: fully eligible + await _seed_resumable_session(db, "impl-ok") + await _seed_plan(db, "plan-ok", impl_session_id="impl-ok") + + _queue( + monkeypatch, tmp_path, + "impl-missing", "impl-archived", "impl-external", "impl-nosdk", "impl-ok", + ) + + flipped = await _recover(db, _notif_stub()) + + assert flipped == 4 + for pid in ("plan-missing", "plan-archived", "plan-external", "plan-nosdk"): + assert (await db.get_plan(pid))["status"] == "failed", pid + assert (await db.get_plan("plan-ok"))["status"] == "implementing" + + +def test_queue_parse_matches_the_engine(monkeypatch, tmp_path): + """T11: byte-parity with ``AgentEngine.resume_enrolled_sessions``' parse. + + Asserted against the engine's own expression rather than a hand-copied + list, so the two readers cannot drift about what counts as enrolled. + """ + # The internal-whitespace id is load-bearing: without it a + # whitespace-splitting parse (``raw.split()``) yields the identical set on + # blank lines / padding / duplicates alone, so the fixture could not tell + # the two apart. With it, splitting shatters one id into three tokens. + raw = "impl-a\n\n impl-b \nimpl-a\n\t\n impl-c\nimpl d with space\n" + _queue(monkeypatch, tmp_path, raw=raw) + + # The engine's parse, verbatim (engine.py resume_enrolled_sessions). + seen: set[str] = set() + ids: list[str] = [] + for line in raw.splitlines(): + sid = line.strip() + if sid and sid not in seen: + seen.add(sid) + ids.append(sid) + + from nerve.agent.plan_service import _enrolled_resume_session_ids + + assert _enrolled_resume_session_ids() == set(ids) + + +@pytest.mark.asyncio +async def test_recovery_does_not_consume_the_queue(db, monkeypatch, tmp_path): + """T12: the queue is READ, never drained. + + The engine's resume task is the sole drainer and runs later in startup; + consuming the file here would silently cancel every enrolled resume. + """ + await _seed_resumable_session(db, "impl-live") + q = _queue(monkeypatch, tmp_path, "impl-live") + before = q.read_bytes() + await _seed_plan(db, "plan-resumed", impl_session_id="impl-live") + + await _recover(db, _notif_stub()) + + assert q.exists() + assert q.read_bytes() == before + + +@pytest.mark.asyncio +async def test_missing_queue_sweeps_but_unreadable_queue_does_not( + db, monkeypatch, tmp_path, +): + """T13: the two error cases go OPPOSITE ways. + + A missing file is a definite answer from the sole writer (the CLI appends + before triggering the restart), so sweep. An ``OSError`` is not an answer: + our read and the engine's happen at different instants, so a transient + error here can succeed there and resume a session whose plan we would have + failed -- hence fail closed. Both arms assert on plan state, since a + single-arm test cannot discriminate the two directions. + """ + # Arm 1: missing queue -> sweep. + _no_queue(monkeypatch, tmp_path) + await _seed_plan(db, "plan-a") + assert await _recover(db, _notif_stub()) == 1 + assert (await db.get_plan("plan-a"))["status"] == "failed" + + # Arm 2: unreadable queue -> sweep nothing, notify nothing. + await _seed_plan(db, "plan-b") + unreadable = tmp_path / "unreadable-queue" + unreadable.write_text("impl-whatever\n") + monkeypatch.setattr( + "nerve.agent.plan_service.RESUME_QUEUE_FILE", unreadable, + ) + notif = _notif_stub() + with patch.object( + type(unreadable), "read_text", side_effect=OSError("EIO"), + ): + assert await _recover(db, notif) == 0 + assert (await db.get_plan("plan-b"))["status"] == "implementing" + assert notif.send_notification.await_count == 0 + + +@pytest.mark.asyncio +async def test_an_already_failed_plan_is_not_reclaimed(db, monkeypatch, tmp_path): + """T16: a ``failed`` + enrolled + eligible plan is left ``failed``. + + IMPORTANT: This documents an ACCEPTED trade, not a bug. An earlier daemon can sweep + a plan before its session is enrolled (the daemon publishes its PID before + startup runs), so a later daemon can resume that session with the plan + already ``failed``. The pass deliberately does NOT reclaim it: the task + still completes via ``task_done`` and ``plan_propose`` stays unblocked, so + only the plan's label differs. Re-adopting it would need durable recovery + provenance -- a legitimately failed plan also keeps its + ``impl_session_id``, and ``PATCH /api/plans/{id}`` can set ``failed`` with + no whitelist -- i.e. a schema change this change avoids. Do not "fix" this + into a reclaim without solving that first. + """ + await _seed_resumable_session(db, "impl-live") + _queue(monkeypatch, tmp_path, "impl-live") + await _seed_plan(db, "plan-swept", status="failed", impl_session_id="impl-live") + notif = _notif_stub() + + assert await _recover(db, notif) == 0 + assert (await db.get_plan("plan-swept"))["status"] == "failed" + assert notif.send_notification.await_count == 0 + + +# --------------------------------------------------------------------------- # +# Wiring: the pass is reachable from daemon startup, and correctly ordered # +# --------------------------------------------------------------------------- # + + +@pytest.fixture +def lifespan_app(tmp_path, monkeypatch): + """Drive the real gateway ``lifespan`` against a temp DB. + + Follows ``tests/test_mcp_http_integration.py``: config built in-process, + ``init_db`` patched to a temp path, heavy components stubbed. Yields + ``(app_factory, db_holder, cron_stub, engine_stub)`` so a test can seed rows + before startup, inspect them after, and re-stub the engine method whose + ordering against recovery is load-bearing (the resume-queue drainer). + """ + import nerve.config as config_module + from nerve.config import NerveConfig + from nerve.gateway import server as gw_server + + config = NerveConfig() + config.workspace = tmp_path / "workspace" + config.workspace.mkdir(parents=True, exist_ok=True) + config.anthropic_api_key = "" + config.telegram.enabled = False + config.mcp_endpoint.enabled = False + config.agent.model_discovery = False + config_module._config = config + + fake_engine = MagicMock() + fake_engine.config = config + fake_engine.db = None + fake_engine._memory_bridge = None + fake_engine._skill_manager = None + fake_engine.notification_service = None + fake_engine.set_notification_service = MagicMock() + fake_engine.shutdown = AsyncMock() + fake_engine.initialize = AsyncMock() + fake_engine.router = MagicMock() + fake_engine.run = AsyncMock() + fake_engine.is_session_running = MagicMock(return_value=False) + fake_engine.sessions = MagicMock() + fake_engine.sessions.run_cleanup = AsyncMock(return_value={}) + fake_engine.run_memorization_sweep = AsyncMock(return_value={}) + fake_engine.run_idle_client_sweep = AsyncMock() + fake_engine.resume_enrolled_sessions = AsyncMock(return_value=0) + from nerve.agent.tools import build_default_registry + + fake_engine.registry = build_default_registry() + + db_path = tmp_path / "lifespan.db" + holder: dict = {} + + async def _init_db_patched(*args, **kwargs): + from nerve.db import Database + + d = Database(db_path) + await d.connect() + fake_engine.db = d + holder["db"] = d + return d + + async def _close_db_patched(): + if fake_engine.db is not None: + await fake_engine.db.close() + + notif_stub = MagicMock( + send_notification=AsyncMock(return_value="notif-stub"), + expire_stale=AsyncMock(return_value=0), + hide_session_label_for=MagicMock(), + ) + cron_stub = MagicMock(start=AsyncMock(), stop=AsyncMock(), + _source_runners=[], _jobs=[]) + + def _build(): + return gw_server.create_app() + + with patch("nerve.gateway.server.AgentEngine", return_value=fake_engine), \ + patch("nerve.gateway.server.init_db", side_effect=_init_db_patched), \ + patch("nerve.gateway.server.close_db", side_effect=_close_db_patched), \ + patch("nerve.notifications.service.NotificationService", + return_value=notif_stub), \ + patch("nerve.gateway.server.init_langfuse"), \ + patch("nerve.cron.service.CronService", return_value=cron_stub): + yield _build, holder, cron_stub, fake_engine + + gw_server._engine = None + gw_server._mcp_manager = None + config_module._config = None + + +def _seed_via_sqlite(db_path, plan_id: str, impl_session_id: str | None = None) -> None: + """Insert an ``implementing`` plan directly, before the app starts. + + Synchronous on purpose: the lifespan tests are sync (``TestClient``), so + they cannot await the async store, and the row must exist before startup. + + ``impl_session_id`` defaults to NULL, which is what an unowned orphan looks + like. Pass it to build a plan the resume exclusion can actually spare: the + exclusion tests ``sid and sid in enrolled``, so a NULL owner is swept however + the queue reads. + """ + import sqlite3 + + conn = sqlite3.connect(db_path) + try: + conn.execute( + "INSERT INTO plans (id, task_id, impl_session_id, status, content, " + "created_at) VALUES (?, ?, ?, 'implementing', 'c', " + "'2026-01-01T00:00:00Z')", + (plan_id, "task-1", impl_session_id), + ) + conn.commit() + finally: + conn.close() + + +def _seed_session_via_sqlite(db_path, session_id: str) -> None: + """Insert a resume-eligible session row directly, before the app starts. + + Eligible = present, not archived, not a satellite, with an SDK session to + resume: the four predicates ``AgentEngine.resume_enrolled_sessions`` skips + on, which ``_resume_eligible`` re-evaluates. Sync for the same reason as + ``_seed_via_sqlite``. + """ + import sqlite3 + + conn = sqlite3.connect(db_path) + try: + conn.execute( + "INSERT INTO sessions (id, source, status, sdk_session_id, created_at) " + "VALUES (?, 'web', 'running', ?, '2026-01-01T00:00:00Z')", + (session_id, f"sdk-{session_id}"), + ) + conn.commit() + finally: + conn.close() + + +def _plan_status(db_path, plan_id: str) -> str | None: + import sqlite3 + + conn = sqlite3.connect(db_path) + try: + row = conn.execute( + "SELECT status FROM plans WHERE id = ?", (plan_id,) + ).fetchone() + return row[0] if row else None + finally: + conn.close() + + +def test_lifespan_recovers_an_orphaned_plan(lifespan_app, tmp_path, monkeypatch): + """T8: the pass is actually reached from daemon startup. + + IMPORTANT: Imports no new symbol and asserts only DB state, which is what makes its + base-arm failure behavioural (the plan is still ``implementing``) rather + than an ``ImportError`` -- and makes it the only test that can observe the + lifespan call being deleted. + """ + from fastapi.testclient import TestClient + + build, holder, _cron, _engine = lifespan_app + db_path = tmp_path / "lifespan.db" + # First start materializes the schema; seed and restart to get an orphan + # that predates startup. + with TestClient(build()): + pass + _seed_via_sqlite(db_path, "plan-orphan") + assert _plan_status(db_path, "plan-orphan") == "implementing" + + with TestClient(build()): + pass + + assert _plan_status(db_path, "plan-orphan") == "failed" + + +def test_recovery_runs_before_cron_starts(lifespan_app, tmp_path): + """T14: cron OBSERVES the plan already reconciled. + + IMPORTANT: Asserted on what cron sees, not on wall-clock order or source line + numbers: a source-order assertion would pass against a version that awaits + something else in between. Ordering is load-bearing -- cron's catch-up can + dispatch a planner run whose ``plan_propose`` reads a stale + ``implementing`` row and permanently skips the task. + """ + from fastapi.testclient import TestClient + + build, holder, cron_stub, _engine = lifespan_app + db_path = tmp_path / "lifespan.db" + with TestClient(build()): + pass + _seed_via_sqlite(db_path, "plan-orphan") + + observed: dict = {} + + async def _record_then_start(): + observed["status"] = _plan_status(db_path, "plan-orphan") + + cron_stub.start = AsyncMock(side_effect=_record_then_start) + + with TestClient(build()): + pass + + assert observed["status"] == "failed" + + +def test_a_raising_recovery_does_not_block_startup(lifespan_app, tmp_path): + """T9: recovery is not startup-critical, so a failure must not take it down.""" + from fastapi.testclient import TestClient + + build, _holder, cron_stub, _engine = lifespan_app + db_path = tmp_path / "lifespan.db" + with TestClient(build()): + pass + _seed_via_sqlite(db_path, "plan-orphan") + + with patch( + "nerve.agent.plan_service.recover_orphaned_plans", + side_effect=RuntimeError("boom"), + ): + with TestClient(build()) as client: + resp = client.get("/health") + + assert resp.status_code == 200 + # The pass raised, so the row is untouched -- startup still completed. + assert _plan_status(db_path, "plan-orphan") == "implementing" + + +def test_recovery_reads_the_resume_queue_before_the_engine_drains_it( + lifespan_app, tmp_path, monkeypatch, +): + """T18: the enrolled plan survives real startup, so recovery read the queue first. + + IMPORTANT: The exclusion is only sound while recovery's read precedes the engine's + drain -- ``resume_enrolled_sessions`` unlinks the queue up front, so a + recovery pass running after it sees no queue at all and sweeps EVERY + enrolled plan. Nothing in the source pins that ordering, and without this + test hoisting the drain above recovery keeps every other test green. + + Asserted on plan STATE with a stub that really unlinks, not on call order or + source line numbers: an order assertion would pass against a version that + awaits something else in between. Same shape as ``T14``, which pins the + other load-bearing ordering (cron) through an observer stub. + """ + from fastapi.testclient import TestClient + + build, _holder, _cron, engine_stub = lifespan_app + db_path = tmp_path / "lifespan.db" + # First start materializes the schema; seed and restart to get an orphan + # that predates startup. + with TestClient(build()): + pass + + # The owner session must be seeded too, and must be eligible: the exclusion + # spares a plan only if its session would really be resumed. + _seed_via_sqlite(db_path, "plan-resumed", impl_session_id="impl-live") + _seed_session_via_sqlite(db_path, "impl-live") + + queue = tmp_path / "resume-queue" + queue.write_text("impl-live\n") + monkeypatch.setattr("nerve.agent.plan_service.RESUME_QUEUE_FILE", queue) + + drained: dict = {} + + async def _draining_resume() -> int: + # What the real engine does: read, then unlink up front. + drained["existed"] = queue.exists() + queue.unlink(missing_ok=True) + return 0 + + engine_stub.resume_enrolled_sessions = AsyncMock(side_effect=_draining_resume) + + with TestClient(build()): + pass + + assert _plan_status(db_path, "plan-resumed") == "implementing" + # Anti-vacuity: without this, a stub that never ran would leave the plan + # ``implementing`` for the wrong reason and the test would pass even against + # a deleted exclusion. + assert drained.get("existed") is True