"""Existing-session reply enrichment through native API; no rollout-file reads.""" import asyncio from copy import deepcopy import pytest import device_gateway.app as gateway from device_gateway.control.codex import CodexAdapter from device_gateway.control.models import ControlError, Project from device_gateway.control.projects import ProjectRegistry def message(text, phase=None, kind="agentMessage"): return dict(id=text, type=kind, text=text, phase=phase) def turn(*items, status="completed"): return dict(id="turn", status=status, items=list(items), itemsView="full") class Native: generation = 1 def __init__(self, cwd, pages): self.cwd, self.pages = str(cwd), pages self.revision = 1 self.status = "notLoaded" self.calls = [] self.failure = None self.entered = self.release = None 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 {} sid = params["threadId"] if method == "thread/read": assert params["includeTurns"] is False return {"thread": dict(id=sid, cwd=self.cwd, updatedAt=self.revision, model="native-model", reasoningEffort="high", status={"type": self.status})} assert method == "thread/turns/list" assert params["sortDirection"] == "desc" and params["itemsView"] == "full" assert params["limit"] == 8 if self.entered: self.entered.set() await self.release.wait() if self.failure: raise self.failure return deepcopy(self.pages.get((sid, params.get("cursor")), {"data": []})) def history_calls(self): return sum(method == "thread/turns/list" for method, _ in self.calls) def setup(tmp_path, pages): projects = ProjectRegistry(tmp_path / "projects.json") projects.add(Project(id="p", name="P", path=str(tmp_path))) native = Native(tmp_path, pages) return CodexAdapter(projects, native), native @pytest.mark.parametrize("items,expected", [ ([message("user", kind="userMessage"), message("tool", kind="commandExecution"), message("thinking", kind="reasoning"), message("progress", "commentary"), message("FINAL", "final_answer"), message("unknown later")], "FINAL"), ([message("earlier"), message("fallback"), message("progress", "commentary")], "fallback"), ([message("progress", "commentary"), message("tool", kind="fileChange")], ""), ([message("", "final_answer"), message("legacy reply")], "legacy reply"), ([], ""), ]) def test_dormant_final_fallback_empty(tmp_path, items, expected): async def run(): adapter, native = setup(tmp_path, {("s", None): {"data": [turn(*items)]}}) s = await adapter.get_session("s") assert s.state == "DORMANT" and s.final_text == expected assert s.model == "native-model" and s.effort == "high" assert native.history_calls() == 1 and not adapter.live asyncio.run(run()) def test_latest_incomplete_user_only_and_commentary_skip_to_previous_completed(tmp_path): async def run(): pages = {("s", None): {"data": [turn(message("cancelled", "final_answer"), status="interrupted"), turn(message("failed", "final_answer"), status="failed"), turn(message("user", kind="userMessage")), turn(message("progress", "commentary"))], "nextCursor": "older"}, ("s", "older"): {"data": [turn(message("last valid", "final_answer")), turn(message("old"))]}} adapter, native = setup(tmp_path, pages) assert (await adapter.get_session("s")).final_text == "last valid" assert native.history_calls() == 2 asyncio.run(run()) def test_snapshot_switch_restart_and_revision_cache(tmp_path, monkeypatch): async def run(): pages = {(sid, None): {"data": [turn(message("Reply " + sid, "final_answer"))]} for sid in ("A", "B")} adapter, native = setup(tmp_path, pages) monkeypatch.setattr(gateway, "adapters", {"codex": adapter}) for sid in ("A", "B", "A", "A", "A"): snapshot = await gateway.agent_target_payload("agent:codex:" + sid) assert snapshot["state"] == "DORMANT" and snapshot["final_text"] == "Reply " + sid assert snapshot["session_id"] == sid assert native.history_calls() == 2 # A=1, B=1 across repeated authoritative pushes. native.revision += 1 native.pages[("A", None)] = {"data": [turn(message("updated A"))]} assert (await adapter.get_session("A")).final_text == "updated A" assert native.history_calls() == 3 fresh = CodexAdapter(adapter.projects, native) assert not fresh.live and not fresh.last_replies assert (await fresh.get_session("A")).final_text == "updated A" assert native.history_calls() == 4 asyncio.run(run()) def test_live_precedence_terminal_and_event_invalidation(tmp_path): async def run(): adapter, native = setup(tmp_path, {("s", None): {"data": [turn(message("OLD"))]}}) assert (await adapter.get_session("s")).final_text == "OLD" native.status = "idle" await adapter.event(dict(method="turn/started", params={"threadId": "s", "turn": {"id": "new"}})) assert "s" not in adapter.last_replies running = await adapter.get_session("s") assert running.state == "RUNNING" and running.final_text == "" await adapter.event(dict(method="item/agentMessage/delta", params={"threadId": "s", "turnId": "new", "delta": "NEW"})) assert (await adapter.get_session("s")).recent_text == "NEW" for state in ("WAITING_APPROVAL", "WAITING_INPUT"): adapter.live["s"]["state"] = state s = await adapter.get_session("s") assert s.state == state and s.final_text == "" for status, state in (("completed", "DONE"), ("failed", "FAILED")): await adapter.event(dict(method="turn/started", params={"threadId": "s", "turn": {"id": status}})) await adapter.event(dict(method="turn/completed", params={"threadId": "s", "turn": dict(id=status, status=status, items=[message("NEW", "final_answer")])})) adapter.last_replies["s"] = (native.revision, "OLD") s = await adapter.get_session("s") assert s.state == state and s.final_text == "NEW" assert native.history_calls() == 1 native.status = "active" s = await adapter.get_session("s") assert s.state == "RUNNING" and s.final_text == "" asyncio.run(run()) def test_live_event_during_recovery_cannot_publish_old_reply(tmp_path): async def run(): adapter, native = setup(tmp_path, {("s", None): {"data": [turn(message("OLD"))]}}) native.entered, native.release = asyncio.Event(), asyncio.Event() pending = asyncio.create_task(adapter.get_session("s")) await native.entered.wait() await adapter.event(dict(method="turn/started", params={"threadId": "s", "turn": {"id": "new"}})) native.release.set() s = await pending assert s.state == "RUNNING" and s.final_text == "" asyncio.run(run()) def test_terminal_event_during_recovery_wins_over_stale_native_metadata(tmp_path): async def run(): adapter, native = setup(tmp_path, {("s", None): {"data": [turn(message("OLD"))]}}) native.entered, native.release = asyncio.Event(), asyncio.Event() pending = asyncio.create_task(adapter.get_session("s")) await native.entered.wait() await adapter.event(dict(method="turn/started", params={"threadId": "s", "turn": {"id": "new"}})) await adapter.event(dict(method="turn/completed", params={"threadId": "s", "turn": dict(id="new", status="completed", items=[message("NEW", "final_answer")])})) native.release.set() s = await pending assert s.state == "DORMANT" and s.final_text == "NEW" asyncio.run(run()) def test_timeout_is_optional_and_idle_current_output_is_preserved(tmp_path): async def run(): adapter, native = setup(tmp_path, {}) adapter.LAST_REPLY_TIMEOUT = 0.01 native.entered, native.release = asyncio.Event(), asyncio.Event() s = await adapter.get_session("s") assert s.state == "DORMANT" and s.final_text == "" and native.history_calls() == 1 native.status = "idle" adapter.live["s"] = dict(state="IDLE", recent_text="current output") s = await adapter.get_session("s") assert s.state == "IDLE" and s.recent_text == "current output" and s.final_text == "" assert native.history_calls() == 1 asyncio.run(run()) @pytest.mark.parametrize("failure", [ControlError("secret native error"), ValueError("private response")]) def test_failure_preserves_metadata_and_warning_is_bounded_cached(tmp_path, failure, caplog): async def run(): adapter, native = setup(tmp_path, {}) native.failure = failure for _ in range(3): s = await adapter.get_session("s") assert s.state == "DORMANT" and s.final_text == "" and s.model == "native-model" assert native.history_calls() == 1 asyncio.run(run()) assert caplog.text.count("last reply unavailable") == 1 assert "secret native error" not in caplog.text and "private response" not in caplog.text def test_concurrent_reads_utf8_cache_and_scan_limits(tmp_path): async def run(): adapter, native = setup(tmp_path, {("s", None): {"data": [turn(message("δΈ­ζ–‡πŸ™‚" * 3000 + " token=private-key"))]}}) sessions = await asyncio.gather(*(adapter.get_session("s") for _ in range(6))) assert native.history_calls() == 1 assert all(0 < len(s.final_text.encode()) <= 2048 and "\ufffd" not in s.final_text and "private-key" not in s.final_text for s in sessions) for i in range(130): await adapter.get_session(str(i)) assert len(adapter.last_replies) == 128 and "s" not in adapter.last_replies for i in range(5): native.pages[("scan", None if i == 0 else str(i))] = {"data": [turn()], "nextCursor": str(i + 1)} before = native.history_calls() assert (await adapter.get_session("scan")).final_text == "" assert native.history_calls() - before == 4 asyncio.run(run()) @pytest.mark.parametrize("response", [{}, {"data": "unsupported"}, {"data": [dict(items=[])], "nextCursor": 7}, {"data": [dict(status="completed", items=[], itemsView="summary")]}]) def test_unsupported_history_response_is_optional(tmp_path, response): async def run(): adapter, native = setup(tmp_path, {("s", None): response}) s = await adapter.get_session("s") assert s.state == "DORMANT" and s.final_text == "" asyncio.run(run())