diff --git a/CHANGELOG.md b/CHANGELOG.md index a1942f4e..4ea42aeb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,6 +17,8 @@ ### Fixed +- **Live Stream: Assistant Turn Anchors now carry a stable run identity through the settled scene.** The activity-scene projection now threads a stable `run_id` (independent of the transport `stream_id`) through Anchor scenes and client hydration, so a settled/reloaded Worklog keeps a consistent identity and can't mis-own or duplicate rows if a producer ever emits `run_id != stream_id`. Behavior-preserving for current writers (which still emit `run_id == stream_id`) — verified byte-identical live/settle/reload render, with malformed/conflicting identities falling back to the transport identity. Thanks @franksong2702. (#6201, #3400) + - **Content search no longer evicts your open sessions from the cache.** A content search scans many sessions, and each scan used to promote/cold-load sessions into the in-memory LRU — evicting the sessions you actually have open. Search now reads through a scan-only accessor that keeps the same disk-freshness, journal-recovery, and state.db reconciliation (so recent content stays searchable) but doesn't disturb the working-set cache. Thanks @ai-ag2026. (#6084, #6083) - **Watching the same terminal from two tabs no longer splits the output stream.** Terminal output is now broadcast to every subscribed viewer instead of draining from one shared queue, so a second tab (or a reconnect) sees the full stream rather than a random half. SSE responses carry event `id:`s and a reconnecting viewer replays only events after its `Last-Event-ID` cursor (fresh viewers still get the full bounded backlog). Thanks @ai-ag2026. (#5836) diff --git a/api/routes.py b/api/routes.py index 4d0998f5..fd57db10 100644 --- a/api/routes.py +++ b/api/routes.py @@ -3132,6 +3132,38 @@ def _run_journal_snapshot_merge_args(existing, incoming): return merged, changed +def _run_journal_envelope_run_id_result(event: dict) -> tuple[str | None, bool]: + raw_run_id = event.get("run_id") + if raw_run_id is None: + return None, False + if not isinstance(raw_run_id, str): + return None, True + run_id = raw_run_id.strip() + if not run_id: + return None, True + raw_event_id = event.get("event_id") + event_id = str(raw_event_id or "").strip() + if event_id: + event_run_id, event_seq = _shared_parse_run_journal_event_id(event_id) + if event_run_id and event_seq is not None and event_run_id != run_id: + return None, True + return run_id, False + + +def _run_journal_snapshot_event_id_for_run( + event: dict, + run_id: str, + event_seq: int, +) -> str | None: + raw_event_id = event.get("event_id") + event_id = str(raw_event_id or "").strip() + if event_id: + event_run_id, parsed_seq = _shared_parse_run_journal_event_id(event_id) + if event_run_id == run_id and parsed_seq is not None: + return event_id + return f"{run_id}:{event_seq}" if event_seq else None + + def _run_journal_live_snapshot(stream_id: str | None, *, handler=None) -> dict | None: stream_id = str(stream_id or "").strip() if not stream_id: @@ -3152,6 +3184,22 @@ def _run_journal_live_snapshot(stream_id: str | None, *, handler=None) -> dict | events = [event for event in (journal.get("events") or []) if isinstance(event, dict)] if not events: return None + event_run_ids: set[str] = set() + malformed_envelope_run_id = False + for event in events: + event_run_id, event_run_id_malformed = _run_journal_envelope_run_id_result(event) + if event_run_id is not None: + event_run_ids.add(event_run_id) + if event_run_id_malformed: + malformed_envelope_run_id = True + # The event envelope is the durable identity authority. Older summaries + # are keyed by the transport id, so only use that fallback when the journal + # does not provide one unambiguous run id. + run_id = ( + next(iter(event_run_ids)) + if not malformed_envelope_run_id and len(event_run_ids) == 1 + else str(summary.get("run_id") or stream_id).strip() + ) assistant_text = "" reasoning_text = "" @@ -3346,7 +3394,7 @@ def _run_journal_live_snapshot(stream_id: str | None, *, handler=None) -> dict | "source_event_type": "token", "event_id": None, "local_id": local_id, - "run_id": stream_id, + "run_id": run_id, "stream_id": stream_id, "seq": None, "status": status, @@ -3354,7 +3402,7 @@ def _run_journal_live_snapshot(stream_id: str | None, *, handler=None) -> dict | "identity": { "event_id": None, "local_id": local_id, - "run_id": stream_id, + "run_id": run_id, "stream_id": stream_id, "seq": None, }, @@ -3389,7 +3437,7 @@ def _run_journal_live_snapshot(stream_id: str | None, *, handler=None) -> dict | "source_event_type": "reasoning", "event_id": None, "local_id": local_id, - "run_id": stream_id, + "run_id": run_id, "stream_id": stream_id, "seq": None, "status": status, @@ -3397,7 +3445,7 @@ def _run_journal_live_snapshot(stream_id: str | None, *, handler=None) -> dict | "identity": { "event_id": None, "local_id": local_id, - "run_id": stream_id, + "run_id": run_id, "stream_id": stream_id, "seq": None, }, @@ -3466,7 +3514,7 @@ def _run_journal_live_snapshot(stream_id: str | None, *, handler=None) -> dict | "source_event_type": "tool_complete" if call.get("done") else "tool", "event_id": None, "local_id": tool_id or row_id, - "run_id": stream_id, + "run_id": run_id, "stream_id": stream_id, "seq": None, "status": status, @@ -3474,7 +3522,7 @@ def _run_journal_live_snapshot(stream_id: str | None, *, handler=None) -> dict | "identity": { "event_id": None, "local_id": tool_id or row_id, - "run_id": stream_id, + "run_id": run_id, "stream_id": stream_id, "seq": None, }, @@ -3588,7 +3636,7 @@ def _run_journal_live_snapshot(stream_id: str | None, *, handler=None) -> dict | "source_event_type": "runtime_journal_snapshot", "event_id": None, "local_id": f"lifecycle:{stream_id}:running", - "run_id": stream_id, + "run_id": run_id, "stream_id": stream_id, "seq": None, "status": "running", @@ -3596,7 +3644,7 @@ def _run_journal_live_snapshot(stream_id: str | None, *, handler=None) -> dict | "identity": { "event_id": None, "local_id": f"lifecycle:{stream_id}:running", - "run_id": stream_id, + "run_id": run_id, "stream_id": stream_id, "seq": None, }, @@ -3626,9 +3674,11 @@ def _run_journal_live_snapshot(stream_id: str | None, *, handler=None) -> dict | 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") - ) + last_event_id = _run_journal_snapshot_event_id_for_run( + events[-1], + run_id, + event_last_seq, + ) or 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") @@ -3656,7 +3706,7 @@ def _run_journal_live_snapshot(stream_id: str | None, *, handler=None) -> dict | "identity": { "session_id": session_id, "stream_id": stream_id, - "run_id": stream_id, + "run_id": run_id, "source_message_refs": [], }, "lifecycle": { diff --git a/static/messages.js b/static/messages.js index cb6ce5cf..58993eba 100644 --- a/static/messages.js +++ b/static/messages.js @@ -3992,8 +3992,12 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ function _hydrateAnchorRegistryFromActivityScene(scene){ if(!_anchorRegistry||!_anchorApi||typeof _anchorApi.applyAssistantTurnAnchorSourceEvent!=='function') return false; if(!scene||scene.version!=='activity_scene_v1'||!Array.isArray(scene.activity_rows)||!scene.activity_rows.length) return false; + const sceneIdentity=(scene.identity&&typeof scene.identity==='object')?scene.identity:{}; + const sceneStreamId=sceneIdentity.stream_id||streamId; + const sceneRunId=sceneIdentity.run_id||sceneStreamId; const sceneKey=[ - scene.identity&&scene.identity.stream_id||streamId||'', + sceneRunId||'', + sceneStreamId||'', scene.activity_rows.length, scene.activity_rows.map(row=>row&&row.row_id||row&&row.local_id||'').join('|'), ].join(':'); @@ -4022,22 +4026,23 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ payload.activitySegmentSeq=payload.activitySegmentSeq||row.group.activity_segment_seq; payload.activityBurstId=payload.activityBurstId||row.group.activity_burst_id; } + const rowIdentity=(row.identity&&typeof row.identity==='object')?row.identity:{}; const sourceEvent={ ...payload, source_event_type:sourceType, - local_id:row.local_id||row.row_id||`snapshot:${streamId}:${i}`, + local_id:row.local_id||row.row_id||`snapshot:${sceneStreamId}:${i}`, event_id:row.event_id||null, seq:row.seq??undefined, status:row.status||undefined, - stream_id:row.stream_id||streamId, - run_id:row.run_id||streamId, + stream_id:row.stream_id||rowIdentity.stream_id||sceneStreamId, + run_id:row.run_id||rowIdentity.run_id||sceneRunId, // Carry the row's persisted creation timestamp through hydration so the // worklog event timestamp (#5700/#5739) survives a settled-snapshot rebuild // (payload may not carry created_at even when the row does). (#5739 gate.) created_at:payload.created_at??row.created_at??undefined, }; try{ - _anchorApi.applyAssistantTurnAnchorSourceEvent(_anchorRegistry,sourceEvent,{session_id:activeSid,stream_id:streamId,run_id:streamId}); + _anchorApi.applyAssistantTurnAnchorSourceEvent(_anchorRegistry,sourceEvent,{session_id:activeSid,stream_id:sceneStreamId,run_id:sceneRunId}); }catch(err){ if(!_anchorShadowWarned&&typeof console!=='undefined'&&console.warn){ _anchorShadowWarned=true; diff --git a/tests/test_live_anchor_stable_run_identity.py b/tests/test_live_anchor_stable_run_identity.py new file mode 100644 index 00000000..e9df7b1d --- /dev/null +++ b/tests/test_live_anchor_stable_run_identity.py @@ -0,0 +1,304 @@ +"""Stable run identity regressions for live Anchor scene projection/hydration.""" + +import json +import shutil +import subprocess +from pathlib import Path + +import pytest + + +ROOT = Path(__file__).resolve().parents[1] +NODE = shutil.which("node") + + +def test_runtime_journal_snapshot_preserves_run_id_separately_from_stream_id( + monkeypatch, +): + from api import routes + + session_id = "session-stable-run" + run_id = "run-stable-1" + stream_id = "stream-transport-1" + events = [ + { + "event": "token", + "seq": 1, + "event_id": f"{run_id}:1", + "run_id": run_id, + "created_at": 1.0, + "payload": {"text": "working"}, + }, + { + "event": "reasoning", + "seq": 2, + "event_id": f"{run_id}:2", + "run_id": run_id, + "created_at": 2.0, + "payload": {"text": "checking"}, + }, + { + "event": "tool", + "seq": 3, + "event_id": f"{run_id}:3", + "run_id": run_id, + "created_at": 3.0, + "payload": {"name": "terminal", "tid": "call-1"}, + }, + { + "event": "tool_complete", + "seq": 4, + "event_id": f"{run_id}:4", + "run_id": run_id, + "created_at": 4.0, + "payload": {"name": "terminal", "tid": "call-1", "preview": "ok"}, + }, + ] + + monkeypatch.setattr( + routes, + "find_run_summary", + lambda lookup_id: { + "session_id": session_id, + # The current summary is keyed by the legacy lookup id, while the + # durable event envelope already carries the stable run identity. + "run_id": stream_id, + "stream_id": stream_id, + "last_seq": 4, + "last_event_id": f"{run_id}:4", + }, + ) + monkeypatch.setattr( + routes, + "read_run_events", + lambda loaded_session_id, lookup_id: {"events": events}, + ) + + snapshot = routes._run_journal_live_snapshot(stream_id) + scene = snapshot["anchor_activity_scene"] + + assert snapshot["stream_id"] == stream_id + assert scene["identity"]["run_id"] == run_id + assert scene["identity"]["stream_id"] == stream_id + assert scene["activity_rows"] + assert [row["role"] for row in scene["activity_rows"]] == [ + "prose", + "thinking", + "tool", + ] + assert {row["run_id"] for row in scene["activity_rows"]} == {run_id} + assert {row["stream_id"] for row in scene["activity_rows"]} == {stream_id} + assert {row["identity"]["run_id"] for row in scene["activity_rows"]} == {run_id} + assert {row["identity"]["stream_id"] for row in scene["activity_rows"]} == { + stream_id + } + + +def test_runtime_journal_lifecycle_shell_preserves_stable_run_id(monkeypatch): + from api import routes + + run_id = "run-stable-lifecycle" + stream_id = "stream-transport-lifecycle" + event = { + "event": "metering", + "seq": 1, + "event_id": f"{run_id}:1", + "run_id": run_id, + "payload": {}, + } + monkeypatch.setattr( + routes, + "find_run_summary", + lambda lookup_id: { + "session_id": "session-stable-lifecycle", + "run_id": stream_id, + "stream_id": stream_id, + "last_seq": 1, + "last_event_id": f"{run_id}:1", + }, + ) + monkeypatch.setattr( + routes, + "read_run_events", + lambda loaded_session_id, lookup_id: {"events": [event]}, + ) + + snapshot = routes._run_journal_live_snapshot(stream_id) + scene = snapshot["anchor_activity_scene"] + row = scene["activity_rows"][0] + + assert row["role"] == "lifecycle" + assert row["run_id"] == run_id + assert row["stream_id"] == stream_id + assert row["identity"]["run_id"] == run_id + assert row["identity"]["stream_id"] == stream_id + + +@pytest.mark.parametrize( + ("event_run_id", "event_id"), + [ + ({"bad": "id"}, None), + ("run-conflicting-envelope", "run-conflicting-event-id:1"), + ], +) +def test_runtime_journal_malformed_envelope_run_id_falls_back_to_transport_cursor( + monkeypatch, + event_run_id, + event_id, +): + from api import routes + + stream_id = "stream-current-1" + event = { + "event": "token", + "seq": 1, + "run_id": event_run_id, + "payload": {"text": "hello"}, + } + if event_id is not None: + event["event_id"] = event_id + + monkeypatch.setattr( + routes, + "find_run_summary", + lambda lookup_id: { + "session_id": "session-malformed-envelope", + "run_id": stream_id, + "stream_id": stream_id, + "last_seq": 1, + "last_event_id": f"{stream_id}:1", + }, + ) + monkeypatch.setattr( + routes, + "read_run_events", + lambda loaded_session_id, lookup_id: {"events": [event]}, + ) + + snapshot = routes._run_journal_live_snapshot(stream_id) + scene = snapshot["anchor_activity_scene"] + row = scene["activity_rows"][0] + + assert scene["identity"]["run_id"] == stream_id + assert row["run_id"] == stream_id + assert row["identity"]["run_id"] == stream_id + assert snapshot["last_event_id"] == f"{stream_id}:1" + assert ( + routes._parse_run_journal_after_seq( + {"after_event_id": [snapshot["last_event_id"]]}, + stream_id, + ) + == 1 + ) + + +@pytest.mark.skipif( + NODE is None, reason="node is required for browser hydration regression coverage" +) +def test_browser_scene_hydration_prefers_stable_run_id_with_legacy_stream_fallback(): + messages_path = ROOT / "static" / "messages.js" + script = f""" +const fs=require('fs'); +const src=fs.readFileSync({json.dumps(str(messages_path))},'utf8'); +function extractFunc(name){{ + const start=src.indexOf('function '+name); + if(start===-1) throw new Error(name+' not found'); + const params=src.indexOf('(',start); + let depth=0,close=-1; + for(let i=params;i