Prevent duplicate private bot retries after a resend

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")

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

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;

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})}},

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."""