From c558385cf3d88461bf3fc3764d47335d509b816c Mon Sep 17 00:00:00 2001 From: Matthew Wesney Date: Thu, 30 Jul 2026 14:23:28 -0400 Subject: [PATCH] Protect chat mutations with revision compare-and-swap --- backend/cortex_backend/api/routes.py | 34 ++- backend/cortex_backend/api/schemas.py | 2 + .../cortex_backend/repositories/__init__.py | 2 + backend/cortex_backend/repositories/chats.py | 240 +++++++++++------- .../repositories/legacy_storage.py | 72 ++++-- contracts/cortex-api.ts | 2 + contracts/openapi.json | 24 ++ tests/test_chat_revision_cas.py | 147 +++++++++++ 8 files changed, 401 insertions(+), 122 deletions(-) create mode 100644 tests/test_chat_revision_cas.py diff --git a/backend/cortex_backend/api/routes.py b/backend/cortex_backend/api/routes.py index 66a2710..f0f490b 100644 --- a/backend/cortex_backend/api/routes.py +++ b/backend/cortex_backend/api/routes.py @@ -27,7 +27,7 @@ title_from_first_message, ) from cortex_backend.core.settings import CortexSettings -from cortex_backend.repositories.chats import ChatRepositoryError +from cortex_backend.repositories.chats import ChatRepositoryError, ChatRevisionConflict from cortex_backend.repositories.settings import SettingsMigrationReport from cortex_backend.services.models import ModelPullProgress from cortex_backend.services.attachments import ( @@ -347,10 +347,13 @@ def add_message( thoughts=payload.thoughts, attachments=[attachment.model_dump(mode="json") for attachment in attachment_refs], thread_title="New Chat" if existing is None else None, + expected_revision=payload.base_revision, ) chat = deps.chats.get_chat(thread_id) except ChatAttachmentError as exc: _raise_chat_attachment_error(exc) + except ChatRevisionConflict as exc: + raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) from exc except ChatDomainError as exc: raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) from exc except Exception as exc: @@ -1133,11 +1136,19 @@ async def regenerate_chat( raise ChatDomainError( "A regeneration request needs user input." ) + current_revision = chat_revision(chat) + if ( + payload.base_revision is not None + and payload.base_revision != current_revision + ): + raise ChatDomainError( + "This chat changed. Reload it before regenerating." + ) generation_payload = GenerationRequest( request_id=payload.request_id, thread_id=thread_id, user_input=user_input, - base_revision=chat_revision(chat), + base_revision=current_revision, attachments=payload.attachments, ) snapshot, _ = await _start_generation_job( @@ -1164,6 +1175,8 @@ async def regenerate_chat( ) from exc except JobConflict as exc: raise HTTPException(status_code=409, detail=str(exc)) from exc + except ChatRevisionConflict as exc: + raise HTTPException(status_code=409, detail=str(exc)) from exc except ChatDomainError as exc: raise HTTPException(status_code=409, detail=str(exc)) from exc return _accepted(snapshot) @@ -1424,6 +1437,7 @@ async def _start_generation_job( try: chat = await asyncio.to_thread(deps.chats.get_chat, thread_id) current_revision = chat_revision(chat) if chat is not None else 0 + admission_revision = current_revision if ( payload.base_revision is not None and current_revision != payload.base_revision @@ -1475,18 +1489,16 @@ async def _start_generation_job( ) user_message_id: str | None = None + prepared_revision: int | None = None prepared_history = history_messages def prepare() -> Mapping[str, Any]: - nonlocal prepared_history, user_message_id + nonlocal prepared_history, prepared_revision, user_message_id current_chat = deps.chats.get_chat(thread_id) current_revision = ( chat_revision(current_chat) if current_chat is not None else 0 ) - if ( - payload.base_revision is not None - and current_revision != payload.base_revision - ): + if current_revision != admission_revision: raise ChatDomainError( "This chat changed. Reload it before generating again." ) @@ -1510,6 +1522,7 @@ def prepare() -> Mapping[str, Any]: "The selected response has no user turn to regenerate." ) prepared_history = current_messages[:target_position] + prepared_revision = current_revision else: user_message_id = deps.chats.add_message( thread_id, @@ -1520,7 +1533,12 @@ def prepare() -> Mapping[str, Any]: for attachment in attachment_refs ], thread_title="New Chat" if current_chat is None else None, + expected_revision=admission_revision, ) + updated_chat = deps.chats.get_chat(thread_id) + if updated_chat is None: + raise ChatRepositoryError("Chat did not persist the user message.") + prepared_revision = chat_revision(updated_chat) return {"user_message_id": user_message_id} def runner(sink, cancel_event): @@ -1568,6 +1586,7 @@ def runner(sink, cancel_event): "assistant", result.response, thoughts=result.thoughts, + expected_revision=prepared_revision, ) else: deps.chats.replace_message( @@ -1575,6 +1594,7 @@ def runner(sink, cancel_event): target_message_id, result.response, thoughts=result.thoughts, + expected_revision=prepared_revision, ) assistant_message_id = target_message_id diff --git a/backend/cortex_backend/api/schemas.py b/backend/cortex_backend/api/schemas.py index fca1538..b54a255 100644 --- a/backend/cortex_backend/api/schemas.py +++ b/backend/cortex_backend/api/schemas.py @@ -135,6 +135,7 @@ class RenameChatRequest(APIModel): class AddMessageRequest(APIModel): role: ChatRole content: str = Field(min_length=1, max_length=100_000) + base_revision: int | None = Field(default=None, ge=0) sources: list[Any] | None = None thoughts: str | None = Field(default=None, max_length=100_000) attachments: list[ChatAttachment] | None = Field(default=None, max_length=MAX_CHAT_ATTACHMENTS) @@ -460,6 +461,7 @@ class RegenerationRequest(APIModel): request_id: str | None = Field(default=None, min_length=1, max_length=200) message_id: str = Field(min_length=1, max_length=200) user_input: str | None = Field(default=None, max_length=100_000) + base_revision: int | None = Field(default=None, ge=0) attachments: list[ChatAttachment] = Field(default_factory=list, max_length=MAX_CHAT_ATTACHMENTS) diff --git a/backend/cortex_backend/repositories/__init__.py b/backend/cortex_backend/repositories/__init__.py index df70175..20e2b8b 100644 --- a/backend/cortex_backend/repositories/__init__.py +++ b/backend/cortex_backend/repositories/__init__.py @@ -3,6 +3,7 @@ from .chats import ( ChatRepository, ChatRepositoryError, + ChatRevisionConflict, InMemoryChatRepository, LegacyDatabaseChatRepository, ) @@ -24,6 +25,7 @@ __all__ = [ "ChatRepository", "ChatRepositoryError", + "ChatRevisionConflict", "InMemoryChatRepository", "LegacyDatabaseChatRepository", "MemoryRepository", diff --git a/backend/cortex_backend/repositories/chats.py b/backend/cortex_backend/repositories/chats.py index 590a2cd..f3c591e 100644 --- a/backend/cortex_backend/repositories/chats.py +++ b/backend/cortex_backend/repositories/chats.py @@ -4,6 +4,7 @@ from copy import deepcopy from datetime import datetime, timezone +from threading import RLock from typing import Any, Protocol @@ -11,6 +12,10 @@ class ChatRepositoryError(RuntimeError): """Safe failure raised by a chat repository.""" +class ChatRevisionConflict(ChatRepositoryError): + """The chat changed after the caller read its expected revision.""" + + class ChatRepository(Protocol): """Durable chat operations required by the versioned API.""" @@ -30,6 +35,7 @@ def add_message( thoughts: str | None = None, attachments: list[dict[str, Any]] | None = None, thread_title: str | None = None, + expected_revision: int | None = None, ) -> str: ... def rename_chat(self, thread_id: str, title: str) -> None: ... @@ -47,6 +53,7 @@ def replace_message( sources: list[Any] | None = None, thoughts: str | None = None, attachments: list[dict[str, Any]] | None = None, + expected_revision: int | None = None, ) -> None: ... @@ -68,7 +75,12 @@ def create_chat(self, thread_id: str, title: str) -> None: def add_message( self, thread_id: str, role: str, content: str, **kwargs: Any ) -> str: - result = self._database.add_message(thread_id, role, content, **kwargs) + try: + result = self._database.add_message(thread_id, role, content, **kwargs) + except Exception as exc: + if getattr(exc, "operation", None) == "chat_revision_conflict": + raise ChatRevisionConflict(str(exc)) from exc + raise if result is None: chat = self._database.load_chat(thread_id) or {} messages = chat.get("messages", []) @@ -108,21 +120,29 @@ def replace_message( sources: list[Any] | None = None, thoughts: str | None = None, attachments: list[dict[str, Any]] | None = None, + expected_revision: int | None = None, ) -> None: - self._database.replace_message( - thread_id, - int(message_id), - content, - sources=sources, - thoughts=thoughts, - attachments=attachments, - ) + try: + self._database.replace_message( + thread_id, + int(message_id), + content, + sources=sources, + thoughts=thoughts, + attachments=attachments, + expected_revision=expected_revision, + ) + except Exception as exc: + if getattr(exc, "operation", None) == "chat_revision_conflict": + raise ChatRevisionConflict(str(exc)) from exc + raise class InMemoryChatRepository: """Deterministic repository used by factory and API tests.""" def __init__(self, chats: list[dict[str, Any]] | None = None): + self._lock = RLock() self._chats: dict[str, dict[str, Any]] = {} self._next_message_id = 1 for chat in chats or []: @@ -153,28 +173,45 @@ def _timestamp() -> str: return datetime.now(timezone.utc).isoformat() def list_summaries(self) -> list[dict[str, Any]]: - return [ - {key: chat.get(key) for key in ("id", "title", "timestamp")} - for chat in sorted( - self._chats.values(), - key=lambda item: str(item.get("timestamp", "")), - reverse=True, - ) - ] + with self._lock: + return [ + {key: chat.get(key) for key in ("id", "title", "timestamp")} + for chat in sorted( + self._chats.values(), + key=lambda item: str(item.get("timestamp", "")), + reverse=True, + ) + ] def get_chat(self, thread_id: str) -> dict[str, Any] | None: - chat = self._chats.get(thread_id) - return deepcopy(chat) if chat is not None else None + with self._lock: + chat = self._chats.get(thread_id) + return deepcopy(chat) if chat is not None else None def create_chat(self, thread_id: str, title: str) -> None: - if thread_id in self._chats: - raise ChatRepositoryError("Chat already exists.") - self._chats[thread_id] = { - "id": thread_id, - "title": title, - "timestamp": self._timestamp(), - "messages": [], - } + with self._lock: + if thread_id in self._chats: + raise ChatRepositoryError("Chat already exists.") + self._chats[thread_id] = { + "id": thread_id, + "title": title, + "timestamp": self._timestamp(), + "messages": [], + } + + @staticmethod + def _check_expected_revision( + chat: dict[str, Any], expected_revision: int | None + ) -> None: + if expected_revision is None: + return + if type(expected_revision) is not int or expected_revision < 0: + raise ValueError("expected_revision must be a non-negative integer") + actual_revision = len(chat.get("messages", ())) + if actual_revision != expected_revision: + raise ChatRevisionConflict( + f"Chat revision changed (expected {expected_revision}, found {actual_revision})." + ) def add_message( self, @@ -186,63 +223,73 @@ def add_message( thoughts: str | None = None, attachments: list[dict[str, Any]] | None = None, thread_title: str | None = None, + expected_revision: int | None = None, ) -> str: - chat = self._chats.get(thread_id) - if chat is None: - if thread_title is None: - raise ChatRepositoryError("Chat does not exist.") - self.create_chat(thread_id, thread_title) - chat = self._chats[thread_id] - message_id = self._new_message_id() - chat["messages"].append( - { - "id": message_id, - "role": role, - "content": content, - "timestamp": self._timestamp(), - "sources": deepcopy(sources), - "thoughts": thoughts, - "attachments": deepcopy(attachments), - } - ) - chat["timestamp"] = self._timestamp() - return message_id + with self._lock: + chat = self._chats.get(thread_id) + if chat is None: + if expected_revision not in (None, 0): + raise ChatRevisionConflict( + f"Chat revision changed (expected {expected_revision}, found 0)." + ) + if thread_title is None: + raise ChatRepositoryError("Chat does not exist.") + self.create_chat(thread_id, thread_title) + chat = self._chats[thread_id] + self._check_expected_revision(chat, expected_revision) + message_id = self._new_message_id() + chat["messages"].append( + { + "id": message_id, + "role": role, + "content": content, + "timestamp": self._timestamp(), + "sources": deepcopy(sources), + "thoughts": thoughts, + "attachments": deepcopy(attachments), + } + ) + chat["timestamp"] = self._timestamp() + return message_id def rename_chat(self, thread_id: str, title: str) -> None: - chat = self._chats.get(thread_id) - if chat is None: - raise ChatRepositoryError("Chat does not exist.") - chat["title"] = title - chat["timestamp"] = self._timestamp() + with self._lock: + chat = self._chats.get(thread_id) + if chat is None: + raise ChatRepositoryError("Chat does not exist.") + chat["title"] = title + chat["timestamp"] = self._timestamp() def delete_chat(self, thread_id: str) -> None: - self._chats.pop(thread_id, None) + with self._lock: + self._chats.pop(thread_id, None) def fork_chat(self, thread_id: str, message_id: str, new_thread_id: str) -> None: - source = self._chats.get(thread_id) - if source is None: - raise ChatRepositoryError("Chat does not exist.") - try: - position = next( - index for index, item in enumerate(source["messages"]) - if str(item.get("id")) == str(message_id) - ) - except StopIteration as exc: - raise ChatRepositoryError("Message does not exist.") from exc - copied = deepcopy(source) - copied["id"] = new_thread_id - copied["title"] = f"Fork of {source.get('title') or 'Untitled Chat'}" - copied["timestamp"] = self._timestamp() - copied["messages"] = [] - for message in source["messages"][: position + 1]: - copied["messages"].append( - { - **deepcopy(message), - "id": self._new_message_id(), - "timestamp": self._timestamp(), - } - ) - self._chats[new_thread_id] = copied + with self._lock: + source = self._chats.get(thread_id) + if source is None: + raise ChatRepositoryError("Chat does not exist.") + try: + position = next( + index for index, item in enumerate(source["messages"]) + if str(item.get("id")) == str(message_id) + ) + except StopIteration as exc: + raise ChatRepositoryError("Message does not exist.") from exc + copied = deepcopy(source) + copied["id"] = new_thread_id + copied["title"] = f"Fork of {source.get('title') or 'Untitled Chat'}" + copied["timestamp"] = self._timestamp() + copied["messages"] = [] + for message in source["messages"][: position + 1]: + copied["messages"].append( + { + **deepcopy(message), + "id": self._new_message_id(), + "timestamp": self._timestamp(), + } + ) + self._chats[new_thread_id] = copied def replace_message( self, @@ -253,22 +300,25 @@ def replace_message( sources: list[Any] | None = None, thoughts: str | None = None, attachments: list[dict[str, Any]] | None = None, + expected_revision: int | None = None, ) -> None: - chat = self._chats.get(thread_id) - if chat is None: - raise ChatRepositoryError("Chat does not exist.") - for message in chat["messages"]: - if str(message.get("id")) == str(message_id): - if message.get("role") != "assistant": - raise ChatRepositoryError("Only assistant messages can be replaced.") - message.update( - content=content, - sources=deepcopy(sources), - thoughts=thoughts, - timestamp=self._timestamp(), - ) - if attachments is not None: - message["attachments"] = deepcopy(attachments) - chat["timestamp"] = self._timestamp() - return - raise ChatRepositoryError("Message does not exist.") + with self._lock: + chat = self._chats.get(thread_id) + if chat is None: + raise ChatRepositoryError("Chat does not exist.") + self._check_expected_revision(chat, expected_revision) + for message in chat["messages"]: + if str(message.get("id")) == str(message_id): + if message.get("role") != "assistant": + raise ChatRepositoryError("Only assistant messages can be replaced.") + message.update( + content=content, + sources=deepcopy(sources), + thoughts=thoughts, + timestamp=self._timestamp(), + ) + if attachments is not None: + message["attachments"] = deepcopy(attachments) + chat["timestamp"] = self._timestamp() + return + raise ChatRepositoryError("Message does not exist.") diff --git a/backend/cortex_backend/repositories/legacy_storage.py b/backend/cortex_backend/repositories/legacy_storage.py index daa6367..dddac3c 100644 --- a/backend/cortex_backend/repositories/legacy_storage.py +++ b/backend/cortex_backend/repositories/legacy_storage.py @@ -330,24 +330,27 @@ def create_chat_from_messages(self, thread_id: str, title: str, messages: list[d cause=exc, ) from exc - def add_message( - self, - thread_id: str, - role: str, - content: str, + def add_message( + self, + thread_id: str, + role: str, + content: str, sources: list | None = None, thoughts: str | None = None, attachments: list | None = None, thread_title: str | None = None, - ): - """Adds a new message to a specific chat thread.""" - try: - with self.connect() as conn: - if thread_title is not None: - conn.execute( - "INSERT OR IGNORE INTO threads (id, title, timestamp) VALUES (?, ?, ?)", - (thread_id, thread_title, _utc_now().isoformat()), - ) + expected_revision: int | None = None, + ): + """Adds a new message to a specific chat thread.""" + try: + with self.connect() as conn: + conn.execute("BEGIN IMMEDIATE") + if thread_title is not None: + conn.execute( + "INSERT OR IGNORE INTO threads (id, title, timestamp) VALUES (?, ?, ?)", + (thread_id, thread_title, _utc_now().isoformat()), + ) + self._check_chat_revision(conn, thread_id, expected_revision) conn.execute(""" INSERT INTO messages (thread_id, role, content, sources, thoughts, attachments, timestamp) VALUES (?, ?, ?, ?, ?, ?, ?) @@ -366,12 +369,36 @@ def add_message( (_utc_now().isoformat(), thread_id) ) return str(conn.execute("SELECT last_insert_rowid()").fetchone()[0]) - except PersistenceError as exc: - raise PersistenceError( - f"Failed to add message to thread {thread_id}.", - operation="add_message", - cause=exc, - ) from exc + except PersistenceError as exc: + if exc.operation == "chat_revision_conflict": + raise + raise PersistenceError( + f"Failed to add message to thread {thread_id}.", + operation="add_message", + cause=exc, + ) from exc + + @staticmethod + def _check_chat_revision( + conn: sqlite3.Connection, + thread_id: str, + expected_revision: int | None, + ) -> None: + if expected_revision is None: + return + if type(expected_revision) is not int or expected_revision < 0: + raise ValueError("expected_revision must be a non-negative integer") + actual_revision = int( + conn.execute( + "SELECT COUNT(*) FROM messages WHERE thread_id = ?", + (thread_id,), + ).fetchone()[0] + ) + if actual_revision != expected_revision: + raise PersistenceError( + f"Chat revision changed (expected {expected_revision}, found {actual_revision}).", + operation="chat_revision_conflict", + ) def load_chat(self, thread_id: str) -> dict | None: """Loads a full chat thread (metadata and messages) from the database.""" @@ -457,10 +484,13 @@ def replace_message( sources: list | None = None, thoughts: str | None = None, attachments: list | None = None, + expected_revision: int | None = None, ) -> None: """Replace one assistant response without disturbing its user turn.""" try: with self.connect() as conn: + conn.execute("BEGIN IMMEDIATE") + self._check_chat_revision(conn, thread_id, expected_revision) cursor = conn.execute( """ UPDATE messages @@ -487,6 +517,8 @@ def replace_message( (_utc_now().isoformat(), thread_id), ) except PersistenceError as exc: + if exc.operation == "chat_revision_conflict": + raise raise PersistenceError( f"Failed to replace message {message_id}.", operation="replace_message", diff --git a/contracts/cortex-api.ts b/contracts/cortex-api.ts index 93bb146..9c5e1de 100644 --- a/contracts/cortex-api.ts +++ b/contracts/cortex-api.ts @@ -8,6 +8,7 @@ export interface AddMemoryRequest { export interface AddMessageRequest { role: "user" | "assistant" | "system"; content: string; + base_revision?: number | null; sources?: Array | null; thoughts?: string | null; attachments?: Array | null; @@ -326,6 +327,7 @@ export interface RegenerationRequest { request_id?: string | null; message_id: string; user_input?: string | null; + base_revision?: number | null; attachments?: Array; } diff --git a/contracts/openapi.json b/contracts/openapi.json index 1512144..ce8c09b 100644 --- a/contracts/openapi.json +++ b/contracts/openapi.json @@ -35,6 +35,18 @@ ], "title": "Attachments" }, + "base_revision": { + "anyOf": [ + { + "minimum": 0.0, + "type": "integer" + }, + { + "type": "null" + } + ], + "title": "Base Revision" + }, "content": { "maxLength": 100000, "minLength": 1, @@ -1936,6 +1948,18 @@ "title": "Attachments", "type": "array" }, + "base_revision": { + "anyOf": [ + { + "minimum": 0.0, + "type": "integer" + }, + { + "type": "null" + } + ], + "title": "Base Revision" + }, "message_id": { "maxLength": 200, "minLength": 1, diff --git a/tests/test_chat_revision_cas.py b/tests/test_chat_revision_cas.py new file mode 100644 index 0000000..cfcb888 --- /dev/null +++ b/tests/test_chat_revision_cas.py @@ -0,0 +1,147 @@ +"""Compare-and-append chat revision boundaries.""" + +from __future__ import annotations + +from concurrent.futures import ThreadPoolExecutor +from threading import Barrier +import json +from pathlib import Path + +import pytest +from fastapi.testclient import TestClient + +from cortex_backend.api import build_demo_dependencies, create_app +from cortex_backend.repositories.chats import ( + ChatRevisionConflict, + InMemoryChatRepository, + LegacyDatabaseChatRepository, +) +from cortex_backend.repositories.legacy_storage import DatabaseManager +from cortex_backend.testing.fake_ollama import FakeOllamaState + + +def test_inmemory_append_is_compare_and_swap_and_atomic(): + repository = InMemoryChatRepository() + repository.create_chat("thread", "Thread") + start = Barrier(3) + + def append(content: str): + start.wait(timeout=2) + try: + return repository.add_message( + "thread", + "user", + content, + expected_revision=0, + ) + except ChatRevisionConflict: + return None + + with ThreadPoolExecutor(max_workers=2) as executor: + futures = [executor.submit(append, content) for content in ("one", "two")] + start.wait(timeout=2) + results = [future.result(timeout=2) for future in futures] + + assert sum(result is not None for result in results) == 1 + assert len(repository.get_chat("thread")["messages"]) == 1 + + +def test_sqlite_chat_revision_conflict_is_atomic(tmp_path: Path): + database = DatabaseManager(db_path=str(tmp_path / "chats.sqlite")) + repository = LegacyDatabaseChatRepository(database) + repository.create_chat("thread", "Thread") + repository.add_message("thread", "user", "one", expected_revision=0) + + with pytest.raises(ChatRevisionConflict): + repository.add_message("thread", "user", "stale", expected_revision=0) + + chat = repository.get_chat("thread") + assert [message["content"] for message in chat["messages"]] == ["one"] + + +def _session(client: TestClient, app) -> dict[str, str]: + token = client.post( + "/api/v1/session/exchange", + json={"bootstrap_token": app.state.session_manager.bootstrap_token}, + ).json()["session_token"] + return {"Authorization": f"Bearer {token}"} + + +def _events(body: str) -> list[dict]: + return [ + json.loads(line.removeprefix("data: ")) + for line in body.splitlines() + if line.startswith("data: ") + ] + + +def test_message_route_rejects_a_stale_base_revision(): + app = create_app(build_demo_dependencies(), allowed_hosts=("testserver",)) + with TestClient(app) as client: + headers = _session(client, app) + created = client.post( + "/api/v1/chats", json={"title": "Thread"}, headers=headers + ).json() + thread_id = created["id"] + assert client.post( + f"/api/v1/chats/{thread_id}/messages", + json={"role": "user", "content": "first"}, + headers=headers, + ).status_code == 200 + + stale = client.post( + f"/api/v1/chats/{thread_id}/messages", + json={"role": "user", "content": "stale", "base_revision": 0}, + headers=headers, + ) + + assert stale.status_code == 409 + assert "revision changed" in stale.json()["detail"].lower() + + +def test_generation_does_not_append_an_assistant_after_a_concurrent_chat_mutation(): + state = FakeOllamaState(generation_delay_seconds=0.2) + app = create_app( + build_demo_dependencies(ollama_state=state), allowed_hosts=("testserver",) + ) + with TestClient(app) as client: + headers = _session(client, app) + accepted = client.post( + "/api/v1/generations", + json={ + "request_id": "cas-generation", + "thread_id": "cas-thread", + "user_input": "first", + "base_revision": 0, + }, + headers=headers, + ) + assert accepted.status_code == 202 + accepted_payload = accepted.json() + + for _ in range(100): + status = client.get( + f"/api/v1/generations/{accepted_payload['job_id']}", headers=headers + ).json() + if status["status"] == "running": + break + else: + raise AssertionError("generation did not begin running") + + concurrent = client.post( + "/api/v1/chats/cas-thread/messages", + json={"role": "user", "content": "concurrent"}, + headers=headers, + ) + assert concurrent.status_code == 200 + + with client.stream( + "GET", + f"/api/v1/generations/{accepted_payload['job_id']}/events", + headers=headers, + ) as response: + events = _events("".join(response.iter_text())) + + assert events[-1]["event"] == "generation.failed" + chat = client.get("/api/v1/chats/cas-thread", headers=headers).json() + assert [message["role"] for message in chat["messages"]] == ["user", "user"]