"""Bounded, read-only Agent History: normalization, cursors, frames, binding.""" import asyncio import json import pytest from device_gateway.control.history import ( HISTORY_MAX_FRAME_BYTES, HISTORY_MAX_MESSAGES, HISTORY_MAX_TEXT_BYTES, build_page, decode_cursor, encode_cursor, normalize_codex_turn, normalize_pi_messages, build_view_page) def test_view_full_utf8_fragments_are_reversible_and_redacted(): text=('开始🙂测试。'*400)+' password=private_value '+('结束'*300) source=[dict(role='assistant',text=text),dict(role='user',text='prompt')] pages=[];cursor=None while True: page=build_view_page(source,cursor,agent='codex',generation=1) assert len(json.dumps(page,ensure_ascii=False).encode())<3000 pages.append(page) cursor=page['next_cursor'] if not cursor:break assert len(pages)>5 reconstructed=''.join(p['messages'][0]['text'] for p in pages[:-1]) assert reconstructed==text.replace('password=private_value','[redacted]') for i in range(len(pages)-1,0,-1): previous=build_view_page(source,pages[i]['previous_cursor'],agent='codex',generation=1) assert previous==pages[i-1] def test_current_keeps_all_native_messages_without_reasoning(): current=turn([{'type':'userMessage','text':'prompt'}, {'type':'agentMessage','text':'first'*400}, {'type':'reasoning','text':'hidden'}, {'type':'agentMessage','text':'second'*400}],status='inProgress') assert [m['text'] for m in normalize_codex_turn(current,current=True)]==['prompt','first'*400,'second'*400] # ---- cursor ----------------------------------------------------------------- def test_cursor_roundtrip_and_rejection(): assert decode_cursor(None) == 0 assert decode_cursor("") == 0 for offset in (0, 1, 4, 1000): assert decode_cursor(encode_cursor(offset)) == offset # Opaque: the device must never interpret it, so malformed input is rejected. for bad in ("!!!!", "not_base64", "a" * 300, "aGVsbG86MQ"): assert decode_cursor(bad) is None # ---- Codex normalization ---------------------------------------------------- def turn(items, status="completed", items_view="full"): return {"status": status, "itemsView": items_view, "items": items} def test_codex_normalization_only_user_and_assistant(): items = [ {"type": "reasoning", "text": "secret chain of thought"}, {"type": "userMessage", "content": [{"type": "text", "text": "fix the network"}]}, {"type": "functionCall", "text": "rm -rf /"}, {"type": "agentMessage", "phase": "commentary", "text": "thinking out loud"}, {"type": "agentMessage", "phase": "final_answer", "text": "done, tests pass"}, {"type": "fileChange", "text": "diff --git"}, ] normalized = normalize_codex_turn(turn(items)) assert [m["role"] for m in normalized] == ["user", "assistant"] assert normalized[0]["text"] == "fix the network" assert normalized[1]["text"] == "done, tests pass" # Reasoning / tool / tool output / file changes never appear. blob = json.dumps(normalized) for forbidden in ("secret chain of thought", "rm -rf", "thinking out loud", "diff --git"): assert forbidden not in blob def test_codex_user_message_reads_content_parts(): # Real Codex schema: UserMessageThreadItem has content[] (no flat text). # Multiple text parts are joined in native order; non-text attachments are # references, not prose, and must not enter the read-only history. item = {"type": "userMessage", "content": [ {"type": "text", "text": "first line"}, {"type": "localImage", "path": "/tmp/secret.png"}, {"type": "text", "text": "second line"}, {"type": "mention", "name": "docs", "path": "/repo/docs"}, {"type": "image", "url": "https://example.test/x.png"}, ]} normalized = normalize_codex_turn(turn([item, {"type": "agentMessage", "text": "answer"}])) assert normalized[0]["role"] == "user" assert normalized[0]["text"] == "first line\nsecond line" blob = json.dumps(normalized) for forbidden in ("/tmp/secret.png", "mention", "example.test", "/repo/docs"): assert forbidden not in blob # A user message with only non-text parts contributes no prose. assert normalize_codex_turn(turn([{"type": "userMessage", "content": [ {"type": "image", "url": "https://example.test/y.png"}]}])) == [] # The legacy flat `text` form still works (older native build). legacy = normalize_codex_turn(turn([{"type": "userMessage", "text": "legacy prompt"}])) assert legacy[0]["text"] == "legacy prompt" def test_codex_assistant_prefers_final_then_last_non_commentary(): # No final_answer: the last non-commentary agentMessage wins. only_commentary = turn([ {"type": "agentMessage", "phase": "commentary", "text": "noise"}, {"type": "agentMessage", "text": "real answer"}, ]) assert normalize_codex_turn(only_commentary)[0]["text"] == "real answer" # final_answer wins over a later non-final message. final_present = turn([ {"type": "agentMessage", "text": "draft"}, {"type": "agentMessage", "phase": "final_answer", "text": "final"}, {"type": "agentMessage", "text": "trailing"}, ]) assert normalize_codex_turn(final_present)[0]["text"] == "final" def test_codex_incomplete_and_non_completed_turns_skipped(): assert normalize_codex_turn(turn([{"type": "userMessage", "content": [ {"type": "text", "text": "hi"}]}], status="inProgress")) == [] assert normalize_codex_turn(turn([{"type": "userMessage", "content": [ {"type": "text", "text": "hi"}]}], items_view="summary")) == [] # ---- Pi normalization ------------------------------------------------------- def test_pi_normalization_only_user_and_assistant(): messages = [ {"role": "system", "content": "system prompt"}, {"role": "user", "content": "hello"}, {"role": "assistant", "content": [{"type": "text", "text": "hi"}, {"type": "toolCall", "name": "bash"}]}, {"role": "toolResult", "content": [{"type": "text", "text": "command output"}]}, {"role": "assistant", "content": [{"type": "reasoning", "text": "hidden"}]}, ] normalized = normalize_pi_messages(messages) assert [m["role"] for m in normalized] == ["user", "assistant"] assert normalized[1]["text"] == "hi" blob = json.dumps(normalized) for forbidden in ("system prompt", "bash", "command output", "hidden"): assert forbidden not in blob # ---- page bounds ------------------------------------------------------------ def messages(count): return [{"role": "user" if i % 2 == 0 else "assistant", "text": f"message {i}"} for i in range(count)] def test_page_is_newest_first_and_bounded(): all_messages = messages(10) page = build_page(all_messages, None, agent="pi", generation=3) assert len(page["messages"]) == HISTORY_MAX_MESSAGES assert page["messages"][0]["text"] == "message 0" # newest first assert page["next_cursor"] assert not page["oldest"] frame = json.dumps({"messages": page["messages"], "next_cursor": page["next_cursor"]}, ensure_ascii=False, separators=(',', ':')) assert len(frame.encode()) <= HISTORY_MAX_FRAME_BYTES def test_cursor_pagination_reaches_oldest_once(): all_messages = messages(9) seen = [] cursor = None for _ in range(10): page = build_page(all_messages, cursor, agent="codex", generation=1) seen.extend(m["text"] for m in page["messages"]) if page["oldest"]: break cursor = page["next_cursor"] assert seen == [m["text"] for m in all_messages] # Past the newest end the cursor is invalid, not silently clamped. from device_gateway.control.models import ControlError with pytest.raises(ControlError): build_page(all_messages, encode_cursor(999), agent="pi", generation=1) def test_long_text_frame_stays_bounded_and_utf8_safe(): # Four very long Chinese messages must still fit one bounded frame. blob = "中文测试内容。" * 400 all_messages = [{"role": "assistant", "text": blob} for _ in range(4)] page = build_page(all_messages, None, agent="pi", generation=1) frame = json.dumps({"messages": page["messages"], "next_cursor": page["next_cursor"]}, ensure_ascii=False, separators=(',', ':')) assert len(frame.encode()) <= HISTORY_MAX_FRAME_BYTES for message in page["messages"]: text = message["text"] assert len(text.encode()) <= HISTORY_MAX_TEXT_BYTES + 3 text.encode("utf-8") # must be valid UTF-8, never a split glyph def test_credentials_redacted(): all_messages = [ {"role": "user", "text": "use Bearer abcdef0123456789 to call it"}, {"role": "assistant", "text": "api_key=sk-abcdefghijklmnop done"}, ] page = build_page(all_messages, None, agent="pi", generation=1) blob = json.dumps(page["messages"]) assert "abcdef0123456789" not in blob assert "sk-abcdefghijklmnop" not in blob assert "[redacted]" in blob # ---- Navigator binding ------------------------------------------------------ from types import SimpleNamespace from device_gateway.control.models import AgentSession, ControlError, Project from device_gateway.control.navigator import Navigator from device_gateway.control.operations import Operations from device_gateway.control.projects import ProjectRegistry from device_gateway.control.user_input import UserInputs from device_gateway.state import TargetStore class HistoryAgent: def __init__(self, agent, path, items=None, history=True): self.agent_type = agent self.agent = agent self.path = str(path) self._messages = items if items is not None else messages(6) if not history: self.read_history = None async def get_session(self, sid): if not str(sid).startswith("s"): raise ControlError("Session disappeared", 404) return AgentSession(agent=self.agent, session_id=sid, project_id="p", name="n", cwd=self.path, updated_at="2026-10-05T00:00:00Z") async def list_sessions(self, project): return [await self.get_session("s1")] async def read_history(self, sid, max_messages=64): return self._messages[:max_messages] def build(tmp_path, agent_messages=None): projects = ProjectRegistry(tmp_path / "projects.json") projects.add(Project(id="p", name="Demo", path=str(tmp_path))) targets = TargetStore(tmp_path / "targets.json") agent = HistoryAgent("codex", tmp_path, agent_messages) navigator = Navigator(projects, {"codex": agent}, targets, Operations(tmp_path / "o.db"), UserInputs(), lambda d, e: None, lambda d: True, lambda target: {}, lambda d: None) return navigator, targets, agent def device(generation_target="agent:codex:s1"): return SimpleNamespace(device_id="d", connection_id="c", send_lock=asyncio.Lock()) def test_view_snapshot_keeps_position_when_native_history_changes(tmp_path): async def run(): navigator,targets,agent=build(tmp_path,[dict(role='assistant',text='old'*1000),dict(role='user',text='prompt')]) targets.set('d','agent:codex:s1','p') d=device();d.capabilities={'view-pagination'} event=dict(target='agent:codex:s1',agent='codex',session_id='s1', expected_target_generation=targets.generation('d'),scope='current') first=await navigator.history(d,event) agent._messages=[dict(role='assistant',text='new response')] second=await navigator.history(d,dict(event,cursor=first['next_cursor'])) assert second['messages'][0]['text'] in 'old'*1000 fresh=await navigator.history(d,event) assert fresh['messages'][0]['text']=='new response' asyncio.run(run()) def test_history_requires_target_generation_and_identity(tmp_path): async def run(): navigator, targets, _ = build(tmp_path) targets.set("d", "agent:codex:s1", "p") generation = targets.generation("d") d = device() event = dict(target="agent:codex:s1", agent="codex", session_id="s1", expected_target_generation=generation, request_id="r1") page = await navigator.history(d, event) assert page["type"] == "history_page" assert page["agent"] == "codex" and page["session_id"] == "s1" assert len(page["messages"]) <= HISTORY_MAX_MESSAGES # A stale generation is rejected. with pytest.raises(ControlError): await navigator.history(d, dict(event, expected_target_generation=generation + 1)) # A different target is rejected even at the current generation. with pytest.raises(ControlError): await navigator.history(d, dict(event, target="agent:codex:other")) # A mismatched agent / session identity is rejected. with pytest.raises(ControlError): await navigator.history(d, dict(event, session_id="other")) with pytest.raises(ControlError): await navigator.history(d, dict(event, agent="pi")) asyncio.run(run()) def test_history_absent_capability_is_reported_not_crashed(tmp_path): async def run(): navigator, targets, agent = build(tmp_path) agent.read_history = None targets.set("d", "agent:codex:s1", "p") generation = targets.generation("d") with pytest.raises(ControlError): await navigator.history(device(), dict(target="agent:codex:s1", agent="codex", session_id="s1", expected_target_generation=generation, request_id="r1")) asyncio.run(run()) def test_history_target_change_during_read_is_ignored(tmp_path): async def run(): navigator, targets, agent = build(tmp_path) targets.set("d", "agent:codex:s1", "p") generation = targets.generation("d") original = agent.read_history async def moving(sid, max_messages=64): # The target moves while the native read is in flight. targets.set("d", "agent:codex:s2", "p") return await original(sid, max_messages) agent.read_history = moving with pytest.raises(ControlError): await navigator.history(device(), dict(target="agent:codex:s1", agent="codex", session_id="s1", expected_target_generation=generation, request_id="r1")) asyncio.run(run()) def test_history_handler_emits_bounded_page_and_error(tmp_path): async def run(): projects = ProjectRegistry(tmp_path / "projects.json") projects.add(Project(id="p", name="Demo", path=str(tmp_path))) targets = TargetStore(tmp_path / "targets.json") agent = HistoryAgent("codex", tmp_path) sent = [] async def send(d, event): sent.append(event) navigator = Navigator(projects, {"codex": agent}, targets, Operations(tmp_path / "o.db"), UserInputs(), send, lambda d: True, lambda target: {}, lambda d: None) targets.set("d", "agent:codex:s1", "p") generation = targets.generation("d") d = device() d.capabilities = {"history"} await navigator.handle(d, dict(type="history_get", target="agent:codex:s1", agent="codex", session_id="s1", expected_target_generation=generation, request_id="r1")) assert sent and sent[0]["type"] == "history_page" assert sent[0]["request_id"] == "r1" frame = json.dumps(sent[0], ensure_ascii=False).encode() assert len(frame) <= HISTORY_MAX_FRAME_BYTES # A stale generation yields history_error, not navigator_error. sent.clear() await navigator.handle(d, dict(type="history_get", target="agent:codex:s1", agent="codex", session_id="s1", expected_target_generation=generation + 5, request_id="r2")) assert sent and sent[0]["type"] == "history_error" asyncio.run(run()) # ---- Native adapter readers ------------------------------------------------- def test_codex_adapter_read_history_pages_turns(tmp_path): from device_gateway.control.codex import CodexAdapter class RPC: generation = 1 def __init__(self): self.calls = [] async def start(self): pass async def write(self, value): pass async def close(self): pass async def call(self, method, params): self.calls.append((method, params)) if method == "initialize": return {} cursor = params.get("cursor") if cursor is None: return {"data": [turn([ {"type": "userMessage", "content": [{"type": "text", "text": "second question"}]}, {"type": "agentMessage", "phase": "final_answer", "text": "second answer"}, ])], "nextCursor": "page2"} return {"data": [turn([ {"type": "userMessage", "content": [{"type": "text", "text": "first question"}]}, {"type": "agentMessage", "phase": "final_answer", "text": "first answer"}, ])], "nextCursor": None} async def run(): projects = ProjectRegistry(tmp_path / "projects.json") projects.add(Project(id="p", name="P", path=str(tmp_path))) adapter = CodexAdapter(projects, RPC()) items = await adapter.read_history("s1", max_messages=64) # Newest first, and only user/assistant (never the turn item types). assert [i["role"] for i in items] == ["assistant", "user", "assistant", "user"] assert items[0]["text"] == "second answer" assert items[-1]["text"] == "first question" asyncio.run(run()) def test_pi_adapter_read_history_normalizes_and_does_not_resume(tmp_path): from device_gateway.control.pi import PiAdapter from device_gateway.integrations.pi import PiSessionRegistry class RPC: def __init__(self): self.calls = [] async def call(self, method, params): self.calls.append((method, params)) if method == "list": return [dict(session_id="s1", name="N", cwd=str(tmp_path), updated_at="2026-10-05T00:00:00Z", live=False, state="DORMANT", recent_text="", final_text="", latest_activity="")] if method == "history": assert params["cwd"] == str(tmp_path) return {"messages": [ {"role": "user", "content": "q"}, {"role": "toolResult", "content": [{"type": "text", "text": "tool output"}]}, {"role": "assistant", "content": [{"type": "text", "text": "a"}]}, ]} raise AssertionError(method) async def run(): rpc = RPC() projects = ProjectRegistry(tmp_path / "projects.json") projects.add(Project(id="p", name="P", path=str(tmp_path))) adapter = PiAdapter(projects, PiSessionRegistry(), rpc) items = await adapter.read_history("s1") # The SDK returns oldest -> newest; the adapter must normalize to # newest-first before the unified history layer sees it. assert [i["role"] for i in items] == ["assistant", "user"] assert items[0]["text"] == "a" assert items[1]["text"] == "q" # Read-only: no resume, no create. assert all(method in ("list", "history") for method, _ in rpc.calls) asyncio.run(run()) def test_pi_adapter_history_is_newest_first(tmp_path): """The Pi SDK tail is oldest-first; the adapter must reverse it.""" from device_gateway.control.pi import PiAdapter from device_gateway.integrations.pi import PiSessionRegistry class RPC: async def call(self, method, params): if method == "list": return [dict(session_id="s1", name="N", cwd=str(tmp_path), updated_at="2026-10-05T00:00:00Z", live=False, state="DORMANT", recent_text="", final_text="", latest_activity="")] if method == "history": return {"messages": [ {"role": "user", "content": "oldest"}, {"role": "assistant", "content": [{"type": "text", "text": "middle"}]}, {"role": "user", "content": "newest"}, ]} raise AssertionError(method) async def run(): projects = ProjectRegistry(tmp_path / "projects.json") projects.add(Project(id="p", name="P", path=str(tmp_path))) adapter = PiAdapter(projects, PiSessionRegistry(), RPC()) items = await adapter.read_history("s1") assert [i["text"] for i in items] == ["newest", "middle", "oldest"] asyncio.run(run()) # ---- adapter -> pagination integration -------------------------------------- class _HistoryRPC: """Minimal Pi SDK host stand-in returning one oldest-first native tail.""" def __init__(self, messages): self.messages = messages async def call(self, method, params): if method == "list": return [dict(session_id="s1", name="N", cwd=self.cwd, updated_at="2026-10-05T00:00:00Z", live=False, state="DORMANT", recent_text="", final_text="", latest_activity="")] if method == "history": return {"messages": self.messages} raise AssertionError(method) def test_adapter_to_pagination_full_chain(tmp_path): """native adapter -> normalized history -> pagination, end to end. The first page must be the newest message, and Older pages must walk strictly backwards in time until the oldest message is reached exactly once. """ from device_gateway.control.pi import PiAdapter from device_gateway.integrations.pi import PiSessionRegistry count = 9 native = [{"role": "user" if i % 2 == 0 else "assistant", "content": f"m{i}"} for i in range(count)] # oldest (m0) -> newest (m8) async def run(): rpc = _HistoryRPC(native) rpc.cwd = str(tmp_path) projects = ProjectRegistry(tmp_path / "projects.json") projects.add(Project(id="p", name="P", path=str(tmp_path))) adapter = PiAdapter(projects, PiSessionRegistry(), rpc) normalized = await adapter.read_history("s1", max_messages=64) assert normalized[0]["text"] == f"m{count - 1}" # newest first assert normalized[-1]["text"] == "m0" # Feed the normalized list straight into the unified pagination layer. seen, cursor, pages = [], None, 0 for _ in range(10): page = build_page(normalized, cursor, agent="pi", generation=1) seen.extend(m["text"] for m in page["messages"]) pages += 1 if page["oldest"]: break cursor = page["next_cursor"] # Every native message appears exactly once, strictly newest -> oldest. assert seen == [f"m{i}" for i in reversed(range(count))] assert pages == 3 # 4 + 4 + 1 messages asyncio.run(run()) def test_codex_adapter_to_pagination_full_chain(tmp_path): """Codex content[] -> normalized history -> pagination, end to end.""" from device_gateway.control.codex import CodexAdapter class RPC: generation = 1 async def start(self): pass async def write(self, value): pass async def close(self): pass async def call(self, method, params): if method == "initialize": return {} # Newest turn first, older turn on the second page. if params.get("cursor") is None: return {"data": [turn([ {"type": "userMessage", "content": [{"type": "text", "text": "newest q"}]}, {"type": "agentMessage", "phase": "final_answer", "text": "newest a"}, ])], "nextCursor": "p2"} return {"data": [turn([ {"type": "userMessage", "content": [{"type": "text", "text": "older q"}]}, {"type": "agentMessage", "phase": "final_answer", "text": "older a"}, ])], "nextCursor": None} async def run(): projects = ProjectRegistry(tmp_path / "projects.json") projects.add(Project(id="p", name="P", path=str(tmp_path))) adapter = CodexAdapter(projects, RPC()) normalized = await adapter.read_history("s1", max_messages=64) # User text is not lost, assistant text is not lost, order is newest-first. assert [m["text"] for m in normalized] == ["newest a", "newest q", "older a", "older q"] page = build_page(normalized, None, agent="codex", generation=1) assert [m["text"] for m in page["messages"]] == ["newest a", "newest q", "older a", "older q"] assert page["oldest"] and not page["next_cursor"] asyncio.run(run())