diff --git a/loopx/chat_store.py b/loopx/chat_store.py index dcbfc69acc..d90119ae2f 100644 --- a/loopx/chat_store.py +++ b/loopx/chat_store.py @@ -86,6 +86,13 @@ def _session_channel(payload: dict[str, Any]) -> str: return f"goal.{payload.get('goal_id')}" +def _require_matching_replay( + existing: dict[str, Any], *, identity: str, request: dict[str, Any] +) -> None: + if any(existing.get(field) != value for field, value in request.items()): + raise ValueError(f"{identity} already belongs to a different request") + + def _atomic_write_json(path: Path, payload: dict[str, Any], *, preserve_mode: bool = False) -> None: previous_mode = path.stat().st_mode & 0o777 if preserve_mode and path.exists() else 0o600 temporary = path.with_name(f".{path.name}.{uuid.uuid4().hex}.tmp") @@ -511,6 +518,11 @@ def create_ingress_receipt( ): existing = _read_json(path) if existing.get("schema_version") == CHAT_INGRESS_SCHEMA_VERSION: + _require_matching_replay( + existing, + identity="client_ingress_id", + request={"mode": _opaque_id(mode, field="mode"), "message": str(message)}, + ) return existing, False now = utc_now() payload = { @@ -575,6 +587,25 @@ def create_turn( ): existing = self.turn_for_client(session_id, client_id) if existing is not None: + # Attachments live in the transcript, not the Turn record. + # This also supports pre-existing Turns without a migration. + original = next( + (row for row in self.messages(session_id) + if row.get("role") == "user" + and row.get("turn_id") == existing["turn_id"]), + None, + ) + if original is None: + raise ValueError("client_turn_id original request is unavailable") + _require_matching_replay( + {**existing, "attachments": original.get("attachments") or None}, + identity="client_turn_id", + request={ + "message": str(message), + "origin": _opaque_id(origin, field="origin"), + "attachments": attachments or None, + }, + ) return existing, False session = self.load_session(session_id) if session is None or session.get("status") == "closed": @@ -652,6 +683,14 @@ def create_queued_turn( ): existing = self.turn_for_client(session_id, client_id) if existing is not None: + _require_matching_replay( + existing, + identity="client_turn_id", + request={ + "message": str(message), + "origin": _opaque_id(origin, field="origin"), + }, + ) return existing, False session = self.load_session(session_id) if session is None or session.get("status") == "closed": diff --git a/tests/test_chat_session_active_turn.py b/tests/test_chat_session_active_turn.py index 7d20d5d059..f10d343bf2 100644 --- a/tests/test_chat_session_active_turn.py +++ b/tests/test_chat_session_active_turn.py @@ -153,6 +153,99 @@ def test_concurrent_managed_turn_creation_claims_active_once( assert current["active_turn_id"] == active_turn_id +def test_chat_idempotency_keys_reject_different_requests(tmp_path: Path) -> None: + store = ChatSessionStore(tmp_path) + session = 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 = str(session["session_id"]) + store.create_turn(session_id, client_turn_id="turn-key", message="first") + store.create_queued_turn(session_id, client_turn_id="queue-key", message="first") + store.create_ingress_receipt( + session_id, + client_ingress_id="ingress-key", + mode="live_steering", + message="first", + ) + + with pytest.raises(ValueError, match="client_turn_id already belongs"): + store.create_turn(session_id, client_turn_id="turn-key", message="second") + with pytest.raises(ValueError, match="client_turn_id already belongs"): + store.create_queued_turn(session_id, client_turn_id="queue-key", message="second") + with pytest.raises(ValueError, match="client_ingress_id already belongs"): + store.create_ingress_receipt( + session_id, + client_ingress_id="ingress-key", + mode="live_steering", + message="second", + ) + + +@pytest.mark.parametrize("restart", [False, True]) +@pytest.mark.parametrize("with_image", [False, True]) +def test_managed_replay_compares_durable_attachments( + tmp_path: Path, restart: bool, with_image: bool, +) -> None: + store = ChatSessionStore(tmp_path) + session_id = 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"] + image = {"id": "image-one", "mime_type": "image/png", "name": "example.png"} + attachments = [image] if with_image else [] + original, created = store.create_turn( + session_id, client_turn_id="request", message="inspect", + attachments=attachments, + ) + assert created + assert "attachments" not in original # No duplicate image payload in Turn state. + if restart: + store = ChatSessionStore(tmp_path) + replay, created = store.create_turn( + session_id, client_turn_id="request", message="inspect", + attachments=attachments or None, + ) + assert not created + assert replay["turn_id"] == original["turn_id"] + for changed in ([{**image, "id": "different"}], [] if with_image else [image]): + with pytest.raises(ValueError, match="different request"): + store.create_turn( + session_id, client_turn_id="request", message="inspect", attachments=changed, + ) + with pytest.raises(ValueError, match="different request"): + store.create_turn( + session_id, client_turn_id="request", message="inspect", + attachments=attachments, origin="external", + ) + assert len(store.messages(session_id)) == 1 + + +def test_managed_replay_rejects_missing_original_message(tmp_path: Path, monkeypatch) -> None: + store = ChatSessionStore(tmp_path) + session_id = 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"] + # Model an interrupted creation after Turn persistence but before transcript append. + def interrupted_append(*args, **kwargs): + raise OSError("interrupted transcript write") + + monkeypatch.setattr(store, "append_message", interrupted_append) + with pytest.raises(OSError, match="interrupted transcript"): + store.create_turn(session_id, client_turn_id="request", message="inspect") + with pytest.raises(ValueError, match="original request is unavailable"): + ChatSessionStore(tmp_path).create_turn( + session_id, client_turn_id="request", message="inspect", + ) + + def test_completed_turn_cannot_release_a_newer_active_turn(tmp_path: Path) -> None: store = ChatSessionStore(tmp_path) session = store.create_session(