From 6bb758632c7cf0de7910403fcbe04a1879075b60 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sat, 7 Nov 2026 01:17:22 +0800 Subject: [PATCH 1/2] fix(chat): scope answer review and recover unstarted queue claims --- .../app-conversation-and-async-inbox-v0.md | 9 +++ ...p-conversation-and-async-inbox-v0.zh-CN.md | 6 ++ loopx/chat_runtime.py | 53 +++++++++++------- .../test_chat_runtime_preparation_recovery.py | 56 +++++++++++++++++++ tests/test_steward_interrupted_answer.py | 15 +++++ 5 files changed, 120 insertions(+), 19 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 3006adefde..42fcf4466d 100644 --- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md +++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md @@ -1835,6 +1835,15 @@ adopts the reserved review instead of granting another attempt. Missing native evidence or capacity remains pending. The failure notice describes this qualified recovery instead of giving every restart the same blanket no-replay message. +This isolated answer review receives its complete original question and verified +visible results at dispatch. It does not import or serially compact the unrelated +conversation before it can answer. Core retains the full transcript, and ordinary +conversation restoration continues to preserve that history. +If restart happens after the queue claim but before native dispatch, the worker +adopts that same unstarted Turn through ordinary preparation and execution +fences. It does not wait forever for a queued Turn to become running by itself; +unfinished managed acceptance and already-started work retain their own owners. + ### Explicit unfinished work after a completed Turn A terminal model Turn is distinct from completed business work. A private 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 08ff985dfd..ec2f2a6c9d 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 @@ -1198,6 +1198,12 @@ provider 当前正文等于已知前文或目标正文。正文改变、效果 与失败通知不变；重启或入队回执不明时采用已预留复核，不产生新次数。缺少原生证据 或额度仍保留待办。失败提示说明这些恢复条件，不再对所有重启一律提示不会自动重放。 +这次独立答案复核在派发时收到完整原问题及已核验的可见结果，不在答复前导入、串行 +压缩无关的整段对话。Core 完整聊天记录保留；普通对话的历史恢复逻辑继续保留全部历史。 +若重启发生在取得队列 claim 后、原生派发前，worker 沿正常准备与执行屏障采用同一 +未启动 Turn，不空等 queued 自己变成 running；尚未发布的受管接收和已启动工作仍由 +各自原 owner 处理。 + ### Turn 完成后的显式未完工作 模型 Turn 终态与业务完成分开。私有项目 worker 核验已结束工作的实际结果后，可在 diff --git a/loopx/chat_runtime.py b/loopx/chat_runtime.py index f4571abfef..2c44e67c82 100644 --- a/loopx/chat_runtime.py +++ b/loopx/chat_runtime.py @@ -949,7 +949,11 @@ def check_route_configuration(): else str(session["upstream_thread_id"]) ), history=( - history + # This isolated review receives the complete original + # request and verified results at dispatch. Importing the + # whole conversation here needlessly compacts unrelated + # history before an unanswered question can be reviewed. + None if answer_fallback else history 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} @@ -1497,6 +1501,7 @@ def _drain_session_queue( or session.get("session_mode") == CHAT_SESSION_MODE_ATTACHED ): return + turn = None active_turn_id = str(session.get("active_turn_id") or "") if active_turn_id: try: @@ -1506,30 +1511,40 @@ def _drain_session_queue( # non-executing lifecycle repair, never the old input. self.closed.wait(0.25) continue - with self.lock: - attached = self.adapters.get(session_id) - if attached is None or not attached.healthcheck(): - try: - self._ensure_adapter( - session, - work_dir=work_dir, - objective=objective, - ) - except Exception as exc: - self._fail_queue_preparation(session_id, active_turn_id, exc) - else: - continue active = self.store.load_turn(session_id, active_turn_id) - if active and active.get("status") not in TERMINAL_TURN_STATES: - self.closed.wait(0.05) - continue - self.store.release_active_turn(session_id, active_turn_id) + if (active and active.get("status") == "queued" and "_acceptance" not in active + and active.get("started_at") is None and active.get("upstream_turn_id") is None + and active.get("delta_count") == 0): + # A crash can leave the queue claim before native + # dispatch. Keep that exact claim and pass through the + # normal preparation and execution fences below. + turn = active + else: + with self.lock: + attached = self.adapters.get(session_id) + if attached is None or not attached.healthcheck(): + try: + self._ensure_adapter( + session, + work_dir=work_dir, + objective=objective, + ) + except Exception as exc: + self._fail_queue_preparation(session_id, active_turn_id, exc) + else: + continue + active = self.store.load_turn(session_id, active_turn_id) + if active and active.get("status") not in TERMINAL_TURN_STATES: + self.closed.wait(0.05) + continue + self.store.release_active_turn(session_id, active_turn_id) refreshed = self.store.load_session(session_id) if refreshed is None: return with self.lock: self.session_queue_wakeups.discard(session_id) - turn = self.store.claim_next_queued_turn(session_id) + if turn is None: + turn = self.store.claim_next_queued_turn(session_id) if turn is None: with self.lock: if session_id in self.session_queue_wakeups: diff --git a/tests/test_chat_runtime_preparation_recovery.py b/tests/test_chat_runtime_preparation_recovery.py index 10f63c7792..055e28b36f 100644 --- a/tests/test_chat_runtime_preparation_recovery.py +++ b/tests/test_chat_runtime_preparation_recovery.py @@ -11,6 +11,62 @@ from loopx.chat_store import _read_json +@pytest.mark.parametrize("prior_dispatch", [None, "native-id", "started"]) +def test_restart_adopts_claimed_unstarted_queue_without_duplicate_dispatch(tmp_path, monkeypatch, prior_dispatch): + import queue + import threading + from types import SimpleNamespace + from loopx.chat_agent import CodexChatAgentSession + from loopx.chat_runtime import ChatRuntimeController, CodexAppServerAdapter + from loopx.chat_store import ChatSessionStore + + store, sid, original_runtime = session_runtime(tmp_path) + original_runtime.close() + original, _ = store.create_queued_turn(sid, client_turn_id="claimed-before-crash", + message="The same original input", origin="web") + assert store.claim_next_queued_turn(sid)["turn_id"] == original["turn_id"] + assert store.load_turn(sid, original["turn_id"])["started_at"] is None + if prior_dispatch: + store.update_turn(sid, original["turn_id"], **( + {"upstream_turn_id": "unknown-native-dispatch"} if prior_dispatch == "native-id" + else {"started_at": "2026-01-01T00:00:00Z"})) + reopened = ChatSessionStore(tmp_path) + runtime = ChatRuntimeController(store=reopened, codex_bin="unused-test-codex") + native = CodexChatAgentSession(process=SimpleNamespace(poll=lambda: None), + messages=queue.Queue(), thread_id="public-thread", work_dir=tmp_path) + adapter = CodexAppServerAdapter(native) + sent = [] + def write(packet): + assert packet["method"] == "turn/start" + sent.append(packet["params"]) + native.messages.put({"id": packet["id"], "result": {"turn": {"id": "original-dispatch"}}}) + native.messages.put({"method": "item/completed", "params": {"item": { + "type": "agentMessage", "phase": "final_answer", "text": "Recovered the original queue."}}}) + native.messages.put({"method": "turn/completed", "params": {"turn": {"status": "completed"}}}) + monkeypatch.setattr(native, "_write", write) + prepared = threading.Event() + def ensure(*_a, **_kw): + prepared.set() + return adapter + monkeypatch.setattr(runtime, "_ensure_adapter", ensure) + try: + runtime.resume_session_queue(session_id=sid, work_dir=tmp_path, objective="Original request") + if prior_dispatch: + assert prepared.wait(timeout=2) + current = reopened.load_turn(sid, original["turn_id"]) + assert current["status"] == "queued" and not sent + assert current.get("upstream_turn_id") == ("unknown-native-dispatch" if prior_dispatch == "native-id" else None) + return + result = runtime.wait_for_turn(session_id=sid, turn_id=original["turn_id"], timeout_sec=3) + assert result["status"] == "completed" + assert len(sent) == 1 and sent[0]["input"][0]["text"].endswith(original["message"]) + assert reopened.turn_for_client(sid, original["client_turn_id"])["turn_id"] == original["turn_id"] + assert len([e for e in reopened.events_after(sid, original["turn_id"], None) + if e["kind"] == "turn.started"]) == 1 + finally: + runtime.close() + + def test_account_preparation_keeps_its_error_before_native_dispatch(tmp_path, monkeypatch): store, sid, runtime = session_runtime(tmp_path) diff --git a/tests/test_steward_interrupted_answer.py b/tests/test_steward_interrupted_answer.py index 8e76c03249..bea53e4d68 100644 --- a/tests/test_steward_interrupted_answer.py +++ b/tests/test_steward_interrupted_answer.py @@ -58,6 +58,16 @@ def observed(kind, payload): transport.reconcile_request(path) failure_bytes = store._turn_path(original["session_id"], original["turn_id"]).read_bytes() notice = deepcopy(json.loads(path.read_text())["deliveries"]) + history_marker = "Unrelated retained conversation. " * 5000 + if fault is None: + store.append_message(original["session_id"], role="agent", text=history_marker, + turn_id="unrelated-historical-turn") + start_adapter = runtime._start_adapter + def start_scoped_review(**kwargs): + if kwargs.get("answer_review_context") is not None: + assert kwargs["history"] is None, "a restricted answer must not import the whole conversation" + return start_adapter(**kwargs) + monkeypatch.setattr(runtime, "_start_adapter", start_scoped_review) class Ahead(datetime): @classmethod def now(cls, tz=None): @@ -129,6 +139,11 @@ def crash(**kwargs): assert opened["config"]["permissions"][opened["permissions"]]["network"]["enabled"] is False assert [t["name"] for t in opened["dynamicTools"]] == ["loopx_manager_read"] assert not any(c["method"] == "thread/resume" for c in calls), "unknown old business must not be resumed" + assert not any(c["method"] in {"thread/inject_items", "thread/compact/start"} for c in calls) + if fault is None: + assert any(item["text"] == history_marker for item in store.messages(original["session_id"])) + review_inputs = [c["params"] for c in calls if c["method"] == "turn/start"] + assert original["message"] in json.dumps(review_inputs, ensure_ascii=False) writes = list(provider.writes) restarted.reconcile_request(path) assert provider.writes == writes From 19a193f15dc2e837861d19a5de9ee4f09b3701f4 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sat, 7 Nov 2026 01:20:32 +0800 Subject: [PATCH 2/2] fix(chat): retain queue ownership across preparation races --- loopx/canary/module_metric_baseline.json | 2 +- loopx/chat_runtime.py | 7 ++++++- 2 files changed, 7 insertions(+), 2 deletions(-) diff --git a/loopx/canary/module_metric_baseline.json b/loopx/canary/module_metric_baseline.json index b873129ae2..e952167c8b 100644 --- a/loopx/canary/module_metric_baseline.json +++ b/loopx/canary/module_metric_baseline.json @@ -39,7 +39,7 @@ "loopx/chat_runtime.py": { "any_count": 38, "dict_any_count": 0, - "lines": 2419 + "lines": 2439 }, "loopx/chat_server.py": { "any_count": 25, diff --git a/loopx/chat_runtime.py b/loopx/chat_runtime.py index 2c44e67c82..1ede8c359c 100644 --- a/loopx/chat_runtime.py +++ b/loopx/chat_runtime.py @@ -1562,11 +1562,16 @@ def _drain_session_queue( # An empty queue needs no provider preparation. Claim real # work first, keeping the existing enqueue/retirement fence # above and the Turn's execution/stop checks below. + current = self.store.load_session(session_id) + if current is None or current.get("active_turn_id") != turn_id: + return adapter = self._ensure_adapter( - self.store.load_session(session_id) or refreshed, + current, work_dir=work_dir, objective=objective, ) + if (self.store.load_session(session_id) or {}).get("active_turn_id") != turn_id: + return except Exception as exc: # The claimed request still owns a terminal outcome when # workspace preparation or adapter restoration fails.