Live Stream: add compression recovery action

This commit is contained in:
Frank Song committed 2026-07-04 21:42:32 +08:00
1 parent 980e98dc7e
commit dfc0686dbd
13 files changed
+690 -19

No files matched your search

+111
View File
@@ -0,0 +1,111 @@
"""Compression-exhausted recovery metadata and intent helpers."""
from __future__ import annotations
import re
import time
from typing import Any
COMPRESSION_RECOVERY_TERMINAL_STATE = "compression_exhausted"
COMPRESSION_RECOVERY_ACTION_START_FOCUSED = "start_focused_continuation"
_GENERIC_CONTINUATION_INTENTS = frozenset(
{
"continue",
"continue please",
"go on",
"keep going",
"resume",
"proceed",
"carry on",
"继续",
"继续吧",
"接着",
"接着做",
"继续做",
"继续执行",
}
)
def _positive_int(value: Any) -> int:
try:
parsed = int(value or 0)
except (TypeError, ValueError):
return 0
return parsed if parsed > 0 else 0
def _normalize_intent_text(text: str) -> str:
raw = str(text or "").strip().lower()
if not raw:
return ""
# Keep CJK letters while dropping punctuation/emoji/noise around short intents.
return re.sub(r"[\W_]+", " ", raw, flags=re.UNICODE).strip()
def is_generic_continuation_intent(text: str) -> bool:
"""Return True only for short, content-free continuation requests."""
normalized = _normalize_intent_text(text)
if normalized in _GENERIC_CONTINUATION_INTENTS:
return True
# A single repeated word such as "continue continue" is still generic, but
# longer prompts like "continue by summarizing file X" are substantive.
parts = normalized.split()
return bool(parts) and len(parts) <= 2 and all(part in _GENERIC_CONTINUATION_INTENTS for part in parts)
def build_compression_recovery_payload(session, *, message: str = "", details: str = "") -> dict:
"""Build the durable UI/route payload for a compression-exhausted turn."""
source_sid = str(getattr(session, "session_id", "") or "")
last_prompt_tokens = _positive_int(getattr(session, "last_prompt_tokens", None))
threshold_tokens = _positive_int(getattr(session, "threshold_tokens", None))
context_length = _positive_int(getattr(session, "context_length", None))
payload = {
"terminal_state": COMPRESSION_RECOVERY_TERMINAL_STATE,
"recommended_action": COMPRESSION_RECOVERY_ACTION_START_FOCUSED,
"source_session_id": source_sid,
"created_at": time.time(),
"title": "Context compression exhausted",
"summary": (
"This run could not safely shrink the conversation enough to continue in place. "
"Start a focused continuation, then describe the next narrow task."
),
"action_label": "Start focused continuation",
"message": str(message or "").strip(),
"details": str(details or "").strip()[:1200],
"last_prompt_tokens": last_prompt_tokens,
"threshold_tokens": threshold_tokens,
"context_length": context_length,
}
return payload
def stamp_compression_exhausted_recovery(session, *, message: str = "", details: str = "") -> dict:
"""Persist recovery metadata on a session and return the message payload."""
payload = build_compression_recovery_payload(session, message=message, details=details)
session.recommended_recovery_action = payload["recommended_action"]
session.compression_recovery = payload
return payload
def compression_recovery_payload_for_session(session) -> dict | None:
payload = getattr(session, "compression_recovery", None)
if not isinstance(payload, dict):
return None
if payload.get("terminal_state") != COMPRESSION_RECOVERY_TERMINAL_STATE:
return None
action = str(payload.get("recommended_action") or getattr(session, "recommended_recovery_action", "") or "")
if action != COMPRESSION_RECOVERY_ACTION_START_FOCUSED:
return None
return payload
def clear_compression_recovery(session) -> None:
session.recommended_recovery_action = None
session.compression_recovery = {}
+7
View File
@@ -1051,6 +1051,8 @@ class Session:
context_engine_state=None,
context_length=None, threshold_tokens=None,
last_prompt_tokens=None,
compression_recovery=None,
recommended_recovery_action=None,
truncation_watermark=None,
truncation_boundary=None,
gateway_routing=None, gateway_routing_history=None,
@@ -1102,6 +1104,8 @@ class Session:
self.context_length = context_length
self.threshold_tokens = threshold_tokens
self.last_prompt_tokens = last_prompt_tokens
self.compression_recovery = compression_recovery if isinstance(compression_recovery, dict) else {}
self.recommended_recovery_action = recommended_recovery_action
self.truncation_watermark = truncation_watermark
self.truncation_boundary = truncation_boundary
self.gateway_routing = gateway_routing if isinstance(gateway_routing, dict) else None
@@ -1172,6 +1176,7 @@ class Session:
'context_engine', 'compression_anchor_engine', 'compression_anchor_mode',
'compression_anchor_details', 'context_engine_state',
'context_length', 'threshold_tokens', 'last_prompt_tokens',
'compression_recovery', 'recommended_recovery_action',
'truncation_watermark',
'truncation_boundary',
'gateway_routing', 'gateway_routing_history', 'llm_title_generated', 'manual_title',
@@ -1419,6 +1424,8 @@ class Session:
'context_length': self.context_length,
'threshold_tokens': self.threshold_tokens,
'last_prompt_tokens': self.last_prompt_tokens,
'compression_recovery': self.compression_recovery,
'recommended_recovery_action': self.recommended_recovery_action,
'gateway_routing': self.gateway_routing,
'gateway_routing_history': self.gateway_routing_history,
'manual_title': self.manual_title,
+136 -4
View File
@@ -42,6 +42,12 @@ from api.agent_sessions import (
read_session_lineage_report,
)
from api.compression_anchor import visible_messages_for_anchor
from api.compression_recovery import (
COMPRESSION_RECOVERY_ACTION_START_FOCUSED,
clear_compression_recovery,
compression_recovery_payload_for_session,
is_generic_continuation_intent,
)
from api.session_events import (
add_session_list_changed_listener,
publish_session_list_changed,
@@ -13099,6 +13105,9 @@ def handle_post(handler, parsed) -> bool:
)
return j(handler, {"session": s.compact() | {"messages": s.messages}})
if parsed.path == "/api/session/compression-recovery/start":
return _handle_session_compression_recovery_start(handler, body)
if parsed.path == "/api/session/duplicate":
try:
sid = body.get("session_id")
@@ -19724,6 +19733,89 @@ def _handle_bg_task_complete_ack(handler, body):
)
def _handle_session_compression_recovery_start(handler, body):
try:
require(body, "session_id")
except ValueError as e:
return bad(handler, str(e))
sid = str(body.get("session_id") or "").strip()
if not sid:
return bad(handler, "session_id is required")
if _session_is_subagent_view_only(sid):
return bad(handler, "Subagent sessions are view-only and cannot start compression recovery from WebUI", 400)
try:
source = get_session(sid)
except KeyError:
return bad(handler, "Session not found", 404)
if not _session_visible_to_active_profile(getattr(source, "profile", None), handler):
return bad(handler, "Session not found", 404)
recovery = compression_recovery_payload_for_session(source)
if not recovery:
return bad(handler, "Session does not have a compression recovery action.", 409)
action = str(recovery.get("recommended_action") or "")
if action != COMPRESSION_RECOVERY_ACTION_START_FOCUSED:
return bad(handler, "Unsupported compression recovery action.", 409)
title = str(getattr(source, "title", None) or "Untitled").strip() or "Untitled"
if not title.endswith(" (focused continuation)"):
title = f"{title} (focused continuation)"
copied_session = Session(
session_id=uuid.uuid4().hex[:12],
title=title,
workspace=getattr(source, "workspace", get_last_workspace()),
model=getattr(source, "model", None),
model_provider=getattr(source, "model_provider", None),
messages=[],
tool_calls=[],
pinned=False,
archived=False,
project_id=getattr(source, "project_id", None),
profile=getattr(source, "profile", None),
personality=getattr(source, "personality", None),
enabled_toolsets=copy.deepcopy(getattr(source, "enabled_toolsets", None)),
context_length=getattr(source, "context_length", None),
threshold_tokens=getattr(source, "threshold_tokens", None),
gateway_routing=copy.deepcopy(getattr(source, "gateway_routing", None)),
gateway_routing_history=copy.deepcopy(getattr(source, "gateway_routing_history", None) or []),
parent_session_id=getattr(source, "session_id", sid),
worktree_path=getattr(source, "worktree_path", None),
worktree_branch=getattr(source, "worktree_branch", None),
worktree_repo_root=getattr(source, "worktree_repo_root", None),
worktree_created_at=getattr(source, "worktree_created_at", None),
)
# Preserve the workspace/model/profile lane, but intentionally start with an
# empty model-facing transcript so a focused follow-up does not replay the
# exhausted state.db/context tail.
copied_session.context_messages = []
copied_session.composer_draft = {"text": "", "files": []}
try:
copied_session.save()
except Exception as e:
logger.exception("failed to persist compression recovery session for %s", sid)
return bad(handler, f"Failed to start compression recovery: {_sanitize_error(e)}", 500)
with LOCK:
SESSIONS[copied_session.session_id] = copied_session
SESSIONS.move_to_end(copied_session.session_id)
_evict_sessions_over_cap()
publish_session_list_changed(
"session_compression_recovery",
profile=getattr(copied_session, "profile", None),
session_id=getattr(copied_session, "session_id", None),
)
session_payload = redact_session_data(copied_session.compact() | {"messages": copied_session.messages})
return j(
handler,
{
"ok": True,
"session": session_payload,
"source_session_id": sid,
"recommended_recovery_action": action,
"message": "Started a focused continuation. Describe the next narrow task to continue.",
},
)
def _handle_goal_command(handler, body):
"""Handle WebUI /goal command controls and optional kickoff stream."""
try:
@@ -19979,6 +20071,19 @@ def _handle_chat_start(handler, body, diag=None):
return bad(handler, "message is required")
diag.stage("normalize_attachments") if diag else None
attachments = _normalize_chat_attachments(body.get("attachments") or [])[:20]
recovery = compression_recovery_payload_for_session(s)
if recovery and not attachments and is_generic_continuation_intent(msg):
return j(
handler,
{
"error": "This session exhausted context compression. Start a focused continuation, then describe the next narrow task.",
"type": "compression_recovery_required",
"recommended_recovery_action": recovery.get("recommended_action"),
"compression_recovery": recovery,
"session_id": getattr(s, "session_id", body["session_id"]),
},
status=409,
)
diag.stage("resolve_workspace") if diag else None
try:
workspace = _resolve_chat_workspace_with_recovery(s, body.get("workspace"))
@@ -20038,16 +20143,43 @@ def _handle_chat_start(handler, body, diag=None):
}
if not gateway_chat_enabled and moa_config is not None:
start_run_kwargs["moa_config"] = moa_config
response = _start_run(
s,
**start_run_kwargs,
)
recovery_cleared_for_start = None
def _restore_cleared_recovery():
if recovery_cleared_for_start is None:
return None
s.compression_recovery = recovery_cleared_for_start
s.recommended_recovery_action = recovery_cleared_for_start.get("recommended_action")
try:
s.save()
except Exception as restore_err:
logger.exception("failed to restore compression recovery after chat start rejection for %s", getattr(s, "session_id", None))
return restore_err
return None
if recovery:
recovery_cleared_for_start = copy.deepcopy(recovery)
clear_compression_recovery(s)
try:
response = _start_run(
s,
**start_run_kwargs,
)
except Exception:
_restore_cleared_recovery()
raise
# Map adapter-selection NotImplementedError (501) onto the legacy
# bad-request response shape that this route exposed historically
# before the helper extraction.
if response.get("_status") == 501 and "error" in response:
restore_err = _restore_cleared_recovery()
if restore_err is not None:
return bad(handler, f"failed to restore compression recovery: {_sanitize_error(restore_err)}", 500)
return j(handler, {"error": response["error"]}, status=501)
status = int(response.pop("_status", 200) or 200)
if status >= 400 and recovery_cleared_for_start is not None:
restore_err = _restore_cleared_recovery()
if restore_err is not None:
return bad(handler, f"failed to restore compression recovery: {_sanitize_error(restore_err)}", 500)
diag.stage("response_write") if diag else None
return j(handler, response, status=status)
finally:
+2
View File
@@ -484,6 +484,8 @@ def _state_db_row_to_sidecar(row: dict) -> dict:
'context_length': None,
'threshold_tokens': None,
'last_prompt_tokens': None,
'compression_recovery': {},
'recommended_recovery_action': None,
'gateway_routing': None,
'gateway_routing_history': [],
'llm_title_generated': False,
+24
View File
@@ -44,6 +44,7 @@ from api.config import (
)
from api.helpers import redact_session_data, _redact_text
from api.compression_anchor import is_context_compression_marker, visible_messages_for_anchor
from api.compression_recovery import stamp_compression_exhausted_recovery
from api.metering import meter
from api.run_journal import RunJournalWriter
from api.todo_state import attach_todo_state, emit_todo_state
@@ -8569,6 +8570,15 @@ def _run_agent_streaming(
'timestamp': int(time.time()),
'_error': True,
}
if _err_type == 'compression_exhausted':
_recovery = stamp_compression_exhausted_recovery(
s,
message=_error_payload.get('message') or _err_label,
details=_error_payload.get('details') or '',
)
_error_message['_compressionRecovery'] = _recovery
_error_payload['compression_recovery'] = _recovery
_error_payload['recommended_recovery_action'] = _recovery.get('recommended_action')
if _error_payload.get('details'):
_error_message['provider_details'] = _error_payload['details']
if _err_type == 'cancelled':
@@ -9461,6 +9471,7 @@ def _run_agent_streaming(
_exc_is_not_found = _classification['type'] == 'model_not_found' # detects '404', 'not found', 'does not exist', and 'invalid model'.
_exc_is_cancelled = _classification['type'] == 'cancelled'
_exc_is_interrupted = _classification['type'] == 'interrupted'
_exc_is_compression_exhausted = _classification['type'] == 'compression_exhausted'
# The user hint still points to Settings / `hermes model` from _classify_provider_error().
if _exc_is_quota:
@@ -9581,6 +9592,10 @@ def _run_agent_streaming(
_exc_label, _exc_type, _exc_hint = (
_classification['label'], _classification['type'], _classification['hint'],
)
elif _exc_is_compression_exhausted:
_exc_label, _exc_type, _exc_hint = (
_classification['label'], _classification['type'], _classification['hint'],
)
else:
_exc_label, _exc_type, _exc_hint = 'Error', 'error', ''
@@ -9620,6 +9635,15 @@ def _run_agent_streaming(
'timestamp': int(time.time()),
'_error': True,
}
if _exc_type == 'compression_exhausted':
_recovery = stamp_compression_exhausted_recovery(
s,
message=_error_payload.get('message') or err_str,
details=_error_payload.get('details') or '',
)
_error_message['_compressionRecovery'] = _recovery
_error_payload['compression_recovery'] = _recovery
_error_payload['recommended_recovery_action'] = _recovery.get('recommended_action')
if _error_payload.get('details'):
_error_message['provider_details'] = _error_payload['details']
if _exc_type == 'cancelled':
+2
View File
@@ -51,6 +51,8 @@ _METADATA_FIELDS = frozenset(
"context_length",
"threshold_tokens",
"last_prompt_tokens",
"compression_recovery",
"recommended_recovery_action",
"truncation_watermark",
"truncation_boundary",
"gateway_routing",
+13 -13
View File
@@ -38,8 +38,8 @@ against current source before any route is added.
- This RFC does **not** implement `GET /api/sessions/{session_id}/events`. No
route, handler, or related code is added in this PR.
- This RFC does **not** modify `GET /api/sessions/events` (the existing global
session-list invalidation stream routed at `api/routes.py:12345-12346` and
implemented by `_handle_session_events_stream()` at `api/routes.py:16177`).
session-list invalidation stream routed at `api/routes.py:12351-12352` and
implemented by `_handle_session_events_stream()` at `api/routes.py:16186`).
- This RFC does **not** replace or modify existing streams: `/api/chat/stream`,
`/api/approval/stream`, or `/api/clarify/stream`.
- This RFC does **not** introduce Android, iOS, or PWA client code.
@@ -53,8 +53,8 @@ against current source before any route is added.
### Existing global session-list stream
`GET /api/sessions/events` is a **different endpoint** from the one this RFC
proposes. It is routed at `api/routes.py:12345-12346` and implemented by
`_handle_session_events_stream()` at `api/routes.py:16177`. It emits bare
proposes. It is routed at `api/routes.py:12351-12352` and implemented by
`_handle_session_events_stream()` at `api/routes.py:16186`. It emits bare
`sessions_changed` events and keepalives for any change to the session list. It
is a global invalidation signal, not a per-session lifecycle stream. The proposed
`GET /api/sessions/{session_id}/events` is per-session and path-distinct.
@@ -73,19 +73,19 @@ Line ranges in this inventory were verified against WebUI `master` when this
RFC was written. Function, constant, and endpoint names are the stable anchors
if source layout moves later.
- `_parse_run_journal_event_id()` (`api/routes.py:15673-15686`) and
`_parse_run_journal_after_seq()` (`api/routes.py:15688-15701`) parse the replay
- `_parse_run_journal_event_id()` (`api/routes.py:15682-15695`) and
`_parse_run_journal_after_seq()` (`api/routes.py:15697-15710`) parse the replay
cursor from the `after_event_id` / `after_seq` **query params** (not the
`Last-Event-ID` header — that header is the *proposed* new-endpoint contract
below, §Reconnect).
- `_runner_event_id()` at `api/routes.py:15765-15772` constructs the event `id`
- `_runner_event_id()` at `api/routes.py:15774-15781` constructs the event `id`
field as `stream_id:seq`.
- SSE frames carry their `id:` via the `_sse_with_id()` helper, emitted on the
live `/api/chat/stream` path at `api/routes.py:15918`, on the runner-observe
path at `api/routes.py:15811`, and during journal replay at
`api/routes.py:15721` / `15734`.
live `/api/chat/stream` path at `api/routes.py:15927`, on the runner-observe
path at `api/routes.py:15820`, and during journal replay at
`api/routes.py:15730` / `15743`.
- `_replay_run_journal()` reads events by `(session_id, stream_id)` at
`api/routes.py:15703-15735`.
`api/routes.py:15712-15744`.
- `api/streaming.py:6265-6285` writes current live agent streams to
`STREAMS[stream_id]`.
- `api/streaming.py:6620-6634` appends SSE events to the run journal and carries
@@ -165,7 +165,7 @@ position.
**`event_id` is opaque to clients.** Its current source-compatible form is
`stream_id:seq`, as constructed by `_runner_event_id()` at
`api/routes.py:15765-15772`. Clients must treat it as an opaque string and must
`api/routes.py:15774-15781`. Clients must treat it as an opaque string and must
not parse or construct cursor values.
**`seq` is monotonic within a stream/run.** It is not a session-global counter
@@ -183,7 +183,7 @@ events. The live `STREAMS[stream_id]` queue (`api/streaming.py:6265-6285`) is
not a reliable replay source because it holds only recent in-memory state.
A future implementation must replay from the run journal via the existing
`_replay_run_journal()` path (`api/routes.py:15703-15735`) and fall back to the
`_replay_run_journal()` path (`api/routes.py:15712-15744`) and fall back to the
snapshot mechanism when journal entries are unavailable for a given cursor.
## Snapshot fallback
+38
View File
@@ -121,6 +121,44 @@ The on-disk locations below assume the default `~/.hermes/webui` state directory
---
## "Context compression exhausted" after a long-running turn
**Symptom.** A long-running session, often with many tool calls or a small
context-window model, ends with a `Context compression exhausted` error instead
of a final answer. The message includes a recovery action labeled `Start focused
continuation`.
**Why.** Automatic compression could not shrink the current conversation enough
to continue safely in the same model-facing context. The exhausted session is
terminal: sending a bare "continue", "go on", or "继续" would usually replay the
same oversized state and fail again, so the WebUI points the user to a focused
linked continuation instead.
**Diagnostic.**
1. Open the session JSON under your WebUI state directory, for example:
```bash
jq '.recommended_recovery_action, .compression_recovery' \
~/.hermes/webui/sessions/<session_id>.json
```
2. A recoverable exhausted turn should report:
- `recommended_recovery_action: "start_focused_continuation"`
- `compression_recovery.terminal_state: "compression_exhausted"`
- the final assistant error message carrying `_compressionRecovery`
**Fix.** Use the `Start focused continuation` action in the exhausted message.
The new linked session preserves the workspace, model, profile, project, and
toolset lane, but intentionally starts with an empty model-facing transcript so
the oversized exhausted tail is not replayed. After the new session opens,
describe the next narrow task explicitly instead of sending a bare continuation.
**When to file a bug.** File a bug if the exhausted message has no recovery
action, the action creates a session with the old oversized context/messages
replayed into the model-facing transcript, or a bare "continue" starts another
turn in the exhausted session instead of being blocked with recovery guidance.
---
## Other troubleshooting
This document grows over time. If a recurring failure mode isn't covered here yet, add it via PR. The format for each entry: **Symptom → Why → Diagnostic commands → Fix → When to file a bug**.
+7 -1
View File
@@ -1326,6 +1326,11 @@ async function send(){
_flushSelectionBlocksToComposer();
text=$('msg').value.trim();
if(!text&&!S.pendingFiles.length){_sendInProgress=false;_sendInProgressSid=null;return;}
if(typeof shouldInterceptCompressionRecoveryContinuation==='function'&&shouldInterceptCompressionRecoveryContinuation(text,S.pendingFiles)){
if(typeof showCompressionRecoveryContinuationHint==='function') showCompressionRecoveryContinuationHint();
_sendInProgress=false;_sendInProgressSid=null;
return;
}
// #5472: snapshot the ORIGINAL user-typed composer state now — before slash
// rewrites (/moa, bundles) mutate `text` and before uploadPendingFiles()
@@ -5680,7 +5685,8 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){
if(typeof _setActiveSessionUrl==='function') _setActiveSessionUrl(S.session.session_id);
}
} else {
S.messages.push({role:'assistant',content:`**${label}:** ${d.message}${hint}`,provider_details:details,provider_details_label:detailsLabel});
const recovery=(d.compression_recovery&&typeof d.compression_recovery==='object')?d.compression_recovery:null;
S.messages.push({role:'assistant',content:`**${label}:** ${d.message}${hint}`,provider_details:details,provider_details_label:detailsLabel,_compressionRecovery:recovery||undefined});
_attachProjectedAnchorSceneToLastAssistant(S.messages);
}
}catch(_){
+8
View File
@@ -5726,6 +5726,14 @@ main.main > #mainPlugin{display:none;}
.status-card-label{color:var(--muted);font-weight:550;background:rgba(0,0,0,.015);}
.status-card-value{color:var(--text);word-break:break-word;font-family:var(--font-mono);}
@media (max-width:700px){.status-card{margin-left:0;}.status-card-head{flex-direction:column;}.status-card-session-copy{max-width:100%;}.status-card-grid{grid-template-columns:1fr;}.status-card-label{padding-bottom:2px;border-top:1px solid var(--border-subtle);}.status-card-value{padding-top:2px;border-top:0;}}
.compression-recovery-card{display:flex;align-items:center;justify-content:space-between;gap:14px;margin:12px 0 4px;padding:14px 16px;border:1px solid var(--border-subtle);border-radius:var(--radius-card);background:var(--surface-subtle);box-shadow:0 8px 20px rgba(0,0,0,.04);}
.compression-recovery-copy{display:flex;flex-direction:column;gap:3px;min-width:0;}
.compression-recovery-title{color:var(--text);font-size:14px;font-weight:650;line-height:1.35;}
.compression-recovery-summary{color:var(--muted);font-size:12px;line-height:1.45;}
.compression-recovery-action{display:inline-flex;align-items:center;justify-content:center;gap:7px;min-height:32px;padding:6px 11px;border:1px solid var(--accent-bg-strong);border-radius:7px;background:var(--accent-bg);color:var(--accent-text);font:inherit;font-size:12px;font-weight:650;white-space:nowrap;cursor:pointer;}
.compression-recovery-action:hover{background:var(--accent-bg-strong);color:var(--accent-text);}
.compression-recovery-action:disabled{opacity:.65;cursor:progress;}
@media (max-width:700px){.compression-recovery-card{flex-direction:column;align-items:stretch;margin-left:0;}.compression-recovery-action{width:100%;}}
.assistant-segment-anchor { display: none; }
/* ── Classic conversation layout: user right, half-width; assistant left ── */
+86 -1
View File
@@ -417,6 +417,89 @@ function _statusCardHtml(card){
</div>`;
}
function _compressionRecoveryHtml(recovery, sessionId){
if(!recovery||typeof recovery!=='object') return '';
if(String(recovery.terminal_state||'')!=='compression_exhausted') return '';
const action=String(recovery.recommended_action||'');
if(action!=='start_focused_continuation') return '';
const sid=String(sessionId||recovery.source_session_id||'');
const title=String(recovery.title||'Context compression exhausted');
const summary=String(recovery.summary||'Start a focused continuation, then describe the next narrow task.');
const actionLabel=String(recovery.action_label||'Start focused continuation');
const icon=(typeof li==='function')?li('git-branch',14):'';
return `<div class="compression-recovery-card" data-compression-recovery-card="1">
<div class="compression-recovery-copy">
<div class="compression-recovery-title">${esc(title)}</div>
<div class="compression-recovery-summary">${esc(summary)}</div>
</div>
<button class="compression-recovery-action" type="button" data-recovery-session-id="${esc(sid)}" onclick="startCompressionRecovery(this);event.stopPropagation()">${icon}<span>${esc(actionLabel)}</span></button>
</div>`;
}
function _activeCompressionRecoveryPayload(){
if(!S||!S.session) return null;
const recovery=S.session.compression_recovery;
if(recovery&&typeof recovery==='object'&&String(recovery.terminal_state||'')==='compression_exhausted') return recovery;
const messages=Array.isArray(S.messages)?S.messages:[];
for(let i=messages.length-1;i>=0;i--){
const msg=messages[i];
const msgRecovery=msg&&msg._compressionRecovery;
if(msgRecovery&&typeof msgRecovery==='object'&&String(msgRecovery.terminal_state||'')==='compression_exhausted') return msgRecovery;
}
return null;
}
function isGenericCompressionContinuationIntent(text){
const raw=String(text||'').trim().toLowerCase();
if(!raw) return false;
const normalized=raw.replace(/[^\p{L}\p{N}]+/gu,' ').trim();
const generic=new Set(['continue','continue please','go on','keep going','resume','proceed','carry on','继续','继续吧','接着','接着做','继续做','继续执行']);
if(generic.has(normalized)) return true;
const parts=normalized.split(/\s+/).filter(Boolean);
return !!parts.length&&parts.length<=2&&parts.every(part=>generic.has(part));
}
function shouldInterceptCompressionRecoveryContinuation(text, files){
const hasFiles=Array.isArray(files)&&files.length>0;
if(hasFiles||!isGenericCompressionContinuationIntent(text)) return false;
const recovery=_activeCompressionRecoveryPayload();
return !!(recovery&&String(recovery.recommended_action||'')==='start_focused_continuation');
}
function showCompressionRecoveryContinuationHint(){
const card=document.querySelector('[data-compression-recovery-card="1"]');
if(card&&typeof card.scrollIntoView==='function'){
try{card.scrollIntoView({block:'center',behavior:'smooth'});}catch(_){card.scrollIntoView();}
const btn=card.querySelector('.compression-recovery-action');
if(btn&&typeof btn.focus==='function') setTimeout(()=>btn.focus(),120);
}
if(typeof showToast==='function') showToast('This session exhausted context compression. Start a focused continuation, then describe the next narrow task.',4500,'warning');
}
async function startCompressionRecovery(btn){
const sourceSid=String((btn&&btn.dataset&&btn.dataset.recoverySessionId)||(S.session&&S.session.session_id)||'').trim();
if(!sourceSid) return;
if(btn){btn.disabled=true;btn.classList.add('loading');}
try{
const data=await api('/api/session/compression-recovery/start',{method:'POST',body:JSON.stringify({session_id:sourceSid})});
const sid=data&&data.session&&data.session.session_id;
if(sid){
try{localStorage.setItem('hermes-webui-session',sid);}catch(_){}
if(typeof loadSession==='function') await loadSession(sid,{preserveActiveInput:false});
else if(data.session){S.session=data.session;S.messages=data.session.messages||[];syncTopbar();renderMessages();}
if(typeof renderSessionList==='function') await renderSessionList();
if(typeof _setActiveSessionUrl==='function') _setActiveSessionUrl(sid);
if(typeof showToast==='function') showToast((data&&data.message)||'Started focused continuation.',3000,'success');
const composer=$('msg');
if(composer&&typeof composer.focus==='function') composer.focus();
}
}catch(e){
if(typeof showToast==='function') showToast('Compression recovery failed: '+(e&&e.message||e),5000,'error');
}finally{
if(btn){btn.disabled=false;btn.classList.remove('loading');}
}
}
const MESSAGE_RENDER_WINDOW_DEFAULT=50;
const MESSAGE_VIRTUAL_THRESHOLD_ROWS=80;
const MESSAGE_VIRTUAL_BUFFER_PX=900;
@@ -13390,6 +13473,8 @@ function renderMessages(options){
const summary=m.provider_details_label||'Provider details';
bodyHtml += `<details class="provider-error-details"><summary>${esc(String(summary))}</summary><pre><code>${esc(String(m.provider_details))}</code></pre></details>`;
}
const recoveryHtml=(!isUser&&m._compressionRecovery) ? _compressionRecoveryHtml(m._compressionRecovery, (S.session&&S.session.session_id)||'') : '';
if(recoveryHtml) bodyHtml += recoveryHtml;
const statusHtml = (!isUser&&m._statusCard) ? _statusCardHtml(m._statusCard) : '';
const isEditableUser=isUser&&rawIdx===lastUserRawIdx;
const editBtn = isEditableUser ? `<button class="msg-action-btn" title="${t('edit_message')}" onclick="editMessage(this)">${li('pencil',13)}</button>` : '';
@@ -13584,7 +13669,7 @@ function renderMessages(options){
if((isCompactWorklogMode()||isTransparentStream())&&_assistantThinkingBelongsInWorklog(m, rawIdx, toolCallAssistantIdxs)) assistantThinking.set(rawIdx, thinkingText);
else if(window._showThinking!==false) seg.insertAdjacentHTML('beforeend', _thinkingCardHtml(thinkingText));
}
const hasVisibleBody=!!(String(content||'').trim()||filesHtml||statusHtml);
const hasVisibleBody=!!(String(content||'').trim()||filesHtml||statusHtml||recoveryHtml);
if(statusHtml){
seg.insertAdjacentHTML('beforeend', statusHtml);
}else if(hasVisibleBody){
@@ -135,6 +135,9 @@ def test_compression_exhausted_after_session_rotation_preserves_snapshot_and_err
assert payload["session"]["session_id"] == new_sid
assert payload["old_session_id"] == old_sid
assert payload["new_session_id"] == new_sid
assert payload["recommended_recovery_action"] == "start_focused_continuation"
assert payload["compression_recovery"]["terminal_state"] == "compression_exhausted"
assert payload["compression_recovery"]["source_session_id"] == new_sid
old_payload = json.loads((session_dir / f"{old_sid}.json").read_text(encoding="utf-8"))
new_payload = json.loads((session_dir / f"{new_sid}.json").read_text(encoding="utf-8"))
@@ -144,7 +147,10 @@ def test_compression_exhausted_after_session_rotation_preserves_snapshot_and_err
assert new_payload["session_id"] == new_sid
assert new_payload["parent_session_id"] == old_sid
assert new_payload["pre_compression_snapshot"] is False
assert new_payload["recommended_recovery_action"] == "start_focused_continuation"
assert new_payload["compression_recovery"]["recommended_action"] == "start_focused_continuation"
assert new_payload["messages"][-1]["_error"] is True
assert new_payload["messages"][-1]["_compressionRecovery"]["recommended_action"] == "start_focused_continuation"
assert "Context compression exhausted" in new_payload["messages"][-1]["content"]
assert old_sid not in streaming.SESSIONS
assert streaming.SESSIONS[new_sid].session_id == new_sid
@@ -428,6 +434,8 @@ def test_apperror_payload_enriched_before_enqueue(tmp_path, monkeypatch):
assert payload_after["session_id"] == new_sid
assert payload_after["old_session_id"] == old_sid
assert payload_after["new_session_id"] == new_sid
assert payload_after["recommended_recovery_action"] == "start_focused_continuation"
assert payload_after["compression_recovery"]["recommended_action"] == "start_focused_continuation"
def test_exception_apperror_payload_includes_session_id_before_enqueue():
+248
View File
@@ -0,0 +1,248 @@
import io
import json
from pathlib import Path
from api import models, routes
from api.compression_recovery import (
compression_recovery_payload_for_session,
is_generic_continuation_intent,
stamp_compression_exhausted_recovery,
)
from api.models import Session
ROOT = Path(__file__).resolve().parents[1]
class _JSONHandler:
headers = {}
def __init__(self):
self.status = None
self.wfile = io.BytesIO()
self.headers_sent = {}
def send_response(self, status):
self.status = status
def send_header(self, key, value):
self.headers_sent[key] = value
def end_headers(self):
pass
def _payload(handler):
raw = handler.wfile.getvalue().decode("utf-8")
return json.loads(raw) if raw else {}
def _isolate_sessions(monkeypatch, tmp_path):
session_dir = tmp_path / "sessions"
session_dir.mkdir()
monkeypatch.setattr(models, "SESSION_DIR", session_dir)
monkeypatch.setattr(models, "SESSION_INDEX_FILE", session_dir / "_index.json")
models.SESSIONS.clear()
routes.SESSIONS.clear()
return session_dir
def test_generic_continuation_intent_is_scoped_to_empty_continue_words():
assert is_generic_continuation_intent("continue")
assert is_generic_continuation_intent("继续吧。")
assert is_generic_continuation_intent("go on")
assert not is_generic_continuation_intent("continue by summarizing the workspace changes")
assert not is_generic_continuation_intent("继续修复 4685 的恢复卡")
def test_chat_start_blocks_generic_continue_after_compression_exhausted(monkeypatch, tmp_path):
_isolate_sessions(monkeypatch, tmp_path)
sid = "recoverychat1"
session = Session(
session_id=sid,
title="Recovery",
workspace=str(tmp_path),
model="gpt-4o",
messages=[{"role": "user", "content": "long task"}],
)
stamp_compression_exhausted_recovery(session, message="Context length exceeded.")
models.SESSIONS[sid] = session
routes.SESSIONS[sid] = session
handler = _JSONHandler()
routes._handle_chat_start(handler, {"session_id": sid, "message": "继续"})
payload = _payload(handler)
assert handler.status == 409
assert payload["type"] == "compression_recovery_required"
assert payload["recommended_recovery_action"] == "start_focused_continuation"
assert payload["compression_recovery"]["terminal_state"] == "compression_exhausted"
def test_chat_start_keeps_recovery_when_substantive_prompt_fails_validation(monkeypatch, tmp_path):
session_dir = _isolate_sessions(monkeypatch, tmp_path)
sid = "recoverychat2"
session = Session(
session_id=sid,
title="Recovery",
workspace=str(tmp_path),
model="gpt-4o",
messages=[{"role": "user", "content": "long task"}],
)
stamp_compression_exhausted_recovery(session, message="Context length exceeded.")
session.save()
models.SESSIONS[sid] = session
routes.SESSIONS[sid] = session
monkeypatch.setattr(
routes,
"_resolve_chat_workspace_with_recovery",
lambda *_args, **_kwargs: (_ for _ in ()).throw(ValueError("bad workspace")),
)
handler = _JSONHandler()
routes._handle_chat_start(handler, {"session_id": sid, "message": "continue by checking the repo"})
saved = json.loads((session_dir / f"{sid}.json").read_text(encoding="utf-8"))
assert handler.status == 400
assert saved["recommended_recovery_action"] == "start_focused_continuation"
assert saved["compression_recovery"]["terminal_state"] == "compression_exhausted"
def test_chat_start_clears_recovery_when_substantive_prompt_starts(monkeypatch, tmp_path):
session_dir = _isolate_sessions(monkeypatch, tmp_path)
sid = "recoverychat3"
session = Session(
session_id=sid,
title="Recovery",
workspace=str(tmp_path),
model="gpt-4o",
messages=[{"role": "user", "content": "long task"}],
)
stamp_compression_exhausted_recovery(session, message="Context length exceeded.")
session.save()
models.SESSIONS[sid] = session
routes.SESSIONS[sid] = session
monkeypatch.setattr(routes, "_resolve_chat_workspace_with_recovery", lambda *_args, **_kwargs: str(tmp_path))
monkeypatch.setattr(routes, "_read_profile_model_config", lambda *_args, **_kwargs: (None, None, {}))
monkeypatch.setattr(
routes,
"_resolve_compatible_session_model_state",
lambda requested_model, requested_provider, **_kwargs: (requested_model or "gpt-4o", requested_provider or "openai", False),
)
monkeypatch.setattr(routes, "webui_gateway_chat_enabled", lambda _config: False)
def _fake_start_run(run_session, **_kwargs):
assert compression_recovery_payload_for_session(run_session) is None
run_session.save()
return {"session_id": sid, "stream_id": "stream1", "_status": 200}
monkeypatch.setattr(routes, "_start_run", _fake_start_run)
handler = _JSONHandler()
routes._handle_chat_start(handler, {"session_id": sid, "message": "continue by checking the repo"})
payload = _payload(handler)
saved = json.loads((session_dir / f"{sid}.json").read_text(encoding="utf-8"))
assert handler.status == 200
assert payload["stream_id"] == "stream1"
assert saved["compression_recovery"] == {}
assert saved["recommended_recovery_action"] is None
def test_chat_start_restores_recovery_when_substantive_prompt_start_is_rejected(monkeypatch, tmp_path):
session_dir = _isolate_sessions(monkeypatch, tmp_path)
sid = "recoverychat4"
session = Session(
session_id=sid,
title="Recovery",
workspace=str(tmp_path),
model="gpt-4o",
messages=[{"role": "user", "content": "long task"}],
)
stamp_compression_exhausted_recovery(session, message="Context length exceeded.")
session.save()
models.SESSIONS[sid] = session
routes.SESSIONS[sid] = session
monkeypatch.setattr(routes, "_resolve_chat_workspace_with_recovery", lambda *_args, **_kwargs: str(tmp_path))
monkeypatch.setattr(routes, "_read_profile_model_config", lambda *_args, **_kwargs: (None, None, {}))
monkeypatch.setattr(
routes,
"_resolve_compatible_session_model_state",
lambda requested_model, requested_provider, **_kwargs: (requested_model or "gpt-4o", requested_provider or "openai", False),
)
monkeypatch.setattr(routes, "webui_gateway_chat_enabled", lambda _config: False)
def _fake_start_run(run_session, **_kwargs):
assert compression_recovery_payload_for_session(run_session) is None
run_session.save()
return {"error": "session already has an active stream", "_status": 409}
monkeypatch.setattr(routes, "_start_run", _fake_start_run)
handler = _JSONHandler()
routes._handle_chat_start(handler, {"session_id": sid, "message": "continue by checking the repo"})
saved = json.loads((session_dir / f"{sid}.json").read_text(encoding="utf-8"))
assert handler.status == 409
assert saved["recommended_recovery_action"] == "start_focused_continuation"
assert saved["compression_recovery"]["terminal_state"] == "compression_exhausted"
def test_recovery_start_creates_focused_linked_session(monkeypatch, tmp_path):
session_dir = _isolate_sessions(monkeypatch, tmp_path)
sid = "recoverysrc1"
session = Session(
session_id=sid,
title="Long task",
workspace=str(tmp_path),
model="gpt-4o",
model_provider="openai",
profile="default",
project_id="proj_1",
messages=[{"role": "user", "content": "long task"}],
context_messages=[{"role": "user", "content": "large context"}],
)
stamp_compression_exhausted_recovery(session, message="Context length exceeded.")
session.save()
models.SESSIONS[sid] = session
routes.SESSIONS[sid] = session
handler = _JSONHandler()
routes._handle_session_compression_recovery_start(handler, {"session_id": sid})
payload = _payload(handler)
assert handler.status == 200
new_session = payload["session"]
assert new_session["session_id"] != sid
assert new_session["parent_session_id"] == sid
assert new_session["workspace"] == str(tmp_path)
assert new_session["model"] == "gpt-4o"
assert new_session["model_provider"] == "openai"
assert new_session["messages"] == []
saved = json.loads((session_dir / f"{new_session['session_id']}.json").read_text(encoding="utf-8"))
assert saved["parent_session_id"] == sid
assert saved["context_messages"] == []
assert compression_recovery_payload_for_session(session)["recommended_action"] == "start_focused_continuation"
def test_recovery_metadata_is_persisted_and_exposed_in_compact_session():
session = Session(session_id="recovermeta1", title="x", messages=[])
recovery = stamp_compression_exhausted_recovery(session, message="Context length exceeded.")
compact = session.compact()
assert recovery["terminal_state"] == "compression_exhausted"
assert compact["recommended_recovery_action"] == "start_focused_continuation"
assert compact["compression_recovery"]["recommended_action"] == "start_focused_continuation"
def test_compression_recovery_ui_wires_card_action_and_send_intercept():
ui = (ROOT / "static/ui.js").read_text(encoding="utf-8")
messages = (ROOT / "static/messages.js").read_text(encoding="utf-8")
assert "function _compressionRecoveryHtml" in ui
assert "data-compression-recovery-card=\"1\"" in ui
assert "api('/api/session/compression-recovery/start'" in ui
assert "function shouldInterceptCompressionRecoveryContinuation" in ui
assert "shouldInterceptCompressionRecoveryContinuation(text,S.pendingFiles)" in messages
assert "_compressionRecovery:recovery||undefined" in messages