fix(chat): fail closed on unreadable ingress receipts

loopx/chat_ingress.py

@@ -335,15 +335,21 @@ def mutate_queued_input(self, session_id: str, turn_id: str, request: dict[str,
 
     def steering_receipt(self, session_id: str, ingress_id: str):
         """A corrupt receipt cannot be treated as absence and resent."""
+        return self._strict_ingress_receipt(session_id, ingress_id)
+
+    def _strict_ingress_receipt(self, session_id: str, ingress_id: str) -> dict[str, Any] | None:
+        """Absent is ``None``; any other unreadable receipt fails closed."""
         from .chat_store import CHAT_INGRESS_SCHEMA_VERSION
         path = self._ingress_path(session_id, ingress_id)
         try:
             value = json.loads(path.read_text(encoding="utf-8"))
         except FileNotFoundError:
             return None
+        except (OSError, ValueError) as exc:
+            raise OSError("Ingress receipt is unreadable") from exc
         if (not isinstance(value, dict) or value.get("schema_version") != CHAT_INGRESS_SCHEMA_VERSION
                 or value.get("session_id") != session_id or value.get("client_ingress_id") != ingress_id):
-            raise OSError("Steering receipt is unreadable")
+            raise OSError("Ingress receipt is unreadable")
         return value
 
     def steering_delivery_lock(self, session_id: str, ingress_id: str):
@@ -588,7 +594,7 @@ def create_ingress_receipt(
     ) -> tuple[dict[str, Any], bool]:
         """Reserve one idempotent external ingress before provider delivery."""
 
-        from .chat_store import _opaque_id, _read_json, _atomic_write_json, utc_now, CHAT_INGRESS_SCHEMA_VERSION
+        from .chat_store import _opaque_id, _atomic_write_json, utc_now, CHAT_INGRESS_SCHEMA_VERSION
 
         if self.load_session(session_id) is None:
             raise KeyError("chat session was not found")
@@ -598,8 +604,10 @@ def create_ingress_receipt(
             agent_id="loopx-chat",
             operation="create_chat_ingress_receipt",
         ):
-            existing = _read_json(path)
-            if existing.get("schema_version") == CHAT_INGRESS_SCHEMA_VERSION:
+            # A damaged receipt may record a delivered send. It is not absence:
+            # recreating it would resend the message and erase that evidence.
+            existing = self._strict_ingress_receipt(session_id, client_ingress_id)
+            if existing is not None:
                 request = {"mode": _opaque_id(mode, field="mode"), "message": str(message)}
                 if expected_turn_id is not None or "expected_turn_id" in existing:
                     request["expected_turn_id"] = expected_turn_id

tests/test_chat_session_active_turn.py

@@ -200,6 +200,31 @@ def test_chat_idempotency_keys_reject_different_requests(tmp_path: Path) -> None
         )
 
 
+@pytest.mark.parametrize("damage", ['{"schema_version": ', "[]", '{"schema_version": "other"}'])
+def test_unreadable_ingress_receipt_is_not_recreated_as_absent(tmp_path: Path, damage: str) -> None:
+    store = ChatSessionStore(tmp_path)
+    session_id = str(
+        store.create_session(
+            goal_id="goal-one",
+            agent_id="codex",
+            executor_endpoint_id="codex",
+            adapter_kind="codex_app_server",
+            upstream_thread_id="thread-one",
+            upstream_mode="chat",
+        )["session_id"]
+    )
+    request = {"client_ingress_id": "ingress-key", "mode": "live_steering", "message": "first"}
+    store.create_ingress_receipt(session_id, **request)
+    store.update_ingress_receipt(session_id, "ingress-key", status="delivered", active_turn_id="turn-one")
+    path = store._ingress_path(session_id, "ingress-key")
+    path.write_text(damage, encoding="utf-8")
+
+    with pytest.raises(OSError, match="unreadable"):
+        store.create_ingress_receipt(session_id, **request)
+
+    assert path.read_text(encoding="utf-8") == damage
+
+
 def test_generated_message_id_skips_transcript_deduplication(
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,

tests/test_chat_turn_steering.py

@@ -440,6 +440,27 @@ def test_http_steering_deduplicates_and_never_retargets(conversation):
     runtime.interrupt_turn(session_id=session_id, turn_id=second)
 
 
+def test_http_steering_never_resends_after_its_receipt_becomes_unreadable(conversation, monkeypatch):
+    store, runtime, session_id, start, post = conversation
+    writes = []
+    original_write = CodexChatAgentSession._write
+
+    def record_write(self, payload):
+        writes.append(json.loads(json.dumps(payload)))
+        return original_write(self, payload)
+
+    monkeypatch.setattr(CodexChatAgentSession, "_write", record_write)
+    turn_id = start("unreadable-receipt")
+    assert post(turn_id)[0] == 200
+    receipt_path = store._ingress_path(session_id, "adjust-one")
+    receipt_path.write_text('{"schema_version": ', encoding="utf-8")
+    status, body = post(turn_id)
+    assert status == 409 and body["delivery_state"] == "unresolved"
+    assert receipt_path.read_text(encoding="utf-8") == '{"schema_version": ', "damaged evidence is preserved"
+    assert len([row for row in writes if row.get("method") == "turn/steer"]) == 1
+    runtime.interrupt_turn(session_id=session_id, turn_id=turn_id)
+
+
 @pytest.mark.parametrize("text,extra", [("", {}), (123, {}), ("x" * 12001, {}), ("hello", {"unexpected": True})])
 def test_http_steering_rejects_invalid_input_before_delivery(conversation, text, extra):
     store, _, session_id, start, post = conversation