diff --git a/docs/api.md b/docs/api.md index d707583eb..70862d3a8 100644 --- a/docs/api.md +++ b/docs/api.md @@ -1116,6 +1116,11 @@ The delete removes the session's events and its agent records and transcripts. Nerve records the delete in `session_deletions` with the signed-in person as actor and cause (see [Session changes](accounts.md#session-changes)). +While a turn of the session runs, a message waits in its turn queue, or a +background continuation of the runtime runs, the delete answers `409`: stop +the session first. The delete expires the +session's pending questions; the notification rows stay. A turn never creates +a deleted session again: a run in it is refused. #### `PATCH /api/sessions/{id}` Update `title`, `starred`, `model` or `parent_session_id`. A `model` change @@ -1187,6 +1192,19 @@ its ancestors stay: `session is running` (a turn is in flight) or archives those too; their agents end as `interrupted` (`archived`). `archived` is `false` when nothing was archived. +An archive deletes the session's pending wakeups (a "Run later" message +included) and expires its pending questions. Approvals stay pending, because +their answer acts on their target, not on the session. No turn runs in an +archived session: a wakeup or a queued message is refused, an answer is +stored but starts no turn, and a channel message goes to a new session. Only these +open an archived session again, each with an `unarchived` event: the +unarchive endpoint, a star, and a person's own message into it (the web chat, +`POST /api/chat` with its `session_id`, "Run later", a plan revision request +for a plan that it proposed). The archive itself is refused (`session is +running`) while a turn runs or waits, or while a background continuation of +the runtime runs. A route that would run a turn in an archived session +answers `409` with `{"detail": "...", "reason": "archived"}`. + ```json Response: { "archived": true, "archived_ids": ["a1b2c3d4"], "skipped": [{ "id": "e5f6a7b8", "reason": "background agents running" }] } ``` @@ -1306,7 +1324,10 @@ Response: { "response": "Hi there!", "session_id": "main" } `client_msg_id` is optional: a retry with an id the session saw in the last 10 minutes runs nothing and answers with an empty `response`. While another turn of the session runs, the message waits in the session's -[turn queue](#get-apisessionsidqueue) (source `api`). +[turn queue](#get-apisessionsidqueue) (source `api`). A `session_id` of an +archived session unarchives it first, in the caller's name. A `session_id` of +a deleted session creates nothing and answers `409` with +`{"detail": "This session was deleted.", "reason": "deleted"}`. For streaming, use the WebSocket endpoint. 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 111b849af..dd25e1195 100644 --- a/nerve/agent/engine.py +++ b/nerve/agent/engine.py @@ -143,6 +143,25 @@ def __init__(self) -> None: super().__init__(_NOT_ACCEPTING_TURNS_MESSAGE) +_SESSION_CLOSED_MESSAGES = { + "archived": "This session is archived. Unarchive it to continue.", + "deleted": "This session was deleted.", +} + + +class SessionClosedError(RuntimeError): + """``AgentEngine.run`` refused a turn because its session is closed. + + ``reason`` is ``archived`` (only unarchive opens the session again) or + ``deleted`` (no session row, and ``session_deletions`` records the + delete). The user message is not stored. + """ + + def __init__(self, reason: str) -> None: + super().__init__(_SESSION_CLOSED_MESSAGES[reason]) + self.reason = reason + + # The runtime's system frame that marks a compacted conversation. _COMPACT_BOUNDARY = "compact_boundary" # A compact turn that ended with no such frame: the runtime had no messages. @@ -472,6 +491,10 @@ def __init__(self, config: NerveConfig, db: Database): # guard asks whether agents still run. self.sessions.on_client_removed = self.agents.on_client_gone self.sessions.has_live_agents = self.agents.has_live + # An archive expires the session's pending questions, and is refused + # while anything holds the session's run lock. + self.sessions.on_archived = self.expire_session_questions + self.sessions.run_lock_held = self._run_lock_held # Generation of the live runtime client per build (any session): a # finished agent is resumable only on the client that ran it. self._client_gen = 0 @@ -1150,6 +1173,41 @@ async def _memorizing(self): async with self._memorize_lock: yield + async def reopen_archived( + self, session_id: str, *, attribution: Attribution, + ) -> bool: + """Unarchive ``session_id`` before a person's own message runs in it. + + A message that a person sends into an archived session (the web + chat, ``POST /api/chat``, a plan revision request) opens it again, + recorded as ``unarchived`` with ``attribution``. Every other run in + an archived session is refused (:class:`SessionClosedError`). + Returns ``True`` when this call unarchived the session. + """ + session = await self.db.get_session(session_id) + if not session or session.get("status") != SessionStatus.ARCHIVED.value: + return False + return await self.sessions.unarchive_session( + session_id, attribution=attribution, + ) + + def _run_lock_held(self, session_id: str) -> bool: + """Something holds the session's run lock: a turn, a run that waits + in the queue (the lock stays held across the handover), an + autonomous turn, an idle drain or a client reap.""" + lock = self._session_locks.get(session_id) + return lock is not None and lock.locked() + + async def expire_session_questions(self, session_id: str) -> None: + """Expire the pending questions of a session that was archived or + deleted, with the same reports as an expiry (web, Telegram, Slack). + Without a notification service (CLI, tests), only the rows change.""" + service = self.notification_service + if service is not None: + await service.expire_session_questions(session_id) + else: + await self.db.expire_pending_questions_for_session(session_id) + async def _memorize_session( self, session_id: str, connected_at_override: str | None = None, ) -> None: @@ -2068,8 +2126,11 @@ async def _refuse_turn( *, entry: QueuedTurn | None = None, client_msg_id: str | None = None, + error: RuntimeError | None = None, ) -> None: - """Tell the session's clients that run() refused the turn, then raise. + """Tell the session's clients that run() refused the turn, then raise + ``error`` (default :class:`NotAcceptingTurnsError`). The error event + says why. A run that has a queue entry or a ``client_msg_id`` is reported in ``queue_update`` with the reason ``refused``. The caller removes the @@ -2080,8 +2141,9 @@ async def _refuse_turn( "queue_id": entry.queue_id if entry is not None else None, "client_msg_id": client_msg_id, "reason": "refused", }]) - await broadcaster.broadcast_error(session_id, _NOT_ACCEPTING_TURNS_MESSAGE) - raise NotAcceptingTurnsError() + error = error or NotAcceptingTurnsError() + await broadcaster.broadcast_error(session_id, str(error)) + raise error # ------------------------------------------------------------------ # # Public API: run, stop, fork, resume # @@ -3978,6 +4040,18 @@ async def _run_turn( # The Stop target from the moment this run holds the run lock: # Stop reaches the turn, never a message queued behind it. self.sessions.set_turn_task(session_id, turn_task) + # The session can be archived or deleted before this run holds + # the lock, or while it waits in the queue. No await since the + # lock (see SessionManager.closing). + closed = ( + self.sessions.closing_reason(session_id) + or await self.db.closed_session_reason(session_id) + ) + if closed: + await self._refuse_turn( + session_id, entry=entry, client_msg_id=client_msg_id, + error=SessionClosedError(closed), + ) if entry is not None or client_msg_id: await self._emit_queue_update(session_id, removed=[{ "queue_id": entry.queue_id if entry is not None else None, diff --git a/nerve/agent/plan_service.py b/nerve/agent/plan_service.py index 257773883..f24ccc447 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. @@ -239,6 +240,13 @@ async def request_plan_revision( # Fall back to the cron planner if the plan was created without # a session attribution (older rows, manual inserts, etc.). session_id = plan.get("session_id") or _FALLBACK_PLANNER_SESSION + # A deleted proposer session is not created again: the cron planner + # takes the revision. + if ( + session_id != _FALLBACK_PLANNER_SESSION + and await db.closed_session_reason(session_id) == "deleted" + ): + session_id = _FALLBACK_PLANNER_SESSION session_title = ( f"Cron: {session_id.split(':')[-1]}" if session_id.startswith("cron:") @@ -251,6 +259,12 @@ async def request_plan_revision( actor=engine.db.system_actor, cause=cause, origin=plan_origin(plan_id, plan["task_id"]), ) + # The person's revision request opens an archived planner session again + # (an ``unarchived`` event in their name); a run in an archived session + # is refused. + await engine.reopen_archived( + session_id, attribution=attribution_for(actor, cause), + ) asyncio.create_task( engine.run( session_id=session_id, user_message=prompt, source="cron", diff --git a/nerve/agent/sessions.py b/nerve/agent/sessions.py index ab2240a53..3e41f8de9 100644 --- a/nerve/agent/sessions.py +++ b/nerve/agent/sessions.py @@ -12,12 +12,14 @@ 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, Callable, Literal +from typing import Any, Awaitable, Callable, Literal from nerve.db import Database from nerve.identity import ( @@ -81,11 +83,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 +114,13 @@ 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 + # ``run_lock_held(session_id)``: something holds the session's run lock + # (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``). _running_listeners: tuple[Callable[[str, bool], None], ...] = () @@ -149,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 # @@ -262,12 +277,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 +306,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 +419,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: @@ -628,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) @@ -995,14 +1036,39 @@ 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 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). 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). """ 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 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) + with self.closing(session_id, "archived"): + archived = await self._archive_marked( + session_id, attribution=attribution, details=details, + ) + 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: @@ -1024,7 +1090,7 @@ async def _memorize_bg(sid: str = session_id) -> None: except Exception: pass now = datetime.now(timezone.utc).isoformat() - archived = await self.db.change_session_status( + return await self.db.change_session_status( session_id, SessionStatus.ARCHIVED.value, fields={ "archived_at": now, @@ -1033,9 +1099,52 @@ async def _memorize_bg(sid: str = session_id) -> None: }, event_type="archived", details=details, attribution=attribution, ) - if archived: - logger.info("Archived session %s", session_id) - return archived + + @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`). + + 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: + 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 + 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 +1225,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 +1305,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 +1321,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 +1347,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..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. @@ -530,6 +540,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..d378a3f5c 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 " @@ -701,55 +703,79 @@ 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 engine.sessions.has_turn_work(session_id): + raise HTTPException(status_code=409, detail=_DELETE_BUSY_DETAIL) + with engine.sessions.closing(session_id, "deleted"): + await _delete_with_engine(engine, db, session_id, actor) 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.""" @@ -823,6 +849,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: @@ -1113,12 +1145,21 @@ 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 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. + # 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), ) + # 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..67841a998 --- /dev/null +++ b/tests/test_closed_sessions.py @@ -0,0 +1,796 @@ +"""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, SessionManager +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_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 + + +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, +): + """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) + 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) + + await db.change_session_status(sid, "archived", event_type="archived") + 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_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) + 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) + assert engine.sessions.closing_reason(sid) == "archived" + + 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 + 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 + 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) + + +@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 +# --------------------------------------------------------------------------- # + +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_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) + 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): + """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 _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" + + 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 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 + + +# --------------------------------------------------------------------------- # +# 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..f0000f853 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 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 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..5cc0bc423 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,25 @@ 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_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/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((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.'); + } } },