fix(stream): rebuild the assistant body when a replay cursor outlives its live state
A transparent-mode SSE drop followed by a run-journal re-attach could settle a turn with its footer (duration / model / TTFT / Done) painted over a blank message body. The cached INFLIGHT entry kept `lastRunJournalSeq` (206 in the report) while its live assistant projection was gone; re-attach passed that cursor to the server as `after_seq`, so every journal event that rebuilds the message body was skipped and only the tail came back. A manual reload - a zero-floor replay - restored the message, which is what proves the server-side journal was complete the whole time. - `_inflightCanSeedJournalReplay()`: only recoverable live output (assistant content/reasoning, `lastAssistantText`/`lastReasoningText`, live DOM, tools, burst anchors, a populated anchor scene) authorises a nonzero replay floor. - `_runJournalReplayFloorForInflight()` / `_runJournalReplayEventIdForInflight()` gate BOTH halves of the cursor. The server reads `after_event_id` before `after_seq`, so zeroing the seq alone would still resume at the stale event id and skip the same range. - `_normalizeInflightReplayCursorForReattach()` re-validates the selected recovery object immediately before reattach (both `loadSession` reattach paths), so a cursor that outlived its projection cannot reach the wire. `_inflightHasVisibleLiveState()` keeps its keep-vs-discard role: a bare user row has to stay visible across a mid-turn reload until pending/live reconciliation runs, so replay authority lives in the floor predicate rather than overloading that contract. Fixes #7640
This commit is contained in:
1 parent
dd69bfd9a6
commit
d9440e8654
3 files changed
+320
-5
No files matched your search
+11
-4
@@ -2722,11 +2722,18 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){
|
||||
let _currentActivityBurstId=Number((INFLIGHT[activeSid]&&INFLIGHT[activeSid].currentActivityBurstId)||0)||0;
|
||||
let _currentLiveSegmentSeq=Number((INFLIGHT[activeSid]&&INFLIGHT[activeSid].currentLiveSegmentSeq)||0)||0;
|
||||
let _assistantSegmentSeq=Number((INFLIGHT[activeSid]&&INFLIGHT[activeSid].currentLiveSegmentSeq)||0)||0;
|
||||
let _lastRunJournalSeq=reconnecting
|
||||
? Number((INFLIGHT[activeSid]&&INFLIGHT[activeSid].lastRunJournalSeq)||0)
|
||||
// #7640: the replay floor is only as trustworthy as the recovery state behind
|
||||
// it. A cache that kept the cursor but lost the live assistant projection must
|
||||
// not raise `after_seq`: the server would then replay only the tail (often just
|
||||
// `stream_end`), and the missing journal range never gets a chance to rebuild
|
||||
// the body — the settled footer paints over a blank message until a reload.
|
||||
// Fall back to the zero floor whenever the state cannot be validated.
|
||||
const _replayCursorInflight=reconnecting?INFLIGHT[activeSid]:null;
|
||||
let _lastRunJournalSeq=(typeof _runJournalReplayFloorForInflight==='function')
|
||||
? _runJournalReplayFloorForInflight(_replayCursorInflight)
|
||||
: 0;
|
||||
let _lastRunJournalEventId=reconnecting
|
||||
? String((INFLIGHT[activeSid]&&INFLIGHT[activeSid].lastRunJournalEventId)||'')
|
||||
let _lastRunJournalEventId=(typeof _runJournalReplayEventIdForInflight==='function')
|
||||
? _runJournalReplayEventIdForInflight(_replayCursorInflight)
|
||||
: '';
|
||||
const _STREAM_FADE_MS=620;
|
||||
const _STREAM_FADE_MAX_MS=900;
|
||||
|
||||
@@ -1017,6 +1017,61 @@ function _inflightHasVisibleLiveState(inflight) {
|
||||
return false;
|
||||
}
|
||||
|
||||
// #7640: a run-journal replay cursor is only safe when the recovery object can
|
||||
// still repaint the live assistant turn. `_inflightHasVisibleLiveState()` above
|
||||
// deliberately counts a plain user row (the optimistic row has to survive a
|
||||
// mid-turn reload), but such an object carries no assistant projection at all:
|
||||
// seeding `after_seq` from it tells the server to skip every journal event that
|
||||
// would rebuild the body, so the settled footer lands over a blank message until
|
||||
// a manual reload replays the journal from the zero floor. Only recoverable live
|
||||
// output may raise the replay floor.
|
||||
function _inflightCanSeedJournalReplay(inflight){
|
||||
if(!inflight||typeof inflight!=='object') return false;
|
||||
if(String(inflight.lastAssistantText||'').trim()) return true;
|
||||
if(String(inflight.lastReasoningText||'').trim()) return true;
|
||||
if(String(inflight.liveTurnHtml||'').trim()) return true;
|
||||
if(Array.isArray(inflight.toolCalls)&&inflight.toolCalls.length) return true;
|
||||
if(Array.isArray(inflight.activityBurstAnchors)&&inflight.activityBurstAnchors.length) return true;
|
||||
const anchorScene=inflight.anchorActivityScene;
|
||||
if(anchorScene&&Array.isArray(anchorScene.activity_rows)&&anchorScene.activity_rows.length) return true;
|
||||
if(Array.isArray(inflight.messages)){
|
||||
return inflight.messages.some((msg)=>{
|
||||
if(!msg||msg.role!=='assistant') return false;
|
||||
const content=msg.content;
|
||||
if(typeof content==='string') return Boolean(content.trim());
|
||||
if(Array.isArray(content)) return content.length>0;
|
||||
return Boolean(content);
|
||||
});
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
function _runJournalReplayFloorForInflight(inflight){
|
||||
if(!_inflightCanSeedJournalReplay(inflight)) return 0;
|
||||
return Math.max(0,Number((inflight&&inflight.lastRunJournalSeq)||0)||0);
|
||||
}
|
||||
|
||||
function _runJournalReplayEventIdForInflight(inflight){
|
||||
if(!_runJournalReplayFloorForInflight(inflight)) return '';
|
||||
return String((inflight&&inflight.lastRunJournalEventId)||'');
|
||||
}
|
||||
|
||||
// #7640: re-validate the selected recovery object immediately before reattach
|
||||
// instead of trusting its existence. A cursor that outlived its assistant
|
||||
// projection is dropped here so it cannot reach the wire as a replay floor; the
|
||||
// replay then restarts from zero, which is the path a manual reload already
|
||||
// proves out. Normalising the object (not just the floor) also keeps a later
|
||||
// persist from resurrecting the stale cursor on the next reattach.
|
||||
function _normalizeInflightReplayCursorForReattach(inflight){
|
||||
if(!inflight||typeof inflight!=='object') return inflight;
|
||||
const seq=Math.max(0,Number(inflight.lastRunJournalSeq||0)||0);
|
||||
if(seq>0&&!_inflightCanSeedJournalReplay(inflight)){
|
||||
inflight.lastRunJournalSeq=0;
|
||||
inflight.lastRunJournalEventId='';
|
||||
}
|
||||
return inflight;
|
||||
}
|
||||
|
||||
function _serverLiveSnapshotToolId(tc){
|
||||
return String(tc&&(tc.tid||tc.id||tc.tool_call_id||tc.tool_use_id||tc.call_id||'')||'').trim();
|
||||
}
|
||||
@@ -2184,6 +2239,11 @@ async function loadSession(sid){
|
||||
if(INFLIGHT[sid].reattach&&activeStreamId&&typeof attachLiveStream==='function'){
|
||||
INFLIGHT[sid].reattach=false;
|
||||
if (!_isCurrentLoad()) return;
|
||||
// #7640: validate the selected recovery object at the moment of reattach
|
||||
// rather than trusting its existence. A cache that kept `lastRunJournalSeq`
|
||||
// but lost the live assistant projection would otherwise seed the replay
|
||||
// floor and skip the journal range that rebuilds the message body.
|
||||
_normalizeInflightReplayCursorForReattach(INFLIGHT[sid]);
|
||||
didReconnect=true;
|
||||
attachLiveStream(sid, activeStreamId, S.session.pending_attachments||[], {reconnecting:true});
|
||||
}
|
||||
@@ -2313,6 +2373,8 @@ async function loadSession(sid){
|
||||
if(activeStreamId){
|
||||
S.busy=true;
|
||||
S.activeStreamId=activeStreamId;
|
||||
// #7640: same reattach validation as the `reattach` branch above.
|
||||
if(INFLIGHT[sid]) _normalizeInflightReplayCursorForReattach(INFLIGHT[sid]);
|
||||
if(typeof attachLiveStream==='function') attachLiveStream(sid, activeStreamId, S.session.pending_attachments||[], {reconnecting:true});
|
||||
else if(typeof watchInflightSession==='function') watchInflightSession(sid, activeStreamId);
|
||||
updateSendBtn();
|
||||
|
||||
@@ -1060,6 +1060,249 @@ def test_load_session_discards_cursor_only_inflight_before_reattach():
|
||||
assert 0 <= guard_pos < inflight_branch_pos
|
||||
|
||||
|
||||
def test_user_only_inflight_must_not_seed_nonzero_replay_floor():
|
||||
"""#7640: a plain user row must not authorize a nonzero replay floor.
|
||||
|
||||
After a mid-stream SSE drop in transparent mode the client re-attaches with
|
||||
`replay=1&after_seq=<cursor>`. When the cached INFLIGHT entry kept the
|
||||
optimistic user row plus `lastRunJournalSeq` but lost the live assistant
|
||||
projection, that cursor told the server to skip every journal event that
|
||||
rebuilds the message body: only the tail (often just `stream_end`) came back
|
||||
and the settled footer painted over a blank message until a manual reload.
|
||||
"""
|
||||
assert NODE, "node not on PATH"
|
||||
script = "\n".join(
|
||||
[
|
||||
"const assert=require('assert');",
|
||||
_function_decl(SESSIONS_JS, "_inflightCanSeedJournalReplay"),
|
||||
_function_decl(SESSIONS_JS, "_runJournalReplayFloorForInflight"),
|
||||
_function_decl(SESSIONS_JS, "_runJournalReplayEventIdForInflight"),
|
||||
"""
|
||||
const userOnly={
|
||||
messages:[{role:'user', content:'a very long ttft prompt'}],
|
||||
lastRunJournalSeq:206,
|
||||
lastRunJournalEventId:'stream-e75e:206',
|
||||
};
|
||||
assert.strictEqual(_inflightCanSeedJournalReplay(userOnly), false);
|
||||
assert.strictEqual(_runJournalReplayFloorForInflight(userOnly), 0);
|
||||
assert.strictEqual(_runJournalReplayEventIdForInflight(userOnly), '');
|
||||
// The floor helper reports, it does not mutate the recovery object.
|
||||
assert.strictEqual(userOnly.lastRunJournalSeq, 206);
|
||||
|
||||
// A user row with a populated but assistant-less transcript is still user-only.
|
||||
assert.strictEqual(
|
||||
_runJournalReplayFloorForInflight({
|
||||
messages:[{role:'user', content:'hi'}, {role:'assistant', content:''}],
|
||||
lastRunJournalSeq:206,
|
||||
}),
|
||||
0
|
||||
);
|
||||
assert.strictEqual(
|
||||
_runJournalReplayFloorForInflight({
|
||||
messages:[{role:'user', content:'hi'}, {role:'assistant', content:' '}],
|
||||
lastRunJournalSeq:206,
|
||||
}),
|
||||
0
|
||||
);
|
||||
|
||||
// Positive inverse: any recoverable live output keeps the cursor.
|
||||
const recovering=[
|
||||
{lastAssistantText:'partial answer'},
|
||||
{lastReasoningText:'thinking...'},
|
||||
{liveTurnHtml:'<div>projected turn</div>'},
|
||||
{toolCalls:[{name:'read_file'}]},
|
||||
{activityBurstAnchors:[{seq:12}]},
|
||||
{anchorActivityScene:{activity_rows:[{kind:'tool'}]}},
|
||||
{messages:[{role:'user', content:'p'}, {role:'assistant', content:'answer'}]},
|
||||
];
|
||||
for (const extra of recovering) {
|
||||
const state={messages:[{role:'user', content:'p'}], ...extra};
|
||||
state.lastRunJournalSeq=206;
|
||||
state.lastRunJournalEventId='stream-e75e:206';
|
||||
assert.strictEqual(
|
||||
_runJournalReplayFloorForInflight(state),
|
||||
206,
|
||||
'recoverable live output must keep the cursor: '+JSON.stringify(extra)
|
||||
);
|
||||
assert.strictEqual(_runJournalReplayEventIdForInflight(state), 'stream-e75e:206');
|
||||
}
|
||||
|
||||
// Degenerate inputs must stay on the zero floor.
|
||||
assert.strictEqual(_runJournalReplayFloorForInflight(null), 0);
|
||||
assert.strictEqual(_runJournalReplayFloorForInflight({}), 0);
|
||||
assert.strictEqual(_runJournalReplayFloorForInflight({lastRunJournalSeq:206, messages:[]}), 0);
|
||||
assert.strictEqual(
|
||||
_runJournalReplayFloorForInflight({lastAssistantText:'partial', lastRunJournalSeq:-4}),
|
||||
0
|
||||
);
|
||||
assert.strictEqual(_runJournalReplayEventIdForInflight({}), '');
|
||||
""",
|
||||
]
|
||||
)
|
||||
result = subprocess.run([NODE, "-e", script], capture_output=True, text=True, check=False)
|
||||
assert result.returncode == 0, result.stderr
|
||||
|
||||
|
||||
def test_reattach_discards_replay_cursor_without_live_assistant_state():
|
||||
"""#7640: validate the recovery object immediately before reattach.
|
||||
|
||||
Trusting the object's existence (rather than its contents) is what let a
|
||||
cursor-only cache reach the wire as a replay floor. Reattach now drops the
|
||||
cursor so the journal replays from zero, which is the path a manual reload
|
||||
already proves out, while a cache that still holds live assistant output
|
||||
keeps its cursor and avoids re-rendering the whole journal.
|
||||
"""
|
||||
assert NODE, "node not on PATH"
|
||||
script = "\n".join(
|
||||
[
|
||||
"const assert=require('assert');",
|
||||
_function_decl(SESSIONS_JS, "_inflightCanSeedJournalReplay"),
|
||||
_function_decl(SESSIONS_JS, "_runJournalReplayFloorForInflight"),
|
||||
_function_decl(SESSIONS_JS, "_normalizeInflightReplayCursorForReattach"),
|
||||
"""
|
||||
const cursorOnly={
|
||||
streamId:'stream-e75e',
|
||||
reattach:true,
|
||||
messages:[{role:'user', content:'a very long ttft prompt'}],
|
||||
lastRunJournalSeq:206,
|
||||
lastRunJournalEventId:'stream-e75e:206',
|
||||
};
|
||||
assert.strictEqual(
|
||||
_normalizeInflightReplayCursorForReattach(cursorOnly),
|
||||
cursorOnly,
|
||||
'must return the same object so the caller keeps mutating the live cache'
|
||||
);
|
||||
assert.strictEqual(cursorOnly.lastRunJournalSeq, 0);
|
||||
assert.strictEqual(cursorOnly.lastRunJournalEventId, '');
|
||||
assert.strictEqual(_runJournalReplayFloorForInflight(cursorOnly), 0);
|
||||
assert.strictEqual(
|
||||
cursorOnly.messages.length,
|
||||
1,
|
||||
'the optimistic user row must survive so pending dedupe still works'
|
||||
);
|
||||
|
||||
const recovering={
|
||||
streamId:'stream-e75e',
|
||||
reattach:true,
|
||||
messages:[{role:'user', content:'p'}],
|
||||
lastAssistantText:'partial answer',
|
||||
lastRunJournalSeq:206,
|
||||
lastRunJournalEventId:'stream-e75e:206',
|
||||
};
|
||||
_normalizeInflightReplayCursorForReattach(recovering);
|
||||
assert.strictEqual(recovering.lastRunJournalSeq, 206);
|
||||
assert.strictEqual(recovering.lastRunJournalEventId, 'stream-e75e:206');
|
||||
|
||||
const noCursor={streamId:'stream-e75e', messages:[{role:'user', content:'p'}]};
|
||||
_normalizeInflightReplayCursorForReattach(noCursor);
|
||||
assert.strictEqual(noCursor.lastRunJournalSeq, undefined);
|
||||
assert.strictEqual(_runJournalReplayFloorForInflight(noCursor), 0);
|
||||
|
||||
assert.strictEqual(_normalizeInflightReplayCursorForReattach(null), null);
|
||||
assert.strictEqual(_normalizeInflightReplayCursorForReattach(undefined), undefined);
|
||||
""",
|
||||
]
|
||||
)
|
||||
result = subprocess.run([NODE, "-e", script], capture_output=True, text=True, check=False)
|
||||
assert result.returncode == 0, result.stderr
|
||||
|
||||
|
||||
def test_reattach_replay_params_from_a_cursor_only_cache_replay_from_zero():
|
||||
"""#7640: end-to-end wire contract for a cursor-only reattach.
|
||||
|
||||
Derives the replay params from the real floor helpers (node) and feeds them
|
||||
to the real server cursor parser. Tying both halves together matters because
|
||||
`api.routes._parse_run_journal_after_seq()` reads `after_event_id` BEFORE
|
||||
`after_seq`: a fix that zeroed only the seq would still resume at the stale
|
||||
event id and skip every journal event that rebuilds the message body.
|
||||
"""
|
||||
assert NODE, "node not on PATH"
|
||||
script = "\n".join(
|
||||
[
|
||||
"const assert=require('assert');",
|
||||
_function_decl(SESSIONS_JS, "_inflightCanSeedJournalReplay"),
|
||||
_function_decl(SESSIONS_JS, "_runJournalReplayFloorForInflight"),
|
||||
_function_decl(SESSIONS_JS, "_runJournalReplayEventIdForInflight"),
|
||||
"""
|
||||
const runId='stream-7640';
|
||||
const userOnly={
|
||||
messages:[{role:'user', content:'a very long ttft prompt'}],
|
||||
lastRunJournalSeq:206,
|
||||
lastRunJournalEventId:runId+':205',
|
||||
};
|
||||
console.log(JSON.stringify({
|
||||
after_seq:String(_runJournalReplayFloorForInflight(userOnly)),
|
||||
after_event_id:String(_runJournalReplayEventIdForInflight(userOnly)||''),
|
||||
}));
|
||||
""",
|
||||
]
|
||||
)
|
||||
result = subprocess.run([NODE, "-e", script], capture_output=True, text=True, check=False)
|
||||
assert result.returncode == 0, result.stderr
|
||||
params = json.loads(result.stdout.strip().splitlines()[-1])
|
||||
|
||||
# What the fixed client puts on the wire for a cache with no live projection.
|
||||
assert params == {"after_seq": "0", "after_event_id": ""}
|
||||
|
||||
from api.routes import _parse_run_journal_after_seq
|
||||
|
||||
run_id = "stream-7640"
|
||||
assert (
|
||||
_parse_run_journal_after_seq(
|
||||
{"after_seq": [params["after_seq"]], "after_event_id": [params["after_event_id"]]},
|
||||
run_id,
|
||||
)
|
||||
== 0
|
||||
)
|
||||
|
||||
# Precedence proof: dropping the seq alone would still resume at the stale id.
|
||||
assert (
|
||||
_parse_run_journal_after_seq(
|
||||
{"after_seq": ["0"], "after_event_id": [f"{run_id}:205"]}, run_id
|
||||
)
|
||||
== 205
|
||||
)
|
||||
# ...and the pre-fix pair, for the record.
|
||||
assert (
|
||||
_parse_run_journal_after_seq(
|
||||
{"after_seq": ["206"], "after_event_id": [f"{run_id}:205"]}, run_id
|
||||
)
|
||||
== 205
|
||||
)
|
||||
|
||||
|
||||
def test_attach_and_reattach_use_the_validated_replay_floor():
|
||||
"""#7640 wiring guard: the validated floor is what reaches the wire.
|
||||
|
||||
The behavioral coverage for the floor helpers lives in
|
||||
`test_user_only_inflight_must_not_seed_nonzero_replay_floor` and
|
||||
`test_reattach_discards_replay_cursor_without_live_assistant_state`; this
|
||||
test pins that `attachLiveStream()` and both `loadSession()` reattach paths
|
||||
actually go through them instead of reading the raw cursor field.
|
||||
"""
|
||||
body = _function_body(MESSAGES_JS, "attachLiveStream")
|
||||
init_pos = body.index("const _replayCursorInflight=")
|
||||
init_block = body[init_pos : body.index(";", body.index("let _lastRunJournalEventId=", init_pos))]
|
||||
assert "INFLIGHT[activeSid]" in init_block, "must still read the live cache"
|
||||
assert "_runJournalReplayFloorForInflight(_replayCursorInflight)" in init_block
|
||||
assert "_runJournalReplayEventIdForInflight(_replayCursorInflight)" in init_block
|
||||
assert "INFLIGHT[activeSid].lastRunJournalSeq" not in init_block, (
|
||||
"attachLiveStream must not read the raw cursor field anymore"
|
||||
)
|
||||
assert "INFLIGHT[activeSid].lastRunJournalEventId" not in init_block
|
||||
|
||||
load_body = re.sub(r"\s+", "", _function_body(SESSIONS_JS, "loadSession"))
|
||||
assert load_body.count("_normalizeInflightReplayCursorForReattach(INFLIGHT[sid])") == 2, (
|
||||
"both reattach paths must re-validate the recovery object"
|
||||
)
|
||||
first = load_body.index("_normalizeInflightReplayCursorForReattach(INFLIGHT[sid])")
|
||||
first_attach = load_body.index("attachLiveStream(sid,activeStreamId", first)
|
||||
assert first < first_attach, "validation must run before the stream is attached"
|
||||
last = load_body.rindex("_normalizeInflightReplayCursorForReattach(INFLIGHT[sid])")
|
||||
last_attach = load_body.rindex("attachLiveStream(sid,activeStreamId", 0, last)
|
||||
assert last_attach < last
|
||||
|
||||
|
||||
def test_live_recovery_prefers_newer_durable_run_journal_snapshot():
|
||||
"""Recovery chooses by stream identity and durable journal progress."""
|
||||
assert NODE, "node not on PATH"
|
||||
@@ -1174,7 +1417,10 @@ assert.strictEqual(inflight.lastRunJournalEventId, 'run-a:7');
|
||||
assert "lastRunJournalEventId:String(stored.lastRunJournalEventId||'')" in load_body
|
||||
assert "lastRunJournalEventId:state.lastRunJournalEventId||''" in compact_body
|
||||
assert "inflight.lastRunJournalEventId||''" in attach_body
|
||||
assert "INFLIGHT[activeSid]&&INFLIGHT[activeSid].lastRunJournalEventId" in attach_body
|
||||
# #7640: the cursor's event id is gated by the same validation as the seq —
|
||||
# reading the raw field here let a cursor outlive its assistant projection.
|
||||
assert "_runJournalReplayEventIdForInflight(_replayCursorInflight)" in attach_body
|
||||
assert "INFLIGHT[activeSid]&&INFLIGHT[activeSid].lastRunJournalEventId" not in attach_body
|
||||
assert "inflight.lastRunJournalEventId=raw" in attach_body
|
||||
assert "INFLIGHT[activeSid].streamId=streamId" in attach_body
|
||||
assert "INFLIGHT[activeSid].lastRunJournalEventId=''" in attach_body
|
||||
|
||||
Reference in new issue
Block a user