plans: reconcile 'implementing' plans orphaned by a daemon restart

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
 

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)

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

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