import json from collections import OrderedDict from types import SimpleNamespace def _client_anchor_scene_message_ref(message): content = message.get("content") if isinstance(message, dict) else "" if isinstance(content, list): parts = [] for part in content: if isinstance(part, dict): parts.append(str(part.get("text") or part.get("content") or part.get("input_text") or "")) else: parts.append(str(part or "")) content = "\n".join(parts) payload = { "role": str(message.get("role") or ""), "content": " ".join(str(content or "").split()), "timestamp": message.get("_ts") or message.get("timestamp") or "", } return json.dumps(payload, ensure_ascii=False, separators=(",", ":")) def _anchor_scene_visible_semantics(scene, *, include_terminal=False): semantics = [] for row in scene.get("activity_rows") or []: role = row.get("role") if role == "terminal" and not include_terminal: continue if role in {"prose", "thinking"}: semantics.append( { "role": role, "kind": row.get("kind"), "text": " ".join(str(row.get("text") or "").split()), } ) continue if role == "tool": tool = row.get("tool") if isinstance(row.get("tool"), dict) else {} payload = row.get("payload") if isinstance(row.get("payload"), dict) else {} args = tool.get("args") if "args" in tool else payload.get("args") if args is None: args = {} semantics.append( { "role": role, "kind": row.get("kind"), "status": row.get("status"), "tool_call_id": row.get("tool_call_id") or tool.get("id") or payload.get("tid"), "name": tool.get("name") or payload.get("name"), "args": args, "done": tool.get("done"), } ) continue if role == "terminal": semantics.append( { "role": role, "kind": row.get("kind"), "source_event_type": row.get("source_event_type"), "status": row.get("status"), } ) return semantics def test_anchor_scene_visible_semantics_preserves_empty_tool_args(): scene = { "activity_rows": [ { "role": "tool", "kind": "tool_completed", "status": "completed", "tool_call_id": "call-1", "tool": {"id": "call-1", "name": "terminal", "args": {}}, "payload": {"tid": "call-1", "name": "terminal", "args": {"command": "stale"}}, } ] } assert _anchor_scene_visible_semantics(scene)[0]["args"] == {} def test_anchor_scene_persistence_round_trip_outside_provider_messages(tmp_path, monkeypatch): from api import models, routes from api.models import Session session_dir = tmp_path / "sessions" session_dir.mkdir() monkeypatch.setattr(models, "SESSION_DIR", session_dir) monkeypatch.setattr(models, "SESSION_INDEX_FILE", session_dir / "_index.json") monkeypatch.setattr(models, "SESSIONS", OrderedDict()) monkeypatch.setattr(routes, "SESSION_DIR", session_dir) monkeypatch.setattr(routes, "SESSIONS", models.SESSIONS) session = Session( session_id="anchorpersist1", title="Anchor persistence", messages=[ {"role": "user", "content": "question"}, {"role": "assistant", "content": "final answer", "timestamp": 10.0}, ], ) session.save(skip_index=True) scene = { "version": "activity_scene_v1", "mode": "compact_worklog", "activity_rows": [ { "row_id": "tool-1", "role": "tool", "tool_call_id": "call-1", "tool": {"id": "call-1", "name": "terminal", "args": {"command": "git status"}}, } ], "final_answer": "final answer", } request_body = { "session_id": "anchorpersist1", "stream_id": "stream-1", "message_index": 1, "message_ref": "stale-ref-after-content-normalization", "scene": scene, } captured = {} monkeypatch.setattr(routes, "_check_csrf", lambda handler: True) monkeypatch.setattr(routes, "read_body", lambda handler: request_body) monkeypatch.setattr( routes, "j", lambda handler, payload, status=200, extra_headers=None: captured.update( payload=payload, status=status ) or True, ) assert routes.handle_post(SimpleNamespace(command="POST"), SimpleNamespace(path="/api/session/anchor-scene")) is True assert captured["status"] == 200 assert captured["payload"]["ok"] is True assert captured["payload"]["message_index"] == 1 raw = json.loads((session_dir / "anchorpersist1.json").read_text(encoding="utf-8")) assert "_anchor_activity_scene" not in raw["messages"][1] assert raw["messages"][1]["content"] == "final answer" records = raw["anchor_activity_scenes"] assert len(records) == 1 record = next(iter(records.values())) assert record["message_index"] == 1 assert record["stream_id"] == "stream-1" assert record["scene"]["version"] == "activity_scene_v1" loaded = Session.load("anchorpersist1") hydrated = routes._hydrate_anchor_activity_scenes( loaded.messages, loaded.anchor_activity_scenes, message_offset=0, ) assert "_anchor_activity_scene" not in loaded.messages[1] assert hydrated[1]["_anchor_stream_id"] == "stream-1" assert hydrated[1]["_anchor_activity_scene"]["final_answer"] == "final answer" assert hydrated[1]["_anchor_activity_scene"]["activity_rows"][0]["tool_call_id"] == "call-1" def test_anchor_scene_persistence_rejects_cross_profile_write(tmp_path, monkeypatch): """#4411 security: /api/session/anchor-scene must not persist a scene onto a session that isn't visible to the active request profile. _get_or_materialize_session loads by id with no profile scoping, so the handler must apply the same _session_visible_to_active_profile guard GET /api/session uses — returning 404 and leaving anchor_activity_scenes untouched (no cross-profile write).""" from api import models, routes from api.models import Session session_dir = tmp_path / "sessions" session_dir.mkdir() monkeypatch.setattr(models, "SESSION_DIR", session_dir) monkeypatch.setattr(models, "SESSION_INDEX_FILE", session_dir / "_index.json") monkeypatch.setattr(models, "SESSIONS", OrderedDict()) monkeypatch.setattr(routes, "SESSION_DIR", session_dir) monkeypatch.setattr(routes, "SESSIONS", models.SESSIONS) session = Session( session_id="foreignprofile1", title="Owned by profile B", messages=[ {"role": "user", "content": "question"}, {"role": "assistant", "content": "final answer", "timestamp": 10.0}, ], ) session.profile = "profile-b" session.save(skip_index=True) scene = { "version": "activity_scene_v1", "mode": "compact_worklog", "activity_rows": [ { "row_id": "tool-1", "role": "tool", "tool_call_id": "call-1", "tool": {"id": "call-1", "name": "terminal", "args": {"command": "git status"}}, } ], "final_answer": "final answer", } request_body = { "session_id": "foreignprofile1", "stream_id": "stream-1", "message_index": 1, "message_ref": "ref", "scene": scene, } captured = {} monkeypatch.setattr(routes, "_check_csrf", lambda handler: True) monkeypatch.setattr(routes, "read_body", lambda handler: request_body) # Request runs under a profile that CANNOT see profile-b's session. monkeypatch.setattr( routes, "_session_visible_to_active_profile", lambda session_profile, handler=None: session_profile not in ("profile-b",), ) monkeypatch.setattr( routes, "bad", lambda handler, msg, status=400, extra_headers=None: captured.update( error=msg, status=status ) or True, ) monkeypatch.setattr( routes, "j", lambda handler, payload, status=200, extra_headers=None: captured.update( payload=payload, status=status ) or True, ) assert routes.handle_post( SimpleNamespace(command="POST"), SimpleNamespace(path="/api/session/anchor-scene"), ) is True # #7710: the generic request-guard now mirrors the detail-load # endpoint's contract — a session owned by a KNOWN other profile # yields 409 ``session_profile_mismatch`` so the client can offer # to switch to it (#5419). The 404 self-heal path is preserved for # the None-profile (unknown/legacy) case. assert captured.get("status") == 409, captured assert captured.get("payload") == { "error": "Session belongs to a different profile", "code": "session_profile_mismatch", "session_id": "foreignprofile1", "profile": "profile-b", } raw = json.loads((session_dir / "foreignprofile1.json").read_text(encoding="utf-8")) assert not raw.get("anchor_activity_scenes"), ( "cross-profile request must NOT persist anchor_activity_scenes" ) def test_anchor_scene_hydration_skips_ambiguous_ref_match(monkeypatch): """#4411 defense-in-depth: when two assistant messages share a ref (byte-identical content + identical _ts), the read-side hydration must NOT attach the same scene to both (which would render duplicate worklog groups) — mirroring the write-side _find_anchor_scene_message ambiguity guard. The ambiguous ref falls through to the positional index match instead.""" from api import routes # Two assistant messages that normalize to the SAME ref. messages = [ {"role": "assistant", "content": "dup answer", "timestamp": 5.0}, {"role": "assistant", "content": "dup answer", "timestamp": 5.0}, ] ref = routes._assistant_anchor_scene_message_ref(messages[0]) assert ref == routes._assistant_anchor_scene_message_ref(messages[1]), "refs must collide for this test" # A single record keyed by that ambiguous ref, index-targeted at message 0. records = { ref: { "version": "anchor_activity_scene_record_v1", "message_index": 0, "message_ref": ref, "scene": {"version": "activity_scene_v1", "activity_rows": [], "final_answer": "dup answer"}, } } out = routes._hydrate_anchor_activity_scenes(messages, records, message_offset=0) attached = [("_anchor_activity_scene" in m) for m in out] # The ambiguous ref must NOT fan the scene out to BOTH messages. assert attached.count(True) <= 1, ( f"ambiguous ref must not double-attach the scene; got {attached}" ) def test_anchor_scene_hydration_rejects_stale_index_fallback_when_final_answer_mismatches(): from api import routes messages = [ {"role": "user", "content": "old question"}, {"role": "assistant", "content": "old final"}, {"role": "user", "content": "new question"}, {"role": "assistant", "content": "new final"}, ] records = { "stale-ref": { "message_index": 3, "message_ref": "missing-ref-after-window-shift", "stream_id": "stream-1", "scene": { "version": "activity_scene_v1", "mode": "compact_worklog", "final_answer": "old final", "activity_rows": [ {"row_id": "old-tool", "role": "tool", "tool_call_id": "old-call"} ], }, } } hydrated = routes._hydrate_anchor_activity_scenes(messages, records) assert "_anchor_activity_scene" not in hydrated[3] def test_anchor_scene_persistence_rejects_invalid_scene(tmp_path, monkeypatch): from api import models, routes from api.models import Session session_dir = tmp_path / "sessions" session_dir.mkdir() monkeypatch.setattr(models, "SESSION_DIR", session_dir) monkeypatch.setattr(models, "SESSION_INDEX_FILE", session_dir / "_index.json") monkeypatch.setattr(models, "SESSIONS", OrderedDict()) monkeypatch.setattr(routes, "SESSION_DIR", session_dir) monkeypatch.setattr(routes, "SESSIONS", models.SESSIONS) Session( session_id="anchorpersist2", messages=[{"role": "assistant", "content": "answer"}], ).save(skip_index=True) captured = {} monkeypatch.setattr(routes, "_check_csrf", lambda handler: True) monkeypatch.setattr( routes, "read_body", lambda handler: { "session_id": "anchorpersist2", "message_index": 0, "scene": {"version": "wrong", "activity_rows": []}, }, ) monkeypatch.setattr( routes, "bad", lambda handler, msg, status=400: captured.update(error=msg, status=status) or True, ) assert routes.handle_post(SimpleNamespace(command="POST"), SimpleNamespace(path="/api/session/anchor-scene")) is True assert captured["status"] == 400 assert "activity_scene_v1" in captured["error"] def test_anchor_scene_persistence_prefers_unique_ref_over_stale_index(tmp_path, monkeypatch): from api import models, routes from api.models import Session session_dir = tmp_path / "sessions" session_dir.mkdir() monkeypatch.setattr(models, "SESSION_DIR", session_dir) monkeypatch.setattr(models, "SESSION_INDEX_FILE", session_dir / "_index.json") monkeypatch.setattr(models, "SESSIONS", OrderedDict()) monkeypatch.setattr(routes, "SESSION_DIR", session_dir) monkeypatch.setattr(routes, "SESSIONS", models.SESSIONS) messages = [ {"role": "user", "content": "old question"}, {"role": "assistant", "content": "old final"}, {"role": "user", "content": "new question"}, {"role": "assistant", "content": "new final"}, ] Session(session_id="anchorpersist_ref", messages=messages).save(skip_index=True) client_ref = _client_anchor_scene_message_ref(messages[3]) assert client_ref != routes._assistant_anchor_scene_message_ref(messages[3]) captured = {} monkeypatch.setattr(routes, "_check_csrf", lambda handler: True) monkeypatch.setattr( routes, "read_body", lambda handler: { "session_id": "anchorpersist_ref", "message_index": 1, "message_ref": client_ref, "scene": { "version": "activity_scene_v1", "mode": "compact_worklog", "activity_rows": [{"row_id": "tool-1", "role": "tool"}], "final_answer": "new final", }, }, ) monkeypatch.setattr( routes, "j", lambda handler, payload, status=200, extra_headers=None: captured.update( payload=payload, status=status ) or True, ) assert routes.handle_post(SimpleNamespace(command="POST"), SimpleNamespace(path="/api/session/anchor-scene")) is True assert captured["payload"]["message_index"] == 3 raw = json.loads((session_dir / "anchorpersist_ref.json").read_text(encoding="utf-8")) record = next(iter(raw["anchor_activity_scenes"].values())) assert record["message_index"] == 3 assert record["scene"]["final_answer"] == "new final" def test_anchor_scene_persistence_rejects_duplicate_client_ref_over_stale_index(tmp_path, monkeypatch): from api import models, routes from api.models import Session session_dir = tmp_path / "sessions" session_dir.mkdir() monkeypatch.setattr(models, "SESSION_DIR", session_dir) monkeypatch.setattr(models, "SESSION_INDEX_FILE", session_dir / "_index.json") monkeypatch.setattr(models, "SESSIONS", OrderedDict()) monkeypatch.setattr(routes, "SESSION_DIR", session_dir) monkeypatch.setattr(routes, "SESSIONS", models.SESSIONS) messages = [ {"role": "user", "content": "old question"}, {"role": "assistant", "content": "same final"}, {"role": "user", "content": "new question"}, {"role": "assistant", "content": "same final"}, ] Session(session_id="anchorpersist_duplicate_ref", messages=messages).save(skip_index=True) captured = {} monkeypatch.setattr(routes, "_check_csrf", lambda handler: True) monkeypatch.setattr( routes, "read_body", lambda handler: { "session_id": "anchorpersist_duplicate_ref", "message_index": 1, "message_ref": _client_anchor_scene_message_ref(messages[3]), "scene": { "version": "activity_scene_v1", "mode": "compact_worklog", "activity_rows": [{"row_id": "tool-1", "role": "tool"}], "final_answer": "same final", }, }, ) monkeypatch.setattr( routes, "bad", lambda handler, msg, status=400: captured.update(error=msg, status=status) or True, ) assert routes.handle_post(SimpleNamespace(command="POST"), SimpleNamespace(path="/api/session/anchor-scene")) is True assert captured["status"] == 404 assert captured["error"] == "Assistant message not found" raw = json.loads((session_dir / "anchorpersist_duplicate_ref.json").read_text(encoding="utf-8")) assert not raw.get("anchor_activity_scenes") def test_anchor_scene_persistence_converts_window_index_to_full_index(tmp_path, monkeypatch): from api import models, routes from api.models import Session session_dir = tmp_path / "sessions" session_dir.mkdir() monkeypatch.setattr(models, "SESSION_DIR", session_dir) monkeypatch.setattr(models, "SESSION_INDEX_FILE", session_dir / "_index.json") monkeypatch.setattr(models, "SESSIONS", OrderedDict()) monkeypatch.setattr(routes, "SESSION_DIR", session_dir) monkeypatch.setattr(routes, "SESSIONS", models.SESSIONS) Session( session_id="anchorpersist_window", messages=[ {"role": "user", "content": "old question"}, {"role": "assistant", "content": "old final"}, {"role": "user", "content": "new question"}, {"role": "assistant", "content": "new final"}, ], ).save(skip_index=True) captured = {} monkeypatch.setattr(routes, "_check_csrf", lambda handler: True) monkeypatch.setattr( routes, "read_body", lambda handler: { "session_id": "anchorpersist_window", "message_index": 1, "message_window_index": 1, "message_offset": 2, "message_ref": "stale-ref-after-content-normalization", "scene": { "version": "activity_scene_v1", "mode": "compact_worklog", "activity_rows": [{"row_id": "tool-1", "role": "tool"}], "final_answer": "new final", }, }, ) monkeypatch.setattr( routes, "j", lambda handler, payload, status=200, extra_headers=None: captured.update( payload=payload, status=status ) or True, ) assert routes.handle_post(SimpleNamespace(command="POST"), SimpleNamespace(path="/api/session/anchor-scene")) is True assert captured["payload"]["message_index"] == 3 raw = json.loads((session_dir / "anchorpersist_window.json").read_text(encoding="utf-8")) record = next(iter(raw["anchor_activity_scenes"].values())) assert record["message_index"] == 3 assert record["message_ref"] == routes._assistant_anchor_scene_message_ref(raw["messages"][3]) def test_anchor_scene_persistence_rejects_unmatched_ref_without_index(tmp_path, monkeypatch): from api import models, routes from api.models import Session session_dir = tmp_path / "sessions" session_dir.mkdir() monkeypatch.setattr(models, "SESSION_DIR", session_dir) monkeypatch.setattr(models, "SESSION_INDEX_FILE", session_dir / "_index.json") monkeypatch.setattr(models, "SESSIONS", OrderedDict()) monkeypatch.setattr(routes, "SESSION_DIR", session_dir) monkeypatch.setattr(routes, "SESSIONS", models.SESSIONS) Session( session_id="anchorpersist_no_index", messages=[ {"role": "user", "content": "question"}, {"role": "assistant", "content": "final"}, ], ).save(skip_index=True) captured = {} monkeypatch.setattr(routes, "_check_csrf", lambda handler: True) monkeypatch.setattr( routes, "read_body", lambda handler: { "session_id": "anchorpersist_no_index", "message_ref": "missing-ref", "scene": { "version": "activity_scene_v1", "mode": "compact_worklog", "activity_rows": [{"row_id": "tool-1", "role": "tool"}], }, }, ) monkeypatch.setattr( routes, "bad", lambda handler, msg, status=400: captured.update(error=msg, status=status) or True, ) assert routes.handle_post(SimpleNamespace(command="POST"), SimpleNamespace(path="/api/session/anchor-scene")) is True assert captured["status"] == 404 assert captured["error"] == "Assistant message not found" def test_anchor_scene_persistence_rejects_ref_miss_stale_index_mismatch(tmp_path, monkeypatch): from api import models, routes from api.models import Session session_dir = tmp_path / "sessions" session_dir.mkdir() monkeypatch.setattr(models, "SESSION_DIR", session_dir) monkeypatch.setattr(models, "SESSION_INDEX_FILE", session_dir / "_index.json") monkeypatch.setattr(models, "SESSIONS", OrderedDict()) monkeypatch.setattr(routes, "SESSION_DIR", session_dir) monkeypatch.setattr(routes, "SESSIONS", models.SESSIONS) Session( session_id="anchorpersist_stale_index_mismatch", messages=[ {"role": "user", "content": "old question"}, {"role": "assistant", "content": "old final"}, {"role": "user", "content": "new question"}, {"role": "assistant", "content": "new final"}, ], ).save(skip_index=True) captured = {} monkeypatch.setattr(routes, "_check_csrf", lambda handler: True) monkeypatch.setattr( routes, "read_body", lambda handler: { "session_id": "anchorpersist_stale_index_mismatch", "message_index": 1, "message_ref": "stale-ref-after-content-normalization", "scene": { "version": "activity_scene_v1", "mode": "compact_worklog", "activity_rows": [{"row_id": "tool-1", "role": "tool"}], "final_answer": "new final", }, }, ) monkeypatch.setattr( routes, "bad", lambda handler, msg, status=400: captured.update(error=msg, status=status) or True, ) assert routes.handle_post(SimpleNamespace(command="POST"), SimpleNamespace(path="/api/session/anchor-scene")) is True assert captured["status"] == 404 assert captured["error"] == "Assistant message not found" raw = json.loads((session_dir / "anchorpersist_stale_index_mismatch.json").read_text(encoding="utf-8")) assert not raw.get("anchor_activity_scenes") def test_anchor_scene_hydration_repairs_tail_only_scene_from_full_turn(): from api import routes final_answer = ( "final answer with enough detail to be the answer and a shared verification paragraph " "about cron proxy fallback removal, 127.0.0.1:7890, direct git pulls, and durable Worklog ordering" ) stale_final_draft = ( "draft answer with enough detail to be the answer and a shared verification paragraph " "about cron proxy fallback removal, 127.0.0.1:7890, direct git pulls, and durable Worklog ordering" ) short_stale_final_draft = "cron 127.0.0.1:7890 fallback " + ("x " * 45) + "Worklog" messages = [ {"role": "user", "content": "question"}, { "role": "assistant", "content": "first progress", "reasoning": "thinking near first progress", "tool_calls": [{"id": "call-1", "function": {"name": "terminal"}}], }, {"role": "tool", "tool_call_id": "call-1", "content": "ok"}, {"role": "assistant", "content": "second progress"}, {"role": "assistant", "content": final_answer}, ] old_scene = { "version": "activity_scene_v1", "mode": "compact_worklog", "final_answer": "", "activity_rows": [ { "row_id": "tail-prose", "role": "prose", "kind": "process_prose", "source_event_type": "token", "text": "second progress", }, { "row_id": "bad-final-prefix", "role": "thinking", "kind": "reasoning", "source_event_type": "reasoning", "text": final_answer, }, { "row_id": "tail-thinking", "role": "thinking", "kind": "reasoning", "source_event_type": "reasoning", "text": "thinking near first progress", }, { "row_id": "tail-final-draft", "role": "prose", "kind": "process_prose", "source_event_type": "token", "text": stale_final_draft, }, { "row_id": "tail-short-final-draft", "role": "prose", "kind": "process_prose", "source_event_type": "token", "text": short_stale_final_draft, }, {"row_id": "done", "role": "terminal", "kind": "terminal_status", "source_event_type": "done"}, ], } records = { "record": { "message_index": 4, "message_ref": routes._assistant_anchor_scene_message_ref(messages[4]), "stream_id": "stream-1", "scene": old_scene, } } hydrated = routes._hydrate_anchor_activity_scenes( messages, records, message_offset=0, tool_calls=[ { "assistant_msg_idx": 1, "tid": "call-1", "name": "terminal", "preview": "running", "snippet": "ok", } ], ) scene = hydrated[4]["_anchor_activity_scene"] rows = scene["activity_rows"] prose_texts = [row.get("text") for row in rows if row.get("role") == "prose"] tool_ids = [row.get("tool_call_id") for row in rows if row.get("role") == "tool"] thinking_texts = [row.get("text") for row in rows if row.get("role") == "thinking"] chronological = [ (row.get("role"), row.get("text") or row.get("tool_call_id")) for row in rows if row.get("role") != "terminal" ] assert scene["final_answer"] == final_answer assert prose_texts == ["first progress", "second progress"] assert tool_ids == ["call-1"] assert thinking_texts == ["thinking near first progress"] assert chronological[:4] == [ ("prose", "first progress"), ("thinking", "thinking near first progress"), ("tool", "call-1"), ("prose", "second progress"), ] assert rows[-1]["role"] == "terminal" def test_anchor_scene_hydration_backfills_turn_duration_from_final_message(): from api import routes messages = [ {"role": "user", "content": "question"}, {"role": "assistant", "content": "final answer", "_turnDuration": 731.2}, ] records = { "record": { "message_index": 1, "message_ref": routes._assistant_anchor_scene_message_ref(messages[1]), "stream_id": "stream-1", "scene": { "version": "activity_scene_v1", "mode": "compact_worklog", "final_answer": "final answer", "activity_rows": [ {"row_id": "done", "role": "terminal", "kind": "terminal_status"} ], }, } } hydrated = routes._hydrate_anchor_activity_scenes(messages, records) assert hydrated[1]["_anchor_activity_scene"]["turn_duration"] == 731.2 def test_anchor_scene_hydration_promotes_final_content_array_tool_use_to_ordered_rows(): from api import routes messages = [ {"role": "user", "content": "question"}, { "role": "assistant", "content": [ "Let me inspect the files first.", {"type": "tool_use", "tool_use_id": "toolu_content", "tool_name": "grep", "args": {"pattern": "TODO"}}, "Found it,", {"type": "text", "text": "here's the fix."}, ], "tool_calls": [ { "id": "toolu_message", "name": "grep", "input": {"pattern": "TODO"}, "snippet": "TODO in static/messages.js", "started_at": 100, } ], }, ] records = { "record": { "message_index": 1, "message_ref": routes._assistant_anchor_scene_message_ref(messages[1]), "stream_id": "stream-1", "scene": { "version": "activity_scene_v1", "mode": "compact_worklog", "final_answer": "", "activity_rows": [ {"row_id": "done", "role": "terminal", "kind": "terminal_status", "source_event_type": "done"} ], }, } } hydrated = routes._hydrate_anchor_activity_scenes( messages, records, tool_calls=[ { "assistant_msg_idx": 1, "tid": "toolu_durable", "name": "grep", "input": {"pattern": "TODO"}, "snippet": "TODO in static/messages.js", "started_at": 100, } ], ) scene = hydrated[1]["_anchor_activity_scene"] rows = scene["activity_rows"] activity = [ (row.get("role"), row.get("text") or row.get("tool_call_id")) for row in rows if row.get("role") != "terminal" ] assert scene["final_answer"] == "Found it,\nhere's the fix." assert activity == [ ("prose", "Let me inspect the files first."), ("tool", "toolu_content"), ] assert rows[1]["tool"]["name"] == "grep" assert rows[1]["tool"]["args"] == {"pattern": "TODO"} assert rows[1]["tool"]["snippet"] == "TODO in static/messages.js" assert len([row for row in rows if row.get("role") == "tool"]) == 1 assert rows[-1]["role"] == "terminal" def test_anchor_scene_hydration_preserves_non_final_post_tool_text(): from api import routes messages = [ {"role": "user", "content": "question"}, { "role": "assistant", "content": [ "I will inspect first.", {"type": "tool_use", "tool_use_id": "toolu_content", "tool_name": "grep"}, {"type": "text", "text": "I found the relevant file."}, {"type": "thinking", "text": "Need one more check."}, ], }, {"role": "assistant", "content": "final answer"}, ] records = { "record": { "message_index": 2, "message_ref": routes._assistant_anchor_scene_message_ref(messages[2]), "stream_id": "stream-1", "scene": { "version": "activity_scene_v1", "mode": "compact_worklog", "final_answer": "final answer", "activity_rows": [], }, } } hydrated = routes._hydrate_anchor_activity_scenes(messages, records, tool_calls=[]) scene = hydrated[2]["_anchor_activity_scene"] rows = scene["activity_rows"] activity = [ (row.get("role"), row.get("text") or row.get("tool_call_id")) for row in rows ] assert scene["final_answer"] == "final answer" assert activity == [ ("prose", "I will inspect first."), ("tool", "toolu_content"), ("prose", "I found the relevant file."), ("thinking", "Need one more check."), ] def test_anchor_scene_hydration_keeps_final_tail_thinking_as_activity_only(): from api import routes messages = [ {"role": "user", "content": "question"}, { "role": "assistant", "content": [ {"type": "text", "text": "Let me check first."}, { "type": "tool_use", "tool_use_id": "toolu_content", "tool_name": "terminal", "args": {"cmd": "ls"}, }, {"type": "thinking", "text": "Tail thinking must stay activity."}, {"type": "reasoning", "reasoning": "Tail reasoning key must stay activity."}, {"type": "text", "text": "Final visible answer."}, ], "tool_calls": [ { "id": "toolu_message", "name": "terminal", "args": {"cmd": "ls"}, "snippet": "OUTPUT", "started_at": 100, } ], }, ] records = { "record": { "message_index": 1, "message_ref": routes._assistant_anchor_scene_message_ref(messages[1]), "stream_id": "stream-1", "scene": { "version": "activity_scene_v1", "mode": "compact_worklog", "final_answer": "", "activity_rows": [], }, } } hydrated = routes._hydrate_anchor_activity_scenes(messages, records, tool_calls=[]) scene = hydrated[1]["_anchor_activity_scene"] rows = scene["activity_rows"] activity = [ (row.get("role"), row.get("text") or row.get("tool_call_id")) for row in rows ] assert scene["final_answer"] == "Final visible answer." assert activity == [ ("prose", "Let me check first."), ("tool", "toolu_content"), ("thinking", "Tail thinking must stay activity."), ("thinking", "Tail reasoning key must stay activity."), ] assert rows[1]["tool"]["snippet"] == "OUTPUT" assert len([row for row in rows if row.get("role") == "tool"]) == 1 def test_anchor_scene_hydration_promotes_output_text_content_tail_to_final_answer(): from api import routes messages = [ {"role": "user", "content": "question"}, { "role": "assistant", "content": [ {"type": "text", "text": "Let me check first."}, { "type": "tool_use", "tool_use_id": "toolu_content", "tool_name": "terminal", "args": {"cmd": "ls"}, }, {"type": "output_text", "content": "Final answer from content field."}, ], "tool_calls": [ { "id": "toolu_message", "name": "terminal", "args": {"cmd": "ls"}, "snippet": "OUTPUT", "started_at": 100, } ], }, ] records = { "record": { "message_index": 1, "message_ref": routes._assistant_anchor_scene_message_ref(messages[1]), "stream_id": "stream-1", "scene": { "version": "activity_scene_v1", "mode": "compact_worklog", "final_answer": "", "activity_rows": [], }, } } hydrated = routes._hydrate_anchor_activity_scenes(messages, records, tool_calls=[]) scene = hydrated[1]["_anchor_activity_scene"] rows = scene["activity_rows"] activity = [ (row.get("role"), row.get("text") or row.get("tool_call_id")) for row in rows ] assert scene["final_answer"] == "Final answer from content field." assert activity == [ ("prose", "Let me check first."), ("tool", "toolu_content"), ] assert rows[1]["tool"]["snippet"] == "OUTPUT" assert len([row for row in rows if row.get("role") == "tool"]) == 1 def test_anchor_scene_hydration_restores_durable_body_after_message_tool_merge(): from api import routes full_output = "X" * 9000 capped_preview = full_output[:4000] messages = [ {"role": "user", "content": "question"}, { "role": "assistant", "content": [ "I will inspect first.", { "type": "tool_use", "tool_use_id": "toolu_content", "tool_name": "terminal", "args": {"cmd": "long-output"}, }, "Done.", ], "tool_calls": [ { "id": "toolu_message", "name": "terminal", "input": {"cmd": "long-output"}, "snippet": capped_preview, "started_at": 100, } ], }, ] records = { "record": { "message_index": 1, "message_ref": routes._assistant_anchor_scene_message_ref(messages[1]), "stream_id": "stream-1", "scene": { "version": "activity_scene_v1", "mode": "compact_worklog", "final_answer": "", "activity_rows": [], }, } } hydrated = routes._hydrate_anchor_activity_scenes( messages, records, tool_calls=[ { "assistant_msg_idx": 1, "tid": "toolu_durable", "name": "terminal", "input": {"cmd": "long-output"}, "snippet": full_output, "started_at": 100, } ], ) tools = [row for row in hydrated[1]["_anchor_activity_scene"]["activity_rows"] if row.get("role") == "tool"] assert len(tools) == 1 assert tools[0]["tool_call_id"] == "toolu_content" assert tools[0]["tool"]["snippet"] == full_output assert tools[0]["payload"]["snippet"] == full_output def test_anchor_scene_hydration_keeps_third_same_command_id_distinct_after_alt_id_match(): from api import routes messages = [ {"role": "user", "content": "question"}, { "role": "assistant", "content": [ "First check.", { "type": "tool_use", "tool_use_id": "content-a", "tool_name": "terminal", "args": {"cmd": "ls"}, }, "Second check.", ], "tool_calls": [ { "id": "message-a", "name": "terminal", "input": {"cmd": "ls"}, "snippet": "OUTPUT A", "started_at": 100, } ], }, ] records = { "record": { "message_index": 1, "message_ref": routes._assistant_anchor_scene_message_ref(messages[1]), "stream_id": "stream-1", "scene": { "version": "activity_scene_v1", "mode": "compact_worklog", "final_answer": "", "activity_rows": [], }, } } hydrated = routes._hydrate_anchor_activity_scenes( messages, records, tool_calls=[ { "assistant_msg_idx": 1, "tid": "durable-b", "name": "terminal", "input": {"cmd": "ls"}, "snippet": "OUTPUT B", "started_at": 200, } ], ) tools = [row for row in hydrated[1]["_anchor_activity_scene"]["activity_rows"] if row.get("role") == "tool"] by_id = {row.get("tool_call_id"): row for row in tools} assert by_id["content-a"]["tool"]["snippet"] == "OUTPUT A" assert "message-a" not in by_id assert by_id["durable-b"]["tool"]["snippet"] == "OUTPUT B" assert len(tools) == 2 def test_anchor_scene_hydration_keeps_identical_output_repeat_distinct_after_alt_id_match(): from api import routes messages = [ {"role": "user", "content": "question"}, { "role": "assistant", "content": [ "First check.", { "type": "tool_use", "tool_use_id": "content-a", "tool_name": "terminal", "args": {"cmd": "ls"}, }, "Second check.", ], "tool_calls": [ { "id": "message-a", "name": "terminal", "input": {"cmd": "ls"}, "snippet": "SAME OUTPUT", } ], }, ] records = { "record": { "message_index": 1, "message_ref": routes._assistant_anchor_scene_message_ref(messages[1]), "stream_id": "stream-1", "scene": { "version": "activity_scene_v1", "mode": "compact_worklog", "final_answer": "", "activity_rows": [], }, } } hydrated = routes._hydrate_anchor_activity_scenes( messages, records, tool_calls=[ { "assistant_msg_idx": 1, "tid": "durable-b", "name": "terminal", "input": {"cmd": "ls"}, "snippet": "SAME OUTPUT", } ], ) tools = [row for row in hydrated[1]["_anchor_activity_scene"]["activity_rows"] if row.get("role") == "tool"] by_id = {row.get("tool_call_id"): row for row in tools} assert by_id["content-a"]["tool"]["snippet"] == "SAME OUTPUT" assert "message-a" not in by_id assert by_id["durable-b"]["tool"]["snippet"] == "SAME OUTPUT" assert len(tools) == 2 def test_anchor_scene_hydration_keeps_same_started_at_repeat_distinct_after_alt_id_match(): from api import routes messages = [ {"role": "user", "content": "question"}, { "role": "assistant", "content": [ "First check.", { "type": "tool_use", "tool_use_id": "content-a", "tool_name": "terminal", "args": {"cmd": "ls"}, }, "Second check.", ], "tool_calls": [ { "id": "message-a", "name": "terminal", "input": {"cmd": "ls"}, "snippet": "OUTPUT A", "started_at": 100, } ], }, ] records = { "record": { "message_index": 1, "message_ref": routes._assistant_anchor_scene_message_ref(messages[1]), "stream_id": "stream-1", "scene": { "version": "activity_scene_v1", "mode": "compact_worklog", "final_answer": "", "activity_rows": [], }, } } hydrated = routes._hydrate_anchor_activity_scenes( messages, records, tool_calls=[ { "assistant_msg_idx": 1, "tid": "durable-b", "name": "terminal", "input": {"cmd": "ls"}, "snippet": "OUTPUT B", "started_at": 100, } ], ) tools = [row for row in hydrated[1]["_anchor_activity_scene"]["activity_rows"] if row.get("role") == "tool"] by_id = {row.get("tool_call_id"): row for row in tools} assert by_id["content-a"]["tool"]["snippet"] == "OUTPUT A" assert "message-a" not in by_id assert by_id["durable-b"]["tool"]["snippet"] == "OUTPUT B" assert len(tools) == 2 def test_anchor_scene_hydration_keeps_short_persisted_body_after_durable_merge(): from api import routes full_output = "short output line\nwith more detail that came later" short_body = "short output line" messages = [ {"role": "user", "content": "question"}, { "role": "assistant", "content": [ "I will inspect first.", { "type": "tool_use", "tool_use_id": "toolu_content", "tool_name": "terminal", "args": {"cmd": "short-output"}, }, "Done.", ], "tool_calls": [ { "id": "toolu_message", "name": "terminal", "input": {"cmd": "short-output"}, "snippet": short_body, "started_at": 100, } ], }, ] records = { "record": { "message_index": 1, "message_ref": routes._assistant_anchor_scene_message_ref(messages[1]), "stream_id": "stream-1", "scene": { "version": "activity_scene_v1", "mode": "compact_worklog", "final_answer": "", "activity_rows": [], }, } } hydrated = routes._hydrate_anchor_activity_scenes( messages, records, tool_calls=[ { "assistant_msg_idx": 1, "tid": "toolu_durable", "name": "terminal", "input": {"cmd": "short-output"}, "snippet": full_output, "started_at": 100, } ], ) tools = [row for row in hydrated[1]["_anchor_activity_scene"]["activity_rows"] if row.get("role") == "tool"] assert len(tools) == 1 assert tools[0]["tool"]["snippet"] == short_body assert tools[0]["payload"]["snippet"] == short_body def test_anchor_scene_hydration_merges_missing_args_after_content_tool_match(): from api import routes messages = [ {"role": "user", "content": "question"}, { "role": "assistant", "content": [ "I will patch the file.", { "type": "tool_use", "tool_use_id": "patch-content", "tool_name": "edit_file", "args": {"path": "x.py"}, }, "Done.", ], "tool_calls": [ { "id": "patch-message", "name": "edit_file", "input": { "path": "x.py", "old_string": "old", "new_string": "new", }, "snippet": "@@ -1 +1 @@\n-old\n+new", } ], }, ] records = { "record": { "message_index": 1, "message_ref": routes._assistant_anchor_scene_message_ref(messages[1]), "stream_id": "stream-1", "scene": { "version": "activity_scene_v1", "mode": "compact_worklog", "final_answer": "", "activity_rows": [], }, } } hydrated = routes._hydrate_anchor_activity_scenes(messages, records, tool_calls=[]) tools = [ row for row in hydrated[1]["_anchor_activity_scene"]["activity_rows"] if row.get("role") == "tool" ] assert len(tools) == 1 assert tools[0]["tool"]["args"] == { "path": "x.py", "old_string": "old", "new_string": "new", } assert tools[0]["payload"]["args"] == { "path": "x.py", "old_string": "old", "new_string": "new", } def test_anchor_scene_hydration_keeps_consumed_different_name_tool_distinct(): from api import routes messages = [ {"role": "user", "content": "edit then inspect"}, { "role": "assistant", "content": [ "Edit the file.", { "type": "tool_use", "tool_use_id": "content-a", "tool_name": "edit_file", "args": {"path": "x.py"}, }, "Inspect it.", ], "tool_calls": [ { "id": "content-a", "name": "edit_file", "input": {"path": "x.py"}, "snippet": "EDITED", } ], }, ] records = { "record": { "message_index": 1, "message_ref": routes._assistant_anchor_scene_message_ref(messages[1]), "stream_id": "stream-1", "scene": { "version": "activity_scene_v1", "mode": "compact_worklog", "final_answer": "", "activity_rows": [], }, } } hydrated = routes._hydrate_anchor_activity_scenes( messages, records, tool_calls=[ { "assistant_msg_idx": 1, "tid": "live-b", "name": "terminal", "input": {"path": "x.py"}, "snippet": "TERMINAL OUTPUT", } ], ) tools = [ row for row in hydrated[1]["_anchor_activity_scene"]["activity_rows"] if row.get("role") == "tool" ] by_id = {row["tool"]["id"]: row for row in tools} assert by_id["content-a"]["tool"]["name"] == "edit_file" assert by_id["content-a"]["tool"]["snippet"] == "EDITED" assert by_id["live-b"]["tool"]["name"] == "terminal" assert by_id["live-b"]["tool"]["snippet"] == "TERMINAL OUTPUT" def test_anchor_scene_hydration_does_not_position_merge_ambiguous_different_id_tools(): from api import routes messages = [ {"role": "user", "content": "question"}, { "role": "assistant", "content": [ "First check.", {"type": "tool_use", "tool_use_id": "content-a", "tool_name": "terminal"}, "Second check.", {"type": "tool_use", "tool_use_id": "content-b", "tool_name": "terminal"}, "Done.", ], }, ] records = { "record": { "message_index": 1, "message_ref": routes._assistant_anchor_scene_message_ref(messages[1]), "stream_id": "stream-1", "scene": { "version": "activity_scene_v1", "mode": "compact_worklog", "final_answer": "", "activity_rows": [], }, } } hydrated = routes._hydrate_anchor_activity_scenes( messages, records, tool_calls=[ {"assistant_msg_idx": 1, "tid": "message-b", "name": "terminal", "snippet": "OUTPUT B"}, {"assistant_msg_idx": 1, "tid": "message-a", "name": "terminal", "snippet": "OUTPUT A"}, ], ) rows = hydrated[1]["_anchor_activity_scene"]["activity_rows"] tools = [row for row in rows if row.get("role") == "tool"] by_id = {row.get("tool_call_id"): row for row in tools} assert by_id["content-a"]["tool"]["snippet"] == "" assert by_id["content-b"]["tool"]["snippet"] == "" assert by_id["message-a"]["tool"]["snippet"] == "OUTPUT A" assert by_id["message-b"]["tool"]["snippet"] == "OUTPUT B" def test_anchor_scene_hydration_does_not_name_merge_remaining_same_name_tool_after_exact_match(): from api import routes messages = [ {"role": "user", "content": "question"}, { "role": "assistant", "content": [ "First check.", {"type": "tool_use", "tool_use_id": "content-a", "tool_name": "terminal"}, "Second check.", {"type": "tool_use", "tool_use_id": "content-b", "tool_name": "terminal"}, "Done.", ], "tool_calls": [ {"id": "content-a", "name": "terminal", "snippet": "OUTPUT A"}, {"id": "message-b", "name": "terminal", "snippet": "OUTPUT B"}, ], }, ] records = { "record": { "message_index": 1, "message_ref": routes._assistant_anchor_scene_message_ref(messages[1]), "stream_id": "stream-1", "scene": { "version": "activity_scene_v1", "mode": "compact_worklog", "final_answer": "", "activity_rows": [], }, } } hydrated = routes._hydrate_anchor_activity_scenes(messages, records, tool_calls=[]) rows = hydrated[1]["_anchor_activity_scene"]["activity_rows"] tools = [row for row in rows if row.get("role") == "tool"] by_id = {row.get("tool_call_id"): row for row in tools} assert by_id["content-a"]["tool"]["snippet"] == "OUTPUT A" assert by_id["content-b"]["tool"]["snippet"] == "" assert by_id["message-b"]["tool"]["snippet"] == "OUTPUT B" def test_anchor_scene_hydration_merges_remaining_matching_tool_after_exact_match(): from api import routes messages = [ {"role": "user", "content": "question"}, { "role": "assistant", "content": [ "First check.", { "type": "tool_use", "tool_use_id": "content-a", "tool_name": "terminal", "args": {"cmd": "ls"}, }, "Second check.", { "type": "tool_use", "tool_use_id": "content-b", "tool_name": "terminal", "args": {"cmd": "pwd"}, }, "Done.", ], "tool_calls": [ {"id": "content-a", "name": "terminal", "input": {"cmd": "ls"}, "snippet": "OUTPUT A"}, {"id": "message-b", "name": "terminal", "input": {"cmd": "pwd"}, "snippet": "OUTPUT B"}, ], }, ] records = { "record": { "message_index": 1, "message_ref": routes._assistant_anchor_scene_message_ref(messages[1]), "stream_id": "stream-1", "scene": { "version": "activity_scene_v1", "mode": "compact_worklog", "final_answer": "", "activity_rows": [], }, } } hydrated = routes._hydrate_anchor_activity_scenes(messages, records, tool_calls=[]) rows = hydrated[1]["_anchor_activity_scene"]["activity_rows"] tools = [row for row in rows if row.get("role") == "tool"] by_id = {row.get("tool_call_id"): row for row in tools} assert by_id["content-a"]["tool"]["snippet"] == "OUTPUT A" assert by_id["content-b"]["tool"]["snippet"] == "OUTPUT B" assert "message-b" not in by_id assert len(tools) == 2 def test_anchor_scene_hydration_keeps_distinct_used_singleton_tool_call(): from api import routes messages = [ {"role": "user", "content": "question"}, { "role": "assistant", "content": [ "First check.", { "type": "tool_use", "tool_use_id": "content-a", "tool_name": "terminal", "args": {"cmd": "ls"}, }, "Second check.", ], "tool_calls": [ { "id": "content-a", "name": "terminal", "input": {"cmd": "ls"}, "snippet": "OUTPUT A", } ], }, ] records = { "record": { "message_index": 1, "message_ref": routes._assistant_anchor_scene_message_ref(messages[1]), "stream_id": "stream-1", "scene": { "version": "activity_scene_v1", "mode": "compact_worklog", "final_answer": "", "activity_rows": [], }, } } hydrated = routes._hydrate_anchor_activity_scenes( messages, records, tool_calls=[ { "assistant_msg_idx": 1, "tid": "durable-b", "name": "terminal", "input": {"cmd": "pwd"}, "snippet": "OUTPUT B", } ], ) rows = hydrated[1]["_anchor_activity_scene"]["activity_rows"] tools = [row for row in rows if row.get("role") == "tool"] by_id = {row.get("tool_call_id"): row for row in tools} assert by_id["content-a"]["tool"]["snippet"] == "OUTPUT A" assert by_id["durable-b"]["tool"]["snippet"] == "OUTPUT B" assert len(tools) == 2 def test_anchor_scene_hydration_keeps_same_command_used_singleton_tool_distinct(): from api import routes messages = [ {"role": "user", "content": "question"}, { "role": "assistant", "content": [ "First check.", { "type": "tool_use", "tool_use_id": "content-a", "tool_name": "terminal", "args": {"cmd": "ls"}, }, "Second check.", ], "tool_calls": [ { "id": "content-a", "name": "terminal", "input": {"cmd": "ls"}, "snippet": "OUTPUT A", } ], }, ] records = { "record": { "message_index": 1, "message_ref": routes._assistant_anchor_scene_message_ref(messages[1]), "stream_id": "stream-1", "scene": { "version": "activity_scene_v1", "mode": "compact_worklog", "final_answer": "", "activity_rows": [], }, } } hydrated = routes._hydrate_anchor_activity_scenes( messages, records, tool_calls=[ { "assistant_msg_idx": 1, "tid": "durable-b", "name": "terminal", "input": {"cmd": "ls"}, "snippet": "OUTPUT B", } ], ) rows = hydrated[1]["_anchor_activity_scene"]["activity_rows"] tools = [row for row in rows if row.get("role") == "tool"] by_id = {row.get("tool_call_id"): row for row in tools} assert by_id["content-a"]["tool"]["snippet"] == "OUTPUT A" assert by_id["durable-b"]["tool"]["snippet"] == "OUTPUT B" assert len(tools) == 2 def test_anchor_scene_hydration_keeps_anonymous_used_singleton_tool_distinct(): from api import routes messages = [ {"role": "user", "content": "question"}, { "role": "assistant", "content": [ "First check.", { "type": "tool_use", "tool_use_id": "content-a", "tool_name": "terminal", "args": {"cmd": "ls"}, }, "Second check.", ], "tool_calls": [ { "id": "content-a", "name": "terminal", "input": {"cmd": "ls"}, "snippet": "OUTPUT A", } ], }, ] records = { "record": { "message_index": 1, "message_ref": routes._assistant_anchor_scene_message_ref(messages[1]), "stream_id": "stream-1", "scene": { "version": "activity_scene_v1", "mode": "compact_worklog", "final_answer": "", "activity_rows": [], }, } } hydrated = routes._hydrate_anchor_activity_scenes( messages, records, tool_calls=[ { "assistant_msg_idx": 1, "name": "terminal", "input": {"cmd": "ls"}, "snippet": "OUTPUT B", } ], ) rows = hydrated[1]["_anchor_activity_scene"]["activity_rows"] tools = [row for row in rows if row.get("role") == "tool"] snippets = sorted(row["tool"]["snippet"] for row in tools) assert len(tools) == 2 assert snippets == ["OUTPUT A", "OUTPUT B"] def test_anchor_scene_hydration_keeps_body_only_distinct_used_singleton_tool_call(): from api import routes messages = [ {"role": "user", "content": "question"}, { "role": "assistant", "content": [ "First check.", { "type": "tool_use", "tool_use_id": "content-a", "tool_name": "terminal", "args": {"cmd": "ls"}, }, "Second check.", ], "tool_calls": [ { "id": "content-a", "name": "terminal", "input": {"cmd": "ls"}, "snippet": "OUTPUT A", } ], }, ] records = { "record": { "message_index": 1, "message_ref": routes._assistant_anchor_scene_message_ref(messages[1]), "stream_id": "stream-1", "scene": { "version": "activity_scene_v1", "mode": "compact_worklog", "final_answer": "", "activity_rows": [], }, } } hydrated = routes._hydrate_anchor_activity_scenes( messages, records, tool_calls=[ { "assistant_msg_idx": 1, "tid": "durable-b", "name": "terminal", "snippet": "OUTPUT B", } ], ) rows = hydrated[1]["_anchor_activity_scene"]["activity_rows"] tools = [row for row in rows if row.get("role") == "tool"] by_id = {row.get("tool_call_id"): row for row in tools} assert by_id["content-a"]["tool"]["snippet"] == "OUTPUT A" assert by_id["durable-b"]["tool"]["snippet"] == "OUTPUT B" assert len(tools) == 2 def test_anchor_scene_hydration_does_not_name_merge_singleton_with_conflicting_args(): from api import routes messages = [ {"role": "user", "content": "question"}, { "role": "assistant", "content": [ "Patch a.py.", { "type": "tool_use", "tool_use_id": "content-a", "tool_name": "edit_file", "args": {"path": "a.py"}, }, ], "tool_calls": [ { "id": "message-b", "name": "edit_file", "input": {"path": "b.py"}, "snippet": "PATCH B", } ], }, ] records = { "record": { "message_index": 1, "message_ref": routes._assistant_anchor_scene_message_ref(messages[1]), "stream_id": "stream-1", "scene": { "version": "activity_scene_v1", "mode": "compact_worklog", "final_answer": "", "activity_rows": [], }, } } hydrated = routes._hydrate_anchor_activity_scenes(messages, records, tool_calls=[]) rows = hydrated[1]["_anchor_activity_scene"]["activity_rows"] tools = [row for row in rows if row.get("role") == "tool"] by_id = {row.get("tool_call_id"): row for row in tools} assert by_id["content-a"]["tool"]["args"] == {"path": "a.py"} assert by_id["content-a"]["tool"]["snippet"] == "" assert by_id["message-b"]["tool"]["args"] == {"path": "b.py"} assert by_id["message-b"]["tool"]["snippet"] == "PATCH B" assert len(tools) == 2 def test_anchor_scene_hydration_dedupes_compression_lifecycle_rows(): from api import routes messages = [ {"role": "user", "content": "question"}, {"role": "assistant", "content": "thinking text", "reasoning": "thinking text"}, {"role": "assistant", "content": "final answer"}, ] records = { "record": { "message_index": 2, "message_ref": routes._assistant_anchor_scene_message_ref(messages[2]), "stream_id": "stream-1", "scene": { "version": "activity_scene_v1", "mode": "compact_worklog", "final_answer": "final answer", "activity_rows": [ { "row_id": "compressing-1", "role": "lifecycle", "kind": "lifecycle_status", "source_event_type": "compressing", "status": "running", "text": "Compressing context", }, { "row_id": "compressing-2", "role": "lifecycle", "kind": "lifecycle_status", "source_event_type": "compressing", "status": "running", "text": "Compressing context", }, { "row_id": "compressed", "role": "lifecycle", "kind": "lifecycle_status", "source_event_type": "compressed", "status": "completed", "text": "Context auto-compressed", }, {"row_id": "done", "role": "terminal", "kind": "terminal_status", "source_event_type": "done"}, ], }, } } hydrated = routes._hydrate_anchor_activity_scenes(messages, records) rows = hydrated[2]["_anchor_activity_scene"]["activity_rows"] compression_rows = [ row for row in rows if row.get("role") == "lifecycle" and row.get("source_event_type") in {"compressing", "compressed"} ] assert len(compression_rows) == 1 assert compression_rows[0]["source_event_type"] == "compressed" assert compression_rows[0]["order_index"] == compression_rows[0]["seq"] def test_anchor_scene_hydration_drops_stale_live_running_thinking_when_settled_thinking_exists(): from api import routes messages = [ {"role": "user", "content": "question"}, { "role": "assistant", "content": "process update", "reasoning": "settled thinking from transcript", }, {"role": "assistant", "content": "final answer"}, ] records = { "record": { "message_index": 2, "message_ref": routes._assistant_anchor_scene_message_ref(messages[2]), "stream_id": "stream-1", "scene": { "version": "activity_scene_v1", "mode": "compact_worklog", "final_answer": "final answer", "activity_rows": [ { "row_id": "live-reasoning:stream-1:2", "local_id": "live-reasoning:stream-1:2", "role": "thinking", "kind": "reasoning", "source_event_type": "reasoning", "status": "running", "text": "stale live reasoning text", }, {"row_id": "done", "role": "terminal", "kind": "terminal_status", "source_event_type": "done"}, ], }, } } hydrated = routes._hydrate_anchor_activity_scenes(messages, records) rows = hydrated[2]["_anchor_activity_scene"]["activity_rows"] thinking_rows = [row for row in rows if row.get("role") == "thinking"] assert [row.get("text") for row in thinking_rows] == ["settled thinking from transcript"] assert not any(str(row.get("row_id") or "").startswith("live-reasoning:") for row in rows) assert not any( row.get("role") in {"thinking", "prose", "tool"} and row.get("status") == "running" for row in rows ) def test_anchor_scene_hydration_seals_unmatched_live_running_activity_rows(): from api import routes messages = [ {"role": "user", "content": "question"}, {"role": "assistant", "content": "final answer"}, ] records = { "record": { "message_index": 1, "message_ref": routes._assistant_anchor_scene_message_ref(messages[1]), "stream_id": "stream-1", "scene": { "version": "activity_scene_v1", "mode": "compact_worklog", "final_answer": "final answer", "activity_rows": [ { "row_id": "live-reasoning:stream-1:1", "local_id": "live-reasoning:stream-1:1", "role": "thinking", "kind": "reasoning", "source_event_type": "reasoning", "status": "running", "text": "only live reasoning", }, { "row_id": "live-prose:stream-1:1", "local_id": "live-prose:stream-1:1", "role": "prose", "kind": "process_prose", "source_event_type": "token", "status": "running", "text": "only live prose", }, { "row_id": "tool:call-1:0", "local_id": "call-1", "role": "tool", "kind": "tool_started", "source_event_type": "tool", "status": "running", "run_id": "run-1", "stream_id": "stream-1", "identity": { "local_id": "call-1", "run_id": "run-1", "stream_id": "stream-1", }, "group": {"group_key": "segment:1", "activity_segment_seq": 1}, "tool_call_id": "call-1", "tool": {"id": "call-1", "name": "terminal", "done": False}, "payload": {"tid": "call-1", "status": "running", "done": False}, }, {"row_id": "done", "role": "terminal", "kind": "terminal_status", "source_event_type": "done"}, ], }, } } hydrated = routes._hydrate_anchor_activity_scenes(messages, records) rows = hydrated[1]["_anchor_activity_scene"]["activity_rows"] activity_rows = [row for row in rows if row.get("role") in {"thinking", "prose", "tool"}] assert [row.get("status") for row in activity_rows] == ["completed", "completed", "completed"] assert [row.get("role") for row in activity_rows] == ["thinking", "prose", "tool"] tool_row = activity_rows[-1] assert tool_row["tool"]["done"] is True assert tool_row["payload"]["status"] == "completed" assert tool_row["payload"]["done"] is True def test_anchor_scene_settlement_does_not_reclassify_transcript_owned_tool_row(): from api import routes row = { "row_id": "hydrated:stream-1:tool:call-1", "local_id": "call-1", "role": "tool", "status": "running", "stream_id": "stream-1", "group": {"assistant_msg_idx": 1}, "tool": {"id": "call-1", "done": False}, "payload": {"tid": "call-1", "status": "running", "done": False}, } assert routes._anchor_scene_settle_live_running_row(row, has_settled_thinking=False) is row assert row["tool"]["done"] is False assert row["payload"]["done"] is False def test_runtime_journal_anchor_scene_matches_settled_hydrated_visible_semantics(tmp_path, monkeypatch): """Runtime journal replay and settled read hydration must preserve the same visible anchor activity semantics for one turn. Compared path: _run_journal_live_snapshot(...).anchor_activity_scene -> persisted anchor_activity_scenes record -> _hydrate_anchor_activity_scenes(...)._anchor_activity_scene. """ from api import models, routes from api.run_journal import RunJournalWriter session_dir = tmp_path / "sessions" session_dir.mkdir() monkeypatch.setattr(models, "SESSION_DIR", session_dir) monkeypatch.setattr(routes, "SESSION_DIR", session_dir) session_id = "anchorparity1" stream_id = "stream-parity-1" process_before_tool = "first progress" thinking = "thinking through plan" process_after_tool = "checkpoint tail" final_answer = "Final answer: keep the activity above this answer." writer = RunJournalWriter(session_id, stream_id, session_dir=session_dir) writer.append_sse_event("token", {"text": process_before_tool}) writer.append_sse_event("reasoning", {"text": thinking}) writer.append_sse_event( "tool", {"name": "terminal", "tid": "call-1", "args": {"command": "pytest"}}, ) writer.append_sse_event( "tool_complete", {"name": "terminal", "tid": "call-1", "preview": "ok"}, ) writer.append_sse_event("token", {"text": f" {process_after_tool}"}) runtime_snapshot = routes._run_journal_live_snapshot(stream_id) assert runtime_snapshot is not None runtime_scene = runtime_snapshot.get("anchor_activity_scene") assert isinstance(runtime_scene, dict) assert _anchor_scene_visible_semantics(runtime_scene) == [ {"role": "prose", "kind": "process_prose", "text": process_before_tool}, {"role": "thinking", "kind": "reasoning", "text": thinking}, { "role": "tool", "kind": "tool_completed", "status": "completed", "tool_call_id": "call-1", "name": "terminal", "args": {"command": "pytest"}, "done": True, }, {"role": "prose", "kind": "process_prose", "text": process_after_tool}, ] persisted_scene = json.loads(json.dumps(runtime_scene)) persisted_scene["activity_rows"].append( { "row_id": "terminal-done", "role": "terminal", "kind": "terminal_status", "source_event_type": "done", "status": "completed", "text": "Response complete", } ) messages = [ {"role": "user", "content": "question"}, { "role": "assistant", "content": process_before_tool, "reasoning": thinking, "tool_calls": [ { "id": "call-1", "name": "terminal", "args": {"command": "pytest"}, "preview": "pytest", "snippet": "ok", } ], }, {"role": "tool", "tool_call_id": "call-1", "content": "ok"}, {"role": "assistant", "content": process_after_tool}, {"role": "assistant", "content": final_answer}, ] message_ref = routes._assistant_anchor_scene_message_ref(messages[4]) records = { message_ref: { "version": "anchor_activity_scene_record_v1", "message_index": 4, "message_ref": message_ref, "stream_id": stream_id, "scene": persisted_scene, } } hydrated = routes._hydrate_anchor_activity_scenes(messages, records) settled_scene = hydrated[4]["_anchor_activity_scene"] assert settled_scene["final_answer"] == final_answer assert final_answer not in [ row.get("text") for row in settled_scene["activity_rows"] if row.get("role") != "terminal" ] assert _anchor_scene_visible_semantics(settled_scene) == _anchor_scene_visible_semantics(runtime_scene) assert _anchor_scene_visible_semantics(settled_scene, include_terminal=True)[-1] == { "role": "terminal", "kind": "terminal_status", "source_event_type": "done", "status": "completed", } assert not any( row.get("role") in {"prose", "thinking", "tool"} and row.get("status") == "running" for row in settled_scene["activity_rows"] ) def test_runtime_journal_snapshot_includes_live_anchor_activity_scene(monkeypatch): from api import routes stream_id = "stream-live-scene" events = [ { "event": "token", "seq": 1, "event_id": f"{stream_id}:1", "created_at": 1.0, "payload": {"text": "first progress"}, }, { "event": "reasoning", "seq": 2, "event_id": f"{stream_id}:2", "created_at": 2.0, "payload": {"text": "thinking through plan"}, }, { "event": "interim_assistant", "seq": 3, "event_id": f"{stream_id}:3", "created_at": 3.0, "payload": {"text": "checkpoint"}, }, { "event": "tool", "seq": 4, "event_id": f"{stream_id}:4", "created_at": 4.0, "payload": {"name": "terminal", "tid": "call-1", "args": {"command": "pytest"}}, }, { "event": "tool_complete", "seq": 5, "event_id": f"{stream_id}:5", "created_at": 5.0, "payload": {"name": "terminal", "tid": "call-1", "preview": "ok"}, }, { "event": "token", "seq": 6, "event_id": f"{stream_id}:6", "created_at": 6.0, "payload": {"text": " tail"}, }, ] monkeypatch.setattr( routes, "find_run_summary", lambda sid: { "session_id": "session-live-scene", "last_seq": 6, "last_event_id": f"{stream_id}:6", }, ) monkeypatch.setattr( routes, "read_run_events", lambda session_id, run_id: {"events": events}, ) snapshot = routes._run_journal_live_snapshot(stream_id) scene = snapshot["anchor_activity_scene"] rows = scene["activity_rows"] assert scene["version"] == "activity_scene_v1" assert snapshot["last_assistant_text"] == "first progress\n\ncheckpoint tail" assert snapshot["last_reasoning_text"] == "thinking through plan" assert [row["role"] for row in rows] == ["prose", "thinking", "tool", "prose"] assert rows[0]["local_id"] == f"live-prose:{stream_id}:1" assert rows[1]["local_id"] == f"live-thinking:{stream_id}:1" assert rows[1]["thinking"]["text"] == "thinking through plan" assert rows[2]["tool_call_id"] == "call-1" assert rows[2]["tool"]["done"] is True assert rows[3]["status"] == "running" def test_runtime_journal_snapshot_dedupes_reasoning_interim_progress_echo(monkeypatch): from api import routes stream_id = "stream-live-reasoning-interim-echo" progress = "我先检查当前仓库状态，然后定位重复渲染路径。" events = [ { "event": "reasoning", "seq": 1, "event_id": f"{stream_id}:1", "created_at": 1.0, "payload": {"text": progress}, }, { "event": "interim_assistant", "seq": 2, "event_id": f"{stream_id}:2", "created_at": 2.0, "payload": {"text": progress}, }, ] monkeypatch.setattr( routes, "find_run_summary", lambda sid: { "session_id": "session-live-reasoning-interim-echo", "last_seq": 2, "last_event_id": f"{stream_id}:2", }, ) monkeypatch.setattr( routes, "read_run_events", lambda session_id, run_id: {"events": events}, ) snapshot = routes._run_journal_live_snapshot(stream_id) rows = snapshot["anchor_activity_scene"]["activity_rows"] assert snapshot["last_assistant_text"] == progress assert snapshot["last_reasoning_text"] == "" assert [row["role"] for row in rows] == ["prose"] assert rows[0]["text"] == progress def test_runtime_journal_snapshot_has_running_anchor_row_before_first_token(monkeypatch): from api import routes stream_id = "stream-live-empty" monkeypatch.setattr( routes, "find_run_summary", lambda sid: { "session_id": "session-live-empty", "last_seq": 1, "last_event_id": f"{stream_id}:1", }, ) monkeypatch.setattr( routes, "read_run_events", lambda session_id, run_id: { "events": [ { "event": "context_status", "seq": 1, "event_id": f"{stream_id}:1", "created_at": 1.0, "payload": {"session_id": "session-live-empty"}, } ] }, ) snapshot = routes._run_journal_live_snapshot(stream_id) rows = snapshot["anchor_activity_scene"]["activity_rows"] assert rows assert rows[0]["role"] == "lifecycle" assert rows[0]["status"] == "running"