fix: resolve durable compressed sessions before WebUI resume
This commit is contained in:
1 parent
034e3edef7
commit
1ee4fc1601
6 files changed
+256
No files matched your search
@@ -89,6 +89,12 @@ Follow that checklist's safety rules:
|
||||
by publishing finalizing under STREAMS_LOCK before the last pending-steer
|
||||
drain; earlier accepted guidance is drained, later guidance is rejected.
|
||||
Test both drain/Steer orderings, including compression-rotated identities.
|
||||
- Inactive-session recovery is separate from live Steer. Resolve durable
|
||||
compression lineage in the session's profile database, read-only, even when
|
||||
the WebUI sidecar has no snapshot flag. Never reopen a sealed parent. Reject
|
||||
stale chat POSTs before workspace/model/pending-state mutation; the browser
|
||||
loads the continuation and preserves the draft without automatic replay.
|
||||
Explicit closures and unknown terminal reasons do not authorize a redirect.
|
||||
- For Docker build changes in `docker_init.bash`, mirror directory exclusions
|
||||
in both the `rsync` and `cp -a` paths — `/opt/hermes` may contain subdirectories
|
||||
with restricted permissions (e.g. `.playwright/`).
|
||||
|
||||
@@ -0,0 +1,60 @@
|
||||
"""Read-only routing hints for externally compressed WebUI sessions.
|
||||
|
||||
SQLite owns compression lineage; sidecar snapshot flags may predate a
|
||||
Desktop/CLI rotation. Never reopen a sealed parent or mutate Agent state here.
|
||||
"""
|
||||
import logging
|
||||
from pathlib import Path
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def durable_compression_continuation(session):
|
||||
"""Return (sealed, resumable tip), without making a recovery write.
|
||||
|
||||
A known sealed parent without a safe tip stays sealed (no sidecar fallback).
|
||||
Older Agent installations without this read API retain legacy behavior.
|
||||
"""
|
||||
from api.profiles import _PROFILE_ID_RE, _resolve_profile_home_for_name
|
||||
|
||||
sid = str(getattr(session, "session_id", "") or "")
|
||||
profile = str(getattr(session, "profile", None) or "default")
|
||||
if not sid or (profile != "default" and not _PROFILE_ID_RE.fullmatch(profile)):
|
||||
return False, None
|
||||
db = None
|
||||
sealed = False
|
||||
try:
|
||||
from hermes_state import SessionDB
|
||||
|
||||
path = Path(_resolve_profile_home_for_name(profile)) / "state.db"
|
||||
if not path.is_file():
|
||||
return False, None
|
||||
db = SessionDB(path, read_only=True)
|
||||
parent = db.get_session(sid)
|
||||
if not parent or parent.get("end_reason") != "compression":
|
||||
return False, None
|
||||
sealed = True
|
||||
tip = db.get_compression_tip(sid)
|
||||
if not tip or tip == sid:
|
||||
return True, None
|
||||
child = db.get_session(tip)
|
||||
if not child or child.get("source") not in {"webui", "tui", "cli"}:
|
||||
return True, None
|
||||
for row in (parent, child):
|
||||
if row.get("profile_name") not in (None, "", profile):
|
||||
return True, None
|
||||
# Only the observed automatic idle closure is resumable here. Explicit
|
||||
# resets/closures and unknown future reasons must not become redirects.
|
||||
if child.get("ended_at") is not None and child.get("end_reason") != "idle_timeout":
|
||||
return True, None
|
||||
if db.get_compression_tip(sid) != tip:
|
||||
return True, None
|
||||
return True, tip
|
||||
except (ImportError, AttributeError, TypeError):
|
||||
return sealed, None
|
||||
except Exception:
|
||||
logger.debug("Could not resolve durable compression continuation", exc_info=True)
|
||||
return sealed, None
|
||||
finally:
|
||||
if db is not None:
|
||||
db.close()
|
||||
@@ -10488,6 +10488,11 @@ def _pre_compression_continuation_session_id(session) -> str | None:
|
||||
exists either in memory or on disk. Follow bounded snapshot-to-snapshot hops
|
||||
so repeated compression still lands on the latest visible continuation.
|
||||
"""
|
||||
from api.compression_continuation import durable_compression_continuation
|
||||
|
||||
sealed, tip = durable_compression_continuation(session)
|
||||
if sealed:
|
||||
return tip
|
||||
if not getattr(session, "pre_compression_snapshot", False):
|
||||
return None
|
||||
sid = _safe_first(getattr(session, "session_id", None))
|
||||
@@ -24355,6 +24360,16 @@ def _handle_chat_start(handler, body, diag=None):
|
||||
s.profile = requested_profile
|
||||
else:
|
||||
return bad(handler, "Session not found", 404)
|
||||
# Resolve durable rotations before any workspace/model/pending mutation.
|
||||
# GET navigation adopts the tip; POST never silently replays a user turn.
|
||||
from api.compression_continuation import durable_compression_continuation
|
||||
sealed, continuation = durable_compression_continuation(s)
|
||||
if sealed:
|
||||
return j(handler, {
|
||||
"error": "This session was compressed. Open its continuation before sending.",
|
||||
"code": "session_rotated",
|
||||
"continuation_session_id": continuation,
|
||||
}, status=409)
|
||||
regeneration = None
|
||||
if body.get("regenerate") is True:
|
||||
if any(key in body for key in ("message", "attachments", "keep_count", "prompt", "prompt_index")):
|
||||
|
||||
@@ -30,6 +30,25 @@ This RFC defines a consistency contract for those layers. It complements the
|
||||
larger run adapter direction in #1925 by documenting what must remain coherent
|
||||
while WebUI still has multiple overlapping state stores.
|
||||
|
||||
## Inactive compression continuation recovery
|
||||
|
||||
The Agent profile's SQLite compression lineage owns the canonical continuation,
|
||||
including when Desktop/CLI compressed a session without updating WebUI's
|
||||
`pre_compression_snapshot` sidecar flag. `GET /api/session` may expose the
|
||||
existing `continuation_session_id` hint from that read-only lineage. Automatic
|
||||
`idle_timeout` closure does not hide the continuation; explicit/unknown terminal
|
||||
reasons, foreign-profile rows and delegated/tool children do not authorize it.
|
||||
This read does not reopen sessions or copy ancestor display history into context.
|
||||
|
||||
A stale `POST /api/chat/start` returns HTTP 409 with `code=session_rotated`
|
||||
and the continuation hint before workspace, model, pending-turn or worker
|
||||
mutation. The browser loads the continuation through normal session access
|
||||
checks and restores the rejected text and attachments as a draft. The user
|
||||
sends again explicitly; there is no automatic POST replay or migration of the
|
||||
parent's workspace binding. Clients without this handling must reload the
|
||||
session before retrying. Server wakeups, regeneration semantics and Gateway
|
||||
routing are not silently retargeted by this recovery path.
|
||||
|
||||
## Goals
|
||||
|
||||
- Define the state layers involved in active and recovered WebUI turns.
|
||||
|
||||
@@ -1296,6 +1296,31 @@ function applySessionTitleUpdate(sid, titleText, options={}){
|
||||
// BEFORE slash rewrites (/moa, bundles) mutate the payload and BEFORE
|
||||
// uploadPendingFiles() drains S.pendingFiles — so we restore what the user
|
||||
// actually typed, not the transformed send payload.
|
||||
async function _recoverCompressedSend(error,sid,draftText,filesSnapshot,clearPromise){
|
||||
let payload;
|
||||
try{ payload=JSON.parse(error&&error.body||'{}'); }catch(_){ return false; }
|
||||
const target=payload&&payload.continuation_session_id;
|
||||
if(!error||error.status!==409||!payload||payload.code!=='session_rotated'||typeof target!=='string'||!target||target===sid) return false;
|
||||
// A failed POST has not admitted a turn. Never resend automatically: the
|
||||
// continuation may already be busy, and attachments must remain a draft.
|
||||
if(!S.session||S.session.session_id!==sid) return false;
|
||||
delete INFLIGHT[sid];
|
||||
if(typeof clearInflightState==='function') clearInflightState(sid);
|
||||
if(typeof clearOptimisticSessionStreaming==='function') clearOptimisticSessionStreaming(sid);
|
||||
stopApprovalPolling();stopClarifyPolling();removeThinking();setBusy(false);
|
||||
try{
|
||||
await loadSession(target);
|
||||
if(!S.session||S.session.session_id===sid) return false;
|
||||
// loadSession can lose its navigation race to another tab selection. Never
|
||||
// place the rejected message into that unrelated session's composer.
|
||||
_restoreComposerDraftAfterFailedSend(draftText,filesSnapshot,target,clearPromise);
|
||||
if(S.session.session_id!==target) return true;
|
||||
setComposerStatus('Session resumed. Your message is preserved; send it when ready.');
|
||||
showToast('Session resumed after compression. Your draft is preserved.',4000);
|
||||
return true;
|
||||
}catch(_){ return false; }
|
||||
}
|
||||
|
||||
function _restoreComposerDraftAfterFailedSend(draftText, filesSnapshot, sid, clearPromise){
|
||||
const restore=String(draftText||'');
|
||||
const files=Array.isArray(filesSnapshot)?filesSnapshot.filter(Boolean):[];
|
||||
@@ -1850,6 +1875,7 @@ async function send(){
|
||||
if(typeof renderSessionList==='function') void renderSessionList();
|
||||
return;
|
||||
}
|
||||
if(await _recoverCompressedSend(e,activeSid,_failedSendDraftText,_failedSendFilesSnapshot,_composerDraftClearPromise)) return;
|
||||
const conflictActiveStream=/session already has an active stream/i.test(errMsg);
|
||||
if(conflictActiveStream){
|
||||
delete INFLIGHT[activeSid];
|
||||
|
||||
@@ -0,0 +1,130 @@
|
||||
"""A stale WebUI sidecar must not hide a durable compression continuation."""
|
||||
import io
|
||||
import json
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
from api import routes, profiles
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def lineage(tmp_path, monkeypatch):
|
||||
SessionDB = pytest.importorskip("hermes_state").SessionDB
|
||||
db = SessionDB(tmp_path / "state.db")
|
||||
db.create_session("sealedparent", source="webui")
|
||||
db.create_session("idlechild", source="tui", parent_session_id="sealedparent")
|
||||
db.append_message("idlechild", "user", "retained task")
|
||||
db.end_session("sealedparent", "compression")
|
||||
db.end_session("idlechild", "idle_timeout")
|
||||
monkeypatch.setattr(profiles, "_resolve_profile_home_for_name", lambda name: str(tmp_path))
|
||||
session = SimpleNamespace(session_id="sealedparent", profile="default", pre_compression_snapshot=False)
|
||||
yield db, session
|
||||
db.close()
|
||||
|
||||
|
||||
def test_reload_resolves_sqlite_rotation_without_sidecar_flag(lineage):
|
||||
db, session = lineage
|
||||
before = db.get_session("idlechild")["ended_at"]
|
||||
assert routes._pre_compression_continuation_session_id(session) == "idlechild"
|
||||
assert db.get_session("idlechild")["ended_at"] == before
|
||||
assert db.get_session("sealedparent")["end_reason"] == "compression"
|
||||
|
||||
|
||||
class Handler:
|
||||
headers = {}
|
||||
def __init__(self):
|
||||
self.wfile = io.BytesIO()
|
||||
def send_response(self, status):
|
||||
self.status = status
|
||||
def send_header(self, *args):
|
||||
pass
|
||||
def end_headers(self):
|
||||
pass
|
||||
|
||||
|
||||
def test_stale_post_rejected_before_workspace_or_worker_mutation(lineage, monkeypatch):
|
||||
_, session = lineage
|
||||
monkeypatch.setattr(routes, "_agent_runtime_barrier_response", lambda **kw: None)
|
||||
monkeypatch.setattr(routes, "_get_or_materialize_session", lambda *a, **kw: session)
|
||||
monkeypatch.setattr(routes, "_get_active_profile_name", lambda: "default")
|
||||
monkeypatch.setattr(routes, "_session_visible_to_active_profile", lambda *a: True)
|
||||
monkeypatch.setattr(routes, "_resolve_chat_workspace_with_recovery", lambda *a: pytest.fail("stale turn must not reach workspace mutation"))
|
||||
h = Handler()
|
||||
routes._handle_chat_start(h, {"session_id": session.session_id, "message": "conclusion?"})
|
||||
assert h.status == 409
|
||||
payload = json.loads(h.wfile.getvalue())
|
||||
assert payload["code"] == "session_rotated"
|
||||
assert payload["continuation_session_id"] == "idlechild"
|
||||
|
||||
|
||||
def test_idle_tip_accepts_normal_persistence_without_reopening_parent(lineage):
|
||||
db, session = lineage
|
||||
tip = routes._pre_compression_continuation_session_id(session)
|
||||
db.append_message(tip, "user", "resume task")
|
||||
assert db.get_session("sealedparent")["end_reason"] == "compression"
|
||||
assert db.get_messages(tip)[-1]["content"] == "resume task"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("reason", ["reset", "user_closed", "compression"])
|
||||
def test_unknown_or_sealed_terminal_tip_is_not_redirected(lineage, reason):
|
||||
db, session = lineage
|
||||
db._conn.execute("UPDATE sessions SET end_reason=? WHERE id='idlechild'", (reason,))
|
||||
db._conn.commit()
|
||||
assert routes._pre_compression_continuation_session_id(session) is None
|
||||
|
||||
|
||||
def test_newer_branch_does_not_replace_canonical_continuation(lineage):
|
||||
db, session = lineage
|
||||
db.create_session("branchchild", source="webui", parent_session_id="sealedparent")
|
||||
db._conn.execute("UPDATE sessions SET message_count=20, model_config=? WHERE id='branchchild'", (json.dumps({'_branched_from': 'sealedparent'}),))
|
||||
db._conn.commit()
|
||||
assert routes._pre_compression_continuation_session_id(session) == "idlechild"
|
||||
|
||||
|
||||
def test_foreign_profile_and_subagent_are_not_redirected(lineage):
|
||||
db, session = lineage
|
||||
db._conn.execute("UPDATE sessions SET profile_name='foreign' WHERE id='idlechild'")
|
||||
db._conn.commit()
|
||||
assert routes._pre_compression_continuation_session_id(session) is None
|
||||
db._conn.execute("UPDATE sessions SET profile_name=NULL, source='subagent' WHERE id='idlechild'")
|
||||
db._conn.commit()
|
||||
assert routes._pre_compression_continuation_session_id(session) is None
|
||||
|
||||
|
||||
def test_noncompressed_parent_and_absent_database_keep_legacy_behavior(lineage, monkeypatch, tmp_path):
|
||||
db, session = lineage
|
||||
db._conn.execute("UPDATE sessions SET end_reason='idle_timeout' WHERE id='sealedparent'")
|
||||
db._conn.commit()
|
||||
assert routes._pre_compression_continuation_session_id(session) is None
|
||||
monkeypatch.setattr(profiles, "_resolve_profile_home_for_name", lambda name: str(tmp_path / 'missing'))
|
||||
assert routes._pre_compression_continuation_session_id(session) is None
|
||||
assert not (tmp_path / 'missing').exists()
|
||||
|
||||
|
||||
def test_browser_rotation_restores_draft_without_reposting():
|
||||
source = (Path(__file__).parents[1] / 'static/messages.js').read_text()
|
||||
helper = source[source.index('async function _recoverCompressedSend('):source.index('function _restoreComposerDraftAfterFailedSend(')]
|
||||
script = r'''
|
||||
const assert = require('node:assert/strict');
|
||||
let S={session:{session_id:'old'}}, INFLIGHT={old:{}};
|
||||
const calls=[];
|
||||
const stopApprovalPolling=()=>{},stopClarifyPolling=()=>{},removeThinking=()=>{},setBusy=()=>{},setComposerStatus=()=>{},showToast=()=>{};
|
||||
const loadSession=async sid=>{calls.push(['load',sid]);S.session={session_id:sid}};
|
||||
const _restoreComposerDraftAfterFailedSend=(...args)=>calls.push(['restore',...args]);
|
||||
const api=()=>{throw Error('must not repost')};
|
||||
(async()=>{
|
||||
const err={status:409,body:JSON.stringify({code:'session_rotated',continuation_session_id:'new'})};
|
||||
const files=[{name:'drawing.png'}], promise=Promise.resolve();
|
||||
assert.equal(await _recoverCompressedSend(err,'old','conclusion?',files,promise),true);
|
||||
assert.deepEqual(calls[0],['load','new']);
|
||||
assert.deepEqual(calls[1],['restore','conclusion?',files,'new',promise]);
|
||||
assert.equal(INFLIGHT.old,undefined);
|
||||
assert.equal(await _recoverCompressedSend({status:500},'new','x',[],promise),false);
|
||||
assert.equal(await _recoverCompressedSend(err,'old','x',[],promise),false);
|
||||
assert.equal(calls.length,2);
|
||||
})().catch(e=>{console.error(e);process.exit(1)});
|
||||
'''
|
||||
subprocess.run(['node', '-e', helper + script], check=True, timeout=15)
|
||||
Reference in new issue
Block a user