fix(chat): batch native command discovery for completed-topic stop

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

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 仍沿原确切中断路径。

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:

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

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,21 +54,59 @@ 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"
     first = finish_group_turn(runtime, transport, message=root["content"])
     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"])