Reuse the original live thread for project result readback

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

@@ -2039,8 +2039,10 @@ command cannot acquire replay authority through this route.
 
 The original worker uses its existing Session, queue and project binding. Fresh
 source and grant checks, an idle queue, the backoff, and stop/steering/attachment
-guards still apply. A live adapter with owned background work is retained. The
-review uses the existing workspace-only read permission profile without network
+guards still apply. With a live adapter, the readback uses that same adapter
+only when it owns the exact thread and has no active or background work. The
+cold reader is used only when no adapter owns the Session. The review
+uses the existing workspace-only read permission profile without network
 access; it may inspect current local artifacts and receipts but cannot write,
 send, delegate or replace the original work declaration. References in the old
 question use its original-time Core context. The original partial Turn and

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

@@ -1340,8 +1340,9 @@ completed Turn 继续明确保留，本路径不提供通用动作或执行重
 已经结清**；失败或被信号终止的命令不能借此取得重放权限。
 
 原 worker 复用已有 Session、队列和项目 binding，仍核验新鲜来源和授权、空闲队列、
-等待间隔，以及停止、steering 和附件边界。仍持有后台工作的 live adapter 必须保留。
-复核使用现有工作区只读权限配置，禁用网络；可检查当前本地产物和回执，不能写入、
+等待间隔，以及停止、steering 和附件边界。原生进程仍在时，只在它持有原线程且
+没有运行中或后台工作时通过该进程读回结果。会话没有现成 adapter 时才启动冷读取。
+复核使用现有工作区只读权限配置并禁用网络。它可检查当前本地产物和回执，不能写入、
 发送、委派或替换原未完声明。旧问题中的指代仅使用原提问之前的 Core 上下文。
 原部分答复 Turn 和投递保持不变，由同一个飞书 worker 使用独立、幂等阶段补回新
 答复；预约崩溃只采用同一个已入队复核。如果可写续办已入队、原请求日志尚未更新，

loopx/capabilities/native_chat/external_conversations.py

@@ -365,7 +365,7 @@ def recover_failed_request(self, request_ref: str, *,
         Core or native no-effects readback; a crash resumes that reservation, never replays the
         failed Turn or fabricates successful execution.
         """
-        from .codex_adapter import CodexAppServerAdapter
+        from .codex_adapter import CodexAppServerAdapter, completed_result_review_readback
         from ...chat_manager import MANAGER_CONTEXT_VERSION
 
         path = self.root / f"{request_ref}.json"
@@ -428,41 +428,33 @@ def source_requests() -> list[dict[str, Any]]:
                             model_id=native_session.get("agent_model_id") or ANTIGRAVITY_DEFAULT_MODEL)
             elif not plan.get("admitted_turn_id") and not preparation_retry:
                 result_review = bool(plan.get("read_only_review") and plan.get("unfinished_work"))
-                if result_review:
-                    with self.controller.lock:
-                        original_adapter = self.controller.adapters.get(row["session_id"])
-                    if original_adapter is not None and (not isinstance(original_adapter, CodexAppServerAdapter)
-                            or not original_adapter.answer_review_replaceable()):
-                        return row  # Do not reserve a review that would abandon owned work.
-                if plan.get("work_attempt"):
-                    # Only the original live adapter can observe its background
-                    # effects. Never turn a cold client's missing thread into
-                    # permission to continue writable work.
+                if result_review or plan.get("work_attempt"):
                     with self.controller.lock:
                         adapter = self.controller.adapters.get(row["session_id"])
-                    if isinstance(adapter, CodexAppServerAdapter):
+                    if not result_review and isinstance(adapter, CodexAppServerAdapter):
+                        # Only the original live owner can prove writable effects.
                         proof = adapter.completed_work_observation(
                             thread_ids=plan["native_thread_ids"], turn_id=turn["upstream_turn_id"],
                             known_turn_ids=self._settled_native_turn_ids(session["session_id"]))
                     if proof is None and (adapter is None or isinstance(adapter, CodexAppServerAdapter)
+                            and adapter.upstream_thread_id in plan["native_thread_ids"]
                             and adapter.answer_review_replaceable()):
-                        # A closed partial answer may be inspected even when its
-                        # effects cannot authorize writing. This cold reader
-                        # supplies result evidence only, never writable proof.
-                        proof = CodexAppServerAdapter.failed_request_observation(
-                            codex_bin=self.controller.codex_bin, codex_home=Path(session["codex_home"]),
-                            work_dir=context["project"], read_only_review=True, completed_result_review=True,
-                            known_turn_ids=self._settled_native_turn_ids(session["session_id"], include_interrupted=True),
-                            thread_ids=plan["native_thread_ids"], turn_id=turn["upstream_turn_id"])
+                        known = self._settled_native_turn_ids(session["session_id"], include_interrupted=True)
+                        if adapter is None:
+                            # Use a cold reader only when there is no live owner.
+                            proof = CodexAppServerAdapter.failed_request_observation(
+                                codex_bin=self.controller.codex_bin, codex_home=Path(session["codex_home"]),
+                                work_dir=context["project"], read_only_review=True, completed_result_review=True,
+                                known_turn_ids=known, thread_ids=plan["native_thread_ids"],
+                                turn_id=turn["upstream_turn_id"])
+                        else:
+                            proof = completed_result_review_readback(
+                                adapter.session, turn["upstream_turn_id"], known_turn_ids=known)
                 else:
                     proof = CodexAppServerAdapter.failed_request_observation(
                         codex_bin=self.controller.codex_bin, codex_home=Path(session["codex_home"]),
                         model_auth_home=model_auth_home,
                         read_only_review=bool(plan.get("read_only_review")) and not quota_answer,
-                        completed_result_review=result_review,
-                        work_dir=context["project"] if result_review else None,
-                        known_turn_ids=(self._settled_native_turn_ids(session["session_id"], include_interrupted=True)
-                                        if result_review else None),
                         thread_ids=plan["native_thread_ids"], turn_id=turn["upstream_turn_id"])
                     if (proof is not None and proof.get("codex_error_info") == "usageLimitExceeded"
                             and proof.get("only_original_user_item") is True and session.get("steward_context")):

tests/test_lark_private_conversations.py

@@ -2049,9 +2049,14 @@ def test_original_worker_continues_explicit_unfinished_project_work_once_after_r
     source = fake.read_text()
     # Persisted history is readable from a cold client. Background terminals
     # belong to the process that loaded the thread, as in native app-server.
-    source = source.replace("turn_count = 0", "loaded = False\nturn_count = 0")
+    source = source.replace("turn_count = 0", "loaded = False\nreview_only = False\nturn_count = 0")
     source = source.replace('    elif method in {"thread/start", "thread/resume"}:',
-        '    elif method in {"thread/start", "thread/resume"}:\n        loaded = True')
+        '    elif method in {"thread/start", "thread/resume"}:\n        loaded = True\n'
+        '        review_only = request["params"].get("permissions") == "loopx_workspace_only_read"')
+    source = source.replace('        result = {"thread": {"id": "durable-thread"},',
+        '        result = {"thread": {"id": "durable-thread"},\n'
+        '                  "activePermissionProfile": {"id": request["params"].get("permissions")},\n'
+        '                  "runtimeWorkspaceRoots": request["params"].get("runtimeWorkspaceRoots"),')
     source = source.replace("turn_count = 0", f"turn_count = len(json.load(open({str(inventory)!r}))[\"turns\"]) if __import__(\"os\").path.exists({str(inventory)!r}) else 0")
     start = source.index("        response = 'Visible runtime response.")
     end = source.index('        print(json.dumps({"method": "item/agentMessage/delta"', start)
@@ -2061,14 +2066,18 @@ def test_original_worker_continues_explicit_unfinished_project_work_once_after_r
         previous = json.loads(path.read_text()) if path.exists() else {{"turns": []}}
         response = {partial!r} if first or {fault == "still-incomplete"!r} else "Verified remaining work completed."
         artifact = Path({str(artifact)!r})
-        if "Newer follow-up" in json.dumps(request["params"]):
+        if review_only:
+            response = "Verified partial result. The remaining work is still pending."
+        elif "Newer follow-up" in json.dumps(request["params"]):
             response = "Newer input processed."
         else:
             artifact.write_text((artifact.read_text() if artifact.exists() else "") + ("first\\n" if first else "remaining\\n"))
         path.write_text(json.dumps({{"turns": [{{"id": active_turn, "status": "completed"}}, *previous["turns"]], "items": [
             {{"turnId": active_turn, "item": {{"id": "user", "type": "userMessage"}}}},
-            {{"turnId": active_turn, "item": {{"id": "effect", "type": "fileChange", "status": "completed"}}}},
-            {{"turnId": active_turn, "item": {{"id": "view", "type": "imageView", "path": str(artifact)}}}},
+            *([] if review_only else [
+                {{"turnId": active_turn, "item": {{"id": "effect", "type": "fileChange", "status": "completed"}}}},
+                {{"turnId": active_turn, "item": {{"id": "view", "type": "imageView", "path": str(artifact)}}}},
+            ]),
             {{"turnId": active_turn, "item": {{"id": "answer", "type": "agentMessage", "phase": "final_answer", "text": response}}}},
             *previous.get("items", [])]}}))
 ''' + source[end:]
@@ -2186,11 +2195,43 @@ def crash(**kwargs):
         if fault in {"queued-newer", "queued-items", "queued-input"}:
             monkeypatch.setattr(runtime, "resume_session_queue", resume)
         selected = worker()
+    artifact_before_recovery = artifact.read_bytes()
     selected.reconcile()
     continued = selected.core.read_request(ref)
-    if fault in {"changed-source", "revoked", "stop", "unknown-native", "background", "later-user", "cold-native", "dead-native", "foreign-native", "failed-effect", "equivalent-later", "later-failed-effect", "unsettled-command"}:
+    reviewed = fault in {"cold-native", "failed-effect", "equivalent-later", "later-failed-effect", "unsettled-command"}
+    if fault in {"changed-source", "revoked", "stop", "unknown-native", "background", "later-user", "dead-native", "foreign-native"}:
         assert continued["turn_id"] == original["turn_id"] and not continued.get("context_recovery")
-        assert artifact.read_text() == ("first\nremaining\n" if fault == "equivalent-later" else "first\n")
+        assert artifact.read_bytes() == artifact_before_recovery
+    elif reviewed:
+        assert continued["turn_id"] != original["turn_id"]
+        assert continued["session_id"] == original["session_id"]
+        recovery = continued["context_recovery"]
+        assert recovery["read_only_review"] is True and "work_attempt" not in recovery
+        assert recovery["native_observation"]["completed_result_review"] is True
+        assert recovery["native_observation"].get("all_effects_settled") is not True
+        review = runtime.wait_for_turn(session_id=continued["session_id"], turn_id=continued["turn_id"], timeout_sec=10)
+        assert review["status"] == "completed", review
+        assert not review["response"].get("unfinished_work")
+        assert artifact.read_bytes() == artifact_before_recovery
+        selected.reconcile_request(path)
+        deliveries = json.loads(path.read_text())["deliveries"]
+        assert deliveries["terminal"] == original_delivery
+        returned = deliveries[presentation_phase(continued, "terminal")]
+        assert returned["verified"] and "The remaining work is still pending." in returned["text"]
+        writes = list(provider.writes)
+        assert not worker().core.context_recovery_candidate(ref)
+        worker().reconcile()
+        assert provider.writes == writes
+
+        recovery_calls = [json.loads(line) for line in capture.read_text().splitlines()][len(before_recovery_calls):]
+        resume, = [row["params"] for row in recovery_calls if row.get("method") == "thread/resume"]
+        assert resume["threadId"] == session_before["upstream_thread_id"]
+        profile = "loopx_workspace_only_read"
+        assert resume["permissions"] == profile and resume["runtimeWorkspaceRoots"] == [str(workspace.resolve())]
+        permissions = resume["config"]["permissions"][profile]
+        assert permissions["filesystem"][":workspace_roots"] == {".": "read"}
+        assert permissions["network"]["enabled"] is False
+        assert sum(row.get("method") == "turn/start" for row in recovery_calls) == 1
     else:
         assert continued["turn_id"] != original["turn_id"]
         assert continued["session_id"] == original["session_id"]
@@ -2228,9 +2269,10 @@ def crash(**kwargs):
             assert "Verified remaining work completed." in bodies
     assert store._turn_path(original["session_id"], original["turn_id"]).read_bytes() == original_turn_bytes
     after_recovery_calls = [json.loads(line) for line in capture.read_text().splitlines()]
-    for method in ("initialize", "thread/start", "thread/resume"):
-        assert sum(row.get("method") == method for row in after_recovery_calls) == sum(
-            row.get("method") == method for row in before_recovery_calls)
+    expected_new = {"initialize": int(reviewed) + int(fault == "cold-native"),
+        "thread/start": 0, "thread/resume": int(reviewed)}
+    for method, count in expected_new.items():
+        assert sum(row.get("method") == method for row in after_recovery_calls[len(before_recovery_calls):]) == count
 
 
 def _app_projection_journey(fixture):