From fb92d785b4e72df5dc47a4a8c9e67a5a55a70aa6 Mon Sep 17 00:00:00 2001 From: ciregenz Date: Mon, 24 Aug 2026 22:59:07 -0700 Subject: [PATCH] [eric] agents: a human's Stop outranks every recovery path, checked before any of them (ENG-402, ENG-369) The liveness door resurrected stopped sessions; a stopped browser agent looks exactly like a dead sidecar. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_018foyDoK19jjbYdudfzQVkZ --- backend/apps/agents/agent_manager.py | 8 + .../manager/streaming/delegation_watchdog.py | 16 ++ backend/tests/test_frozen_sidecar_bound.py | 54 ++++++ backend/tests/test_overcorrection_audit.py | 156 ++++++++++++++++++ 4 files changed, 234 insertions(+) create mode 100644 backend/tests/test_overcorrection_audit.py diff --git a/backend/apps/agents/agent_manager.py b/backend/apps/agents/agent_manager.py index 39f4d00c..f6e90502 100644 --- a/backend/apps/agents/agent_manager.py +++ b/backend/apps/agents/agent_manager.py @@ -80,6 +80,14 @@ class AgentManager(SessionLifecycle, SessionHistory, SessionPersistence, Messagi logger.info(f"continuation for {session_id} superseded by a user message while held at the machine-turn ceiling") await self.p_settle_unstarted_continuation(session_id) return + # Messaging refuses a machine send to a session a human ended, but it refuses SILENTLY, and + # arming already promised "running" (ENG-390). Without this the promise is never released + # and the card spins forever on a chat the user stopped. Checked here so the reliance on the + # chokepoint is explicit rather than a coincidence two files apart. + if getattr(p_now, "ended_by_user", False): + logger.info(f"continuation for {session_id} stood down: a human ended this chat") + await self.p_settle_unstarted_continuation(session_id) + return try: await self.send_message(session_id, prompt, hidden=True) except Exception: diff --git a/backend/apps/agents/manager/streaming/delegation_watchdog.py b/backend/apps/agents/manager/streaming/delegation_watchdog.py index b40a58ee..6797469f 100644 --- a/backend/apps/agents/manager/streaming/delegation_watchdog.py +++ b/backend/apps/agents/manager/streaming/delegation_watchdog.py @@ -87,6 +87,14 @@ def no_child_ever_born(session_id: str, since: float) -> bool: return True +@typechecked +def p_ended_by_user(session_id: str) -> bool: + """True when a person ended this session. Nothing automatic may act on it after that.""" + from backend.apps.agents.agent_manager import agent_manager + p_s = agent_manager.sessions.get(session_id) + return p_s is not None and bool(getattr(p_s, "ended_by_user", False)) + + @typechecked def sidecar_is_dead(session_id: str) -> bool: """A stale heartbeat means the PROCESS stopped, which no amount of patience fixes. @@ -120,6 +128,14 @@ def arm_delegation_watchdog(ctx: object, tool_use_id: str, tool_name: str) -> No session_id = getattr(ctx, "session_id", "") if not session_id: return + # A HUMAN'S STOP OUTRANKS EVERY RECOVERY PATH, and it is checked HERE rather than inside one + # of the predicates below. It used to live only inside delegation_children_settled, where a + # False meant "do not recover"; the liveness door then overrode that False to True and + # resurrected the very session the user had just stopped, because a stopped browser agent + # looks exactly like a dead sidecar (no child, no heartbeat). That is ENG-369 coming back + # through a side door. At the top, no future predicate can reintroduce it (ENG-402). + if p_ended_by_user(session_id): + return try: settled = delegation_children_settled(session_id, started) except Exception: diff --git a/backend/tests/test_frozen_sidecar_bound.py b/backend/tests/test_frozen_sidecar_bound.py index 1113672d..dcd2173a 100644 --- a/backend/tests/test_frozen_sidecar_bound.py +++ b/backend/tests/test_frozen_sidecar_bound.py @@ -88,3 +88,57 @@ def test_a_child_from_an_EARLIER_delegation_does_not_count(monkeypatch): assert uw.no_child_ever_born("par-old", time.time()) is True finally: am.agent_manager.sessions.pop("kid-old", None) + + +def test_a_stopped_agent_is_never_resurrected_by_the_liveness_door(monkeypatch): + """The regression this fix nearly shipped, caught by Eric asking the right question. + + A browser agent the user STOPS looks identical to a frozen sidecar: no child was born and the + heartbeat is stale. The human-Stop guard used to live only inside `delegation_children_settled`, + where it returns False meaning "do not recover" -- and the liveness door then overrode that + False to True. That is ENG-369 ("stage 3 resent RETRY_PROMPT into the session the user had just + stopped, every ~150s") coming back through a side door. + """ + import asyncio + from backend.apps.agents.core.models import AgentSession + from backend.apps.agents import agent_manager as am + from backend.apps.agents.manager.streaming import delegation_watchdog as dw + + fired = {} + monkeypatch.setattr(dw, "unwedge", lambda *a: fired.setdefault("unwedge", a), raising=True) + monkeypatch.setattr(dw, "arm_retry", lambda s: fired.setdefault("retry", True), raising=True) + monkeypatch.setattr(dw, "DELEGATION_CHECK_SECONDS", 0.05, raising=True) + # The frozen-sidecar condition: dead heartbeat, and no child ever born. + monkeypatch.setattr(dw, "heartbeat_age", lambda sid: dw.HEARTBEAT_FRESH_S + 99, raising=True) + + sess = AgentSession(id="par-stopped", name="p", model="sonnet") + sess.ended_by_user = True + am.agent_manager.sessions["par-stopped"] = sess + + class P_Ctx: + def __init__(self): + self.session = sess + self.session_id = "par-stopped" + self.tool_start_times = {"tu-1": 0.0} + + async def main(): + dw.arm_delegation_watchdog(P_Ctx(), "tu-1", "mcp__openswarm-core__CreateBrowserAgent") + await asyncio.sleep(0.4) + + try: + asyncio.run(main()) + finally: + am.agent_manager.sessions.pop("par-stopped", None) + + assert "unwedge" not in fired, "a session the user stopped must never be unwedged back to life" + assert "retry" not in fired, "and must never be handed a retry prompt" + + +def test_the_human_stop_check_runs_BEFORE_any_settled_predicate(): + # Placement is the fix. Inside a predicate it can be overridden; at the top it cannot. + src = open("backend/apps/agents/manager/streaming/delegation_watchdog.py").read() + i_stop = src.index("if p_ended_by_user(session_id):") + i_settled = src.index("settled = delegation_children_settled(session_id, started)") + i_door = src.index("sidecar_is_dead(session_id)", i_settled) + assert i_stop < i_settled < i_door, \ + "a human's Stop has to short-circuit before anything can set settled" diff --git a/backend/tests/test_overcorrection_audit.py b/backend/tests/test_overcorrection_audit.py new file mode 100644 index 00000000..71f58345 --- /dev/null +++ b/backend/tests/test_overcorrection_audit.py @@ -0,0 +1,156 @@ +"""Every guard added since 1.7.9, checked against the LEGITIMATE case that resembles the bad one. + +A fix that fires beyond its intent is a trade, not a win. One of these caught a real regression the +day it was written: the frozen-sidecar liveness door resurrected sessions the user had stopped, +because a stopped browser agent looks exactly like a dead sidecar. These pin the innocent case for +each guard, so the next over-broad edit fails a test instead of shipping. +""" + +import asyncio + +from backend.apps.agents.core.models import AgentSession + + +# --------------------------------------------------------------- the machine-turn ceiling (ENG-398) + +def test_a_session_stopped_during_a_gate_hold_never_gets_the_continuation(monkeypatch): + """The gate can hold a continuation for up to a minute. If the user stops the chat in that + window, the held send must not land. The gate itself does not check this; it relies on the + Messaging chokepoint, so the reliance is what gets pinned.""" + from backend.apps.agents.agent_manager import agent_manager + from backend.apps.agents.manager import machine_turn_gate as gate + + session = AgentSession(name="stopped-mid-hold", model="sonnet", dashboard_id="d") + session.ended_by_user = True + agent_manager.sessions[session.id] = session + landed = [] + + async def p_capture(sid, prompt, hidden=False, by_user=False, **kw): + # Mirrors Messaging.py's guard: hidden + not by_user + ended_by_user -> refused. + p_s = agent_manager.sessions.get(sid) + if hidden and not by_user and getattr(p_s, "ended_by_user", False): + return + landed.append(prompt) + + monkeypatch.setattr(agent_manager, "send_message", p_capture, raising=True) + gate.reset_for_test() + try: + asyncio.run(agent_manager.dispatch_hidden_continuation(session.id, "carry on", 0)) + finally: + agent_manager.sessions.pop(session.id, None) + gate.reset_for_test() + assert landed == [], "a chat the user stopped must not be continued after a gate hold" + + +def test_the_gate_never_holds_a_human_send(): + # The capability this must not cost: a person pressing send is never delayed, at any rate. + src = open("backend/apps/agents/agent_manager.py").read() + head = src[:src.index("async def dispatch_hidden_continuation")] + assert "wait_for_machine_turn_slot" not in head + + +def test_a_delayed_continuation_still_reaches_the_gate_in_the_right_order(monkeypatch): + """Codex rotation waits 75s so the retry lands AFTER the token rotates. The ceiling must come + after that wait, not replace it, or the retry burns its one shot inside the rotation window.""" + src = open("backend/apps/agents/agent_manager.py").read() + body = src[src.index("async def dispatch_hidden_continuation"):] + i_delay = body.index("if delay_s > 0:") + i_gate = body.index("await wait_for_machine_turn_slot(session_id") + assert i_delay < i_gate, "the rotation wait has to run before the ceiling, never instead of it" + + +# ------------------------------------------------------------------ tool-output shaping (ENG-385) + +def test_shaping_never_removes_without_a_way_back(): + """The guarantee is not the line-matching heuristic, it is recoverability. Any body that gets + cut must name where the full text lives, or a wrong guess costs the answer instead of a re-read.""" + from backend.apps.agents.manager.streaming.tool_output_shaper import shape_text + out = shape_text("q" * 9_000, "/blobs/x-model.txt") + assert "/blobs/x-model.txt" in out + + +def test_shaping_leaves_a_normal_result_completely_alone(): + from backend.apps.agents.manager.streaming.tool_output_shaper import shape_tool_response + for benign in ("ok", {"stdout": "3 passed", "stderr": ""}, [{"type": "text", "text": "done"}]): + assert shape_tool_response(benign, "/b.txt")[0] is None + + +# --------------------------------------------------------- permanent-vs-transient classify (ENG-395) + +def test_a_real_throttle_is_still_retried(): + """The fix stops a malformed request being retried forever. It must not stop a genuine 429, + an expiring token, or a traceback line number from behaving as before.""" + from backend.apps.agents.core.error_classify import is_transient_capacity_error as t + assert t(RuntimeError('API Error: 429 {"message":"rate_limit_error (reset after 21s)"}')) is True + assert t(RuntimeError("API Error: 401 unauthorized (reset after 1m 57s)")) is True + assert t(RuntimeError("File runner.py, line 400, in execute (reset after 3s)")) is True + assert t(RuntimeError("overloaded_error")) is True + + +# ------------------------------------------------------------------- no transcript replay (ENG-396) + +def test_the_users_own_words_and_the_result_both_survive(): + """Stripping model prose once went too far and made InvokeWorkflow return a trail with no + answer in it. The user's ask is not model output, and one final result is not a replay.""" + from backend.apps.agents.manager.session.history_compaction import render_agent_trail + + class M: + def __init__(s, role, content): + s.role, s.content, s.id, s.hidden = role, content, "m", False + + out = render_agent_trail([ + M("user", "find the bug"), + M("tool_call", {"tool": "Bash", "input": {"command": "pytest -q"}}), + M("tool_result", {"tool_name": "Bash", "text": "3 failed"}), + M("assistant", "the parser is at fault"), + ]) + assert "find the bug" in out + assert "pytest -q" in out and "3 failed" in out + assert "the parser is at fault" in out, "the outcome is why the caller invoked the run" + + +def test_the_real_messaging_guard_is_what_the_gate_relies_on(): + """Wire check, not a mirror. The gate test above simulates the refusal; this asserts the actual + chokepoint carries it at BOTH doors (the in-memory path and the disk-reload path), because a + late watchdog retry reopening a closed card is exactly how ENG-369/384 happened.""" + src = open("backend/apps/agents/manager/Messaging.py").read() + guard = "if hidden and not by_user and session.ended_by_user:" + assert src.count(guard) == 2, \ + "both the reload path and the live path must refuse a machine send to a stopped session" + # And a human's own click must still get through, or the Resume chip reappears forever. + assert "if session.ended_by_user and (not hidden or by_user):" in src + + +def test_a_stopped_session_is_not_left_spinning_after_a_refused_continuation(): + """The other half: a refused send returns normally rather than raising, so the settle path is + never reached. That is only safe because a human Stop already wrote a terminal status.""" + src = open("backend/apps/agents/agent_manager.py").read() + i = src.index("async def p_settle_unstarted_continuation") + assert 'status != "running"' in src[i:i + 600], \ + "settling must no-op on an already-terminal session rather than rewriting it" + + +def test_a_continuation_armed_for_a_stopped_chat_releases_its_running_promise(monkeypatch): + """The race between ENG-390 and ENG-384: arming a continuation promises `running`, Messaging + refuses the send SILENTLY for a stopped chat, and a silent refusal never reaches the settle + path. Left alone the card spins forever on a chat the user ended.""" + from backend.apps.agents.agent_manager import agent_manager + from backend.apps.agents.manager import machine_turn_gate as gate + + session = AgentSession(name="raced", model="sonnet", dashboard_id="d") + session.ended_by_user = True + session.status = "running" + agent_manager.sessions[session.id] = session + sent = [] + monkeypatch.setattr(agent_manager, "send_message", + lambda *a, **k: sent.append(a), raising=True) + gate.reset_for_test() + try: + asyncio.run(agent_manager.dispatch_hidden_continuation(session.id, "go", 0)) + status = agent_manager.sessions[session.id].status + finally: + agent_manager.sessions.pop(session.id, None) + gate.reset_for_test() + + assert sent == [], "no machine send into a chat a human ended" + assert status != "running", "and the running promise must be released, or the card spins forever"