diff --git a/loopx/chat_ingress.py b/loopx/chat_ingress.py index 7052737e04..0922bed57d 100644 --- a/loopx/chat_ingress.py +++ b/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 diff --git a/tests/test_chat_session_active_turn.py b/tests/test_chat_session_active_turn.py index 93381af6c2..eb921e148e 100644 --- a/tests/test_chat_session_active_turn.py +++ b/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, diff --git a/tests/test_chat_turn_steering.py b/tests/test_chat_turn_steering.py index 0cd54da825..8958197117 100644 --- a/tests/test_chat_turn_steering.py +++ b/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