From b4aa8493a13aa2cfc92d2ed77527b968edd54988 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Fri, 6 Nov 2026 12:37:53 +0800 Subject: [PATCH] fix(chat): compact bounded history before new native turns Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../app-conversation-and-async-inbox-v0.md | 21 +- ...p-conversation-and-async-inbox-v0.zh-CN.md | 13 +- .../capabilities/native_chat/codex_adapter.py | 184 ++++++++++++++- loopx/chat_agent.py | 14 +- loopx/chat_runtime.py | 102 ++++----- tests/test_chat_agent.py | 38 ++++ tests/test_chat_codex_history.py | 215 +++++++++++++++++- 7 files changed, 515 insertions(+), 72 deletions(-) diff --git a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md index 76d515a39a..75251ded78 100644 --- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md +++ b/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; diff --git a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md index 4ed9b5d4ed..49ac3a5edf 100644 --- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md +++ b/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， diff --git a/loopx/capabilities/native_chat/codex_adapter.py b/loopx/capabilities/native_chat/codex_adapter.py index 59d9d4e430..2b82ed21aa 100644 --- a/loopx/capabilities/native_chat/codex_adapter.py +++ b/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,7 +18,7 @@ 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 @@ -25,6 +26,166 @@ 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, diff --git a/loopx/chat_agent.py b/loopx/chat_agent.py index ee9ea8b096..f744fc5a8b 100644 --- a/loopx/chat_agent.py +++ b/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.""" diff --git a/loopx/chat_runtime.py b/loopx/chat_runtime.py index 7b68ea45b7..e858edf7ca 100644 --- a/loopx/chat_runtime.py +++ b/loopx/chat_runtime.py @@ -24,7 +24,7 @@ ) from .chat.coordination import PROJECT_COORDINATION_GUIDANCE, PROJECT_CONTEXT_VERSION, apply_context_handoff from .capabilities.native_chat import codex_context, project_context as project_context_policy -from .capabilities.native_chat.codex_adapter import CodexAppServerAdapter, start_attachment_turn +from .capabilities.native_chat.codex_adapter import CodexAppServerAdapter, NativeHistoryRestoration, start_attachment_turn from .capabilities.native_chat.machine_defaults import codex_profile_home, load_codex_chat_defaults from .capabilities.native_chat.steward_managed_mode import prepare_managed_session_route, require_managed_agent_session from .capabilities.native_chat.turn_release import TurnReleaseWakeup @@ -273,6 +273,7 @@ def __init__( self.hard_timeout_sec = hard_timeout_sec self.endpoint_registry = endpoint_registry or AgentEndpointRegistry(store.root) self.adapters: dict[str, ChatRuntimeAdapter] = {} + self.compacting_history: set[str] = set() self.cancelled_turns: set[tuple[str, str]] = set() self.turn_event_buffers: dict[tuple[str, str], _TurnEventBuffer] = {} self.turn_done_events: dict[tuple[str, str], threading.Event] = {} @@ -363,6 +364,7 @@ def _start_adapter( read_only_review: bool = False, metadata_review_context: dict[str, Any] | None = None, answer_review_context: dict[str, Any] | None = None, + history_check: Callable[[], None] | None = None, ) -> ChatRuntimeAdapter: if project_context is not None: project_context_policy.validate_project_native_executor(agent_id, project_context) @@ -419,7 +421,8 @@ def _start_adapter( resume_thread_id=resume_thread_id, execution_mode=execution_mode, project_context=project_context, project_coordination=project_coordination, loopx_tools=loopx_tools, codex_home_override=codex_home_override, read_only_review=read_only_review, - metadata_review_context=metadata_review_context, answer_review_context=answer_review_context) + metadata_review_context=metadata_review_context, answer_review_context=answer_review_context, + history_check=history_check) if agent_id == "claude-code": return ClaudeCodeAdapter.start( claude_bin=self.claude_bin, @@ -924,42 +927,13 @@ def check_route_configuration(): payload={"error_code": "server_restarted", "message": failed["error"]}, ) native_recovery = None + history_restore = NativeHistoryRestoration(self, session, accepted_turn_id, commit, check_route_configuration, manager_runtime is not None) try: - stored_messages = self.store.messages(session_id) - history = [ - { - "role": "assistant" if item.get("role") == "agent" else "user", - "content": str(item.get("text") or ""), - } - for item in stored_messages - if item.get("role") in {"user", "agent"} - and ( - accepted_turn_id is None - or item.get("turn_id") != accepted_turn_id - ) - ] - legacy_manager_context = ( - is_manager_channel(session.get("channel_id")) - and session.get("upstream_mode") != CODEX_GOAL_CHAT_MODE - and ( - session.get("manager_context_version") != MANAGER_CONTEXT_VERSION - or manager_profile_changed - or model_allocation is not None - or quota_fallback # Do not resume a prior tools-disabled formatter as business execution. - or answer_fallback or session.get("answer_review") is True - ) - ) - legacy_codex_goal_thread = ( - session.get("agent_id") == "codex" - and session.get("upstream_mode") not in {"chat", CODEX_GOAL_CHAT_MODE} - ) - retry_failed_claude_session = ( - session.get("agent_id") == "claude-code" - and ( - session.get("last_error_code") == "provider_unavailable" - or bool(stored_messages and stored_messages[-1].get("role") == "error") - ) - ) + stored_messages = history_restore.capture() + history = history_restore.messages(stored_messages) + legacy_manager_context = history_restore.manager_migration(refresh=(manager_profile_changed + or model_allocation is not None or quota_fallback or answer_fallback)) + legacy_provider_context = history_restore.provider_migration(stored_messages) adapter = self._start_adapter( agent_id=str(selected_agent), work_dir=work_dir, @@ -971,12 +945,12 @@ def check_route_configuration(): objective=objective, resume_thread_id=( None - if legacy_codex_goal_thread or retry_failed_claude_session or legacy_manager_context or legacy_project_context + if legacy_provider_context or legacy_manager_context or legacy_project_context else str(session["upstream_thread_id"]) ), history=( history - if legacy_codex_goal_thread or retry_failed_claude_session or legacy_manager_context or legacy_project_context + if legacy_provider_context or legacy_manager_context or legacy_project_context or session.get("agent_id") in {"anthropic-api", "openai-api", MANAGED_TURN_HOST} else None @@ -997,6 +971,7 @@ def check_route_configuration(): read_only_review=bool(recovery.get("read_only_review")), metadata_review_context=session.get("steward_context") if metadata_fallback else None, answer_review_context=session.get("steward_context") if answer_fallback else None, + history_check=history_restore.check, ) try: # Recovery reads back the provider-restored configuration. @@ -1011,14 +986,16 @@ def check_route_configuration(): adapter.close_session() raise except Exception as exc: + history_restore.finish() configuration_error = isinstance(exc, CodexChatAgentError) and exc.error_code == "unsupported_configuration" - with commit(): - self.store.update_session( - session_id, - status="busy" if accepted_turn_id is not None else "ready" if settings_changed else "resume_failed", - active_turn_id=accepted_turn_id, - last_error_code="unsupported_configuration" if settings_changed and configuration_error else "resume_failed", - ) + if not history_restore.invalidated: + with commit(): + self.store.update_session( + session_id, + status="busy" if accepted_turn_id is not None else "ready" if settings_changed else "resume_failed", + active_turn_id=accepted_turn_id, + last_error_code="unsupported_configuration" if settings_changed and configuration_error else "resume_failed", + ) if settings_changed and configuration_error: raise gate = exc.gate if isinstance(exc, CodexChatAgentError) else None @@ -1028,6 +1005,8 @@ def check_route_configuration(): gate=gate, ) from exc try: + if history_restore.checked: + history_restore.check() with commit(): if quota_route: check_route_configuration() @@ -1082,7 +1061,9 @@ def check_route_configuration(): changes["codex_home"] = str(selected_codex_home) if session["channel_id"] == "manager": changes["goal_id"] = MANAGER_AGENT_GOAL_ID - self.store.update_session(session_id, **changes) + # An idle Close may have won while native preparation was + # finishing. It must not acquire the replacement identity. + self.store.update_session(session_id, expected_statuses={session["status"], "resuming"}, **changes) elif (conversation_scope(session)["kind"] == "owner_goal" and session.get("upstream_mode") != CODEX_GOAL_CHAT_MODE): self.store.update_session(session_id, coordination_context_version=PROJECT_CONTEXT_VERSION) @@ -1097,8 +1078,10 @@ def check_route_configuration(): owns_adapter = self.adapters.get(session_id) is adapter if owns_adapter: self.adapters.pop(session_id, None) + history_restore.finish() adapter.close_session() raise + history_restore.finish() return adapter def submit_turn( @@ -2157,6 +2140,16 @@ def _fail_turn( ) def interrupt_turn(self, *, session_id: str, turn_id: str) -> dict[str, Any]: + # A new adapter can compact history while holding the adapter lock. + # Publish an atomic stop for a still-queued managed Turn first, so its + # live restoration guard can abort before the user's work is started. + # Running/attached Turns retain the existing native stop protocol. + with self.lock: + compacting = session_id in self.compacting_history + session = self.store.load_session(session_id) if compacting else None + queued_interrupt = (self.store.begin_turn_interrupt(session_id, turn_id, expected_statuses={"queued"}) + if session and session.get("session_mode") != CHAT_SESSION_MODE_ATTACHED + and self.store.prepared_managed_turn_request(session_id, turn_id=turn_id) is None else None) with self._session_adapter_lock(session_id): turn = self.store.load_turn(session_id, turn_id) if turn is None: @@ -2200,7 +2193,7 @@ def interrupt_turn(self, *, session_id: str, turn_id: str) -> dict[str, Any]: # checks the status and writes the intent under the Turn lock, so a # Turn that completed while this call was reading it keeps its # terminal event history and is not relabelled as interrupted. - interrupting = self.store.begin_turn_interrupt( + interrupting = turn if queued_interrupt is not None and turn.get("status") == "interrupting" else self.store.begin_turn_interrupt( session_id, turn_id, expected_statuses={"queued", "starting", "running", "interrupting"}, @@ -2239,7 +2232,7 @@ def interrupt_turn(self, *, session_id: str, turn_id: str) -> dict[str, Any]: adapter.interrupt_turn(upstream_turn_id) elif adapter is not None: adapter.interrupt_turn(upstream_turn_id or None) - elif upstream_turn_id or turn.get("status") in {"running", "interrupting"}: + elif upstream_turn_id or queued_interrupt is None and turn.get("status") in {"running", "interrupting"}: raise RuntimeError("active Turn transport is unavailable") if done_event is not None and not done_event.wait(timeout=5.0): raise TimeoutError("Turn worker has not acknowledged stop") @@ -2305,14 +2298,21 @@ def wait_for_turn(self, *, session_id: str, turn_id: str, timeout_sec: float = 9 (done_event.wait if done_event else time.sleep)(remaining if done_event else min(0.02, remaining)) def close_session(self, session_id: str) -> bool: + with self.lock: + compacting = session_id in self.compacting_history + session = self.store.load_session(session_id) if compacting else None + # The store atomically rejects any live/queued Turn. For an idle + # managed Session, make Close visible to in-flight native restoration + # before waiting for its adapter lock and transport cleanup. + preclosed = bool(session and session.get("session_mode") != CHAT_SESSION_MODE_ATTACHED + and self.store.close_managed_session(session_id)) with self._session_adapter_lock(session_id): session = self.store.load_session(session_id) if session is None: return False if session.get("session_mode") == CHAT_SESSION_MODE_ATTACHED: return self.store.close_attached_session(session_id) - closed = self.store.close_managed_session(session_id) - if not closed: + if not preclosed and not self.store.close_managed_session(session_id): return False with self.lock: adapter = self.adapters.pop(session_id, None) diff --git a/tests/test_chat_agent.py b/tests/test_chat_agent.py index c387e8932e..d887a9afdb 100644 --- a/tests/test_chat_agent.py +++ b/tests/test_chat_agent.py @@ -771,6 +771,44 @@ def write(packet): assert proof is None +@pytest.mark.parametrize("fault", [None, "missing-item", "interrupted", "failed", "foreign-turn", "unknown-item", "tool", "ack-only"]) +def test_history_compaction_requires_its_native_item_and_terminal_turn(tmp_path, monkeypatch, fault): + session = chat_agent.CodexChatAgentSession(process=SimpleNamespace(poll=lambda: None), + messages=queue.Queue(), thread_id="history-thread", work_dir=tmp_path, response_timeout_sec=.02, idle_timeout_sec=.02) + calls = [] + def write(packet): + calls.append(packet["method"]) + assert packet["method"] == "thread/compact/start" and packet["params"] == {"threadId": "history-thread"} + session.messages.put({"id": packet["id"], "result": {}}) + if fault == "ack-only": + return + rows = [ + {"method": "turn/started", "params": {"threadId": "history-thread", "turn": {"id": "compact-turn"}}}, + {"method": "item/started", "params": {"threadId": "history-thread", "turnId": "compact-turn", + "item": {"id": "compact-item", "type": "contextCompaction"}}}, + {"method": "item/completed", "params": {"threadId": "history-thread", "turnId": "compact-turn", + "item": {"id": "compact-item", "type": "contextCompaction"}}}, + {"method": "turn/completed", "params": {"threadId": "history-thread", "turn": { + "id": "compact-turn", "status": "interrupted" if fault == "interrupted" else "failed" if fault == "failed" else "completed", + "error": {"codexErrorInfo": "contextWindowExceeded"} if fault == "failed" else None}}}, + ] + if fault == "missing-item": + rows.pop(2) + if fault == "foreign-turn": + rows[2]["params"]["turnId"] = "another-turn" + if fault in {"unknown-item", "tool"}: + rows[1]["params"]["item"]["type"] = "futureEffect" if fault == "unknown-item" else "commandExecution" + for event in rows: + session.messages.put(event) + monkeypatch.setattr(session, "_write", write) + if fault is None: + session.compact_history() + else: + with pytest.raises(chat_agent.CodexChatAgentError): + session.compact_history() + assert calls == ["thread/compact/start"] + + @pytest.mark.parametrize("fault", [None, "older", "host-other", "active", "tool", "answer", "correction", "foreign", "unknown", "changed-error", "policy", "usage", "unauthorized", "http401"]) def test_historical_context_recovery_reads_only_proven_pre_execution_failure(tmp_path, monkeypatch, fault): diff --git a/tests/test_chat_codex_history.py b/tests/test_chat_codex_history.py index 379d2ccfed..263c4cdab8 100644 --- a/tests/test_chat_codex_history.py +++ b/tests/test_chat_codex_history.py @@ -7,6 +7,127 @@ from loopx import chat_agent, chat_runtime from loopx.capabilities.native_chat.codex_adapter import CodexAppServerAdapter +from test_chat_manager_model_adoption import conversation, configure # noqa: F401 +from test_native_steward_private import steward # noqa: F401 + + +def test_source_revocation_during_history_compaction_preserves_original_audience(steward, monkeypatch): + import threading + from concurrent.futures import ThreadPoolExecutor + + store, runtime, provider, transport, binding, capture, workspace = steward + transport.admit("steward-app", provider.event("steward-app", "original-question", "A first question")) + row = transport.core.pending()[0] + sid = row["session_id"] + assert runtime.wait_for_turn(session_id=sid, turn_id=row["turn_id"], timeout_sec=10)["status"] == "completed" + original = store.update_session(sid, manager_context_version=20) + runtime.adapters.pop(sid).close_session() + native_calls = capture.read_bytes() + entered = threading.Event() + + def native_start(**kwargs): + kwargs["history_check"]() + entered.set() + for _ in range(200): + kwargs["history_check"]() + threading.Event().wait(.01) + pytest.fail("restoration ignored the revoked source") + + monkeypatch.setattr(runtime, "_start_adapter", native_start) + with ThreadPoolExecutor(max_workers=2) as pool: + preparing = pool.submit(runtime._ensure_adapter, original, work_dir=workspace, objective="manager") + assert entered.wait(2) + revoking = pool.submit(transport.bindings.disconnect, binding["binding_id"], + expected_revision=transport.bindings.read()["revision"]) + with pytest.raises(chat_agent.CodexChatAgentError): + preparing.result(timeout=3) + revoking.result(timeout=3) + current = store.load_session(sid) + assert current["channel_id"] == original["channel_id"] + assert current["upstream_thread_id"] == original["upstream_thread_id"] + assert current["steward_context"] == original["steward_context"] + assert sid not in runtime.adapters and capture.read_bytes() == native_calls + + +@pytest.mark.parametrize("action", ["stop", "close"]) +def test_native_history_preparation_observes_user_stop_before_its_adapter_lock(conversation, monkeypatch, action): + import threading + from concurrent.futures import ThreadPoolExecutor + + root, store, runtime, session, _ = conversation + sid = session["session_id"] + before = store.messages(sid) + configure(root, model="gpt-6.1-sol", effort="high") + entered = threading.Event() + turn = None + if action == "stop": + turn, _ = store.create_turn(sid, client_turn_id="original", message="Original question") + session = store.update_session(sid, active_turn_id=turn["turn_id"], status="busy") + + def native_start(**kwargs): + entered.set() + # The native model may be silent for minutes. Its guard is polled even + # without notifications; Stop/Close must not wait for model completion. + for _ in range(200): + kwargs["history_check"]() + threading.Event().wait(.01) + pytest.fail("user action waited behind native history preparation") + + monkeypatch.setattr(runtime, "_start_adapter", native_start) + + def restore(): + with runtime._session_adapter_lock(sid): + return runtime._ensure_adapter_locked(session, work_dir=root, objective="manager", + **({"accepted_turn_id": turn["turn_id"]} if turn else {})) + + with ThreadPoolExecutor(max_workers=2) as pool: + preparing = pool.submit(restore) + assert entered.wait(1) + action_result = pool.submit(runtime.interrupt_turn, session_id=sid, turn_id=turn["turn_id"]) if turn else pool.submit(runtime.close_session, sid) + with pytest.raises(chat_agent.CodexChatAgentError, match="could not be restored"): + preparing.result(timeout=3) + result = action_result.result(timeout=3) + current = store.load_session(sid) + assert current["upstream_thread_id"] == session["upstream_thread_id"] + assert sid not in runtime.adapters + if turn: + assert result["status"] == "interrupted" and current["active_turn_id"] is None + assert not store.load_turn(sid, turn["turn_id"]).get("upstream_turn_id") + events = store.events_after(sid, turn["turn_id"], None) + assert sum(event["kind"] == "turn.interrupt_requested" for event in events) == 1 + else: + assert result is True and current["status"] == "closed" + assert store.messages(sid) == before + runtime.close() + + +@pytest.mark.parametrize("change", ["close", "settings", "requested-model"]) +def test_history_guard_rechecks_after_native_start_before_publishing(conversation, monkeypatch, change): + from test_chat_manager_model_adoption import Adapter + root, store, runtime, session, _ = conversation + sid = session["session_id"] + configure(root, model="gpt-6.1-sol", effort="high") + replacement = Adapter(99) + + def native_start(**kwargs): + kwargs["history_check"]() + if change == "close": + store.close_managed_session(sid) + elif change == "settings": + configure(root, model="gpt-6-luna", effort="high") + else: + store.update_session(sid, requested_agent_model_id="gpt-6-luna") + return replacement + + monkeypatch.setattr(runtime, "_start_adapter", native_start) + with pytest.raises((ValueError, chat_agent.CodexChatAgentError)): + runtime._ensure_adapter(session, work_dir=root, objective="manager") + current = store.load_session(sid) + assert current["upstream_thread_id"] == session["upstream_thread_id"] + assert replacement.closed and sid not in runtime.adapters + if change == "close": + assert current["status"] == "closed" + runtime.close() @pytest.mark.parametrize("fault", [None, "image-view", "later-user", "live", "failed", "background", "unknown", "live-tool", @@ -62,37 +183,113 @@ def items(method, params, deadline): @pytest.mark.parametrize("resume", [False, True]) -def test_history_is_persisted_once_before_publishing_a_new_adapter(monkeypatch, tmp_path, resume): +@pytest.mark.parametrize("count", [2, 400]) +def test_history_is_persisted_once_before_publishing_a_new_adapter(monkeypatch, tmp_path, resume, count): 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": f"Evidence {i}: " + "x" * 1000} for i in range(count)], {"role": "user", "content": ""}, ] calls = [] native = SimpleNamespace(thread_id="native-thread") - native._request = lambda method, params: calls.append((method, params)) or {} + active_bytes = 0 + + def request(method, params): + nonlocal active_bytes + if method == "thread/inject_items": + active_bytes += sum(len(item["content"][0]["text"].encode()) for item in params["items"]) + # Model the finite next-request window independently of the batching + # constant. An unbounded injection fails before a user Turn can run. + assert active_bytes < 256 * 1024 + calls.append((method, params)) + return {} + + def compact(): + nonlocal active_bytes + calls.append(("thread/compact/start", {})) + active_bytes = 128 # The native compactor's opaque retained state. + + native._request, native.compact_history = request, compact 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, + history_check=(lambda: pytest.fail("small/resumed history must keep its existing path")) if resume or count == 2 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": [{ + injected = [params for method, params in calls if method == "thread/inject_items"] + assert all(params["threadId"] == "native-thread" for params in injected) + assert [item for params in injected for item in params["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"]], - } + } for item in history if item["content"]] + methods = [method for method, _ in calls] + assert methods == ["thread/name/set", "thread/inject_items", *[ + method for _ in range(len(injected) - 1) for method in ("thread/compact/start", "thread/inject_items")]] + assert len(injected) == 1 if count == 2 else len(injected) > 1 assert all(method != "turn/start" for method, _ in calls) +def test_one_large_utf8_message_survives_bounded_native_restore(monkeypatch, tmp_path): + text = ('原始约束😀\\\"\n' * 40000) + "Keep the final constraint." + injected, calls = [], [] + native = SimpleNamespace(thread_id="native-thread", compact_history=lambda: calls.append("compact")) + def request(method, params): + if method == "thread/inject_items": + import json + assert len(json.dumps(params["items"], ensure_ascii=False).encode()) < 256 * 1024 + injected.extend(params["items"]) + calls.append(method) + native._request = request + monkeypatch.setattr(chat_agent.CodexChatAgentSession, "start", lambda **kwargs: native) + CodexAppServerAdapter.start(codex_bin="fixture", work_dir=tmp_path, goal_id="fixture", + objective="Stable objective", history=[{"role": "user", "content": text}]) + assert "".join(item["content"][0]["text"] for item in injected) == text + assert all(item["role"] == "user" for item in injected) + assert "compact" in calls and "turn/start" not in calls + + +def test_unfinished_history_compaction_never_injects_the_next_prefix(monkeypatch, tmp_path): + calls = [] + native = SimpleNamespace(thread_id="native-thread", close=lambda: calls.append("closed")) + native._request = lambda method, params: calls.append(method) + def compact(): + calls.append("compact") + raise chat_agent.CodexChatTimeoutError("unknown compaction outcome", error_code="response_timeout", gate=None) + native.compact_history = compact + monkeypatch.setattr(chat_agent.CodexChatAgentSession, "start", lambda **kwargs: native) + with pytest.raises(chat_agent.CodexChatTimeoutError, match="unknown compaction outcome"): + CodexAppServerAdapter.start(codex_bin="fixture", work_dir=tmp_path, goal_id="fixture", + objective="Stable objective", history=[{"role": "user", "content": "x" * 500000}]) + assert calls == ["thread/name/set", "thread/inject_items", "compact", "closed"] + + +def test_history_prefixes_share_the_original_hard_time_budget(monkeypatch, tmp_path): + import loopx.capabilities.native_chat.codex_adapter as provider + clock, calls = [0.0], [] + native = SimpleNamespace(thread_id="native-thread", close=lambda: calls.append("closed")) + native._request = lambda method, params: calls.append(method) + + def compact(*, check): + clock[0] += 4 + check() + calls.append("compact") + + native.compact_history = compact + monkeypatch.setattr(provider.time, "monotonic", lambda: clock[0]) + monkeypatch.setattr(chat_agent.CodexChatAgentSession, "start", lambda **kwargs: native) + with pytest.raises(chat_agent.CodexChatTimeoutError, match="time budget"): + CodexAppServerAdapter.start(codex_bin="fixture", work_dir=tmp_path, goal_id="fixture", + objective="Stable objective", hard_timeout_sec=6, + history=[{"role": "user", "content": "x" * 500000}]) + assert calls == ["thread/name/set", "thread/inject_items", "compact", "thread/inject_items", "closed"] + + @pytest.mark.parametrize("error", [ RuntimeError("unknown injection outcome"), chat_agent.CodexChatTimeoutError("unknown injection outcome", error_code="response_timeout", gate=None),