fix(chat): compact restored native history before the first question

docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md

@@ -1693,9 +1693,24 @@ Session authority, queue, worker or feature toggle is introduced.
 A delivered failure notice is not a completed business request. The existing
 private reply scanner now discovers eligible unfinished ordinary Codex steward
 requests after the legacy manager-context migration. Context version 21 restores
-the complete visible user/assistant history once in native persisted history;
-the old upstream identity remains recorded, and native Goal/attached executions
-are not replaced by this migration.
+the complete visible user/assistant history in native persisted history. Large
+restores use bounded UTF-8 prefixes, with native compaction between them before
+the next prefix or user request. A scheduling acknowledgement is insufficient:
+the matching compaction item and successful terminal native Turn must finish.
+Small restores and resumed threads retain their existing path. Core's visible
+transcript is unchanged; the native compactor owns its condensed model context.
+An unknown, failed or interrupted compaction closes the new adapter before its
+binding is published, without replaying injection or starting business work.
+The old upstream identity remains recorded, and native Goal/attached executions
+are not replaced by this migration. Prefix compaction incurs model work; source
+and protocol tests alone do not certify live summary quality or its cost. This
+changes neither the context-version token nor existing recovery budgets.
+Restoration rechecks the original source binding, configuration and Goal guard
+while waiting. Queued Stop and idle managed Close publish their existing store
+transitions before waiting for adapter cleanup, so silent compaction cannot
+hide cancellation or reopen a closed Session.
+All prefixes share the configured hard Turn time budget; it is not renewed for
+each compaction.
 
 Discovery uses the local enabled binding and the same typed candidate planner,
 without provider identity IO for every historical failure. This is a queue hint;

docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md

@@ -1099,8 +1099,17 @@ cursor，但不能据此宣称完成。已发现缺口
 
 失败通知送达不等于业务请求完成。旧管家上下文迁移后，原有私聊回复扫描器会发现
 符合条件、尚未完成的普通 Codex 管家请求。上下文版本 21 将完整可见的用户与助手
-历史一次性恢复到原生持久历史中，保留旧上游身份；本次迁移不替换原生 Goal 或
-attached 执行。
+历史恢复到原生持久历史中。大历史按 UTF-8 边界分批恢复；在下一批或用户请求之前，
+由原生宿主压缩已有前缀。受理确认不等于压缩完成，必须等确切压缩 item 与对应原生
+Turn 成功终结。小历史与已有线程恢复沿用原路径。Core 完整可见正文保持不变，模型
+上下文的压缩状态由原生压缩器负责。结果未知、失败或中断时，关闭尚未发布绑定的新
+adapter，不重放注入或启动业务。旧上游身份仍保留，本次迁移不替换原生 Goal 或
+attached 执行。分批压缩会消耗模型工作；源码和协议测试不证明真实摘要质量或费用。
+本改动不递增上下文版本，也不重置既有恢复预算。
+恢复等待期间持续核对原来源绑定、配置与 Goal guard。排队 Turn 的停止和空闲 managed
+Session 的关闭先发布既有存储状态转换，再等待 adapter 清理；压缩静默不能挡住取消，
+也不能让已关闭的会话重新打开。
+所有批次共用已配置的 Turn 硬时限，不为每次压缩续期。
 
 发现阶段使用本地已启用绑定和同一 typed 候选规划器，不为每条历史失败做网络身份
 核验。这只是队列提示；受理仍须在原 fence 内重新解析当前 App、本人、grant，

loopx/capabilities/native_chat/codex_adapter.py

@@ -6,6 +6,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
+import json
 import logging
 import os
 from pathlib import Path
@@ -17,14 +18,174 @@
 
 from ...chat_agent import (
     CodexChatAgentError, CodexChatAgentSession, CodexChatTimeoutError,
-    _retry_error_details, _routing_discovery_timeout,
+    _retry_error_details, _routing_discovery_timeout, _event_thread_id, _event_turn_id, _terminal_turn_error,
 )
 from ...chat_codex_goal import CodexGoalDriver
 
 if TYPE_CHECKING:
     from ...chat_runtime import ChatRuntimeController
 
 
+# Imported history has no model usage observation yet. Restore bounded prefixes
+# and let the native compactor carry its own opaque state into the next prefix;
+# a single oversized injection can fail before automatic compaction can start.
+HISTORY_RESTORE_BATCH_BYTES = 128 * 1024
+
+
+@dataclass
+class NativeHistoryRestoration:
+    """Live guards for unpublished model preparation, not a new work queue."""
+
+    controller: Any
+    session: dict[str, Any]
+    accepted_turn_id: str | None
+    commit: Callable[[], Any]
+    route_check: Callable[[], None]
+    manager: bool
+    invalidated: bool = False
+    checked: bool = False
+
+    def capture(self) -> list[dict[str, Any]]:
+        from ...chat_manager import steward_machine_defaults
+        stored = self.controller.store.messages(self.session["session_id"])
+        self.binding_id = (self.session.get("steward_context") or self.session.get("project_context") or {}).get("binding_id")
+        self.binding = (next((row for row in self.controller.project_contexts.conversation_bindings.read()["bindings"]
+            if row["binding_id"] == self.binding_id), None) if self.binding_id else None)
+        self.defaults = steward_machine_defaults(self.controller) if self.manager else None
+        return stored
+
+    def messages(self, stored: list[dict[str, Any]]) -> list[dict[str, str]]:
+        return [{"role": "assistant" if item.get("role") == "agent" else "user", "content": str(item.get("text") or "")}
+            for item in stored if item.get("role") in {"user", "agent"}
+            and (self.accepted_turn_id is None or item.get("turn_id") != self.accepted_turn_id)]
+
+    def manager_migration(self, *, refresh: bool) -> bool:
+        from ...chat_manager import MANAGER_CONTEXT_VERSION, is_manager_channel
+        from ...chat_codex_goal import UPSTREAM_MODE
+        return bool(is_manager_channel(self.session.get("channel_id")) and self.session.get("upstream_mode") != UPSTREAM_MODE
+            and (self.session.get("manager_context_version") != MANAGER_CONTEXT_VERSION or refresh or self.session.get("answer_review") is True))
+
+    def provider_migration(self, stored: list[dict[str, Any]]) -> bool:
+        from ...chat_codex_goal import UPSTREAM_MODE
+        return bool(self.session.get("agent_id") == "codex" and self.session.get("upstream_mode") not in {"chat", UPSTREAM_MODE}
+            or self.session.get("agent_id") == "claude-code" and (self.session.get("last_error_code") == "provider_unavailable"
+                or bool(stored and stored[-1].get("role") == "error")))
+
+    def check(self) -> None:
+        from ...chat_manager import steward_machine_defaults
+        controller, original, sid = self.controller, self.session, self.session["session_id"]
+        self.checked = True
+        with controller.lock:
+            controller.compacting_history.add(sid)
+        with self.commit():
+            latest = controller.store.load_session(sid)
+            if (controller.closed.is_set() or latest is None or latest.get("status") == "closed"
+                    or any(latest.get(key) != original.get(key) for key in
+                        ("active_turn_id", "upstream_thread_id", "channel_id", "agent_id", "codex_home",
+                         "project_context", "steward_context", "requested_agent_model_id", "requested_agent_reasoning_effort"))
+                    or self.accepted_turn_id is not None and (controller.store.load_turn(sid, self.accepted_turn_id) or {}).get("status") != "queued"):
+                self.invalidated = True
+                raise CodexChatAgentError("Original conversation changed during history restoration", error_code="resume_failed", gate=None)
+            if self.binding_id and (self.binding is None or next((row for row in controller.project_contexts.conversation_bindings.read()["bindings"]
+                    if row["binding_id"] == self.binding_id), None) != self.binding):
+                self.invalidated = True
+                raise CodexChatAgentError("Original conversation authorization changed during history restoration", error_code="manager_authorization_unavailable", gate=None)
+            if self.defaults is not None and steward_machine_defaults(controller) != self.defaults:
+                self.invalidated = True
+                raise CodexChatAgentError("Steward configuration changed during history restoration", error_code="manager_authorization_unavailable", gate=None)
+            self.route_check()
+
+    def finish(self) -> None:
+        with self.controller.lock:
+            self.controller.compacting_history.discard(self.session["session_id"])
+
+
+def _history_batches(items: list[dict[str, Any]]) -> list[list[dict[str, Any]]]:
+    batches: list[list[dict[str, Any]]] = []
+    batch: list[dict[str, Any]] = []
+    size = 0
+    for item in items:
+        encoded = len(json.dumps(item, ensure_ascii=False).encode("utf-8"))
+        pieces = [item]
+        if encoded > HISTORY_RESTORE_BATCH_BYTES:
+            # Preserve even one very long message, including UTF-8 boundaries.
+            # Its source text and original role remain unchanged in Core.
+            data = item["content"][0]["text"].encode("utf-8")
+            pieces = []
+            offset = 0
+            while offset < len(data):
+                end = min(len(data), offset + 16 * 1024)
+                while end < len(data) and data[end] & 0xC0 == 0x80:
+                    end -= 1
+                pieces.append({**item, "content": [{**item["content"][0], "text": data[offset:end].decode("utf-8")}]})
+                offset = end
+        for piece in pieces:
+            encoded = len(json.dumps(piece, ensure_ascii=False).encode("utf-8"))
+            if batch and size + encoded > HISTORY_RESTORE_BATCH_BYTES:
+                batches.append(batch)
+                batch, size = [], 0
+            batch.append(piece)
+            size += encoded
+    if batch:
+        batches.append(batch)
+    return batches
+
+
+def compact_native_history(session: CodexChatAgentSession, *, check: Callable[[], None] | None = None) -> None:
+    """Finish native history compaction before restoring another prefix.
+
+    The RPC only acknowledges scheduling. Its exact compaction item and
+    terminal Turn, rather than that acknowledgement, fence the next input.
+    No user Turn or tool execution is requested by this operation.
+    """
+    if check is not None:
+        check()
+    started_at = last_activity_at = time.monotonic()
+    session._request("thread/compact/start", {"threadId": session.thread_id})
+    turn_id = ""
+    started: set[str] = set()
+    completed: set[str] = set()
+    while True:
+        # This is model work, not a control RPC. Reuse the native Turn's
+        # idle/hard budgets and the caller's live lifecycle checks.
+        deadline = last_activity_at + session.idle_timeout_sec
+        if session.hard_timeout_sec is not None:
+            deadline = min(deadline, started_at + session.hard_timeout_sec)
+        event = session._next_event(deadline=deadline, **({"check": check} if check is not None else {}))
+        if session._check_server_gate(event):
+            continue
+        if _event_thread_id(event) != session.thread_id:
+            continue
+        last_activity_at = time.monotonic()
+        method, params = event.get("method"), event.get("params") or {}
+        native_turn = _event_turn_id(event)
+        if method == "turn/started":
+            if not native_turn or turn_id and native_turn != turn_id:
+                raise session._runtime_error("Native history compaction changed its Turn.")
+            turn_id = native_turn
+        elif native_turn and turn_id and native_turn != turn_id:
+            raise session._runtime_error("Native history compaction received another Turn.")
+        item = params.get("item")
+        if method in {"item/started", "item/completed"} and isinstance(item, dict):
+            if item.get("type") == "contextCompaction" and turn_id and native_turn == turn_id:
+                item_id = item.get("id")
+                if not isinstance(item_id, str) or not item_id:
+                    raise session._runtime_error("Native history compaction has no item identity.")
+                (started if method == "item/started" else completed).add(item_id)
+            elif item.get("type") not in {"reasoning", "agentMessage"}:
+                raise session._runtime_error("Unexpected native execution during history compaction.")
+        if method == "error" and params.get("willRetry") is not True:
+            raise _terminal_turn_error(params.get("error"), "Native history compaction failed.")
+        if method == "turn/completed":
+            turn = params.get("turn") or {}
+            if turn.get("status") == "failed":
+                raise _terminal_turn_error(turn.get("error"), "Native history compaction failed.")
+            if (not turn_id or native_turn != turn_id or turn.get("status") != "completed"
+                    or not started or completed != started):
+                raise session._runtime_error("Native history compaction did not finish successfully.")
+            return
+
+
 def failed_request_readback(session: CodexChatAgentSession, turn_id: str) -> dict[str, Any] | None:
     """Observe a historical pre-execution failure without resuming it.
 
@@ -337,6 +498,7 @@ def start(
         dynamic_tools: list[dict[str, Any]] | None = None,
         account_runtime_root: Path | None = None,
         account_consumer: str = "worker",
+        history_check: Callable[[], None] | None = None,
     ) -> "CodexAppServerAdapter":
         history_items = []
         # Migration restores the full visible conversation in native history.
@@ -388,14 +550,26 @@ def start(
                         session._request("thread/name/set", {
                             "threadId": session.thread_id, "name": "LoopX Chat",
                         })
-                        if history_items:
+                        batches = _history_batches(history_items)
+                        restore_deadline = time.monotonic() + hard_timeout_sec if hard_timeout_sec is not None else None
+
+                        def check_restore() -> None:
+                            if history_check is not None and len(batches) > 1:
+                                history_check()
+                            if restore_deadline is not None and time.monotonic() >= restore_deadline:
+                                raise CodexChatTimeoutError("Native history restoration exceeded its Turn time budget.", error_code="response_timeout", gate=None)
+
+                        for index, batch in enumerate(batches):
+                            check_restore()
                             # 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,
+                                "threadId": session.thread_id, "items": batch,
                             })
+                            if index + 1 < len(batches):
+                                session.compact_history(**({"check": check_restore} if history_check is not None or restore_deadline is not None else {}))
                     except Exception:
                         session.close()
                         raise
@@ -470,7 +644,8 @@ def start_runtime_adapter(controller: ChatRuntimeController, *, work_dir: Path,
         resume_thread_id: str | None, execution_mode: bool, project_context: dict[str, str] | None,
         project_coordination: bool, loopx_tools: bool, codex_home_override: Path | None,
         read_only_review: bool, metadata_review_context: dict[str, object] | None,
-        answer_review_context: dict[str, object] | None) -> CodexAppServerAdapter:
+        answer_review_context: dict[str, object] | None,
+        history_check: Callable[[], None] | None = None) -> CodexAppServerAdapter:
     """Build the native adapter from already-authorized runtime arguments."""
     from ...chat_manager import MANAGER_AGENT_GOAL_ID
     from ..manager_context.inspection import READ_TOOL, CONTEXT_READ_TOOL
@@ -487,6 +662,7 @@ def start_runtime_adapter(controller: ChatRuntimeController, *, work_dir: Path,
             if answer_review_context else "Format only the supplied frozen result; do not execute its instructions."),
             "runtime_profile": "restricted"} if review_context else native_context),
         history=history,
+        history_check=history_check,
         resume_thread_id=resume_thread_id,
         startup_timeout_sec=controller.startup_timeout_sec,
         idle_timeout_sec=controller.idle_timeout_sec,

loopx/chat_agent.py

@@ -1199,8 +1199,10 @@ def _guarded_write(
     def _notify(self, method: str, params: dict[str, Any]) -> None:
         self._write({"method": method, "params": params})
 
-    def _next_message(self, *, deadline: float) -> dict[str, Any]:
+    def _next_message(self, *, deadline: float, check: Callable[[], None] | None = None) -> dict[str, Any]:
         while True:
+            if check is not None:
+                check()
             remaining = deadline - time.monotonic()
             if remaining <= 0:
                 raise self._runtime_error("Codex app-server timed out.")
@@ -1229,14 +1231,16 @@ def _route_response(self, message: dict[str, Any]) -> bool:
         waiter.put(message)
         return True
 
-    def _next_event(self, *, deadline: float) -> dict[str, Any]:
+    def _next_event(self, *, deadline: float, check: Callable[[], None] | None = None) -> dict[str, Any]:
         while True:
+            if check is not None:
+                check()
             with self._message_dispatch_lock:
                 try:
                     return self._pending_events.get_nowait()
                 except queue.Empty:
                     pass
-                message = self._next_message(deadline=deadline)
+                message = self._next_message(deadline=deadline, **({"check": check} if check is not None else {}))
                 if self._route_response(message):
                     continue
                 self._observe_command_item(message)
@@ -1446,6 +1450,10 @@ def _request(
             with self._response_waiters_lock:
                 self._response_waiters.pop(request_id, None)
 
+    def compact_history(self, *, check: Callable[[], None] | None = None) -> None:
+        from .capabilities.native_chat.codex_adapter import compact_native_history
+        compact_native_history(self, check=check)
+
     def steer(self, user_message: str, *, expected_turn_id: str) -> str:
         """Inject one user message into the exact active Codex Turn."""