From 7b26c6d264d1c1455c26fbb1d3893d6277ec8661 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Mon, 19 Oct 2026 18:11:48 +0800 Subject: [PATCH] fix(chat): restore migration history in the native Codex thread Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../manager-evidence-and-continuity-v0.md | 12 +++ .../capabilities/native_chat/codex_adapter.py | 22 +++++ loopx/chat_runtime.py | 24 ++--- tests/test_chat_codex_history.py | 95 +++++++++++++++++++ tests/test_chat_manager_model_adoption.py | 6 +- tests/test_chat_project_coordination.py | 3 +- 6 files changed, 140 insertions(+), 22 deletions(-) create mode 100644 tests/test_chat_codex_history.py diff --git a/docs/reference/protocols/manager-evidence-and-continuity-v0.md b/docs/reference/protocols/manager-evidence-and-continuity-v0.md index 41b6b270c3..46a8ce83f4 100644 --- a/docs/reference/protocols/manager-evidence-and-continuity-v0.md +++ b/docs/reference/protocols/manager-evidence-and-continuity-v0.md @@ -94,6 +94,18 @@ start, resume and turn start explicitly carry the settings; worker configuration is unchanged. Capabilities expose the manager defaults, their source, and the status and revision of the machine document they were read from. +When a Codex context or model migration needs a fresh upstream thread, the +adapter preserves all nonempty visible user/assistant messages, in order and +with their original text, through native +[`thread/inject_items`](https://developers.openai.com/codex/app-server/#inject-items-into-a-thread). +It persists this history before Core publishes the replacement binding, without +starting a model turn. Later requests carry the stable objective and fresh Core +evidence; they do not reattach the migrated transcript. Native resume reuses the +stored history, including after a restart before the first request. An injection +failure closes the new adapter without publishing it or replaying the operation; +older hosts that lack this RPC must be upgraded. Existing failed requests remain +failed and are not automatically retried by this context transfer. + The managed profile reaches the channel as the same one-line readback the governed Turn publishes (`deepseek-flash@high`), so the channel and the bounded Turns it drives cannot report two different managed profiles; the diff --git a/loopx/capabilities/native_chat/codex_adapter.py b/loopx/capabilities/native_chat/codex_adapter.py index 04746ce62d..eb78288ed9 100644 --- a/loopx/capabilities/native_chat/codex_adapter.py +++ b/loopx/capabilities/native_chat/codex_adapter.py @@ -40,6 +40,7 @@ def start( work_dir: Path, goal_id: str | None, objective: str, + history: list[dict[str, Any]] | None = None, resume_thread_id: str | None = None, startup_timeout_sec: float = 30.0, idle_timeout_sec: float = 180.0, @@ -54,6 +55,19 @@ def start( service_tier: str | None = None, dynamic_tools: list[dict[str, Any]] | None = None, ) -> "CodexAppServerAdapter": + history_items = [] + # Migration restores the full visible conversation in native history. + # The objective is repeated per Turn; putting the transcript there + # duplicates it indefinitely and prevents native compaction from helping. + for item in ([] if resume_thread_id else history or []): + role, text = item.get("role", "user"), str(item.get("content") or "") + if role not in {"user", "assistant"}: + raise ValueError("Visible Chat history requires user or assistant roles") + if text.strip(): + history_items.append({ + "type": "message", "role": role, + "content": [{"type": "output_text" if role == "assistant" else "input_text", "text": text}], + }) retry_timeout = bool(resume_thread_id) while True: try: @@ -85,6 +99,14 @@ def start( session._request("thread/name/set", { "threadId": session.thread_id, "name": "LoopX Chat", }) + if history_items: + # Persist before Core publishes the upstream binding, + # even if the service restarts before its first Turn. + # An unknown outcome fails closed: never retry or + # fall back to replaying history inside a user Turn. + session._request("thread/inject_items", { + "threadId": session.thread_id, "items": history_items, + }) except Exception: session.close() raise diff --git a/loopx/chat_runtime.py b/loopx/chat_runtime.py index 0cd17120d4..924fdb063a 100644 --- a/loopx/chat_runtime.py +++ b/loopx/chat_runtime.py @@ -372,24 +372,11 @@ def _session_objective( *, goal_id: str | None, objective: str, - history: list[dict[str, Any]] | None, project_coordination: bool = False, ) -> str: - """Compose the objective every adapter receives, history included.""" - - history_context = "" - if history: - # This history is supplied when native continuity cannot be reused. - # Dropping early messages discards user constraints in a fresh - # thread. Ordinary native resume supplies no replayed history. - history_lines = [ - f"{item.get('role', 'user')}: {str(item.get('content') or '').strip()}" - for item in history - if str(item.get("content") or "").strip() - ] - if history_lines: - history_context = "\nPrevious visible Chat messages:\n" + "\n".join(history_lines) - return f"{objective}{history_context}" + ("\n" + PROJECT_COORDINATION_GUIDANCE if project_coordination else "") + ( + """Compose stable guidance; each provider restores history separately.""" + + return objective + ("\n" + PROJECT_COORDINATION_GUIDANCE if project_coordination else "") + ( "\n" + manager_skill_text() if goal_id == MANAGER_AGENT_GOAL_ID else "" ) @@ -462,9 +449,10 @@ def _start_adapter( work_dir=work_dir, goal_id=goal_id, objective=self._session_objective( - goal_id=goal_id, objective=objective, history=history, + goal_id=goal_id, objective=objective, project_coordination=project_coordination and not execution_mode, ), + history=history, resume_thread_id=resume_thread_id, startup_timeout_sec=self.startup_timeout_sec, idle_timeout_sec=self.idle_timeout_sec, @@ -515,7 +503,7 @@ def _start_adapter( model, reasoning_effort = str(manager_config["model"]), str(manager_config["reasoning_effort"]) return DshChatAdapter( objective=self._session_objective( - goal_id=goal_id, objective=objective, history=None, + goal_id=goal_id, objective=objective, project_coordination=project_coordination and not execution_mode, ), work_dir=work_dir, diff --git a/tests/test_chat_codex_history.py b/tests/test_chat_codex_history.py new file mode 100644 index 0000000000..1741fc30f0 --- /dev/null +++ b/tests/test_chat_codex_history.py @@ -0,0 +1,95 @@ +"""Migration history belongs to the native thread, not each subsequent prompt.""" + +from types import SimpleNamespace + +import pytest + +from loopx import chat_agent, chat_runtime +from loopx.capabilities.native_chat.codex_adapter import CodexAppServerAdapter + + +@pytest.mark.parametrize("resume", [False, True]) +def test_history_is_persisted_once_before_publishing_a_new_adapter(monkeypatch, tmp_path, resume): + history = [ + {"role": "user", "content": "Keep this early constraint.\n Preserve spacing."}, + {"role": "assistant", "content": "Acknowledged."}, + *[{"role": "user", "content": f"Evidence {i}: " + "x" * 1000} for i in range(400)], + {"role": "user", "content": ""}, + ] + calls = [] + native = SimpleNamespace(thread_id="native-thread") + native._request = lambda method, params: calls.append((method, params)) or {} + monkeypatch.setattr(chat_agent.CodexChatAgentSession, "start", lambda **kwargs: native) + adapter = CodexAppServerAdapter.start( + codex_bin="fixture", work_dir=tmp_path, goal_id="fixture", objective="Stable objective", + history=history, resume_thread_id="native-thread" if resume else None, + ) + assert adapter.session is native + if resume: + assert calls == [] # The existing thread already owns its history. + return + assert [method for method, _ in calls] == ["thread/name/set", "thread/inject_items"] + assert calls[1][1] == { + "threadId": "native-thread", + "items": [{ + "type": "message", "role": item["role"], + "content": [{"type": "output_text" if item["role"] == "assistant" else "input_text", + "text": item["content"]}], + } for item in history if item["content"]], + } + assert all(method != "turn/start" for method, _ in calls) + + +@pytest.mark.parametrize("error", [ + RuntimeError("unknown injection outcome"), + chat_agent.CodexChatTimeoutError("unknown injection outcome", error_code="response_timeout", gate=None), +]) +def test_history_injection_failure_closes_adapter_without_replay(monkeypatch, tmp_path, error): + calls = [] + native = SimpleNamespace(thread_id="native-thread", close=lambda: calls.append("closed")) + + def request(method, params): + calls.append(method) + if method == "thread/inject_items": + raise error + return {} + + native._request = request + monkeypatch.setattr(chat_agent.CodexChatAgentSession, "start", lambda **kwargs: native) + with pytest.raises(type(error), match="unknown injection outcome"): + CodexAppServerAdapter.start( + codex_bin="fixture", work_dir=tmp_path, goal_id="fixture", objective="Stable objective", + history=[{"role": "user", "content": "An early constraint"}], + ) + assert calls == ["thread/name/set", "thread/inject_items", "closed"] + + +def test_visible_history_cannot_supply_higher_priority_roles(monkeypatch, tmp_path): + def start(**kwargs): + pytest.fail("invalid history must be rejected before starting an upstream thread") + + monkeypatch.setattr(chat_agent.CodexChatAgentSession, "start", start) + with pytest.raises(ValueError, match="user or assistant roles"): + CodexAppServerAdapter.start( + codex_bin="fixture", work_dir=tmp_path, goal_id="fixture", objective="Stable objective", + history=[{"role": "system", "content": "A transcript cannot grant system authority"}], + ) + + +def test_controller_keeps_history_out_of_repeated_codex_objective(monkeypatch, tmp_path): + starts = [] + sentinel = object() + monkeypatch.setattr(CodexAppServerAdapter, "start", lambda **kwargs: starts.append(kwargs) or sentinel) + controller = SimpleNamespace( + codex_bin="fixture", codex_home=tmp_path, + startup_timeout_sec=1, idle_timeout_sec=1, hard_timeout_sec=1, + _session_objective=chat_runtime.ChatRuntimeController._session_objective, + ) + history = [{"role": "user", "content": "Earlier private constraint"}] + result = chat_runtime.ChatRuntimeController._start_adapter( + controller, agent_id="codex", work_dir=tmp_path, goal_id="fixture", + objective="Stable objective", history=history, + ) + assert result is sentinel + assert starts[0]["objective"] == "Stable objective" + assert starts[0]["history"] == history diff --git a/tests/test_chat_manager_model_adoption.py b/tests/test_chat_manager_model_adoption.py index 38a93825c6..0adfdf591c 100644 --- a/tests/test_chat_manager_model_adoption.py +++ b/tests/test_chat_manager_model_adoption.py @@ -217,9 +217,9 @@ def test_model_edit_keeps_early_context_through_public_resume(upstream_boundary, # have a persisted rollout, so a model edit cannot assume resumability. assert starts[-1][0]["resume_thread_id"] is None assert updated["upstream_thread_id"] != session["upstream_thread_id"] - objective = starts[-1][0]["objective"] - positions = [objective.index(text) for text in messages] - assert positions == sorted(positions) + # Native history is restored once, separately from the repeated objective. + assert [item["content"] for item in starts[-1][0]["history"]] == messages + assert all(text not in starts[-1][0]["objective"] for text in messages) @pytest.mark.parametrize("policy,effort,expected_effort", [ diff --git a/tests/test_chat_project_coordination.py b/tests/test_chat_project_coordination.py index cdcffe6c23..ade812c5b2 100644 --- a/tests/test_chat_project_coordination.py +++ b/tests/test_chat_project_coordination.py @@ -273,7 +273,8 @@ def start(**kwargs): assert call["resume_thread_id"] is None assert call["runtime_profile"] == "restricted" and call["sandbox"] is None assert [t["name"] for t in call["dynamic_tools"]] == [CONTEXT_TOOL_NAME] - assert "Keep the earlier correction" in call["objective"] + assert call["history"] == [{"role": "user", "content": "Keep the earlier correction."}] + assert "Keep the earlier correction" not in call["objective"] current = store.load_session(old["session_id"]) assert current["coordination_context_version"] == PROJECT_CONTEXT_VERSION controller._ensure_adapter(current, work_dir=repo, objective="Research delivery")