Merge pull request #3949 from nesquena/release/stage-3427

Release LQ — v0.51.353 (#3427 cross-client live-turn recovery)
This commit is contained in:
nesquena-hermes 2026-06-10 13:27:09 -07:00 committed by GitHub
commit a212fcd330
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 468 additions and 6 deletions

View file

@ -7,6 +7,12 @@
- **New RFC: Stable Assistant Turn Anchors for Live-to-Final rendering.** Defines a frontend presentation/reconciliation model for anchoring one assistant turn across live streaming, settlement, replay/reload/recovery, Compact Worklog, Transparent Stream, terminal states, artifacts, and side effects. (#3926)
## [v0.51.353] — 2026-06-10 — Release LQ (cross-client live-turn recovery)
### Fixed
- **Opening an in-progress session from a fresh client now shows the already-streamed text and tool activity.** Switching away from an active session and returning from another device — or a tab with no in-memory snapshot — could previously show an empty or stalled turn until the next token arrived, because the already-emitted progress lived only in browser-local memory. The server now builds a live snapshot from the run journal (`runtime_journal_snapshot` in `GET /api/session`) so a fresh client rebuilds the visible assistant text + tool cards from the server. It composes with the existing run-journal replay cursor (seeds the replay-from sequence so events aren't duplicated) and keys tool cards by the same id aliases as the live path so SSE replay replaces rather than duplicates the restored cards. (#3427, part of the #3400 live-to-final epic)
## [v0.51.352] — 2026-06-10 — Release LP (medium round: jump-to-response, STATE_DIR warning, J/K nav)
### Added

View file

@ -433,7 +433,8 @@ def _redact_value(v, *, _enabled: bool | None = None):
def redact_session_data(session_dict: dict) -> dict:
"""Redact credentials from message content, tool data, and session sidecars.
Applies to: messages[], tool_calls[], todo_state, and title.
Applies to: messages[], tool_calls[], todo_state, runtime_journal_snapshot,
and title.
The underlying session file is not modified; redaction is response-layer only.
Reads the ``api_redact_enabled`` setting ONCE for the entire response and
@ -453,6 +454,11 @@ def redact_session_data(session_dict: dict) -> dict:
result['tool_calls'] = _redact_value(result['tool_calls'], _enabled=_enabled)
if 'todo_state' in result:
result['todo_state'] = _redact_value(result['todo_state'], _enabled=_enabled)
if 'runtime_journal_snapshot' in result:
result['runtime_journal_snapshot'] = _redact_value(
result['runtime_journal_snapshot'],
_enabled=_enabled,
)
return result

View file

@ -1330,6 +1330,222 @@ def _run_journal_status_payload(summary: dict, *, active: bool = False) -> dict:
}
_RUN_JOURNAL_TOOL_ID_KEYS = ("tid", "id", "tool_call_id", "tool_use_id", "call_id")
def _run_journal_snapshot_tool_id(payload: dict | None) -> str:
if not isinstance(payload, dict):
return ""
for key in _RUN_JOURNAL_TOOL_ID_KEYS:
value = str(payload.get(key) or "").strip()
if value:
return value
return ""
def _truncate_journal_snapshot_value(value, *, limit: int = 120):
if isinstance(value, str):
return value if len(value) <= limit else value[:limit] + "..."
if isinstance(value, dict):
return {str(k): _truncate_journal_snapshot_value(v, limit=limit) for k, v in value.items()}
if isinstance(value, list):
return [_truncate_journal_snapshot_value(v, limit=limit) for v in value[:20]]
return value
def _run_journal_live_snapshot(stream_id: str | None) -> dict | None:
stream_id = str(stream_id or "").strip()
if not stream_id:
return None
summary = find_run_summary(stream_id)
if not summary:
return None
session_id = str(summary.get("session_id") or "")
if not session_id:
return None
journal = read_run_events(session_id, stream_id)
events = [event for event in (journal.get("events") or []) if isinstance(event, dict)]
if not events:
return None
assistant_text = ""
reasoning_text = ""
messages: list[dict] = []
tool_calls: list[dict] = []
activity_burst_anchors: list[dict] = []
current_activity_burst_id = 0
fresh_segment = True
last_ts = None
def mark_boundary() -> int:
nonlocal current_activity_burst_id
text_end = len(assistant_text)
if text_end <= 0:
return current_activity_burst_id
last_end = max(
[int(anchor.get("textEnd") or 0) for anchor in activity_burst_anchors]
or [0]
)
if text_end > last_end:
current_activity_burst_id += 1
activity_burst_anchors.append(
{"id": current_activity_burst_id, "textEnd": text_end}
)
return current_activity_burst_id
def update_completed_tool(payload: dict) -> None:
tool_id = _run_journal_snapshot_tool_id(payload)
name = str(payload.get("name") or "").strip()
for call in reversed(tool_calls):
if call.get("done"):
continue
call_id = _run_journal_snapshot_tool_id(call)
if (tool_id and call_id == tool_id) or (not tool_id and name and call.get("name") == name):
call["done"] = True
if payload.get("preview") is not None:
call["snippet"] = str(payload.get("preview") or "")
call["preview"] = call.get("preview") or call["snippet"]
if payload.get("duration") is not None:
call["duration"] = payload.get("duration")
if payload.get("is_error") is not None:
call["is_error"] = bool(payload.get("is_error"))
return
if not name or name == "clarify":
return
call = {
"name": name,
"preview": str(payload.get("preview") or ""),
"snippet": str(payload.get("preview") or ""),
"args": _truncate_journal_snapshot_value(payload.get("args") or {}),
"done": True,
"_live": True,
"_journal_snapshot": True,
"_journal_stream_id": stream_id,
}
tool_id = _run_journal_snapshot_tool_id(payload)
if tool_id:
call["tid"] = tool_id
for key in _RUN_JOURNAL_TOOL_ID_KEYS:
if payload.get(key):
call[key] = str(payload.get(key))
if current_activity_burst_id:
call["activityBurstId"] = current_activity_burst_id
call["activitySegmentSeq"] = current_activity_burst_id
tool_calls.append(call)
for event in events:
event_name = str(event.get("event") or event.get("type") or "")
payload = event.get("payload") if isinstance(event.get("payload"), dict) else {}
last_ts = event.get("created_at", last_ts)
if event_name == "token":
text = str(payload.get("text") or "")
if text:
assistant_text += text
fresh_segment = False
continue
if event_name == "reasoning":
reasoning_text += str(payload.get("text") or "")
continue
if event_name == "interim_assistant":
visible = str(payload.get("text") or "").strip()
if visible:
if payload.get("already_streamed"):
if not assistant_text:
assistant_text = visible
else:
assistant_text = f"{assistant_text}\n\n{visible}" if assistant_text else visible
mark_boundary()
fresh_segment = True
continue
if event_name == "tool":
name = str(payload.get("name") or "").strip()
if not name or name == "clarify":
continue
boundary_id = mark_boundary()
tool_id = _run_journal_snapshot_tool_id(payload)
call = {
"name": name,
"preview": str(payload.get("preview") or ""),
"args": _truncate_journal_snapshot_value(payload.get("args") or {}),
"done": False,
"_live": True,
"_journal_snapshot": True,
"_journal_stream_id": stream_id,
}
if tool_id:
call["tid"] = tool_id
for key in _RUN_JOURNAL_TOOL_ID_KEYS:
if payload.get(key):
call[key] = str(payload.get(key))
if boundary_id:
call["activityBurstId"] = boundary_id
call["activitySegmentSeq"] = boundary_id
tool_calls.append(call)
fresh_segment = True
continue
if event_name == "tool_complete":
update_completed_tool(payload)
fresh_segment = True
if assistant_text or reasoning_text:
message = {
"role": "assistant",
"content": assistant_text,
"_live": True,
"_journal_snapshot": True,
"_journal_stream_id": stream_id,
}
if reasoning_text:
message["reasoning"] = reasoning_text
if last_ts is not None:
message["_ts"] = last_ts
messages.append(message)
if not messages and not tool_calls:
return None
visible_anchors = [
anchor
for anchor in activity_burst_anchors
if int(anchor.get("textEnd") or 0) < len(assistant_text)
]
segment_count = len(visible_anchors) + (1 if assistant_text else 0)
current_live_segment_seq = max(segment_count, len(activity_burst_anchors), 0)
try:
summary_last_seq = max(0, int(summary.get("last_seq") or 0))
except (TypeError, ValueError):
summary_last_seq = 0
try:
event_last_seq = max(0, int(events[-1].get("seq") or 0))
except (TypeError, ValueError):
event_last_seq = 0
if event_last_seq >= summary_last_seq:
last_seq = event_last_seq
last_event_id = events[-1].get("event_id") or (
f"{stream_id}:{event_last_seq}" if event_last_seq else summary.get("last_event_id")
)
else:
last_seq = summary_last_seq
last_event_id = summary.get("last_event_id") or events[-1].get("event_id")
return {
"session_id": session_id,
"stream_id": stream_id,
"last_seq": last_seq,
"last_event_id": last_event_id,
"event_count": len(events),
"fresh_segment": fresh_segment,
"messages": messages,
"tool_calls": tool_calls,
"last_assistant_text": assistant_text,
"last_reasoning_text": reasoning_text,
"activity_burst_anchors": activity_burst_anchors,
"current_activity_burst_id": current_activity_burst_id,
"current_live_segment_seq": current_live_segment_seq,
}
def _ensure_full_session_before_mutation(sid: str, session):
"""Reload cached metadata-only sessions before mutating persisted fields.
@ -5691,6 +5907,19 @@ def handle_get(handler, parsed) -> bool:
journal,
active=journal_active,
)
if journal_active:
try:
snapshot = _run_journal_live_snapshot(original_stream_id)
except Exception:
logger.debug(
"Failed to build runtime journal snapshot for %s",
original_stream_id,
exc_info=True,
)
snapshot = None
if snapshot:
raw["runtime_journal_snapshot"] = snapshot
raw["pending_attachments"] = getattr(s, "pending_attachments", []) or []
# Cold-load: derive the latest settled todo snapshot from the full
# merged transcript, not the truncated display window. This keeps
# the Todos panel correct after refresh even when the latest todo

View file

@ -1898,13 +1898,13 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){
const byTid=new Map();
liveCalls.forEach((tc,idx)=>{
if(!tc||typeof tc!=='object') return;
const tid=tc.tid||tc.id||tc.tool_call_id||tc.call_id||'';
const tid=tc.tid||tc.id||tc.tool_call_id||tc.tool_use_id||tc.call_id||'';
if(tid&&!byTid.has(tid)) byTid.set(tid,{tc,idx});
});
const used=new Set();
return (rawCalls||[]).map((raw,idx)=>{
const next={...(raw||{}),done:true};
const tid=next.tid||next.id||next.tool_call_id||next.call_id||'';
const tid=next.tid||next.id||next.tool_call_id||next.tool_use_id||next.call_id||'';
let matchEntry=tid?byTid.get(tid):null;
if(!matchEntry){
const name=next.name||((next.function||{}).name)||'';
@ -2463,7 +2463,7 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){
}
function _liveToolTid(d, activityBurstId, activitySegmentSeq){
const explicit=String(d&&d.tid||'').trim();
const explicit=String(d&&(d.tid||d.id||d.tool_call_id||d.tool_use_id||d.call_id)||'').trim();
if(explicit) return explicit;
return `live-${activeSid}-${_hashString(_toolCallSignature(d,activityBurstId,activitySegmentSeq))}`;
}
@ -2491,7 +2491,7 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){
const candidate=toolCalls[i];
if(!candidate||typeof candidate!=='object') continue;
if(!allowDone&&candidate.done===true) continue;
const candidateTid=String(candidate.tid||candidate.id||candidate.tool_call_id||candidate.call_id||'');
const candidateTid=String(candidate.tid||candidate.id||candidate.tool_call_id||candidate.tool_use_id||candidate.call_id||'');
if(candidateTid&&candidateTid===wantedTid) return i;
}
}
@ -2561,7 +2561,7 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){
if(!Array.isArray(inflight.toolCalls)) inflight.toolCalls=[];
if(!Array.isArray(inflight.messages)) inflight.messages=[...(inflight.messages||[])];
const explicitTid=String(d&&d.tid||'').trim();
const explicitTid=String(d&&d.tid||d&&d.id||d&&d.tool_call_id||d&&d.tool_use_id||d&&d.call_id||'').trim();
const isComplete=phase==='complete';
let signature=_toolCallSignature(d,current.burstId,current.segmentSeq);
let index=-1;

View file

@ -454,6 +454,63 @@ function _inflightHasVisibleLiveState(inflight) {
return false;
}
function _serverLiveSnapshotToolId(tc){
return String(tc&&(tc.tid||tc.id||tc.tool_call_id||tc.tool_use_id||tc.call_id||'')||'').trim();
}
function _serverLiveSnapshotInflight(snapshot, uploaded){
if(!snapshot||typeof snapshot!=='object') return null;
const rawMessages=Array.isArray(snapshot.messages)?snapshot.messages:[];
const messages=rawMessages
.filter(m=>m&&m.role)
.map(m=>({...m,_live:m._live!==false,_journal_snapshot:true}));
const rawToolCalls=Array.isArray(snapshot.tool_calls)?snapshot.tool_calls:[];
const toolCalls=rawToolCalls
.filter(tc=>tc&&tc.name)
.map(tc=>{
const next={...tc,_live:true,_journal_snapshot:true};
const tid=_serverLiveSnapshotToolId(next);
if(tid&&!next.tid) next.tid=tid;
return next;
});
let lastAssistantText=String(snapshot.last_assistant_text||snapshot.lastAssistantText||'');
let lastReasoningText=String(snapshot.last_reasoning_text||snapshot.lastReasoningText||'');
const lastLiveAssistant=[...messages].reverse().find(m=>m&&m.role==='assistant'&&m._live);
if(lastLiveAssistant){
if(!lastAssistantText&&typeof lastLiveAssistant.content==='string') lastAssistantText=lastLiveAssistant.content;
if(!lastReasoningText&&typeof lastLiveAssistant.reasoning==='string') lastReasoningText=lastLiveAssistant.reasoning;
}
if((lastAssistantText||lastReasoningText)&&!lastLiveAssistant){
messages.push({
role:'assistant',
content:lastAssistantText,
reasoning:lastReasoningText||undefined,
_live:true,
_journal_snapshot:true,
});
}
if(!messages.length&&!toolCalls.length&&!lastAssistantText&&!lastReasoningText) return null;
const replayAfterSeq=Number(snapshot.last_seq||0);
const activityBurstAnchors=Array.isArray(snapshot.activity_burst_anchors)
? snapshot.activity_burst_anchors
: (Array.isArray(snapshot.activityBurstAnchors)?snapshot.activityBurstAnchors:[]);
return {
messages,
uploaded:Array.isArray(uploaded)?[...uploaded]:[],
toolCalls,
todos:null,
todoStateMeta:null,
reattach:true,
journalSnapshot:true,
lastAssistantText,
lastReasoningText,
lastRunJournalSeq:Number.isFinite(replayAfterSeq)?Math.max(0,replayAfterSeq):0,
currentActivityBurstId:Number(snapshot.current_activity_burst_id||snapshot.currentActivityBurstId||0)||0,
currentLiveSegmentSeq:Number(snapshot.current_live_segment_seq||snapshot.currentLiveSegmentSeq||0)||0,
activityBurstAnchors,
};
}
function _rememberRenderedSessionSnapshot(s) {
if (!s || !s.session_id) return;
const previous = _sessionListSnapshotById.get(s.session_id);
@ -999,6 +1056,13 @@ async function loadSession(sid){
if(typeof clearInflightState==='function') clearInflightState(sid);
}
const serverLiveSnapshot=activeStreamId
? _serverLiveSnapshotInflight(S.session.runtime_journal_snapshot, S.session.pending_attachments||[])
: null;
if(serverLiveSnapshot&&(!INFLIGHT[sid]||!_inflightHasVisibleLiveState(INFLIGHT[sid]))){
INFLIGHT[sid]=serverLiveSnapshot;
}
if(INFLIGHT[sid]){
_ensureInflightLiveAssistantMessage(INFLIGHT[sid]);
const inflightMessages=_projectInflightMessagesForActivityBursts(INFLIGHT[sid]);

View file

@ -3,6 +3,7 @@ from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
MESSAGES_SRC = (ROOT / "static" / "messages.js").read_text()
SESSIONS_SRC = (ROOT / "static" / "sessions.js").read_text()
def test_reattach_path_uses_replay_when_status_reports_journal():
@ -107,3 +108,34 @@ def test_run_journal_cursor_tracks_every_long_task_timeline_event():
assert f"'{event_name}'" in cursor_loop, (
f"{event_name} must advance the replay cursor to avoid duplicate timeline replay"
)
def test_server_runtime_journal_snapshot_restores_structured_inflight_state():
helper_pos = SESSIONS_SRC.index("function _serverLiveSnapshotToolId")
helper_block = SESSIONS_SRC[helper_pos : helper_pos + 2600]
load_pos = SESSIONS_SRC.index("async function loadSession")
load_block = SESSIONS_SRC[load_pos : load_pos + 18000]
assert "runtime_journal_snapshot" in load_block
assert "_serverLiveSnapshotInflight(S.session.runtime_journal_snapshot" in load_block
assert "!_inflightHasVisibleLiveState(INFLIGHT[sid])" in load_block
assert "journalSnapshot:true" in helper_block
assert "lastRunJournalSeq" in helper_block
assert "last_assistant_text" in helper_block
assert "activity_burst_anchors" in helper_block
for key in ("tid", "id", "tool_call_id", "tool_use_id", "call_id"):
assert key in helper_block
def test_live_tool_matching_uses_the_same_aliases_as_live_card_dedup():
live_tid_pos = MESSAGES_SRC.index("function _liveToolTid")
live_tid_block = MESSAGES_SRC[live_tid_pos : live_tid_pos + 450]
find_pos = MESSAGES_SRC.index("function _findPendingLiveToolCallIndex")
find_block = MESSAGES_SRC[find_pos : find_pos + 900]
upsert_pos = MESSAGES_SRC.index("function upsertLiveToolCall")
upsert_block = MESSAGES_SRC[upsert_pos : upsert_pos + 600]
for key in ("tid", "id", "tool_call_id", "tool_use_id", "call_id"):
assert key in live_tid_block
assert key in find_block
assert key in upsert_block

View file

@ -264,11 +264,108 @@ def test_replay_emits_event_ids_and_stale_restart_diagnostic():
def test_session_payload_exposes_runtime_journal_for_stale_streams():
assert "original_stream_id = getattr(s, \"active_stream_id\", None)" in ROUTES_SRC
assert '"runtime_journal"' in ROUTES_SRC
assert '"runtime_journal_snapshot"' in ROUTES_SRC
assert "_run_journal_live_snapshot(original_stream_id)" in ROUTES_SRC
assert 'terminal_state = "lost-worker-bookkeeping"' in ROUTES_SRC
assert "active=journal_active" in ROUTES_SRC
assert "journal_active = bool(original_stream_id in active_stream_ids)" in ROUTES_SRC
def test_live_journal_snapshot_reconstructs_visible_progress_and_tool_aliases(monkeypatch):
import api.routes as routes
monkeypatch.setattr(
routes,
"find_run_summary",
lambda stream_id: {
"session_id": "session_1",
"run_id": stream_id,
"last_seq": 4,
"last_event_id": f"{stream_id}:4",
},
)
monkeypatch.setattr(
routes,
"read_run_events",
lambda session_id, run_id: {
"events": [
{
"seq": 1,
"event": "token",
"payload": {"text": "First segment."},
"event_id": f"{run_id}:1",
"created_at": 1000.0,
},
{
"seq": 2,
"event": "tool",
"payload": {
"name": "terminal",
"preview": "running tests",
"tool_use_id": "toolu_123",
"args": {"command": "pytest -q", "extra": "x" * 200},
},
"event_id": f"{run_id}:2",
},
{
"seq": 3,
"event": "tool_complete",
"payload": {
"name": "terminal",
"preview": "passed",
"tool_use_id": "toolu_123",
"duration": 1.25,
},
"event_id": f"{run_id}:3",
},
{
"seq": 4,
"event": "reasoning",
"payload": {"text": "Checked result."},
"event_id": f"{run_id}:4",
},
{
"seq": 5,
"event": "token",
"payload": {"text": " Second segment."},
"event_id": f"{run_id}:5",
"created_at": 1001.0,
},
]
},
)
snapshot = routes._run_journal_live_snapshot("run_1")
assert snapshot["last_seq"] == 5
assert snapshot["last_event_id"] == "run_1:5"
assert snapshot["last_assistant_text"] == "First segment. Second segment."
assert snapshot["last_reasoning_text"] == "Checked result."
assert snapshot["current_live_segment_seq"] == 2
assert snapshot["activity_burst_anchors"] == [{"id": 1, "textEnd": len("First segment.")}]
assert snapshot["messages"] == [
{
"role": "assistant",
"content": "First segment. Second segment.",
"reasoning": "Checked result.",
"_live": True,
"_journal_snapshot": True,
"_journal_stream_id": "run_1",
"_ts": 1001.0,
}
]
tool = snapshot["tool_calls"][0]
assert tool["name"] == "terminal"
assert tool["done"] is True
assert tool["tid"] == "toolu_123"
assert tool["tool_use_id"] == "toolu_123"
assert tool["activityBurstId"] == 1
assert tool["activitySegmentSeq"] == 1
assert tool["snippet"] == "passed"
assert tool["duration"] == 1.25
assert len(tool["args"]["extra"]) <= 123
def test_status_payload_marks_non_terminal_dead_journal_as_stale():
import api.routes as routes

View file

@ -304,6 +304,34 @@ def test_redact_session_data_todo_state_sidecar():
assert result["todo_state"]["todos"][0]["status"] == "pending"
def test_redact_session_data_runtime_journal_snapshot():
"""redact_session_data masks credentials in active run-journal snapshots."""
from api.helpers import redact_session_data
session = {
"session_id": "journal-redact",
"messages": [],
"tool_calls": [],
"runtime_journal_snapshot": {
"messages": [
{"role": "assistant", "content": f"token echoed: {_FAKE_GITHUB_PAT}"},
],
"tool_calls": [
{
"name": "terminal",
"preview": f"using {_FAKE_SK_KEY}",
"args": {"command": f"export TOKEN={_FAKE_GITHUB_PAT}"},
},
],
"last_assistant_text": f"assistant saw {_FAKE_HF_TOKEN}",
"last_reasoning_text": f"reasoning saw {_FAKE_AWS_KEY}",
},
}
result = redact_session_data(session)
dump = json.dumps(result)
_assert_no_plaintext_credentials(dump, "runtime_journal_snapshot redaction")
assert result["runtime_journal_snapshot"]["tool_calls"][0]["name"] == "terminal"
def test_redact_session_data_multiple_cred_types():
"""redact_session_data handles sk-, ghp_, hf_, and AKIA keys."""
from api.helpers import redact_session_data