diff --git a/CHANGELOG.md b/CHANGELOG.md index 79893c42..851b2b9c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/api/helpers.py b/api/helpers.py index 432f3c76..4a39ef9e 100644 --- a/api/helpers.py +++ b/api/helpers.py @@ -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 diff --git a/api/routes.py b/api/routes.py index 451fa140..e7ab46a2 100644 --- a/api/routes.py +++ b/api/routes.py @@ -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 diff --git a/static/messages.js b/static/messages.js index 69e5a3d7..89cec5c2 100644 --- a/static/messages.js +++ b/static/messages.js @@ -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; diff --git a/static/sessions.js b/static/sessions.js index e89de870..89f82b55 100644 --- a/static/sessions.js +++ b/static/sessions.js @@ -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]); diff --git a/tests/test_run_journal_frontend_static.py b/tests/test_run_journal_frontend_static.py index 9a100584..d8e816e3 100644 --- a/tests/test_run_journal_frontend_static.py +++ b/tests/test_run_journal_frontend_static.py @@ -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 diff --git a/tests/test_run_journal_routes.py b/tests/test_run_journal_routes.py index bae7aacd..6767384b 100644 --- a/tests/test_run_journal_routes.py +++ b/tests/test_run_journal_routes.py @@ -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 diff --git a/tests/test_security_redaction.py b/tests/test_security_redaction.py index 86a22e50..457c2625 100644 --- a/tests/test_security_redaction.py +++ b/tests/test_security_redaction.py @@ -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