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 7b532cf72d..d0afb9a319 100644 --- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md +++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md @@ -170,6 +170,11 @@ later answer. Core freezes the original Session and canonical candidate Turn identities in the existing request journal **before** native inventory or transport resume can fail. Native observation may only narrow that frozen set; a retry cannot add later completed/active Turns or switch to a newer Session. +Completed-command discovery reads the exact native thread's complete paginated +item inventory once, retaining only command ownership for the frozen candidates. +This avoids a separate inventory round trip for each historical Turn. Missing or +conflicting item ownership and incomplete pagination remain unconfirmed; the +existing exact-Turn cancellation and independent terminal readback still run. The existing Lark listener returns after this durable freeze; its existing reply workers recover the prepared request to observe and cancel native background commands. Prepared recovery uses admission's fresh authority check inside both 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 de5fdd0160..f00d66bab0 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 @@ -119,7 +119,10 @@ App 身份时，并行读取两条原 provider 消息。两项只读观测全部 普通 Codex 话题空闲时，`/stop` 也覆盖同一原生 Session 中已完成 Turn 遗留的后台命令。 Core 先在原请求日志固定 Session 与候选 Turn，原 listener 随即继续受理消息；现有 回复工作池再恢复这条 prepared 请求，读取原生命令清单并取消原目标。恢复只能缩小 -固定候选，不能追停后来的 Turn 或新 Session。prepared 恢复复用受理路径在原请求与 +固定候选，不能追停后来的 Turn 或新 Session。 +发现已完成命令时，一次分页读完确切原生线程的 item 清单，仅保留冻结候选的命令 +归属，避免按历史 Turn 逐个往返。归属缺失或冲突、分页不完整仍不能确认；之后仍须 +执行原有的确切 Turn 取消与独立终端清单读回。prepared 恢复复用受理路径在原请求与 来源路由锁内的新鲜授权核验，不在等待锁前重复探测 provider；等待期间撤权即阻止 控制和投递。慢速或失败的原生控制不阻塞入站， 也不提前回复“已停止”；没有固定目标的旧请求保持未知，活跃 Turn 仍沿原确切中断路径。 diff --git a/loopx/chat_agent.py b/loopx/chat_agent.py index 63a3c36df8..14c73b9ab6 100644 --- a/loopx/chat_agent.py +++ b/loopx/chat_agent.py @@ -1364,12 +1364,41 @@ def owned_terminals() -> list[dict[str, Any]]: def background_turn_ids(self, turn_ids: list[str]) -> list[str]: """Observe live terminals owned by the supplied canonical native Turns.""" + candidates = list(dict.fromkeys(turn_ids)) + if not candidates: + return [] deadline = time.monotonic() + self.response_timeout_sec live_items = {terminal["itemId"] for terminal in self._background_terminals(deadline)} if not live_items: return [] - return [turn_id for turn_id in dict.fromkeys(turn_ids) - if live_items & self._command_item_ids(turn_id, deadline)] + # The native endpoint can return all items in this exact thread. Scan + # its complete pagination once instead of one round trip per historical + # Turn; newer/foreign Turns never enter the frozen candidate set. + owners: dict[str, str] = {} + commands: dict[str, set[str]] = {turn_id: set() for turn_id in candidates} + for entry in self._control_items("thread/items/list", {}, deadline): + turn_id, item = entry.get("turnId"), entry.get("item") + if not isinstance(turn_id, str) or not turn_id or not isinstance(item, dict): + raise self._runtime_error("Codex returned an unbound thread item.") + if item.get("type") == "commandExecution": + item_id = item.get("id") + if not isinstance(item_id, str) or not item_id: + raise self._runtime_error("Codex returned an unbound command item.") + if item_id in owners and owners[item_id] != turn_id: + raise self._runtime_error("Codex returned conflicting command ownership.") + owners[item_id] = turn_id + if turn_id in commands: + commands[turn_id].add(item_id) + # RPC dispatch can observe command events omitted by native history. + # Keep their existing exact thread/Turn proof, within this observation. + with self._command_items_lock: + for turn_id, items in commands.items(): + for item_id in self._command_items.get(turn_id, set()): + if item_id in owners and owners[item_id] != turn_id: + raise self._runtime_error("Codex returned conflicting command ownership.") + owners[item_id] = turn_id + items.add(item_id) + return [turn_id for turn_id in candidates if live_items & commands[turn_id]] def _command_item_ids(self, turn_id: str, deadline: float) -> set[str]: with self._command_items_lock: diff --git a/tests/test_chat_background_inventory.py b/tests/test_chat_background_inventory.py new file mode 100644 index 0000000000..f8914b0856 --- /dev/null +++ b/tests/test_chat_background_inventory.py @@ -0,0 +1,101 @@ +"""Completed-command discovery uses fresh native ownership in one thread scan.""" +import queue +import time +from types import SimpleNamespace + +import pytest + +from loopx.chat_agent import CodexChatAgentError, CodexChatAgentSession + + +def inventory(tmp_path, monkeypatch, *, fault=None): + session = CodexChatAgentSession(process=SimpleNamespace(poll=lambda: None), + messages=queue.Queue(), thread_id="original-thread", work_dir=tmp_path, + response_timeout_sec=.5) + turns = [f"completed-{i}" for i in range(23)] + calls = [] + entries = [{"turnId": turn, "item": {"type": "agentMessage", "id": f"answer-{i}"}} + for i, turn in enumerate(turns)] + entries += [ + {"turnId": turns[-1], "item": {"type": "commandExecution", "id": "owned-command"}}, + {"turnId": "newer-turn", "item": {"type": "commandExecution", "id": "newer-command"}}, + ] + + def write(packet): + method, params = packet["method"], packet["params"] + calls.append((method, params)) + assert params["threadId"] == "original-thread" + if fault == "rpc" and method == "thread/items/list": + session.messages.put({"id": packet["id"], "error": {"message": "unavailable"}}) + return + if method == "thread/backgroundTerminals/list": + result = {"data": [] if fault == "idle" else [ + {"itemId": "owned-command", "processId": "owned-pty"}, + {"itemId": "newer-command", "processId": "newer-pty"}, + {"itemId": "wire-command", "processId": "wire-pty"}, + ]} + else: + assert method == "thread/items/list" + # Preserve the former per-Turn protocol for the counterfactual. + if params.get("turnId"): + result = {"data": [entry for entry in entries if entry["turnId"] == params["turnId"]]} + else: + page = 1 if params.get("cursor") else 0 + result = {"data": entries[20:] if page else entries[:20], + "nextCursor": "second" if not page or fault == "cursor-cycle" else None} + if page and fault == "missing-data": + result.pop("data") + elif page and fault == "missing-turn": + result["data"] = [{"item": {"type": "commandExecution", "id": "owned-command"}}] + elif page and fault == "missing-item": + result["data"] = [{"turnId": turns[-1]}] + elif page and fault == "unbound-command": + result["data"] = [{"turnId": turns[-1], "item": {"type": "commandExecution", "id": ""}}] + elif page and fault == "ambiguous-item": + result["data"].append({"turnId": "newer-turn", "item": { + "type": "commandExecution", "id": "owned-command"}}) + session.messages.put({"id": packet["id"], "result": result}) + + monkeypatch.setattr(session, "_write", write) + # Native command events can outlive their history projection, but only + # this exact thread/Turn association may supplement the fresh inventory. + session.messages.put({"method": "item/completed", "params": { + "threadId": "original-thread", "turnId": turns[0], + "item": {"type": "commandExecution", "id": "wire-command"}}}) + if fault == "wire-conflict": + session.messages.put({"method": "item/completed", "params": { + "threadId": "original-thread", "turnId": turns[1], + "item": {"type": "commandExecution", "id": "wire-command"}}}) + return session, turns, calls + + +def test_completed_history_discovery_scans_all_pages_once_and_keeps_frozen_order(tmp_path, monkeypatch): + session, turns, calls = inventory(tmp_path, monkeypatch) + assert session.background_turn_ids(turns + [turns[0]]) == [turns[0], turns[-1]] + item_reads = [params for method, params in calls if method == "thread/items/list"] + assert len(item_reads) == 2 + assert all("turnId" not in params for params in item_reads) + assert item_reads[-1]["cursor"] == "second" + assert all(method in {"thread/items/list", "thread/backgroundTerminals/list"} for method, _ in calls) + + +@pytest.mark.parametrize("fault", ["rpc", "missing-data", "missing-turn", "missing-item", + "unbound-command", "cursor-cycle", "ambiguous-item", "wire-conflict"]) +def test_incomplete_or_ambiguous_thread_inventory_never_supplies_stop_targets(tmp_path, monkeypatch, fault): + session, turns, calls = inventory(tmp_path, monkeypatch, fault=fault) + with pytest.raises(CodexChatAgentError): + session.background_turn_ids(turns) + assert all(not method.endswith("/terminate") for method, _ in calls) + + +def test_idle_background_inventory_never_reads_old_command_history(tmp_path, monkeypatch): + session, turns, calls = inventory(tmp_path, monkeypatch, fault="idle") + assert session.background_turn_ids(turns) == [] + assert [method for method, _ in calls] == ["thread/backgroundTerminals/list"] + + +def test_empty_frozen_candidate_set_does_not_query_native_thread(tmp_path, monkeypatch): + session, _, calls = inventory(tmp_path, monkeypatch) + started = time.monotonic() + assert session.background_turn_ids([]) == [] + assert calls == [] and time.monotonic() - started < .5 diff --git a/tests/test_chat_completed_background_stop.py b/tests/test_chat_completed_background_stop.py index 90d0fe9aba..40498c05c7 100644 --- a/tests/test_chat_completed_background_stop.py +++ b/tests/test_chat_completed_background_stop.py @@ -32,7 +32,7 @@ def write(packet): elif method == "thread/items/list": result = {"data": [{"turnId": turn_id, "item": { "type": "commandExecution", "id": "owned-command", "status": "completed"}}] - if params["turnId"] == turn_id else []} + if params.get("turnId") in (None, turn_id) else []} elif method == "thread/turns/list": result = {"data": [{"id": turn_id, "status": "completed"}]} elif method == "thread/backgroundTerminals/terminate": @@ -54,7 +54,7 @@ def close(self): child.wait(timeout=5) -def _completed_with_later_answer(fixture, monkeypatch): +def _completed_with_later_answer(fixture, monkeypatch, *, history_size=2): store, runtime, provider, transport = connect_group(fixture) root = provider.topic("background-root", "Start bounded background work") assert transport.admit("notes-app", root)["status"] == "durably_accepted" @@ -62,13 +62,51 @@ def _completed_with_later_answer(fixture, monkeypatch): sid, tid = first["session_id"], first["turn_id"] native_id = store.load_turn(sid, tid)["upstream_turn_id"] wire = BackgroundWire(runtime.adapters[sid].session, native_id, monkeypatch) - later = provider.topic("later-answer", "Keep the corrected filename", root=root["message_id"]) - assert transport.admit("notes-app", later)["status"] == "durably_accepted" - finish_group_turn(runtime, transport, message=later["content"]) + for index in range(history_size - 1): + later = provider.topic(f"later-answer-{index}", f"Keep the corrected filename {index}", root=root["message_id"]) + assert transport.admit("notes-app", later)["status"] == "durably_accepted" + finish_group_turn(runtime, transport, message=later["content"]) assert store.load_session(sid)["active_turn_id"] is None return store, runtime, provider, transport, root, first, wire +def test_long_completed_history_stop_reaches_original_provider_with_one_thread_scan(ordinary, monkeypatch): # noqa: F811 + store, runtime, provider, transport, root, first, wire = _completed_with_later_answer( + ordinary, monkeypatch, history_size=23) + from loopx.capabilities.native_chat.codex_background_control import completed_turn_ids + + sid = first["session_id"] + frozen = completed_turn_ids(runtime, sid) + before = {turn: store.load_turn(sid, turn) for turn in frozen} + assert len(frozen) == 23 + wire.calls.clear() + stop = provider.topic("long-history-stop", "/stop", root=root["message_id"]) + try: + # Exercise the listener's durable admission and existing reply worker + # boundary, not only the inventory helper or a replacement controller. + assert transport.admit("notes-app", stop, defer_background_stop=True)["status"] == "command_recorded" + row = next(r for r in transport.core.pending() if r["command"] == "stop") + assert row["status"] == "prepared" and set(row["background_turn_ids"]) == set(frozen) + assert not wire.calls and wire.children["owned"].poll() is None + transport.reconcile() + delivered = transport.core.read_request(row["request_ref"]) + assert delivered["status"] == "command_completed" and delivered["delivery_verified"] + assert delivered["background_turn_ids"] == [first["turn_id"]] + assert wire.children["owned"].poll() is not None and wire.children["unowned"].poll() is None + assert all(store.load_turn(sid, turn) == original for turn, original in before.items()) + reads = [params for method, params in wire.calls if method == "thread/items/list"] + assert len([params for params in reads if "turnId" not in params]) == 1 + assert len(reads) == 2 # Discovery plus fresh exact-Turn cancellation proof. + assert sum(topic == root["message_id"] and "停止" in text + for _, topic, text in provider.topic_writes) == 1 + calls, writes = list(wire.calls), list(provider.writes) + transport.reconcile() + assert wire.calls == calls and provider.writes == writes + finally: + wire.close() + runtime.close() + + def test_idle_topic_stop_finds_earlier_completed_commands_and_keeps_receipts_and_peers(ordinary, monkeypatch): # noqa: F811 store, runtime, provider, transport, root, first, wire = _completed_with_later_answer(ordinary, monkeypatch) sid, tid = first["session_id"], first["turn_id"] @@ -249,9 +287,12 @@ def write(packet): result = {"data": [{"id": key, "status": "interrupted" if key in interrupted_native else "inProgress" if value["status"] in {"running", "interrupting"} else value["status"]} for key, value in turns.items()]} - elif method == "thread/items/list" and params["turnId"] == newer_native: + elif method == "thread/items/list" and params.get("turnId") in (None, newer_native): result = {"data": [{"turnId": newer_native, "item": { "type": "commandExecution", "id": "future-command", "status": "completed"}}]} + if "turnId" not in params: + result["data"].append({"turnId": native, "item": { + "type": "commandExecution", "id": "owned-command", "status": "completed"}}) else: if method == "turn/interrupt": interrupted_native.add(params["turnId"])