diff --git a/loopx/capabilities/native_chat/external_conversations.py b/loopx/capabilities/native_chat/external_conversations.py index 8e08e0f06e..3ab8a0bae6 100644 --- a/loopx/capabilities/native_chat/external_conversations.py +++ b/loopx/capabilities/native_chat/external_conversations.py @@ -8,6 +8,7 @@ from collections.abc import Callable, Mapping, Sequence from copy import deepcopy +from functools import cache from pathlib import Path import hashlib import json @@ -26,7 +27,7 @@ from ...file_lock import exclusive_file_lock, try_exclusive_file_lock from . import result_review_evidence from .codex_context import checked_session_home -from .file_discovery import FileChatDiscovery +from .file_discovery import FileChatDiscovery, later_equivalent_source_input, source_control_fields # Request journals are named by their 24-hex request_ref; lock metadata and # other siblings in this directory are not requests. @@ -187,6 +188,8 @@ def _context_recovery_plan(self, row: dict[str, Any], *, _, open_literal, _ = _review_openings(str((turn.get("response") or {}).get("message") or ""), 0) preparation = None later_equivalent = None + read_source_controls = (discovery_reads.source_controls if discovery_reads is not None + else cache(self.source_controls)) image_failure = turn.get("error_code") == "runtime_error" and bool(row.get("attachments")) runtime_failure = turn.get("error_code") in {"model_auth_unavailable", "resume_failed"} if turn.get("error_code") == "manager_authorization_unavailable" or image_failure or runtime_failure: @@ -207,6 +210,10 @@ def _context_recovery_plan(self, row: dict[str, Any], *, and datetime.fromisoformat(item["created_at"]) > created for item in (discovery_reads.user_messages(session["session_id"]) if discovery_reads is not None else store.messages(session["session_id"]))) + elif turn.get("status") == "failed" and session.get("agent_id") == "codex": + # Native history cannot see a newer request that is still in the FIFO. + # Read the Core receipt when the source journal has not settled yet. + later_equivalent = later_equivalent_source_input(row, read_source_controls, store.turn_for_client) from ...chat_manager import steward_machine_defaults recovery_row = {key: value for key, value in row.items() if key != "attachments"} review_preparation = None @@ -239,15 +246,14 @@ def _context_recovery_plan(self, row: dict[str, Any], *, "attachments_present": bool(row.get("attachments")), "steering_observed": any(event.get("kind") == "turn.steered" for event in events), "stop_observed": self._source_stop_observed(row, turn, events, - source_controls=discovery_reads.source_controls if discovery_reads is not None else None), + source_controls=read_source_controls), "preparation_observation": preparation, "later_equivalent_input_present": later_equivalent, "preparation_scope_available": preparation_scope_available, "later_source_input_present": any(item.get("binding_id") == row["binding_id"] and item.get("source") == row["source"] and item.get("command") is None and datetime.fromisoformat(item["created_at"]) > datetime.fromisoformat(row["created_at"]) - for item in (discovery_reads.source_controls() if discovery_reads is not None - else self.source_controls())) if turn.get("error_code") == "antigravity_step_error" else None, + for item in read_source_controls()) if turn.get("error_code") == "antigravity_step_error" else None, "native_input": anchors[0] if len(anchors) == 1 else None, **({"retained_result_review": retained_review} if retained_review is not None else {}), **({"live_work_owner": live_work_owner} if unfinished or related_review is not None else {}), @@ -1687,7 +1693,7 @@ def pending(self) -> list[dict[str, Any]]: if _REQUEST_JOURNAL_STEM.fullmatch(path.stem)] def source_controls(self) -> list[dict[str, Any]]: - """Observe exact source-control fields without reparsing unchanged bodies. + """Observe source controls and private text digests without reparsing unchanged bodies. Recovery/stop rules remain with their original typed owner and callers. A new, replaced or modified journal is read again; pending() and exact @@ -1696,14 +1702,8 @@ def source_controls(self) -> list[dict[str, Any]]: paths = (path for path in sorted(self.root.glob("*.json")) if _REQUEST_JOURNAL_STEM.fullmatch(path.stem)) - def project(_path, row): - fields = ("binding_id", "source", "command", "created_at") - # The tolerant reader can return {} after transient IO failure. - # An incomplete read must not seed an unchanged-file observation. - return {key: row[key] for key in fields} if all(key in row for key in fields) else None - return [row for _path, row in self._source_control_discovery.rows(self.root, paths, - read=_read_json, project=project)] + read=_read_json, project=source_control_fields)] def read_request(self, request_ref: str) -> dict[str, Any]: return _read_json(self.root / f"{request_ref}.json") diff --git a/loopx/capabilities/native_chat/file_discovery.py b/loopx/capabilities/native_chat/file_discovery.py index 625ee8ad6b..7fd1a5a27f 100644 --- a/loopx/capabilities/native_chat/file_discovery.py +++ b/loopx/capabilities/native_chat/file_discovery.py @@ -3,12 +3,79 @@ from collections.abc import Callable, Iterable, Iterator from copy import deepcopy +from datetime import datetime +import hashlib from pathlib import Path import sys from threading import Lock from typing import Any +def _ordinary_request_text_digest(row: dict[str, Any]) -> str | None: + """Match ordinary private text without retaining its body in discovery.""" + message = row.get("message") + source = row.get("source") + status = row.get("status") + if (status not in ("accepted", "prepared") or row.get("command") is not None + or row.get("capture_origin") is not None or row.get("agent_target") is not None + or row.get("attachments") not in (None, []) or not isinstance(message, str) or not message + or not row.get("session_id") + or not isinstance(source, dict) or source.get("private_human_message") is not True): + return None + return hashlib.sha256(message.encode("utf-8", errors="surrogatepass")).hexdigest() + + +def source_control_fields(path: Path, row: dict[str, Any]) -> dict[str, Any] | None: + """Keep exact source controls and the metadata needed to prove admission.""" + fields = ("binding_id", "source", "command", "created_at") + # A tolerant reader can return an incomplete journal after a transient + # failure. Do not cache that read as an unchanged-file observation. + if not all(key in row for key in fields): + return None + result = {key: row[key] for key in fields} + digest = _ordinary_request_text_digest(row) + if digest is None: + return result + if row["status"] == "accepted" and row.get("turn_id"): + result.update(text_sha256=digest, session_id=row["session_id"]) + elif (row["status"] == "prepared" and row.get("turn_id") is None + and row.get("request_ref") == path.stem): + result.update(text_sha256=digest, session_id=row["session_id"], + prepared_client_turn_id=f"external-{path.stem}", origin=row.get("origin", "lark")) + return result + + +def later_equivalent_source_input( + row: dict[str, Any], + read_controls: Callable[[], list[dict[str, Any]]], + read_turn_for_client: Callable[[str, str], dict[str, Any] | None], +) -> bool | None: + """Check canonical acceptance in the same source and Session, even before dispatch.""" + if row.get("status") != "accepted" or not row.get("turn_id"): + return None + digest = _ordinary_request_text_digest(row) + if digest is None: + return None + created = datetime.fromisoformat(row["created_at"]) + for item in read_controls(): + if (item.get("text_sha256") != digest or item["binding_id"] != row["binding_id"] + or item["source"] != row["source"] or item["session_id"] != row["session_id"] + or datetime.fromisoformat(item["created_at"]) <= created): + continue + client_turn_id = item.get("prepared_client_turn_id") + if client_turn_id is None: + return True + # Core owns the durable queue receipt even if the outer journal has + # not finished its write. Never treat preparation alone as admission. + turn = read_turn_for_client(item["session_id"], client_turn_id) + if (turn is not None and turn.get("origin") == item["origin"] + and turn.get("_queue_original_message", turn.get("message")) == row["message"] + and turn.get("attachments") in (None, []) + and turn.get("external_agent_target") is None and turn.get("collaboration_request") is None): + return True + return False + + def file_change_fingerprint(path: Path) -> tuple[int, ...] | None: """A local mutation token, or no reuse when the platform lacks one.""" # Windows stat ctime can be creation time, not a mutation stamp. Keep diff --git a/loopx/control_plane/collaboration/conversation_binding.ts b/loopx/control_plane/collaboration/conversation_binding.ts index b951c2324e..7c3148aa2d 100644 --- a/loopx/control_plane/collaboration/conversation_binding.ts +++ b/loopx/control_plane/collaboration/conversation_binding.ts @@ -948,9 +948,9 @@ function planContextRecovery(params: JsonObject, row: JsonObject, current: JsonO admitted_turn_id: sessionIdentity(admitted.turn_id, "admitted recovery Turn"), session_id: current.session_id, native_thread_ids: []}; } - // A later source stop vetoes new execution. Correlation repair for an - // already admitted Turn above remains read-only and keeps its exact identity. - if (recovery.stop_observed === true) return no; + // A source stop or a newer admitted equivalent vetoes new execution. + // Correlation repair above remains read-only and keeps its exact identity. + if (recovery.stop_observed === true || recovery.later_equivalent_input_present === true) return no; const routing = planRoutingRecovery(recovery, row, current, failed, previous, version); if (routing !== null) return routing; if (!steward) return no; diff --git a/tests/control_plane_ts/conversation_binding.test.ts b/tests/control_plane_ts/conversation_binding.test.ts index 058aeafbd3..1be4b07419 100644 --- a/tests/control_plane_ts/conversation_binding.test.ts +++ b/tests/control_plane_ts/conversation_binding.test.ts @@ -1349,7 +1349,8 @@ for (const kind of ["project", "steward"]) for (const retryKind of ["routing", " context_recovery: {...packet.context_recovery, native_observation: null, admitted_turn: admitted}}; assert.equal(planBoundConversationRequest(crash).admitted_turn_id, "already-admitted"); assert.equal(planBoundConversationRequest({...crash, - context_recovery: {...crash.context_recovery, stop_observed: true}}).admitted_turn_id, "already-admitted"); + context_recovery: {...crash.context_recovery, stop_observed: true, + later_equivalent_input_present: true}}).admitted_turn_id, "already-admitted"); for (const change of [{session_id: "foreign"}, {client_turn_id: "foreign"}, {message: "another request"}, {attachments: [{type: "image"}]}]) { assert.equal(planBoundConversationRequest({...crash, @@ -1362,6 +1363,7 @@ for (const kind of ["project", "steward"]) for (const retryKind of ["routing", " assert.equal(planBoundConversationRequest({...next, request: {...next.request, context_recovery: {...reservation, [attemptKey]: 3}}}).recovery_candidate, false); for (const change of [{source_verified: false}, {steering_observed: true}, {stop_observed: true}, + {later_equivalent_input_present: true}, {observed_at: "2026-01-01T00:00:59Z"}, {observed_at: "invalid"}, {native_observation: {...observed, ...(retryKind === "quota" ? {only_original_user_item: false} : {only_pre_execution_items: false})}}, diff --git a/tests/test_chat_conversation_bindings.py b/tests/test_chat_conversation_bindings.py index b824abebb4..97ee9cf4d6 100644 --- a/tests/test_chat_conversation_bindings.py +++ b/tests/test_chat_conversation_bindings.py @@ -186,6 +186,106 @@ def now(cls, tz=None): runtime.close() +@pytest.mark.parametrize("lost_receipt", [False, True], ids=["accepted", "lost-outer-receipt"]) +def test_original_retry_yields_to_a_newer_equivalent_lark_request(ordinary, monkeypatch, lost_receipt): + from datetime import datetime, timedelta, timezone + + from loopx.capabilities.native_chat import external_conversations + from loopx.capabilities.native_chat.codex_adapter import CodexAppServerAdapter + from test_lark_private_conversations import connect + + store, runtime, provider, transport = connect(ordinary) + fake = ordinary[-2] + fake.write_text(fake.read_text().replace('"codexErrorInfo": "cyberPolicy"', + '"codexErrorInfo": "usageLimitExceeded"').replace( + ' print(json.dumps({"method": "item/agentMessage/delta", "params": {"threadId": "durable-thread", "turnId": active_turn, "delta": "Partial answer."}}), flush=True)\n', '')) + + def admit(profile, name, text): + event = provider.event(profile, name, text) + transport.admit(profile, event) + provider_ref = transport._binding(profile)["provider_ref"] + return transport.core.read_request(identity_ref(provider_ref, event["message_id"])) + + text = "typed terminal failure" + try: + original = admit("notes-app", "quota-original", text) + sid, tid = original["session_id"], original["turn_id"] + failed = runtime.wait_for_turn(session_id=sid, turn_id=tid, timeout_sec=10) + assert failed["error_code"] == "usage_limit_exceeded" and failed["delta_count"] == 0 + transport.reconcile() + original = transport.core.read_request(original["request_ref"]) + failed_bytes = store._turn_path(sid, tid).read_bytes() + lark_failure_writes = list(provider.writes) + assert original["delivery_verified"] and lark_failure_writes + + # Leave new source inputs in the real FIFO before they become native Turns. + monkeypatch.setattr(runtime, "resume_session_queue", lambda **_: None) + unrelated = admit("steward-app", "other-source", text) + distinct = admit("notes-app", "different-text", "Another request") + assert unrelated["session_id"] != sid and distinct["session_id"] == sid + + class AfterBackoff(datetime): + @classmethod + def now(cls, tz=None): + return datetime.now(timezone.utc) + timedelta(seconds=61) + + monkeypatch.setattr(external_conversations, "datetime", AfterBackoff) + monkeypatch.setattr(CodexAppServerAdapter, "failed_request_observation", staticmethod(lambda **_: { + "thread_id": "durable-thread", "turn_id": failed["upstream_turn_id"], + "only_original_user_item": True, "codex_error_info": "usageLimitExceeded", + "quota_available": True, "quota_observed_at": AfterBackoff.now().timestamp() - 1})) + assert transport.core.context_recovery_candidate(original["request_ref"]) + + resend = provider.event("notes-app", "same-source-resend", text) + resend_ref = identity_ref(transport._binding("notes-app")["provider_ref"], resend["message_id"]) + resend_client = f"external-{resend_ref}" + if lost_receipt: + def fail_before_queue(**_): + raise OSError("Core queue unavailable") + + with monkeypatch.context() as fault: + fault.setattr(runtime, "enqueue_turn", fail_before_queue) + with pytest.raises(OSError, match="Core queue unavailable"): + transport.admit("notes-app", resend) + assert transport.core.read_request(resend_ref)["session_id"] == sid + assert store.turn_for_client(sid, resend_client) is None + assert transport.core.context_recovery_candidate(original["request_ref"]) + + write = external_conversations._atomic_write_json + + def lose_outer_receipt(path, row, **kwargs): + if path.stem == resend_ref and row.get("status") == "accepted": + raise OSError("outer receipt unavailable") + write(path, row, **kwargs) + + with monkeypatch.context() as fault: + fault.setattr(external_conversations, "_atomic_write_json", lose_outer_receipt) + with pytest.raises(OSError, match="outer receipt unavailable"): + transport.admit("notes-app", resend) + prepared = transport.core.read_request(resend_ref) + assert prepared["status"] == "prepared" and prepared["turn_id"] is None + else: + transport.admit("notes-app", resend) + assert transport.core.read_request(resend_ref)["status"] == "accepted" + + resent = store.turn_for_client(sid, resend_client) + assert resent["session_id"] == sid and resent["status"] == "queued" + assert any(item["kind"] == "turn.queued" for item in store.events_after(sid, resent["turn_id"], None)) + expected_queue = {distinct["turn_id"], resent["turn_id"]} + assert {row["turn_id"] for row in store.queued_turns(sid)} == expected_queue + transport.reconcile_request(transport.root / f"{original['request_ref']}.json") + assert transport.core.read_request(original["request_ref"]) == original + assert not transport.core.context_recovery_candidate(original["request_ref"]) + assert {row["turn_id"] for row in store.queued_turns(sid)} == expected_queue + assert store._turn_path(sid, tid).read_bytes() == failed_bytes + assert provider.writes == lark_failure_writes + transport.admit("notes-app", resend) + assert transport.core.read_request(resend_ref)["turn_id"] == resent["turn_id"] + assert {row["turn_id"] for row in store.queued_turns(sid)} == expected_queue + finally: + runtime.close() + + @pytest.fixture def recorded_return_continuation(private_return, monkeypatch): # noqa: F811 """Use the real admission/recovery owners; only model/provider IO is fake."""