"""Tests for SqlAlchemyConversationStore.""" from __future__ import annotations import pytest from sqlalchemy import text from omnigent.db.utils import get_or_create_engine from omnigent.entities import ( ErrorData, FunctionCallData, FunctionCallOutputData, MessageData, NewConversationItem, ReasoningData, ) from omnigent.server.auth import RESERVED_USER_LOCAL from omnigent.stores.agent_store.sqlalchemy_store import SqlAlchemyAgentStore from omnigent.stores.conversation_store.sqlalchemy_store import ( SqlAlchemyConversationStore, ) from omnigent.stores.host_store import HostStore # ── CRUD ────────────────────────────────────────────── def test_create_and_get(conversation_store: SqlAlchemyConversationStore) -> None: conv = conversation_store.create_conversation() assert conv.id.startswith("conv_") fetched = conversation_store.get_conversation(conv.id) assert fetched is not None assert fetched.id == conv.id def test_get_nonexistent(conversation_store: SqlAlchemyConversationStore) -> None: assert conversation_store.get_conversation("conv_none") is None def test_get_conversations_bulk( conversation_store: SqlAlchemyConversationStore, ) -> None: """ ``get_conversations`` returns one entry per resolvable id, omits unknown ids, and carries each row's batched labels — matching what a per-id ``get_conversation`` fan-out would produce, which is what the ``WS /v1/sessions/updates`` rescan relies on. """ a = conversation_store.create_conversation(title="alpha") b = conversation_store.create_conversation(title="beta") # Label only one row to prove labels are joined per-id, not smeared # across the batch or dropped for the unlabeled row. conversation_store.set_labels(a.id, {"omnigent.ui": "terminal"}) result = conversation_store.get_conversations([a.id, b.id, "conv_missing"]) # The unknown id is omitted rather than mapped to None — the caller # treats absence as "no longer resolves". assert set(result) == {a.id, b.id} # Titles prove the real rows came back, not placeholder shells. assert result[a.id].title == "alpha" assert result[b.id].title == "beta" # Labels are attached to the row they belong to and only that row. assert result[a.id].labels == {"omnigent.ui": "terminal"} assert result[b.id].labels == {} def test_get_conversations_empty_input_skips_query( conversation_store: SqlAlchemyConversationStore, ) -> None: """Empty id list returns an empty map without a database round-trip.""" assert conversation_store.get_conversations([]) == {} def test_list_latest_message_items_for_conversations( conversation_store: SqlAlchemyConversationStore, ) -> None: """ ``list_latest_message_items_for_conversations`` returns newest message rows per conversation in one batched call. The child-session summary route depends on this helper to avoid an event-loop-blocking N+1 ``list_items`` fan-out. This test seeds two conversations with interleaved item types, asks for two messages per conversation, and asserts exact per-conversation ordering and filtering. :param conversation_store: SQLAlchemy conversation store fixture. """ conv_a = conversation_store.create_conversation(title="alpha") conv_b = conversation_store.create_conversation(title="beta") conv_empty = conversation_store.create_conversation(title="empty") conversation_store.append( conv_a.id, [ NewConversationItem( type="message", response_id="resp_a", data=MessageData( role="assistant", content=[{"type": "output_text", "text": "alpha old"}], agent="worker", ), ), NewConversationItem( type="reasoning", response_id="resp_a", data=ReasoningData( agent="worker", summary=[{"type": "summary_text", "text": "thinking"}], ), ), NewConversationItem( type="message", response_id="resp_a", data=MessageData( role="assistant", content=[{"type": "output_text", "text": "alpha new"}], agent="worker", ), ), ], ) conversation_store.append( conv_b.id, [ NewConversationItem( type="message", response_id="resp_b", data=MessageData( role="assistant", content=[{"type": "output_text", "text": "bravo old"}], agent="worker", ), ), NewConversationItem( type="message", response_id="resp_b", data=MessageData( role="assistant", content=[{"type": "output_text", "text": "bravo middle"}], agent="worker", ), ), NewConversationItem( type="message", response_id="resp_b", data=MessageData( role="assistant", content=[{"type": "output_text", "text": "bravo new"}], agent="worker", ), ), ], ) result = conversation_store.list_latest_message_items_for_conversations( [conv_a.id, conv_b.id, conv_empty.id, "conv_missing"], per_conversation_limit=2, ) def _texts(conversation_id: str) -> list[str]: """Extract assistant text from the returned message items. :param conversation_id: Conversation id whose returned messages should be inspected. :returns: Ordered list of text blocks from that conversation's returned message items. """ texts: list[str] = [] for item in result[conversation_id]: assert isinstance(item.data, MessageData) texts.append(item.data.content[0]["text"]) return texts assert set(result) == {conv_a.id, conv_b.id, conv_empty.id, "conv_missing"} assert _texts(conv_a.id) == ["alpha new", "alpha old"] assert _texts(conv_b.id) == ["bravo new", "bravo middle"] assert result[conv_empty.id] == [] assert result["conv_missing"] == [] def test_update_title(conversation_store: SqlAlchemyConversationStore) -> None: conv = conversation_store.create_conversation() updated = conversation_store.update_conversation(conv.id, title="Chat 1") assert updated is not None assert updated.title == "Chat 1" assert conversation_store.update_conversation("conv_none", title="x") is None def test_update_archived_round_trip( conversation_store: SqlAlchemyConversationStore, ) -> None: """ ``update_conversation(archived=...)`` persists the flag both ways and a fresh conversation defaults to not-archived. Re-fetching via ``get_conversation`` (a separate read from the DB) proves the column was actually written, not just reflected on the in-session ORM object. A failure here means the archive column isn't persisted — the sidebar's archive button would appear to do nothing after a refresh. """ conv = conversation_store.create_conversation() # Default state: a brand-new session is not archived. Read it back # from the DB so we exercise the server_default / column mapping. assert conversation_store.get_conversation(conv.id).archived is False archived = conversation_store.update_conversation(conv.id, archived=True) assert archived is not None assert archived.archived is True # Persisted, not just on the returned object. assert conversation_store.get_conversation(conv.id).archived is True unarchived = conversation_store.update_conversation(conv.id, archived=False) assert unarchived is not None assert unarchived.archived is False assert conversation_store.get_conversation(conv.id).archived is False def test_update_archived_none_leaves_unchanged( conversation_store: SqlAlchemyConversationStore, ) -> None: """ ``archived=None`` (the default) must not touch the stored flag. The PATCH route passes ``archived=body.archived`` on every session update — including title-only edits where ``archived`` is ``None``. If ``None`` were treated as "set to false", renaming an archived session would silently unarchive it. This guards that. """ conv = conversation_store.create_conversation() conversation_store.update_conversation(conv.id, archived=True) # A title-only update (archived defaults to None) must leave the # archived flag set. conversation_store.update_conversation(conv.id, title="Renamed") refetched = conversation_store.get_conversation(conv.id) assert refetched.title == "Renamed" assert refetched.archived is True, ( "archived=None must leave the flag unchanged; a title-only edit " "unarchived the session, which would lose archive state on rename." ) def test_update_archived_bumps_updated_at( conversation_store: SqlAlchemyConversationStore, monkeypatch: pytest.MonkeyPatch, ) -> None: """ Toggling archived advances ``updated_at`` (like title/effort do). The clock is monkeypatched to a fixed, larger value so the bump is deterministic regardless of wall-clock resolution. The web client relies on this bump being acknowledged (``markConversationSeen``) so a self-initiated archive isn't mistaken for new activity — if the bump regressed to a no-op, that contract would silently change. """ conv = conversation_store.create_conversation() created_at = conversation_store.get_conversation(conv.id).updated_at # Pin the clock past created_at so the new updated_at is unambiguous. monkeypatch.setattr( "omnigent.stores.conversation_store.sqlalchemy_store.now_epoch", lambda: created_at + 100, ) updated = conversation_store.update_conversation(conv.id, archived=True) assert updated is not None assert updated.updated_at == created_at + 100, ( f"archiving should stamp updated_at via now_epoch(); expected " f"{created_at + 100}, got {updated.updated_at}. If it equals " f"{created_at}, the archive write didn't mark the row changed." ) # ── Append & list items ────────────────────────────── def test_append_and_list_items(conversation_store: SqlAlchemyConversationStore) -> None: conv = conversation_store.create_conversation() items = conversation_store.append( conv.id, [ NewConversationItem( type="message", response_id="resp_001", data=MessageData(role="user", content=[{"type": "input_text", "text": "Hello"}]), ), NewConversationItem( type="message", response_id="resp_001", data=MessageData( role="assistant", content=[{"type": "output_text", "text": "Hi there!"}], agent="test-agent", ), ), ], ) assert len(items) == 2 assert items[0].id.startswith("msg_") assert items[1].id.startswith("msg_") page = conversation_store.list_items(conv.id) assert len(page.data) == 2 assert page.data[0].data.role == "user" assert page.data[1].data.role == "assistant" def test_append_records_human_author_attribution( conversation_store: SqlAlchemyConversationStore, ) -> None: """A human-authored item round-trips its author identity. Analogue of the comment ``created_by`` contract: the actor who posted the message is recorded on the persisted item and read back on listing. """ conv = conversation_store.create_conversation() [persisted] = conversation_store.append( conv.id, [ NewConversationItem( type="message", response_id="resp_attr", data=MessageData( role="user", content=[{"type": "input_text", "text": "Please review"}], ), created_by="alice@example.com", ) ], ) assert persisted.created_by == "alice@example.com" [read_back] = conversation_store.list_items(conv.id).data assert read_back.created_by == "alice@example.com" def test_append_leaves_created_by_none_for_agent_items( conversation_store: SqlAlchemyConversationStore, ) -> None: """Items appended without an actor (agent/tool/system) read back None. Keeps agent output distinguishable from human-authored messages. """ conv = conversation_store.create_conversation() [persisted] = conversation_store.append( conv.id, [ NewConversationItem( type="message", response_id="resp_agent", data=MessageData( role="assistant", agent="my-agent", content=[{"type": "output_text", "text": "Done"}], ), ) ], ) assert persisted.created_by is None [read_back] = conversation_store.list_items(conv.id).data assert read_back.created_by is None def test_append_function_call_items( conversation_store: SqlAlchemyConversationStore, ) -> None: conv = conversation_store.create_conversation() items = conversation_store.append( conv.id, [ NewConversationItem( type="function_call", response_id="resp_002", data=FunctionCallData( agent="test-agent", name="get_weather", arguments='{"city": "SF"}', call_id="call_001", ), ), NewConversationItem( type="function_call_output", response_id="resp_002", data=FunctionCallOutputData( call_id="call_001", output='{"temp": 65}', ), ), ], ) assert items[0].id.startswith("fc_") assert items[1].id.startswith("fco_") def test_append_tool_output_with_nul_bytes( conversation_store: SqlAlchemyConversationStore, db_uri: str, ) -> None: """ Tool output containing NUL (0x00) bytes must still persist. Reproduces the production failure where a tool returned bytes from a binary file: the embedded NUL hit a Postgres text column and aborted the whole INSERT (``psycopg.DataError: PostgreSQL text fields cannot contain NUL (0x00) bytes``), so the function call output never persisted. SQLite tolerates NUL, so the deterministic guard below is reading the persisted columns and asserting no raw NUL survived — that fails if the store stops stripping NUL. """ conv = conversation_store.create_conversation() # "marker_unique_token" is space-delimited from the NUL run so we # can later prove it remained FTS-indexable after sanitization. nul_output = "binary chunk \x00\x00\x00\x00 marker_unique_token" items = conversation_store.append( conv.id, [ NewConversationItem( type="function_call_output", response_id="resp_nul", data=FunctionCallOutputData( call_id="call_nul", output=nul_output, ), ), ], ) # append() returning normally (not raising DataError) is the # primary reproduction: pre-fix this INSERT aborted on Postgres. item_id = items[0].id assert item_id.startswith("fco_") # The payload round-trips faithfully: NUL is preserved in the data # column (json.dumps escapes it to the literal 6-char # backslash-u0000 sequence, which Postgres accepts and json.loads # decodes back). Stripping is lossy for the FTS index only, never # for the stored output. page = conversation_store.list_items(conv.id) assert page.data[0].data.output == nul_output # Deterministic backend-agnostic guard: inspect the persisted # columns directly. search_text must have had its raw NUL removed # (this is the line that fails if append() stops calling # strip_nul_bytes); data must carry no raw NUL either (json # escaping keeps it as the literal backslash-u0000 sequence). engine = get_or_create_engine(db_uri) with engine.connect() as conn: row = conn.execute( text("SELECT data, search_text FROM conversation_items WHERE id = :id"), {"id": item_id}, ).one() assert "\x00" not in row.search_text assert "\x00" not in row.data # The meaningful, NUL-adjacent token is still searchable, proving # sanitization dropped only the NUL and left real tokens indexed. results = conversation_store.search("marker_unique_token", conversation_id=conv.id) assert [r.id for r in results] == [item_id] def test_append_reasoning_item(conversation_store: SqlAlchemyConversationStore) -> None: conv = conversation_store.create_conversation() items = conversation_store.append( conv.id, [ NewConversationItem( type="reasoning", response_id="resp_003", data=ReasoningData( agent="test-agent", summary=[{"type": "summary_text", "text": "Thinking..."}], ), ), ], ) assert items[0].id.startswith("rs_") def test_append_error_item_round_trips_for_history( conversation_store: SqlAlchemyConversationStore, ) -> None: """ Persisted ``error`` items survive the real SQLAlchemy store path and flatten to the same shape ``GET /sessions/{id}/items`` returns. A regression here reproduces the web symptom where a live ``response.error`` banner appears but disappears after refresh: either append/list rejects the new item type, or the API shape no longer contains the fields ``itemsToBlocks`` needs to rebuild the banner. :param conversation_store: SQLAlchemy conversation store fixture. """ conv = conversation_store.create_conversation() [persisted] = conversation_store.append( conv.id, [ NewConversationItem( type="error", response_id="resp_failed", data=ErrorData( source="execution", code="native_terminal_start_failed", message="Native Codex requires the 'codex' CLI on PATH.", ), ), ], ) assert persisted.id.startswith("err_") [read_back] = conversation_store.list_items(conv.id).data assert read_back.id == persisted.id assert read_back.response_id == "resp_failed" assert isinstance(read_back.data, ErrorData) assert read_back.data.message == "Native Codex requires the 'codex' CLI on PATH." assert read_back.to_api_dict() == { "id": persisted.id, "response_id": "resp_failed", "type": "error", "status": "completed", "source": "execution", "code": "native_terminal_start_failed", "message": "Native Codex requires the 'codex' CLI on PATH.", } # ── Ordering & cursors ─────────────────────────────── def test_position_ordering(conversation_store: SqlAlchemyConversationStore) -> None: conv = conversation_store.create_conversation() conversation_store.append( conv.id, [ NewConversationItem( type="message", response_id="resp_a", data=MessageData(role="user", content=[{"type": "input_text", "text": "First"}]), ), ], ) conversation_store.append( conv.id, [ NewConversationItem( type="message", response_id="resp_b", data=MessageData(role="user", content=[{"type": "input_text", "text": "Second"}]), ), ], ) page = conversation_store.list_items(conv.id) assert len(page.data) == 2 texts = [page.data[i].data.content[0]["text"] for i in range(2)] assert texts == ["First", "Second"] def test_unique_position_constraint( conversation_store: SqlAlchemyConversationStore, ) -> None: """ The (conversation_id, position) pair has a unique index. Verify that manually inserting a duplicate position raises IntegrityError, confirming the safety net is in place. """ from sqlalchemy.exc import IntegrityError from omnigent.db.db_models import SqlConversationItem from omnigent.db.enum_codecs import encode_item_status, encode_item_type from omnigent.db.utils import generate_item_id conv = conversation_store.create_conversation() conversation_store.append( conv.id, [ NewConversationItem( type="message", response_id="resp_dup", data=MessageData( role="user", content=[{"type": "input_text", "text": "first"}], ), ), ], ) # Directly insert a row at position 0 (already taken) to # confirm the unique constraint rejects it. with pytest.raises(IntegrityError): with conversation_store._session() as session: session.add( SqlConversationItem( id=generate_item_id("message"), conversation_id=conv.id, response_id="resp_dup", created_at=0, status=encode_item_status("completed"), position=0, # duplicate type=encode_item_type("message"), data='{"role":"user","content":[]}', search_text="", ) ) def test_concurrent_appends_do_not_collide_on_position( conversation_store: SqlAlchemyConversationStore, ) -> None: """ Two concurrent ``append()`` calls on the same conversation must not collide on position assignment. Reproduces the UNIQUE-constraint race observed 2026-04-30 in a live REPL session with the user's 20-shell scenario: the agent loop's incremental tool-call persist and the steering inbox's auto-injection of idle-notification user messages both ran ``append()`` for the same conversation_id. Both transactions read ``max(position) = N``, both INSERTed at position ``N+1``, the loser crashed with ``IntegrityError: UNIQUE constraint failed: conversation_items.conversation_id, conversation_items.position``. The race window came from SQLite starting transactions as DEFERRED (no write lock until first DML), so the ``select(max(position))`` in :meth:`append` ran without holding any lock. Fixed by upgrading :meth:`_lock_conversation` from a no-op on SQLite to an UPDATE that escalates the transaction to RESERVED. What this test proves and what a failure means: - Running N appends concurrently on the same conversation produces N items with N distinct positions, no ``IntegrityError``. If a thread raises, the transaction-escalation fix regressed. - Final positions are contiguous from 0 to ``items_per_thread * threads - 1``. Gaps would mean a write succeeded but its position was reused — a worse failure mode than the IntegrityError. Does NOT use mocks: real :class:`SqlAlchemyConversationStore`, real SQLite (the conftest's per-test temp DB), real threads so SQLite's actual lock semantics participate. """ import threading conv = conversation_store.create_conversation() threads_count = 8 items_per_thread = 5 errors: list[Exception] = [] errors_lock = threading.Lock() def _append_n(thread_idx: int) -> None: try: for i in range(items_per_thread): conversation_store.append( conv.id, [ NewConversationItem( type="message", response_id=f"resp_t{thread_idx}", data=MessageData( role="user", content=[ { "type": "input_text", "text": f"t{thread_idx}-{i}", } ], ), ), ], ) except Exception as exc: with errors_lock: errors.append(exc) threads = [threading.Thread(target=_append_n, args=(i,)) for i in range(threads_count)] for t in threads: t.start() for t in threads: t.join(timeout=30.0) # No thread raised — the lock escalation prevented every # IntegrityError. If non-empty, the race regressed and the # user-reported "UNIQUE constraint failed" symptom is back. assert errors == [], ( f"Concurrent appends raised {len(errors)} error(s); the " f"first was: {errors[0]!r}. The position-race fix in " f"_lock_conversation regressed — SQLite transactions are " f"running concurrent SELECT max(position) without a " f"write lock again." ) # All items persisted with distinct IDs. Position # uniqueness is enforced by the # ``ix_conversation_items_conversation_id_position`` UNIQUE # index on the SQL table, so any race that produced # duplicate positions would have raised IntegrityError — # caught above. The remaining check here is that no append # silently dropped: every (thread, item) pair surfaced as # exactly one ConversationItem in the public listing. items = conversation_store.list_items(conv.id, limit=1000).data expected_count = threads_count * items_per_thread assert len(items) == expected_count, ( f"expected {expected_count} items; got {len(items)}. " f"Some appends silently dropped despite not raising — " f"that would be a worse regression than the original " f"IntegrityError." ) item_ids = {item.id for item in items} assert len(item_ids) == expected_count, ( f"expected {expected_count} distinct item IDs; got " f"{len(item_ids)}. Duplicate IDs would mean an item was " f"persisted twice or list_items returned duplicates." ) # Positions are contiguous 0..N-1. The UNIQUE index catches *reused* # positions (IntegrityError, asserted above), but a counter that # over-advances — or an append silently skipped — leaves a *gap* with no # error. list_items hides position, so assert on the raw column directly. assert _stored_positions(conversation_store, conv.id) == list(range(expected_count)), ( "concurrent appends must allocate a gap-free 0..N-1 position sequence" ) def test_heavy_batch_racing_steering_append_does_not_collide( conversation_store: SqlAlchemyConversationStore, ) -> None: """ Models the exact user-reported race shape from 2026-04-30: one path appends a heavy batch (assistant message + function_call + function_call_output items, mirroring the workflow's incremental tool-call persist via ``_handle_executor_event``), while a second path concurrently appends a single user message (mirroring the steering inbox's auto-injection of an idle notification). The unit test ``test_concurrent_appends_do_not_collide_on_position`` proves the SQL-level race exists with N homogeneous threads. This test pins the SHAPE of the user's actual failure: a multi-item append (which holds the reserve longer because it inserts more rows) racing a single-item append (the steering write). If a regression introduces a DIFFERENT race window that only surfaces under this asymmetric pattern, this test catches it. On revert: the heavy-batch thread holds RESERVED while inserting message + fc + fco, the steering thread blocks on busy_timeout, then re-reads max_pos. With the fix, no IntegrityError. Without the fix, the steering thread reads max_pos before the heavy batch commits and one of them crashes. """ import threading conv = conversation_store.create_conversation() barrier = threading.Barrier(2, timeout=10.0) errors: list[tuple[str, Exception]] = [] errors_lock = threading.Lock() def _heavy_batch() -> None: """Mimics ``_handle_executor_event``: 1 message + 1 fc + 1 fco per call.""" try: barrier.wait() for i in range(20): conversation_store.append( conv.id, [ NewConversationItem( type="message", response_id="resp_workflow", data=MessageData( role="assistant", content=[ { "type": "output_text", "text": f"calling tool {i}", } ], agent="agent_test", ), ), NewConversationItem( type="function_call", response_id="resp_workflow", data=FunctionCallData( name="sys_terminal_launch", arguments=f'{{"session": "sh{i}"}}', call_id=f"call_{i}", agent="agent_test", ), ), NewConversationItem( type="function_call_output", response_id="resp_workflow", data=FunctionCallOutputData( call_id=f"call_{i}", output='{"status": "launched"}', ), ), ], ) except Exception as exc: with errors_lock: errors.append(("heavy_batch", exc)) def _steering_appends() -> None: """Mimics the steering inbox auto-injecting idle notifications.""" try: barrier.wait() for i in range(20): conversation_store.append( conv.id, [ NewConversationItem( type="message", response_id="resp_steering", data=MessageData( role="user", content=[ { "type": "input_text", "text": (f"[System: terminal sh{i} is idle]"), } ], ), ), ], ) except Exception as exc: with errors_lock: errors.append(("steering", exc)) heavy = threading.Thread(target=_heavy_batch) steering = threading.Thread(target=_steering_appends) heavy.start() steering.start() heavy.join(timeout=30.0) steering.join(timeout=30.0) # Neither thread crashed — the lock escalation serialized # both. Without the fix, one of them raises # ``IntegrityError: UNIQUE constraint failed: # conversation_items.conversation_id, conversation_items.position``. assert errors == [], ( f"Heavy-batch + steering append race produced {len(errors)} " f"error(s). First: {errors[0]!r}. The user-reported " f"2026-04-30 IntegrityError is back." ) # Total = 20 batches × 3 items + 20 steering items = 80. # Position uniqueness is enforced by the SQL UNIQUE index # on (conversation_id, position) — any race that produced a # duplicate would have raised above. So count + ID # distinctness is sufficient evidence that the lock # escalation worked. items = conversation_store.list_items(conv.id, limit=1000).data assert len(items) == 80, f"expected 80 items; got {len(items)}" item_ids = {item.id for item in items} assert len(item_ids) == 80, ( f"expected 80 distinct item IDs; got {len(item_ids)}. " f"Duplicate IDs would mean an item was persisted twice." ) # Gap-free 0..79 across both threads: the heavy batch advances the # next_position counter by 3 and the steering append by 1, so a counter # that mis-advances under this asymmetric race would skip or reuse a # position without tripping the UNIQUE index. Assert the raw column. assert _stored_positions(conversation_store, conv.id) == list(range(80)), ( "heavy-batch + steering appends must allocate a gap-free 0..79 sequence" ) def _make_5_items(conversation_store: SqlAlchemyConversationStore, conv_id: str): """Helper: append 5 messages and return the persisted items.""" return conversation_store.append( conv_id, [ NewConversationItem( type="message", response_id="resp_x", data=MessageData( role="user", content=[{"type": "input_text", "text": f"msg-{i}"}], ), ) for i in range(5) ], ) def test_list_items_after_cursor( conversation_store: SqlAlchemyConversationStore, ) -> None: conv = conversation_store.create_conversation() items = _make_5_items(conversation_store, conv.id) page = conversation_store.list_items(conv.id, after=items[1].id, limit=2) assert len(page.data) == 2 assert page.data[0].id == items[2].id def test_list_items_desc_order( conversation_store: SqlAlchemyConversationStore, ) -> None: conv = conversation_store.create_conversation() _make_5_items(conversation_store, conv.id) page_asc = conversation_store.list_items(conv.id, order="asc") page_desc = conversation_store.list_items(conv.id, order="desc") assert [it.id for it in page_asc.data] == list(reversed([it.id for it in page_desc.data])) def test_list_items_desc_with_after_cursor( conversation_store: SqlAlchemyConversationStore, ) -> None: """In desc order, 'after' means items with lower position.""" conv = conversation_store.create_conversation() items = _make_5_items(conversation_store, conv.id) # desc full page: [4, 3, 2, 1, 0] page1 = conversation_store.list_items(conv.id, limit=2, order="desc") assert page1.data[0].id == items[4].id assert page1.data[1].id == items[3].id assert page1.has_more is True page2 = conversation_store.list_items(conv.id, limit=2, order="desc", after=page1.last_id) assert page2.data[0].id == items[2].id assert page2.data[1].id == items[1].id def test_list_items_before_cursor( conversation_store: SqlAlchemyConversationStore, ) -> None: conv = conversation_store.create_conversation() items = _make_5_items(conversation_store, conv.id) # asc order: [0, 1, 2, 3, 4]; before item[3] should give [0, 1, 2] page = conversation_store.list_items(conv.id, before=items[3].id, order="asc") assert [it.id for it in page.data] == [items[i].id for i in range(3)] # ── Conversation ID / response ID lookups ──────────── def test_search(conversation_store: SqlAlchemyConversationStore) -> None: conv = conversation_store.create_conversation() conversation_store.append( conv.id, [ NewConversationItem( type="message", response_id="resp_s1", data=MessageData( role="user", content=[{"type": "input_text", "text": "weather in Paris"}], ), ), NewConversationItem( type="message", response_id="resp_s1", data=MessageData( role="assistant", content=[{"type": "output_text", "text": "sunny and warm"}], agent="test-agent", ), ), ], ) results = conversation_store.search("Paris") assert len(results) == 1 assert results[0].type == "message" results = conversation_store.search("sunny") assert len(results) == 1 assert conversation_store.search("nonexistent") == [] def test_search_scoped_to_conversation( conversation_store: SqlAlchemyConversationStore, ) -> None: conv1 = conversation_store.create_conversation() conv2 = conversation_store.create_conversation() conversation_store.append( conv1.id, [ NewConversationItem( type="message", response_id="r1", data=MessageData( role="user", content=[{"type": "input_text", "text": "hello world"}], ), ), ], ) conversation_store.append( conv2.id, [ NewConversationItem( type="message", response_id="r2", data=MessageData( role="user", content=[{"type": "input_text", "text": "hello universe"}], ), ), ], ) # Unscoped: both match "hello" assert len(conversation_store.search("hello")) == 2 # Scoped: only one per conversation assert len(conversation_store.search("hello", conversation_id=conv1.id)) == 1 assert len(conversation_store.search("hello", conversation_id=conv2.id)) == 1 def test_search_function_call_item( conversation_store: SqlAlchemyConversationStore, ) -> None: """FTS indexes function_call items by name and arguments.""" conv = conversation_store.create_conversation() conversation_store.append( conv.id, [ NewConversationItem( type="function_call", response_id="resp_fc", data=FunctionCallData( agent="test-agent", name="get_weather", arguments='{"city": "Tokyo"}', call_id="call_1", ), ), ], ) assert len(conversation_store.search("get_weather")) == 1 assert len(conversation_store.search("Tokyo")) == 1 assert conversation_store.search("nonexistent") == [] def test_list_conversations_search_query_matches_content( conversation_store: SqlAlchemyConversationStore, ) -> None: """ ``list_conversations(search_query=...)`` matches conversations whose title OR item content contains the query substring. :param conversation_store: The conversation store fixture. """ conv_title = conversation_store.create_conversation() conversation_store.update_conversation(conv_title.id, title="deployment runbook") conv_content = conversation_store.create_conversation() conversation_store.update_conversation(conv_content.id, title="General chat") conversation_store.append( conv_content.id, [ NewConversationItem( type="message", response_id="resp_gc1", data=MessageData( role="user", content=[{"type": "input_text", "text": "fix the deployment pipeline"}], ), ), ], ) conv_neither = conversation_store.create_conversation() conversation_store.update_conversation(conv_neither.id, title="Unrelated session") page = conversation_store.list_conversations(search_query="deployment") matched_ids = {c.id for c in page.data} assert conv_title.id in matched_ids, "title match should be included" assert conv_content.id in matched_ids, "content match should be included" assert conv_neither.id not in matched_ids, "non-matching conversation excluded" def test_list_conversations_search_query_content_only( conversation_store: SqlAlchemyConversationStore, ) -> None: """ A conversation with no title but matching item content is returned by ``search_query``. :param conversation_store: The conversation store fixture. """ conv = conversation_store.create_conversation() conversation_store.append( conv.id, [ NewConversationItem( type="message", response_id="resp_co1", data=MessageData( role="assistant", content=[{"type": "output_text", "text": "the server is healthy"}], agent="test-agent", ), ), ], ) page = conversation_store.list_conversations(search_query="healthy") assert len(page.data) == 1 assert page.data[0].id == conv.id def test_list_conversations_search_snippet_on_content_match( conversation_store: SqlAlchemyConversationStore, ) -> None: """ A content match carries a ``search_snippet`` excerpt of the matching text; a title-only match leaves it ``None``. :param conversation_store: The conversation store fixture. """ conv_content = conversation_store.create_conversation() conversation_store.update_conversation(conv_content.id, title="General chat") conversation_store.append( conv_content.id, [ NewConversationItem( type="message", response_id="resp_snip1", data=MessageData( role="user", content=[{"type": "input_text", "text": "please fix the deployment pipeline"}], ), ), ], ) conv_title = conversation_store.create_conversation() conversation_store.update_conversation(conv_title.id, title="deployment runbook") by_id = { c.id: c for c in conversation_store.list_conversations(search_query="deployment").data } # Content match: snippet present and contains the query term. assert by_id[conv_content.id].search_snippet is not None assert "deployment" in by_id[conv_content.id].search_snippet.lower() # Title-only match: no snippet (the title already shows the hit). assert by_id[conv_title.id].search_snippet is None def test_list_conversations_search_snippet_absent_without_query( conversation_store: SqlAlchemyConversationStore, ) -> None: """ Non-search listings never populate ``search_snippet``. :param conversation_store: The conversation store fixture. """ conv = conversation_store.create_conversation() conversation_store.append( conv.id, [ NewConversationItem( type="message", response_id="resp_snip2", data=MessageData( role="user", content=[{"type": "input_text", "text": "hello world"}], ), ), ], ) page = conversation_store.list_conversations() assert all(c.search_snippet is None for c in page.data) def test_list_conversations_search_snippet_uses_earliest_match( conversation_store: SqlAlchemyConversationStore, ) -> None: """ With multiple matching turns, the snippet comes from the earliest one. Exercises the ``MIN(position)`` join path: two turns match the query; the snippet must be built from the first turn's text, not a later one. :param conversation_store: The conversation store fixture. """ conv = conversation_store.create_conversation() conversation_store.append( conv.id, [ NewConversationItem( type="message", response_id="resp_early", data=MessageData( role="user", content=[{"type": "input_text", "text": "deployment first mention"}], ), ), NewConversationItem( type="message", response_id="resp_late", data=MessageData( role="assistant", content=[{"type": "output_text", "text": "deployment second mention"}], agent="test-agent", ), ), ], ) by_id = { c.id: c for c in conversation_store.list_conversations(search_query="deployment").data } snippet = by_id[conv.id].search_snippet assert snippet is not None assert "first mention" in snippet assert "second mention" not in snippet def test_list_conversations_excludes_archived_by_default( conversation_store: SqlAlchemyConversationStore, ) -> None: """ ``list_conversations`` hides archived rows unless ``include_archived=True``. Two conversations, one archived: the default listing returns only the active one; ``include_archived=True`` returns both. This is the exact contract the sidebar's default view and "Show archived" toggle depend on. A failure means archived sessions either leak into the default sidebar (filter not applied) or can never be surfaced (toggle has no effect). """ active = conversation_store.create_conversation() archived = conversation_store.create_conversation() conversation_store.update_conversation(archived.id, archived=True) default_ids = {c.id for c in conversation_store.list_conversations().data} assert active.id in default_ids, "active session must appear in the default listing" assert archived.id not in default_ids, "archived session must be hidden by default" all_ids = {c.id for c in conversation_store.list_conversations(include_archived=True).data} assert all_ids >= {active.id, archived.id}, ( f"include_archived=True must return both active and archived sessions; got {all_ids}" ) # ── Delete ─────────────────────────────────────────── @pytest.mark.asyncio async def test_delete_conversation( conversation_store: SqlAlchemyConversationStore, ) -> None: conv = conversation_store.create_conversation() conversation_store.append( conv.id, [ NewConversationItem( type="message", response_id="resp_del", data=MessageData(role="user", content=[{"type": "input_text", "text": "bye"}]), ), ], ) assert await conversation_store.delete_conversation(conv.id) is True assert conversation_store.get_conversation(conv.id) is None assert conversation_store.list_items(conv.id).data == [] assert await conversation_store.delete_conversation(conv.id) is False @pytest.mark.asyncio async def test_delete_conversation_with_items( conversation_store: SqlAlchemyConversationStore, ) -> None: """Deleting a conversation with items removes the conversation and all its items. :param conversation_store: The conversation store fixture. """ conv = conversation_store.create_conversation() conversation_store.append( conv.id, [ NewConversationItem( type="message", response_id="resp_x", data=MessageData( role="user", content=[{"type": "input_text", "text": "hi"}], ), ), ], ) assert await conversation_store.delete_conversation(conv.id) is True assert conversation_store.get_conversation(conv.id) is None # ── List conversations pagination ──────────────────── def test_list_conversations_pagination( conversation_store: SqlAlchemyConversationStore, ) -> None: for _ in range(4): conversation_store.create_conversation() page1 = conversation_store.list_conversations(limit=2) assert len(page1.data) == 2 assert page1.has_more is True page2 = conversation_store.list_conversations(limit=2, after=page1.last_id) assert len(page2.data) == 2 assert page2.has_more is False def test_list_conversations_order_asc( conversation_store: SqlAlchemyConversationStore, ) -> None: for _ in range(3): conversation_store.create_conversation() page_desc = conversation_store.list_conversations(order="desc") page_asc = conversation_store.list_conversations(order="asc") assert [c.id for c in page_asc.data] == list(reversed([c.id for c in page_desc.data])) def test_list_conversations_asc_with_after_cursor( conversation_store: SqlAlchemyConversationStore, ) -> None: for _ in range(5): conversation_store.create_conversation() page1 = conversation_store.list_conversations(limit=2, order="asc") page2 = conversation_store.list_conversations(limit=2, order="asc", after=page1.last_id) page3 = conversation_store.list_conversations(limit=2, order="asc", after=page2.last_id) all_ids = [c.id for c in page1.data + page2.data + page3.data] full_asc = conversation_store.list_conversations(limit=100, order="asc") assert all_ids == [c.id for c in full_asc.data] # ── list_items type filter ──────────────────────────── def test_list_items_type_filter_returns_only_matching_type( conversation_store: SqlAlchemyConversationStore, ) -> None: """ list_items(type=...) returns only items of the specified type, while list_items() without a filter returns all types. """ from omnigent.entities import CompactionData conv = conversation_store.create_conversation() # Append a mix of message and compaction items conversation_store.append( conv.id, [ NewConversationItem( type="message", response_id="resp_001", data=MessageData(role="user", content=[{"type": "input_text", "text": "hi"}]), ), ], ) conversation_store.append( conv.id, [ NewConversationItem( type="compaction", response_id="resp_001", data=CompactionData( summary="Summary text", last_item_id="msg_001", model="openai/gpt-4o", token_count=50, ), ), ], ) conversation_store.append( conv.id, [ NewConversationItem( type="message", response_id="resp_002", data=MessageData( role="assistant", content=[{"type": "output_text", "text": "hello"}], agent="test-agent", ), ), ], ) compaction_items = conversation_store.list_items(conv.id, type="compaction") message_items = conversation_store.list_items(conv.id, type="message") all_items = conversation_store.list_items(conv.id) # Only the one compaction item must be returned. assert len(compaction_items.data) == 1, ( f"Expected 1 compaction item, got {len(compaction_items.data)}. " "Failure means type filter did not exclude message items." ) assert compaction_items.data[0].type == "compaction" # Only message items (2) must be returned. assert len(message_items.data) == 2, ( f"Expected 2 message items, got {len(message_items.data)}. " "Failure means type filter did not exclude the compaction item." ) assert all(i.type == "message" for i in message_items.data) # No filter returns all 3 items. assert len(all_items.data) == 3, ( f"Expected 3 total items (2 message + 1 compaction), got {len(all_items.data)}." ) def test_list_items_type_filter_with_order_and_limit( conversation_store: SqlAlchemyConversationStore, ) -> None: """ list_items(type="compaction", order="desc", limit=1) returns only the most recently appended compaction item. """ from omnigent.entities import CompactionData conv = conversation_store.create_conversation() # Append two compaction items conversation_store.append( conv.id, [ NewConversationItem( type="compaction", response_id="resp_001", data=CompactionData( summary="First summary", last_item_id="msg_010", model="openai/gpt-4o", token_count=100, ), ), ], ) conversation_store.append( conv.id, [ NewConversationItem( type="compaction", response_id="resp_002", data=CompactionData( summary="Second summary", last_item_id="msg_020", model="openai/gpt-4o", token_count=120, ), ), ], ) result = conversation_store.list_items(conv.id, type="compaction", order="desc", limit=1) # Only one item returned (limit=1). assert len(result.data) == 1, f"Expected 1 item with limit=1, got {len(result.data)}." # The most recent compaction item (second) should be returned (order=desc). assert result.data[0].data.summary == "Second summary", ( f"Expected the latest compaction item with 'Second summary', " f"got: {result.data[0].data.summary!r}. " "Failure means order=desc with limit=1 did not return the newest item." ) # ── Sub-agent conversation isolation ──────────────── def test_subagent_conversations_are_isolated( conversation_store: SqlAlchemyConversationStore, ) -> None: """ Two sub-agent conversations created independently must have fully isolated item sets. list_items on one must never return items belonging to the other. This is the foundational invariant that prevents sub-agent "pollution": each sub-agent writes to its own conversation, and the agent loop loads history via ``list_items(conversation_id)`` — so items from sibling sub-agents are structurally invisible. A failure here means the WHERE clause on ``conversation_id`` in ``list_items`` is broken, which would cause sub-agents to see each other's messages and produce incoherent LLM prompts. """ conv_a = conversation_store.create_conversation(kind="sub_agent") conv_b = conversation_store.create_conversation(kind="sub_agent") # Append distinct items to each conversation. conversation_store.append( conv_a.id, [ NewConversationItem( type="message", response_id="task_a", data=MessageData( role="user", content=[{"type": "input_text", "text": "alpha input"}], ), ), NewConversationItem( type="message", response_id="task_a", data=MessageData( role="assistant", content=[{"type": "output_text", "text": "alpha output"}], agent="researcher", ), ), ], ) conversation_store.append( conv_b.id, [ NewConversationItem( type="message", response_id="task_b", data=MessageData( role="user", content=[{"type": "input_text", "text": "bravo input"}], ), ), NewConversationItem( type="message", response_id="task_b", data=MessageData( role="assistant", content=[{"type": "output_text", "text": "bravo output"}], agent="researcher", ), ), ], ) # List items for conv_a — must contain only alpha items. page_a = conversation_store.list_items(conv_a.id) # 2 items: user + assistant for the alpha sub-agent. assert len(page_a.data) == 2, ( f"Expected 2 items in conv_a, got {len(page_a.data)}. " "If > 2, items from conv_b leaked into conv_a's listing." ) texts_a = [item.data.content[0]["text"] for item in page_a.data] assert texts_a == ["alpha input", "alpha output"], ( f"Expected alpha items only in conv_a, got {texts_a}. " "If bravo items appear, the conversation_id filter is broken." ) # Every item must carry the correct response_id. for item in page_a.data: assert item.response_id == "task_a", ( f"Item {item.id} in conv_a has response_id {item.response_id!r}, expected 'task_a'." ) # List items for conv_b — must contain only bravo items. page_b = conversation_store.list_items(conv_b.id) # 2 items: user + assistant for the bravo sub-agent. assert len(page_b.data) == 2, ( f"Expected 2 items in conv_b, got {len(page_b.data)}. " "If > 2, items from conv_a leaked into conv_b's listing." ) texts_b = [item.data.content[0]["text"] for item in page_b.data] assert texts_b == ["bravo input", "bravo output"], ( f"Expected bravo items only in conv_b, got {texts_b}. " "If alpha items appear, the conversation_id filter is broken." ) for item in page_b.data: assert item.response_id == "task_b", ( f"Item {item.id} in conv_b has response_id {item.response_id!r}, expected 'task_b'." ) # Cross-check: item IDs must be disjoint. ids_a = {item.id for item in page_a.data} ids_b = {item.id for item in page_b.data} assert ids_a.isdisjoint(ids_b), ( f"Item IDs overlap between conversations: " f"{ids_a & ids_b}. Each conversation must have " "unique item IDs." ) # ── updated_at ───────────────────────────────────────── def test_create_sets_updated_at_equal_to_created_at( conversation_store: SqlAlchemyConversationStore, ) -> None: """ A newly created conversation has updated_at == created_at. """ conv = conversation_store.create_conversation() assert conv.updated_at == conv.created_at, ( f"Expected updated_at ({conv.updated_at}) to equal " f"created_at ({conv.created_at}) on a brand-new conversation." ) def test_append_bumps_updated_at( conversation_store: SqlAlchemyConversationStore, monkeypatch: pytest.MonkeyPatch, ) -> None: """ Appending items to a conversation advances updated_at to the current time. """ import omnigent.stores.conversation_store.sqlalchemy_store as store_mod # Freeze time at creation monkeypatch.setattr(store_mod, "now_epoch", lambda: 1000) conv = conversation_store.create_conversation() assert conv.updated_at == 1000 # Advance time, then append monkeypatch.setattr(store_mod, "now_epoch", lambda: 2000) conversation_store.append( conv.id, [ NewConversationItem( type="message", response_id="resp_bump", data=MessageData( role="user", content=[{"type": "input_text", "text": "hi"}], ), ), ], ) fetched = conversation_store.get_conversation(conv.id) assert fetched is not None assert fetched.updated_at == 2000, ( f"Expected updated_at to advance to 2000 after append, got {fetched.updated_at}." ) def test_update_title_bumps_updated_at( conversation_store: SqlAlchemyConversationStore, monkeypatch: pytest.MonkeyPatch, ) -> None: """ Updating the title of a conversation advances updated_at. """ import omnigent.stores.conversation_store.sqlalchemy_store as store_mod monkeypatch.setattr(store_mod, "now_epoch", lambda: 1000) conv = conversation_store.create_conversation() assert conv.updated_at == 1000 monkeypatch.setattr(store_mod, "now_epoch", lambda: 3000) updated = conversation_store.update_conversation(conv.id, title="New title") assert updated is not None assert updated.updated_at == 3000, ( f"Expected updated_at to advance to 3000 after title update, got {updated.updated_at}." ) # ── sort_by=updated_at ──────────────────────────────── def test_list_conversations_sort_by_updated_at( conversation_store: SqlAlchemyConversationStore, monkeypatch: pytest.MonkeyPatch, ) -> None: """ Sorting by updated_at returns conversations in order of last activity, not creation order. """ import omnigent.stores.conversation_store.sqlalchemy_store as store_mod # Create conv_a at t=100, conv_b at t=200 monkeypatch.setattr(store_mod, "now_epoch", lambda: 100) conv_a = conversation_store.create_conversation() monkeypatch.setattr(store_mod, "now_epoch", lambda: 200) conv_b = conversation_store.create_conversation() # Append to conv_a at t=300, making it the most recently updated monkeypatch.setattr(store_mod, "now_epoch", lambda: 300) conversation_store.append( conv_a.id, [ NewConversationItem( type="message", response_id="resp_sort", data=MessageData( role="user", content=[{"type": "input_text", "text": "hello"}], ), ), ], ) # sort_by=created_at desc → conv_b first (created later) by_created = conversation_store.list_conversations( sort_by="created_at", order="desc", kind=None, ) assert by_created.data[0].id == conv_b.id, ( "Expected conv_b first when sorting by created_at desc." ) # sort_by=updated_at desc → conv_a first (updated more recently) by_updated = conversation_store.list_conversations( sort_by="updated_at", order="desc", kind=None, ) assert by_updated.data[0].id == conv_a.id, ( "Expected conv_a first when sorting by updated_at desc, " "because it was updated at t=300 vs conv_b at t=200." ) def test_list_conversations_sort_by_updated_at_with_pagination( conversation_store: SqlAlchemyConversationStore, monkeypatch: pytest.MonkeyPatch, ) -> None: """ Cursor-based pagination works correctly when sorting by updated_at. """ import omnigent.stores.conversation_store.sqlalchemy_store as store_mod # Create 3 conversations with distinct updated_at values ids = [] for t in (100, 200, 300): monkeypatch.setattr(store_mod, "now_epoch", lambda _t=t: _t) conv = conversation_store.create_conversation() ids.append(conv.id) # Reverse the update order: bump the oldest conversation last monkeypatch.setattr(store_mod, "now_epoch", lambda: 400) conversation_store.append( ids[0], [ NewConversationItem( type="message", response_id="resp_pg", data=MessageData( role="user", content=[{"type": "input_text", "text": "pg"}], ), ), ], ) # sort_by=updated_at desc: ids[0] (400), ids[2] (300), ids[1] (200) page1 = conversation_store.list_conversations( limit=2, sort_by="updated_at", order="desc", kind=None, ) # 2 results with has_more=True assert len(page1.data) == 2 assert page1.has_more is True assert page1.data[0].id == ids[0] assert page1.data[1].id == ids[2] page2 = conversation_store.list_conversations( limit=2, sort_by="updated_at", order="desc", after=page1.last_id, kind=None, ) # 1 result remaining assert len(page2.data) == 1 assert page2.has_more is False assert page2.data[0].id == ids[1] # ─── Phase 4: parent_conversation_id + name uniqueness ────── def test_create_conversation_with_parent_pointer_and_title( conversation_store: SqlAlchemyConversationStore, ) -> None: """Setting ``parent_conversation_id`` + ``title`` round-trips through the row.""" parent = conversation_store.create_conversation() child = conversation_store.create_conversation( kind="sub_agent", title="coder:auth", parent_conversation_id=parent.id, ) # Both fields surface on the entity — proves the row was # populated AND the converter pulls the column. Without the # converter update, parent_conversation_id would always be # None on the returned entity even after the row stores it. fetched = conversation_store.get_conversation(child.id) assert fetched is not None assert fetched.title == "coder:auth" assert fetched.parent_conversation_id == parent.id assert fetched.kind == "sub_agent" def test_create_duplicate_title_under_same_parent_raises( conversation_store: SqlAlchemyConversationStore, ) -> None: """G36: partial unique index rejects ``(parent_id, title)`` duplicates.""" from omnigent.stores.conversation_store import NameAlreadyExistsError parent = conversation_store.create_conversation() conversation_store.create_conversation( kind="sub_agent", title="coder:auth", parent_conversation_id=parent.id, ) # Without the partial unique index + IntegrityError-to- # NameAlreadyExistsError translation, the second create # would either succeed silently (creating a duplicate row) # or raise a raw sqlalchemy IntegrityError that would leak # through to the LLM as an opaque error. with pytest.raises(NameAlreadyExistsError): conversation_store.create_conversation( kind="sub_agent", title="coder:auth", parent_conversation_id=parent.id, ) def test_create_same_title_under_different_parents_succeeds( conversation_store: SqlAlchemyConversationStore, ) -> None: """The unique constraint is per-parent — ``(p1, "auth")`` and ``(p2, "auth")`` coexist.""" p1 = conversation_store.create_conversation() p2 = conversation_store.create_conversation() conversation_store.create_conversation( kind="sub_agent", title="coder:auth", parent_conversation_id=p1.id ) # Same title, different parent — no conflict. conversation_store.create_conversation( kind="sub_agent", title="coder:auth", parent_conversation_id=p2.id ) # Both children must exist; if the unique constraint were # global (not partial-by-parent), the second create would # raise. p1_children = conversation_store.list_conversations( kind="sub_agent", parent_conversation_id=p1.id, ) p2_children = conversation_store.list_conversations( kind="sub_agent", parent_conversation_id=p2.id, ) assert len(p1_children.data) == 1 assert len(p2_children.data) == 1 assert p1_children.data[0].title == "coder:auth" assert p2_children.data[0].title == "coder:auth" def test_create_null_parent_allows_duplicate_titles( conversation_store: SqlAlchemyConversationStore, ) -> None: """Top-level conversations (NULL parent) are NOT subject to the unique constraint.""" # Both conversations share title="" and parent=None. The unique index on # (parent_conversation_id, title) still allows this: a NULL in any indexed # column makes the key distinct, so top-level rows never collide even # without a WHERE predicate. a = conversation_store.create_conversation() b = conversation_store.create_conversation() assert a.id != b.id assert a.parent_conversation_id is None assert b.parent_conversation_id is None def test_list_conversations_filtered_by_parent_returns_children_only( conversation_store: SqlAlchemyConversationStore, ) -> None: """``parent_conversation_id`` filter scopes results to one parent's sub-tree.""" parent_a = conversation_store.create_conversation() parent_b = conversation_store.create_conversation() conversation_store.create_conversation( kind="sub_agent", title="coder:auth", parent_conversation_id=parent_a.id ) conversation_store.create_conversation( kind="sub_agent", title="coder:payments", parent_conversation_id=parent_a.id ) conversation_store.create_conversation( kind="sub_agent", title="coder:other", parent_conversation_id=parent_b.id ) page = conversation_store.list_conversations( kind="sub_agent", parent_conversation_id=parent_a.id, ) # Exactly 2 children for parent_a — proves the WHERE clause # excludes parent_b's child. If the filter were a no-op, # all 3 sub-agent rows would appear. titles = sorted(c.title for c in page.data if c.title) assert titles == ["coder:auth", "coder:payments"] def test_list_conversations_filtered_by_title( conversation_store: SqlAlchemyConversationStore, ) -> None: """``title`` filter returns only children with an exact title match.""" parent = conversation_store.create_conversation() conversation_store.create_conversation( kind="sub_agent", title="coder:auth", parent_conversation_id=parent.id ) conversation_store.create_conversation( kind="sub_agent", title="coder:payments", parent_conversation_id=parent.id ) page = conversation_store.list_conversations( kind="sub_agent", parent_conversation_id=parent.id, title="coder:auth", ) assert len(page.data) == 1 assert page.data[0].title == "coder:auth" empty = conversation_store.list_conversations( kind="sub_agent", parent_conversation_id=parent.id, title="coder:nonexistent", ) assert len(empty.data) == 0 def test_list_child_conversation_ids_by_parent_groups_direct_subagents( conversation_store: SqlAlchemyConversationStore, ) -> None: """ ``list_child_conversation_ids_by_parent`` groups direct sub-agent children. The sessions list uses this helper to roll child runner status onto parent rows without issuing one ``child_sessions`` query per sidebar row. This test proves the helper returns every requested parent key, excludes other parents and non-sub-agent children, and does not widen from direct children to nested descendants. """ parent_a = conversation_store.create_conversation() parent_b = conversation_store.create_conversation() child_a1 = conversation_store.create_conversation( kind="sub_agent", title="coder:auth", parent_conversation_id=parent_a.id ) child_a2 = conversation_store.create_conversation( kind="sub_agent", title="coder:payments", parent_conversation_id=parent_a.id ) child_b = conversation_store.create_conversation( kind="sub_agent", title="coder:other", parent_conversation_id=parent_b.id ) default_child = conversation_store.create_conversation( kind="default", title="default:child", parent_conversation_id=parent_a.id ) nested = conversation_store.create_conversation( kind="sub_agent", title="reviewer:nested", parent_conversation_id=child_a1.id ) result = conversation_store.list_child_conversation_ids_by_parent( [parent_a.id, parent_b.id, "conv_missing", parent_a.id] ) assert set(result) == {parent_a.id, parent_b.id, "conv_missing"} assert len(result[parent_a.id]) == 2 assert sorted(result[parent_a.id]) == sorted([child_a1.id, child_a2.id]) assert result[parent_b.id] == [child_b.id] assert result["conv_missing"] == [] assert default_child.id not in result[parent_a.id] assert nested.id not in result[parent_a.id] # ── agent_id filter on list_conversations ────────────── def test_list_conversations_filtered_by_agent_id_returns_matching_only( conversation_store: SqlAlchemyConversationStore, agent_store: SqlAlchemyAgentStore, ) -> None: """ Powers Omnigent mode ``--continue`` (resume the most-recent conversation for *this agent*). Two agents, three conversations: agent_alpha owns convs 1+2, agent_beta owns conv 3. Filtering by agent_alpha returns exactly the two conversations agent_alpha touched. The filter uses ``conversations.agent_id`` directly (the tasks table has been removed). What breaks if this fails: ``--continue`` either resumes a conversation belonging to the wrong agent (privacy / correctness regression) or returns nothing when prior history exists. :param conversation_store: The conversation store fixture. :param agent_store: The agent store fixture. """ alpha = agent_store.create(agent_id="ag_alpha", name="alpha", bundle_location="ag_alpha/h") beta = agent_store.create(agent_id="ag_beta", name="beta", bundle_location="ag_beta/h") conv1 = conversation_store.create_conversation(agent_id=alpha.id) conv2 = conversation_store.create_conversation(agent_id=alpha.id) conv3 = conversation_store.create_conversation(agent_id=beta.id) _ = conv3 # ensure conv3 exists but is not returned page = conversation_store.list_conversations(agent_id=alpha.id) returned_ids = {c.id for c in page.data} # Exactly two conversations, both owned by alpha — beta's # conversation must be excluded. assert returned_ids == {conv1.id, conv2.id} def test_list_conversations_agent_id_none_disables_filter( conversation_store: SqlAlchemyConversationStore, agent_store: SqlAlchemyAgentStore, ) -> None: """ The default (``agent_id=None``) returns every conversation, including ones without an agent binding. Pinning this so the filter stays opt-in and existing callers (the ``/switch`` slash command, list endpoint without filter) keep their cross-agent visibility. :param conversation_store: The conversation store fixture. :param agent_store: The agent store fixture. """ alpha = agent_store.create(agent_id="ag_alpha2", name="alpha2", bundle_location="ag_alpha2/h") conv_with_agent = conversation_store.create_conversation(agent_id=alpha.id) conv_without_agent = conversation_store.create_conversation() page = conversation_store.list_conversations() returned_ids = {c.id for c in page.data} # Both conversations present — no agent_id filter applied. assert conv_with_agent.id in returned_ids assert conv_without_agent.id in returned_ids def test_list_conversations_filter_distinct_by_agent_id( conversation_store: SqlAlchemyConversationStore, agent_store: SqlAlchemyAgentStore, ) -> None: """ A conversation bound to an agent appears exactly once in the result when filtered by that agent's id. The filter now uses ``conversations.agent_id`` directly (the tasks table has been removed) — each conversation appears at most once by definition. :param conversation_store: The conversation store fixture. :param agent_store: The agent store fixture. """ alpha = agent_store.create(agent_id="ag_alpha3", name="alpha3", bundle_location="ag_alpha3/h") conv = conversation_store.create_conversation(agent_id=alpha.id) page = conversation_store.list_conversations(agent_id=alpha.id) assert [c.id for c in page.data] == [conv.id] def test_list_conversations_filter_orders_by_sort_by( conversation_store: SqlAlchemyConversationStore, agent_store: SqlAlchemyAgentStore, monkeypatch: pytest.MonkeyPatch, ) -> None: """ With ``agent_id`` AND ``sort_by="updated_at"``, the result is ordered by the conversation's updated_at. This is what ``--continue`` relies on: "the conversation I most recently *did anything in*", which is reflected in the conversation's own ``updated_at`` (bumped on every append). The filter uses ``conversations.agent_id`` directly (the tasks table has been removed). :param conversation_store: The conversation store fixture. :param agent_store: The agent store fixture. :param monkeypatch: Pytest monkeypatch for time control. """ import omnigent.stores.conversation_store.sqlalchemy_store as store_mod alpha = agent_store.create(agent_id="ag_alpha4", name="alpha4", bundle_location="ag_alpha4/h") # Create two conversations at distinct timestamps so # ``updated_at`` differs. monkeypatch.setattr(store_mod, "now_epoch", lambda: 100) older = conversation_store.create_conversation(agent_id=alpha.id) monkeypatch.setattr(store_mod, "now_epoch", lambda: 200) newer = conversation_store.create_conversation(agent_id=alpha.id) _ = newer # both created; newer has later created_at # Bump older's updated_at so it becomes the most recently updated. monkeypatch.setattr(store_mod, "now_epoch", lambda: 300) conversation_store.append( older.id, [ NewConversationItem( type="message", response_id="resp_recency", data=MessageData( role="user", content=[{"type": "input_text", "text": "bump"}], ), ), ], ) page = conversation_store.list_conversations( agent_id=alpha.id, sort_by="updated_at", order="desc", limit=1, ) # ``older`` is now the most-recently-updated conversation # despite being created earlier. assert [c.id for c in page.data] == [older.id] def test_cascade_delete_removes_descendants( conversation_store: SqlAlchemyConversationStore, ) -> None: """Deleting a parent recursively removes children + grandchildren (FK CASCADE).""" import asyncio parent = conversation_store.create_conversation() child = conversation_store.create_conversation( kind="sub_agent", title="coder:auth", parent_conversation_id=parent.id ) grandchild = conversation_store.create_conversation( kind="sub_agent", title="reviewer:nested", parent_conversation_id=child.id ) # Delete the root — both descendants must vanish via the # ON DELETE CASCADE on parent_conversation_id. asyncio.run(conversation_store.delete_conversation(parent.id)) assert conversation_store.get_conversation(parent.id) is None assert conversation_store.get_conversation(child.id) is None, ( "Child not cascaded — FK ondelete=CASCADE missing or migration didn't apply it" ) assert conversation_store.get_conversation(grandchild.id) is None, ( "Grandchild not cascaded — recursive FK cascade missing" ) # ── Runner pinning (designs/RUNNER.md §5) ───────────── def test_runner_id_default_null( conversation_store: SqlAlchemyConversationStore, ) -> None: """Fresh conversations have no runner pin until first dispatch claims one.""" conv = conversation_store.create_conversation() fetched = conversation_store.get_conversation(conv.id) assert fetched is not None # ``runner_id`` is the load-bearing assertion — proves the column flowed # all the way from the DB row to the entity. A non-None default would # mean the entity dataclass is masking the SQL NULL. assert fetched.runner_id is None def test_replace_runner_id_allows_internal_non_session_conversation( conversation_store: SqlAlchemyConversationStore, ) -> None: """Internal sub-agent conversations can inherit runner bindings.""" conv = conversation_store.create_conversation() updated = conversation_store.replace_runner_id(conv.id, "runner-uuid-1") assert updated.runner_id == "runner-uuid-1" fetched = conversation_store.get_conversation(conv.id) assert fetched is not None assert fetched.runner_id == "runner-uuid-1" def test_list_conversations_by_runner_id_filters( conversation_store: SqlAlchemyConversationStore, ) -> None: """Only conversations pinned to the queried runner are returned. The runner tunnel's connect/disconnect callbacks use this lookup to find the sessions whose ``create_session`` handshake (and the claude-native terminal bootstrap behind it) must be driven on reconnect — a wrong or empty result here means the terminal is silently never created. """ bound = conversation_store.create_conversation(runner_id="runner_token_a") conversation_store.create_conversation(runner_id="runner_token_b") conversation_store.create_conversation() # unbound result = conversation_store.list_conversations_by_runner_id("runner_token_a") # Exactly the bound conversation — proves the filter matched on the # column rather than returning a superset the caller must re-filter. assert [c.id for c in result] == [bound.id] assert result[0].runner_id == "runner_token_a" # ── Host id ───────────────────────────────────────── def _register_host(db_uri: str, host_id: str) -> None: """ Insert a ``hosts`` row so a conversation can reference ``host_id``. ``conversations.host_id`` is an FK to ``hosts.host_id`` (enforced under ``PRAGMA foreign_keys=ON``), so the referenced host must exist before a conversation can be bound to it. Shares ``db_uri`` with the ``conversation_store`` fixture so both hit the same DB. :param db_uri: SQLite URI shared with the conversation store. :param host_id: Host identifier to register, e.g. ``"host_abc123"``. """ HostStore(db_uri).upsert_on_connect(host_id, f"laptop-{host_id}", RESERVED_USER_LOCAL) def test_host_id_defaults_to_none( conversation_store: SqlAlchemyConversationStore, ) -> None: """ A freshly created conversation has ``host_id=None``. If not None, the entity dataclass or the row→entity converter is fabricating a default instead of reflecting the SQL NULL. """ conv = conversation_store.create_conversation() assert conv.host_id is None fetched = conversation_store.get_conversation(conv.id) assert fetched is not None assert fetched.host_id is None def test_create_conversation_with_host_id( conversation_store: SqlAlchemyConversationStore, db_uri: str, ) -> None: """ Verify that host_id passed to create_conversation is persisted and survives a round-trip through the DB. Pass ``workspace`` alongside ``host_id`` because the schema's ``ck_conversations_workspace_required_for_host`` constraint forbids the (host_id NOT NULL, workspace NULL) combination — sessions targeting a host always need a path to launch in. If the fetched host_id doesn't match, either the INSERT is dropping the column or the row→entity converter is skipping it. """ # host_id is an FK to hosts.host_id, so the host must exist first. _register_host(db_uri, "host_abc123") conv = conversation_store.create_conversation( host_id="host_abc123", workspace="/Users/corey/projects/myapp", ) assert conv.host_id == "host_abc123" assert conv.workspace == "/Users/corey/projects/myapp" fetched = conversation_store.get_conversation(conv.id) assert fetched is not None assert fetched.host_id == "host_abc123" assert fetched.workspace == "/Users/corey/projects/myapp" def test_create_conversation_with_git_branch( conversation_store: SqlAlchemyConversationStore, ) -> None: """ Verify git_branch passed to create_conversation persists and round-trips through the DB. If the fetched git_branch doesn't match, either the INSERT drops the column (bad migration / model) or the row→entity converter skips it — both would break the delete-dialog cleanup gate, which keys off ``git_branch IS NOT NULL``. """ conv = conversation_store.create_conversation(git_branch="feature/login") assert conv.git_branch == "feature/login" fetched = conversation_store.get_conversation(conv.id) assert fetched is not None assert fetched.git_branch == "feature/login" def test_create_conversation_git_branch_defaults_none( conversation_store: SqlAlchemyConversationStore, ) -> None: """ Verify git_branch defaults to None for sessions with no worktree. A non-None default would make every session look like a worktree session and wrongly surface the "delete local branch" checkbox. """ conv = conversation_store.create_conversation() fetched = conversation_store.get_conversation(conv.id) assert fetched is not None assert fetched.git_branch is None def test_set_host_id( conversation_store: SqlAlchemyConversationStore, db_uri: str, ) -> None: """ Verify that set_host_id updates the column and persists. The conversation is created with ``workspace`` already set so that the post-update row satisfies ``ck_conversations_workspace_required_for_host``. Without a pre-existing workspace, the UPDATE that introduces ``host_id`` would be blocked by the check constraint. If the fetched value is still None after set_host_id, the UPDATE statement is not executing or not committing. """ # host_id is an FK to hosts.host_id, so the host must exist first. _register_host(db_uri, "host_def456") # Pre-set workspace so the host_id update doesn't violate the # workspace-required-for-host check constraint. conv = conversation_store.create_conversation( workspace="/Users/corey/projects/myapp", ) assert conv.host_id is None assert conv.workspace == "/Users/corey/projects/myapp" updated = conversation_store.set_host_id(conv.id, "host_def456") assert updated.host_id == "host_def456" fetched = conversation_store.get_conversation(conv.id) assert fetched is not None assert fetched.host_id == "host_def456" assert fetched.workspace == "/Users/corey/projects/myapp" def test_set_host_id_missing_conversation_raises( conversation_store: SqlAlchemyConversationStore, ) -> None: """ Verify that set_host_id raises ConversationNotFoundError for a nonexistent conversation. If it silently succeeds, the guard clause is missing and a stale host_id could be written to a phantom row. """ from omnigent.stores.conversation_store import ConversationNotFoundError with pytest.raises(ConversationNotFoundError): conversation_store.set_host_id("conv_nonexistent", "host_xyz") def test_set_host_id_with_workspace_satisfies_constraint( conversation_store: SqlAlchemyConversationStore, db_uri: str, ) -> None: """ Verify set_host_id(host_id, workspace) writes both columns so the row satisfies ck_conversations_workspace_required_for_host. The launch-runner endpoint binds host_id post-create on rows that may have NULL workspace; without the workspace argument, the UPDATE would violate the check constraint and 500 the request. """ # host_id is an FK to hosts.host_id, so the host must exist first. _register_host(db_uri, "host_with_workspace") conv = conversation_store.create_conversation() assert conv.host_id is None assert conv.workspace is None updated = conversation_store.set_host_id( conv.id, "host_with_workspace", workspace="/Users/corey/projects/myapp", ) assert updated.host_id == "host_with_workspace" assert updated.workspace == "/Users/corey/projects/myapp" def test_clear_host_binding_nulls_all_binding_fields( conversation_store: SqlAlchemyConversationStore, db_uri: str, ) -> None: """ clear_host_binding NULLs host_id/workspace/git_branch/runner_id together. This is the failed-bind rollback primitive: after a worktree launch fails and the worktree is removed, the session must not keep pointing at the deleted worktree/branch or stay runner-bound. Unlike set_host_id (None = leave untouched, so it can't clear git_branch), this fully reverts to unbound. A leftover git_branch here would wrongly satisfy worktree-cleanup paths (git_branch IS NOT NULL); a leftover runner_id would block the picker's retry on the atomic set_runner_id CAS. """ _register_host(db_uri, "host_to_unbind") conv = conversation_store.create_conversation() # Fully bind it the way a worktree launch would: host + worktree path # + branch, then a runner. conversation_store.set_host_id( conv.id, "host_to_unbind", workspace="/Users/corey/repo-worktrees/feature-x", git_branch="feature/x", ) assert conversation_store.set_runner_id(conv.id, "runner_token_abc") is True bound = conversation_store.get_conversation(conv.id) assert bound is not None # Precondition: every binding field is set (so the assertions below # prove clearing, not a no-op on already-empty fields). assert bound.host_id == "host_to_unbind" assert bound.workspace == "/Users/corey/repo-worktrees/feature-x" assert bound.git_branch == "feature/x" assert bound.runner_id == "runner_token_abc" cleared = conversation_store.clear_host_binding(conv.id) assert cleared.host_id is None assert cleared.workspace is None assert cleared.git_branch is None assert cleared.runner_id is None # Persisted, not just returned: a re-fetch must agree. refetched = conversation_store.get_conversation(conv.id) assert refetched is not None assert refetched.host_id is None assert refetched.workspace is None assert refetched.git_branch is None assert refetched.runner_id is None # And the session can be re-bound (the CAS sees runner_id IS NULL). assert conversation_store.set_runner_id(conv.id, "runner_token_retry") is True def test_clear_host_binding_missing_conversation_raises( conversation_store: SqlAlchemyConversationStore, ) -> None: """clear_host_binding raises for an unknown conversation id.""" from omnigent.stores.conversation_store import ConversationNotFoundError with pytest.raises(ConversationNotFoundError): conversation_store.clear_host_binding("conv_nonexistent") def test_create_session_with_agent_records_workspace( conversation_store: SqlAlchemyConversationStore, ) -> None: """ Verify create_session_with_agent stores workspace= on the conversation row. The CLI calls this path with ``workspace=os.getcwd()`` so the Web UI can show "running locally in " for sessions started outside the Web UI. If the column isn't populated, the Web UI shows "(no workspace)" and the user can't tell where the session is running. """ created = conversation_store.create_session_with_agent( agent_id="ag_session_ws", agent_name="cli-test-agent", agent_bundle_location="ag_session_ws/bundle1", agent_description=None, workspace="/Users/corey/projects/cli-launch", ) fetched = conversation_store.get_conversation(created.conversation.id) assert fetched is not None assert fetched.workspace == "/Users/corey/projects/cli-launch" assert fetched.host_id is None def test_create_session_with_agent_workspace_defaults_to_none( conversation_store: SqlAlchemyConversationStore, ) -> None: """ Verify create_session_with_agent leaves workspace NULL when no value is passed. Headless API callers (no host, no terminal cwd) shouldn't be forced to invent a workspace. The check constraint allows NULL workspace when host_id is also NULL — this test pins that path through the public method. """ created = conversation_store.create_session_with_agent( agent_id="ag_session_no_ws", agent_name="no-ws-agent", agent_bundle_location="ag_session_no_ws/bundle1", agent_description=None, ) fetched = conversation_store.get_conversation(created.conversation.id) assert fetched is not None assert fetched.workspace is None def test_create_session_with_agent_records_terminal_launch_args( conversation_store: SqlAlchemyConversationStore, ) -> None: """ Verify create_session_with_agent persists terminal_launch_args as a JSON list that round-trips back to a real ``list[str]``. The native wrappers set these at create-time so a daemon- or server-launched runner can reconstruct the terminal command. If the JSON encode (row builder) or decode (converter) is broken, the fetched value would be a raw JSON string or ``None`` instead of the list, and the runner would launch with the wrong args. """ created = conversation_store.create_session_with_agent( agent_id="ag_session_tla", agent_name="cli-test-agent", agent_bundle_location="ag_session_tla/bundle1", agent_description=None, terminal_launch_args=["--dangerously-skip-permissions", "--model", "opus"], ) fetched = conversation_store.get_conversation(created.conversation.id) assert fetched is not None # Exact list (order + content) proves the value traversed # encode→DB→decode intact, not just "something non-null". assert fetched.terminal_launch_args == [ "--dangerously-skip-permissions", "--model", "opus", ] def test_create_session_with_agent_terminal_launch_args_defaults_to_none( conversation_store: SqlAlchemyConversationStore, ) -> None: """ Verify create_session_with_agent leaves terminal_launch_args NULL when no value is passed. Non-native sessions (the common case) must not carry launch args. A NULL column must decode to ``None`` — not ``[]`` or ``"null"`` — so downstream code can distinguish "no native launch" from "native launch with zero extra args". """ created = conversation_store.create_session_with_agent( agent_id="ag_session_no_tla", agent_name="no-tla-agent", agent_bundle_location="ag_session_no_tla/bundle1", agent_description=None, ) fetched = conversation_store.get_conversation(created.conversation.id) assert fetched is not None assert fetched.terminal_launch_args is None def test_create_session_with_agent_links_parent_and_inherits_root( conversation_store: SqlAlchemyConversationStore, ) -> None: """ Verify create_session_with_agent with parent_conversation_id creates a sub-agent child that inherits the parent's spawn-tree root and runner binding. This is the bundle-mode ``sys_session_create`` path: the multipart create must produce the same parent/kind/root/runner shape as the JSON child create, or tree-scoped reads (``sys_session_get_history`` matches on ``root_conversation_id``) and runner co-location would silently break for bundle-created children. """ parent = conversation_store.create_conversation(runner_id="runner_swa1") created = conversation_store.create_session_with_agent( agent_id="ag_session_child", agent_name="bundle-child-agent", agent_bundle_location="ag_session_child/bundle1", agent_description=None, parent_conversation_id=parent.id, runner_id=parent.runner_id, ) fetched = conversation_store.get_conversation(created.conversation.id) assert fetched is not None assert fetched.parent_conversation_id == parent.id # Child rows are sub-agent kind — sys_session_list filters on this. assert fetched.kind == "sub_agent" # Root must be the PARENT's root (== parent.id for a top-level # parent), not the child's own id — otherwise the child lands in # its own one-row tree and tree-scoped tools can't see it. assert fetched.root_conversation_id == parent.root_conversation_id assert fetched.runner_id == "runner_swa1" def test_create_session_with_agent_top_level_unchanged( conversation_store: SqlAlchemyConversationStore, ) -> None: """ Verify the no-parent path still creates a top-level default row. The parent support must not disturb the existing multipart create contract (CLI ``omnigent run`` sessions): kind stays "default", no parent link, and the row roots its own tree. """ created = conversation_store.create_session_with_agent( agent_id="ag_session_top", agent_name="top-level-agent", agent_bundle_location="ag_session_top/bundle1", agent_description=None, ) fetched = conversation_store.get_conversation(created.conversation.id) assert fetched is not None assert fetched.parent_conversation_id is None assert fetched.kind == "default" assert fetched.root_conversation_id == created.conversation.id assert fetched.runner_id is None def test_create_session_with_agent_missing_parent_fails_loud( conversation_store: SqlAlchemyConversationStore, ) -> None: """ Verify a nonexistent parent_conversation_id raises ConversationNotFoundError instead of silently creating an orphan. The route authorizes the parent before calling the store, but the parent can be deleted between the check and the insert — the store must fail loud so no half-linked child row (and no orphaned agent row) is committed. """ from omnigent.stores.conversation_store import ConversationNotFoundError with pytest.raises(ConversationNotFoundError): conversation_store.create_session_with_agent( agent_id="ag_session_orphan", agent_name="orphan-agent", agent_bundle_location="ag_session_orphan/bundle1", agent_description=None, parent_conversation_id="conv_does_not_exist", ) # The transaction rolled back: no agent row leaked either. assert conversation_store.get_conversation("conv_does_not_exist") is None def test_create_conversation_records_terminal_launch_args( conversation_store: SqlAlchemyConversationStore, ) -> None: """ Verify create_conversation persists terminal_launch_args as a JSON list that round-trips back to a real ``list[str]``. This is the JSON ``POST /v1/sessions`` path (web permission-mode selector), distinct from create_session_with_agent (multipart). If the row builder skipped the JSON encode or the converter the decode, the fetched value would be ``None`` or a raw string and the runner would launch claude without the chosen --permission-mode. """ created = conversation_store.create_conversation( terminal_launch_args=["--permission-mode", "bypassPermissions"], ) fetched = conversation_store.get_conversation(created.id) assert fetched is not None # Exact list (order preserved) proves the value traversed # encode→DB→decode intact. assert fetched.terminal_launch_args == ["--permission-mode", "bypassPermissions"] def test_create_conversation_terminal_launch_args_defaults_to_none( conversation_store: SqlAlchemyConversationStore, ) -> None: """ Verify create_conversation leaves terminal_launch_args NULL when no value is passed. Non-native / auto-mode-off sessions (the common case) must not carry launch args; a NULL column must decode to ``None``, not ``[]``, so the runner distinguishes "no native launch" from "native launch with zero extra args". """ created = conversation_store.create_conversation() fetched = conversation_store.get_conversation(created.id) assert fetched is not None assert fetched.terminal_launch_args is None def test_update_conversation_replaces_terminal_launch_args( conversation_store: SqlAlchemyConversationStore, ) -> None: """ Verify update_conversation replaces terminal_launch_args wholesale (last-write-wins) and that ``None`` leaves the stored value unchanged. This pins the resume semantics from designs/NATIVE_RUNNER_SERVER_LAUNCH.md: a cold resume with new flags overwrites the prior set rather than appending to it. If the store appended instead of replacing, the second assertion would see the concatenation and fail — which is exactly the bug that would make repeated resumes accumulate stale/conflicting flags. """ created = conversation_store.create_session_with_agent( agent_id="ag_session_tla_update", agent_name="update-agent", agent_bundle_location="ag_session_tla_update/bundle1", agent_description=None, terminal_launch_args=["--model", "opus"], ) conv_id = created.conversation.id replaced = conversation_store.update_conversation( conv_id, terminal_launch_args=["--verbose"], ) assert replaced is not None # Replaced, NOT appended: ["--model", "opus", "--verbose"] would # mean the store concatenated across launches (the bug). assert replaced.terminal_launch_args == ["--verbose"] # None must leave the prior value intact (matches the # reasoning_effort / model_override "None = unchanged" contract). unchanged = conversation_store.update_conversation(conv_id, title="renamed") assert unchanged is not None assert unchanged.terminal_launch_args == ["--verbose"] def test_update_conversation_terminal_launch_args_empty_list_distinct_from_none( conversation_store: SqlAlchemyConversationStore, ) -> None: """ Verify an explicitly-empty arg list round-trips as ``[]`` and stays distinct from the NULL/None "leave unchanged" sentinel. The store uses ``None`` to mean "leave unchanged", so an empty list must be storable as a real, retrievable ``[]`` — otherwise a caller clearing args back to none-extra would be silently ignored. """ created = conversation_store.create_session_with_agent( agent_id="ag_session_tla_empty", agent_name="empty-agent", agent_bundle_location="ag_session_tla_empty/bundle1", agent_description=None, terminal_launch_args=["--model", "opus"], ) updated = conversation_store.update_conversation( created.conversation.id, terminal_launch_args=[], ) assert updated is not None # "[]" is a truthy JSON string, so the converter must decode it to # an empty list — not collapse it to None the way a NULL column does. assert updated.terminal_launch_args == [] def test_set_host_id_no_workspace_fails_when_row_has_none( conversation_store: SqlAlchemyConversationStore, ) -> None: """ Verify that calling set_host_id without a workspace argument on a row whose workspace is NULL raises IntegrityError. Pins the contract that set_host_id can't bypass the check constraint by accident. Callers must pass a workspace when binding to a host on a row that doesn't already have one. """ from sqlalchemy.exc import IntegrityError, OperationalError conv = conversation_store.create_conversation() assert conv.workspace is None with pytest.raises((IntegrityError, OperationalError)): conversation_store.set_host_id(conv.id, "host_no_ws") # ── Workspace ─────────────────────────────────────── def test_workspace_defaults_to_none( conversation_store: SqlAlchemyConversationStore, ) -> None: """ A freshly created conversation has ``workspace=None`` when no workspace is passed. Load-bearing because the entity must mirror the DB's NULL state rather than substituting an empty string — empty-string defaults on path columns mask the "never set" case and would launch the runner in the host daemon's process cwd, which is rarely correct. """ conv = conversation_store.create_conversation() assert conv.workspace is None fetched = conversation_store.get_conversation(conv.id) assert fetched is not None assert fetched.workspace is None def test_workspace_persists_for_cli_session_without_host_id( conversation_store: SqlAlchemyConversationStore, ) -> None: """ A CLI session can record its starting cwd without a host_id. Pairs with the schema-level test in ``tests/db/test_migration_workspace.py`` that proves the check constraint is one-way (host_id requires workspace, not the reverse). This test exercises the same path through the store interface to catch regressions where the store layer might add a stricter rule than the schema enforces. """ conv = conversation_store.create_conversation( workspace="/Users/corey/projects/cli-launched", ) assert conv.host_id is None assert conv.workspace == "/Users/corey/projects/cli-launched" fetched = conversation_store.get_conversation(conv.id) assert fetched is not None assert fetched.host_id is None assert fetched.workspace == "/Users/corey/projects/cli-launched" def test_create_conversation_with_host_id_no_workspace_raises( conversation_store: SqlAlchemyConversationStore, ) -> None: """ Creating a conversation with host_id but no workspace raises IntegrityError from the check constraint. The store deliberately does NOT add a Python-side guard here: the check constraint is the canonical enforcement, and adding a duplicate guard at the store would diverge from the schema in confusing ways. This test pins the contract: violating the constraint surfaces as IntegrityError, callers can catch it, and the row is never written. """ from sqlalchemy.exc import IntegrityError, OperationalError with pytest.raises((IntegrityError, OperationalError)): conversation_store.create_conversation(host_id="host_abc") # ── External session id ───────────────────────────── def test_external_session_id_defaults_to_none( conversation_store: SqlAlchemyConversationStore, ) -> None: """ A freshly created conversation has ``external_session_id=None``. Load-bearing because a non-None default would mean the entity dataclass is masking the SQL NULL — the wrapper bridge wouldn't be able to tell "not yet observed" from "set to empty". """ conv = conversation_store.create_conversation() assert conv.external_session_id is None fetched = conversation_store.get_conversation(conv.id) assert fetched is not None # Read-back through get_conversation goes through the row→entity # converter; this proves the column flows from DB row to entity. assert fetched.external_session_id is None def test_set_external_session_id_first_call_persists( conversation_store: SqlAlchemyConversationStore, ) -> None: """First write transitions NULL → value and is visible on read-back.""" conv = conversation_store.create_conversation() updated = conversation_store.set_external_session_id( conv.id, "a1b2c3d4-1234-5678-9abc-def012345678", ) # The returned entity reflects the write — the route's # response builder reads this snapshot rather than issuing a # follow-up GET. assert updated.external_session_id == "a1b2c3d4-1234-5678-9abc-def012345678" # Independent read-back proves the column was actually committed # (not just held in the returned dataclass). fetched = conversation_store.get_conversation(conv.id) assert fetched is not None assert fetched.external_session_id == "a1b2c3d4-1234-5678-9abc-def012345678" def test_set_external_session_id_same_value_is_idempotent( conversation_store: SqlAlchemyConversationStore, ) -> None: """ Re-writing the same value is a no-op and does not raise. The wrapper bridge may observe the Claude session id across many hook events; the latch in the forwarder limits this to one PATCH per process, but a second forwarder process for the same conv (server bounce, hot reload) would race-write the same value. The store must accept that without surfacing a spurious error. """ conv = conversation_store.create_conversation() first = conversation_store.set_external_session_id(conv.id, "sid-1") second = conversation_store.set_external_session_id(conv.id, "sid-1") assert first.external_session_id == "sid-1" assert second.external_session_id == "sid-1" fetched = conversation_store.get_conversation(conv.id) assert fetched is not None assert fetched.external_session_id == "sid-1" def test_set_external_session_id_rejects_overwrite_with_different_value( conversation_store: SqlAlchemyConversationStore, ) -> None: """ Attempting to overwrite an existing value raises ValueError. A divergent write signals a real bug (the bridge captured a different Claude session id than was previously recorded). The store surfaces it loudly so the route can return 400 instead of silently destroying the prior mapping. """ conv = conversation_store.create_conversation() conversation_store.set_external_session_id(conv.id, "sid-1") with pytest.raises(ValueError, match="already has external_session_id"): conversation_store.set_external_session_id(conv.id, "sid-2") fetched = conversation_store.get_conversation(conv.id) assert fetched is not None # First-writer-wins — the rejected second call must not have # mutated the row. If this assertion fails, the store silently # overwrote on conflict (the very bug the ValueError exists to # prevent). assert fetched.external_session_id == "sid-1" def test_set_external_session_id_missing_conversation_raises( conversation_store: SqlAlchemyConversationStore, ) -> None: """ Writing to a nonexistent conversation raises ConversationNotFoundError. Mirrors replace_runner_id / clear_runner_id — the public PATCH routes translate this into a 404, so silently no-oping here would let the route return 200 for a write that never happened. """ from omnigent.stores.conversation_store import ConversationNotFoundError with pytest.raises(ConversationNotFoundError): conversation_store.set_external_session_id( "conv_does_not_exist", "sid-1", ) # ── Fork conversation ──────────────────────────────── def test_fork_conversation_copies_items( conversation_store: SqlAlchemyConversationStore, agent_store: SqlAlchemyAgentStore, ) -> None: """Fork creates a new conversation with deep-copied items. Items in the fork must have fresh IDs but identical data. The source conversation must be untouched — no items removed or mutated. """ agent_store.create( agent_id="ag_fork_test", name="fork-test", bundle_location="ag_fork_test/fakehash", ) source = conversation_store.create_conversation( agent_id="ag_fork_test", title="Original", ) conversation_store.append( source.id, [ NewConversationItem( type="message", response_id="resp_001", data=MessageData( role="user", content=[{"type": "input_text", "text": "Hello"}], ), ), NewConversationItem( type="message", response_id="resp_001", data=MessageData( role="assistant", content=[{"type": "output_text", "text": "Hi there!"}], agent="test-agent", ), ), ], ) fork = conversation_store.fork_conversation(source.id, title="My Fork") # The fork is a new conversation with a different ID. assert fork.id != source.id assert fork.id.startswith("conv_") assert fork.title == "My Fork" # Agent binding is copied from the source. assert fork.agent_id == "ag_fork_test" # Items are deep-copied — same count, different IDs, same data. fork_items = conversation_store.list_items(fork.id) source_items = conversation_store.list_items(source.id) # Both conversations have 2 items. assert len(fork_items.data) == 2, ( f"Fork should have 2 items (same as source), got {len(fork_items.data)}" ) assert len(source_items.data) == 2, ( f"Source should still have 2 items after fork, got {len(source_items.data)}" ) for src_item, fork_item in zip(source_items.data, fork_items.data, strict=True): # IDs must differ (fresh generation). assert fork_item.id != src_item.id, "Fork item IDs must be fresh, not reused from source" # Type and response_id are preserved. assert fork_item.type == src_item.type assert fork_item.response_id == src_item.response_id # Data content is identical. assert fork_item.data == src_item.data def test_fork_conversation_preserves_created_by( conversation_store: SqlAlchemyConversationStore, agent_store: SqlAlchemyAgentStore, ) -> None: """Forking carries per-item actor attribution into the fork. Attribution history travels with the items; cloning the conversation does not blank out who authored each message. """ agent_store.create( agent_id="ag_fork_attr", name="fork-attr", bundle_location="ag_fork_attr/fakehash", ) source = conversation_store.create_conversation(agent_id="ag_fork_attr") conversation_store.append( source.id, [ NewConversationItem( type="message", response_id="resp_001", data=MessageData( role="user", content=[{"type": "input_text", "text": "Hello"}], ), created_by="alice@example.com", ), NewConversationItem( type="message", response_id="resp_001", data=MessageData( role="assistant", content=[{"type": "output_text", "text": "Hi!"}], agent="test-agent", ), ), ], ) fork = conversation_store.fork_conversation(source.id) fork_items = conversation_store.list_items(fork.id).data assert fork_items[0].created_by == "alice@example.com" assert fork_items[1].created_by is None def test_fork_conversation_default_title( conversation_store: SqlAlchemyConversationStore, agent_store: SqlAlchemyAgentStore, ) -> None: """When no title is given, fork derives one from the source title.""" agent_store.create( agent_id="ag_fork_title", name="fork-title", bundle_location="ag_fork_title/fakehash", ) source = conversation_store.create_conversation( agent_id="ag_fork_title", title="Chat about Python", ) fork = conversation_store.fork_conversation(source.id) assert fork.title == "Fork of Chat about Python" def test_fork_conversation_empty_source( conversation_store: SqlAlchemyConversationStore, agent_store: SqlAlchemyAgentStore, ) -> None: """Forking a conversation with no items produces an empty fork.""" agent_store.create( agent_id="ag_fork_empty", name="fork-empty", bundle_location="ag_fork_empty/fakehash", ) source = conversation_store.create_conversation(agent_id="ag_fork_empty") fork = conversation_store.fork_conversation(source.id) fork_items = conversation_store.list_items(fork.id) assert fork_items.data == [], "Fork of an empty conversation should have no items" assert fork.agent_id == "ag_fork_empty" def test_fork_conversation_nonexistent_raises( conversation_store: SqlAlchemyConversationStore, ) -> None: """Forking a non-existent conversation raises LookupError.""" with pytest.raises(LookupError, match="conversation not found"): conversation_store.fork_conversation("conv_does_not_exist") def test_fork_conversation_copies_labels( conversation_store: SqlAlchemyConversationStore, agent_store: SqlAlchemyAgentStore, ) -> None: """Labels on the source conversation are copied to the fork.""" agent_store.create( agent_id="ag_fork_labels", name="fork-labels", bundle_location="ag_fork_labels/fakehash", ) source = conversation_store.create_conversation(agent_id="ag_fork_labels") conversation_store.set_labels(source.id, {"sensitivity": "high", "dept": "eng"}) fork = conversation_store.fork_conversation(source.id) # Both labels must be copied — a mismatch means the store's fork # skipped the label-copy step or only copied partial keys. assert fork.labels == {"sensitivity": "high", "dept": "eng"}, ( "Fork should inherit all labels from the source conversation" ) def test_fork_conversation_drops_instance_scoped_labels( conversation_store: SqlAlchemyConversationStore, agent_store: SqlAlchemyAgentStore, ) -> None: """ Instance-scoped labels are NOT copied to the fork. The native bridge-id labels and the context metrics belong to the source's running instance. Copying the bridge-id in particular pointed a forked claude-native session at the SOURCE's bridge — the launched terminal + web injection keyed off it and hit "session no longer active after /clear" because the bridge's active-session marker wasn't the clone. The fork must drop them (and re-bind its own runtime), while ordinary labels still copy. The DANGEROUS codex full-bypass directive is in the same set for a different reason: a fork is a new session + workspace, so re-arming ``--dangerously-bypass-approvals-and-sandbox`` there with no typed re-confirmation would violate the "impossible to enable accidentally" contract (#657). It must be dropped so the clone opts in afresh. """ agent_store.create( agent_id="ag_fork_instance", name="fork-instance", bundle_location="ag_fork_instance/fakehash", ) source = conversation_store.create_conversation(agent_id="ag_fork_instance") conversation_store.set_labels( source.id, { "omnigent.claude_native.bridge_id": source.id, "omnigent.codex_native.bridge_id": source.id, "omnigent.last_context_tokens": "39903", "omnigent.last_context_window": "1000000", # The dangerous bypass opt-in must NOT ride into the fork. "omnigent.codex_native.bypass_sandbox": "1", # An ordinary, non-instance label that SHOULD carry over. "omnigent.wrapper": "claude-code-native-ui", }, ) fork = conversation_store.fork_conversation(source.id) # The clone keeps the harness identity (wrapper) but none of the # source's per-instance state. A bridge-id here would re-introduce # the cross-bridge bug; the metrics would show the source's stale # usage. assert fork.labels == {"omnigent.wrapper": "claude-code-native-ui"}, ( f"Fork must drop instance-scoped labels, kept {fork.labels!r}" ) def test_fork_conversation_stamps_source_external_session_id( conversation_store: SqlAlchemyConversationStore, agent_store: SqlAlchemyAgentStore, ) -> None: """ The source's native session id is stamped on the fork as a one-shot resume directive, while the clone's own ``external_session_id`` stays NULL. A native harness launching the clone uses ``omnigent.fork.source_external_session_id`` to resume + branch the source's local transcript (Claude Code ``--fork-session``), so the clone opens with prior history. The clone is NOT that session, so its own ``external_session_id`` must remain unset until it captures its own on first launch. Sources without a native session get no label. """ agent_store.create( agent_id="ag_fork_ext", name="fork-ext", bundle_location="ag_fork_ext/fakehash", ) source = conversation_store.create_conversation(agent_id="ag_fork_ext") conversation_store.set_external_session_id(source.id, "claude-uuid-abc") fork = conversation_store.fork_conversation(source.id) # Directive carries the SOURCE's claude uuid for the resume+fork launch. assert fork.labels.get("omnigent.fork.source_external_session_id") == "claude-uuid-abc", ( f"Fork should carry the source's external session id, got {fork.labels!r}" ) # The clone is a fresh session — it has no native session of its own # yet. Copying external_session_id would make two Omnigent sessions claim # the same Claude session. reloaded = conversation_store.get_conversation(fork.id) assert reloaded is not None assert reloaded.external_session_id is None, ( "Fork must not inherit the source's external_session_id" ) def test_fork_conversation_no_external_session_id_no_directive( conversation_store: SqlAlchemyConversationStore, agent_store: SqlAlchemyAgentStore, ) -> None: """A source with no native session id stamps no fork directive.""" agent_store.create( agent_id="ag_fork_noext", name="fork-noext", bundle_location="ag_fork_noext/fakehash", ) source = conversation_store.create_conversation(agent_id="ag_fork_noext") fork = conversation_store.fork_conversation(source.id) assert "omnigent.fork.source_external_session_id" not in fork.labels, ( f"No source native session → no resume directive, got {fork.labels!r}" ) def _append_three_responses( conversation_store: SqlAlchemyConversationStore, conversation_id: str, ) -> None: """ Append three user/assistant turns under distinct response ids. Builds the fixture history for the fork-truncation tests: response ``resp_001`` ("Q1"/"A1"), ``resp_002`` ("Q2"/"A2"), and ``resp_003`` ("Q3"/"A3"), six items total in chronological order. :param conversation_store: Store to append into. :param conversation_id: Target conversation id, e.g. ``"conv_abc123"``. """ for index in (1, 2, 3): conversation_store.append( conversation_id, [ NewConversationItem( type="message", response_id=f"resp_00{index}", data=MessageData( role="user", content=[{"type": "input_text", "text": f"Q{index}"}], ), ), NewConversationItem( type="message", response_id=f"resp_00{index}", data=MessageData( role="assistant", content=[{"type": "output_text", "text": f"A{index}"}], agent="test-agent", ), ), ], ) def test_fork_conversation_up_to_response_truncates_items( conversation_store: SqlAlchemyConversationStore, agent_store: SqlAlchemyAgentStore, ) -> None: """``up_to_response_id`` copies history through that response's last item. Forking at the middle response must copy every item up to and including the LAST item of that response (never ending mid-turn) and drop everything after it, while leaving the source untouched. """ agent_store.create( agent_id="ag_fork_trunc", name="fork-trunc", bundle_location="ag_fork_trunc/fakehash", ) source = conversation_store.create_conversation(agent_id="ag_fork_trunc") _append_three_responses(conversation_store, source.id) fork = conversation_store.fork_conversation(source.id, up_to_response_id="resp_002") fork_texts = [ part["text"] for item in conversation_store.list_items(fork.id).data for part in item.data.content # type: ignore[union-attr] ] # Both items of resp_002 are included (the cutoff is the response's # LAST item) and resp_003 is dropped. A1-only would mean the cutoff # used the response's FIRST item; any Q3/A3 means no truncation. assert fork_texts == ["Q1", "A1", "Q2", "A2"], ( f"Fork should contain history through resp_002 only, got {fork_texts}" ) # The source keeps its full 6-item history — fork must never mutate it. assert len(conversation_store.list_items(source.id).data) == 6 def test_fork_conversation_truncated_drops_external_session_directive( conversation_store: SqlAlchemyConversationStore, agent_store: SqlAlchemyAgentStore, ) -> None: """A truncated fork omits the native resume directive but keeps carry-history. If ``omnigent.fork.source_external_session_id`` were stamped, the runner would clone the source's FULL native transcript and resume it — resurrecting the truncated turns. The directive must be omitted so the runner's carry-history fork-rebuild path synthesizes the transcript from the truncated items instead. """ from omnigent.stores.conversation_store import ( FORK_CARRY_HISTORY_LABEL_KEY, FORK_SOURCE_EXTERNAL_SESSION_LABEL_KEY, ) agent_store.create( agent_id="ag_fork_trunc_ext", name="fork-trunc-ext", bundle_location="ag_fork_trunc_ext/fakehash", ) source = conversation_store.create_conversation(agent_id="ag_fork_trunc_ext") conversation_store.set_external_session_id(source.id, "claude-uuid-trunc") _append_three_responses(conversation_store, source.id) fork = conversation_store.fork_conversation( source.id, carry_history_into_native=True, up_to_response_id="resp_001", ) assert FORK_SOURCE_EXTERNAL_SESSION_LABEL_KEY not in fork.labels, ( f"Truncated fork must not carry the full-transcript resume directive, got {fork.labels!r}" ) # Carry-history still stamps: the runner needs it to rebuild the # native transcript from the truncated items rather than launch fresh. assert fork.labels.get(FORK_CARRY_HISTORY_LABEL_KEY) == "1" def test_fork_conversation_cross_family_drops_external_session_directive( conversation_store: SqlAlchemyConversationStore, agent_store: SqlAlchemyAgentStore, ) -> None: """``resume_source_native_session=False`` omits the native resume directive. A cross-family fork (e.g. codex-native source → claude-native target) must not stamp ``omnigent.fork.source_external_session_id``: the source's native transcript is the wrong format for the target harness, and the runner's clone path launches FRESH when its clone attempt fails — silently losing history. Omitting the directive routes the runner to the carry-history rebuild path (native transcript built from the copied Omnigent items) instead. """ from omnigent.stores.conversation_store import ( FORK_CARRY_HISTORY_LABEL_KEY, FORK_SOURCE_EXTERNAL_SESSION_LABEL_KEY, ) agent_store.create( agent_id="ag_fork_xfam", name="fork-xfam", bundle_location="ag_fork_xfam/fakehash", ) source = conversation_store.create_conversation(agent_id="ag_fork_xfam") # A codex thread id on the source: resumable only by a codex target. conversation_store.set_external_session_id(source.id, "codex-thread-xfam") _append_three_responses(conversation_store, source.id) fork = conversation_store.fork_conversation( source.id, carry_history_into_native=True, resume_source_native_session=False, ) # Despite a FULL (untruncated) fork of a source with a native session, # the directive must be absent — present would mean the cross-family # gate regressed and the runner will attempt a doomed transcript clone. assert FORK_SOURCE_EXTERNAL_SESSION_LABEL_KEY not in fork.labels, ( f"Cross-family fork must not stamp the source's native session id, got {fork.labels!r}" ) # Carry-history still stamps so the runner rebuilds from the copied items. assert fork.labels.get(FORK_CARRY_HISTORY_LABEL_KEY) == "1", ( f"cross-family fork must stamp carry-history for rebuild, got {fork.labels!r}" ) def test_fork_conversation_up_to_last_response_keeps_external_directive( conversation_store: SqlAlchemyConversationStore, agent_store: SqlAlchemyAgentStore, ) -> None: """Truncating at the LAST response is treated as a full fork. The copy is equivalent to a full fork, so the resume directive is kept — the runner can still clone the source's native transcript verbatim (full fidelity) instead of rebuilding from items. """ from omnigent.stores.conversation_store import FORK_SOURCE_EXTERNAL_SESSION_LABEL_KEY agent_store.create( agent_id="ag_fork_trunc_last", name="fork-trunc-last", bundle_location="ag_fork_trunc_last/fakehash", ) source = conversation_store.create_conversation(agent_id="ag_fork_trunc_last") conversation_store.set_external_session_id(source.id, "claude-uuid-last") _append_three_responses(conversation_store, source.id) fork = conversation_store.fork_conversation(source.id, up_to_response_id="resp_003") # All 6 items copied — the cutoff at the last response drops nothing. assert len(conversation_store.list_items(fork.id).data) == 6 assert fork.labels.get(FORK_SOURCE_EXTERNAL_SESSION_LABEL_KEY) == "claude-uuid-last", ( f"Cutoff at the last response is a full fork and must keep the resume " f"directive, got {fork.labels!r}" ) def test_fork_conversation_up_to_unknown_response_raises( conversation_store: SqlAlchemyConversationStore, agent_store: SqlAlchemyAgentStore, ) -> None: """An ``up_to_response_id`` matching no item raises ValueError. Silently copying the full history would fork far more context than the user selected (stale client state); the store must fail loud and create nothing. """ agent_store.create( agent_id="ag_fork_trunc_bad", name="fork-trunc-bad", bundle_location="ag_fork_trunc_bad/fakehash", ) source = conversation_store.create_conversation(agent_id="ag_fork_trunc_bad") _append_three_responses(conversation_store, source.id) with pytest.raises(ValueError, match="resp_nope"): conversation_store.fork_conversation(source.id, up_to_response_id="resp_nope") def test_fork_clone_agent_is_session_scoped( conversation_store: SqlAlchemyConversationStore, agent_store: SqlAlchemyAgentStore, ) -> None: """A fork that clones an agent creates a session-scoped row, not a built-in. The clone must be born with ``kind='session'`` so it never appears in the built-in agent list (``kind='template'``) that backs the fork picker — the regression that surfaced as duplicate "Claude Code" / "Codex" entries in the fork dialog. """ agent_store.create( agent_id="ag_fork_src", name="claude-native-ui", bundle_location="ag_fork_src/hash", ) source = conversation_store.create_conversation(agent_id="ag_fork_src") fork = conversation_store.fork_conversation( source.id, agent_id="ag_clone_ok", cloned_agent_name="claude-native-ui (fork ag_clone_o)", cloned_agent_bundle_location="ag_fork_src/hash", cloned_agent_description=None, ) assert fork.agent_id == "ag_clone_ok" cloned = agent_store.get("ag_clone_ok") assert cloned is not None assert cloned.session_id == fork.id, "clone must be bound to the fork session" # The clone is session-scoped, so it must NOT leak into the built-in # list (the source built-in is the only template-name row). builtin_ids = {a.id for a in agent_store.list(limit=100).data} assert "ag_clone_ok" not in builtin_ids assert "ag_fork_src" in builtin_ids def test_fork_clone_agent_failure_leaves_no_orphan( conversation_store: SqlAlchemyConversationStore, agent_store: SqlAlchemyAgentStore, ) -> None: """A failed clone-fork rolls the agent row back — no orphaned built-in. Pre-fix the route pre-created the clone in its own committed transaction, so a fork failure (here a stale ``up_to_response_id``) orphaned a ``session_id IS NULL`` row that polluted the built-in agent catalog. Creating the clone inside the fork transaction means the failure rolls it back too. """ agent_store.create( agent_id="ag_fork_src2", name="codex-native-ui", bundle_location="ag_fork_src2/hash", ) source = conversation_store.create_conversation(agent_id="ag_fork_src2") _append_three_responses(conversation_store, source.id) before = {a.id for a in agent_store.list(limit=100).data} with pytest.raises(ValueError, match="resp_nope"): conversation_store.fork_conversation( source.id, agent_id="ag_clone_orphan", cloned_agent_name="codex-native-ui (fork ag_clone_o)", cloned_agent_bundle_location="ag_fork_src2/hash", up_to_response_id="resp_nope", ) # The clone must not exist at all, and the built-in list is unchanged. assert agent_store.get("ag_clone_orphan") is None after = {a.id for a in agent_store.list(limit=100).data} assert after == before def test_instance_scoped_label_keys_match_harness_constants() -> None: """ The store's instance-scoped denylist matches the harness label keys. The store hard-codes the bridge-id literals (to avoid importing harness modules into the persistence layer). This guards against drift: if a harness renames its bridge-id label key, the literal in :data:`_INSTANCE_SCOPED_LABEL_KEYS` would silently stop matching and forks would re-inherit the source's bridge. Importing the real constants here makes that rename fail loudly at test time. """ from omnigent.claude_native_bridge import BRIDGE_ID_LABEL_KEY from omnigent.codex_native_bridge import CODEX_NATIVE_BRIDGE_ID_LABEL_KEY from omnigent.stores.conversation_store import _INSTANCE_SCOPED_LABEL_KEYS # Each harness's canonical bridge-id key must be in the denylist; a # miss means a rename slipped past the store's hard-coded literal. assert BRIDGE_ID_LABEL_KEY in _INSTANCE_SCOPED_LABEL_KEYS assert CODEX_NATIVE_BRIDGE_ID_LABEL_KEY in _INSTANCE_SCOPED_LABEL_KEYS def test_fork_conversation_copies_reasoning_effort( conversation_store: SqlAlchemyConversationStore, agent_store: SqlAlchemyAgentStore, ) -> None: """Fork inherits the source's reasoning_effort setting.""" agent_store.create( agent_id="ag_fork_re", name="fork-reasoning", bundle_location="ag_fork_re/fakehash", ) source = conversation_store.create_conversation(agent_id="ag_fork_re") conversation_store.update_conversation(source.id, reasoning_effort="high") fork = conversation_store.fork_conversation(source.id) # A wrong value means fork_conversation didn't copy the # reasoning_effort column — the fork would silently use the # default effort level instead of the source's setting. assert fork.reasoning_effort == "high", "Fork should inherit reasoning_effort from the source" def test_fork_conversation_copies_terminal_launch_args( conversation_store: SqlAlchemyConversationStore, agent_store: SqlAlchemyAgentStore, ) -> None: """Fork inherits the source's terminal_launch_args setting.""" agent_store.create( agent_id="ag_fork_tla", name="fork-tla", bundle_location="ag_fork_tla/fakehash", ) source = conversation_store.create_conversation(agent_id="ag_fork_tla") conversation_store.update_conversation( source.id, terminal_launch_args=["--dangerously-skip-permissions"], ) fork = conversation_store.fork_conversation(source.id) # A wrong value means fork_conversation didn't copy the # terminal_launch_args column — a forked native session would # silently lose its launch flags (consistent with how the fork # copies reasoning_effort / model_override). assert fork.terminal_launch_args == ["--dangerously-skip-permissions"], ( "Fork should inherit terminal_launch_args from the source" ) def test_fork_conversation_copy_model_settings_false_resets( conversation_store: SqlAlchemyConversationStore, agent_store: SqlAlchemyAgentStore, ) -> None: """``copy_model_settings=False`` drops the source's model settings. A model id is provider-bound, so a fork that switches to a different provider family must NOT inherit the source's ``model_override`` / ``reasoning_effort`` — they'd name a model the new provider can't serve. Both must reset to ``None`` (the bound agent's defaults), while the default (``True``) still copies them. """ agent_store.create( agent_id="ag_fork_cms", name="fork-cms", bundle_location="ag_fork_cms/fakehash", ) source = conversation_store.create_conversation(agent_id="ag_fork_cms") conversation_store.update_conversation( source.id, reasoning_effort="high", model_override="claude-opus-4" ) fork = conversation_store.fork_conversation(source.id, copy_model_settings=False) reloaded = conversation_store.get_conversation(fork.id) assert reloaded is not None # Both reset — a non-None value means the cross-family reset didn't # apply and the fork would launch pointing at an incompatible model. assert reloaded.model_override is None, ( "copy_model_settings=False must drop the source's model_override" ) assert reloaded.reasoning_effort is None, ( "copy_model_settings=False must drop the source's reasoning_effort" ) # Control: the default still copies, proving the reset is gated on the # flag and not a blanket drop. fork_default = conversation_store.fork_conversation(source.id) reloaded_default = conversation_store.get_conversation(fork_default.id) assert reloaded_default is not None assert reloaded_default.model_override == "claude-opus-4" assert reloaded_default.reasoning_effort == "high" def test_fork_conversation_carry_history_into_native_stamps_label( conversation_store: SqlAlchemyConversationStore, agent_store: SqlAlchemyAgentStore, ) -> None: """``carry_history_into_native=True`` stamps the carry-history directive. The runner reads ``omnigent.fork.carry_history`` to decide whether a native target rebuilds its transcript (vs launching fresh). The label must be set only when the flag is passed; the default leaves it off so a normal fork into a native target doesn't trigger a rebuild from the wrong items. """ from omnigent.stores.conversation_store import FORK_CARRY_HISTORY_LABEL_KEY agent_store.create( agent_id="ag_fork_carry", name="fork-carry", bundle_location="ag_fork_carry/fakehash", ) source = conversation_store.create_conversation(agent_id="ag_fork_carry") carried = conversation_store.fork_conversation(source.id, carry_history_into_native=True) assert carried.labels.get(FORK_CARRY_HISTORY_LABEL_KEY) == "1", ( f"carry_history_into_native=True must stamp the directive, got {carried.labels!r}" ) # Default (False): no directive, so a native target launches fresh. plain = conversation_store.fork_conversation(source.id) assert FORK_CARRY_HISTORY_LABEL_KEY not in plain.labels, ( f"Default fork must not stamp the carry-history directive, got {plain.labels!r}" ) def test_fork_conversation_agent_id_override( conversation_store: SqlAlchemyConversationStore, agent_store: SqlAlchemyAgentStore, ) -> None: """When agent_id is passed, the fork binds to the override instead of the source's agent. """ agent_store.create( agent_id="ag_original", name="original-agent", bundle_location="ag_original/fakehash", ) agent_store.create( agent_id="ag_cloned", name="cloned-agent", bundle_location="ag_original/fakehash", ) source = conversation_store.create_conversation(agent_id="ag_original") fork = conversation_store.fork_conversation( source.id, agent_id="ag_cloned", ) assert fork.agent_id == "ag_cloned", ( "Fork should use the overridden agent_id, not the source's" ) def test_switch_conversation_agent_cross_family_resets_and_relabels( conversation_store: SqlAlchemyConversationStore, agent_store: SqlAlchemyAgentStore, ) -> None: """In-place switch deletes the old agent, binds the new, and on a cross-family switch resets model settings, clears the native session id, and replaces the harness-presentation labels. """ from omnigent._wrapper_labels import ( CODEX_NATIVE_WRAPPER_VALUE, UI_MODE_LABEL_KEY, UI_MODE_TERMINAL_VALUE, WRAPPER_LABEL_KEY, ) from omnigent.stores.conversation_store import ( FORK_CARRY_HISTORY_LABEL_KEY, SWITCH_PREVIOUS_BUILTIN_LABEL_KEY, ) # An instance-scoped label (belongs to the running instance, dropped on a # switch). Uses a literal still in _INSTANCE_SCOPED_LABEL_KEYS — the old # omnigent.stopped marker was retired upstream. instance_label = "omnigent.last_context_tokens" # A real session binds a session-scoped agent (agent.session_id == conv). created = conversation_store.create_session_with_agent( agent_id="ag_switch_old", agent_name="claude (switch src)", agent_bundle_location="ag_switch_old/hash", agent_description="old", ) conv_id = created.conversation.id # Give the session model settings, a native session id, and labels that a # switch must touch (instance-scoped stopped marker + the old harness's # ui/wrapper) so we can assert they're handled correctly. conversation_store.update_conversation( conv_id, model_override="claude-opus-4-7", reasoning_effort="high" ) conversation_store.set_external_session_id(conv_id, "old-native-uuid") conversation_store.set_labels( conv_id, { instance_label: "1", # DANGEROUS codex bypass opt-in: in the instance-scoped set so a # switch (a new agent/harness context) drops it rather than # silently re-arming bypass without a fresh typed confirmation. "omnigent.codex_native.bypass_sandbox": "1", UI_MODE_LABEL_KEY: UI_MODE_TERMINAL_VALUE, WRAPPER_LABEL_KEY: "claude-code-native-ui", }, ) conversation_store.append( conv_id, [ NewConversationItem( type="message", response_id="resp_1", data=MessageData(role="user", content=[{"type": "input_text", "text": "hi"}]), ) ], ) target_labels = { UI_MODE_LABEL_KEY: UI_MODE_TERMINAL_VALUE, WRAPPER_LABEL_KEY: CODEX_NATIVE_WRAPPER_VALUE, } updated = conversation_store.switch_conversation_agent( conv_id, new_agent_id="ag_switch_new", new_agent_name="codex (switch new)", new_agent_bundle_location="ag_switch_new/hash", new_agent_description="new", copy_model_settings=False, # cross-family carry_history_into_native=True, # native target presentation_labels=target_labels, previous_builtin_id="ag_builtin_claude", ) # New agent bound; old session-scoped agent deleted (unique session_id # index would otherwise be violated by leaving both). assert updated.agent_id == "ag_switch_new" assert agent_store.get("ag_switch_old") is None, ( "old session-scoped agent must be deleted on switch" ) new_agent = agent_store.get("ag_switch_new") assert new_agent is not None and new_agent.session_id == conv_id, ( "new agent must be session-scoped to this conversation" ) # Cross-family → provider-bound model id is meaningless, so both reset. assert updated.model_override is None assert updated.reasoning_effort is None # Native runtime state belongs to the old harness → cleared so the next # turn cold-starts and rebuilds from items. assert updated.external_session_id is None # Labels: target ui/wrapper applied, carry-history + previous-builtin # stamped, and the old instance-scoped stopped marker dropped. assert updated.labels[UI_MODE_LABEL_KEY] == UI_MODE_TERMINAL_VALUE assert updated.labels[WRAPPER_LABEL_KEY] == CODEX_NATIVE_WRAPPER_VALUE assert updated.labels[FORK_CARRY_HISTORY_LABEL_KEY] == "1" assert updated.labels[SWITCH_PREVIOUS_BUILTIN_LABEL_KEY] == "ag_builtin_claude" assert instance_label not in updated.labels, "instance-scoped labels must not survive a switch" assert "omnigent.codex_native.bypass_sandbox" not in updated.labels, ( "the dangerous bypass opt-in must not survive a switch (re-confirm per context)" ) # Transcript is untouched (in place, not copied). assert len(conversation_store.list_items(conv_id).data) == 1 def test_switch_conversation_agent_same_family_keeps_model_settings( conversation_store: SqlAlchemyConversationStore, agent_store: SqlAlchemyAgentStore, ) -> None: """A same-family switch keeps model settings; an SDK target (empty presentation labels) drops the old ui/wrapper labels and does not stamp the carry-history directive. """ from omnigent._wrapper_labels import ( UI_MODE_LABEL_KEY, UI_MODE_TERMINAL_VALUE, WRAPPER_LABEL_KEY, ) from omnigent.stores.conversation_store import ( FORK_CARRY_HISTORY_LABEL_KEY, SWITCH_PREVIOUS_BUILTIN_LABEL_KEY, ) created = conversation_store.create_session_with_agent( agent_id="ag_switch_old2", agent_name="claude-native-ui", agent_bundle_location="ag_switch_old2/hash", agent_description=None, ) conv_id = created.conversation.id conversation_store.update_conversation( conv_id, model_override="claude-opus-4-7", reasoning_effort="high" ) conversation_store.set_labels( conv_id, { UI_MODE_LABEL_KEY: UI_MODE_TERMINAL_VALUE, WRAPPER_LABEL_KEY: "claude-code-native-ui", # A stale previous-builtin pointer from an earlier switch. SWITCH_PREVIOUS_BUILTIN_LABEL_KEY: "ag_stale_builtin", }, ) updated = conversation_store.switch_conversation_agent( conv_id, new_agent_id="ag_switch_new2", new_agent_name="claude (switch new)", new_agent_bundle_location="ag_switch_new2/hash", new_agent_description=None, copy_model_settings=True, # same family (anthropic native → sdk) carry_history_into_native=False, # SDK target rebuilds nothing presentation_labels={}, # SDK → chat mode (drop ui/wrapper) previous_builtin_id=None, ) # Same family → model settings carry over unchanged. assert updated.model_override == "claude-opus-4-7" assert updated.reasoning_effort == "high" # SDK target → terminal-first ui/wrapper labels removed (chat mode). assert UI_MODE_LABEL_KEY not in updated.labels assert WRAPPER_LABEL_KEY not in updated.labels # No native rebuild for an SDK target. assert FORK_CARRY_HISTORY_LABEL_KEY not in updated.labels # Stale previous-builtin pointer dropped (None passed → not re-stamped), # so a later "switch back" can't offer a wrong target. assert SWITCH_PREVIOUS_BUILTIN_LABEL_KEY not in updated.labels def test_get_session_connectivity_batches_runner_and_host( conversation_store: SqlAlchemyConversationStore, db_uri: str, ) -> None: """ ``get_session_connectivity`` returns runner/host per id. This is the bulk read powering the ``/health`` online-dot path: it replaced an N+1 fan-out of ``get_conversation`` (plus a labels query each). Liveness is now purely "is the tunnel up / is the host fresh" — the retired ``omnigent.stopped`` marker is no longer a field on the result. The test pins both binding fields the dot decision needs, across a mix of bindings in one call: - a runner-bound session reports its ``runner_id`` and no host; - a host-bound session reports its ``host_id`` and no runner; - an unknown id is absent from the result (callers treat that as reachable, matching the legacy single-row path). """ _register_host(db_uri, "host_conn") runner_bound = conversation_store.create_conversation(runner_id="runner_xyz") host_bound = conversation_store.create_conversation(host_id="host_conn", workspace="/tmp/ws") result = conversation_store.get_session_connectivity( [runner_bound.id, host_bound.id, "conv_unknown"] ) assert set(result) == {runner_bound.id, host_bound.id} assert result[runner_bound.id].runner_id == "runner_xyz" assert result[runner_bound.id].host_id is None assert result[host_bound.id].host_id == "host_conn" assert result[host_bound.id].runner_id is None # None of these are forks of a coding session, so needs_workspace is # off across the board — a True here would wrongly force the online # dot off for a normally-bound session. assert result[runner_bound.id].needs_workspace is False assert result[host_bound.id].needs_workspace is False def test_get_session_connectivity_reports_needs_workspace_for_fork( conversation_store: SqlAlchemyConversationStore, ) -> None: """ The fork-source label surfaces as ``needs_workspace=True``. A fork of a session that had a working directory carries the ``omnigent.fork.source_id`` label (set by ``fork_conversation``). ``get_session_connectivity`` must report ``needs_workspace=True`` for it — that flag is what makes ``_bulk_session_liveness`` mark the unbound clone offline so the UI prompts for a directory instead of dropping the first message. """ fork = conversation_store.create_conversation() conversation_store.set_labels(fork.id, {"omnigent.fork.source_id": "conv_src"}) plain = conversation_store.create_conversation() result = conversation_store.get_session_connectivity([fork.id, plain.id]) # Fork-source label present → needs_workspace on. A False here means # the label SELECT or the flag computation dropped the fork key. assert result[fork.id].needs_workspace is True # No label → in-process resumable, needs_workspace off. This is the # CUJ-2 chat-only fork path that must stay reachable. assert result[plain.id].needs_workspace is False def test_get_session_connectivity_empty_input_skips_query( conversation_store: SqlAlchemyConversationStore, ) -> None: """ ``get_session_connectivity([])`` returns ``{}`` without a query. The single-session ``/health`` variant and an empty sidebar both hit this; it must short-circuit rather than issue an ``IN ()``. """ assert conversation_store.get_session_connectivity([]) == {} # ── Per-user daily cost rollup ──────────────────────── def test_get_daily_cost_missing_returns_zero( conversation_store: SqlAlchemyConversationStore, ) -> None: """A (user, day) with no recorded spend reads as ``0.0``.""" assert conversation_store.get_daily_cost("alice@example.com", "2026-06-05") == 0.0 def test_add_daily_cost_accumulates( conversation_store: SqlAlchemyConversationStore, ) -> None: """Repeated adds for the same (user, day) sum into one total.""" conversation_store.add_daily_cost("alice@example.com", "2026-06-05", 1.25) conversation_store.add_daily_cost("alice@example.com", "2026-06-05", 0.75) conversation_store.add_daily_cost("alice@example.com", "2026-06-05", 2.00) # 1.25 + 0.75 + 2.00 = 4.00 proves each delta was added, not overwritten. assert conversation_store.get_daily_cost("alice@example.com", "2026-06-05") == pytest.approx( 4.00 ) def test_add_daily_cost_isolated_by_user_and_day( conversation_store: SqlAlchemyConversationStore, ) -> None: """Spend is partitioned by both user and UTC day; no cross-bleed.""" conversation_store.add_daily_cost("alice@example.com", "2026-06-05", 5.0) conversation_store.add_daily_cost("alice@example.com", "2026-06-06", 3.0) conversation_store.add_daily_cost("bob@example.com", "2026-06-05", 9.0) # Same user, different day: only that day's delta. assert conversation_store.get_daily_cost("alice@example.com", "2026-06-05") == pytest.approx( 5.0 ) assert conversation_store.get_daily_cost("alice@example.com", "2026-06-06") == pytest.approx( 3.0 ) # Different user, same day: isolated from alice. assert conversation_store.get_daily_cost("bob@example.com", "2026-06-05") == pytest.approx(9.0) def test_add_daily_cost_nonpositive_is_noop( conversation_store: SqlAlchemyConversationStore, ) -> None: """``delta <= 0`` never creates or mutates a row.""" conversation_store.add_daily_cost("alice@example.com", "2026-06-05", 0.0) conversation_store.add_daily_cost("alice@example.com", "2026-06-05", -1.0) assert conversation_store.get_daily_cost("alice@example.com", "2026-06-05") == 0.0 # A subsequent positive add still works and isn't polluted by the no-ops. conversation_store.add_daily_cost("alice@example.com", "2026-06-05", 2.5) assert conversation_store.get_daily_cost("alice@example.com", "2026-06-05") == pytest.approx( 2.5 ) def test_get_session_owner_returns_highest_level_grantee( conversation_store: SqlAlchemyConversationStore, db_uri: str, ) -> None: """The owner is the max-``level`` grantee, regardless of grant order.""" from omnigent.stores.permission_store.sqlalchemy_store import ( SqlAlchemyPermissionStore, ) conv = conversation_store.create_conversation() perms = SqlAlchemyPermissionStore(db_uri) for user in ("reader@example.com", "alice@example.com", "editor@example.com"): # session_permissions.user_id is an FK to users.id. perms.ensure_user(user) perms.grant("reader@example.com", conv.id, 1) # read perms.grant("alice@example.com", conv.id, 4) # owner (LEVEL_OWNER) perms.grant("editor@example.com", conv.id, 2) # edit # Owner (level 4) outranks the read/edit grants; if this returned a # lower-level grantee the ORDER BY level DESC ranking would be wrong. assert conversation_store.get_session_owner(conv.id) == "alice@example.com" def test_get_session_owner_none_when_no_grants( conversation_store: SqlAlchemyConversationStore, ) -> None: """A session with no permission grants (single-user mode) has no owner.""" conv = conversation_store.create_conversation() assert conversation_store.get_session_owner(conv.id) is None def test_get_session_owner_excludes_public_sentinel( conversation_store: SqlAlchemyConversationStore, db_uri: str, ) -> None: """A session with only a public grant (no real owner) returns None.""" from omnigent.server.auth import RESERVED_USER_PUBLIC from omnigent.stores.permission_store.sqlalchemy_store import ( SqlAlchemyPermissionStore, ) conv = conversation_store.create_conversation() perms = SqlAlchemyPermissionStore(db_uri) perms.ensure_user(RESERVED_USER_PUBLIC) perms.grant(RESERVED_USER_PUBLIC, conv.id, 1) # public read, no owner grant # Without the public-sentinel filter this would return "__public__" # (the only — hence highest-level — grant); the filter makes a # session with no real owner read as None instead. assert conversation_store.get_session_owner(conv.id) is None # ── Per-user daily cost: ask-approved state ─────────── def test_get_daily_cost_state_missing_returns_zeros( conversation_store: SqlAlchemyConversationStore, ) -> None: """A (user, day) with no row reads as zeros for both fields.""" state = conversation_store.get_daily_cost_state("alice@example.com", "2026-06-06") assert state == {"cost_usd": 0.0, "ask_approved_usd": 0.0} def test_set_daily_ask_approved_does_not_clobber_cost( conversation_store: SqlAlchemyConversationStore, ) -> None: """Recording an approved checkpoint leaves accumulated cost intact.""" conversation_store.add_daily_cost("alice@example.com", "2026-06-06", 3.0) conversation_store.set_daily_ask_approved("alice@example.com", "2026-06-06", 2.0) state = conversation_store.get_daily_cost_state("alice@example.com", "2026-06-06") # cost must survive the approval write (set touches only ask_approved_usd); # if it dropped to 0 the UPSERT wrongly overwrote cost_usd. assert state["cost_usd"] == pytest.approx(3.0) assert state["ask_approved_usd"] == pytest.approx(2.0) def test_add_daily_cost_does_not_clobber_ask_approved( conversation_store: SqlAlchemyConversationStore, ) -> None: """Accumulating cost after an approval leaves the approval intact.""" conversation_store.set_daily_ask_approved("alice@example.com", "2026-06-06", 2.0) conversation_store.add_daily_cost("alice@example.com", "2026-06-06", 1.5) state = conversation_store.get_daily_cost_state("alice@example.com", "2026-06-06") # The increment UPSERT must only touch cost_usd; ask_approved survives. assert state["cost_usd"] == pytest.approx(1.5) assert state["ask_approved_usd"] == pytest.approx(2.0) def test_set_daily_ask_approved_creates_row_when_absent( conversation_store: SqlAlchemyConversationStore, ) -> None: """Approving with no prior row inserts a cost=0 row carrying the approval.""" conversation_store.set_daily_ask_approved("alice@example.com", "2026-06-06", 0.05) state = conversation_store.get_daily_cost_state("alice@example.com", "2026-06-06") assert state["cost_usd"] == 0.0 assert state["ask_approved_usd"] == pytest.approx(0.05) def test_add_daily_cost_stacks_after_ask_approved( conversation_store: SqlAlchemyConversationStore, ) -> None: """Cost increments stack (not overwrite) even after an approval is set.""" conversation_store.set_daily_ask_approved("alice@example.com", "2026-06-06", 2.0) conversation_store.add_daily_cost("alice@example.com", "2026-06-06", 1.5) conversation_store.add_daily_cost("alice@example.com", "2026-06-06", 1.0) state = conversation_store.get_daily_cost_state("alice@example.com", "2026-06-06") # 1.5 + 1.0 proves the second add incremented rather than overwrote; # ask_approved untouched throughout. assert state["cost_usd"] == pytest.approx(2.5) assert state["ask_approved_usd"] == pytest.approx(2.0) # ── set_session_state ───────────────────────────────────────────────────── def test_set_session_state_persists( conversation_store: SqlAlchemyConversationStore, ) -> None: """set_session_state writes a JSON-serializable dict to the conversation.""" conv = conversation_store.create_conversation() state = {"cursor": 42, "flags": ["a", "b"]} conversation_store.set_session_state(conv.id, state) fetched = conversation_store.get_conversation(conv.id) assert fetched is not None assert fetched.session_state == state def test_set_session_state_overwrites( conversation_store: SqlAlchemyConversationStore, ) -> None: """set_session_state replaces the entire state dict.""" conv = conversation_store.create_conversation() conversation_store.set_session_state(conv.id, {"v": 1}) conversation_store.set_session_state(conv.id, {"v": 2, "new_key": True}) fetched = conversation_store.get_conversation(conv.id) assert fetched is not None assert fetched.session_state == {"v": 2, "new_key": True} def test_set_session_state_empty_dict( conversation_store: SqlAlchemyConversationStore, ) -> None: """set_session_state with empty dict clears state.""" conv = conversation_store.create_conversation() conversation_store.set_session_state(conv.id, {"old": True}) conversation_store.set_session_state(conv.id, {}) fetched = conversation_store.get_conversation(conv.id) assert fetched is not None assert fetched.session_state == {} # ── set_session_usage ───────────────────────────────────────────────────── def test_set_session_usage_persists( conversation_store: SqlAlchemyConversationStore, ) -> None: """set_session_usage writes token usage to the conversation.""" conv = conversation_store.create_conversation() usage = {"input_tokens": 1500, "output_tokens": 350, "total_tokens": 1850} conversation_store.set_session_usage(conv.id, usage) fetched = conversation_store.get_conversation(conv.id) assert fetched is not None assert fetched.session_usage == usage def test_set_session_usage_overwrites( conversation_store: SqlAlchemyConversationStore, ) -> None: """set_session_usage replaces the entire usage dict.""" conv = conversation_store.create_conversation() conversation_store.set_session_usage(conv.id, {"input_tokens": 100}) conversation_store.set_session_usage(conv.id, {"input_tokens": 200, "output_tokens": 50}) fetched = conversation_store.get_conversation(conv.id) assert fetched is not None assert fetched.session_usage == {"input_tokens": 200, "output_tokens": 50} # ── next_position counter (write-path MAX(position) scan removal) ────── def _user_message(text: str, response_id: str = "resp_pos") -> NewConversationItem: """A minimal user message item for position-counter tests.""" return NewConversationItem( type="message", response_id=response_id, data=MessageData(role="user", content=[{"type": "input_text", "text": text}]), ) def _stored_next_position( conversation_store: SqlAlchemyConversationStore, conversation_id: str ) -> int | None: """Read the raw ``conversations.next_position`` counter for assertions.""" from omnigent.db.db_models import SqlConversation with conversation_store._session() as session: row = session.get(SqlConversation, (0, conversation_id)) assert row is not None return row.next_position def _stored_positions( conversation_store: SqlAlchemyConversationStore, conversation_id: str ) -> list[int]: """Raw item positions for a conversation, ascending — the source of truth ``list_items`` (which hides ``position``) cannot assert on.""" from sqlalchemy import select from omnigent.db.db_models import SqlConversationItem with conversation_store._session() as session: return sorted( session.execute( select(SqlConversationItem.position).where( SqlConversationItem.conversation_id == conversation_id ) ) .scalars() .all() ) def test_new_conversation_seeds_next_position_zero( conversation_store: SqlAlchemyConversationStore, ) -> None: """A freshly created conversation starts its position allocator at 0, so the first append reads the counter rather than scanning MAX(position).""" conv = conversation_store.create_conversation() assert _stored_next_position(conversation_store, conv.id) == 0 @pytest.mark.parametrize("batch_sizes", [[1], [1, 1, 1], [3], [2, 1, 4]]) def test_append_allocates_dense_positions_and_advances_counter( conversation_store: SqlAlchemyConversationStore, batch_sizes: list[int], ) -> None: """append() assigns contiguous positions from next_position and advances the counter by the batch size, so the stored counter always equals the total items appended — across single- and multi-item batches. Real store, real SQLite; asserts on the raw position column and counter, no mocks. """ conv = conversation_store.create_conversation() total = 0 for batch in batch_sizes: conversation_store.append(conv.id, [_user_message(f"m{total + i}") for i in range(batch)]) total += batch assert _stored_positions(conversation_store, conv.id) == list(range(total)) # The counter points one past the last item — the next position to hand out. assert _stored_next_position(conversation_store, conv.id) == total def test_append_reads_counter_not_max_scan( conversation_store: SqlAlchemyConversationStore, ) -> None: """append() allocates from the maintained counter, not a MAX(position) scan: advancing the counter past the real max makes the next item land at the counter value, which a scan-based implementation could never produce. """ from omnigent.db.db_models import SqlConversation conv = conversation_store.create_conversation() conversation_store.append(conv.id, [_user_message("a"), _user_message("b")]) # Real max position is 1; jump the counter ahead to 100. with conversation_store._session() as session: session.get(SqlConversation, (0, conv.id)).next_position = 100 conversation_store.append(conv.id, [_user_message("c")]) # Position 100 (counter), not 2 (max + 1) — proves the scan path is unused. assert _stored_positions(conversation_store, conv.id) == [0, 1, 100] assert _stored_next_position(conversation_store, conv.id) == 101 @pytest.mark.parametrize("preexisting", [0, 1, 3]) def test_append_falls_back_to_scan_when_counter_null( conversation_store: SqlAlchemyConversationStore, preexisting: int, ) -> None: """A conversation written before the counter existed has next_position = NULL. The next append falls back to a one-time MAX(position) scan to place items correctly, then persists the advanced counter so subsequent appends are scan-free. """ from omnigent.db.db_models import SqlConversation conv = conversation_store.create_conversation() if preexisting: conversation_store.append(conv.id, [_user_message(f"pre{i}") for i in range(preexisting)]) # Simulate a pre-counter row: clear the maintained counter. with conversation_store._session() as session: session.get(SqlConversation, (0, conv.id)).next_position = None assert _stored_next_position(conversation_store, conv.id) is None conversation_store.append(conv.id, [_user_message("new")]) # The fallback scan placed the new item right after the existing max... assert _stored_positions(conversation_store, conv.id) == list(range(preexisting + 1)) # ...and the counter is now backfilled, so the next append won't scan. assert _stored_next_position(conversation_store, conv.id) == preexisting + 1 def test_fork_seeds_next_position_from_copied_items( conversation_store: SqlAlchemyConversationStore, agent_store: SqlAlchemyAgentStore, ) -> None: """A full fork seeds the clone's allocator from the number of copied items, so the clone's first append is scan-free and collision-free.""" agent_store.create(agent_id="ag_fork_pos", name="fork-pos", bundle_location="ag_fork_pos/h") source = conversation_store.create_conversation(agent_id="ag_fork_pos") conversation_store.append( source.id, [_user_message(f"s{i}", response_id="resp_1") for i in range(3)] ) fork = conversation_store.fork_conversation(source.id, title="fork") # 3 items copied (dense positions 0..2) → allocator starts at 3. assert _stored_next_position(conversation_store, fork.id) == 3 conversation_store.append(fork.id, [_user_message("after")]) assert _stored_positions(conversation_store, fork.id) == [0, 1, 2, 3] def test_truncated_fork_seeds_next_position_from_copied_items( conversation_store: SqlAlchemyConversationStore, agent_store: SqlAlchemyAgentStore, ) -> None: """A truncated fork seeds the allocator from the count of the *copied* items, not the source length, so the shorter clone stays collision-free.""" agent_store.create( agent_id="ag_fork_trunc", name="fork-trunc", bundle_location="ag_fork_trunc/h" ) source = conversation_store.create_conversation(agent_id="ag_fork_trunc") conversation_store.append( source.id, [_user_message("a", "resp_1"), _user_message("b", "resp_1")], ) conversation_store.append( source.id, [_user_message("c", "resp_2"), _user_message("d", "resp_2")], ) fork = conversation_store.fork_conversation(source.id, up_to_response_id="resp_1") # Only resp_1's 2 items are copied → allocator starts at 2. assert _stored_positions(conversation_store, fork.id) == [0, 1] assert _stored_next_position(conversation_store, fork.id) == 2 conversation_store.append(fork.id, [_user_message("after")]) assert _stored_positions(conversation_store, fork.id) == [0, 1, 2] def test_append_many_batches_stay_contiguous( conversation_store: SqlAlchemyConversationStore, ) -> None: """End-to-end: many sequential appends produce a contiguous, gap-free position sequence and a counter equal to the item count — the invariant the maintained allocator must preserve across a long session (the scan-per-write pattern this replaces grew with that length).""" conv = conversation_store.create_conversation() total = 0 for turn in range(25): conversation_store.append( conv.id, [_user_message(f"t{turn}-{i}", response_id=f"resp_{turn}") for i in range(3)], ) total += 3 assert _stored_positions(conversation_store, conv.id) == list(range(total)) assert _stored_next_position(conversation_store, conv.id) == total listed = conversation_store.list_items(conv.id, limit=total) assert len(listed.data) == total # ── Projects (conversation_labels key="omni_project") ─────── def test_list_projects_returns_distinct_names_sorted( conversation_store: SqlAlchemyConversationStore, ) -> None: """``list_projects`` returns each distinct project name once, ordered alphabetically. Sessions with no project label don't create phantom projects, and a project shared by two sessions appears a single time.""" a1 = conversation_store.create_conversation() a2 = conversation_store.create_conversation() b1 = conversation_store.create_conversation() conversation_store.create_conversation() # unfiled — must not appear conversation_store.set_labels(a1.id, {"omni_project": "Sprint 42"}) conversation_store.set_labels(a2.id, {"omni_project": "Sprint 42"}) conversation_store.set_labels(b1.id, {"omni_project": "Customer X"}) # Alphabetical, de-duplicated. A missing DISTINCT would list "Sprint 42" # twice. assert conversation_store.list_projects() == ["Customer X", "Sprint 42"] def test_list_projects_empty_when_no_project_labels( conversation_store: SqlAlchemyConversationStore, ) -> None: """Non-project labels (e.g. guardrail keys) never surface as projects.""" conv = conversation_store.create_conversation() conversation_store.set_labels(conv.id, {"integrity": "1", "sensitivity": "public"}) assert conversation_store.list_projects() == [] def test_list_projects_excludes_all_archived_projects( conversation_store: SqlAlchemyConversationStore, ) -> None: """A project whose every member is archived drops out of the list (this is what makes "Delete project" — which archives all members — remove the folder), while the label is preserved so unarchiving restores it. A project with a mix of archived and active members still appears.""" solo = conversation_store.create_conversation() mix_archived = conversation_store.create_conversation() mix_active = conversation_store.create_conversation() conversation_store.set_labels(solo.id, {"omni_project": "Gone"}) conversation_store.set_labels(mix_archived.id, {"omni_project": "Mixed"}) conversation_store.set_labels(mix_active.id, {"omni_project": "Mixed"}) # "Gone" has one member; archiving it empties the project. "Mixed" keeps a # live member, so it stays. conversation_store.update_conversation(solo.id, archived=True) conversation_store.update_conversation(mix_archived.id, archived=True) assert conversation_store.list_projects() == ["Mixed"] # Unarchiving the lone member brings its project back — the label was kept. conversation_store.update_conversation(solo.id, archived=False) assert conversation_store.list_projects() == ["Gone", "Mixed"] def test_list_projects_scoped_by_accessible_by( conversation_store: SqlAlchemyConversationStore, db_uri: str, ) -> None: """When ``accessible_by`` is set, only projects on sessions the user has a permission row for are returned — mirroring the list_conversations ACL.""" from omnigent.stores.permission_store.sqlalchemy_store import ( SqlAlchemyPermissionStore, ) mine = conversation_store.create_conversation() theirs = conversation_store.create_conversation() conversation_store.set_labels(mine.id, {"omni_project": "Mine"}) conversation_store.set_labels(theirs.id, {"omni_project": "Theirs"}) perms = SqlAlchemyPermissionStore(db_uri) for user in ("alice@example.com", "bob@example.com"): perms.ensure_user(user) perms.grant("alice@example.com", mine.id, 4) perms.grant("bob@example.com", theirs.id, 4) # Alice only sees her project; Theirs is invisible to her. assert conversation_store.list_projects(accessible_by="alice@example.com") == ["Mine"] def test_delete_label_removes_only_target_key( conversation_store: SqlAlchemyConversationStore, ) -> None: """``delete_label`` drops the named key and leaves siblings intact — so removing a session from its project doesn't wipe guardrail labels.""" conv = conversation_store.create_conversation() conversation_store.set_labels(conv.id, {"omni_project": "X", "integrity": "1"}) conversation_store.delete_label(conv.id, "omni_project") got = conversation_store.get_conversation(conv.id) assert got is not None assert got.labels == {"integrity": "1"} def test_delete_label_is_noop_when_absent( conversation_store: SqlAlchemyConversationStore, ) -> None: """Deleting a label that doesn't exist is a no-op, not an error.""" conv = conversation_store.create_conversation() conversation_store.delete_label(conv.id, "omni_project") # must not raise got = conversation_store.get_conversation(conv.id) assert got is not None assert got.labels == {} def test_list_conversations_filters_by_project( conversation_store: SqlAlchemyConversationStore, ) -> None: """``project="X"`` returns only sessions carrying that exact project label.""" filed = conversation_store.create_conversation() other = conversation_store.create_conversation() conversation_store.create_conversation() # unfiled conversation_store.set_labels(filed.id, {"omni_project": "X"}) conversation_store.set_labels(other.id, {"omni_project": "Y"}) ids = {c.id for c in conversation_store.list_conversations(project="X").data} assert ids == {filed.id} def test_list_conversations_empty_project_returns_unfiled( conversation_store: SqlAlchemyConversationStore, ) -> None: """``project=""`` returns only sessions with NO project label (Unfiled).""" filed = conversation_store.create_conversation() unfiled = conversation_store.create_conversation() conversation_store.set_labels(filed.id, {"omni_project": "X"}) ids = {c.id for c in conversation_store.list_conversations(project="").data} assert unfiled.id in ids assert filed.id not in ids def test_list_conversations_project_none_disables_filter( conversation_store: SqlAlchemyConversationStore, ) -> None: """``project=None`` (the default) returns filed and unfiled alike.""" filed = conversation_store.create_conversation() unfiled = conversation_store.create_conversation() conversation_store.set_labels(filed.id, {"omni_project": "X"}) ids = {c.id for c in conversation_store.list_conversations().data} assert ids >= {filed.id, unfiled.id} def test_list_projects_owned_by_excludes_shared_only_projects( conversation_store: SqlAlchemyConversationStore, db_uri: str, ) -> None: """``owned_by`` restricts to projects the user OWNS, not ones merely shared with them — so a project whose sessions are only shared to the user (owned by someone else) does not surface as one of their own sidebar folders.""" from omnigent.stores.permission_store.sqlalchemy_store import ( SqlAlchemyPermissionStore, ) mine = conversation_store.create_conversation() shared = conversation_store.create_conversation() conversation_store.set_labels(mine.id, {"omni_project": "Mine"}) conversation_store.set_labels(shared.id, {"omni_project": "Shared"}) perms = SqlAlchemyPermissionStore(db_uri) for user in ("alice@example.com", "bob@example.com"): perms.ensure_user(user) # Bob owns both; Alice only gets a read (level 1) grant on the shared one. perms.grant("bob@example.com", mine.id, 4) perms.grant("alice@example.com", mine.id, 4) perms.grant("bob@example.com", shared.id, 4) perms.grant("alice@example.com", shared.id, 1) # accessible_by would leak "Shared" — Alice can access it. owned_by must not. assert conversation_store.list_projects(accessible_by="alice@example.com") == [ "Mine", "Shared", ] assert conversation_store.list_projects(owned_by="alice@example.com") == ["Mine"] def test_list_conversations_owned_by_excludes_shared_sessions( conversation_store: SqlAlchemyConversationStore, db_uri: str, ) -> None: """``owned_by`` on a project filter returns only sessions the user owns; a session shared with them (read grant) under the same project is excluded so it stays out of the owner-only project folder.""" from omnigent.stores.permission_store.sqlalchemy_store import ( SqlAlchemyPermissionStore, ) mine = conversation_store.create_conversation() shared = conversation_store.create_conversation() conversation_store.set_labels(mine.id, {"omni_project": "X"}) conversation_store.set_labels(shared.id, {"omni_project": "X"}) perms = SqlAlchemyPermissionStore(db_uri) for user in ("alice@example.com", "bob@example.com"): perms.ensure_user(user) perms.grant("alice@example.com", mine.id, 4) perms.grant("bob@example.com", shared.id, 4) perms.grant("alice@example.com", shared.id, 1) ids = { c.id for c in conversation_store.list_conversations( project="X", owned_by="alice@example.com" ).data } assert ids == {mine.id}