From 596c0479ca02ea6d591d52eadb8f0460c9dfd23a Mon Sep 17 00:00:00 2001 From: ciregenz Date: Mon, 24 Aug 2026 10:16:02 -0700 Subject: [PATCH] [eric] agents: a ceiling on harness-started turns, so a runaway loop cannot spend an account (ENG-398) Measured across 79 installs: median 1/min, 76 of 79 under 10/min, and one install at 186/min that produced 44 of 45 policy blocks. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_018foyDoK19jjbYdudfzQVkZ --- backend/apps/agents/agent_manager.py | 14 ++ .../apps/agents/manager/machine_turn_gate.py | 103 ++++++++++ backend/tests/test_machine_turn_gate.py | 181 ++++++++++++++++++ 3 files changed, 298 insertions(+) create mode 100644 backend/apps/agents/manager/machine_turn_gate.py create mode 100644 backend/tests/test_machine_turn_gate.py diff --git a/backend/apps/agents/agent_manager.py b/backend/apps/agents/agent_manager.py index 7f101ba8..39f4d00c 100644 --- a/backend/apps/agents/agent_manager.py +++ b/backend/apps/agents/agent_manager.py @@ -66,6 +66,20 @@ class AgentManager(SessionLifecycle, SessionHistory, SessionPersistence, Messagi if p_tail: logger.info(f"continuation for {session_id} superseded by a user message during the {delay_s}s wait") return + # The ceiling on harness-started turns. Sits HERE because both dispatch sites route through + # this one function, so a future third caller inherits it instead of being a new hole. + from backend.apps.agents.manager.machine_turn_gate import wait_for_machine_turn_slot + p_gate_before = len(getattr(self.sessions.get(session_id), "messages", []) or []) + await wait_for_machine_turn_slot(session_id, "continuation") + # A hold can run to a minute, so it gets the SAME no-stomp guard as the delay above: a user + # message during the wait already resumes the work, and firing anyway would talk over them. + p_now = self.sessions.get(session_id) + if p_now is None: + return + if [m for m in p_now.messages[p_gate_before:] if getattr(m, "role", "") == "user"]: + 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 try: await self.send_message(session_id, prompt, hidden=True) except Exception: diff --git a/backend/apps/agents/manager/machine_turn_gate.py b/backend/apps/agents/manager/machine_turn_gate.py new file mode 100644 index 00000000..6d94b4bd --- /dev/null +++ b/backend/apps/agents/manager/machine_turn_gate.py @@ -0,0 +1,103 @@ +"""A ceiling on how fast the HARNESS may start turns on the user's account. + +Measured across 79 installs over 14 days: the median install starts **1** machine-initiated turn +per minute and 76 of 79 never pass 10. Three outliers sit at 16, 48 and **186 per minute**, and the +186 install is the same one that produced 44 of the 45 policy blocks in that window. Association, +not proof, but an unbounded self-heal loop spending someone's subscription is a bug at any rate. + +Why a TOKEN BUCKET and not a semaphore: a bucket can only ever DELAY, so it cannot deadlock. A +parent waiting on a child while holding a slot is the classic way a gate like this eats an app, and +that failure is unrepresentable here because nothing is ever held. + +What it deliberately does NOT cover: a human pressing send (never delayed, at any rate), and the +provider requests the CLI makes inside a turn (we do not send those; only 9router sees them). This +bounds the one thing we actually control, which is how often we start a turn nobody asked for. +""" + +import asyncio +import logging +import time +from typing import Dict, List + +from typeguard import typechecked + +logger = logging.getLogger(__name__) + +# Sits in the gap between the busiest legitimate install (9/min) and the pathological ones (16, 48, +# 186). Raising it needs a new measurement, not a hunch: a ceiling nothing reaches is not a ceiling. +MACHINE_TURNS_PER_MINUTE = 20 +WINDOW_S = 60.0 +# Waking in slices rather than one long sleep: the hold stays responsive to a window that rolls +# early, and a caller checking for a user message afterwards is never more than a slice stale. +SLICE_S = 2.0 + +p_starts: List[float] = [] +p_stats: Dict[str, float] = {"admitted": 0, "delayed": 0, "delayed_s": 0.0} + + +@typechecked +def wait_needed(now: float) -> float: + """Seconds until a slot frees, 0 when one is free. Pure, so the ceiling is testable without sleeping.""" + while p_starts and now - p_starts[0] > WINDOW_S: + p_starts.pop(0) + if len(p_starts) < MACHINE_TURNS_PER_MINUTE: + return 0.0 + return max(0.0, WINDOW_S - (now - p_starts[0])) + + +@typechecked +async def wait_for_machine_turn_slot(session_id: str, reason: str) -> None: + """Hold a machine-started turn until the account has room. Never rejects, never drops work.""" + waited = 0.0 + said = False + while True: + delay = wait_needed(time.monotonic()) + if delay <= 0: + break + waited += min(delay, SLICE_S) + # Says which session it is holding and why: a guard that throttles in silence reads to the + # next person as "the app randomly got slow". Once per hold, not once per slice. + if not said: + said = True + logger.warning( + f"machine-turn ceiling reached ({MACHINE_TURNS_PER_MINUTE}/min): holding the {reason} " + f"for {session_id} up to {delay:.1f}s. A human send is never held." + ) + await asyncio.sleep(min(delay, SLICE_S)) + p_starts.append(time.monotonic()) + p_stats["admitted"] += 1 + if waited > 0: + p_stats["delayed"] += 1 + p_stats["delayed_s"] += waited + + +@typechecked +def gate_report() -> str: + """What the ceiling actually did, so 'it never fires' is a fact rather than an assumption.""" + return (f"machine-turn gate: {int(p_stats['admitted'])} starts admitted, " + f"{int(p_stats['delayed'])} held, {p_stats['delayed_s']:.1f}s total") + + +@typechecked +def admitted_count() -> int: + return int(p_stats["admitted"]) + + +@typechecked +def reset_for_test() -> None: + """Forget the window. Named for what it is, so nobody calls it from production by accident.""" + p_starts.clear() + p_stats.update({"admitted": 0, "delayed": 0, "delayed_s": 0.0}) + + +@typechecked +def note_start_for_test(n: int) -> None: + """Pretend n machine-started turns just happened, so a ceiling test needs no real turns.""" + now = time.monotonic() + p_starts.extend([now] * n) + + +@typechecked +def roll_window_for_test(seconds: float) -> None: + """Age every recorded start, so a test can reach the far side of the window without sleeping it.""" + p_starts[:] = [t - seconds for t in p_starts] diff --git a/backend/tests/test_machine_turn_gate.py b/backend/tests/test_machine_turn_gate.py new file mode 100644 index 00000000..576b9082 --- /dev/null +++ b/backend/tests/test_machine_turn_gate.py @@ -0,0 +1,181 @@ +"""A ceiling on harness-started turns, and proof it can only ever delay. + +Measured 2026-08-24 across 79 installs: the median install starts 1 machine-initiated turn per +minute, 76 of 79 never pass 10, and three outliers sit at 16, 48 and 186. The 186 install produced +44 of the 45 policy blocks that window. These pin that the ceiling clips the pathological rate, +leaves every measured real install alone, and can neither deadlock nor drop work. +""" + +import asyncio +import time + +import pytest + +from backend.tests.log_capture import LogCapture +from backend.apps.agents.manager import machine_turn_gate as gate + + +@pytest.fixture(autouse=True) +def p_clean(): + gate.reset_for_test() + yield + gate.reset_for_test() + gate.SLICE_S = 2.0 + + +def test_the_ceiling_leaves_every_measured_real_install_alone(): + # Fleet p90 is 6/min and the busiest legitimate install peaks at 9. + gate.note_start_for_test(9) + assert gate.wait_needed(time.monotonic()) == 0.0, "a real install must never be held" + + +def test_the_ceiling_clips_the_runaway(): + gate.note_start_for_test(186) + assert gate.wait_needed(time.monotonic()) > 0 + + +def test_the_window_rolls_so_a_hold_is_never_permanent(): + gate.note_start_for_test(gate.MACHINE_TURNS_PER_MINUTE) + assert gate.wait_needed(time.monotonic()) > 0 + gate.roll_window_for_test(gate.WINDOW_S + 1) + assert gate.wait_needed(time.monotonic()) == 0.0, \ + "a ceiling that never reopens is an outage, not a guard" + + +def test_it_delays_and_never_rejects(): + # Why this is a token bucket and not a semaphore: nothing is ever held, so nothing can deadlock, + # and the caller always eventually proceeds with the work intact. + async def main(): + for _ in range(gate.MACHINE_TURNS_PER_MINUTE): + await gate.wait_for_machine_turn_slot("s1", "continuation") + gate.roll_window_for_test(gate.WINDOW_S + 1) + await asyncio.wait_for(gate.wait_for_machine_turn_slot("s1", "continuation"), timeout=2) + asyncio.run(main()) + assert gate.admitted_count() == gate.MACHINE_TURNS_PER_MINUTE + 1 + + +def test_a_hold_wakes_when_the_window_rolls_rather_than_sleeping_it_out(): + # One long sleep would sit out the full minute even after room appeared. Slices keep it honest. + async def main(): + gate.note_start_for_test(gate.MACHINE_TURNS_PER_MINUTE) + gate.SLICE_S = 0.05 + + async def p_release(): + await asyncio.sleep(0.1) + gate.roll_window_for_test(gate.WINDOW_S + 1) + + asyncio.create_task(p_release()) + t0 = time.monotonic() + await asyncio.wait_for(gate.wait_for_machine_turn_slot("s1", "continuation"), timeout=5) + return time.monotonic() - t0 + + took = asyncio.run(main()) + assert took < 2.0, f"held {took:.1f}s after the window rolled; it slept the full delay" + + +def test_a_hold_names_the_session_and_says_a_human_is_never_held(): + async def main(): + gate.note_start_for_test(gate.MACHINE_TURNS_PER_MINUTE) + gate.SLICE_S = 0.05 + + async def p_release(): + await asyncio.sleep(0.1) + gate.roll_window_for_test(gate.WINDOW_S + 1) + + asyncio.create_task(p_release()) + with LogCapture("backend.apps.agents.manager.machine_turn_gate") as cap: + await asyncio.wait_for(gate.wait_for_machine_turn_slot("sess-abc", "continuation"), timeout=5) + return cap.text + + text = asyncio.run(main()) + assert "sess-abc" in text and "human send is never held" in text + + +def test_the_gate_sits_at_the_one_chokepoint_both_callers_use(): + src = open("backend/apps/agents/agent_manager.py").read() + assert src.count("dispatch_hidden_continuation(") >= 3, "both dispatch sites plus the def" + i_gate = src.index("wait_for_machine_turn_slot(session_id") + i_send = src.index("await self.send_message(session_id, prompt, hidden=True)") + assert i_gate < i_send, "the ceiling has to be reached before the send, not after" + + +def test_a_human_send_never_reaches_the_gate(): + 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_user_message_during_a_hold_is_never_talked_over(): + src = open("backend/apps/agents/agent_manager.py").read() + i_gate = src.index("await wait_for_machine_turn_slot(session_id") + after = src[i_gate:i_gate + 900] + assert "superseded by a user message while held" in after, \ + "a minute-long hold needs the same no-stomp guard as the delay path" + assert "p_settle_unstarted_continuation" in after, \ + "standing down must release the running promise, or the card spins forever" + + +def test_the_report_makes_a_dead_ceiling_visible(): + assert "0 starts admitted" in gate.gate_report() + + +def test_the_real_dispatcher_holds_the_runaway_and_loses_no_work(monkeypatch): + """Drive the ACTUAL dispatch path: a gate can be perfect and wired to nothing.""" + from backend.apps.agents.agent_manager import agent_manager + from backend.apps.agents.core.models import AgentSession + + session = AgentSession(name="gate", model="sonnet", dashboard_id="d") + agent_manager.sessions[session.id] = session + sent = [] + + async def p_fake_send(sid, prompt, hidden=False, **kw): + sent.append(prompt) + + monkeypatch.setattr(agent_manager, "send_message", p_fake_send, raising=True) + gate.SLICE_S = 0.05 + + async def main(): + for _ in range(gate.MACHINE_TURNS_PER_MINUTE): + await agent_manager.dispatch_hidden_continuation(session.id, "go", 0) + admitted_fast = len(sent) + task = asyncio.create_task(agent_manager.dispatch_hidden_continuation(session.id, "held", 0)) + await asyncio.sleep(0.15) + held = len(sent) == admitted_fast + gate.roll_window_for_test(gate.WINDOW_S + 1) + await asyncio.wait_for(task, timeout=5) + return admitted_fast, held, sent[-1] + + admitted_fast, held, last = asyncio.run(main()) + agent_manager.sessions.pop(session.id, None) + assert admitted_fast == gate.MACHINE_TURNS_PER_MINUTE, "normal traffic must pass straight through" + assert held, "the 21st start in a minute must be held" + assert last == "held", "a held continuation must still run; delaying is not dropping" + + +def test_a_user_message_during_a_real_hold_stands_the_continuation_down(monkeypatch): + from backend.apps.agents.agent_manager import agent_manager + from backend.apps.agents.core.models import AgentSession, Message + + session = AgentSession(name="gate2", model="sonnet", dashboard_id="d") + agent_manager.sessions[session.id] = session + sent = [] + + async def p_fake_send(sid, prompt, hidden=False, **kw): + sent.append(prompt) + + monkeypatch.setattr(agent_manager, "send_message", p_fake_send, raising=True) + gate.SLICE_S = 0.05 + + async def main(): + gate.note_start_for_test(gate.MACHINE_TURNS_PER_MINUTE) + task = asyncio.create_task(agent_manager.dispatch_hidden_continuation(session.id, "stale", 0)) + await asyncio.sleep(0.1) + session.messages.append(Message(role="user", content="actually do this instead")) + gate.roll_window_for_test(gate.WINDOW_S + 1) + await asyncio.wait_for(task, timeout=5) + + asyncio.run(main()) + status = agent_manager.sessions[session.id].status + agent_manager.sessions.pop(session.id, None) + assert sent == [], "the user already resumed the work; a held continuation must not talk over them" + assert status != "running", "standing down must release the running promise, or the card spins forever"