Fix unfinished-work observation through the original native adapter

loopx/capabilities/native_chat/codex_adapter.py

@@ -116,7 +116,7 @@ def completed_work_readback(session: CodexChatAgentSession, turn_id: str) -> dic
     """
     from ...chat import parse_agent_response
 
-    deadline = time.monotonic() + session.response_timeout_sec
+    deadline = time.monotonic() + min(session.response_timeout_sec, 5.0)
     try:
         def turns():
             return list(session._control_items("thread/turns/list", {
@@ -152,7 +152,7 @@ def turns():
             elif kind in {"commandExecution", "fileChange", "mcpToolCall"}:
                 if item.get("status") != "completed":
                     return None  # Declined/failed/live/unknown effects cannot request writable continuation.
-            elif kind not in {"reasoning", "webSearch", "contextCompaction"}:
+            elif kind not in {"reasoning", "webSearch", "imageView", "contextCompaction"}:
                 return None
         if (users != 1 or len(finals) != 1 or turns() != inventory
                 or list(session._control_items("thread/items/list", {"turnId": turn_id}, deadline)) != items
@@ -223,11 +223,21 @@ class CodexAppServerAdapter:
     session: CodexChatAgentSession
     goal_driver: CodexGoalDriver | None = None
 
+    def completed_work_observation(self, *, thread_ids: list[str], turn_id: str) -> dict[str, Any] | None:
+        """Read live effects through the adapter that owns the original thread.
+
+        A cold app-server can read persisted items, but cannot inventory another
+        process's background terminals. Missing ownership is not an empty list.
+        Do not resume or replace a native thread to manufacture that proof.
+        """
+        if self.upstream_thread_id not in thread_ids or not self.healthcheck():
+            return None
+        return completed_work_readback(self.session, turn_id)
+
     @staticmethod
     def failed_request_observation(*, codex_bin: str, codex_home: Path,
                                    thread_ids: list[str], turn_id: str,
                                    read_only_review: bool = False,
-                                   completed_work: bool = False,
                                    work_dir: Path | None = None,
                                    model_auth_home: Path | None = None) -> dict[str, Any] | None:
         """Read persisted native metadata/items; create no native conversation.
@@ -253,8 +263,7 @@ def failed_request_observation(*, codex_bin: str, codex_home: Path,
             for thread_id in dict.fromkeys(thread_ids):
                 native.thread_id = thread_id
                 native.response_timeout_sec = max(.01, deadline - time.monotonic())
-                reader = (completed_work_readback if completed_work else
-                          interrupted_result_readback if read_only_review else failed_request_readback)
+                reader = interrupted_result_readback if read_only_review else failed_request_readback
                 proof = reader(native, turn_id)
                 if proof is not None:
                     if proof.get("codex_error_info") == "usageLimitExceeded":

loopx/capabilities/native_chat/external_conversations.py

@@ -415,12 +415,21 @@ def source_requests() -> list[dict[str, Any]]:
                             cascade_id=native_session["upstream_thread_id"],
                             model_id=native_session.get("agent_model_id") or ANTIGRAVITY_DEFAULT_MODEL)
             elif not plan.get("admitted_turn_id") and not preparation_retry:
-                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": True} if plan.get("read_only_review") else {}),
-                    **({"completed_work": True, "work_dir": Path(context["project"])} if plan.get("work_attempt") else {}),
-                    thread_ids=plan["native_thread_ids"], turn_id=turn["upstream_turn_id"])
+                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.
+                    with self.controller.lock:
+                        adapter = self.controller.adapters.get(row["session_id"])
+                    if isinstance(adapter, CodexAppServerAdapter):
+                        proof = adapter.completed_work_observation(
+                            thread_ids=plan["native_thread_ids"], turn_id=turn["upstream_turn_id"])
+                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": True} if plan.get("read_only_review") else {}),
+                        thread_ids=plan["native_thread_ids"], turn_id=turn["upstream_turn_id"])
             facts = proof.pop("retained_result", None) if proof is not None and metadata_review else None
             selected = self.bindings.resolve(binding_id=row["binding_id"], **row["source"], include_observation=True)
             source = source_observation(selected)

tests/test_chat_codex_history.py

@@ -9,7 +9,7 @@
 from loopx.capabilities.native_chat.codex_adapter import CodexAppServerAdapter
 
 
-@pytest.mark.parametrize("fault", [None, "later-user", "live", "failed", "background", "unknown", "live-tool",
+@pytest.mark.parametrize("fault", [None, "image-view", "later-user", "live", "failed", "background", "unknown", "live-tool",
     "declined", "duplicate", "foreign", "second-user", "no-final", "legacy-prose", "moving-turn", "moving-items"])
 def test_completed_work_readback_requires_latest_stable_explicit_settled_inventory(fault):
     from loopx.chat import CHAT_REVIEW_OPEN_TAG, CHAT_REVIEW_CLOSE_TAG
@@ -34,6 +34,8 @@ def items(method, params, deadline):
         effect = {"id": "effect", "type": "fileChange", "status": "completed"}
         if fault == "unknown":
             effect["type"] = "futureEffect"
+        if fault == "image-view":
+            effect = {"id": "view", "type": "imageView", "path": "/synthetic/notes/figure.png"}
         if fault in {"live-tool", "declined"}:
             effect["status"] = "inProgress" if fault == "live-tool" else "declined"
         rows = [{"turnId": "original-turn", "item": {"id": "user", "type": "userMessage"}},
@@ -50,7 +52,7 @@ def items(method, params, deadline):
     native = SimpleNamespace(thread_id="native", work_dir=Path("/synthetic/notes"), response_timeout_sec=.1, _control_items=items,
         _background_terminals=lambda deadline: [{"processId": "live"}] if fault == "background" else [])
     proof = completed_work_readback(native, "original-turn")
-    if fault is None:
+    if fault in {None, "image-view"}:
         assert proof == {"thread_id": "native", "turn_id": "original-turn", "completed_work": True,
                          "all_effects_settled": True, "unfinished_work": work}
         assert counts == {"thread/turns/list": 2, "thread/items/list": 2}
@@ -133,6 +135,7 @@ def test_controller_keeps_history_out_of_repeated_codex_objective(monkeypatch, t
     monkeypatch.setattr(CodexAppServerAdapter, "start", lambda **kwargs: starts.append(kwargs) or sentinel)
     controller = SimpleNamespace(
         codex_bin="fixture", codex_home=tmp_path,
+        coordination_runtime_root=None,
         startup_timeout_sec=1, idle_timeout_sec=1, hard_timeout_sec=1,
         _session_objective=chat_runtime.ChatRuntimeController._session_objective,
     )

tests/test_lark_private_conversations.py

@@ -1598,8 +1598,9 @@ def counted_revision():
 
 
 @pytest.mark.parametrize("fault", [None, "changed-source", "revoked", "stop", "unknown-native",
-    "background", "later-user", "reservation-crash", "post-enqueue-crash", "queued-newer", "queued-input", "private-path", "still-incomplete"])
+    "background", "later-user", "reservation-crash", "post-enqueue-crash", "queued-newer", "queued-input", "private-path", "still-incomplete", "cold-native", "dead-native", "foreign-native", "failed-effect"])
 def test_original_worker_continues_explicit_unfinished_project_work_once_after_restart(ordinary, monkeypatch, fault):  # noqa: F811
+    """Reply-worker restart retains native ownership; native loss stays fenced."""
     from datetime import datetime, timedelta, timezone
     from pathlib import Path
     from loopx.capabilities.native_chat import external_conversations
@@ -1617,6 +1618,11 @@ def test_original_worker_continues_explicit_unfinished_project_work_once_after_r
     partial = "Partial result retained; remaining work is blocked.\n<loopx-review-json>" + json.dumps({
         "message": "", "unfinished_work": work}) + "</loopx-review-json>"
     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('    elif method in {"thread/start", "thread/resume"}:',
+        '    elif method in {"thread/start", "thread/resume"}:\n        loaded = True')
     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)
@@ -1633,6 +1639,7 @@ def test_original_worker_continues_explicit_unfinished_project_work_once_after_r
         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)}}}},
             {{"turnId": active_turn, "item": {{"id": "answer", "type": "agentMessage", "phase": "final_answer", "text": response}}}}]}}))
 ''' + source[end:]
     start = source.index('    elif method == "thread/turns/list":')
@@ -1641,6 +1648,9 @@ def test_original_worker_continues_explicit_unfinished_project_work_once_after_r
         snapshot = json.load(open({str(inventory)!r}))
         result = {{"data": snapshot["turns" if method == "thread/turns/list" else "items"]}}
     elif method == "thread/backgroundTerminals/list":
+        if not loaded:
+            print(json.dumps({{"id": request["id"], "error": {{"code": -32600, "message": "thread not found"}}}}), flush=True)
+            continue
         result = {{"data": [{{"processId": "still-live"}}] if {fault == "background"!r} else []}}
 ''' + source[end:]
     fake.write_text(source)
@@ -1660,7 +1670,13 @@ def test_original_worker_continues_explicit_unfinished_project_work_once_after_r
     original_turn_bytes = store._turn_path(original["session_id"], original["turn_id"]).read_bytes()
     session_before = store.load_session(original["session_id"])
     Path(session_before["codex_home"]).mkdir(parents=True, exist_ok=True)
-    runtime.adapters.pop(original["session_id"]).close_session()
+    if fault == "cold-native":
+        runtime.adapters.pop(original["session_id"]).close_session()
+    elif fault == "dead-native":
+        runtime.adapters[original["session_id"]].close_session()
+    elif fault == "foreign-native":
+        runtime.adapters[original["session_id"]].session.thread_id = "foreign-thread"
+    before_recovery_calls = [json.loads(line) for line in capture.read_text().splitlines()]
     clock = datetime.now(timezone.utc) + timedelta(seconds=61)
     class AfterBackoff(datetime):
         @classmethod
@@ -1673,10 +1689,12 @@ def now(cls, tz=None):
         transport.bindings.disconnect(binding["binding_id"], expected_revision=transport.bindings.read()["revision"])
     if fault == "stop":
         transport.admit("notes-app", provider.event("notes-app", "stop-later", "/stop"))
-    if fault in {"unknown-native", "later-user"}:
+    if fault in {"unknown-native", "later-user", "failed-effect"}:
         snapshot = json.loads(inventory.read_text())
         if fault == "unknown-native":
             snapshot["items"].append({"turnId": turn["upstream_turn_id"], "item": {"id": "unknown", "type": "futureEffect"}})
+        elif fault == "failed-effect":
+            snapshot["items"][1]["item"]["status"] = "failed"
         else:
             snapshot["turns"].insert(0, {"id": "newer-user", "status": "completed"})
         inventory.write_text(json.dumps(snapshot))
@@ -1686,6 +1704,11 @@ def worker():
     selected = worker()
     if fault in {"reservation-crash", "post-enqueue-crash", "queued-newer", "queued-input"}:
         enqueue = runtime.enqueue_turn
+        resume = runtime.resume_session_queue
+        if fault in {"queued-newer", "queued-input"}:
+            # Hold dispatch before enqueue, rather than relying on a slow
+            # adapter startup to let the newer-input fixture win a race.
+            monkeypatch.setattr(runtime, "resume_session_queue", lambda **kwargs: None)
         def crash(**kwargs):
             if fault in {"post-enqueue-crash", "queued-newer", "queued-input"}:
                 enqueue(**kwargs)
@@ -1699,14 +1722,13 @@ def crash(**kwargs):
             snapshot["turns"].insert(0, {"id": "newer-after-reservation", "status": "completed"})
             inventory.write_text(json.dumps(snapshot))
         if fault == "queued-input":
-            resume = runtime.resume_session_queue
-            monkeypatch.setattr(runtime, "resume_session_queue", lambda **kwargs: None)
             transport.admit("notes-app", provider.event("notes-app", "newer-input", "Newer follow-up: inspect this correction first."))
+        if fault in {"queued-newer", "queued-input"}:
             monkeypatch.setattr(runtime, "resume_session_queue", resume)
         selected = worker()
     selected.reconcile()
     continued = selected.core.read_request(ref)
-    if fault in {"changed-source", "revoked", "stop", "unknown-native", "background", "later-user"}:
+    if fault in {"changed-source", "revoked", "stop", "unknown-native", "background", "later-user", "cold-native", "dead-native", "foreign-native", "failed-effect"}:
         assert continued["turn_id"] == original["turn_id"] and not continued.get("context_recovery")
         assert artifact.read_text() == "first\n"
     else:
@@ -1745,6 +1767,10 @@ def crash(**kwargs):
         if fault is None:
             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)
 
 
 def _app_projection_journey(fixture):