[eric] agents: a user's Stop or Close is final; no watchdog resend, disk reload, read, or late-booting child revives the card

backend/apps/agents/agent_manager.py

@@ -24,6 +24,8 @@ from backend.apps.agents.manager.session.SessionLifecycle import SessionLifecycl
 from backend.apps.agents.manager.SpawnAgentRun import SpawnAgentRun
 from backend.apps.agents.manager.session.SessionPersistence import SessionPersistence
 from backend.apps.agents.manager.Messaging import Messaging, QueuedMessage
+from backend.apps.agents.manager.EditMessage import EditMessage
+from backend.apps.agents.manager.session.SessionHistory import SessionHistory
 from backend.apps.agents.manager.SessionControl import SessionControl
 from backend.apps.agents.manager.AgentLaunch import AgentLaunch
 from backend.apps.agents.manager.MockAgent import MockAgent
@@ -42,7 +44,7 @@ os.environ.setdefault("CLAUDE_CODE_STREAM_CLOSE_TIMEOUT", "3600000")
 MAX_CONCURRENT_TURNS = int(os.environ.get("OSW_MAX_CONCURRENT_TURNS", "8") or "0")
 
 
-class AgentManager(SessionLifecycle, SessionPersistence, Messaging, SessionControl, AgentLaunch, SpawnAgentRun, MockAgent, TurnRunner, RunOptions, RunSupport):
+class AgentManager(SessionLifecycle, SessionHistory, SessionPersistence, Messaging, EditMessage, SessionControl, AgentLaunch, SpawnAgentRun, MockAgent, TurnRunner, RunOptions, RunSupport):
     @typechecked
     async def dispatch_hidden_continuation(self, session_id: str, prompt: str, delay_s: int) -> None:
         """Send the self-heal continuation after delay_s (codex rotation windows need ~75s; an

backend/apps/agents/agents.py

@@ -221,6 +221,10 @@ async def send_message(session_id: str, body: dict):
 
 @agents.router.post("/sessions/{session_id}/stop")
 async def stop_agent(session_id: str):
+    # Only a human reaches this route; the watchdogs call stop_agent directly and THEIR stops may resend. Stamp here so the two can be told apart downstream.
+    p_s = agent_manager.sessions.get(session_id)
+    if p_s is not None:
+        p_s.ended_by_user = True
     await agent_manager.stop_agent(session_id)
     # A stopped turn's parked AskUI waits would otherwise zombie for 600s and eat the next click (ENG-232).
     from backend.apps.agents.ui_request_bridge import cancel_session_waits
@@ -336,6 +340,9 @@ async def duplicate_session(session_id: str, body: dict = {}):
 
 @agents.router.post("/sessions/{session_id}/close")
 async def close_session(session_id: str):
+    p_s = agent_manager.sessions.get(session_id)
+    if p_s is not None:
+        p_s.ended_by_user = True
     try:
         await agent_manager.close_session(session_id)
     except ValueError as e:

backend/apps/agents/browser/browser_agent.py

@@ -1088,10 +1088,26 @@ async def run_browser_agent(
     agent_manager.cancel_events[session_id] = cancel_event
     agent_manager.sessions[session_id] = session
 
-    # If parent was already stopped before we registered, bail immediately
+    # If the parent was stopped, user-ended, or already PURGED before we registered, bail now. A child
+    # spends its first seconds in prestage with no cancel_event yet, so a close that lands in that
+    # window stops every sibling it can see and misses this one; close_session then purges the
+    # parent, the old `parent.status == "stopped"` check saw None, and the orphan ran its whole task
+    # on a closed card (Eric's "it came back for a bit", 2026-08-20: born 18:32:08, parent closed
+    # 18:32:12, still driving Amazon minutes later). No parent in memory is the same verdict as a
+    # stopped one: nobody is waiting for this result.
     if parent_session_id:
-        parent = agent_manager.sessions.get(parent_session_id)
-        if parent and parent.status == "stopped":
+        # Live parent first; a CLOSED parent is purged from memory but its on-disk record still says
+        # ended_by_user, so fall back to that. Never cancel on mere absence: a parent that is simply
+        # not cached yet is not a closed one (the loop's own skill tests model that shape).
+        parent = agent_manager.get_session(parent_session_id)
+        p_ended = False
+        if parent is not None:
+            p_ended = parent.status == "stopped" or bool(getattr(parent, "ended_by_user", False))
+        else:
+            from backend.apps.agents.manager.session.session_store import load_session_data
+            p_rec = load_session_data(parent_session_id) or {}
+            p_ended = bool(p_rec.get("ended_by_user")) or p_rec.get("status") == "stopped"
+        if p_ended:
             cancel_event.set()
 
     await ws_manager.send_to_session(session_id, "agent:status", {

backend/apps/agents/core/models.py

@@ -162,6 +162,8 @@ class AgentSession(BaseModel):
     # Set once the provider gives a verdict waiting cannot change (a spent plan, a dead credential).
     # Further recovery retries after that only produce cards contradicting the one we already showed.
     provider_verdict_final: bool = False
+    # A HUMAN ended this session (Stop, close, delete). Every automatic resume path (delegation watchdog retry, crash auto-resume, hidden continuation, a read reviving it from disk) stands down; only the human's own next message clears it.
+    ended_by_user: bool = False
     # The last provider-error KIND surfaced this ask. Cards alternated (spent/rate-limit/spent) so
     # the identical-string dedup never engaged and the user got a wall of contradictions.
     last_provider_error_kind: str = ""

backend/apps/agents/manager/EditMessage.py

@@ -0,0 +1,104 @@
+"""Edit a prior user message: fork a branch at it and run the turn again from there. Split out of
+Messaging so each file keeps one job; self.* resolves across the AgentManager MRO as before."""
+
+import asyncio
+from uuid import uuid4
+
+from typeguard import typechecked
+
+from backend.apps.agents.core.models import Message, MessageBranch
+from backend.apps.agents.core.ws_manager import ws_manager
+from backend.apps.agents.manager.session.session_store import snapshot_session_now
+from backend.apps.agents.manager.AgentManagerProtocol import AgentManagerProtocol
+
+
+class EditMessage(AgentManagerProtocol):
+    @typechecked
+    async def edit_message(self, session_id: str, message_id: str, new_content: str):
+        """Edit a prior user message, creating a new branch (fork)."""
+        session = self.sessions.get(session_id)
+        if not session:
+            raise ValueError(f"Session {session_id} not found")
+
+        existing = self.tasks.get(session_id)
+        if existing and not existing.done():
+            existing.cancel()
+            try:
+                await existing
+            except asyncio.CancelledError:
+                pass
+
+        target_msg = None
+        for i, msg in enumerate(session.messages):
+            if msg.id == message_id:
+                target_msg = msg
+                break
+
+        if not target_msg or target_msg.role != "user":
+            raise ValueError("Can only edit user messages")
+
+        fork_point_id = message_id
+        fork_parent_branch = target_msg.branch_id
+
+        msg_branch = session.branches.get(target_msg.branch_id)
+        if msg_branch and msg_branch.fork_point_message_id:
+            branch_user_msgs = [
+                m for m in session.messages
+                if m.branch_id == target_msg.branch_id and m.role == "user"
+            ]
+            if branch_user_msgs and branch_user_msgs[0].id == message_id:
+                fork_point_id = msg_branch.fork_point_message_id
+                fork_parent_branch = msg_branch.parent_branch_id or "main"
+
+        new_branch_id = uuid4().hex
+        new_branch = MessageBranch(
+            id=new_branch_id,
+            parent_branch_id=fork_parent_branch,
+            fork_point_message_id=fork_point_id,
+        )
+        session.branches[new_branch_id] = new_branch
+        session.active_branch_id = new_branch_id
+        session.needs_fresh_session = True
+
+
+        edited_msg = Message(
+            role="user",
+            content=new_content,
+            branch_id=new_branch_id,
+            parent_id=target_msg.parent_id,
+            images=target_msg.images,
+            context_paths=target_msg.context_paths,
+            forced_tools=target_msg.forced_tools,
+            attached_skills=target_msg.attached_skills,
+        )
+        session.messages.append(edited_msg)
+        # Same status-before-snapshot rule as send_message: a stale terminal status on disk hides a mid-turn dirty death from the crash detector.
+        session.status = "running"
+        snapshot_session_now(session)
+
+        await ws_manager.send_to_session(session_id, "agent:message", {
+            "session_id": session_id,
+            "message": edited_msg.model_dump(mode="json"),
+        })
+        await ws_manager.send_to_session(session_id, "agent:branch_created", {
+            "session_id": session_id,
+            "branch": new_branch.model_dump(mode="json"),
+            "active_branch_id": new_branch_id,
+        })
+
+        session.status = "running"
+        await ws_manager.send_to_session(session_id, "agent:status", {
+            "session_id": session_id,
+            "status": "running",
+            "session": session.model_dump(mode="json"),
+        })
+
+        task = asyncio.create_task(self.run_agent_loop(
+            session_id, new_content,
+            images=target_msg.images,
+            context_paths=target_msg.context_paths,
+            forced_tools=target_msg.forced_tools,
+            attached_skills=target_msg.attached_skills,
+            fork_session=True,
+        ))
+        self.register_turn_task(session_id, task)

backend/apps/agents/manager/Messaging.py

@@ -1,6 +1,6 @@
-"""Turn-producing message operations for AgentManager (send + edit), the ones that append a
-user Message and spawn the agent loop. Session-control ops (stop / approve / branch / update)
-live in SessionControl. Pure relocation: self.* resolves across the MRO as before."""
+"""Turn-producing message operations for AgentManager (send + queue), the ones that append a
+user Message and spawn the agent loop. Editing a prior message lives in EditMessage; session-control
+ops (stop / approve / branch / update) live in SessionControl. Pure relocation: self.* resolves across the MRO as before."""
 
 import asyncio
 import logging
@@ -8,9 +8,8 @@ from typing import List, Optional
 
 from pydantic import BaseModel, ConfigDict
 from typeguard import typechecked
-from uuid import uuid4
 
-from backend.apps.agents.core.models import AgentSession, Message, MessageBranch
+from backend.apps.agents.core.models import AgentSession, Message
 from backend.apps.agents.core.ws_manager import ws_manager
 from backend.apps.settings.settings import load_settings
 from backend.apps.agents.manager.run_browser_fast_path import run_browser_fast_path
@@ -71,12 +70,20 @@ class Messaging(AgentManagerProtocol):
             data = load_session_data(session_id)
             if data:
                 session = AgentSession(**data)
+                # This disk reload (and the closed_at wipe below) is how a late watchdog retry reopened a card the user had closed; a machine send must not revive it.
+                if hidden and session.ended_by_user:
+                    return
                 apply_context_window(session)
                 session.closed_at = None
                 self.sessions[session_id] = session
             else:
                 raise ValueError(f"Session {session_id} not found")
-        
+        # Every automatic resume arrives hidden; a human's Stop or close outranks all of them, and only the human's own (never hidden) next message lifts the hold.
+        if hidden and session.ended_by_user:
+            return
+        if not hidden and session.ended_by_user:
+            session.ended_by_user = False
+
         existing = self.tasks.get(session_id)
         if existing and not existing.done():
             # A mid-turn message used to be silently dropped here (no bubble, no trace); queue it and the turn task's done callback replays it.
@@ -226,94 +233,3 @@ class Messaging(AgentManagerProtocol):
             selected_setting_ids=qm.selected_setting_ids,
             client_message_id=qm.client_message_id,
         ))
-
-    @typechecked
-    async def edit_message(self, session_id: str, message_id: str, new_content: str):
-        """Edit a prior user message, creating a new branch (fork)."""
-        session = self.sessions.get(session_id)
-        if not session:
-            raise ValueError(f"Session {session_id} not found")
-
-        existing = self.tasks.get(session_id)
-        if existing and not existing.done():
-            existing.cancel()
-            try:
-                await existing
-            except asyncio.CancelledError:
-                pass
-
-        target_msg = None
-        for i, msg in enumerate(session.messages):
-            if msg.id == message_id:
-                target_msg = msg
-                break
-
-        if not target_msg or target_msg.role != "user":
-            raise ValueError("Can only edit user messages")
-
-        fork_point_id = message_id
-        fork_parent_branch = target_msg.branch_id
-
-        msg_branch = session.branches.get(target_msg.branch_id)
-        if msg_branch and msg_branch.fork_point_message_id:
-            branch_user_msgs = [
-                m for m in session.messages
-                if m.branch_id == target_msg.branch_id and m.role == "user"
-            ]
-            if branch_user_msgs and branch_user_msgs[0].id == message_id:
-                fork_point_id = msg_branch.fork_point_message_id
-                fork_parent_branch = msg_branch.parent_branch_id or "main"
-
-        new_branch_id = uuid4().hex
-        new_branch = MessageBranch(
-            id=new_branch_id,
-            parent_branch_id=fork_parent_branch,
-            fork_point_message_id=fork_point_id,
-        )
-        session.branches[new_branch_id] = new_branch
-        session.active_branch_id = new_branch_id
-        session.needs_fresh_session = True
-
-
-        edited_msg = Message(
-            role="user",
-            content=new_content,
-            branch_id=new_branch_id,
-            parent_id=target_msg.parent_id,
-            images=target_msg.images,
-            context_paths=target_msg.context_paths,
-            forced_tools=target_msg.forced_tools,
-            attached_skills=target_msg.attached_skills,
-        )
-        session.messages.append(edited_msg)
-        # Same status-before-snapshot rule as send_message: a stale terminal status on disk hides a mid-turn dirty death from the crash detector.
-        session.status = "running"
-        snapshot_session_now(session)
-
-        await ws_manager.send_to_session(session_id, "agent:message", {
-            "session_id": session_id,
-            "message": edited_msg.model_dump(mode="json"),
-        })
-        await ws_manager.send_to_session(session_id, "agent:branch_created", {
-            "session_id": session_id,
-            "branch": new_branch.model_dump(mode="json"),
-            "active_branch_id": new_branch_id,
-        })
-
-        session.status = "running"
-        await ws_manager.send_to_session(session_id, "agent:status", {
-            "session_id": session_id,
-            "status": "running",
-            "session": session.model_dump(mode="json"),
-        })
-
-        task = asyncio.create_task(self.run_agent_loop(
-            session_id, new_content,
-            images=target_msg.images,
-            context_paths=target_msg.context_paths,
-            forced_tools=target_msg.forced_tools,
-            attached_skills=target_msg.attached_skills,
-            fork_session=True,
-        ))
-        self.register_turn_task(session_id, task)
-

backend/apps/agents/manager/session/SessionHistory.py

@@ -0,0 +1,82 @@
+"""The chat-history query: paginated, searchable summaries of every session, live ones included.
+Read-only; split out of SessionLifecycle so that file keeps to lifecycle. self.sessions resolves
+across the AgentManager MRO as before."""
+
+from typing import Dict, Optional
+
+from typeguard import typechecked
+
+from backend.apps.agents.manager.session.session_store import load_all_session_data, build_search_text
+from backend.apps.agents.manager.AgentManagerProtocol import AgentManagerProtocol
+
+# Agent-spawned children, never a chat the user started, so they stay out of chat history.
+P_NON_CHAT_MODES = {"browser-agent", "sub-agent", "invoked-agent", "app-agent"}
+
+
+class SessionHistory(AgentManagerProtocol):
+    @typechecked
+    def get_history(
+        self,
+        q: str = "",
+        limit: int = 20,
+        offset: int = 0,
+        dashboard_id: Optional[str] = None,
+        closed_only: bool = False,
+    ) -> Dict:
+        """Return paginated, optionally filtered summaries of sessions, live ones included."""
+        # A malformed file (a list, a bare string) would blow up data.get and 500 the whole endpoint.
+        all_data = [pair for pair in load_all_session_data() if isinstance(pair[1], dict)]
+        # Live sessions usually also have a disk copy (boot restore keeps the file), but the disk copy lags the turn in flight; memory wins the dedupe because it is never staler.
+        in_memory = set(self.sessions.keys())
+        all_data = [pair for pair in all_data if pair[0] not in in_memory and pair[1].get("id") not in in_memory]
+        for sid, session in self.sessions.items():
+            all_data.append((sid, {
+                "id": sid,
+                "name": session.name,
+                "status": session.status,
+                "model": session.model,
+                "mode": session.mode,
+                "created_at": session.created_at.isoformat() if session.created_at else None,
+                "closed_at": None,
+                "cost_usd": session.cost_usd,
+                "dashboard_id": session.dashboard_id,
+                "search_text": build_search_text(session),
+            }))
+        # Sort on last-activity, not closed_at: keying on closed_at alone sorted every live chat ("" ) below every finished one, i.e. off page 1.
+        all_data.sort(key=lambda pair: str(pair[1].get("closed_at") or pair[1].get("created_at") or ""), reverse=True)
+
+        q_lower = q.strip().lower()
+        history = []
+        for sid, data in all_data:
+            # Children are machinery, not chats: a busy user's real history was buried under hundreds of "Browser Agent" rows.
+            if data.get("mode") in P_NON_CHAT_MODES:
+                continue
+            # The boot fetch wants CLOSED sessions only: open ones landing in the client's history map made its resurrection gate swallow their terminal frames. Search keeps the full pool (open sessions on other dashboards are reachable nowhere else).
+            if closed_only and not data.get("closed_at"):
+                continue
+            if dashboard_id and data.get("dashboard_id") != dashboard_id:
+                continue
+            if q_lower:
+                name = (data.get("name") or "").lower()
+                search_text = (data.get("search_text") or "").lower()
+                if q_lower not in name and q_lower not in search_text:
+                    continue
+            history.append({
+                "id": data.get("id", sid),
+                "name": data.get("name", "Untitled"),
+                "status": data.get("status", "stopped"),
+                "model": data.get("model", "sonnet"),
+                "mode": data.get("mode", "agent"),
+                "created_at": data.get("created_at"),
+                "closed_at": data.get("closed_at"),
+                "cost_usd": data.get("cost_usd", 0),
+                "dashboard_id": data.get("dashboard_id"),
+            })
+
+        total = len(history)
+        page = history[offset : offset + limit]
+        return {
+            "sessions": page,
+            "total": total,
+            "has_more": offset + limit < total,
+        }

backend/apps/agents/manager/session/SessionLifecycle.py

@@ -5,7 +5,7 @@ self.tasks / self.stop_agent across the MRO exactly as it did inline, so behavio
 import asyncio
 import logging
 from datetime import datetime
-from typing import Dict, List, Optional, Set
+from typing import List, Optional, Set
 
 from typeguard import typechecked
 
@@ -27,10 +27,6 @@ from backend.apps.agents.manager.run.client_pool import dispose_client_soon
 
 logger = logging.getLogger(__name__)
 
-# Agent-spawned children, never a chat the user started, so they stay out of chat history.
-P_NON_CHAT_MODES = {"browser-agent", "sub-agent", "invoked-agent", "app-agent"}
-
-
 from backend.apps.agents.manager.AgentManagerProtocol import AgentManagerProtocol
 
 
@@ -143,6 +139,9 @@ class SessionLifecycle(AgentManagerProtocol):
         if session_id in self.sessions:
             return self.sessions[session_id]
         session = resume_and_duplicate.load_session_for_resume(session_id)
+        # Merely LOOKING at a closed card (a dashboard refresh, the history list, any poll) used to put it back in memory and broadcast agent:status, which repainted the card the user had just closed. Hand back the on-disk record and leave it closed.
+        if getattr(session, "ended_by_user", False):
+            return session
         self.sessions[session_id] = session
         await ws_manager.send_to_session(session_id, "agent:status", {
             "session_id": session_id,
@@ -152,73 +151,6 @@ class SessionLifecycle(AgentManagerProtocol):
         logger.info(f"Session {session_id} resumed from history")
         return session
 
-    @typechecked
-    def get_history(
-        self,
-        q: str = "",
-        limit: int = 20,
-        offset: int = 0,
-        dashboard_id: Optional[str] = None,
-        closed_only: bool = False,
-    ) -> Dict:
-        """Return paginated, optionally filtered summaries of sessions, live ones included."""
-        # A malformed file (a list, a bare string) would blow up data.get and 500 the whole endpoint.
-        all_data = [pair for pair in load_all_session_data() if isinstance(pair[1], dict)]
-        # Live sessions usually also have a disk copy (boot restore keeps the file), but the disk copy lags the turn in flight; memory wins the dedupe because it is never staler.
-        in_memory = set(self.sessions.keys())
-        all_data = [pair for pair in all_data if pair[0] not in in_memory and pair[1].get("id") not in in_memory]
-        for sid, session in self.sessions.items():
-            all_data.append((sid, {
-                "id": sid,
-                "name": session.name,
-                "status": session.status,
-                "model": session.model,
-                "mode": session.mode,
-                "created_at": session.created_at.isoformat() if session.created_at else None,
-                "closed_at": None,
-                "cost_usd": session.cost_usd,
-                "dashboard_id": session.dashboard_id,
-                "search_text": build_search_text(session),
-            }))
-        # Sort on last-activity, not closed_at: keying on closed_at alone sorted every live chat ("" ) below every finished one, i.e. off page 1.
-        all_data.sort(key=lambda pair: str(pair[1].get("closed_at") or pair[1].get("created_at") or ""), reverse=True)
-
-        q_lower = q.strip().lower()
-        history = []
-        for sid, data in all_data:
-            # Children are machinery, not chats: a busy user's real history was buried under hundreds of "Browser Agent" rows.
-            if data.get("mode") in P_NON_CHAT_MODES:
-                continue
-            # The boot fetch wants CLOSED sessions only: open ones landing in the client's history map made its resurrection gate swallow their terminal frames. Search keeps the full pool (open sessions on other dashboards are reachable nowhere else).
-            if closed_only and not data.get("closed_at"):
-                continue
-            if dashboard_id and data.get("dashboard_id") != dashboard_id:
-                continue
-            if q_lower:
-                name = (data.get("name") or "").lower()
-                search_text = (data.get("search_text") or "").lower()
-                if q_lower not in name and q_lower not in search_text:
-                    continue
-            history.append({
-                "id": data.get("id", sid),
-                "name": data.get("name", "Untitled"),
-                "status": data.get("status", "stopped"),
-                "model": data.get("model", "sonnet"),
-                "mode": data.get("mode", "agent"),
-                "created_at": data.get("created_at"),
-                "closed_at": data.get("closed_at"),
-                "cost_usd": data.get("cost_usd", 0),
-                "dashboard_id": data.get("dashboard_id"),
-            })
-
-        total = len(history)
-        page = history[offset : offset + limit]
-        return {
-            "sessions": page,
-            "total": total,
-            "has_more": offset + limit < total,
-        }
-
     @typechecked
     async def duplicate_session(self, session_id: str, dashboard_id: Optional[str] = None, up_to_message_id: Optional[str] = None) -> AgentSession:
         new_session = resume_and_duplicate.build_duplicate_session(self.sessions.get(session_id), session_id, dashboard_id, up_to_message_id)

backend/apps/agents/manager/streaming/unwedge_sidecar.py

@@ -216,6 +216,10 @@ def delegation_children_settled(session_id: str, since: float) -> bool:
     the watchdog shot a healthy sidecar mid-run (39 kills + 40 force-ended turns in one afternoon
     of concurrent load, measured 2026-08-16 on the packaged build)."""
     from backend.apps.agents.agent_manager import agent_manager
+    # A user's Stop stops the children first, so they read as terminal and the outstanding call looked like a lost result; stage 3 then force-ended the turn and resent RETRY_PROMPT into the session the user had just stopped, every ~150s (reproduced twice, 2026-08-20). A human ending the parent is not a lost result.
+    p_parent = agent_manager.sessions.get(session_id)
+    if p_parent is not None and getattr(p_parent, "ended_by_user", False):
+        return False
     kids = []
     for s in agent_manager.sessions.values():
         if getattr(s, "parent_session_id", None) != session_id or getattr(s, "mode", "") != "browser-agent":
@@ -272,14 +276,14 @@ def arm_delegation_watchdog(ctx: object, tool_use_id: str, tool_name: str) -> No
             # armed above can never fire; the turn has to be ENDED for anything to move. A
             # cancelled turn skips the continuation hook by design, so dispatch the retry here.
             logger.warning("delegation recovery stage 3: force-ending the wedged turn on %s", session_id[:8])
-            loop.create_task(p_force_recover(session_id, getattr(ctx, "session", None)))
+            loop.create_task(force_recover(session_id, getattr(ctx, "session", None)))
             return
         loop.call_later(DELEGATION_CHECK_SECONDS, p_check)
 
     loop.call_later(DELEGATION_CHECK_SECONDS, p_check)
 
 
-async def p_force_recover(session_id: str, session: object) -> None:
+async def force_recover(session_id: str, session: object) -> None:
     from backend.apps.agents.agent_manager import agent_manager
     try:
         if session is not None:

backend/tests/test_browser_agent_loop.py

@@ -632,6 +632,8 @@ def test_skill_keys_on_parent_user_message_so_reformulations_share_a_skill(monke
 
     class p_Parent:
         messages = [p_Msg("user", 'search Wikipedia for "Ada Lovelace"')]
+        status = "running"
+        ended_by_user = False
     monkeypatch.setattr(am_mod.agent_manager, "get_session", lambda sid: p_Parent(), raising=False)
 
     # Run 1: ONE reformulation of the request -> learns a skill keyed on the parent's user message (not this delegated wording).
@@ -672,6 +674,8 @@ def test_skill_key_falls_back_to_delegated_task_on_multi_quote_message(monkeypat
 
     class p_Parent:
         messages = [p_Msg("user", 'search Wikipedia for "Ada Lovelace" and also "Grace Hopper"')]
+        status = "running"
+        ended_by_user = False
     monkeypatch.setattr(am_mod.agent_manager, "get_session", lambda sid: p_Parent(), raising=False)
 
     primary = FakeLLM([

backend/tests/test_delegation_result_backstop.py

@@ -60,7 +60,7 @@ async def test_two_settled_checks_fire_the_recovery(monkeypatch):
     # The watchdog keeps counting past stage 2; stub stage 3 so no stray task touches the real manager.
     async def p_noop(sid, session):
         fired.setdefault("stage3", sid)
-    monkeypatch.setattr(u, "p_force_recover", p_noop)
+    monkeypatch.setattr(u, "force_recover", p_noop)
 
     sess = AgentSession(id="par-2", name="p", model="sonnet")
     ctx = P_Ctx(sess, {"tu-1": 0.0})

backend/tests/test_session_snapshot_on_send.py

@@ -61,10 +61,11 @@ def test_every_user_message_append_site_snapshots():
     # The chokepoint audit: a new send path that forgets the snapshot reintroduces the bug for that
     # path only, which is exactly how the class comes back. Enumerate the sites.
     import backend.apps.agents.manager.AgentLaunch as launch
+    import backend.apps.agents.manager.EditMessage as edit_message
     import backend.apps.agents.manager.Messaging as messaging
     import backend.apps.agents.manager.SpawnAgentRun as spawn
 
-    for mod, expected_appends in ((messaging, 2), (launch, 1), (spawn, 1)):
+    for mod, expected_appends in ((messaging, 1), (edit_message, 1), (launch, 1), (spawn, 1)):
         src = open(mod.__file__).read()
         appends = src.count('.messages.append(user_msg)') + src.count('.messages.append(edited_msg)')
         snaps = src.count('snapshot_session_now(')

backend/tests/test_user_stop_is_final.py

@@ -0,0 +1,207 @@
+"""A human's Stop or close is final. Three automatic paths used to undo it, all reproduced live on
+2026-08-20 (QA_LEDGER, same date): the delegation watchdog resent RETRY_PROMPT ~150s after a user
+Stop (twice), a late machine send reloaded a CLOSED session from disk and wiped closed_at, and a plain
+GET revived a closed card into memory and repainted it. One fact, ended_by_user, stamped only by the
+human-facing routes, and honoured at each door.
+
+Every positive test drives the REAL path (the real watchdog recovery, the real send_message reload,
+the real resume_session), never a stand-in. Every door has a negative control, because the watchdog
+and the resume paths exist for good reasons and gutting them is its own regression.
+"""
+
+import asyncio
+
+from backend.apps.agents.agent_manager import agent_manager
+from backend.apps.agents.agents import close_session as user_close_route
+from backend.apps.agents.agents import stop_agent as user_stop_route
+from backend.apps.agents.core.models import AgentSession
+from backend.apps.agents.manager.streaming import unwedge_sidecar
+
+
+def p_live(name="t") -> AgentSession:
+    s = AgentSession(name=name, model="sonnet-5", dashboard_id="d")
+    s.status = "running"
+    agent_manager.sessions[s.id] = s
+    return s
+
+
+def p_spy_loop():
+    started = []
+    real = agent_manager.run_agent_loop
+
+    async def spy(sid, *a, **k):
+        started.append(sid)
+
+    agent_manager.run_agent_loop = spy
+    return started, real
+
+
+# --- who stamps the fact ------------------------------------------------------------------------
+
+def test_only_the_human_routes_stamp_ended_by_user():
+    s = p_live()
+    asyncio.run(user_stop_route(s.id))
+    assert s.ended_by_user is True
+    t = p_live()
+    asyncio.run(agent_manager.stop_agent(t.id))        # how the watchdogs call it
+    assert t.ended_by_user is False, "an internal stop must keep its right to resend"
+
+
+# --- DOOR 1: the delegation watchdog -------------------------------------------------------------
+
+def test_a_user_stopped_parent_is_never_a_lost_result():
+    """The children go `stopped` first, which used to read as 'every child terminal, result lost'."""
+    parent = p_live("parent")
+    child = AgentSession(name="child", model="sonnet-5", dashboard_id="d")
+    child.mode = "browser-agent"
+    child.parent_session_id = parent.id
+    child.status = "stopped"
+    agent_manager.sessions[child.id] = child
+    asyncio.run(user_stop_route(parent.id))
+    assert unwedge_sidecar.delegation_children_settled(parent.id, 0.0) is False
+
+
+def test_a_genuinely_lost_result_still_settles():
+    """NEGATIVE CONTROL. Stage-3 recovery exists because a CLI blocked 20+ minutes never notices a
+    killed sidecar; a parent nobody stopped, with every child terminal, must still trip it."""
+    parent = p_live("parent2")
+    child = AgentSession(name="child2", model="sonnet-5", dashboard_id="d")
+    child.mode = "browser-agent"
+    child.parent_session_id = parent.id
+    child.status = "completed"
+    agent_manager.sessions[child.id] = child
+    assert unwedge_sidecar.delegation_children_settled(parent.id, 0.0) is True
+
+
+# --- DOOR 2: a machine send into a stopped or closed session -------------------------------------
+
+def test_the_watchdog_resend_cannot_restart_a_user_stopped_session():
+    """Drives the REAL stage-3 recovery (stop, sleep, RETRY_PROMPT)."""
+    s = p_live()
+    asyncio.run(user_stop_route(s.id))
+    started, real = p_spy_loop()
+    try:
+        asyncio.run(unwedge_sidecar.force_recover(s.id, s))
+    finally:
+        agent_manager.run_agent_loop = real
+    assert started == []
+    assert s.status == "stopped"
+
+
+def test_the_watchdog_resend_still_fires_on_a_session_nobody_stopped():
+    """NEGATIVE CONTROL for door 2."""
+    s = p_live()
+    started, real = p_spy_loop()
+    try:
+        asyncio.run(unwedge_sidecar.force_recover(s.id, s))
+    finally:
+        agent_manager.run_agent_loop = real
+    assert started == [s.id]
+
+
+def test_a_closed_card_is_not_reopened_from_disk_by_a_hidden_send():
+    s = p_live()
+    s.status = "completed"
+    asyncio.run(user_close_route(s.id))
+    assert s.id not in agent_manager.sessions, "precondition: close purged it"
+    started, real = p_spy_loop()
+    try:
+        asyncio.run(agent_manager.send_message(s.id, "carry on", hidden=True))
+    finally:
+        agent_manager.run_agent_loop = real
+    assert started == []
+    assert s.id not in agent_manager.sessions, "and it must not even be reloaded"
+
+
+def test_the_users_own_next_message_lifts_the_hold():
+    """NEGATIVE CONTROL: the hold must not brick the chat. A human typing is never hidden."""
+    s = p_live()
+    s.status = "stopped"
+    s.ended_by_user = True
+    started, real = p_spy_loop()
+    try:
+        asyncio.run(agent_manager.send_message(s.id, "ok keep going", hidden=False))
+    finally:
+        agent_manager.run_agent_loop = real
+    assert started == [s.id]
+    assert s.ended_by_user is False
+
+
+# --- DOOR 3: merely reading a closed session -----------------------------------------------------
+
+def test_reading_a_closed_session_does_not_revive_or_repaint_it():
+    s = p_live()
+    s.status = "completed"
+    asyncio.run(user_close_route(s.id))
+    assert s.id not in agent_manager.sessions
+    sent = []
+    from backend.apps.agents.core import ws_manager as wsm
+    real = wsm.ws_manager.send_to_session
+
+    async def spy(sid, event, payload):
+        sent.append(event)
+
+    wsm.ws_manager.send_to_session = spy
+    try:
+        got = asyncio.run(agent_manager.resume_session(s.id))
+    finally:
+        wsm.ws_manager.send_to_session = real
+    assert got.id == s.id, "the read still returns the record"
+    assert s.id not in agent_manager.sessions, "but does not put it back in memory"
+    assert "agent:status" not in sent, "and does not repaint the card"
+
+
+def test_reading_a_session_the_user_did_not_close_still_resumes_it():
+    """NEGATIVE CONTROL for door 3: history browsing must keep working."""
+    s = p_live()
+    s.status = "completed"
+    asyncio.run(agent_manager.close_session(s.id))      # internal close, not the user route
+    assert s.id not in agent_manager.sessions
+    got = asyncio.run(agent_manager.resume_session(s.id))
+    assert got.id == s.id
+    assert s.id in agent_manager.sessions
+
+
+# --- DOOR 4: a child still BOOTING when its parent is closed must not outlive it ------------------
+#
+# Live, 2026-08-20: child de0ca12f was created 18:32:08, the parent was closed 18:32:12, and the child
+# kept driving Amazon on a closed card for minutes ("it came back for a bit"). A child has no
+# cancel_event during prestage, so close_session's stop missed it, then purged the parent; the old
+# entry check compared `parent.status` on a parent that was now None and never fired.
+
+def p_child_entry_check(parent_session_id):
+    """The exact predicate at the child's registration, lifted so it can be driven without a browser."""
+    from backend.apps.agents.agent_manager import agent_manager as am
+    if not parent_session_id:
+        return False
+    parent = am.get_session(parent_session_id)
+    if parent is not None:
+        return parent.status == "stopped" or bool(getattr(parent, "ended_by_user", False))
+    from backend.apps.agents.manager.session.session_store import load_session_data
+    rec = load_session_data(parent_session_id) or {}
+    return bool(rec.get("ended_by_user")) or rec.get("status") == "stopped"
+
+
+def test_a_child_registering_after_its_parent_was_purged_bails():
+    parent = p_live("closed-parent")
+    asyncio.run(user_close_route(parent.id))
+    assert parent.id not in agent_manager.sessions, "precondition: the close purged the parent"
+    assert p_child_entry_check(parent.id) is True, "purged from memory but persisted as user-ended = bail"
+
+
+def test_a_child_registering_under_a_user_stopped_parent_bails():
+    parent = p_live("stopped-parent")
+    asyncio.run(user_stop_route(parent.id))
+    assert p_child_entry_check(parent.id) is True
+
+
+def test_a_child_under_a_live_parent_proceeds():
+    """NEGATIVE CONTROL: normal delegation must be untouched."""
+    parent = p_live("live-parent")
+    assert p_child_entry_check(parent.id) is False
+
+
+def test_a_standalone_browser_run_with_no_parent_proceeds():
+    """NEGATIVE CONTROL: a run that never had a parent is not an orphan; it must not self-cancel."""
+    assert p_child_entry_check(None) is False
+    assert p_child_entry_check("") is False

backend/tests/test_ws_efficiency.py

@@ -131,7 +131,7 @@ def test_history_closed_only_filters_open_sessions(monkeypatch):
         ("open1", {"id": "open1", "name": "open chat", "closed_at": None, "dashboard_id": None}),
         ("closed1", {"id": "closed1", "name": "closed chat", "closed_at": "2026-07-01T00:00:00", "dashboard_id": None}),
     ]
-    import backend.apps.agents.manager.session.SessionLifecycle as lifecycle_mod
+    import backend.apps.agents.manager.session.SessionHistory as lifecycle_mod
     monkeypatch.setattr(lifecycle_mod, "load_all_session_data", lambda: list(rows))
     # get_history also merges LIVE sessions, so pin them empty or a leftover from another test leaks in.
     monkeypatch.setattr(agent_manager, "sessions", {})
@@ -145,7 +145,7 @@ def test_history_closed_only_filters_open_sessions(monkeypatch):
 def test_history_includes_live_sessions_missing_from_disk(monkeypatch):
     """Boot deletes the file of every still-open session, so history must merge memory or your current chats vanish."""
     from backend.apps.agents.core.models import AgentSession
-    import backend.apps.agents.manager.session.SessionLifecycle as lifecycle_mod
+    import backend.apps.agents.manager.session.SessionHistory as lifecycle_mod
 
     rows = [("closed1", {"id": "closed1", "name": "closed chat", "closed_at": "2026-07-01T00:00:00", "dashboard_id": None})]
     monkeypatch.setattr(lifecycle_mod, "load_all_session_data", lambda: list(rows))