diff --git a/backend/apps/agents/agent_manager.py b/backend/apps/agents/agent_manager.py index 5e97b0b8..e11855e1 100644 --- a/backend/apps/agents/agent_manager.py +++ b/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 diff --git a/backend/apps/agents/agents.py b/backend/apps/agents/agents.py index 95f0e012..597e0a93 100644 --- a/backend/apps/agents/agents.py +++ b/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: diff --git a/backend/apps/agents/browser/browser_agent.py b/backend/apps/agents/browser/browser_agent.py index 712475c5..489f7149 100644 --- a/backend/apps/agents/browser/browser_agent.py +++ b/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", { diff --git a/backend/apps/agents/core/models.py b/backend/apps/agents/core/models.py index a4c390a0..236a9e6a 100644 --- a/backend/apps/agents/core/models.py +++ b/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 = "" diff --git a/backend/apps/agents/manager/EditMessage.py b/backend/apps/agents/manager/EditMessage.py new file mode 100644 index 00000000..30f8185c --- /dev/null +++ b/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) diff --git a/backend/apps/agents/manager/Messaging.py b/backend/apps/agents/manager/Messaging.py index 30b60e7d..e90d955d 100644 --- a/backend/apps/agents/manager/Messaging.py +++ b/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) - diff --git a/backend/apps/agents/manager/session/SessionHistory.py b/backend/apps/agents/manager/session/SessionHistory.py new file mode 100644 index 00000000..b607eec8 --- /dev/null +++ b/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, + } diff --git a/backend/apps/agents/manager/session/SessionLifecycle.py b/backend/apps/agents/manager/session/SessionLifecycle.py index fe1897a2..e117c970 100644 --- a/backend/apps/agents/manager/session/SessionLifecycle.py +++ b/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) diff --git a/backend/apps/agents/manager/streaming/unwedge_sidecar.py b/backend/apps/agents/manager/streaming/unwedge_sidecar.py index ff5ff245..c6f3aa74 100644 --- a/backend/apps/agents/manager/streaming/unwedge_sidecar.py +++ b/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: diff --git a/backend/tests/test_browser_agent_loop.py b/backend/tests/test_browser_agent_loop.py index d60b200f..96513b10 100644 --- a/backend/tests/test_browser_agent_loop.py +++ b/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([ diff --git a/backend/tests/test_delegation_result_backstop.py b/backend/tests/test_delegation_result_backstop.py index f8c0fcf2..ff7bcb4c 100644 --- a/backend/tests/test_delegation_result_backstop.py +++ b/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}) diff --git a/backend/tests/test_session_snapshot_on_send.py b/backend/tests/test_session_snapshot_on_send.py index b282313d..460838f4 100644 --- a/backend/tests/test_session_snapshot_on_send.py +++ b/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(') diff --git a/backend/tests/test_user_stop_is_final.py b/backend/tests/test_user_stop_is_final.py new file mode 100644 index 00000000..73f5d052 --- /dev/null +++ b/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 diff --git a/backend/tests/test_ws_efficiency.py b/backend/tests/test_ws_efficiency.py index 31b09a16..d6795180 100644 --- a/backend/tests/test_ws_efficiency.py +++ b/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))