From 147746e83ee2821d146683554078b52dc9791c0a Mon Sep 17 00:00:00 2001
From: Alex Soffronow Pagonidis
 <237136924+alex-clickhouse@users.noreply.github.com>
Date: Tue, 10 Nov 2026 06:13:23 +0000
Subject: [PATCH 1/5] Sessions: archived and deleted sessions stay closed
 (#671)

An archived session leaves `archived` only through unarchive, and a run
never creates a row for a deleted session id.

- Engine: run() raises SessionClosedError for an archived or deleted
  session. It checks after the run lock and again under the client lock
  while it builds the client. The queue entry leaves as `refused`, and
  the error event says why. Routes answer 409.
- mark_active/idle/stopped/error write nothing to an archived session.
- Archive: refused while a turn holds the run lock (also while it waits
  for a concurrency slot). The check and the status write hold the
  client lock. After the archive, pending wakeups are deleted and
  pending questions expire with the normal reports. Approvals stay.
- Channels: an archived session is not sticky. The router sends a
  message refused by an archive again, once, to a new session.
- Delete: 409 while a turn runs or waits; a run cannot start during the
  delete. The session's questions expire. The web chat shows an alert.
- An answer to an archived or deleted session is stored, no turn.
- A person's own message (web chat, POST /api/chat, plan revision)
  unarchives the session in their name, then runs.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 docs/api.md                       |  18 +-
 nerve/agent/engine.py             | 151 ++++++-
 nerve/agent/plan_service.py       |   9 +-
 nerve/agent/sessions.py           | 185 ++++++---
 nerve/channels/router.py          |  18 +
 nerve/cron/service.py             |   9 +-
 nerve/db/notifications.py         |  27 +-
 nerve/db/sessions.py              |  17 +
 nerve/gateway/routes/sessions.py  | 116 ++++--
 nerve/gateway/server.py           |  24 +-
 nerve/notifications/service.py    |  42 +-
 tests/test_closed_sessions.py     | 670 ++++++++++++++++++++++++++++++
 tests/test_drain.py               |   5 +-
 tests/test_external_auth.py       |   3 +
 tests/test_plan_actions.py        |   3 +
 tests/test_plan_revise.py         |   7 +
 tests/test_session_attribution.py |   6 +-
 web/src/stores/chatStore.test.ts  |  32 ++
 web/src/stores/chatStore.ts       |   4 +
 19 files changed, 1217 insertions(+), 129 deletions(-)
 create mode 100644 tests/test_closed_sessions.py

diff --git a/docs/api.md b/docs/api.md
index d707583eb..b0d6f2e24 100644
--- a/docs/api.md
+++ b/docs/api.md
@@ -1116,6 +1116,10 @@ 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, or a message waits in its turn queue, 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 +1191,17 @@ 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`, a plan revision request for a plan
+that it proposed). 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 +1321,8 @@ 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.
 
 For streaming, use the WebSocket endpoint.
 
diff --git a/nerve/agent/engine.py b/nerve/agent/engine.py
index 111b849af..b39f0816e 100644
--- a/nerve/agent/engine.py
+++ b/nerve/agent/engine.py
@@ -24,7 +24,7 @@
 from collections.abc import Iterator
 from dataclasses import dataclass, field
 from datetime import datetime, timedelta, timezone
-from typing import Any
+from typing import Any, Literal
 
 from nerve.agent.agents import AgentRegistry
 from nerve.agent.agents.errors import (
@@ -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,8 @@ 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.
+        self.sessions.on_archived = self.expire_session_questions
         # 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 +1171,66 @@ 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 has_turn_work(self, session_id: str) -> bool:
+        """A turn of the session runs, waits for a concurrency slot, or
+        waits in the session's turn queue."""
+        return (
+            self.sessions.has_running_work(session_id)
+            or bool(self.queue_view(session_id))
+        )
+
+    def _deleting_sessions(self) -> set[str]:
+        """Sessions whose delete is in progress (created on first use, so
+        engines built with ``__new__`` work too)."""
+        return self.__dict__.setdefault("_deleting", set())
+
+    def begin_delete(self, session_id: str) -> bool:
+        """Start the delete of a session: ``False`` while it has turn work
+        (:meth:`has_turn_work`), else mark it and return ``True``. Call
+        :meth:`end_delete` when the rows are gone.
+
+        No await here. A run sets its turn task and then checks the mark,
+        with no await in between either. So a run either sees the mark and
+        is refused, or makes this call return ``False``.
+        """
+        if self.has_turn_work(session_id):
+            return False
+        self._deleting_sessions().add(session_id)
+        return True
+
+    def end_delete(self, session_id: str) -> None:
+        """End what :meth:`begin_delete` started. The ``session_deletions``
+        row refuses later runs."""
+        self._deleting_sessions().discard(session_id)
+
+    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:
@@ -1496,6 +1577,12 @@ async def _get_or_create_client(
 
             # Session row first — backend resolution is sticky on it.
             session = await self.db.get_session(session_id)
+            # Under the client lock, which archive_session also holds: a
+            # session archived (or deleted) after the checks in _run_turn
+            # gets no client, and its runtime gets no turn.
+            closed = await self._session_closed_reason(session_id, session)
+            if closed:
+                raise SessionClosedError(closed)
             backend = self._backend_for(session, source)
             requested_model = (
                 model or (session or {}).get("model")
@@ -2062,6 +2149,50 @@ async def drain(self, timeout: float) -> bool:
         logger.info("Drain: all running turns ended")
         return True
 
+    async def _session_closed_reason(
+        self, session_id: str, session: dict | None | Literal[False] = False,
+    ) -> str | None:
+        """``archived`` or ``deleted`` when no turn may run in the session,
+        else ``None``. ``session`` is the row when the caller has read it
+        already (``None`` for no row). A delete in progress counts as
+        ``deleted``; that check comes before the first await."""
+        if session_id in self._deleting_sessions():
+            return "deleted"
+        if session is False:
+            session = await self.db.get_session(session_id)
+        if session is not None:
+            if session.get("status") == SessionStatus.ARCHIVED.value:
+                return "archived"
+            return None
+        if await self.db.get_session_deletion(session_id) is not None:
+            return "deleted"
+        return None
+
+    async def _refuse_closed_session(
+        self,
+        session_id: str,
+        reason: str,
+        *,
+        entry: QueuedTurn | None = None,
+        client_msg_id: str | None = None,
+    ) -> None:
+        """Tell the session's clients that run() refused the turn because the
+        session is closed, then raise :class:`SessionClosedError`.
+
+        As :meth:`_refuse_turn`: a run that has a queue entry or a
+        ``client_msg_id`` leaves the queue with the reason ``refused``. The
+        error event says why.
+        """
+        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,
+                "client_msg_id": client_msg_id, "reason": "refused",
+            }])
+        error = SessionClosedError(reason)
+        logger.info("Session %s: refused a turn (%s)", session_id, reason)
+        await broadcaster.broadcast_error(session_id, str(error))
+        raise error
+
     async def _refuse_turn(
         self,
         session_id: str,
@@ -3976,8 +4107,16 @@ async def _run_turn(
                     session_id, entry=entry, client_msg_id=client_msg_id,
                 )
             # The Stop target from the moment this run holds the run lock:
-            # Stop reaches the turn, never a message queued behind it.
+            # Stop reaches the turn, never a message queued behind it. From
+            # here on, an archive of the session is refused.
             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.
+            closed = await self._session_closed_reason(session_id)
+            if closed:
+                await self._refuse_closed_session(
+                    session_id, closed, entry=entry, client_msg_id=client_msg_id,
+                )
             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,
@@ -4531,6 +4670,14 @@ def _with_session_preamble(base: TurnInput) -> TurnInput:
             await self.schedule_memorize(session_id)
             return TurnResult(partial, outcome)
 
+        except SessionClosedError as e:
+            # Archived or deleted while this turn waited for its client: the
+            # runtime got nothing, so there is no reply and no error state.
+            self._end_turn(session_id, TurnOutcome("interrupted", str(e)))
+            logger.info("Session %s: turn refused (%s)", session_id, e.reason)
+            await broadcaster.broadcast_error(session_id, str(e))
+            raise
+
         except Exception as e:
             error_msg = f"Agent error: {e}"
             logger.error(error_msg, exc_info=True)
diff --git a/nerve/agent/plan_service.py b/nerve/agent/plan_service.py
index 257773883..fe992d5ee 100644
--- a/nerve/agent/plan_service.py
+++ b/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.
@@ -251,6 +252,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",
diff --git a/nerve/agent/sessions.py b/nerve/agent/sessions.py
index ab2240a53..ff0b646ad 100644
--- a/nerve/agent/sessions.py
+++ b/nerve/agent/sessions.py
@@ -17,7 +17,7 @@
 import uuid
 from datetime import datetime, timedelta, timezone
 from enum import StrEnum
-from typing import Any, Callable, Literal
+from typing import Any, Awaitable, Callable, Literal
 
 from nerve.db import Database
 from nerve.identity import (
@@ -81,11 +81,14 @@ class SessionStatus(StrEnum):
 
 # The reason an archive refuses a session whose agents still run.
 BUSY_AGENTS_REASON = "background agents running"
+BUSY_RUNNING_REASON = "session is running"
 
 
 class SessionBusyError(RuntimeError):
-    """The session cannot be archived now: background agents still run in
-    it (archiving would stop them). Pass ``force=True`` to archive anyway."""
+    """The session cannot be archived now: a turn of it runs or waits for a
+    concurrency slot (``BUSY_RUNNING_REASON``), or background agents still
+    run in it (``BUSY_AGENTS_REASON``, archiving would stop them; pass
+    ``force=True`` to archive anyway)."""
 
 
 class SessionManager:
@@ -109,6 +112,10 @@ class SessionManager:
     # ``has_live_agents(session_id)``: background agents still run in the
     # session (the archive guard). Wired to the agent registry.
     has_live_agents: Callable[[str], bool] | None = None
+    # Awaited as ``on_archived(session_id)`` after an archive changed the
+    # status (the engine wires the expiry of the session's questions).
+    # Never raises into the caller.
+    on_archived: Callable[[str], Awaitable[None]] | None = None
     # Called as ``listener(session_id, running)`` right after a session
     # starts or stops running a turn (see ``add_running_listener``).
     _running_listeners: tuple[Callable[[str, bool], None], ...] = ()
@@ -262,12 +269,16 @@ async def transition(
                 attribution=attribution,
             )
 
+    # The ``mark_*`` methods record the state of the runtime. They write
+    # nothing to an archived or missing session: only unarchive takes a
+    # session out of ``archived``. Each returns ``True`` when it wrote.
+
     async def mark_active(
         self,
         session_id: str,
         sdk_session_id: str | None = None,
         connected_at: str | None = None,
-    ) -> None:
+    ) -> bool:
         """Mark session as active with an SDK client.
 
         ``connected_at`` is the (stable) time the SDK client connected; callers
@@ -287,41 +298,49 @@ async def mark_active(
         }
         if sdk_session_id is not None:
             fields["sdk_session_id"] = sdk_session_id
-        await self.db.update_session_fields(session_id, fields)
+        if not await self.db.update_unarchived_session_fields(session_id, fields):
+            return False
         await self.db.log_session_event(session_id, "started", {
             "sdk_session_id": sdk_session_id,
         })
+        return True
 
     async def mark_idle(
         self, session_id: str, preserve_sdk_id: bool = True,
-    ) -> None:
+    ) -> bool:
         """Mark session as idle (client disconnected but potentially resumable)."""
         fields: dict[str, Any] = {"status": SessionStatus.IDLE.value}
         if not preserve_sdk_id:
             fields["sdk_session_id"] = None
             fields["connected_at"] = None
-        await self.db.update_session_fields(session_id, fields)
+        if not await self.db.update_unarchived_session_fields(session_id, fields):
+            return False
         await self.db.log_session_event(session_id, "idle", {
             "resumable": preserve_sdk_id,
         })
+        return True
 
-    async def mark_stopped(self, session_id: str) -> None:
+    async def mark_stopped(self, session_id: str) -> bool:
         """Mark session as user-stopped."""
-        await self.db.update_session_fields(
+        if not await self.db.update_unarchived_session_fields(
             session_id, {"status": SessionStatus.STOPPED.value},
-        )
+        ):
+            return False
         await self.db.log_session_event(session_id, "stopped", {})
+        return True
 
-    async def mark_error(self, session_id: str, error_msg: str) -> None:
+    async def mark_error(self, session_id: str, error_msg: str) -> bool:
         """Mark session as errored (SDK crashed)."""
-        await self.db.update_session_fields(session_id, {
+        if not await self.db.update_unarchived_session_fields(session_id, {
             "status": SessionStatus.ERROR.value,
             "sdk_session_id": None,
             "connected_at": None,
-        })
+        }):
+            return False
         await self.db.log_session_event(session_id, "error", {
             "error": error_msg[:500],
         })
+        return True
 
     # ------------------------------------------------------------------ #
     #  Channel mapping (DB-persisted)                                      #
@@ -392,8 +411,14 @@ def _is_within_sticky_period(self, session: dict) -> bool:
         Idle sessions fall back to the time-based check so channels
         that have been quiet for longer than sticky_period_minutes
         roll over to a new session as before.
+
+        An archived session is never sticky: the next message of the
+        channel starts a new session.
         """
-        if session.get("status") == SessionStatus.ACTIVE.value:
+        status = session.get("status")
+        if status == SessionStatus.ARCHIVED.value:
+            return False
+        if status == SessionStatus.ACTIVE.value:
             return True
         ts = session.get("last_activity_at") or session.get("updated_at")
         if not ts:
@@ -995,48 +1020,93 @@ async def archive_session(
         agent still running in it. Unless ``force``, a session with live
         agents is refused with :class:`SessionBusyError`; with ``force`` its
         agents end as interrupted (``archived``).
+
+        A session with a turn that holds its run lock (it runs, or waits for
+        a concurrency slot) is refused with :class:`SessionBusyError`, also
+        with ``force``. The check and the status write hold the session's
+        client lock, the lock under which a turn builds its client and checks
+        that the session is not archived. So the runtime never gets a turn of
+        a session that this call archives.
+
+        After the archive, the session's pending wakeups are deleted and
+        ``on_archived`` runs (it expires the session's pending questions).
         """
         session = await self.db.get_session(session_id)
         if not session:
             raise ValueError(f"Session {session_id} not found")
         if session.get("status") == SessionStatus.ARCHIVED.value:
             return False
-        if not force and self._agents_live(session_id):
-            raise SessionBusyError(BUSY_AGENTS_REASON)
-        # Memorize in the background: memU indexing can take ~90s; awaiting it
-        # here would hang the archive (and, in a cascade, the whole request on the root).
-        if self._on_memorize:
-            async def _memorize_bg(sid: str = session_id) -> None:
+        lock = self.get_lock(session_id)
+        async with lock:
+            # Read again under the lock: another archive that held it can
+            # have archived the session meanwhile.
+            session = await self.db.get_session(session_id)
+            if not session or session.get("status") == SessionStatus.ARCHIVED.value:
+                return False
+            if self.has_running_work(session_id):
+                raise SessionBusyError(BUSY_RUNNING_REASON)
+            if not force and self._agents_live(session_id):
+                raise SessionBusyError(BUSY_AGENTS_REASON)
+            # Memorize in the background: memU indexing can take ~90s; awaiting it
+            # here would hang the archive (and, in a cascade, the whole request on the root).
+            if self._on_memorize:
+                async def _memorize_bg(sid: str = session_id) -> None:
+                    try:
+                        await self._on_memorize(sid)
+                    except Exception as e:
+                        logger.warning("Background memorize for %s failed: %s", sid, e)
+                asyncio.create_task(_memorize_bg())
+
+            client = self.remove_client(session_id, reason="archived")
+            # remove_client drops the session's client lock from the map. Keep
+            # the held lock there until the status is written, so that a turn
+            # that builds a client meanwhile waits for it.
+            self._client_locks[session_id] = lock
+            if client is None and self._agents_live(session_id):
+                # Records still live without a runtime: nothing runs them any
+                # more, so settle them with the archive.
+                self._notify_client_removed(session_id, "archived")
+            if client:
                 try:
-                    await self._on_memorize(sid)
-                except Exception as e:
-                    logger.warning("Background memorize for %s failed: %s", sid, e)
-            asyncio.create_task(_memorize_bg())
-
-        client = self.remove_client(session_id, reason="archived")
-        if client is None and self._agents_live(session_id):
-            # Records still live without a runtime: nothing runs them any
-            # more, so settle them with the archive.
-            self._notify_client_removed(session_id, "archived")
-        if client:
-            try:
-                await client.disconnect()
-            except Exception:
-                pass
-        now = datetime.now(timezone.utc).isoformat()
-        archived = await self.db.change_session_status(
-            session_id, SessionStatus.ARCHIVED.value,
-            fields={
-                "archived_at": now,
-                "connected_at": None,
-                "sdk_session_id": None,
-            },
-            event_type="archived", details=details, attribution=attribution,
-        )
+                    await client.disconnect()
+                except Exception:
+                    pass
+            now = datetime.now(timezone.utc).isoformat()
+            archived = await self.db.change_session_status(
+                session_id, SessionStatus.ARCHIVED.value,
+                fields={
+                    "archived_at": now,
+                    "connected_at": None,
+                    "sdk_session_id": None,
+                },
+                event_type="archived", details=details, attribution=attribution,
+            )
+        if self._client_locks.get(session_id) is lock:
+            self._client_locks.pop(session_id)
         if archived:
             logger.info("Archived session %s", session_id)
+            await self._close_archived_work(session_id)
         return archived
 
+    async def _close_archived_work(self, session_id: str) -> None:
+        """End the work that targets a session that was just archived: its
+        pending wakeups (deleted) and, through ``on_archived``, its pending
+        questions. Best effort: a failure is logged, the archive stands."""
+        try:
+            cancelled = await self.db.cancel_wakeups_for_session(session_id)
+            if cancelled:
+                logger.info(
+                    "Archive of %s: deleted %d pending wakeup(s)",
+                    session_id, cancelled,
+                )
+        except Exception as e:
+            logger.warning("Archive of %s: wakeup cancel failed: %s", session_id, e)
+        if self.on_archived is not None:
+            try:
+                await self.on_archived(session_id)
+            except Exception as e:
+                logger.warning("Archive of %s: on_archived failed: %s", session_id, e)
+
     async def _resolve_descendants(self, root_id: str) -> list[str]:
         """Every session in ``root_id``'s subtree, discovery order (root first).
 
@@ -1116,8 +1186,8 @@ async def archive_session_cascade(
                     force=force,
                 )
                 archived.append(sid)
-            except SessionBusyError:  # agents started meanwhile
-                skipped[sid] = BUSY_AGENTS_REASON
+            except SessionBusyError as e:  # a turn or agents started meanwhile
+                skipped[sid] = str(e)
             except Exception as e:  # keep going; sibling subtrees still complete
                 logger.warning("Cascade archive failed for %s: %s", sid, e)
                 skipped[sid] = str(e)
@@ -1196,8 +1266,8 @@ async def run_cleanup(
         When ``interactive_archive_after_hours`` > 0 (opt-in; 0 disables and leaves the
         default behavior unchanged), interactive sessions (web/telegram/…)
         idle longer than that are archived and memorized promptly — including
-        ones parked on an unanswered ``ask_user`` question, whose pending
-        notification is expired so it doesn't dangle on a closed session.
+        ones parked on an unanswered ``ask_user`` question. Every archive
+        expires the session's pending questions (see :meth:`archive_session`).
         Cron/persistent sessions are excluded from the short cutoff and only
         hit the ``archive_after_days`` backstop. Starred sessions are exempt
         from every auto-archival path (short cutoff, backstop, and the
@@ -1212,17 +1282,16 @@ async def run_cleanup(
         system = attribution_for(self.db.system_actor)
 
         async def _archive(sid: str, reason: str) -> bool:
-            """Archive one session; a session whose background agents still
-            run is skipped (an automatic archive never ends them)."""
+            """Archive one session; a session with a running turn or with
+            background agents is skipped (an automatic archive never ends
+            them)."""
             try:
                 await self.archive_session(
                     sid, attribution=system, details={"reason": reason},
                 )
                 return True
-            except SessionBusyError:
-                logger.info(
-                    "Session cleanup: skipping %s (%s)", sid, BUSY_AGENTS_REASON,
-                )
+            except SessionBusyError as e:
+                logger.info("Session cleanup: skipping %s (%s)", sid, e)
                 return False
 
         # Opt-in short idle cutoff for interactive sessions. Skipped entirely
@@ -1239,12 +1308,6 @@ async def _archive(sid: str, reason: str) -> bool:
                         "Session cleanup: skipping %s (%s)", s["id"], BUSY_AGENTS_REASON,
                     )
                     continue
-                try:
-                    await self.db.expire_pending_questions_for_session(s["id"])
-                except Exception as e:
-                    logger.warning(
-                        "Expire pending questions failed for %s: %s", s["id"], e,
-                    )
                 if await _archive(s["id"], "idle_interactive"):
                     archived_interactive += 1
 
diff --git a/nerve/channels/router.py b/nerve/channels/router.py
index 639613cc2..585191b34 100644
--- a/nerve/channels/router.py
+++ b/nerve/channels/router.py
@@ -18,6 +18,7 @@
 from dataclasses import asdict
 from typing import Any, TYPE_CHECKING
 
+from nerve.agent.engine import SessionClosedError
 from nerve.agent.interactive import get_handler
 from nerve.agent.streaming import broadcaster
 from nerve.channels.access import Decision
@@ -200,7 +201,24 @@ async def handle_message(self, msg: InboundMessage) -> str:
         separate turn, in arrival order.
 
         Called by channel implementations when they receive a user message.
+
+        A message that the channel's session takes, but that run() refuses
+        because the session was archived meanwhile, is sent again once: the
+        channel then gets a new session, as after the sticky period.
         """
+        try:
+            return await self._handle_message(msg)
+        except SessionClosedError as exc:
+            if msg.session_id or exc.reason != "archived":
+                raise
+            logger.info(
+                "Channel %s: session archived before the message ran; "
+                "sending it to a new session", msg.channel_key,
+            )
+            return await self._handle_message(msg)
+
+    async def _handle_message(self, msg: InboundMessage) -> str:
+        """:meth:`handle_message` for one session resolution."""
         channel = self._channels.get(msg.channel_name)
         if not channel:
             raise ValueError(f"Unknown channel: {msg.channel_name}")
diff --git a/nerve/cron/service.py b/nerve/cron/service.py
index 2b62ad85d..38caf273d 100644
--- a/nerve/cron/service.py
+++ b/nerve/cron/service.py
@@ -27,7 +27,7 @@
 from apscheduler.triggers.interval import IntervalTrigger
 
 from nerve import paths
-from nerve.agent.engine import AgentEngine, NotAcceptingTurnsError
+from nerve.agent.engine import AgentEngine, NotAcceptingTurnsError, SessionClosedError
 from nerve.agent.sessions import new_cron_run_id
 from nerve.config import ConfigError, NerveConfig, _is_within, workspace_reviewed_paths
 from nerve.cron.jobs import (
@@ -2215,7 +2215,12 @@ def _dispatch_wakeup(self, session_id: str, wakeup: dict) -> None:
 
         def _done(t: asyncio.Task) -> None:
             exc = t.exception() if not t.cancelled() else None
-            if exc is not None:
+            if isinstance(exc, SessionClosedError):
+                logger.info(
+                    "Wakeup %s dropped: session %s is %s",
+                    wakeup["id"], session_id, exc.reason,
+                )
+            elif exc is not None:
                 logger.error(
                     "Wakeup %s run failed for session %s: %s",
                     wakeup["id"], session_id, exc,
diff --git a/nerve/db/notifications.py b/nerve/db/notifications.py
index 3d34463d3..11531a2a9 100644
--- a/nerve/db/notifications.py
+++ b/nerve/db/notifications.py
@@ -395,28 +395,33 @@ async def expire_notification(self, notification_id: str) -> bool:
             )
         return True
 
-    async def expire_pending_questions_for_session(self, session_id: str) -> int:
+    async def expire_pending_questions_for_session(
+        self, session_id: str,
+    ) -> list[dict]:
         """Expire a session's pending ``question`` notifications.
 
-        Called when an idle session is auto-archived so the user isn't left
-        with a phantom pending question on a closed session (and the periodic
-        expiry pass has nothing to inject into the now-archived session).
-        Returns the number of questions expired.
+        Called when a session is archived or deleted, so that no question
+        stays answerable for a session that runs no more turns. Approvals
+        stay pending: their answer acts on their target, not on the session.
+        Returns the expired rows, with ``status`` set to ``'expired'``, for
+        the service layer to report (as :meth:`expire_due_notifications`).
         """
         async with self._atomic():
             async with self.db.execute(
-                """SELECT id FROM notifications
+                """SELECT * FROM notifications
                    WHERE session_id = ? AND status = 'pending' AND type = 'question'""",
                 (session_id,),
             ) as cursor:
-                ids = [row[0] async for row in cursor]
-            if ids:
-                ph = ",".join("?" for _ in ids)
+                rows = [dict(row) async for row in cursor]
+            if rows:
+                ph = ",".join("?" for _ in rows)
                 await self.db.execute(
                     f"UPDATE notifications SET status = 'expired' WHERE id IN ({ph})",
-                    tuple(ids),
+                    tuple(r["id"] for r in rows),
                 )
-        return len(ids)
+        for r in rows:
+            r["status"] = "expired"
+        return rows
 
     async def snooze_notification(
         self, notification_id: str, redeliver_at: str, new_expires_at: str,
diff --git a/nerve/db/sessions.py b/nerve/db/sessions.py
index 234d472bf..9632ba8ce 100644
--- a/nerve/db/sessions.py
+++ b/nerve/db/sessions.py
@@ -530,6 +530,23 @@ async def update_session_fields(self, session_id: str, fields: dict) -> None:
             return
         await self._write(*built)
 
+    async def update_unarchived_session_fields(
+        self, session_id: str, fields: dict,
+    ) -> bool:
+        """Like :meth:`update_session_fields`, but only for a session that is
+        not archived.
+
+        The runtime's status writes use it, so that only unarchive takes a
+        session out of ``archived``. Returns ``True`` when a row was written,
+        ``False`` for an archived or missing session.
+        """
+        built = self._build_session_fields_update(session_id, fields)
+        if built is None:
+            return False
+        sql, params = built
+        result = await self._write(f"{sql} AND status != 'archived'", params)
+        return (result.rowcount or 0) > 0
+
     async def change_session_fields(
         self,
         session_id: str,
diff --git a/nerve/gateway/routes/sessions.py b/nerve/gateway/routes/sessions.py
index c032936c3..687f12929 100644
--- a/nerve/gateway/routes/sessions.py
+++ b/nerve/gateway/routes/sessions.py
@@ -701,55 +701,81 @@ async def delete_session(session_id: str, actor: Actor = Depends(require_auth)):
     db = deps.db
 
     if engine:
-        # Disconnect client
-        client = engine.sessions.remove_client(session_id, reason="deleted")
-        if client:
-            try:
-                await client.disconnect()
-            except Exception:
-                pass
-        # Drop the session's agent records, queued writes and pending emits
-        # before the rows go (a write already in flight fails the foreign key
-        # and is dropped).
-        agents = getattr(engine, "agents", None)
-        if agents is not None:
-            await agents.forget_session(session_id)
-        # Snapshot messages for background memorization before deleting
-        session = await db.get_session(session_id)
-        connected_at = session.get("connected_at") if session else None
-        messages = await db.get_messages(session_id, limit=10000) if connected_at else []
-        # Delete from DB immediately (fast). The delete is recorded with
-        # the person, because the session's own events go with it.
-        await db.delete_session(session_id, attribution=attribution_for(actor))
-        # Cleanup uploaded files for this session
-        await _cleanup_uploaded_files(db, session_id)
-        # Memorize in background from snapshot
-        if messages and connected_at and engine._memory_bridge and engine._memory_bridge.available:
-            async def _bg_memorize():
-                try:
-                    context_msgs = []
-                    for msg in messages:
-                        created = msg.get("created_at", "")
-                        if created:
-                            norm = created if "T" in created else created.replace(" ", "T") + "Z"
-                            if norm >= connected_at:
-                                context_msgs.append(msg)
-                    if context_msgs:
-                        with keep_awake(LEASE_MEMORIZE):
-                            await engine._memory_bridge.memorize_conversation(
-                                session_id, context_msgs,
-                            )
-                        # Same window to xmemory (opt-in, fire-and-forget).
-                        engine.schedule_xmemory_transcript(session_id, context_msgs)
-                except Exception as e:
-                    logger.warning("Background memorize for deleted session %s failed: %s", session_id, e)
-            asyncio.create_task(_bg_memorize())
+        # A turn that runs or waits would write into the deleted rows, or
+        # create the session again. The person stops it first.
+        if not engine.begin_delete(session_id):
+            raise HTTPException(status_code=409, detail=_DELETE_BUSY_DETAIL)
+        try:
+            await _delete_with_engine(engine, db, session_id, actor)
+        finally:
+            engine.end_delete(session_id)
     else:
         await db.delete_session(session_id, attribution=attribution_for(actor))
         await _cleanup_uploaded_files(db, session_id)
     return {"deleted": True}
 
 
+_DELETE_BUSY_DETAIL = (
+    "The session is running or has queued messages. Stop it, then delete it."
+)
+
+
+async def _delete_with_engine(engine, db, session_id: str, actor: Actor) -> None:
+    """Delete a session that has no turn work: its client, agent records,
+    rows and uploads; memorize it from a snapshot; expire its questions."""
+    # Disconnect client
+    client = engine.sessions.remove_client(session_id, reason="deleted")
+    if client:
+        try:
+            await client.disconnect()
+        except Exception:
+            pass
+    # Drop the session's agent records, queued writes and pending emits
+    # before the rows go (a write already in flight fails the foreign key
+    # and is dropped).
+    agents = getattr(engine, "agents", None)
+    if agents is not None:
+        await agents.forget_session(session_id)
+    # Snapshot messages for background memorization before deleting
+    session = await db.get_session(session_id)
+    connected_at = session.get("connected_at") if session else None
+    messages = await db.get_messages(session_id, limit=10000) if connected_at else []
+    # Delete from DB immediately (fast). The delete is recorded with
+    # the person, because the session's own events go with it.
+    await db.delete_session(session_id, attribution=attribution_for(actor))
+    # Cleanup uploaded files for this session
+    await _cleanup_uploaded_files(db, session_id)
+    # Memorize in background from snapshot
+    if messages and connected_at and engine._memory_bridge and engine._memory_bridge.available:
+        async def _bg_memorize():
+            try:
+                context_msgs = []
+                for msg in messages:
+                    created = msg.get("created_at", "")
+                    if created:
+                        norm = created if "T" in created else created.replace(" ", "T") + "Z"
+                        if norm >= connected_at:
+                            context_msgs.append(msg)
+                if context_msgs:
+                    with keep_awake(LEASE_MEMORIZE):
+                        await engine._memory_bridge.memorize_conversation(
+                            session_id, context_msgs,
+                        )
+                    # Same window to xmemory (opt-in, fire-and-forget).
+                    engine.schedule_xmemory_transcript(session_id, context_msgs)
+            except Exception as e:
+                logger.warning("Background memorize for deleted session %s failed: %s", session_id, e)
+        asyncio.create_task(_bg_memorize())
+    # No question stays answerable for a deleted session. The rows stay
+    # as history.
+    try:
+        await engine.expire_session_questions(session_id)
+    except Exception as e:
+        logger.warning(
+            "Expire questions of deleted session %s failed: %s", session_id, e,
+        )
+
+
 @router.get("/api/sessions/{session_id}/status")
 async def session_status(session_id: str, actor: Actor = Depends(require_auth)):
     """Enhanced session status with lifecycle info."""
@@ -1119,6 +1145,10 @@ async def chat(req: MessageRequest, actor: Actor = Depends(require_auth)):
         await deps.engine.sessions.get_or_create(
             session_id, source="web", actor=actor, origin=Origin(TRIGGER_API),
         )
+        # The caller's own message into an archived session opens it again.
+        await deps.engine.reopen_archived(
+            session_id, attribution=attribution_for(actor),
+        )
     response = await deps.engine.run(
         session_id=session_id,
         user_message=req.message,
diff --git a/nerve/gateway/server.py b/nerve/gateway/server.py
index cf45b5c3c..19b40bb3a 100644
--- a/nerve/gateway/server.py
+++ b/nerve/gateway/server.py
@@ -31,7 +31,7 @@
 
 from nerve import paths
 from nerve.agent.agents.emitter import agent_channel
-from nerve.agent.engine import AgentEngine, NotAcceptingTurnsError
+from nerve.agent.engine import AgentEngine, NotAcceptingTurnsError, SessionClosedError
 from nerve.agent.streaming import broadcaster
 from nerve.config import EXTERNAL_VERIFICATION_LOG, NerveConfig, get_config
 from nerve.db import Database, init_db, close_db
@@ -185,7 +185,8 @@ def _connections_changed() -> None:
 
 
 def _ws_run_done(task: asyncio.Task) -> None:
-    """Forget a finished WebSocket run. A run refused by a drain is no error."""
+    """Forget a finished WebSocket run. A run refused by a drain or by a
+    closed session is no error."""
     _ws_runs.discard(task)
     if task.cancelled():
         return
@@ -193,6 +194,9 @@ def _ws_run_done(task: asyncio.Task) -> None:
     if isinstance(exc, NotAcceptingTurnsError):
         # The engine already told the session's clients.
         logger.info("Draining: refused a queued web turn")
+    elif isinstance(exc, SessionClosedError):
+        # The engine already told the session's clients.
+        logger.info("Refused a web turn: session %s", exc.reason)
     elif exc is not None:
         logger.error("Web turn failed", exc_info=exc)
 
@@ -1207,6 +1211,15 @@ async def _not_accepting_turns_handler(  # noqa: ANN001
     ):
         return JSONResponse(status_code=503, content={"detail": str(exc)})
 
+    # A route that starts a turn in an archived or deleted session.
+    @app.exception_handler(SessionClosedError)
+    async def _session_closed_handler(  # noqa: ANN001
+        request, exc: SessionClosedError,
+    ):
+        return JSONResponse(
+            status_code=409, content={"detail": str(exc), "reason": exc.reason},
+        )
+
     # FastAPI validation errors echo the rejected input, which can include the
     # setup token. Return them without the input.
     @app.exception_handler(RequestValidationError)
@@ -1385,6 +1398,13 @@ async def ws_broadcast(session_id: str, message: dict):
                         })
                         continue
 
+                    # The person's own message into an archived chat opens
+                    # it again (an ``unarchived`` event in their name). Every
+                    # other run in an archived session is refused.
+                    await _engine.reopen_archived(
+                        session_id, attribution=attribution_for(connection.actor),
+                    )
+
                     # Load uploaded files if any
                     images = None
                     image_refs = None
diff --git a/nerve/notifications/service.py b/nerve/notifications/service.py
index 836e3f790..8f13611c2 100644
--- a/nerve/notifications/service.py
+++ b/nerve/notifications/service.py
@@ -561,7 +561,8 @@ async def handle_answer(
           maintenance tick (:meth:`redeliver_due`) fans it out again
           with a fresh card.
         - For ``type=question`` rows (legacy): persist the answer,
-          inject it back into the originating session, broadcast.
+          inject it back into the originating session, broadcast. For an
+          archived or deleted session, persist and broadcast only: no turn.
         - Fire-and-forget ``type=notify`` rows do not flow through this
           method; the dismiss endpoint handles those.
 
@@ -589,10 +590,16 @@ async def handle_answer(
         is_external = bool(
             session_record and session_record.get("source") == "external"
         )
+        # An archived or deleted session runs no more turns: the answer is
+        # stored and shown, but nothing is delivered to the session.
+        is_closed = (
+            session_record is None
+            or session_record.get("status") == "archived"
+        )
 
         # During a drain the turn that delivers the answer is refused. Keep
         # the question pending, so the person can answer after the restart.
-        if not is_external and not self.engine.accepting_turns:
+        if not is_external and not is_closed and not self.engine.accepting_turns:
             from nerve.agent.engine import NotAcceptingTurnsError
 
             raise NotAcceptingTurnsError()
@@ -610,7 +617,13 @@ async def handle_answer(
 
         from nerve.agent.streaming import broadcaster
 
-        if is_external:
+        if is_closed:
+            logger.info(
+                "Answer to %s stored; session %s is %s, no turn",
+                notification_id, session_id,
+                "missing" if session_record is None else "archived",
+            )
+        if is_external or is_closed:
             await broadcaster.broadcast("__global__", {
                 "type": "notification_answered",
                 "notification_id": notification_id,
@@ -938,7 +951,11 @@ def _on_answer_task_done(self, task: asyncio.Task) -> None:
         if task.cancelled():
             return
         exc = task.exception()
-        if exc:
+        from nerve.agent.engine import SessionClosedError
+
+        if isinstance(exc, SessionClosedError):
+            logger.info("Answer injection dropped: %s", exc)
+        elif exc:
             logger.error("Answer injection task failed: %s", exc)
 
     async def handle_dismiss(
@@ -1661,6 +1678,23 @@ async def expire_stale(self) -> int:
             await self._report_expired(rows)
         return len(rows)
 
+    async def expire_session_questions(self, session_id: str) -> int:
+        """Expire the pending questions of a session that was archived or
+        deleted, and report each one as :meth:`expire_stale` does (web
+        broadcast, Telegram and Slack card edits). The session gets no note:
+        :meth:`_takes_note` refuses an archived or missing session.
+        Approvals stay pending, because their answer acts on their target,
+        not on the session. Returns the number of expired questions.
+        """
+        rows = await self.db.expire_pending_questions_for_session(session_id)
+        if rows:
+            logger.info(
+                "Expired %d pending question(s) of closed session %s",
+                len(rows), session_id,
+            )
+            await self._report_expired(rows)
+        return len(rows)
+
     async def _report_expired(self, rows: list[dict[str, Any]]) -> None:
         """Report expired questions/approvals to every interested party.
 
diff --git a/tests/test_closed_sessions.py b/tests/test_closed_sessions.py
new file mode 100644
index 000000000..b0da0a4c2
--- /dev/null
+++ b/tests/test_closed_sessions.py
@@ -0,0 +1,670 @@
+"""Archived and deleted sessions stay closed (ClickHouse/nerve#671).
+
+A session in ``archived`` leaves that state only through unarchive. A run
+never creates a row for a deleted session id. Work that targets an archived or
+deleted session is refused or cancelled. The tests drive the real
+``AgentEngine`` (turn loop, run lock, client build, SQLite store) with a
+scripted backend in place of the CLI.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+import pytest_asyncio
+from fastapi import HTTPException
+
+from nerve.agent.backends import events as ev
+from nerve.agent.backends.base import BackendCapabilities
+from nerve.agent.engine import AgentEngine, SessionClosedError
+from nerve.agent.sessions import BUSY_RUNNING_REASON, SessionBusyError
+from nerve.config import NerveConfig, NotificationsConfig
+from nerve.cron.service import CronService
+from nerve.identity import attribution_for
+from nerve.notifications.service import NotificationService
+
+from tests.actor_rows import ensure_system_principal
+
+
+@pytest_asyncio.fixture
+async def db(db):  # noqa: F811 - the conftest database, with the system actor
+    await ensure_system_principal(db)
+    return db
+
+
+@pytest.fixture(autouse=True)
+def broadcasts(monkeypatch):
+    """Every broadcast of the test, as ``(channel, message)``."""
+    captured: list[tuple[str, dict]] = []
+
+    async def _broadcast(channel, message, *args, **kwargs):
+        captured.append((channel, message))
+
+    from nerve.agent.streaming import broadcaster
+    monkeypatch.setattr(broadcaster, "broadcast", _broadcast)
+    return captured
+
+
+class _Client:
+    """A runtime client: each turn streams one text delta, then ends, or
+    waits until ``release`` when ``hold`` is set."""
+
+    def __init__(self, native_id: str, spec) -> None:
+        self.spec = spec
+        self._native_id = native_id
+        self.model = "claude-test"
+        self.turns: list[str] = []
+        self.hold = False
+        self.release = asyncio.Event()
+        self.closed = asyncio.Event()
+        self.got_content = asyncio.Event()
+        self.disconnect_gate: asyncio.Event | None = None
+
+    @property
+    def native_session_id(self) -> str:
+        return self._native_id
+
+    async def connect(self) -> None:
+        pass
+
+    async def start_turn(self, turn) -> None:
+        self.turns.append(turn.text)
+
+    async def receive_turn(self):
+        yield ev.TextDelta(text=f"reply-from-{self._native_id}")
+        self.got_content.set()
+        if self.hold:
+            await self.release.wait()
+        yield ev.TurnCompleted(
+            native_session_id=self._native_id,
+            usage=ev.NormalizedUsage(input_tokens=1, output_tokens=1),
+        )
+
+    async def interrupt(self, **_kw) -> None:
+        self.release.set()
+
+    def is_alive(self) -> bool:
+        return not self.closed.is_set()
+
+    async def disconnect(self, timeout: float = 5.0) -> None:
+        if self.disconnect_gate is not None:
+            await self.disconnect_gate.wait()
+        self.closed.set()
+        self.release.set()
+
+    def try_receive_idle_events(self):
+        return None
+
+    def buffer_used(self) -> int:
+        return 0
+
+
+class _Backend:
+    name = "claude"
+
+    def __init__(self) -> None:
+        self.capabilities = BackendCapabilities(
+            supports_idle_stream=False, supports_cache_ttl=False,
+        )
+        self.clients: list[_Client] = []
+        self.hold_new_clients = False
+
+    def default_model(self, source):
+        return "claude-test"
+
+    def excluded_tools(self):
+        return set()
+
+    def validate_resume_target(self, native_id, cwd):
+        return True
+
+    async def create_client(self, spec):
+        client = _Client(f"native-{len(self.clients)}", spec)
+        client.hold = self.hold_new_clients
+        self.clients.append(client)
+        return client
+
+
+def _engine(tmp_path, db, *, max_concurrent: int = 3) -> tuple[AgentEngine, _Backend]:
+    ws = tmp_path / "ws"
+    ws.mkdir(exist_ok=True)
+    cfg = NerveConfig.from_dict({
+        "workspace": str(ws),
+        "agent": {
+            "backend": "claude", "effort": "high", "cron_effort": "low",
+            "max_concurrent": max_concurrent,
+        },
+    })
+    engine = AgentEngine(cfg, db)
+    backend = _Backend()
+    engine._backends["claude"] = backend
+    return engine, backend
+
+
+def _cron(engine: AgentEngine, db) -> CronService:
+    config = MagicMock()
+    config.timezone = "UTC"
+    return CronService(config, engine, db)
+
+
+def _notifications(engine: AgentEngine, db, tmp_path: Path) -> NotificationService:
+    cfg = NerveConfig()
+    cfg.workspace = tmp_path
+    cfg.notifications = NotificationsConfig(
+        channels=["web"], telegram_chat_id=None, default_expiry_hours=48,
+        priority_prefixes={"high": "", "urgent": ""},
+    )
+    service = NotificationService(cfg, db, engine)
+    engine.set_notification_service(service)
+    return service
+
+
+async def _until(pred, timeout: float = 5.0) -> None:
+    async def _loop():
+        while not pred():
+            await asyncio.sleep(0.01)
+    await asyncio.wait_for(_loop(), timeout=timeout)
+
+
+def _past() -> str:
+    return (datetime.now(timezone.utc) - timedelta(minutes=1)).isoformat()
+
+
+async def _events(db, sid: str) -> list[str]:
+    return [e["event_type"] for e in await db.get_session_events(sid, limit=100)]
+
+
+async def _new_session(engine: AgentEngine, db, sid: str) -> None:
+    """A web session with one completed turn (it has a native conversation)."""
+    await engine.sessions.get_or_create(sid, source="web", backend="claude", actor=None)
+    await engine.run(sid, "first message", source="web", channel="web", actor=None)
+
+
+def _queue_removals(broadcasts) -> list[dict]:
+    return [
+        removed
+        for _channel, msg in broadcasts if msg.get("type") == "queue_update"
+        for removed in msg.get("removed") or ()
+    ]
+
+
+# --------------------------------------------------------------------------- #
+# Archive: wakeups, status writes, the run lock
+# --------------------------------------------------------------------------- #
+
+@pytest.mark.asyncio
+async def test_archive_deletes_pending_wakeups_and_the_sweep_fires_nothing(tmp_path, db):
+    engine, backend = _engine(tmp_path, db)
+    sid = "s-loop"
+    await _new_session(engine, db, sid)
+    await db.add_wakeup(sid, prompt="check the deploy", fire_at=_past())
+
+    result = await engine.sessions.archive_session_cascade(sid)
+
+    assert result["archived"] == [sid]
+    assert await db.list_pending_wakeups(sid) == []
+    await _cron(engine, db)._sweep_wakeups()
+    await asyncio.sleep(0.05)
+    assert len(backend.clients) == 1
+    row = await db.get_session(sid)
+    assert row["status"] == "archived"
+    assert "unarchived" not in await _events(db, sid)
+
+
+@pytest.mark.asyncio
+async def test_a_wakeup_of_an_archived_session_is_refused(tmp_path, db, broadcasts):
+    """A wakeup that is pending on an archived session runs nothing. (The
+    archive deletes pending wakeups, so this one is added after it.)"""
+    engine, backend = _engine(tmp_path, db)
+    sid = "s-old-archive"
+    await _new_session(engine, db, sid)
+    await engine.sessions.archive_session(sid)
+    await db.add_wakeup(sid, prompt="check the deploy", fire_at=_past())
+    messages_before = len(await db.get_messages(sid))
+
+    with pytest.raises(SessionClosedError) as info:
+        await engine.run(sid, "check the deploy", source="wakeup", internal=True, actor=None)
+
+    assert info.value.reason == "archived"
+    assert len(backend.clients) == 1
+    assert len(await db.get_messages(sid)) == messages_before
+    assert (await db.get_session(sid))["status"] == "archived"
+    assert any(
+        msg.get("type") == "error" and "archived" in msg.get("error", "")
+        for _c, msg in broadcasts
+    )
+
+
+@pytest.mark.asyncio
+async def test_runtime_status_writes_keep_an_archived_session_archived(tmp_path, db):
+    engine, _backend = _engine(tmp_path, db)
+    sid = "s-marks"
+    await engine.sessions.get_or_create(sid, source="web", actor=None)
+    await engine.sessions.archive_session(sid)
+    before = await _events(db, sid)
+
+    assert await engine.sessions.mark_active(sid, sdk_session_id="native-x") is False
+    assert await engine.sessions.mark_idle(sid) is False
+    assert await engine.sessions.mark_stopped(sid) is False
+    assert await engine.sessions.mark_error(sid, "boom") is False
+
+    row = await db.get_session(sid)
+    assert row["status"] == "archived"
+    assert row["sdk_session_id"] is None
+    assert await _events(db, sid) == before
+
+
+@pytest.mark.asyncio
+async def test_runtime_status_writes_still_work_for_an_open_session(tmp_path, db):
+    engine, _backend = _engine(tmp_path, db)
+    sid = "s-open"
+    await engine.sessions.get_or_create(sid, source="web", actor=None)
+
+    assert await engine.sessions.mark_active(sid, sdk_session_id="native-x") is True
+    assert (await db.get_session(sid))["status"] == "active"
+    assert await engine.sessions.mark_idle(sid) is True
+    assert (await db.get_session(sid))["status"] == "idle"
+    assert "started" in await _events(db, sid)
+
+
+@pytest.mark.asyncio
+async def test_archive_is_refused_while_a_turn_waits_for_a_concurrency_slot(tmp_path, db):
+    engine, backend = _engine(tmp_path, db, max_concurrent=1)
+    sid = "s-slot"
+    await engine.sessions.get_or_create(sid, source="web", backend="claude", actor=None)
+    # Another session holds the only slot.
+    await engine._semaphore.acquire()
+    turn = asyncio.create_task(
+        engine.run(sid, "waits for a slot", source="web", channel="web", actor=None),
+    )
+    await _until(lambda: engine.sessions.turn_task(sid) is not None)
+    assert not engine.sessions.is_running(sid)
+
+    with pytest.raises(SessionBusyError) as info:
+        await engine.sessions.archive_session(sid)
+    assert str(info.value) == BUSY_RUNNING_REASON
+    result = await engine.sessions.archive_session_cascade(sid)
+    assert result["archived"] == []
+    assert result["skipped"] == [{"id": sid, "reason": BUSY_RUNNING_REASON}]
+
+    engine._semaphore.release()
+    await asyncio.wait_for(turn, timeout=5)
+    assert (await db.get_session(sid))["status"] == "active"
+    assert backend.clients[0].turns
+
+
+@pytest.mark.asyncio
+async def test_a_queued_turn_is_refused_when_the_session_was_archived_meanwhile(
+    tmp_path, db, broadcasts,
+):
+    engine, backend = _engine(tmp_path, db)
+    sid = "s-queued"
+    await _new_session(engine, db, sid)
+    # The run lock is held by work that is not a turn (an idle drain or a
+    # client reap), so the next message waits in the queue.
+    lock = engine._session_locks[sid]
+    await lock.acquire()
+    turn = asyncio.create_task(
+        engine.run(sid, "queued message", source="web", channel="web",
+                   actor=None, client_msg_id="m-1"),
+    )
+    await _until(lambda: len(engine.queue_view(sid)) == 1)
+
+    assert (await engine.sessions.archive_session_cascade(sid))["archived"] == [sid]
+    engine._pass_run_lock(sid, lock)
+
+    with pytest.raises(SessionClosedError):
+        await asyncio.wait_for(turn, timeout=5)
+    assert len(backend.clients) == 1
+    assert (await db.get_session(sid))["status"] == "archived"
+    users = [m["content"] for m in await db.get_messages(sid) if m["role"] == "user"]
+    assert "queued message" not in users
+    removal = _queue_removals(broadcasts)[-1]
+    assert (removal["client_msg_id"], removal["reason"]) == ("m-1", "refused")
+    assert any(
+        msg.get("type") == "error" and msg.get("error") == str(SessionClosedError("archived"))
+        for _c, msg in broadcasts
+    )
+    assert not lock.locked()
+
+
+@pytest.mark.asyncio
+async def test_a_session_archived_while_its_turn_waits_for_the_client_gets_no_client(
+    tmp_path, db,
+):
+    """The check under the client lock: the archive lands after the checks
+    in ``_run_turn``, while the turn waits to build its client."""
+    engine, backend = _engine(tmp_path, db)
+    sid = "s-build"
+    await engine.sessions.get_or_create(sid, source="web", backend="claude", actor=None)
+    client_lock = engine.sessions.get_lock(sid)
+    await client_lock.acquire()
+    turn = asyncio.create_task(
+        engine.run(sid, "a message", source="web", channel="web", actor=None),
+    )
+    await _until(lambda: engine.sessions.is_running(sid))
+    await db.change_session_status(sid, "archived", event_type="archived")
+    client_lock.release()
+
+    with pytest.raises(SessionClosedError):
+        await asyncio.wait_for(turn, timeout=5)
+    assert backend.clients == []
+    row = await db.get_session(sid)
+    assert row["status"] == "archived"
+    assistant = [m for m in await db.get_messages(sid) if m["role"] == "assistant"]
+    assert assistant == []
+    assert "error" not in await _events(db, sid)
+
+
+@pytest.mark.asyncio
+async def test_a_turn_that_starts_during_an_archive_gets_no_client(tmp_path, db):
+    """The archive holds the client lock: a turn that passed its checks while
+    the archive disconnected the old client builds no new one."""
+    engine, backend = _engine(tmp_path, db)
+    sid = "s-mid-archive"
+    await _new_session(engine, db, sid)
+    gate = asyncio.Event()
+    backend.clients[0].disconnect_gate = gate
+    archive = asyncio.create_task(engine.sessions.archive_session(sid))
+    await _until(lambda: engine.sessions.get_client(sid) is None)
+    turn = asyncio.create_task(
+        engine.run(sid, "sent during the archive", source="web", channel="web", actor=None),
+    )
+    await _until(lambda: engine.sessions.is_running(sid))
+    gate.set()
+
+    assert await asyncio.wait_for(archive, timeout=5) is True
+    with pytest.raises(SessionClosedError):
+        await asyncio.wait_for(turn, timeout=5)
+    assert len(backend.clients) == 1
+    assert (await db.get_session(sid))["status"] == "archived"
+
+
+# --------------------------------------------------------------------------- #
+# Channels and direct messages
+# --------------------------------------------------------------------------- #
+
+@pytest.mark.asyncio
+async def test_a_channel_message_after_an_archive_gets_a_new_session(tmp_path, db):
+    engine, _backend = _engine(tmp_path, db)
+    key = "telegram:42"
+    first = await engine.sessions.get_active_session(key, source="telegram", actor=None)
+    await engine.sessions.mark_active(first)  # last activity: now
+    assert await engine.sessions.get_active_session(key, source="telegram", actor=None) == first
+
+    await engine.sessions.archive_session(first)
+    second = await engine.sessions.get_active_session(key, source="telegram", actor=None)
+
+    assert second != first
+    assert (await db.get_channel_session(key))["session_id"] == second
+    assert (await db.get_session(first))["status"] == "archived"
+
+
+@pytest.mark.asyncio
+async def test_the_router_sends_a_refused_message_again_to_a_new_session(tmp_path, db):
+    from nerve.channels.base import InboundMessage
+    from nerve.channels.router import ChannelRouter
+
+    router = ChannelRouter(MagicMock())
+    calls: list[str] = []
+
+    async def _handle(msg):
+        calls.append(msg.text)
+        if len(calls) == 1:
+            raise SessionClosedError("archived")
+        return "answer"
+
+    router._handle_message = _handle
+    msg = InboundMessage(
+        channel_name="telegram", channel_key="telegram:42", sender_id="42",
+        text="hello",
+    )
+    assert await router.handle_message(msg) == "answer"
+    assert calls == ["hello", "hello"]
+
+    # An explicit target is not sent anywhere else.
+    calls.clear()
+    explicit = InboundMessage(
+        channel_name="telegram", channel_key="telegram:42", sender_id="42",
+        text="hello", session_id="s-explicit",
+    )
+    with pytest.raises(SessionClosedError):
+        await router.handle_message(explicit)
+    assert calls == ["hello"]
+
+
+@pytest.mark.asyncio
+async def test_a_persons_message_unarchives_the_session_in_their_name(tmp_path, db):
+    engine, backend = _engine(tmp_path, db)
+    sid = "s-reopen"
+    await _new_session(engine, db, sid)
+    await engine.sessions.archive_session(sid)
+    person = db.system_actor
+
+    assert await engine.reopen_archived(sid, attribution=attribution_for(person)) is True
+    assert await engine.reopen_archived(sid, attribution=attribution_for(person)) is False
+    await engine.run(sid, "continue please", source="web", channel="web", actor=None)
+
+    row = await db.get_session(sid)
+    assert row["status"] == "active"
+    assert row["archived_at"] is None
+    events = await db.get_session_events(sid, limit=100)
+    unarchived = [e for e in events if e["event_type"] == "unarchived"]
+    assert len(unarchived) == 1
+    assert unarchived[0]["actor_id"] == person.actor_id
+    assert "continue please" in backend.clients[-1].turns[-1]
+
+
+@pytest.mark.asyncio
+async def test_the_chat_api_unarchives_an_archived_session_it_names(tmp_path, db):
+    from nerve.gateway.routes import _deps as deps_mod
+    from nerve.gateway.routes import sessions as routes
+
+    engine, _backend = _engine(tmp_path, db)
+    sid = "s-api"
+    await _new_session(engine, db, sid)
+    await engine.sessions.archive_session(sid)
+
+    saved = deps_mod._deps
+    deps_mod.init_deps(engine, db)
+    try:
+        req = routes.MessageRequest(session_id=sid, message="from the API")
+        response = await routes.chat(req, actor=db.system_actor)
+    finally:
+        deps_mod._deps = saved
+
+    assert response.session_id == sid
+    assert (await db.get_session(sid))["status"] == "active"
+    assert "unarchived" in await _events(db, sid)
+
+
+# --------------------------------------------------------------------------- #
+# Delete
+# --------------------------------------------------------------------------- #
+
+async def _delete(engine, db, sid):
+    from nerve.gateway.routes import _deps as deps_mod
+    from nerve.gateway.routes import sessions as routes
+
+    saved = deps_mod._deps
+    deps_mod.init_deps(engine, db)
+    try:
+        return await routes.delete_session(sid, actor=db.system_actor)
+    finally:
+        deps_mod._deps = saved
+
+
+@pytest.mark.asyncio
+async def test_delete_is_refused_while_a_turn_runs_and_one_waits(tmp_path, db):
+    engine, backend = _engine(tmp_path, db)
+    sid = "s-busy"
+    await engine.sessions.get_or_create(sid, source="web", backend="claude", actor=None)
+    backend.hold_new_clients = True
+    turn1 = asyncio.create_task(
+        engine.run(sid, "long task", source="web", channel="web", actor=None),
+    )
+    await _until(lambda: bool(backend.clients) and backend.clients[0].got_content.is_set())
+    turn2 = asyncio.create_task(
+        engine.run(sid, "queued follow-up", source="web", channel="web", actor=None),
+    )
+    await _until(lambda: len(engine.queue_view(sid)) == 1)
+
+    with pytest.raises(HTTPException) as info:
+        await _delete(engine, db, sid)
+
+    assert info.value.status_code == 409
+    assert await db.get_session(sid) is not None
+    assert await db.get_session_deletion(sid) is None
+    backend.clients[0].release.set()
+    await asyncio.wait_for(turn1, timeout=5)
+    await asyncio.wait_for(turn2, timeout=5)
+    users = [m["content"] for m in await db.get_messages(sid) if m["role"] == "user"]
+    assert users == ["long task", "queued follow-up"]
+
+
+@pytest.mark.asyncio
+async def test_a_run_never_creates_a_deleted_session_again(tmp_path, db):
+    engine, backend = _engine(tmp_path, db)
+    sid = "s-gone"
+    await _new_session(engine, db, sid)
+    assert await _delete(engine, db, sid) == {"deleted": True}
+
+    with pytest.raises(SessionClosedError) as info:
+        await engine.run(sid, "[Answer to: Proceed?]\n\nyes", source="notification:web",
+                         actor=None)
+
+    assert info.value.reason == "deleted"
+    assert await db.get_session(sid) is None
+    assert len(backend.clients) == 1
+
+
+@pytest.mark.asyncio
+async def test_a_run_during_a_delete_is_refused(tmp_path, db):
+    engine, backend = _engine(tmp_path, db)
+    sid = "s-deleting"
+    await engine.sessions.get_or_create(sid, source="web", backend="claude", actor=None)
+
+    assert engine.begin_delete(sid) is True
+    try:
+        with pytest.raises(SessionClosedError) as info:
+            await engine.run(sid, "a message", source="web", channel="web", actor=None)
+    finally:
+        engine.end_delete(sid)
+
+    assert info.value.reason == "deleted"
+    assert backend.clients == []
+    # After the delete ends, an existing session runs again.
+    await engine.run(sid, "a message", source="web", channel="web", actor=None)
+    assert len(backend.clients) == 1
+
+
+# --------------------------------------------------------------------------- #
+# Questions
+# --------------------------------------------------------------------------- #
+
+@pytest.mark.asyncio
+async def test_archive_expires_questions_with_a_report_and_keeps_approvals(
+    tmp_path, db, broadcasts,
+):
+    engine, _backend = _engine(tmp_path, db)
+    service = _notifications(engine, db, tmp_path)
+    sid = "s-asks"
+    await engine.sessions.get_or_create(sid, source="web", actor=None)
+    question = await service.ask_question(sid, "Proceed?", options=["yes", "no"])
+    # An approval acts on its target, not on the session: it stays open.
+    await db.create_notification(
+        notification_id="n-approval", session_id=sid, type="approval",
+        title="Merge?", target_kind="mechanical-action", target_id="a-1",
+    )
+    runs = AsyncMock()
+    engine.run = runs  # an expiry note would be a run
+
+    await engine.sessions.archive_session(sid)
+
+    row = await db.get_notification(question["notification_id"])
+    assert row["status"] == "expired"
+    assert any(
+        msg.get("type") == "notification_expired"
+        and msg.get("notification_id") == question["notification_id"]
+        for _c, msg in broadcasts
+    )
+    runs.assert_not_called()
+    assert (await db.get_notification("n-approval"))["status"] == "pending"
+
+
+@pytest.mark.asyncio
+async def test_delete_expires_questions(tmp_path, db, broadcasts):
+    engine, _backend = _engine(tmp_path, db)
+    service = _notifications(engine, db, tmp_path)
+    sid = "s-asks-deleted"
+    await engine.sessions.get_or_create(sid, source="web", actor=None)
+    question = await service.ask_question(sid, "Proceed?", options=["yes", "no"])
+
+    await _delete(engine, db, sid)
+
+    assert (await db.get_notification(question["notification_id"]))["status"] == "expired"
+    assert any(msg.get("type") == "notification_expired" for _c, msg in broadcasts)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("close", ["delete", "archive"])
+async def test_an_answer_to_a_closed_session_is_stored_and_starts_no_turn(
+    tmp_path, db, close,
+):
+    engine, _backend = _engine(tmp_path, db)
+    service = _notifications(engine, db, tmp_path)
+    sid = "s-answered"
+    await engine.sessions.get_or_create(sid, source="web", actor=None)
+    question = await service.ask_question(sid, "Delete branch?", options=["yes", "no"])
+    nid = question["notification_id"]
+    # Close the session without the archive or delete paths, which would
+    # expire the question.
+    if close == "delete":
+        await db.delete_session(sid)
+    else:
+        await db.change_session_status(sid, "archived", event_type="archived")
+    runs = AsyncMock()
+    engine.run = runs
+
+    assert await service.handle_answer(nid, "yes", answered_by="web") is True
+    await asyncio.sleep(0)
+
+    runs.assert_not_called()
+    assert (await db.get_notification(nid))["status"] == "answered"
+    if close == "delete":
+        assert await db.get_session(sid) is None
+    else:
+        assert (await db.get_session(sid))["status"] == "archived"
+
+
+@pytest.mark.asyncio
+async def test_idle_cleanup_expires_questions_only_after_the_archive(tmp_path, db):
+    engine, _backend = _engine(tmp_path, db)
+    service = _notifications(engine, db, tmp_path)
+    old = (datetime.now(timezone.utc) - timedelta(hours=5)).isoformat()
+    busy, idle = "s-busy", "s-idle"
+    questions = {}
+    for sid in (busy, idle):
+        await engine.sessions.get_or_create(sid, source="web", actor=None)
+        await db.update_session_fields(sid, {"status": "idle", "last_activity_at": old})
+        await db.db.execute("UPDATE sessions SET updated_at = ? WHERE id = ?", (old, sid))
+        await db.db.commit()
+        questions[sid] = (await service.ask_question(sid, "Proceed?", options=["yes"]))["notification_id"]
+    # A turn of the busy session holds its run lock, so its archive fails.
+    pending_turn = asyncio.create_task(asyncio.Event().wait())
+    engine.sessions.set_turn_task(busy, pending_turn)
+
+    try:
+        stats = await engine.sessions.run_cleanup(interactive_archive_after_hours=1)
+    finally:
+        pending_turn.cancel()
+
+    assert stats["archived_interactive"] == 1
+    assert (await db.get_session(busy))["status"] == "idle"
+    assert (await db.get_notification(questions[busy]))["status"] == "pending"
+    assert (await db.get_session(idle))["status"] == "archived"
+    assert (await db.get_notification(questions[idle]))["status"] == "expired"
diff --git a/tests/test_drain.py b/tests/test_drain.py
index 536deb6f8..f0477e234 100644
--- a/tests/test_drain.py
+++ b/tests/test_drain.py
@@ -45,7 +45,10 @@ def _make_engine() -> AgentEngine:
     config = MagicMock()
     config.agent.max_concurrent = 3
     config.mcp_servers = []
-    engine = AgentEngine(config, AsyncMock())
+    db = AsyncMock()
+    # A run reads the session row: an open session.
+    db.get_session.return_value = {"status": "idle"}
+    engine = AgentEngine(config, db)
     engine._broadcast_session_running = AsyncMock()
     engine._DRAIN_POLL_SECONDS = 0.01
     return engine
diff --git a/tests/test_external_auth.py b/tests/test_external_auth.py
index fa4b491da..30bc37aee 100644
--- a/tests/test_external_auth.py
+++ b/tests/test_external_auth.py
@@ -814,6 +814,9 @@ def is_session_running(self, *_args, **_kwargs):
     async def run(self, **kwargs):
         self.runs.append(kwargs)
 
+    async def reopen_archived(self, *_args, **_kwargs):
+        return False
+
     def register_task(self, *_args, **_kwargs):
         return None
 
diff --git a/tests/test_plan_actions.py b/tests/test_plan_actions.py
index 2ea6c999e..a1a9dbb00 100644
--- a/tests/test_plan_actions.py
+++ b/tests/test_plan_actions.py
@@ -94,6 +94,9 @@ async def run(
     def register_task(self, session_id, task) -> None:
         self.registered.append(session_id)
 
+    async def reopen_archived(self, session_id, *, attribution) -> bool:
+        return False
+
 
 async def _setup(
     db: Database, tmp_path, *, plan_type: str = "generic", status: str = "pending",
diff --git a/tests/test_plan_revise.py b/tests/test_plan_revise.py
index d81035e33..d4715c45b 100644
--- a/tests/test_plan_revise.py
+++ b/tests/test_plan_revise.py
@@ -82,6 +82,7 @@ def __init__(self, db: Database | None = None) -> None:
         # Event flipped after every run() call so tests can await
         # dispatch deterministically instead of sleeping.
         self.run_event = asyncio.Event()
+        self.reopened: list[str] = []
 
     async def run(
         self,
@@ -101,6 +102,10 @@ async def run(
         })
         self.run_event.set()
 
+    async def reopen_archived(self, session_id: str, *, attribution) -> bool:
+        self.reopened.append(session_id)
+        return False
+
 
 async def _setup(db: Database, tmp_path) -> tuple[FakeEngine, str]:
     """Create a task on disk + DB and a pending plan, then wire tools."""
@@ -195,6 +200,8 @@ async def test_routes_to_proposer_session(self, db: Database, tmp_path):
         # must route the feedback there, not to the cron fallback.
         assert engine.runs[0]["session_id"] == "sess-proposer"
         assert engine.sessions.calls[0]["session_id"] == "sess-proposer"
+        # An archived proposer session is unarchived before the run.
+        assert engine.reopened == ["sess-proposer"]
 
     async def test_falls_back_to_cron_planner_when_session_unknown(
         self, db: Database, tmp_path,
diff --git a/tests/test_session_attribution.py b/tests/test_session_attribution.py
index 7caea551a..5c01e3f53 100644
--- a/tests/test_session_attribution.py
+++ b/tests/test_session_attribution.py
@@ -349,7 +349,10 @@ async def test_concurrent_archives_and_restores_record_one_change(install, monke
     alice, bob = install.actor_id("alice"), install.actor_id("bob")
     async with _client(install.app()) as client:
         session_id = await _new_session(client, install.headers("alice"))
-        _hold_until_all_arrive(monkeypatch, install.db, "change_session_status")
+        # Two archives of one session take turns on its client lock, so they
+        # wait for each other before the lock: both have read the session,
+        # and the second one finds it archived.
+        _hold_until_all_arrive(monkeypatch, install.db, "list_child_sessions")
 
         async def both(action: str) -> None:
             responses = await asyncio.gather(*(
@@ -361,6 +364,7 @@ async def both(action: str) -> None:
             assert [r.status_code for r in responses] == [200, 200]
 
         await both("archive")
+        _hold_until_all_arrive(monkeypatch, install.db, "change_session_status")
         await both("unarchive")
 
     events = await install.events(session_id, "archived", "unarchived")
diff --git a/web/src/stores/chatStore.test.ts b/web/src/stores/chatStore.test.ts
index 61e81413e..52ad1454a 100644
--- a/web/src/stores/chatStore.test.ts
+++ b/web/src/stores/chatStore.test.ts
@@ -34,6 +34,7 @@ vi.mock('../api/client', () => ({
     getSessionAgents: vi.fn(async (sessionId: string) => ({ session_id: sessionId, agents: [], queue: [] })),
     getAgentTranscript: vi.fn(),
     archiveSession: vi.fn(),
+    deleteSession: vi.fn(),
     stopSession: vi.fn(),
     createSession: vi.fn(),
     getSessionTasks: vi.fn(async () => ({ tasks: [] })),
@@ -745,6 +746,37 @@ describe('archiving a session with running agents', () => {
   });
 });
 
+describe('deleting a running session', () => {
+  beforeEach(() => {
+    onSession();
+    localStorage.setItem(`nerve_draft_${SID}`, 'unsent');
+    useChatStore.setState({ drafts: { [SID]: 'unsent' } });
+  });
+
+  it('says why the server refused, and keeps the draft', async () => {
+    const alert = vi.spyOn(window, 'alert').mockImplementation(() => {});
+    vi.mocked(api.deleteSession).mockRejectedValueOnce(
+      new Error('409: {"detail":"The session is running or has queued messages. Stop it, then delete it."}'),
+    );
+
+    await useChatStore.getState().deleteSession(SID);
+
+    expect(alert).toHaveBeenCalledWith('Could not delete this chat: it is running or has queued messages. Stop it, then delete it.');
+    expect(useChatStore.getState().drafts[SID]).toBe('unsent');
+    alert.mockRestore();
+  });
+
+  it('shows nothing for another error', async () => {
+    const alert = vi.spyOn(window, 'alert').mockImplementation(() => {});
+    vi.mocked(api.deleteSession).mockRejectedValueOnce(new Error('500: boom'));
+
+    await useChatStore.getState().deleteSession(SID);
+
+    expect(alert).not.toHaveBeenCalled();
+    alert.mockRestore();
+  });
+});
+
 describe('stopping the turn', () => {
   it('leaves streaming mode when the server has no turn running', async () => {
     onSession({ isStreaming: true, streamingBlocks: [{ type: 'text', content: 'partial' }] });
diff --git a/web/src/stores/chatStore.ts b/web/src/stores/chatStore.ts
index e081a1b6b..84143c110 100644
--- a/web/src/stores/chatStore.ts
+++ b/web/src/stores/chatStore.ts
@@ -1119,6 +1119,10 @@ export const useChatStore = create<ChatState>((set, get) => ({
       }
     } catch (e) {
       console.error('Failed to delete session:', e);
+      // 409: a turn of the session runs or waits in its queue.
+      if (e instanceof Error && e.message.startsWith('409:')) {
+        window.alert('Could not delete this chat: it is running or has queued messages. Stop it, then delete it.');
+      }
     }
   },
 

From ecf2aa6625d9f5a8cd09b7e5ddc2afe816200d72 Mon Sep 17 00:00:00 2001
From: Alex Soffronow Pagonidis
 <237136924+alex-clickhouse@users.noreply.github.com>
Date: Tue, 10 Nov 2026 06:46:42 +0000
Subject: [PATCH 2/5] Sessions: close the review findings on archived and
 deleted sessions

- Archive: refused while anything holds the session's run lock (a
  queued message, a background continuation, an idle drain) or a
  registered task runs. After the checks, with no await, the session is
  marked as archiving; a run checks the mark right after it takes the
  run lock, so it is refused before it stores anything. This replaces
  the hold of the client lock.
- Delete: also 409 while the run lock is held (the queue handoff).
- POST /api/chat with the id of a deleted session creates nothing and
  answers 409. A plan revision for a deleted proposer session goes to
  cron:task-planner.
- Run later into an archived session unarchives it first.
- Unstarring an archived session starts no star_project_hook turn.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 docs/api.md                       |  15 ++-
 docs/plans.md                     |   2 +-
 nerve/agent/engine.py             |  37 ++++---
 nerve/agent/plan_service.py       |   8 ++
 nerve/agent/sessions.py           | 131 ++++++++++++++----------
 nerve/gateway/routes/sessions.py  |  21 +++-
 tests/test_closed_sessions.py     | 161 +++++++++++++++++++++++++++---
 tests/test_plan_revise.py         |  17 ++++
 tests/test_run_later.py           |   2 +
 tests/test_session_attribution.py |   6 +-
 10 files changed, 308 insertions(+), 92 deletions(-)

diff --git a/docs/api.md b/docs/api.md
index b0d6f2e24..70862d3a8 100644
--- a/docs/api.md
+++ b/docs/api.md
@@ -1116,8 +1116,9 @@ 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, or a message waits in its turn queue, the
-delete answers `409`: stop the session first. The delete expires the
+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.
 
@@ -1198,8 +1199,10 @@ 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`, a plan revision request for a plan
-that it proposed). A route that would run a turn in an archived session
+`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
@@ -1322,7 +1325,9 @@ Response: { "response": "Hi there!", "session_id": "main" }
 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`). A `session_id` of an
-archived session unarchives it first, in the caller's name.
+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.
 
diff --git a/docs/plans.md b/docs/plans.md
index a5f67b44c..b2662da67 100644
--- a/docs/plans.md
+++ b/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?)`
 
diff --git a/nerve/agent/engine.py b/nerve/agent/engine.py
index b39f0816e..faff6c11f 100644
--- a/nerve/agent/engine.py
+++ b/nerve/agent/engine.py
@@ -491,8 +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.
+        # 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
@@ -1189,12 +1191,19 @@ async def reopen_archived(
             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()
+
     def has_turn_work(self, session_id: str) -> bool:
-        """A turn of the session runs, waits for a concurrency slot, or
-        waits in the session's turn queue."""
+        """A turn of the session runs or waits for a concurrency slot, or
+        anything holds its run lock (see :meth:`_run_lock_held`)."""
         return (
             self.sessions.has_running_work(session_id)
-            or bool(self.queue_view(session_id))
+            or self._run_lock_held(session_id)
         )
 
     def _deleting_sessions(self) -> set[str]:
@@ -1207,7 +1216,7 @@ def begin_delete(self, session_id: str) -> bool:
         (:meth:`has_turn_work`), else mark it and return ``True``. Call
         :meth:`end_delete` when the rows are gone.
 
-        No await here. A run sets its turn task and then checks the mark,
+        No await here. A run takes the run lock and then checks the mark,
         with no await in between either. So a run either sees the mark and
         is refused, or makes this call return ``False``.
         """
@@ -1577,9 +1586,8 @@ async def _get_or_create_client(
 
             # Session row first — backend resolution is sticky on it.
             session = await self.db.get_session(session_id)
-            # Under the client lock, which archive_session also holds: a
-            # session archived (or deleted) after the checks in _run_turn
-            # gets no client, and its runtime gets no turn.
+            # Check again before the runtime gets anything: the session row
+            # read here is the one the client is built for.
             closed = await self._session_closed_reason(session_id, session)
             if closed:
                 raise SessionClosedError(closed)
@@ -2154,10 +2162,13 @@ async def _session_closed_reason(
     ) -> str | None:
         """``archived`` or ``deleted`` when no turn may run in the session,
         else ``None``. ``session`` is the row when the caller has read it
-        already (``None`` for no row). A delete in progress counts as
-        ``deleted``; that check comes before the first await."""
+        already (``None`` for no row). A delete or an archive in progress
+        counts as ``deleted`` or ``archived``; those checks come before the
+        first await."""
         if session_id in self._deleting_sessions():
             return "deleted"
+        if self.sessions.is_archiving(session_id):
+            return "archived"
         if session is False:
             session = await self.db.get_session(session_id)
         if session is not None:
@@ -4107,11 +4118,11 @@ async def _run_turn(
                     session_id, entry=entry, client_msg_id=client_msg_id,
                 )
             # The Stop target from the moment this run holds the run lock:
-            # Stop reaches the turn, never a message queued behind it. From
-            # here on, an archive of the session is refused.
+            # 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.
+            # the lock, or while it waits in the queue. No await since the
+            # lock: an archive or a delete that starts later sees this run.
             closed = await self._session_closed_reason(session_id)
             if closed:
                 await self._refuse_closed_session(
diff --git a/nerve/agent/plan_service.py b/nerve/agent/plan_service.py
index fe992d5ee..edc999eef 100644
--- a/nerve/agent/plan_service.py
+++ b/nerve/agent/plan_service.py
@@ -240,6 +240,14 @@ 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.get_session(session_id) is None
+        and await db.get_session_deletion(session_id) is not None
+    ):
+        session_id = _FALLBACK_PLANNER_SESSION
     session_title = (
         f"Cron: {session_id.split(':')[-1]}"
         if session_id.startswith("cron:")
diff --git a/nerve/agent/sessions.py b/nerve/agent/sessions.py
index ff0b646ad..69e311964 100644
--- a/nerve/agent/sessions.py
+++ b/nerve/agent/sessions.py
@@ -116,6 +116,9 @@ class SessionManager:
     # status (the engine wires the expiry of the session's questions).
     # Never raises into the caller.
     on_archived: Callable[[str], Awaitable[None]] | None = None
+    # ``run_lock_held(session_id)``: something holds the session's run lock
+    # (the archive guard). Wired to the engine's run locks.
+    run_lock_held: Callable[[str], bool] | None = None
     # Called as ``listener(session_id, running)`` right after a session
     # starts or stops running a turn (see ``add_running_listener``).
     _running_listeners: tuple[Callable[[str, bool], None], ...] = ()
@@ -1021,12 +1024,14 @@ async def archive_session(
         agents is refused with :class:`SessionBusyError`; with ``force`` its
         agents end as interrupted (``archived``).
 
-        A session with a turn that holds its run lock (it runs, or waits for
-        a concurrency slot) is refused with :class:`SessionBusyError`, also
-        with ``force``. The check and the status write hold the session's
-        client lock, the lock under which a turn builds its client and checks
-        that the session is not archived. So the runtime never gets a turn of
-        a session that this call archives.
+        A session with turn work is refused with :class:`SessionBusyError`,
+        also with ``force``: a turn that runs or waits for a concurrency
+        slot, or any holder of the session's run lock (a queued message, an
+        autonomous turn, an idle drain). Right after these checks, with no
+        await in between, the session is marked as archiving until its status
+        is written (:meth:`is_archiving`). A run checks the mark right after
+        it holds the run lock, also with no await in between. So a run either
+        makes this call fail or is refused.
 
         After the archive, the session's pending wakeups are deleted and
         ``on_archived`` runs (it expires the session's pending questions).
@@ -1036,58 +1041,80 @@ async def archive_session(
             raise ValueError(f"Session {session_id} not found")
         if session.get("status") == SessionStatus.ARCHIVED.value:
             return False
-        lock = self.get_lock(session_id)
-        async with lock:
-            # Read again under the lock: another archive that held it can
-            # have archived the session meanwhile.
-            session = await self.db.get_session(session_id)
-            if not session or session.get("status") == SessionStatus.ARCHIVED.value:
-                return False
-            if self.has_running_work(session_id):
-                raise SessionBusyError(BUSY_RUNNING_REASON)
-            if not force and self._agents_live(session_id):
-                raise SessionBusyError(BUSY_AGENTS_REASON)
-            # Memorize in the background: memU indexing can take ~90s; awaiting it
-            # here would hang the archive (and, in a cascade, the whole request on the root).
-            if self._on_memorize:
-                async def _memorize_bg(sid: str = session_id) -> None:
-                    try:
-                        await self._on_memorize(sid)
-                    except Exception as e:
-                        logger.warning("Background memorize for %s failed: %s", sid, e)
-                asyncio.create_task(_memorize_bg())
-
-            client = self.remove_client(session_id, reason="archived")
-            # remove_client drops the session's client lock from the map. Keep
-            # the held lock there until the status is written, so that a turn
-            # that builds a client meanwhile waits for it.
-            self._client_locks[session_id] = lock
-            if client is None and self._agents_live(session_id):
-                # Records still live without a runtime: nothing runs them any
-                # more, so settle them with the archive.
-                self._notify_client_removed(session_id, "archived")
-            if client:
-                try:
-                    await client.disconnect()
-                except Exception:
-                    pass
-            now = datetime.now(timezone.utc).isoformat()
-            archived = await self.db.change_session_status(
-                session_id, SessionStatus.ARCHIVED.value,
-                fields={
-                    "archived_at": now,
-                    "connected_at": None,
-                    "sdk_session_id": None,
-                },
-                event_type="archived", details=details, attribution=attribution,
+        if self.has_running_work(session_id) or self._run_lock_held(session_id):
+            raise SessionBusyError(BUSY_RUNNING_REASON)
+        if not force and self._agents_live(session_id):
+            raise SessionBusyError(BUSY_AGENTS_REASON)
+        marks = self._archiving_marks()
+        marks[session_id] = marks.get(session_id, 0) + 1
+        try:
+            archived = await self._archive_marked(
+                session_id, attribution=attribution, details=details,
             )
-        if self._client_locks.get(session_id) is lock:
-            self._client_locks.pop(session_id)
+        finally:
+            left = marks.pop(session_id, 1) - 1
+            if left:
+                marks[session_id] = left
         if archived:
             logger.info("Archived session %s", session_id)
             await self._close_archived_work(session_id)
         return archived
 
+    async def _archive_marked(
+        self, session_id: str, *, attribution: Attribution, details: dict | None,
+    ) -> bool:
+        """The work of :meth:`archive_session` while the session is marked:
+        memorize, disconnect the client, write the status."""
+        # Memorize in the background: memU indexing can take ~90s; awaiting it
+        # here would hang the archive (and, in a cascade, the whole request on the root).
+        if self._on_memorize:
+            async def _memorize_bg(sid: str = session_id) -> None:
+                try:
+                    await self._on_memorize(sid)
+                except Exception as e:
+                    logger.warning("Background memorize for %s failed: %s", sid, e)
+            asyncio.create_task(_memorize_bg())
+
+        client = self.remove_client(session_id, reason="archived")
+        if client is None and self._agents_live(session_id):
+            # Records still live without a runtime: nothing runs them any
+            # more, so settle them with the archive.
+            self._notify_client_removed(session_id, "archived")
+        if client:
+            try:
+                await client.disconnect()
+            except Exception:
+                pass
+        now = datetime.now(timezone.utc).isoformat()
+        return await self.db.change_session_status(
+            session_id, SessionStatus.ARCHIVED.value,
+            fields={
+                "archived_at": now,
+                "connected_at": None,
+                "sdk_session_id": None,
+            },
+            event_type="archived", details=details, attribution=attribution,
+        )
+
+    def _archiving_marks(self) -> dict[str, int]:
+        """Archives in progress per session (created on first use, so
+        managers built with ``__new__`` work too)."""
+        return self.__dict__.setdefault("_archiving", {})
+
+    def is_archiving(self, session_id: str) -> bool:
+        """An archive of the session is in progress (see :meth:`archive_session`)."""
+        return session_id in self._archiving_marks()
+
+    def _run_lock_held(self, session_id: str) -> bool:
+        """``run_lock_held(session_id)``; ``False`` when it is not wired."""
+        if self.run_lock_held is None:
+            return False
+        try:
+            return bool(self.run_lock_held(session_id))
+        except Exception:
+            logger.exception("run-lock check failed for %s", session_id)
+            return False
+
     async def _close_archived_work(self, session_id: str) -> None:
         """End the work that targets a session that was just archived: its
         pending wakeups (deleted) and, through ``on_archived``, its pending
diff --git a/nerve/gateway/routes/sessions.py b/nerve/gateway/routes/sessions.py
index 687f12929..b506758a4 100644
--- a/nerve/gateway/routes/sessions.py
+++ b/nerve/gateway/routes/sessions.py
@@ -650,9 +650,11 @@ async def update_session(session_id: str, req: dict, actor: Actor = Depends(requ
     await deps.db.change_session_fields(session_id, fields, attribution=attribution)
     updated = await deps.db.get_session(session_id)
     # Star = opt-in project registration (sessions.star_project_hook, default
-    # off). On a starred transition fire a one-shot internal turn.
+    # off). On a starred transition fire a one-shot internal turn. An
+    # archived session runs no turn (unstarring keeps it archived).
     if (get_config().sessions.star_project_hook and "starred" in fields
-            and deps.engine and fields["starred"] != old_starred):
+            and deps.engine and fields["starred"] != old_starred
+            and (updated or {}).get("status") != "archived"):
         promoted = fields["starred"] == 1
         trigger = (
             "[system:star-promote] This session was just STARRED, which registers it as a "
@@ -849,6 +851,12 @@ async def run_later(req: RunLaterRequest, actor: Actor = Depends(require_auth)):
     message = (req.message or "").strip()
     if not message and not req.file_ids:
         raise HTTPException(status_code=400, detail="Nothing to defer")
+    if deps.engine:
+        # The person's own message into an archived session opens it again,
+        # so that the deferred turn can run.
+        await deps.engine.reopen_archived(
+            req.session_id, attribution=attribution_for(actor),
+        )
 
     delay = (req.delay or "none").strip().lower()
     if delay != "none" and delay not in _RUN_LATER_TIMED:
@@ -1139,9 +1147,14 @@ async def chat(req: MessageRequest, actor: Actor = Depends(require_auth)):
         session_id = await deps.engine.sessions.get_active_session(
             "api:default", source="api", actor=actor,
         )
-    elif deps.engine.accepting_turns:
+    elif deps.engine.accepting_turns and not (
+        await deps.db.get_session(session_id) is None
+        and await deps.db.get_session_deletion(session_id) is not None
+    ):
         # A new session id from the caller: a session of the API. A draining
-        # engine creates nothing: run() refuses the turn with a 503.
+        # engine creates nothing: run() refuses the turn with a 503. The id of
+        # a deleted session creates nothing either: run() refuses it with a
+        # 409.
         await deps.engine.sessions.get_or_create(
             session_id, source="web", actor=actor, origin=Origin(TRIGGER_API),
         )
diff --git a/tests/test_closed_sessions.py b/tests/test_closed_sessions.py
index b0da0a4c2..9db09d40f 100644
--- a/tests/test_closed_sessions.py
+++ b/tests/test_closed_sessions.py
@@ -297,15 +297,54 @@ async def test_archive_is_refused_while_a_turn_waits_for_a_concurrency_slot(tmp_
     assert backend.clients[0].turns
 
 
+@pytest.mark.asyncio
+async def test_archive_is_refused_while_a_registered_task_runs(tmp_path, db):
+    """A workflow leg between two of its turns: its task is registered for
+    the session, and nothing holds the run lock."""
+    engine, _backend = _engine(tmp_path, db)
+    sid = "s-leg"
+    await _new_session(engine, db, sid)
+    leg = asyncio.create_task(asyncio.Event().wait())
+    engine.register_task(sid, leg)
+    try:
+        assert not engine._session_locks[sid].locked()
+        with pytest.raises(SessionBusyError) as info:
+            await engine.sessions.archive_session(sid)
+        assert str(info.value) == BUSY_RUNNING_REASON
+    finally:
+        leg.cancel()
+    await asyncio.sleep(0)
+    assert await engine.sessions.archive_session(sid) is True
+
+
+@pytest.mark.asyncio
+async def test_archive_is_refused_while_anything_holds_the_run_lock(tmp_path, db):
+    """An autonomous turn, an idle drain or a queued message holds the run
+    lock with no turn task: the archive is refused all the same."""
+    engine, _backend = _engine(tmp_path, db)
+    sid = "s-run-lock"
+    await _new_session(engine, db, sid)
+
+    async with engine._holding_run_lock(sid, engine._session_locks[sid]):
+        assert engine.sessions.turn_task(sid) is None
+        with pytest.raises(SessionBusyError) as info:
+            await engine.sessions.archive_session(sid)
+        assert str(info.value) == BUSY_RUNNING_REASON
+        result = await engine.sessions.archive_session_cascade(sid)
+        assert result["skipped"] == [{"id": sid, "reason": BUSY_RUNNING_REASON}]
+
+    assert await engine.sessions.archive_session(sid) is True
+
+
 @pytest.mark.asyncio
 async def test_a_queued_turn_is_refused_when_the_session_was_archived_meanwhile(
     tmp_path, db, broadcasts,
 ):
+    """The check after the run lock: a session that another path archives
+    (as the Codex sync does for its mirror sessions) while a run waits."""
     engine, backend = _engine(tmp_path, db)
     sid = "s-queued"
     await _new_session(engine, db, sid)
-    # The run lock is held by work that is not a turn (an idle drain or a
-    # client reap), so the next message waits in the queue.
     lock = engine._session_locks[sid]
     await lock.acquire()
     turn = asyncio.create_task(
@@ -314,7 +353,7 @@ async def test_a_queued_turn_is_refused_when_the_session_was_archived_meanwhile(
     )
     await _until(lambda: len(engine.queue_view(sid)) == 1)
 
-    assert (await engine.sessions.archive_session_cascade(sid))["archived"] == [sid]
+    await db.change_session_status(sid, "archived", event_type="archived")
     engine._pass_run_lock(sid, lock)
 
     with pytest.raises(SessionClosedError):
@@ -361,9 +400,11 @@ async def test_a_session_archived_while_its_turn_waits_for_the_client_gets_no_cl
 
 
 @pytest.mark.asyncio
-async def test_a_turn_that_starts_during_an_archive_gets_no_client(tmp_path, db):
-    """The archive holds the client lock: a turn that passed its checks while
-    the archive disconnected the old client builds no new one."""
+async def test_a_turn_that_starts_during_an_archive_is_refused_before_it_stores_anything(
+    tmp_path, db,
+):
+    """The archive mark: a turn that takes the run lock while the archive
+    disconnects the old client is refused before its message is stored."""
     engine, backend = _engine(tmp_path, db)
     sid = "s-mid-archive"
     await _new_session(engine, db, sid)
@@ -371,16 +412,21 @@ async def test_a_turn_that_starts_during_an_archive_gets_no_client(tmp_path, db)
     backend.clients[0].disconnect_gate = gate
     archive = asyncio.create_task(engine.sessions.archive_session(sid))
     await _until(lambda: engine.sessions.get_client(sid) is None)
-    turn = asyncio.create_task(
-        engine.run(sid, "sent during the archive", source="web", channel="web", actor=None),
-    )
-    await _until(lambda: engine.sessions.is_running(sid))
+    assert engine.sessions.is_archiving(sid)
+
+    with pytest.raises(SessionClosedError):
+        await asyncio.wait_for(
+            engine.run(sid, "sent during the archive", source="web",
+                       channel="web", actor=None),
+            timeout=5,
+        )
     gate.set()
 
     assert await asyncio.wait_for(archive, timeout=5) is True
-    with pytest.raises(SessionClosedError):
-        await asyncio.wait_for(turn, timeout=5)
+    assert not engine.sessions.is_archiving(sid)
     assert len(backend.clients) == 1
+    users = [m["content"] for m in await db.get_messages(sid) if m["role"] == "user"]
+    assert "sent during the archive" not in users
     assert (await db.get_session(sid))["status"] == "archived"
 
 
@@ -482,6 +528,58 @@ async def test_the_chat_api_unarchives_an_archived_session_it_names(tmp_path, db
     assert "unarchived" in await _events(db, sid)
 
 
+@pytest.mark.asyncio
+async def test_run_later_into_an_archived_session_unarchives_it(tmp_path, db):
+    from nerve.gateway.routes import _deps as deps_mod
+    from nerve.gateway.routes import sessions as routes
+
+    engine, _backend = _engine(tmp_path, db)
+    sid = "s-later"
+    await _new_session(engine, db, sid)
+    await engine.sessions.archive_session(sid)
+
+    saved = deps_mod._deps
+    deps_mod.init_deps(engine, db)
+    try:
+        req = routes.RunLaterRequest(session_id=sid, message="do it later", delay="30m")
+        result = await routes.run_later(req, actor=db.system_actor)
+    finally:
+        deps_mod._deps = saved
+
+    assert result["scheduled"] is True
+    assert (await db.get_session(sid))["status"] == "idle"
+    assert len(await db.list_pending_wakeups(sid)) == 1
+    assert "unarchived" in await _events(db, sid)
+
+
+@pytest.mark.asyncio
+async def test_unstarring_an_archived_session_runs_no_hook(tmp_path, db, monkeypatch):
+    from nerve.gateway.routes import _deps as deps_mod
+    from nerve.gateway.routes import sessions as routes
+
+    engine, _backend = _engine(tmp_path, db)
+    engine.config.sessions.star_project_hook = True
+    monkeypatch.setattr(routes, "get_config", lambda: engine.config)
+    sid = "s-starred"
+    await engine.sessions.get_or_create(sid, source="web", actor=None)
+    await db.change_session_fields(sid, {"starred": 1})
+    await engine.sessions.archive_session(sid)
+    runs = AsyncMock()
+    engine.run = runs
+
+    saved = deps_mod._deps
+    deps_mod.init_deps(engine, db)
+    try:
+        await routes.update_session(sid, {"starred": False}, actor=db.system_actor)
+    finally:
+        deps_mod._deps = saved
+    await asyncio.sleep(0)
+
+    runs.assert_not_called()
+    row = await db.get_session(sid)
+    assert row["status"] == "archived" and not row["starred"]
+
+
 # --------------------------------------------------------------------------- #
 # Delete
 # --------------------------------------------------------------------------- #
@@ -526,6 +624,45 @@ async def test_delete_is_refused_while_a_turn_runs_and_one_waits(tmp_path, db):
     assert users == ["long task", "queued follow-up"]
 
 
+@pytest.mark.asyncio
+async def test_delete_is_refused_while_the_run_lock_is_held_without_a_turn(tmp_path, db):
+    """The queue handoff, an autonomous turn or an idle drain: the run lock
+    is held, and no turn task is set."""
+    engine, _backend = _engine(tmp_path, db)
+    sid = "s-handoff"
+    await _new_session(engine, db, sid)
+
+    async with engine._holding_run_lock(sid, engine._session_locks[sid]):
+        with pytest.raises(HTTPException) as info:
+            await _delete(engine, db, sid)
+    assert info.value.status_code == 409
+    assert await db.get_session(sid) is not None
+
+
+@pytest.mark.asyncio
+async def test_the_chat_api_does_not_create_a_deleted_session_again(tmp_path, db):
+    from nerve.gateway.routes import _deps as deps_mod
+    from nerve.gateway.routes import sessions as routes
+
+    engine, backend = _engine(tmp_path, db)
+    sid = "s-api-deleted"
+    await _new_session(engine, db, sid)
+    await _delete(engine, db, sid)
+
+    saved = deps_mod._deps
+    deps_mod.init_deps(engine, db)
+    try:
+        req = routes.MessageRequest(session_id=sid, message="still there?")
+        with pytest.raises(SessionClosedError) as info:
+            await routes.chat(req, actor=db.system_actor)
+    finally:
+        deps_mod._deps = saved
+
+    assert info.value.reason == "deleted"
+    assert await db.get_session(sid) is None
+    assert len(backend.clients) == 1
+
+
 @pytest.mark.asyncio
 async def test_a_run_never_creates_a_deleted_session_again(tmp_path, db):
     engine, backend = _engine(tmp_path, db)
diff --git a/tests/test_plan_revise.py b/tests/test_plan_revise.py
index d4715c45b..5cc0bc423 100644
--- a/tests/test_plan_revise.py
+++ b/tests/test_plan_revise.py
@@ -203,6 +203,23 @@ async def test_routes_to_proposer_session(self, db: Database, tmp_path):
         # An archived proposer session is unarchived before the run.
         assert engine.reopened == ["sess-proposer"]
 
+    async def test_falls_back_to_cron_planner_when_proposer_was_deleted(
+        self, db: Database, tmp_path,
+    ):
+        engine, _ = await _setup(db, tmp_path)
+        await db.create_session("sess-proposer", actor=None)
+        await db.delete_session("sess-proposer")
+
+        await request_plan_revision(
+            db=db, engine=engine,
+            plan_id="plan-orig", feedback="rev",
+            actor=None, cause=None,
+        )
+        await asyncio.wait_for(engine.run_event.wait(), timeout=1.0)
+
+        assert engine.runs[0]["session_id"] == "cron:task-planner"
+        assert await db.get_session("sess-proposer") is None
+
     async def test_falls_back_to_cron_planner_when_session_unknown(
         self, db: Database, tmp_path,
     ):
diff --git a/tests/test_run_later.py b/tests/test_run_later.py
index fa19e383f..7ad2e3af5 100644
--- a/tests/test_run_later.py
+++ b/tests/test_run_later.py
@@ -41,6 +41,8 @@ async def _install(db: Database, engine) -> None:
 def _fake_engine() -> MagicMock:
     engine = MagicMock()
     engine.run = AsyncMock()  # must never be awaited by run-later
+    # Run-later unarchives an archived session first; these sessions are open.
+    engine.reopen_archived = AsyncMock(return_value=False)
     return engine
 
 
diff --git a/tests/test_session_attribution.py b/tests/test_session_attribution.py
index 5c01e3f53..7caea551a 100644
--- a/tests/test_session_attribution.py
+++ b/tests/test_session_attribution.py
@@ -349,10 +349,7 @@ async def test_concurrent_archives_and_restores_record_one_change(install, monke
     alice, bob = install.actor_id("alice"), install.actor_id("bob")
     async with _client(install.app()) as client:
         session_id = await _new_session(client, install.headers("alice"))
-        # Two archives of one session take turns on its client lock, so they
-        # wait for each other before the lock: both have read the session,
-        # and the second one finds it archived.
-        _hold_until_all_arrive(monkeypatch, install.db, "list_child_sessions")
+        _hold_until_all_arrive(monkeypatch, install.db, "change_session_status")
 
         async def both(action: str) -> None:
             responses = await asyncio.gather(*(
@@ -364,7 +361,6 @@ async def both(action: str) -> None:
             assert [r.status_code for r in responses] == [200, 200]
 
         await both("archive")
-        _hold_until_all_arrive(monkeypatch, install.db, "change_session_status")
         await both("unarchive")
 
     events = await install.events(session_id, "archived", "unarchived")

From f826f82edffc34db656d362f0588e5c9d057006b Mon Sep 17 00:00:00 2001
From: Alex Soffronow Pagonidis
 <237136924+alex-clickhouse@users.noreply.github.com>
Date: Tue, 10 Nov 2026 07:32:23 +0000
Subject: [PATCH 3/5] Sessions: one closing mark for archive and delete

- SessionManager.closing(session_id, reason) marks a session while an
  archive or a delete is in progress; closing_reason() reports it. The
  engine's own delete set, begin_delete and end_delete are gone.
- SessionManager.has_turn_work() is the one turn-work check for archive
  and the delete route.
- Database.closed_session_reason() is the one "archived or deleted"
  lookup for the engine, POST /api/chat and the plan revision.
- The race explanation is in one docstring (SessionManager.closing).
- The delete test goes through the delete route, so it also covers the
  route's mark. A unit test covers nested marks and their cleanup.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 nerve/agent/engine.py            | 68 +++++-------------------------
 nerve/agent/plan_service.py      |  3 +-
 nerve/agent/sessions.py          | 72 +++++++++++++++++++-------------
 nerve/db/sessions.py             | 10 +++++
 nerve/gateway/routes/sessions.py | 12 +++---
 tests/test_closed_sessions.py    | 43 +++++++++++++------
 tests/test_drain.py              |  4 +-
 7 files changed, 101 insertions(+), 111 deletions(-)

diff --git a/nerve/agent/engine.py b/nerve/agent/engine.py
index faff6c11f..cbad19ddf 100644
--- a/nerve/agent/engine.py
+++ b/nerve/agent/engine.py
@@ -24,7 +24,7 @@
 from collections.abc import Iterator
 from dataclasses import dataclass, field
 from datetime import datetime, timedelta, timezone
-from typing import Any, Literal
+from typing import Any
 
 from nerve.agent.agents import AgentRegistry
 from nerve.agent.agents.errors import (
@@ -1198,38 +1198,6 @@ def _run_lock_held(self, session_id: str) -> bool:
         lock = self._session_locks.get(session_id)
         return lock is not None and lock.locked()
 
-    def has_turn_work(self, session_id: str) -> bool:
-        """A turn of the session runs or waits for a concurrency slot, or
-        anything holds its run lock (see :meth:`_run_lock_held`)."""
-        return (
-            self.sessions.has_running_work(session_id)
-            or self._run_lock_held(session_id)
-        )
-
-    def _deleting_sessions(self) -> set[str]:
-        """Sessions whose delete is in progress (created on first use, so
-        engines built with ``__new__`` work too)."""
-        return self.__dict__.setdefault("_deleting", set())
-
-    def begin_delete(self, session_id: str) -> bool:
-        """Start the delete of a session: ``False`` while it has turn work
-        (:meth:`has_turn_work`), else mark it and return ``True``. Call
-        :meth:`end_delete` when the rows are gone.
-
-        No await here. A run takes the run lock and then checks the mark,
-        with no await in between either. So a run either sees the mark and
-        is refused, or makes this call return ``False``.
-        """
-        if self.has_turn_work(session_id):
-            return False
-        self._deleting_sessions().add(session_id)
-        return True
-
-    def end_delete(self, session_id: str) -> None:
-        """End what :meth:`begin_delete` started. The ``session_deletions``
-        row refuses later runs."""
-        self._deleting_sessions().discard(session_id)
-
     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).
@@ -1586,9 +1554,8 @@ async def _get_or_create_client(
 
             # Session row first — backend resolution is sticky on it.
             session = await self.db.get_session(session_id)
-            # Check again before the runtime gets anything: the session row
-            # read here is the one the client is built for.
-            closed = await self._session_closed_reason(session_id, session)
+            # Check again before the runtime gets anything.
+            closed = await self._session_closed_reason(session_id)
             if closed:
                 raise SessionClosedError(closed)
             backend = self._backend_for(session, source)
@@ -2157,27 +2124,14 @@ async def drain(self, timeout: float) -> bool:
         logger.info("Drain: all running turns ended")
         return True
 
-    async def _session_closed_reason(
-        self, session_id: str, session: dict | None | Literal[False] = False,
-    ) -> str | None:
+    async def _session_closed_reason(self, session_id: str) -> str | None:
         """``archived`` or ``deleted`` when no turn may run in the session,
-        else ``None``. ``session`` is the row when the caller has read it
-        already (``None`` for no row). A delete or an archive in progress
-        counts as ``deleted`` or ``archived``; those checks come before the
-        first await."""
-        if session_id in self._deleting_sessions():
-            return "deleted"
-        if self.sessions.is_archiving(session_id):
-            return "archived"
-        if session is False:
-            session = await self.db.get_session(session_id)
-        if session is not None:
-            if session.get("status") == SessionStatus.ARCHIVED.value:
-                return "archived"
-            return None
-        if await self.db.get_session_deletion(session_id) is not None:
-            return "deleted"
-        return None
+        else ``None``. An archive or a delete in progress counts; that check
+        comes before the first await."""
+        return (
+            self.sessions.closing_reason(session_id)
+            or await self.db.closed_session_reason(session_id)
+        )
 
     async def _refuse_closed_session(
         self,
@@ -4122,7 +4076,7 @@ async def _run_turn(
             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: an archive or a delete that starts later sees this run.
+            # lock (see SessionManager.closing).
             closed = await self._session_closed_reason(session_id)
             if closed:
                 await self._refuse_closed_session(
diff --git a/nerve/agent/plan_service.py b/nerve/agent/plan_service.py
index edc999eef..f24ccc447 100644
--- a/nerve/agent/plan_service.py
+++ b/nerve/agent/plan_service.py
@@ -244,8 +244,7 @@ async def request_plan_revision(
     # takes the revision.
     if (
         session_id != _FALLBACK_PLANNER_SESSION
-        and await db.get_session(session_id) is None
-        and await db.get_session_deletion(session_id) is not None
+        and await db.closed_session_reason(session_id) == "deleted"
     ):
         session_id = _FALLBACK_PLANNER_SESSION
     session_title = (
diff --git a/nerve/agent/sessions.py b/nerve/agent/sessions.py
index 69e311964..3e41f8de9 100644
--- a/nerve/agent/sessions.py
+++ b/nerve/agent/sessions.py
@@ -12,9 +12,11 @@
 from __future__ import annotations
 
 import asyncio
+import contextlib
 import logging
 import secrets
 import uuid
+from collections.abc import Iterator
 from datetime import datetime, timedelta, timezone
 from enum import StrEnum
 from typing import Any, Awaitable, Callable, Literal
@@ -117,7 +119,7 @@ class SessionManager:
     # Never raises into the caller.
     on_archived: Callable[[str], Awaitable[None]] | None = None
     # ``run_lock_held(session_id)``: something holds the session's run lock
-    # (the archive guard). Wired to the engine's run locks.
+    # (see :meth:`has_turn_work`). Wired to the engine's run locks.
     run_lock_held: Callable[[str], bool] | None = None
     # Called as ``listener(session_id, running)`` right after a session
     # starts or stops running a turn (see ``add_running_listener``).
@@ -159,6 +161,9 @@ def __init__(
         # Callback for memorizing session before close/archive/delete.
         # Set by AgentEngine after init to wire up memU bridge.
         self._on_memorize: Any | None = None
+        # session_id -> one reason per archive or delete in progress
+        # (see :meth:`closing`).
+        self._closing: dict[str, list[str]] = {}
 
     # ------------------------------------------------------------------ #
     #  Lifecycle: Create / Get                                             #
@@ -656,6 +661,14 @@ def has_running_work(self, session_id: str) -> bool:
             or (task is not None and not task.done())
         )
 
+    def has_turn_work(self, session_id: str) -> bool:
+        """:meth:`has_running_work`, or something holds the session's run
+        lock (a queued message, an autonomous turn, an idle drain). Archive
+        and delete refuse a session with turn work."""
+        return self.has_running_work(session_id) or bool(
+            self.run_lock_held and self.run_lock_held(session_id)
+        )
+
     def get_running_ids(self) -> set[str]:
         """Return the set of currently running session IDs."""
         return set(self._running_sessions)
@@ -1027,11 +1040,8 @@ async def archive_session(
         A session with turn work is refused with :class:`SessionBusyError`,
         also with ``force``: a turn that runs or waits for a concurrency
         slot, or any holder of the session's run lock (a queued message, an
-        autonomous turn, an idle drain). Right after these checks, with no
-        await in between, the session is marked as archiving until its status
-        is written (:meth:`is_archiving`). A run checks the mark right after
-        it holds the run lock, also with no await in between. So a run either
-        makes this call fail or is refused.
+        autonomous turn, an idle drain). The session is then marked as
+        closing until its status is written (:meth:`closing`).
 
         After the archive, the session's pending wakeups are deleted and
         ``on_archived`` runs (it expires the session's pending questions).
@@ -1041,20 +1051,14 @@ async def archive_session(
             raise ValueError(f"Session {session_id} not found")
         if session.get("status") == SessionStatus.ARCHIVED.value:
             return False
-        if self.has_running_work(session_id) or self._run_lock_held(session_id):
+        if self.has_turn_work(session_id):
             raise SessionBusyError(BUSY_RUNNING_REASON)
         if not force and self._agents_live(session_id):
             raise SessionBusyError(BUSY_AGENTS_REASON)
-        marks = self._archiving_marks()
-        marks[session_id] = marks.get(session_id, 0) + 1
-        try:
+        with self.closing(session_id, "archived"):
             archived = await self._archive_marked(
                 session_id, attribution=attribution, details=details,
             )
-        finally:
-            left = marks.pop(session_id, 1) - 1
-            if left:
-                marks[session_id] = left
         if archived:
             logger.info("Archived session %s", session_id)
             await self._close_archived_work(session_id)
@@ -1096,24 +1100,32 @@ async def _memorize_bg(sid: str = session_id) -> None:
             event_type="archived", details=details, attribution=attribution,
         )
 
-    def _archiving_marks(self) -> dict[str, int]:
-        """Archives in progress per session (created on first use, so
-        managers built with ``__new__`` work too)."""
-        return self.__dict__.setdefault("_archiving", {})
-
-    def is_archiving(self, session_id: str) -> bool:
-        """An archive of the session is in progress (see :meth:`archive_session`)."""
-        return session_id in self._archiving_marks()
+    @contextlib.contextmanager
+    def closing(self, session_id: str, reason: str) -> Iterator[None]:
+        """Mark the session as closing (``archived`` or ``deleted``) for the
+        duration of the block (see :meth:`closing_reason`).
 
-    def _run_lock_held(self, session_id: str) -> bool:
-        """``run_lock_held(session_id)``; ``False`` when it is not wired."""
-        if self.run_lock_held is None:
-            return False
+        The caller checks :meth:`has_turn_work` and enters the block with no
+        await in between. A run checks the mark right after it holds the run
+        lock, also with no await in between. So a run either makes the
+        caller's check fail, or is refused.
+        """
+        self._closing.setdefault(session_id, []).append(reason)
         try:
-            return bool(self.run_lock_held(session_id))
-        except Exception:
-            logger.exception("run-lock check failed for %s", session_id)
-            return False
+            yield
+        finally:
+            marks = self._closing[session_id]
+            marks.remove(reason)
+            if not marks:
+                del self._closing[session_id]
+
+    def closing_reason(self, session_id: str) -> str | None:
+        """``deleted`` or ``archived`` while :meth:`closing` marks the
+        session (``deleted`` wins), else ``None``."""
+        marks = self._closing.get(session_id)
+        if not marks:
+            return None
+        return "deleted" if "deleted" in marks else "archived"
 
     async def _close_archived_work(self, session_id: str) -> None:
         """End the work that targets a session that was just archived: its
diff --git a/nerve/db/sessions.py b/nerve/db/sessions.py
index 9632ba8ce..5e2b7bfd7 100644
--- a/nerve/db/sessions.py
+++ b/nerve/db/sessions.py
@@ -465,6 +465,16 @@ async def get_session_deletion(self, session_id: str) -> dict | None:
             row = await cursor.fetchone()
             return dict(row) if row else None
 
+    async def closed_session_reason(self, session_id: str) -> str | None:
+        """``archived`` for an archived session, ``deleted`` for an id that
+        has no row and a ``session_deletions`` record, else ``None``."""
+        session = await self.get_session(session_id)
+        if session is not None:
+            return "archived" if session.get("status") == "archived" else None
+        if await self.get_session_deletion(session_id) is not None:
+            return "deleted"
+        return None
+
     async def update_session_metadata(self, session_id: str, metadata: dict) -> None:
         """Update the metadata JSON for a session.
 
diff --git a/nerve/gateway/routes/sessions.py b/nerve/gateway/routes/sessions.py
index b506758a4..d378a3f5c 100644
--- a/nerve/gateway/routes/sessions.py
+++ b/nerve/gateway/routes/sessions.py
@@ -705,12 +705,10 @@ async def delete_session(session_id: str, actor: Actor = Depends(require_auth)):
     if engine:
         # A turn that runs or waits would write into the deleted rows, or
         # create the session again. The person stops it first.
-        if not engine.begin_delete(session_id):
+        if engine.sessions.has_turn_work(session_id):
             raise HTTPException(status_code=409, detail=_DELETE_BUSY_DETAIL)
-        try:
+        with engine.sessions.closing(session_id, "deleted"):
             await _delete_with_engine(engine, db, session_id, actor)
-        finally:
-            engine.end_delete(session_id)
     else:
         await db.delete_session(session_id, attribution=attribution_for(actor))
         await _cleanup_uploaded_files(db, session_id)
@@ -1147,9 +1145,9 @@ async def chat(req: MessageRequest, actor: Actor = Depends(require_auth)):
         session_id = await deps.engine.sessions.get_active_session(
             "api:default", source="api", actor=actor,
         )
-    elif deps.engine.accepting_turns and not (
-        await deps.db.get_session(session_id) is None
-        and await deps.db.get_session_deletion(session_id) is not None
+    elif (
+        deps.engine.accepting_turns
+        and await deps.db.closed_session_reason(session_id) != "deleted"
     ):
         # A new session id from the caller: a session of the API. A draining
         # engine creates nothing: run() refuses the turn with a 503. The id of
diff --git a/tests/test_closed_sessions.py b/tests/test_closed_sessions.py
index 9db09d40f..d3b91d74f 100644
--- a/tests/test_closed_sessions.py
+++ b/tests/test_closed_sessions.py
@@ -21,7 +21,7 @@
 from nerve.agent.backends import events as ev
 from nerve.agent.backends.base import BackendCapabilities
 from nerve.agent.engine import AgentEngine, SessionClosedError
-from nerve.agent.sessions import BUSY_RUNNING_REASON, SessionBusyError
+from nerve.agent.sessions import BUSY_RUNNING_REASON, SessionBusyError, SessionManager
 from nerve.config import NerveConfig, NotificationsConfig
 from nerve.cron.service import CronService
 from nerve.identity import attribution_for
@@ -336,6 +336,16 @@ async def test_archive_is_refused_while_anything_holds_the_run_lock(tmp_path, db
     assert await engine.sessions.archive_session(sid) is True
 
 
+def test_closing_marks_nest_and_clear():
+    manager = SessionManager(MagicMock())
+    with manager.closing("s", "deleted"):
+        with manager.closing("s", "archived"):
+            assert manager.closing_reason("s") == "deleted"
+        assert manager.closing_reason("s") == "deleted"
+    assert manager.closing_reason("s") is None
+    assert manager._closing == {}
+
+
 @pytest.mark.asyncio
 async def test_a_queued_turn_is_refused_when_the_session_was_archived_meanwhile(
     tmp_path, db, broadcasts,
@@ -412,7 +422,7 @@ async def test_a_turn_that_starts_during_an_archive_is_refused_before_it_stores_
     backend.clients[0].disconnect_gate = gate
     archive = asyncio.create_task(engine.sessions.archive_session(sid))
     await _until(lambda: engine.sessions.get_client(sid) is None)
-    assert engine.sessions.is_archiving(sid)
+    assert engine.sessions.closing_reason(sid) == "archived"
 
     with pytest.raises(SessionClosedError):
         await asyncio.wait_for(
@@ -423,7 +433,7 @@ async def test_a_turn_that_starts_during_an_archive_is_refused_before_it_stores_
     gate.set()
 
     assert await asyncio.wait_for(archive, timeout=5) is True
-    assert not engine.sessions.is_archiving(sid)
+    assert engine.sessions.closing_reason(sid) is None
     assert len(backend.clients) == 1
     users = [m["content"] for m in await db.get_messages(sid) if m["role"] == "user"]
     assert "sent during the archive" not in users
@@ -681,21 +691,28 @@ async def test_a_run_never_creates_a_deleted_session_again(tmp_path, db):
 
 @pytest.mark.asyncio
 async def test_a_run_during_a_delete_is_refused(tmp_path, db):
+    """The delete mark: a run that takes the run lock while the delete
+    disconnects the client is refused and creates nothing."""
     engine, backend = _engine(tmp_path, db)
     sid = "s-deleting"
-    await engine.sessions.get_or_create(sid, source="web", backend="claude", actor=None)
+    await _new_session(engine, db, sid)
+    gate = asyncio.Event()
+    backend.clients[0].disconnect_gate = gate
+    delete = asyncio.create_task(_delete(engine, db, sid))
+    await _until(lambda: engine.sessions.get_client(sid) is None)
+    assert engine.sessions.closing_reason(sid) == "deleted"
 
-    assert engine.begin_delete(sid) is True
-    try:
-        with pytest.raises(SessionClosedError) as info:
-            await engine.run(sid, "a message", source="web", channel="web", actor=None)
-    finally:
-        engine.end_delete(sid)
+    with pytest.raises(SessionClosedError) as info:
+        await asyncio.wait_for(
+            engine.run(sid, "a message", source="web", channel="web", actor=None),
+            timeout=5,
+        )
+    gate.set()
 
     assert info.value.reason == "deleted"
-    assert backend.clients == []
-    # After the delete ends, an existing session runs again.
-    await engine.run(sid, "a message", source="web", channel="web", actor=None)
+    assert await asyncio.wait_for(delete, timeout=5) == {"deleted": True}
+    assert engine.sessions.closing_reason(sid) is None
+    assert await db.get_session(sid) is None
     assert len(backend.clients) == 1
 
 
diff --git a/tests/test_drain.py b/tests/test_drain.py
index f0477e234..f0000f853 100644
--- a/tests/test_drain.py
+++ b/tests/test_drain.py
@@ -46,8 +46,8 @@ def _make_engine() -> AgentEngine:
     config.agent.max_concurrent = 3
     config.mcp_servers = []
     db = AsyncMock()
-    # A run reads the session row: an open session.
-    db.get_session.return_value = {"status": "idle"}
+    # A run checks that the session is open.
+    db.closed_session_reason.return_value = None
     engine = AgentEngine(config, db)
     engine._broadcast_session_running = AsyncMock()
     engine._DRAIN_POLL_SECONDS = 0.01

From 0a33168f06ca62c21cc5ab1076d3087e9de65123 Mon Sep 17 00:00:00 2001
From: Alex Soffronow Pagonidis
 <237136924+alex-clickhouse@users.noreply.github.com>
Date: Tue, 10 Nov 2026 07:56:09 +0000
Subject: [PATCH 4/5] Sessions: one closed-session check per run

run() checks the session right after it holds the run lock. From there
the run lock stays held for the whole turn, and archive and delete
refuse a session whose run lock is held, so the second check at the
client build and its except branch in _run_inner had no caller that
could reach them. The test that wrote status='archived' straight to the
database to reach them goes too.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 nerve/agent/engine.py         | 26 ++++----------------------
 tests/test_closed_sessions.py | 28 ----------------------------
 2 files changed, 4 insertions(+), 50 deletions(-)

diff --git a/nerve/agent/engine.py b/nerve/agent/engine.py
index cbad19ddf..e9d5e6326 100644
--- a/nerve/agent/engine.py
+++ b/nerve/agent/engine.py
@@ -1554,10 +1554,6 @@ async def _get_or_create_client(
 
             # Session row first — backend resolution is sticky on it.
             session = await self.db.get_session(session_id)
-            # Check again before the runtime gets anything.
-            closed = await self._session_closed_reason(session_id)
-            if closed:
-                raise SessionClosedError(closed)
             backend = self._backend_for(session, source)
             requested_model = (
                 model or (session or {}).get("model")
@@ -2124,15 +2120,6 @@ async def drain(self, timeout: float) -> bool:
         logger.info("Drain: all running turns ended")
         return True
 
-    async def _session_closed_reason(self, session_id: str) -> str | None:
-        """``archived`` or ``deleted`` when no turn may run in the session,
-        else ``None``. An archive or a delete in progress counts; that check
-        comes before the first await."""
-        return (
-            self.sessions.closing_reason(session_id)
-            or await self.db.closed_session_reason(session_id)
-        )
-
     async def _refuse_closed_session(
         self,
         session_id: str,
@@ -4077,7 +4064,10 @@ async def _run_turn(
             # 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 = await self._session_closed_reason(session_id)
+            closed = (
+                self.sessions.closing_reason(session_id)
+                or await self.db.closed_session_reason(session_id)
+            )
             if closed:
                 await self._refuse_closed_session(
                     session_id, closed, entry=entry, client_msg_id=client_msg_id,
@@ -4635,14 +4625,6 @@ def _with_session_preamble(base: TurnInput) -> TurnInput:
             await self.schedule_memorize(session_id)
             return TurnResult(partial, outcome)
 
-        except SessionClosedError as e:
-            # Archived or deleted while this turn waited for its client: the
-            # runtime got nothing, so there is no reply and no error state.
-            self._end_turn(session_id, TurnOutcome("interrupted", str(e)))
-            logger.info("Session %s: turn refused (%s)", session_id, e.reason)
-            await broadcaster.broadcast_error(session_id, str(e))
-            raise
-
         except Exception as e:
             error_msg = f"Agent error: {e}"
             logger.error(error_msg, exc_info=True)
diff --git a/tests/test_closed_sessions.py b/tests/test_closed_sessions.py
index d3b91d74f..67841a998 100644
--- a/tests/test_closed_sessions.py
+++ b/tests/test_closed_sessions.py
@@ -381,34 +381,6 @@ async def test_a_queued_turn_is_refused_when_the_session_was_archived_meanwhile(
     assert not lock.locked()
 
 
-@pytest.mark.asyncio
-async def test_a_session_archived_while_its_turn_waits_for_the_client_gets_no_client(
-    tmp_path, db,
-):
-    """The check under the client lock: the archive lands after the checks
-    in ``_run_turn``, while the turn waits to build its client."""
-    engine, backend = _engine(tmp_path, db)
-    sid = "s-build"
-    await engine.sessions.get_or_create(sid, source="web", backend="claude", actor=None)
-    client_lock = engine.sessions.get_lock(sid)
-    await client_lock.acquire()
-    turn = asyncio.create_task(
-        engine.run(sid, "a message", source="web", channel="web", actor=None),
-    )
-    await _until(lambda: engine.sessions.is_running(sid))
-    await db.change_session_status(sid, "archived", event_type="archived")
-    client_lock.release()
-
-    with pytest.raises(SessionClosedError):
-        await asyncio.wait_for(turn, timeout=5)
-    assert backend.clients == []
-    row = await db.get_session(sid)
-    assert row["status"] == "archived"
-    assistant = [m for m in await db.get_messages(sid) if m["role"] == "assistant"]
-    assert assistant == []
-    assert "error" not in await _events(db, sid)
-
-
 @pytest.mark.asyncio
 async def test_a_turn_that_starts_during_an_archive_is_refused_before_it_stores_anything(
     tmp_path, db,

From 9b787f6641655cb59ddcb15964601baa20e7f4cc Mon Sep 17 00:00:00 2001
From: Alex Soffronow Pagonidis
 <237136924+alex-clickhouse@users.noreply.github.com>
Date: Tue, 10 Nov 2026 07:57:47 +0000
Subject: [PATCH 5/5] Sessions: one refusal path for a drain and a closed
 session

_refuse_turn takes the error to raise (default NotAcceptingTurnsError),
so a run in a closed session uses it with SessionClosedError, and the
copy in _refuse_closed_session goes. Every caller that gets a
SessionClosedError already logs it.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 nerve/agent/engine.py | 40 ++++++++++------------------------------
 1 file changed, 10 insertions(+), 30 deletions(-)

diff --git a/nerve/agent/engine.py b/nerve/agent/engine.py
index e9d5e6326..dd25e1195 100644
--- a/nerve/agent/engine.py
+++ b/nerve/agent/engine.py
@@ -2120,39 +2120,17 @@ async def drain(self, timeout: float) -> bool:
         logger.info("Drain: all running turns ended")
         return True
 
-    async def _refuse_closed_session(
-        self,
-        session_id: str,
-        reason: str,
-        *,
-        entry: QueuedTurn | None = None,
-        client_msg_id: str | None = None,
-    ) -> None:
-        """Tell the session's clients that run() refused the turn because the
-        session is closed, then raise :class:`SessionClosedError`.
-
-        As :meth:`_refuse_turn`: a run that has a queue entry or a
-        ``client_msg_id`` leaves the queue with the reason ``refused``. The
-        error event says why.
-        """
-        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,
-                "client_msg_id": client_msg_id, "reason": "refused",
-            }])
-        error = SessionClosedError(reason)
-        logger.info("Session %s: refused a turn (%s)", session_id, reason)
-        await broadcaster.broadcast_error(session_id, str(error))
-        raise error
-
     async def _refuse_turn(
         self,
         session_id: str,
         *,
         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
@@ -2163,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                                 #
@@ -4069,8 +4048,9 @@ async def _run_turn(
                 or await self.db.closed_session_reason(session_id)
             )
             if closed:
-                await self._refuse_closed_session(
-                    session_id, closed, entry=entry, client_msg_id=client_msg_id,
+                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=[{