mirror of
https://github.com/openswarm-ai/openswarm.git
synced 2026-09-12 04:37:44 +02:00
[eric] agents: a finished turn always leaves the user something to read
Every silent-stop fix so far was a detector for a shape somebody had already found in the field, which is why the class kept coming back wearing a new hat. turn_spoke.py moves the question down a tier: at the one exit every terminal path passes through, ask whether anything readable appeared since the user last spoke, and if not, say the honest line. Cause no longer has to be enumerated for the user to be answered. proactive_prune.py is the hermes trigger we were missing. Their own tests say our bug out loud: on a large window, a percentage-of-window compaction check almost never fires, so aged tool output rides in history and is re-sent verbatim every turn. Measured here, our shaping cut 0.0% at every session size; with a fixed 60K-token cost trigger it cuts 88% at 12 turns, 93% at 30, 94% at 60. The prompt-cache contract is load-bearing rather than optional, because our prune is a rebuild: it commits only when it reclaims enough to pay for the busted prefix, then disarms until history has regrown a full runway. Also: lane preflight now treats only 401/403 as a dead credential, since testStatus=="unavailable" conflated a throttled lane with a revoked one and told users to reconnect a merely rate-limited Gemini; and awaiting_reconnect is cleared when the retry budget is spent, so a stale flag can no longer muzzle the floor and end an ask in total silence. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_014wtspwSFzZmjCx9UNPAorQ
This commit is contained in:
ciregenz
co-authored by
Claude Opus 5
Claude Opus 5
parent
9a7570092f
commit
0f2543db65
@@ -378,6 +378,16 @@ class AgentManager(SessionLifecycle, SessionPersistence, Messaging, SessionContr
|
||||
p_is_live_task = self.tasks.get(session_id) is asyncio.current_task()
|
||||
if p_is_live_task:
|
||||
self.live_partial.pop(session_id, None)
|
||||
# The floor: every terminal path funnels through here, so "turn ended with nothing readable" stops being representable instead of being caught shape by shape.
|
||||
try:
|
||||
from backend.apps.agents.manager.run.turn_spoke import ensure_turn_spoke
|
||||
if ensure_turn_spoke(session, session_id):
|
||||
await ws_manager.send_to_session(session_id, "agent:message", {
|
||||
"session_id": session_id,
|
||||
"message": session.messages[-1].model_dump(mode="json"),
|
||||
})
|
||||
except Exception:
|
||||
logger.exception("terminal turn-spoke invariant failed")
|
||||
if session_id in self.sessions and p_is_live_task:
|
||||
# For canvas-launched App Builder sessions, the workspace folder IS the session_id (see launch_agent), so meta.json lives at outputs_workspace/<session_id>/meta.json. Read it and propagate name/description into the Output row before the terminal status fires; without this, the row stays "Untitled App" forever because no React component polls the file on the canvas path. Best-effort, only acts when the row's name is still the default placeholder.
|
||||
if session.mode == "view-builder":
|
||||
|
||||
@@ -154,6 +154,8 @@ class AgentSession(BaseModel):
|
||||
crash_interrupt_count: int = 0
|
||||
# Outage rounds spent on this ask: the in-turn ladder covers only 335s, and the work is checkpointed, so a longer drop is waited out rather than ending the task.
|
||||
# True when the preflight found the router had already given up on this lane, so an auth failure this turn is a dead credential, not a rotation window worth waiting out.
|
||||
# Input-token level history must regrow past before another proactive prune may commit; a rebuild busts the prompt cache, so one per runway, never one per turn.
|
||||
proactive_prune_rearm_tokens: int = 0
|
||||
lane_credential_dead: bool = False
|
||||
reconnect_attempts: int = 0
|
||||
# True while a turn is parked waiting for the connection back; persisted so a quit DURING the wait is still an owed turn at next boot.
|
||||
|
||||
@@ -189,6 +189,14 @@ async def handle_run_error(e: Exception, session: AgentSession, session_id: str,
|
||||
# 335s of ladder is a blip's worth of patience, and a closed lid or switched network outlasts it, so park and retry before conceding a turn the user never chose to end.
|
||||
from backend.apps.agents.manager.run.reconnect_resume import arm_reconnect_resume
|
||||
p_delay = arm_reconnect_resume(session, parse_retry_after(e, p_stderr_tail), is_connection_lost(e))
|
||||
if p_delay is None:
|
||||
# Budget spent: this turn is OVER, not parked. Leaving the flag set muzzles the
|
||||
# terminal floor (which stays quiet for parked turns) and the ask ends in silence,
|
||||
# which is the exact failure the floor exists to prevent. Caught live on a rate-limited
|
||||
# Gemini run, 2026-08-20: status=completed, awaiting_reconnect=True, attempts=3, and
|
||||
# not one word to the user.
|
||||
from backend.apps.agents.manager.run.reconnect_resume import clear_reconnect_wait
|
||||
clear_reconnect_wait(session)
|
||||
if p_delay is not None:
|
||||
logger.info(f"Agent {session_id}: connection lost past the in-turn budget; retrying in {p_delay}s")
|
||||
await ws_manager.send_to_session(session_id, "agent:reconnect_wait", {
|
||||
|
||||
@@ -70,9 +70,15 @@ def provider_for_model(resolved_model: str) -> Optional[str]:
|
||||
|
||||
@typechecked
|
||||
def connection_is_dead(conn: Dict) -> bool:
|
||||
"""A connection the router has given up on. Deliberately narrow: only the states that mean a dispatch is guaranteed to fail, never a slow or merely idle one."""
|
||||
if conn.get("testStatus") == "unavailable":
|
||||
return True
|
||||
"""A credential that needs the USER, as opposed to one having a bad minute.
|
||||
|
||||
Only auth-shaped failures qualify. `testStatus: "unavailable"` alone does NOT: the router
|
||||
stamps it for rate limits and upstream 5xx too, and a live 2026-08-20 run proved the cost of
|
||||
conflating them, telling Eric to reconnect a Google account whose credential was valid for
|
||||
another half hour and merely 429'd. Advising a reconnect for a throttle is the same lie as
|
||||
"just rotated" for a dead token, pointing the other way, so the bar here is evidence that
|
||||
waiting cannot help: 401 or 403.
|
||||
"""
|
||||
return conn.get("errorCode") in (401, 403)
|
||||
|
||||
|
||||
@@ -120,11 +126,18 @@ async def preflight_lane(resolved_model: str,
|
||||
f"lane preflight: {provider} is {dead.get('testStatus')} (errorCode={dead.get('errorCode')}); "
|
||||
"bouncing the router once, then letting the turn decide"
|
||||
)
|
||||
p_back_up = False
|
||||
try:
|
||||
from backend.apps.nine_router.bounce_after_connect import bounce_router_after_connect
|
||||
await bounce_router_after_connect(provider)
|
||||
p_back_up = await bounce_router_after_connect(provider)
|
||||
except Exception:
|
||||
logger.debug("lane preflight bounce failed", exc_info=True)
|
||||
if not p_back_up:
|
||||
# Dispatching into a router that has not come back is a guaranteed connection error, and
|
||||
# the user would read that as the model failing rather than us restarting something.
|
||||
logger.warning("lane preflight: the router did not come back after the bounce; not dispatching into it")
|
||||
return ("The local AI connection is restarting. This clears itself in a few seconds; "
|
||||
"send your message again.")
|
||||
# Deliberately no post-bounce health re-read: see the module docstring. Dispatch is the test.
|
||||
return None
|
||||
|
||||
|
||||
@@ -25,6 +25,30 @@ def merge_hard_blocked_tools(effective_disallowed: List[str]) -> List[str]:
|
||||
# `manager` is the AgentManager; it isn't annotated because typing it would import agent_manager back into a module agent_manager already imports (a cycle). Same reason self is never annotated.
|
||||
@typechecked
|
||||
async def pre_send_context_guard(manager, session: AgentSession, session_id: str) -> None:
|
||||
# The second trigger (hermes lift): the threshold below fires at a percentage of the window,
|
||||
# which on a 1M lane almost never arrives, so history rides untouched to the cliff. This one
|
||||
# fires on COST and pays for its own cache miss.
|
||||
try:
|
||||
from backend.apps.agents.manager.session.proactive_prune import (
|
||||
arm_proactive_prune,
|
||||
estimate_aged_rebuild_tokens,
|
||||
should_proactively_prune,
|
||||
)
|
||||
if should_proactively_prune(session):
|
||||
arm_proactive_prune(session)
|
||||
await ws_manager.send_to_session(session_id, "agent:context_status", {
|
||||
"session_id": session_id,
|
||||
"reason": "compacted",
|
||||
"compacted_through_msg_id": session.compacted_through_msg_id,
|
||||
})
|
||||
await manager.emit_context_update(
|
||||
session_id, session,
|
||||
input_tokens=estimate_aged_rebuild_tokens(session),
|
||||
output_tokens=session.tokens.get("output", 0),
|
||||
)
|
||||
except Exception:
|
||||
logger.exception("proactive prune failed; proceeding without it")
|
||||
|
||||
try:
|
||||
if manager.maybe_compact(session):
|
||||
# A mark alone never applies on the resume path (the CLI replays its own untrimmed transcript), so pay for the rebuild too: next turn drops the SDK convo and rebuilds with the cutoff + distilled summary. One respawn per compaction epoch is the price of never reaching the wall.
|
||||
|
||||
@@ -0,0 +1,91 @@
|
||||
"""One invariant at the single exit: a finished turn always left the user something to read.
|
||||
|
||||
Every silent-stop fix so far has been a DETECTOR: empty finish, vanishing quit, stalled nudge,
|
||||
produced-nothing, exhausted budget. Each one enumerates a shape somebody already found in the
|
||||
field, which means the next shape nobody has found yet still ships as silence. That is the wrong
|
||||
tier, and the ENG-354 history is the proof: the class kept coming back wearing a different hat.
|
||||
|
||||
So this does not classify anything. It asks the only question the user asks, at the one place every
|
||||
terminal path passes through:
|
||||
|
||||
since you last spoke to me, has anything appeared that I can read?
|
||||
|
||||
If not, the honest line goes in. It does not matter whether the cause was a silent quit, a shape
|
||||
nobody has named, or a bug written next year: the state "turn ended, nothing to read" stops being
|
||||
representable, rather than being caught case by case.
|
||||
|
||||
Deliberately last: the detectors upstream produce BETTER messages because they know why. This only
|
||||
fires when every one of them declined, so it is a floor, never a replacement.
|
||||
"""
|
||||
|
||||
import logging
|
||||
from typing import List
|
||||
|
||||
from typeguard import typechecked
|
||||
|
||||
from backend.apps.agents.core.models import AgentSession, Message
|
||||
from backend.apps.agents.manager.run.empty_finish import EXHAUSTED_NOTE, P_ANSWER_TOOL_MARKERS
|
||||
from backend.apps.agents.manager.session.history_compaction import get_branch_messages
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# States where the turn is genuinely over. "stopped" is excluded on purpose: the user pressed stop,
|
||||
# so they know exactly why it ended and do not need to be told.
|
||||
P_TERMINAL = ("completed", "error")
|
||||
|
||||
|
||||
@typechecked
|
||||
def p_tool_name_of(msg: object) -> str:
|
||||
content = getattr(msg, "content", None)
|
||||
if isinstance(content, dict):
|
||||
return str(content.get("tool") or content.get("tool_name") or "")
|
||||
return ""
|
||||
|
||||
|
||||
@typechecked
|
||||
def turn_left_the_user_with_nothing(session: AgentSession) -> bool:
|
||||
"""True when nothing readable has appeared since the user's last visible message."""
|
||||
msgs: List = [m for m in get_branch_messages(session) if not getattr(m, "hidden", False)]
|
||||
if not msgs:
|
||||
return False
|
||||
|
||||
p_last_user = -1
|
||||
for i, m in enumerate(msgs):
|
||||
if getattr(m, "role", "") == "user":
|
||||
p_last_user = i
|
||||
if p_last_user < 0:
|
||||
return False
|
||||
|
||||
for m in msgs[p_last_user + 1:]:
|
||||
role = getattr(m, "role", "")
|
||||
if role == "system":
|
||||
return False
|
||||
if role == "assistant":
|
||||
text = m.content if isinstance(m.content, str) else ""
|
||||
if text.strip():
|
||||
return False
|
||||
if role == "tool_call" and any(mk in p_tool_name_of(m) for mk in P_ANSWER_TOOL_MARKERS):
|
||||
# A rendered widget IS the answer; the user is looking at it.
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
@typechecked
|
||||
def ensure_turn_spoke(session: AgentSession, session_id: str) -> bool:
|
||||
"""Append the honest line if the turn is over and said nothing. Returns whether it fired."""
|
||||
if session.status not in P_TERMINAL:
|
||||
return False
|
||||
# A parked or continuing turn is not over; speaking now would be the lie.
|
||||
if getattr(session, "pending_continuation", False) or getattr(session, "awaiting_reconnect", False):
|
||||
return False
|
||||
if not turn_left_the_user_with_nothing(session):
|
||||
return False
|
||||
|
||||
session.messages.append(
|
||||
Message(role="system", content=EXHAUSTED_NOTE, branch_id=session.active_branch_id)
|
||||
)
|
||||
logger.warning(
|
||||
f"Agent {session_id}: turn ended with nothing readable and no detector claimed it; "
|
||||
"the floor spoke instead of leaving the user with a bare Done"
|
||||
)
|
||||
return True
|
||||
@@ -0,0 +1,134 @@
|
||||
"""The trigger hermes has and we did not (NousResearch/hermes-agent, MIT).
|
||||
|
||||
Their own test says our bug out loud: "On large-window models should_compress() (~50% of the
|
||||
window) rarely fires, so old tool outputs ride in history and are re-sent verbatim on every
|
||||
subsequent turn." Measured here: our shaping cuts 0.0% at EVERY session size, because the only
|
||||
lever is a 50KB per-message cap no single tool result reaches, so a 218K-token history ships
|
||||
untouched right up to a cliff the model chokes at (Haik quits around 149K, below our 180K trigger).
|
||||
|
||||
The aging itself was already lifted (aged_recap_lines). What was missing is that it only ran at a
|
||||
compaction boundary, which on a 1M window almost never arrives. So this is the second, INDEPENDENT
|
||||
trigger: cheap, deterministic, no LLM call, fired on cost rather than on a percentage of a window
|
||||
nobody reaches.
|
||||
|
||||
One adaptation, because our architecture is not theirs. Hermes owns its message list and rewrites
|
||||
it in place. Our transcript lives inside the CLI, so the only way to make the provider see an aged
|
||||
history is to rebuild on a fresh session whose recap is the aged one. The rebuild IS our prune.
|
||||
|
||||
That makes hermes's PROMPT-CACHE CONTRACT load-bearing rather than optional: a rebuild busts the
|
||||
cached prefix, so it commits only when it reclaims enough to be worth that, and then disarms until
|
||||
history has regrown a full runway. Without both gates this would trade tokens for cache misses and
|
||||
come out slower.
|
||||
"""
|
||||
|
||||
import logging
|
||||
|
||||
from typeguard import typechecked
|
||||
|
||||
from backend.apps.agents.core.models import AgentSession
|
||||
from backend.apps.agents.manager.session.aged_recap_lines import TAIL_BUDGET_CHARS, TAIL_COUNT_FLOOR
|
||||
from backend.apps.agents.manager.session.history_compaction import get_branch_messages
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Deliberately NOT a fraction of the context window: that is the mistake this exists to correct.
|
||||
# A fixed cost trigger fires on a 1M lane and a 200K lane alike, because the tokens cost the same.
|
||||
PROACTIVE_PRUNE_TOKENS = 60_000
|
||||
|
||||
# A rebuild is only worth its cache miss if it reclaims real weight.
|
||||
MIN_RECLAIM_TOKENS = 20_000
|
||||
|
||||
# After a commit, stay disarmed until history has regrown a full trigger-sized runway, so a session
|
||||
# hovering near the line cannot rebuild every turn.
|
||||
REARM_GROWTH_TOKENS = PROACTIVE_PRUNE_TOKENS
|
||||
|
||||
|
||||
# What one aged stub costs: the tool name, its arguments capped, and a size note.
|
||||
STUB_COST_CHARS = 220
|
||||
CHARS_PER_TOKEN = 4
|
||||
|
||||
|
||||
@typechecked
|
||||
def p_live_chars(session: AgentSession) -> int:
|
||||
"""Characters of history that would actually be SENT: everything after the compaction cutoff.
|
||||
|
||||
Compaction only marks a cutoff, it never deletes, so counting the whole message list keeps
|
||||
reporting the pre-prune size forever and the rearm gate can never disarm (caught by this
|
||||
module's own test). What matters is the live tail, because that is what a rebuild ships.
|
||||
"""
|
||||
msgs = [m for m in get_branch_messages(session) if not getattr(m, "hidden", False)]
|
||||
p_cut = getattr(session, "compacted_through_msg_id", None)
|
||||
if p_cut:
|
||||
for i, m in enumerate(msgs):
|
||||
if m.id == p_cut:
|
||||
msgs = msgs[i + 1:]
|
||||
break
|
||||
return sum(len(m.content if isinstance(m.content, str) else str(m.content)) for m in msgs)
|
||||
|
||||
|
||||
@typechecked
|
||||
def history_tokens(session: AgentSession) -> int:
|
||||
"""What the conversation itself costs, ignoring framework overhead.
|
||||
|
||||
The reported input total also carries the system prompt, tool schemas and MCP descriptions
|
||||
(~35K on a loaded session). None of that is reclaimable by pruning history, so measuring
|
||||
reclaim against it overstates the win and would fire this on sessions with nothing to give.
|
||||
"""
|
||||
return p_live_chars(session) // CHARS_PER_TOKEN
|
||||
|
||||
|
||||
@typechecked
|
||||
def estimate_aged_rebuild_tokens(session: AgentSession) -> int:
|
||||
"""What the history would cost AFTER an aged rebuild.
|
||||
|
||||
Deliberately not estimate_post_compact_input: that one measures what survives a cutoff that has
|
||||
already been marked, so before a commit it reports the whole history and makes every reclaim
|
||||
look negative (caught by this module's own tests). This measures the thing we would actually
|
||||
send: a verbatim tail inside its budget, plus a one-line stub per older tool result.
|
||||
"""
|
||||
msgs = [m for m in get_branch_messages(session) if not getattr(m, "hidden", False)]
|
||||
if not msgs:
|
||||
return 0
|
||||
p_total = p_live_chars(session)
|
||||
p_stubs = STUB_COST_CHARS * max(0, len(msgs) - TAIL_COUNT_FLOOR)
|
||||
p_after = min(p_total, TAIL_BUDGET_CHARS + p_stubs)
|
||||
return p_after // CHARS_PER_TOKEN
|
||||
|
||||
|
||||
@typechecked
|
||||
def should_proactively_prune(session: AgentSession) -> bool:
|
||||
"""True when an aged rebuild would pay for itself right now."""
|
||||
p_history = history_tokens(session)
|
||||
if p_history < PROACTIVE_PRUNE_TOKENS:
|
||||
return False
|
||||
|
||||
# Never duplicate the work the real compaction trigger is about to do anyway; that one reads
|
||||
# the reported input, because it is guarding the provider's hard wall rather than our cost.
|
||||
from backend.apps.agents.manager.context_budget import compact_trigger_tokens
|
||||
if int(session.tokens.get("input", 0) or 0) >= compact_trigger_tokens(session):
|
||||
return False
|
||||
|
||||
p_rearm = int(getattr(session, "proactive_prune_rearm_tokens", 0) or 0)
|
||||
if p_rearm and p_history < p_rearm:
|
||||
return False
|
||||
|
||||
p_after = estimate_aged_rebuild_tokens(session)
|
||||
p_reclaim = p_history - p_after
|
||||
if p_reclaim < MIN_RECLAIM_TOKENS:
|
||||
return False
|
||||
|
||||
logger.info(
|
||||
f"proactive prune: {p_history} tokens of history, an aged rebuild reclaims ~{p_reclaim}; committing"
|
||||
)
|
||||
return True
|
||||
|
||||
|
||||
@typechecked
|
||||
def arm_proactive_prune(session: AgentSession) -> None:
|
||||
"""Commit the prune: mark history aged and force the rebuild that actually applies it."""
|
||||
from backend.apps.agents.manager.context_budget import maybe_compact
|
||||
maybe_compact(session, force=True)
|
||||
session.needs_fresh_session = True
|
||||
session.proactive_prune_rearm_tokens = (
|
||||
estimate_aged_rebuild_tokens(session) + REARM_GROWTH_TOKENS
|
||||
)
|
||||
@@ -117,10 +117,46 @@ def test_unreadable_health_lets_the_turn_proceed(monkeypatch):
|
||||
|
||||
|
||||
def test_only_terminal_states_count_as_dead():
|
||||
assert lp.connection_is_dead({"testStatus": "unavailable"}) is True
|
||||
# "unavailable" alone is NOT enough: the router stamps it for throttles and 5xx as well, so it
|
||||
# cannot distinguish a dead credential from a bad minute (corrected after a live false positive).
|
||||
assert lp.connection_is_dead({"testStatus": "unavailable"}) is False
|
||||
assert lp.connection_is_dead({"errorCode": 401}) is True
|
||||
assert lp.connection_is_dead({"errorCode": 403}) is True
|
||||
# A slow, rate-limited or merely idle connection is NOT dead; grounding those would be the bug.
|
||||
assert lp.connection_is_dead({"testStatus": "active", "errorCode": 429}) is False
|
||||
assert lp.connection_is_dead({"testStatus": "active", "errorCode": 502}) is False
|
||||
assert lp.connection_is_dead({}) is False
|
||||
|
||||
|
||||
def test_never_dispatches_into_a_router_that_did_not_come_back(monkeypatch):
|
||||
"""A bounce that fails to restart leaves nothing listening. Dispatching into that is a
|
||||
guaranteed connection error the user would read as the model failing, rather than as us
|
||||
restarting something underneath them."""
|
||||
async def fake_get_providers():
|
||||
return P_DEAD
|
||||
|
||||
async def failed_bounce(provider):
|
||||
return False
|
||||
|
||||
import backend.apps.nine_router as nr
|
||||
import backend.apps.nine_router.bounce_after_connect as ba
|
||||
monkeypatch.setattr(nr, "get_providers", fake_get_providers, raising=True)
|
||||
monkeypatch.setattr(ba, "bounce_router_after_connect", failed_bounce, raising=True)
|
||||
|
||||
msg = asyncio.run(lp.preflight_lane("cx/gpt-5.6"))
|
||||
assert msg and "restarting" in msg.lower()
|
||||
assert "reconnect" not in msg.lower(), "this is our restart, not the user's credential"
|
||||
|
||||
|
||||
def test_a_rate_limited_lane_is_not_a_dead_credential():
|
||||
"""Live 2026-08-20: antigravity sat at testStatus=unavailable with errorCode=429 and a
|
||||
credential valid for another 30 minutes. Telling that user to reconnect is the same lie as
|
||||
"just rotated" for a dead token, aimed the other way."""
|
||||
assert lp.connection_is_dead({"testStatus": "unavailable", "errorCode": 429}) is False
|
||||
assert lp.connection_is_dead({"testStatus": "unavailable", "errorCode": 503}) is False
|
||||
assert lp.connection_is_dead({"testStatus": "unavailable", "errorCode": None}) is False
|
||||
|
||||
|
||||
def test_only_auth_shaped_failures_send_the_user_to_settings():
|
||||
assert lp.connection_is_dead({"testStatus": "unavailable", "errorCode": 401}) is True
|
||||
assert lp.connection_is_dead({"testStatus": "active", "errorCode": 403}) is True
|
||||
|
||||
@@ -0,0 +1,88 @@
|
||||
"""The trigger that fires on cost instead of on a percentage of the window.
|
||||
|
||||
Measured bug this closes: our shaping cut 0.0% at every session size because the compaction
|
||||
threshold is a fraction of the context window, and on a 1M lane that fraction is never reached, so
|
||||
a 218K history shipped verbatim to a cliff the model chokes at first. Hermes hit the same wall and
|
||||
solved it with a second, independent trigger (MIT, NousResearch/hermes-agent).
|
||||
"""
|
||||
|
||||
from backend.apps.agents.core.models import AgentSession, Message
|
||||
from backend.apps.agents.manager.session.proactive_prune import (
|
||||
MIN_RECLAIM_TOKENS,
|
||||
PROACTIVE_PRUNE_TOKENS,
|
||||
arm_proactive_prune,
|
||||
should_proactively_prune,
|
||||
)
|
||||
|
||||
|
||||
def p_session(input_tokens: int, bulky_msgs: int = 40, window: int = 1_000_000) -> AgentSession:
|
||||
s = AgentSession(name="t", model="sonnet-5", dashboard_id="d")
|
||||
s.context_window = window
|
||||
s.tokens["input"] = input_tokens
|
||||
s.messages.append(Message(role="user", content="go", branch_id=s.active_branch_id))
|
||||
for i in range(bulky_msgs):
|
||||
s.messages.append(Message(role="tool_call", content={"tool": "Read", "input": {"n": i}},
|
||||
branch_id=s.active_branch_id))
|
||||
s.messages.append(Message(role="tool_result", content={"text": "x" * 20_000},
|
||||
branch_id=s.active_branch_id))
|
||||
return s
|
||||
|
||||
|
||||
def test_a_big_history_on_a_1m_window_is_finally_pruned():
|
||||
"""The exact case that shipped 0.0%: far past any sane cost, nowhere near 50% of 1M."""
|
||||
s = p_session(120_000)
|
||||
assert should_proactively_prune(s) is True
|
||||
|
||||
|
||||
def test_a_small_session_is_left_alone():
|
||||
"""Negative control: pruning a cheap session spends a prompt cache for nothing."""
|
||||
assert should_proactively_prune(p_session(5_000, bulky_msgs=2)) is False
|
||||
|
||||
|
||||
def test_it_never_duplicates_the_work_the_real_threshold_is_about_to_do():
|
||||
"""Above the compaction trigger the existing path owns it; two rebuilds would be one wasted."""
|
||||
s = p_session(700_000)
|
||||
assert should_proactively_prune(s) is False
|
||||
|
||||
|
||||
def test_a_prune_that_reclaims_little_is_refused():
|
||||
"""The prompt-cache contract: a rebuild rewrites bytes the provider cached, so it must earn it.
|
||||
A long conversation of SHORT messages has nothing worth reclaiming."""
|
||||
s = AgentSession(name="t", model="sonnet-5", dashboard_id="d")
|
||||
s.context_window = 1_000_000
|
||||
s.tokens["input"] = PROACTIVE_PRUNE_TOKENS + 10_000
|
||||
for i in range(60):
|
||||
s.messages.append(Message(role="user", content=f"q{i}", branch_id=s.active_branch_id))
|
||||
s.messages.append(Message(role="assistant", content=f"a{i}", branch_id=s.active_branch_id))
|
||||
assert should_proactively_prune(s) is False
|
||||
|
||||
|
||||
def test_committing_disarms_until_history_regrows():
|
||||
"""A session hovering at the line must not rebuild every single turn."""
|
||||
s = p_session(120_000)
|
||||
assert should_proactively_prune(s) is True
|
||||
arm_proactive_prune(s)
|
||||
assert s.needs_fresh_session is True, "the rebuild is what actually applies the aging"
|
||||
assert s.proactive_prune_rearm_tokens > 0
|
||||
|
||||
# Same history again right after: disarmed.
|
||||
assert should_proactively_prune(s) is False
|
||||
|
||||
# History genuinely regrown past the runway: armed again.
|
||||
for i in range(40):
|
||||
s.messages.append(Message(role="tool_call", content={"tool": "Read", "input": {"n": 900 + i}},
|
||||
branch_id=s.active_branch_id))
|
||||
s.messages.append(Message(role="tool_result", content={"text": "y" * 20_000},
|
||||
branch_id=s.active_branch_id))
|
||||
s.compacted_through_msg_id = None
|
||||
assert should_proactively_prune(s) is True
|
||||
|
||||
|
||||
def test_the_trigger_is_not_a_fraction_of_the_window():
|
||||
"""The whole correction: identical history fires on a 200K lane and a 1M lane alike, because
|
||||
the tokens cost the same money either way. Tying this to a percentage of the window is what
|
||||
let a 218K history sail through untouched on the big lane."""
|
||||
small = p_session(50_000, window=200_000)
|
||||
big = p_session(50_000, window=1_000_000)
|
||||
assert should_proactively_prune(small) is True
|
||||
assert should_proactively_prune(big) is True
|
||||
@@ -210,3 +210,25 @@ def test_a_dead_socket_respawns_the_cli_but_a_429_does_not(monkeypatch):
|
||||
throttled, _ = p_drive(monkeypatch, Exception("429 rate_limit_error: overloaded"))
|
||||
assert throttled.needs_fresh_session is False, "a refusal is not a broken pipe"
|
||||
assert throttled.awaiting_reconnect is True, "but it is still worth waiting out"
|
||||
|
||||
|
||||
def test_a_spent_budget_stops_pretending_the_turn_is_parked(monkeypatch):
|
||||
"""Live catch, 2026-08-20 (rate-limited Gemini): after three outage rounds the budget is gone
|
||||
and the ask is over, but awaiting_reconnect stayed True. The terminal floor deliberately keeps
|
||||
quiet for parked turns, so the stale flag muzzled it and the run ended in total silence: the
|
||||
precise failure both features exist to prevent, created by one of them."""
|
||||
session = p_session()
|
||||
for _ in RECONNECT_BACKOFFS:
|
||||
session.pending_continuation = False
|
||||
p_drive(monkeypatch, ConnectionError("network is unreachable"), session=session)
|
||||
assert session.awaiting_reconnect is True, "still parked while the budget lasts"
|
||||
|
||||
session.pending_continuation = False
|
||||
p_drive(monkeypatch, ConnectionError("network is unreachable"), session=session)
|
||||
assert session.awaiting_reconnect is False, "budget spent: the turn is over, not parked"
|
||||
|
||||
# And with the flag honest, the floor can finally speak for this session.
|
||||
from backend.apps.agents.manager.run.turn_spoke import ensure_turn_spoke
|
||||
session.messages.append(Message(role="tool_call", content={"tool": "Read"},
|
||||
branch_id=session.active_branch_id))
|
||||
assert ensure_turn_spoke(session, "sid") is True
|
||||
|
||||
@@ -0,0 +1,91 @@
|
||||
"""The floor: a finished turn always left the user something to read.
|
||||
|
||||
Every earlier silent-stop fix was a detector for one shape found in the field, which means the
|
||||
NEXT shape nobody has found yet still ships as silence. This asks the user's own question at the
|
||||
one exit every terminal path passes through, so the state stops being representable rather than
|
||||
being enumerated.
|
||||
"""
|
||||
|
||||
from backend.apps.agents.core.models import AgentSession, Message
|
||||
from backend.apps.agents.manager.run.empty_finish import EXHAUSTED_NOTE
|
||||
from backend.apps.agents.manager.run.turn_spoke import ensure_turn_spoke, turn_left_the_user_with_nothing
|
||||
|
||||
|
||||
def p_session(*msgs, status="completed") -> AgentSession:
|
||||
s = AgentSession(name="t", model="sonnet", dashboard_id="d")
|
||||
s.status = status
|
||||
for role, content in msgs:
|
||||
s.messages.append(Message(role=role, content=content, branch_id=s.active_branch_id))
|
||||
return s
|
||||
|
||||
|
||||
def p_cards(s):
|
||||
return [m for m in s.messages if m.role == "system"]
|
||||
|
||||
|
||||
def test_a_shape_no_detector_knows_about_still_gets_a_line():
|
||||
"""The whole point: this must fire for causes nobody has enumerated. Here the turn ended with
|
||||
only a tool call, which no existing detector claimed."""
|
||||
s = p_session(("user", "audit the repo"), ("tool_call", {"tool": "Bash"}))
|
||||
assert turn_left_the_user_with_nothing(s) is True
|
||||
assert ensure_turn_spoke(s, "sid") is True
|
||||
assert [m.content for m in p_cards(s)] == [EXHAUSTED_NOTE]
|
||||
|
||||
|
||||
def test_it_never_speaks_over_a_turn_that_answered():
|
||||
s = p_session(("user", "hi"), ("assistant", "here is your answer"))
|
||||
assert ensure_turn_spoke(s, "sid") is False
|
||||
assert p_cards(s) == []
|
||||
|
||||
|
||||
def test_it_never_doubles_up_on_a_detector_that_already_spoke():
|
||||
"""Upstream detectors write BETTER messages because they know why; the floor must stay quiet
|
||||
whenever one of them already did the job."""
|
||||
s = p_session(("user", "go"), ("tool_call", {"tool": "Read"}),
|
||||
("system", "Your ChatGPT subscription needs reconnecting."))
|
||||
assert ensure_turn_spoke(s, "sid") is False
|
||||
assert len(p_cards(s)) == 1
|
||||
|
||||
|
||||
def test_a_rendered_widget_counts_as_an_answer():
|
||||
s = p_session(("user", "pick one"), ("tool_call", {"tool": "mcp__openswarm-ui__AskUI"}))
|
||||
assert ensure_turn_spoke(s, "sid") is False
|
||||
|
||||
|
||||
def test_a_still_running_turn_is_left_alone():
|
||||
s = p_session(("user", "go"), ("tool_call", {"tool": "Read"}), status="running")
|
||||
assert ensure_turn_spoke(s, "sid") is False
|
||||
|
||||
|
||||
def test_a_parked_turn_is_not_over_so_it_stays_quiet():
|
||||
"""Speaking over a turn that is about to resume would be the lie this exists to prevent."""
|
||||
s = p_session(("user", "go"), ("tool_call", {"tool": "Read"}))
|
||||
s.awaiting_reconnect = True
|
||||
assert ensure_turn_spoke(s, "sid") is False
|
||||
s.awaiting_reconnect = False
|
||||
s.pending_continuation = True
|
||||
assert ensure_turn_spoke(s, "sid") is False
|
||||
|
||||
|
||||
def test_a_user_stop_needs_no_explanation():
|
||||
"""They pressed stop; telling them it stopped is noise, not honesty."""
|
||||
s = p_session(("user", "go"), ("tool_call", {"tool": "Read"}), status="stopped")
|
||||
assert ensure_turn_spoke(s, "sid") is False
|
||||
|
||||
|
||||
def test_only_work_since_the_LAST_ask_counts():
|
||||
"""A reply to the previous question must not excuse silence on the current one."""
|
||||
s = p_session(("user", "first"), ("assistant", "answered the first"),
|
||||
("user", "second"), ("tool_call", {"tool": "Read"}))
|
||||
assert turn_left_the_user_with_nothing(s) is True
|
||||
|
||||
|
||||
def test_hidden_machinery_is_invisible_to_the_check():
|
||||
"""Our own hidden continuations are not the user speaking, and not us answering."""
|
||||
s = AgentSession(name="t", model="sonnet", dashboard_id="d")
|
||||
s.status = "completed"
|
||||
s.messages.append(Message(role="user", content="go", branch_id=s.active_branch_id))
|
||||
s.messages.append(Message(role="tool_call", content={"tool": "Read"}, branch_id=s.active_branch_id))
|
||||
s.messages.append(Message(role="user", content="[Automated] continue",
|
||||
branch_id=s.active_branch_id, hidden=True))
|
||||
assert turn_left_the_user_with_nothing(s) is True
|
||||
Reference in New Issue
Block a user