fix(chat): bind idempotency keys to requests

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

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(