jinghui/hermes-webui
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Release: Live Stream stable Anchor run identity (#6201, @franksong2702) (#6346)

Browse Source 
* fix: preserve live anchor run identity

* Validate envelope run ids before snapshot cursor use

* docs(changelog): stamp #6201 stable Anchor run identity

---------

Co-authored-by: Frank Song <franksong2702@gmail.com>
Co-authored-by: nesquena-hermes <agent@nesquena-hermes>
This commit is contained in:
nesquena-hermesFrank Songnesquena-hermes
 authored and GitHub committed 2026-07-20 04:10:06 +00:00
1 parent 98a67ee6cf
commit 02b2b0aac1
4 files changed
+378 -17
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

CHANGELOG.md
+2
View File
@@ -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)
api/routes.py
+62 -12
View File
@@ -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": {
static/messages.js
+10 -5
View File
@@ -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;
tests/test_live_anchor_stable_run_identity.py
+304
View File
@@ -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<src.length;i+=1){{
if(src[i]==='(') depth+=1;
else if(src[i]===')'){{
depth-=1;
if(depth===0){{ close=i; break; }}
}}
}}
const brace=src.indexOf('{{',close);
depth=0;
for(let i=brace;i<src.length;i+=1){{
if(src[i]==='{{') depth+=1;
else if(src[i]==='}}'){{
depth-=1;
if(depth===0) return src.slice(start,i+1);
}}
}}
throw new Error(name+' body did not close');
}}
const activeSid='session-stable-run';
const streamId='stream-transport-1';
let _anchorShadowWarned=false;
let _anchorRegistry={{anchor:{{identity:{{run_id:null,stream_id:streamId}}}}}};
const applied=[];
const _anchorApi={{
applyAssistantTurnAnchorSourceEvent(registry,event,context){{
applied.push({{event,context}});
registry.anchor.identity.run_id=event.run_id||context.run_id||null;
registry.anchor.identity.stream_id=event.stream_id||context.stream_id||null;
return event;
}},
}};
eval(extractFunc('_sourceEventTypeForSnapshotAnchorRow'));
eval(extractFunc('_hydrateAnchorRegistryFromActivityScene'));
function scene(identity,rowId){{
return {{
version:'activity_scene_v1',
identity,
activity_rows:[{{
row_id:rowId,
role:'prose',
kind:'process_prose',
source_event_type:'token',
payload:{{text:'working'}},
text:'working',
}}],
}};
}}
const stableHydrated=_hydrateAnchorRegistryFromActivityScene(scene({{
run_id:'run-stable-1',
stream_id:streamId,
}},'stable-row'));
const stable=applied[0];
_anchorRegistry={{anchor:{{identity:{{run_id:null,stream_id:streamId}}}}}};
const legacyHydrated=_hydrateAnchorRegistryFromActivityScene(scene({{
stream_id:streamId,
}},'legacy-row'));
const legacy=applied[1];
const sceneScopedStreamId='stream-scene-scoped';
_anchorRegistry={{anchor:{{identity:{{run_id:null,stream_id:sceneScopedStreamId}}}}}};
const fallbackHydrated=_hydrateAnchorRegistryFromActivityScene(scene({{
stream_id:sceneScopedStreamId,
}},null));
const fallback=applied[2];
console.log(JSON.stringify({{
stableHydrated,
legacyHydrated,
fallbackHydrated,
stable,
legacy,
fallback,
}}));
"""
result = subprocess.run(
[NODE, "-e", script],
text=True,
capture_output=True,
check=False,
)
assert result.returncode == 0, result.stderr
data = json.loads(result.stdout)
assert data["stableHydrated"] is True
assert data["stable"]["event"]["run_id"] == "run-stable-1"
assert data["stable"]["event"]["stream_id"] == "stream-transport-1"
assert data["stable"]["context"]["run_id"] == "run-stable-1"
assert data["stable"]["context"]["stream_id"] == "stream-transport-1"
assert data["legacyHydrated"] is True
assert data["legacy"]["event"]["run_id"] == "stream-transport-1"
assert data["legacy"]["context"]["run_id"] == "stream-transport-1"
assert data["fallbackHydrated"] is True
assert data["fallback"]["event"]["local_id"] == "snapshot:stream-scene-scoped:0"