Sessions: archived and deleted sessions stay closed

docs/api.md

@@ -1116,6 +1116,11 @@ The delete removes the session's events and its agent records and
 transcripts. Nerve records the delete in
 `session_deletions` with the signed-in person as actor and cause (see
 [Session changes](accounts.md#session-changes)).
+While a turn of the session runs, a message waits in its turn queue, or a
+background continuation of the runtime runs, the delete answers `409`: stop
+the session first. The delete expires the
+session's pending questions; the notification rows stay. A turn never creates
+a deleted session again: a run in it is refused.
 
 #### `PATCH /api/sessions/{id}`
 Update `title`, `starred`, `model` or `parent_session_id`. A `model` change
@@ -1187,6 +1192,19 @@ its ancestors stay: `session is running` (a turn is in flight) or
 archives those too; their agents end as `interrupted` (`archived`).
 `archived` is `false` when nothing was archived.
 
+An archive deletes the session's pending wakeups (a "Run later" message
+included) and expires its pending questions. Approvals stay pending, because
+their answer acts on their target, not on the session. No turn runs in an
+archived session: a wakeup or a queued message is refused, an answer is
+stored but starts no turn, and a channel message goes to a new session. Only these
+open an archived session again, each with an `unarchived` event: the
+unarchive endpoint, a star, and a person's own message into it (the web chat,
+`POST /api/chat` with its `session_id`, "Run later", a plan revision request
+for a plan that it proposed). The archive itself is refused (`session is
+running`) while a turn runs or waits, or while a background continuation of
+the runtime runs. A route that would run a turn in an archived session
+answers `409` with `{"detail": "...", "reason": "archived"}`.
+
 ```json
 Response: { "archived": true, "archived_ids": ["a1b2c3d4"], "skipped": [{ "id": "e5f6a7b8", "reason": "background agents running" }] }
 ```
@@ -1306,7 +1324,10 @@ Response: { "response": "Hi there!", "session_id": "main" }
 `client_msg_id` is optional: a retry with an id the session saw in the last
 10 minutes runs nothing and answers with an empty `response`. While another
 turn of the session runs, the message waits in the session's
-[turn queue](#get-apisessionsidqueue) (source `api`).
+[turn queue](#get-apisessionsidqueue) (source `api`). A `session_id` of an
+archived session unarchives it first, in the caller's name. A `session_id` of
+a deleted session creates nothing and answers `409` with
+`{"detail": "This session was deleted.", "reason": "deleted"}`.
 
 For streaming, use the WebSocket endpoint.
 

docs/plans.md

@@ -30,7 +30,7 @@ User reviews (via /plans UI or chat tools)
 | `plan_read` | Read full plan content. Used to review details before approving/declining. |
 | `plan_approve` | Approve a pending plan and spawn an implementation session. With `implement_here=true`, record the approval for the calling session and start nothing. |
 | `plan_decline` | Decline a pending plan with optional feedback. Moves the related task to `done` (declining is treated as abandoning the effort, not pausing it). |
-| `plan_revise` | Request revision of a pending plan — sends feedback to the session that proposed it (the planner session for cron-made plans; `cron:task-planner` when the plan records no session), which calls `plan_update` to produce v+1. |
+| `plan_revise` | Request revision of a pending plan — sends feedback to the session that proposed it (the planner session for cron-made plans; `cron:task-planner` when the plan records no session or its session was deleted; an archived session is unarchived first), which calls `plan_update` to produce v+1. |
 
 ### `plan_propose(task_id, content, plan_type?, summary?)`
 

nerve/agent/engine.py

@@ -143,6 +143,25 @@ def __init__(self) -> None:
         super().__init__(_NOT_ACCEPTING_TURNS_MESSAGE)
 
 
+_SESSION_CLOSED_MESSAGES = {
+    "archived": "This session is archived. Unarchive it to continue.",
+    "deleted": "This session was deleted.",
+}
+
+
+class SessionClosedError(RuntimeError):
+    """``AgentEngine.run`` refused a turn because its session is closed.
+
+    ``reason`` is ``archived`` (only unarchive opens the session again) or
+    ``deleted`` (no session row, and ``session_deletions`` records the
+    delete). The user message is not stored.
+    """
+
+    def __init__(self, reason: str) -> None:
+        super().__init__(_SESSION_CLOSED_MESSAGES[reason])
+        self.reason = reason
+
+
 # The runtime's system frame that marks a compacted conversation.
 _COMPACT_BOUNDARY = "compact_boundary"
 # A compact turn that ended with no such frame: the runtime had no messages.
@@ -472,6 +491,10 @@ def __init__(self, config: NerveConfig, db: Database):
         # guard asks whether agents still run.
         self.sessions.on_client_removed = self.agents.on_client_gone
         self.sessions.has_live_agents = self.agents.has_live
+        # An archive expires the session's pending questions, and is refused
+        # while anything holds the session's run lock.
+        self.sessions.on_archived = self.expire_session_questions
+        self.sessions.run_lock_held = self._run_lock_held
         # Generation of the live runtime client per build (any session): a
         # finished agent is resumable only on the client that ran it.
         self._client_gen = 0
@@ -1150,6 +1173,41 @@ async def _memorizing(self):
             async with self._memorize_lock:
                 yield
 
+    async def reopen_archived(
+        self, session_id: str, *, attribution: Attribution,
+    ) -> bool:
+        """Unarchive ``session_id`` before a person's own message runs in it.
+
+        A message that a person sends into an archived session (the web
+        chat, ``POST /api/chat``, a plan revision request) opens it again,
+        recorded as ``unarchived`` with ``attribution``. Every other run in
+        an archived session is refused (:class:`SessionClosedError`).
+        Returns ``True`` when this call unarchived the session.
+        """
+        session = await self.db.get_session(session_id)
+        if not session or session.get("status") != SessionStatus.ARCHIVED.value:
+            return False
+        return await self.sessions.unarchive_session(
+            session_id, attribution=attribution,
+        )
+
+    def _run_lock_held(self, session_id: str) -> bool:
+        """Something holds the session's run lock: a turn, a run that waits
+        in the queue (the lock stays held across the handover), an
+        autonomous turn, an idle drain or a client reap."""
+        lock = self._session_locks.get(session_id)
+        return lock is not None and lock.locked()
+
+    async def expire_session_questions(self, session_id: str) -> None:
+        """Expire the pending questions of a session that was archived or
+        deleted, with the same reports as an expiry (web, Telegram, Slack).
+        Without a notification service (CLI, tests), only the rows change."""
+        service = self.notification_service
+        if service is not None:
+            await service.expire_session_questions(session_id)
+        else:
+            await self.db.expire_pending_questions_for_session(session_id)
+
     async def _memorize_session(
         self, session_id: str, connected_at_override: str | None = None,
     ) -> None:
@@ -2068,8 +2126,11 @@ async def _refuse_turn(
         *,
         entry: QueuedTurn | None = None,
         client_msg_id: str | None = None,
+        error: RuntimeError | None = None,
     ) -> None:
-        """Tell the session's clients that run() refused the turn, then raise.
+        """Tell the session's clients that run() refused the turn, then raise
+        ``error`` (default :class:`NotAcceptingTurnsError`). The error event
+        says why.
 
         A run that has a queue entry or a ``client_msg_id`` is reported in
         ``queue_update`` with the reason ``refused``. The caller removes the
@@ -2080,8 +2141,9 @@ async def _refuse_turn(
                 "queue_id": entry.queue_id if entry is not None else None,
                 "client_msg_id": client_msg_id, "reason": "refused",
             }])
-        await broadcaster.broadcast_error(session_id, _NOT_ACCEPTING_TURNS_MESSAGE)
-        raise NotAcceptingTurnsError()
+        error = error or NotAcceptingTurnsError()
+        await broadcaster.broadcast_error(session_id, str(error))
+        raise error
 
     # ------------------------------------------------------------------ #
     #  Public API: run, stop, fork, resume                                 #
@@ -3978,6 +4040,18 @@ async def _run_turn(
             # The Stop target from the moment this run holds the run lock:
             # Stop reaches the turn, never a message queued behind it.
             self.sessions.set_turn_task(session_id, turn_task)
+            # The session can be archived or deleted before this run holds
+            # the lock, or while it waits in the queue. No await since the
+            # lock (see SessionManager.closing).
+            closed = (
+                self.sessions.closing_reason(session_id)
+                or await self.db.closed_session_reason(session_id)
+            )
+            if closed:
+                await self._refuse_turn(
+                    session_id, entry=entry, client_msg_id=client_msg_id,
+                    error=SessionClosedError(closed),
+                )
             if entry is not None or client_msg_id:
                 await self._emit_queue_update(session_id, removed=[{
                     "queue_id": entry.queue_id if entry is not None else None,

nerve/agent/plan_service.py

@@ -168,7 +168,8 @@ async def request_plan_revision(
         - Writes a ``Revision requested for {plan_id}`` note to the task.
         - Dispatches ``engine.run()`` on the original proposer session
           (``plan.session_id``), falling back to ``cron:task-planner``
-          if the proposer is unknown/deleted.
+          if the proposer is unknown/deleted. An archived proposer session
+          is unarchived first, in the name of ``actor``.
         - The planner is instructed to call
           ``plan_update(plan_id=..., content=..., summary=..., feedback=...)``
           so version history stays linked.
@@ -239,6 +240,13 @@ async def request_plan_revision(
     #    Fall back to the cron planner if the plan was created without
     #    a session attribution (older rows, manual inserts, etc.).
     session_id = plan.get("session_id") or _FALLBACK_PLANNER_SESSION
+    # A deleted proposer session is not created again: the cron planner
+    # takes the revision.
+    if (
+        session_id != _FALLBACK_PLANNER_SESSION
+        and await db.closed_session_reason(session_id) == "deleted"
+    ):
+        session_id = _FALLBACK_PLANNER_SESSION
     session_title = (
         f"Cron: {session_id.split(':')[-1]}"
         if session_id.startswith("cron:")
@@ -251,6 +259,12 @@ async def request_plan_revision(
         actor=engine.db.system_actor, cause=cause,
         origin=plan_origin(plan_id, plan["task_id"]),
     )
+    # The person's revision request opens an archived planner session again
+    # (an ``unarchived`` event in their name); a run in an archived session
+    # is refused.
+    await engine.reopen_archived(
+        session_id, attribution=attribution_for(actor, cause),
+    )
     asyncio.create_task(
         engine.run(
             session_id=session_id, user_message=prompt, source="cron",