From c51907c47ba3ad71715a4a1ccc14c6677ad77401 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 5 Nov 2026 23:13:56 +0800 Subject: [PATCH] fix(chat): observe unfinished work through its original native adapter Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../capabilities/native_chat/codex_adapter.py | 19 +++++++--- .../native_chat/external_conversations.py | 21 +++++++--- tests/test_chat_codex_history.py | 7 +++- tests/test_lark_private_conversations.py | 38 ++++++++++++++++--- 4 files changed, 66 insertions(+), 19 deletions(-) diff --git a/loopx/capabilities/native_chat/codex_adapter.py b/loopx/capabilities/native_chat/codex_adapter.py index fdae3e2fff..59d9d4e430 100644 --- a/loopx/capabilities/native_chat/codex_adapter.py +++ b/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": diff --git a/loopx/capabilities/native_chat/external_conversations.py b/loopx/capabilities/native_chat/external_conversations.py index 991f37f62c..1b4269319d 100644 --- a/loopx/capabilities/native_chat/external_conversations.py +++ b/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) diff --git a/tests/test_chat_codex_history.py b/tests/test_chat_codex_history.py index 80eca414ac..379d2ccfed 100644 --- a/tests/test_chat_codex_history.py +++ b/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, ) diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py index 15885fa136..00eb12d0a9 100644 --- a/tests/test_lark_private_conversations.py +++ b/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" + json.dumps({ "message": "", "unfinished_work": work}) + "" 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):