diff --git a/api/oss/src/apis/fastapi/sessions/models.py b/api/oss/src/apis/fastapi/sessions/models.py index 3a1da94dde..2ac7817168 100644 --- a/api/oss/src/apis/fastapi/sessions/models.py +++ b/api/oss/src/apis/fastapi/sessions/models.py @@ -4,6 +4,7 @@ from pydantic import BaseModel, ConfigDict, Field, model_validator +from oss.src.core.sessions.dtos import SessionListItem from oss.src.core.sessions.streams.dtos import ( SessionStream, ) @@ -33,11 +34,15 @@ class SessionQueryRequest(BaseModel): include_ended: bool = False # Include archived sessions — off by default (archive hides); on for the archived view. include_archived: bool = False + # Case-insensitive substring match over the session title (`session_streams.name`). + search: Optional[str] = None class SessionsResponse(BaseModel): count: int = 0 - sessions: List[SessionStream] = Field(default_factory=list) + # `SessionListItem` = `SessionStream` + the latest turn's `references` (WP0-R3), + # absent (excluded by response_model_exclude_none) when the session has no turns yet. + sessions: List[SessionListItem] = Field(default_factory=list) class SessionResponse(BaseModel): diff --git a/api/oss/src/apis/fastapi/sessions/router.py b/api/oss/src/apis/fastapi/sessions/router.py index 4968dcae55..5792e8c550 100644 --- a/api/oss/src/apis/fastapi/sessions/router.py +++ b/api/oss/src/apis/fastapi/sessions/router.py @@ -1562,6 +1562,7 @@ async def query_sessions( references=body.references, include_ended=body.include_ended, include_archived=body.include_archived, + search=body.search, ), windowing=body.windowing, ) diff --git a/api/oss/src/core/sessions/dtos.py b/api/oss/src/core/sessions/dtos.py index 8c413a1dd7..2c12c6eb00 100644 --- a/api/oss/src/core/sessions/dtos.py +++ b/api/oss/src/core/sessions/dtos.py @@ -2,9 +2,20 @@ from pydantic import BaseModel +from oss.src.core.sessions.streams.dtos import SessionStream from oss.src.core.shared.dtos import Reference +class SessionListItem(SessionStream): + """A `/sessions/query` row, enriched at READ time with the session's HIGHEST + `turn_index` turn's `references` — the agent/workflow that produced the latest turn. + + Hydrated by `SessionsService.query_sessions` via a batch turns lookup; never + denormalized onto `session_streams` (see that method's docstring).""" + + references: Optional[List[Reference]] = None + + class SessionQuery(BaseModel): """Root `/sessions/query` filter: reference-scoped, joined through the turns' references (WP1's GIN `.contains()`), not denormalized onto the stream row.""" @@ -15,3 +26,5 @@ class SessionQuery(BaseModel): include_ended: bool = False # Include archived sessions — off by default (archive hides); on for the archived view. include_archived: bool = False + # Case-insensitive substring match over the session title (`session_streams.name`). + search: Optional[str] = None diff --git a/api/oss/src/core/sessions/service.py b/api/oss/src/core/sessions/service.py index a161bf8203..952f63fcbb 100644 --- a/api/oss/src/core/sessions/service.py +++ b/api/oss/src/core/sessions/service.py @@ -15,7 +15,7 @@ from uuid import UUID from oss.src.core.shared.dtos import Reference, Windowing -from oss.src.core.sessions.dtos import SessionQuery +from oss.src.core.sessions.dtos import SessionListItem, SessionQuery from oss.src.core.sessions.streams.dtos import SessionStream, SessionStreamQuery from oss.src.core.sessions.streams.service import SessionStreamsService from oss.src.core.sessions.turns.dtos import SessionTurnQuery @@ -45,7 +45,7 @@ async def query_sessions( # query: Optional[SessionQuery] = None, windowing: Optional[Windowing] = None, - ) -> List[SessionStream]: + ) -> List[SessionListItem]: """List/filter sessions, newest -> oldest, windowed. Reads the merged stream rows; when `references` is set, first joins the @@ -53,6 +53,10 @@ async def query_sessions( `session_id`s, then filters the stream query to that set. No denormalization onto the stream row (B3) — revisit only if the join proves hot. + + Each row is enriched (READ-time only, see `SessionListItem`) with its latest + turn's `references` via a single batch lookup keyed on every listed + `session_id` — never one `latest_turn` call per row (WP0-R3). """ session_ids: Optional[List[str]] = None @@ -66,16 +70,36 @@ async def query_sessions( if not session_ids: return [] - return await self.streams_service.query_streams( + streams = await self.streams_service.query_streams( project_id=project_id, filter=SessionStreamQuery( include_ended=bool(query and query.include_ended), include_archived=bool(query and query.include_archived), + search=query.search if query else None, ), windowing=windowing, session_ids=session_ids, ) + if not streams: + return [] + + latest_turns = await self.turns_service.latest_turn_per_session( + project_id=project_id, + session_ids=[stream.session_id for stream in streams], + ) + + items = [] + for stream in streams: + turn = latest_turns.get(stream.session_id) + items.append( + SessionListItem( + **stream.model_dump(), + references=turn.references if turn else None, + ) + ) + return items + async def delete_session( self, *, diff --git a/api/oss/src/core/sessions/streams/dtos.py b/api/oss/src/core/sessions/streams/dtos.py index f58506765f..6d69beb36c 100644 --- a/api/oss/src/core/sessions/streams/dtos.py +++ b/api/oss/src/core/sessions/streams/dtos.py @@ -71,6 +71,8 @@ class SessionStreamQuery(BaseModel): # Include archived (deliberately-hidden) rows — off by default so archive hides; on for the # archived view. Orthogonal to `include_ended` (a row can be killed OR archived). include_archived: bool = False + # Case-insensitive substring match over `name` (the session title). + search: Optional[str] = None class CommandMode(str, Enum): diff --git a/api/oss/src/core/sessions/turns/interfaces.py b/api/oss/src/core/sessions/turns/interfaces.py index 30f4b20fc6..647f263703 100644 --- a/api/oss/src/core/sessions/turns/interfaces.py +++ b/api/oss/src/core/sessions/turns/interfaces.py @@ -1,5 +1,5 @@ from abc import ABC, abstractmethod -from typing import List, Optional +from typing import Dict, List, Optional from uuid import UUID from oss.src.core.shared.dtos import Windowing @@ -68,6 +68,14 @@ async def latest_turn_per_harness_kind( harness_kind: HarnessKind, ) -> Optional[SessionTurn]: ... + @abstractmethod + async def latest_turn_per_session( + self, + *, + project_id: UUID, + session_ids: List[str], + ) -> Dict[str, SessionTurn]: ... + @abstractmethod async def delete_by_session_id( self, diff --git a/api/oss/src/core/sessions/turns/service.py b/api/oss/src/core/sessions/turns/service.py index 068d219911..b42bf2e7e4 100644 --- a/api/oss/src/core/sessions/turns/service.py +++ b/api/oss/src/core/sessions/turns/service.py @@ -5,7 +5,7 @@ turn_index DESC LIMIT 1). """ -from typing import List, Optional +from typing import Dict, List, Optional from uuid import UUID from oss.src.core.shared.dtos import Windowing @@ -109,6 +109,19 @@ async def latest_turn_per_harness_kind( harness_kind=harness_kind, ) + async def latest_turn_per_session( + self, + *, + project_id: UUID, + session_ids: List[str], + ) -> Dict[str, SessionTurn]: + """Batch resume-read across many sessions — one query, not N `latest_turn` calls + (WP0-R3, the `/sessions/query` list-row `references` hydration).""" + return await self._dao.latest_turn_per_session( + project_id=project_id, + session_ids=session_ids, + ) + async def delete_by_session_id( self, *, diff --git a/api/oss/src/dbs/postgres/sessions/streams/dao.py b/api/oss/src/dbs/postgres/sessions/streams/dao.py index 6d8a7f43b6..aa5083a813 100644 --- a/api/oss/src/dbs/postgres/sessions/streams/dao.py +++ b/api/oss/src/dbs/postgres/sessions/streams/dao.py @@ -141,16 +141,42 @@ async def query( ) if flags_filter: stmt = stmt.where(SessionStreamDBE.flags.contains(flags_filter)) + term = filter.search.strip() if filter.search else "" + if term: + # Escape LIKE metacharacters so a literal `%`/`_` in the search term + # doesn't act as a wildcard. + escaped = ( + term.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") + ) + stmt = stmt.where( + SessionStreamDBE.name.ilike(f"%{escaped}%", escape="\\") + ) if windowing: stmt = apply_windowing( stmt=stmt, DBE=SessionStreamDBE, - attribute="id", + # Last-activity ordering: updated_at is fed by heartbeat/edit/archive, + # so a resumed session bumps to the top instead of sorting by its + # original (uuid7) creation time. The coalesce(updated_at, created_at) + # expression can't use the (project_id, created_at)/archived_at indexes + # (no expression index for it), but per-project session-list sizes keep + # the in-memory sort acceptable — a deliberate choice, not an oversight. + attribute="updated_at", order="descending", windowing=windowing, ) else: - stmt = stmt.order_by(SessionStreamDBE.created_at.desc()) + # No windowing here means this is the liveness-index caller + # (`query_session_streams`), not the paginated session list — ordering + # isn't load-bearing there, but keep it consistent with the windowed path, + # coalescing onto created_at for rows never touched since creation (same + # rationale as `apply_windowing`'s updated_at branch). + stmt = stmt.order_by( + func.coalesce( + SessionStreamDBE.updated_at, SessionStreamDBE.created_at + ).desc(), + SessionStreamDBE.id.desc(), + ) result = await session.execute(stmt) dbes = result.scalars().all() return [map_stream_dbe_to_dto(stream_dbe=dbe) for dbe in dbes] diff --git a/api/oss/src/dbs/postgres/sessions/turns/dao.py b/api/oss/src/dbs/postgres/sessions/turns/dao.py index d5d88c7a9b..44b6c69d80 100644 --- a/api/oss/src/dbs/postgres/sessions/turns/dao.py +++ b/api/oss/src/dbs/postgres/sessions/turns/dao.py @@ -1,4 +1,4 @@ -from typing import List, Optional +from typing import Dict, List, Optional from uuid import UUID from sqlalchemy import and_, delete as sa_delete, or_, select @@ -273,6 +273,35 @@ async def latest_turn_per_harness_kind( return None return map_turn_dbe_to_dto(turn_dbe=dbe) + async def latest_turn_per_session( + self, + *, + project_id: UUID, + session_ids: List[str], + ) -> Dict[str, SessionTurn]: + """Batch resume-read: one row per `session_id`, the highest `turn_index` (WP0-R3 + list-row hydration — a single query instead of N `latest_turn` calls). The IN-list + is bounded by the caller's page (windowing limit); it is unbounded only for the + currently-unpaginated OSS `/sessions` list.""" + if not session_ids: + return {} + async with self.engine.session() as session: + stmt = ( + select(SessionTurnDBE) + .distinct(SessionTurnDBE.session_id) + .where( + SessionTurnDBE.project_id == project_id, + SessionTurnDBE.session_id.in_(session_ids), + ) + .order_by( + SessionTurnDBE.session_id, + SessionTurnDBE.turn_index.desc(), + ) + ) + result = await session.execute(stmt) + dbes = result.scalars().all() + return {dbe.session_id: map_turn_dbe_to_dto(turn_dbe=dbe) for dbe in dbes} + async def delete_by_session_id( self, *, diff --git a/api/oss/src/dbs/postgres/shared/utils.py b/api/oss/src/dbs/postgres/shared/utils.py index 2ebc5f4bd8..3e4dc61cc2 100644 --- a/api/oss/src/dbs/postgres/shared/utils.py +++ b/api/oss/src/dbs/postgres/shared/utils.py @@ -1,4 +1,4 @@ -from sqlalchemy import Select, and_, or_ +from sqlalchemy import Select, and_, func, or_ from oss.src.core.shared.dtos import Windowing @@ -17,16 +17,30 @@ def apply_windowing( id_attribute = span_id_attribute or entity_id_attribute or None created_at_attribute = DBE.created_at if getattr(DBE, "created_at", None) else None # type: ignore start_time_attribute = DBE.start_time if getattr(DBE, "start_time", None) else None # type: ignore + updated_at_attribute = DBE.updated_at if getattr(DBE, "updated_at", None) else None # type: ignore + # updated_at is nullable (never touched since row creation) — coalesce onto created_at + # so "last activity" degrades to "creation time" instead of sorting a never-touched row + # first under `DESC` (Postgres puts NULLs first). Mirrors the FE's `activity()` helper + # (updated_at ?? created_at). + if updated_at_attribute is not None and created_at_attribute is not None: + updated_at_attribute = func.coalesce(updated_at_attribute, created_at_attribute) + # updated_at rides its own cursor (last-activity ordering); default time_attribute + # stays start_time/created_at so unrelated callers are unaffected. time_attribute = start_time_attribute or created_at_attribute or None + if attribute.lower() == "updated_at" and updated_at_attribute is not None: + time_attribute = updated_at_attribute # UUID7 -> id ---------------------------------------------------- # order_attribute = { "id": id_attribute, "span_id": span_id_attribute, "created_at": created_at_attribute, "start_time": start_time_attribute, + "updated_at": updated_at_attribute, }.get(attribute.lower(), created_at_attribute) - if not order_attribute or not time_attribute or not id_attribute: + # `order_attribute`/`time_attribute` may be a `func.coalesce(...)` expression (no + # truthy `__bool__`) rather than a plain column — compare against `None` explicitly. + if order_attribute is None or time_attribute is None or id_attribute is None: return stmt # ---------------------------------------------------------------- # ascending_order = order_attribute.asc() # type: ignore @@ -93,7 +107,15 @@ def apply_windowing( if order_attribute is id_attribute: stmt = stmt.order_by(windowing_order) else: - stmt = stmt.order_by(windowing_order, id_attribute) + # Tiebreak direction must match the cursor predicate's direction (`id <` on the + # descending branch, `id >` on ascending) — an ASC tiebreak under a DESC cursor + # splits a tie group across the page boundary (duplicate/skip rows). + id_tiebreak = ( + id_attribute.desc() + if windowing_order == descending_order + else id_attribute.asc() # type: ignore + ) + stmt = stmt.order_by(windowing_order, id_tiebreak) if windowing.limit: stmt = stmt.limit(windowing.limit) diff --git a/api/oss/tests/pytest/unit/sessions/test_query_sessions_references.py b/api/oss/tests/pytest/unit/sessions/test_query_sessions_references.py new file mode 100644 index 0000000000..c1207dc982 --- /dev/null +++ b/api/oss/tests/pytest/unit/sessions/test_query_sessions_references.py @@ -0,0 +1,277 @@ +"""Unit tests for WP0-R3: echo each session's latest-turn `references` on +`/sessions/query` rows. + +(a) Service-level: fake streams + fake turns services, mirroring +`test_sessions_root_service.py`'s fake pattern. Asserts each row carries the +HIGHEST `turn_index` turn's `references` (or `None` when the session has no +turns), and that the turns lookup batches — ONE call across every listed +`session_id`, never one `latest_turn` call per row. +(b) DAO-level: statement-compilation only (no DB), mirroring +`test_query_sessions_search.py`'s dummy-engine monkeypatch pattern. Asserts +`latest_turn_per_session` compiles to `DISTINCT ON (session_turns.session_id)` +ordered by `session_id, turn_index DESC`. +""" + +from typing import Dict, List, Optional +from uuid import uuid4 + +import pytest +from sqlalchemy.dialects import postgresql + +from oss.src.core.sessions.dtos import SessionListItem +from oss.src.core.sessions.service import SessionsService +from oss.src.core.sessions.streams.dtos import SessionStream +from oss.src.core.sessions.turns.dtos import HarnessKind, SessionTurn +from oss.src.core.shared.dtos import Reference + + +_PROJECT = uuid4() + + +def _stream(session_id: str) -> SessionStream: + return SessionStream(id=uuid4(), project_id=_PROJECT, session_id=session_id) + + +def _turn( + session_id: str, + turn_index: int, + references: Optional[List[Reference]] = None, +) -> SessionTurn: + return SessionTurn( + id=uuid4(), + project_id=_PROJECT, + session_id=session_id, + stream_id=uuid4(), + turn_index=turn_index, + harness_kind=HarnessKind.PI, + references=references, + ) + + +# ---------------------------------------------------------------------------------- # +# (a) Service — batch hydration +# ---------------------------------------------------------------------------------- # + + +class _FakeStreamsService: + def __init__(self, rows: List[SessionStream]): + self.rows = rows + + async def query_streams( + self, *, project_id, filter, windowing=None, session_ids=None + ): + if session_ids is not None: + return [s for s in self.rows if s.session_id in session_ids] + return list(self.rows) + + +class _FakeTurnsService: + """`turns` maps session_id -> every turn for that session (any order); the fake + computes the highest-turn_index row itself, mirroring the real DAO's + `DISTINCT ON (session_id) ORDER BY session_id, turn_index DESC` semantics.""" + + def __init__(self, turns_by_session: Dict[str, List[SessionTurn]]): + self.turns_by_session = turns_by_session + self.latest_turn_per_session_calls: list[dict] = [] + + async def query_turns(self, *, project_id, query=None, windowing=None): + return [] + + async def latest_turn_per_session( + self, *, project_id, session_ids: List[str] + ) -> Dict[str, SessionTurn]: + self.latest_turn_per_session_calls.append( + {"project_id": project_id, "session_ids": list(session_ids)} + ) + result: Dict[str, SessionTurn] = {} + for session_id in session_ids: + turns = self.turns_by_session.get(session_id) or [] + if not turns: + continue + result[session_id] = max(turns, key=lambda t: t.turn_index) + return result + + +class _FakeInteractionsService: + pass + + +class _FakeMountsService: + pass + + +def _service( + *, streams: List[SessionStream], turns_by_session: Dict[str, List[SessionTurn]] +): + streams_svc = _FakeStreamsService(rows=streams) + turns_svc = _FakeTurnsService(turns_by_session=turns_by_session) + svc = SessionsService( + streams_service=streams_svc, + turns_service=turns_svc, + interactions_service=_FakeInteractionsService(), + mounts_service=_FakeMountsService(), + ) + return svc, streams_svc, turns_svc + + +@pytest.mark.asyncio +async def test_query_sessions_echoes_highest_turn_index_references(): + session_with_turns = "session-with-turns" + session_without_turns = "session-without-turns" + + early_ref = Reference(id=uuid4(), slug="early-workflow", version="v1") + latest_ref = Reference(id=uuid4(), slug="latest-workflow", version="v2") + + streams = [_stream(session_with_turns), _stream(session_without_turns)] + turns_by_session = { + session_with_turns: [ + _turn(session_with_turns, turn_index=0, references=[early_ref]), + _turn(session_with_turns, turn_index=1, references=[latest_ref]), + ], + session_without_turns: [], + } + + svc, _, turns_svc = _service(streams=streams, turns_by_session=turns_by_session) + + result = await svc.query_sessions(project_id=_PROJECT) + + assert len(result) == 2 + for item in result: + assert isinstance(item, SessionListItem) + + by_session = {item.session_id: item for item in result} + assert by_session[session_with_turns].references == [latest_ref] + assert by_session[session_without_turns].references is None + + +@pytest.mark.asyncio +async def test_query_sessions_batches_latest_turn_lookup_into_one_call(): + session_a, session_b, session_c = "session-a", "session-b", "session-c" + streams = [_stream(session_a), _stream(session_b), _stream(session_c)] + turns_by_session = { + session_a: [_turn(session_a, turn_index=0)], + session_b: [_turn(session_b, turn_index=0)], + session_c: [], + } + + svc, _, turns_svc = _service(streams=streams, turns_by_session=turns_by_session) + + await svc.query_sessions(project_id=_PROJECT) + + # One batch call covering every listed session_id -- never one lookup per row. + assert len(turns_svc.latest_turn_per_session_calls) == 1 + assert set(turns_svc.latest_turn_per_session_calls[0]["session_ids"]) == { + session_a, + session_b, + session_c, + } + + +@pytest.mark.asyncio +async def test_query_sessions_no_streams_skips_turns_lookup_entirely(): + svc, _, turns_svc = _service(streams=[], turns_by_session={}) + + result = await svc.query_sessions(project_id=_PROJECT) + + assert result == [] + assert turns_svc.latest_turn_per_session_calls == [] + + +@pytest.mark.asyncio +async def test_query_sessions_preserves_stream_ordering(): + session_first, session_second = "session-first", "session-second" + streams = [_stream(session_first), _stream(session_second)] + + svc, _, _ = _service(streams=streams, turns_by_session={}) + + result = await svc.query_sessions(project_id=_PROJECT) + + assert [item.session_id for item in result] == [session_first, session_second] + + +# ---------------------------------------------------------------------------------- # +# (b) DAO — statement compilation for `latest_turn_per_session` +# ---------------------------------------------------------------------------------- # + + +class _DummyScalars: + def all(self): + return [] + + +class _DummyResult: + def scalars(self): + return _DummyScalars() + + +class _DummySession: + def __init__(self): + self.captured_stmt = None + + async def execute(self, stmt): + self.captured_stmt = stmt + return _DummyResult() + + +class _DummySessionContext: + def __init__(self, session): + self.session = session + + async def __aenter__(self): + return self.session + + async def __aexit__(self, exc_type, exc, tb): + return False + + +@pytest.mark.asyncio +async def test_latest_turn_per_session_compiles_to_distinct_on_session_id(monkeypatch): + from oss.src.dbs.postgres.sessions.turns import dao as turns_dao_module + + session = _DummySession() + mock_engine = type( + "MockEngine", (), {"session": lambda self: _DummySessionContext(session)} + )() + monkeypatch.setattr( + turns_dao_module, "get_transactions_engine", lambda: mock_engine + ) + + await turns_dao_module.SessionTurnsDAO().latest_turn_per_session( + project_id=_PROJECT, + session_ids=["session-a", "session-b"], + ) + + stmt = session.captured_stmt + sql = str( + stmt.compile( + dialect=postgresql.dialect(), compile_kwargs={"literal_binds": True} + ) + ) + + assert "DISTINCT ON (session_turns.session_id)" in sql + order_by_fragment = sql.split("ORDER BY", 1)[1] + assert order_by_fragment.strip().startswith( + "session_turns.session_id, session_turns.turn_index DESC" + ) + + +@pytest.mark.asyncio +async def test_latest_turn_per_session_empty_session_ids_short_circuits(monkeypatch): + from oss.src.dbs.postgres.sessions.turns import dao as turns_dao_module + + session = _DummySession() + mock_engine = type( + "MockEngine", (), {"session": lambda self: _DummySessionContext(session)} + )() + monkeypatch.setattr( + turns_dao_module, "get_transactions_engine", lambda: mock_engine + ) + + result = await turns_dao_module.SessionTurnsDAO().latest_turn_per_session( + project_id=_PROJECT, + session_ids=[], + ) + + assert result == {} + # never even touches the engine -- no session_ids means no query + assert session.captured_stmt is None diff --git a/api/oss/tests/pytest/unit/sessions/test_query_sessions_search.py b/api/oss/tests/pytest/unit/sessions/test_query_sessions_search.py new file mode 100644 index 0000000000..435b2c9658 --- /dev/null +++ b/api/oss/tests/pytest/unit/sessions/test_query_sessions_search.py @@ -0,0 +1,185 @@ +"""Unit tests for WP0-R2: free-text title search on `/sessions/query`. + +`search` is a case-insensitive substring filter over `session_streams.name` (the +session title, HeaderDBA). Known coverage caveat (accepted): auto-titling is FE-only +today, so untitled sessions won't match. + +(a) DAO-level: statement-compilation only (no DB), mirroring +`test_query_sessions_windowing.py`'s dummy-engine monkeypatch pattern. +(b) Service-level: fake streams service, mirroring `test_sessions_root_service.py`'s +fake pattern, asserting `search` forwards from the core `SessionQuery` into the +`SessionStreamQuery` built for `streams_service.query_streams`. +""" + +from typing import Optional +from uuid import uuid4 + +import pytest + +from oss.src.core.sessions.dtos import SessionQuery +from oss.src.core.sessions.service import SessionsService +from oss.src.core.sessions.streams.dtos import SessionStream, SessionStreamQuery +from oss.src.dbs.postgres.sessions.streams import dao as dao_module + + +# ---------------------------------------------------------------------------------- # +# (a) DAO — statement compilation +# ---------------------------------------------------------------------------------- # + + +class _DummyScalars: + def all(self): + return [] + + +class _DummyResult: + def scalars(self): + return _DummyScalars() + + +class _DummySession: + def __init__(self): + self.captured_stmt = None + + async def execute(self, stmt): + self.captured_stmt = stmt + return _DummyResult() + + +class _DummySessionContext: + def __init__(self, session): + self.session = session + + async def __aenter__(self): + return self.session + + async def __aexit__(self, exc_type, exc, tb): + return False + + +async def _run_query(monkeypatch, *, search: Optional[str]): + session = _DummySession() + mock_engine = type( + "MockEngine", (), {"session": lambda self: _DummySessionContext(session)} + )() + monkeypatch.setattr(dao_module, "get_transactions_engine", lambda: mock_engine) + + await dao_module.SessionStreamsDAO().query( + project_id=uuid4(), + filter=SessionStreamQuery(search=search), + ) + return session.captured_stmt + + +@pytest.mark.asyncio +async def test_search_compiles_to_case_insensitive_like_on_name(monkeypatch): + stmt = await _run_query(monkeypatch, search="Refund") + + sql = str(stmt.compile(compile_kwargs={"literal_binds": True})) + params = stmt.compile().params + + # Generic-dialect ilike renders as lower(...) LIKE lower(...); assert the actual + # rendering rather than assuming ILIKE (that's postgres-dialect-specific). + assert "lower(session_streams.name) LIKE lower(" in sql + assert params["name_1"] == "%Refund%" + + +@pytest.mark.asyncio +async def test_search_escapes_like_special_characters(monkeypatch): + stmt = await _run_query(monkeypatch, search="50%_done\\x") + + sql = str(stmt.compile(compile_kwargs={"literal_binds": True})) + params = stmt.compile().params + + assert params["name_1"] == "%50\\%\\_done\\\\x%" + assert "ESCAPE '\\'" in sql + + +@pytest.mark.asyncio +async def test_absent_search_has_no_like_clause(monkeypatch): + stmt = await _run_query(monkeypatch, search=None) + + sql = str(stmt.compile(compile_kwargs={"literal_binds": True})) + + assert "LIKE" not in sql.upper() + + +@pytest.mark.asyncio +async def test_blank_search_has_no_like_clause(monkeypatch): + stmt = await _run_query(monkeypatch, search="") + + sql = str(stmt.compile(compile_kwargs={"literal_binds": True})) + + assert "LIKE" not in sql.upper() + + +@pytest.mark.asyncio +async def test_whitespace_only_search_has_no_like_clause(monkeypatch): + stmt = await _run_query(monkeypatch, search=" ") + + sql = str(stmt.compile(compile_kwargs={"literal_binds": True})) + + assert "LIKE" not in sql.upper() + + +# ---------------------------------------------------------------------------------- # +# (b) Service — forwarding into the streams query +# ---------------------------------------------------------------------------------- # + + +_PROJECT = uuid4() +_SESSION = "session-wp0-r2" + + +def _stream(session_id: str = _SESSION) -> SessionStream: + return SessionStream(id=uuid4(), project_id=_PROJECT, session_id=session_id) + + +class _FakeStreamsService: + """Mirrors `test_sessions_root_service.py`'s `_FakeStreamsService`, plus capturing + the `filter` kwarg (unused by the existing fake) so `search` forwarding is + observable.""" + + def __init__(self, row: Optional[SessionStream] = None): + self.row = row + self.query_calls: list[dict] = [] + + async def query_streams( + self, *, project_id, filter, windowing=None, session_ids=None + ): + self.query_calls.append({"project_id": project_id, "filter": filter}) + return [self.row] if self.row else [] + + +class _FakeTurnsService: + async def query_turns(self, *, project_id, query=None, windowing=None): + return [] + + async def latest_turn_per_session(self, *, project_id, session_ids): + return {} + + +class _FakeInteractionsService: + pass + + +class _FakeMountsService: + pass + + +@pytest.mark.asyncio +async def test_service_forwards_search_into_stream_query(): + streams = _FakeStreamsService(row=_stream()) + svc = SessionsService( + streams_service=streams, + turns_service=_FakeTurnsService(), + interactions_service=_FakeInteractionsService(), + mounts_service=_FakeMountsService(), + ) + + await svc.query_sessions( + project_id=_PROJECT, + query=SessionQuery(search="x"), + ) + + assert streams.query_calls[0]["filter"].search == "x" diff --git a/api/oss/tests/pytest/unit/sessions/test_query_sessions_windowing.py b/api/oss/tests/pytest/unit/sessions/test_query_sessions_windowing.py new file mode 100644 index 0000000000..51eacf9d47 --- /dev/null +++ b/api/oss/tests/pytest/unit/sessions/test_query_sessions_windowing.py @@ -0,0 +1,185 @@ +"""Unit tests (statement-compilation only, no DB) for `apply_windowing` support of +`updated_at` ordering — the sessions list must sort by last activity, not by the +uuid7 `id` creation order (WP0-R1). + +`updated_at` is nullable (never touched since row creation), so the effective ordering/ +cursor attribute is `coalesce(updated_at, created_at)` — a never-updated row degrades to +its creation time instead of sorting first under `ORDER BY ... DESC` (Postgres puts NULLs +first). This mirrors the FE's `activity()` helper (`updated_at ?? created_at`).""" + +from datetime import datetime, timezone + +import pytest + +from sqlalchemy import select + +from uuid_utils.compat import uuid7 + +from oss.src.core.shared.dtos import Windowing +from oss.src.core.sessions.streams.dtos import SessionStreamQuery +from oss.src.dbs.postgres.sessions.streams import dao as dao_module +from oss.src.dbs.postgres.sessions.streams.dbes import SessionStreamDBE +from oss.src.dbs.postgres.shared.utils import apply_windowing + +COALESCE_EXPR = "coalesce(session_streams.updated_at, session_streams.created_at)" + + +def _compile(stmt) -> str: + return str(stmt.compile(compile_kwargs={"literal_binds": True})) + + +def _assert_created_at_only_inside_coalesce(sql_fragment: str) -> None: + """`created_at` may appear, but only as part of the coalesce expression — never as a + bare `session_streams.created_at` comparison/order-by riding on its own. Callers pass + a WHERE/ORDER-BY fragment, not a full SELECT (whose column list legitimately lists + `created_at` as a plain selected column).""" + stripped = sql_fragment.replace(COALESCE_EXPR, "") + assert "created_at" not in stripped + + +def test_updated_at_attribute_orders_by_coalesced_updated_at_with_id_tiebreak(): + stmt = apply_windowing( + stmt=select(SessionStreamDBE), + DBE=SessionStreamDBE, + attribute="updated_at", + order="descending", + windowing=Windowing(limit=20), + ) + sql = _compile(stmt) + order_by_fragment = sql.split("ORDER BY", 1)[1] + + # Tiebreak direction must match the cursor's DESC semantics (`id <`) — an ASC + # tiebreak here would split a tie group across the page boundary. + assert f"ORDER BY {COALESCE_EXPR} DESC, session_streams.id DESC" in sql + _assert_created_at_only_inside_coalesce(order_by_fragment) + + +def test_updated_at_attribute_ascending_orders_with_matching_id_tiebreak(): + """Mirror of the descending case: the ascending branch's cursor uses `id >`, so + the tiebreak must be `id ASC`, not the bare (previously-ASC-by-default) column.""" + stmt = apply_windowing( + stmt=select(SessionStreamDBE), + DBE=SessionStreamDBE, + attribute="updated_at", + order="ascending", + windowing=Windowing(limit=20, order="ascending"), + ) + sql = _compile(stmt) + + assert f"ORDER BY {COALESCE_EXPR} ASC, session_streams.id ASC" in sql + + +def test_updated_at_cursor_rides_coalesced_updated_at(): + newest = datetime.now(timezone.utc) + next_id = uuid7() + stmt = apply_windowing( + stmt=select(SessionStreamDBE), + DBE=SessionStreamDBE, + attribute="updated_at", + order="descending", + windowing=Windowing(newest=newest, next=next_id, limit=20), + ) + # Assert against the WHERE clause specifically: created_at also appears in the + # SELECT column list (select(SessionStreamDBE) pulls every column), so grepping + # the full compiled statement would pass even if the cursor were mis-anchored. + where = str(stmt.whereclause.compile(compile_kwargs={"literal_binds": True})) + + assert f"{COALESCE_EXPR} <=" in where + assert f"{COALESCE_EXPR} <" in where + assert "session_streams.id <" in where + # `created_at` must ride only inside the coalesce expression, never bare. + _assert_created_at_only_inside_coalesce(where) + + +def test_created_at_attribute_behavior_is_unchanged(): + """Regression pin: current behavior for `attribute="created_at"` (observed + directly against `apply_windowing` before this change) must not shift.""" + newest = datetime.now(timezone.utc) + next_id = uuid7() + + no_cursor_sql = _compile( + apply_windowing( + stmt=select(SessionStreamDBE), + DBE=SessionStreamDBE, + attribute="created_at", + order="descending", + windowing=Windowing(limit=20), + ) + ) + assert ( + "ORDER BY session_streams.created_at DESC, session_streams.id DESC" + in no_cursor_sql + ) + + cursor_sql = _compile( + apply_windowing( + stmt=select(SessionStreamDBE), + DBE=SessionStreamDBE, + attribute="created_at", + order="descending", + windowing=Windowing(newest=newest, next=next_id, limit=20), + ) + ) + assert "session_streams.created_at <=" in cursor_sql + assert "session_streams.created_at <" in cursor_sql + assert "session_streams.id <" in cursor_sql + + +# ---------------------------------------------------------------------------------- # +# DAO fallback (no windowing) — the liveness-index caller (`query_session_streams`) +# hits this branch. Ordering isn't load-bearing there, but it must stay consistent +# with the windowed path rather than silently reverting to bare updated_at/created_at. +# ---------------------------------------------------------------------------------- # + + +class _DummyScalars: + def all(self): + return [] + + +class _DummyResult: + def scalars(self): + return _DummyScalars() + + +class _DummySession: + def __init__(self): + self.captured_stmt = None + + async def execute(self, stmt): + self.captured_stmt = stmt + return _DummyResult() + + +class _DummySessionContext: + def __init__(self, session): + self.session = session + + async def __aenter__(self): + return self.session + + async def __aexit__(self, exc_type, exc, tb): + return False + + +@pytest.mark.asyncio +async def test_dao_query_fallback_orders_by_coalesced_updated_at(monkeypatch): + from uuid import uuid4 + + session = _DummySession() + mock_engine = type( + "MockEngine", (), {"session": lambda self: _DummySessionContext(session)} + )() + monkeypatch.setattr(dao_module, "get_transactions_engine", lambda: mock_engine) + + await dao_module.SessionStreamsDAO().query( + project_id=uuid4(), + filter=SessionStreamQuery(), + windowing=None, + ) + + sql = str(session.captured_stmt.compile(compile_kwargs={"literal_binds": True})) + order_by_fragment = sql.split("ORDER BY", 1)[1] + + assert f"ORDER BY {COALESCE_EXPR} DESC, session_streams.id DESC" in sql + _assert_created_at_only_inside_coalesce(order_by_fragment) diff --git a/api/oss/tests/pytest/unit/sessions/test_sessions_root_service.py b/api/oss/tests/pytest/unit/sessions/test_sessions_root_service.py index 9950b27874..6e4abe22ae 100644 --- a/api/oss/tests/pytest/unit/sessions/test_sessions_root_service.py +++ b/api/oss/tests/pytest/unit/sessions/test_sessions_root_service.py @@ -122,6 +122,16 @@ async def query_turns(self, *, project_id, query=None, windowing=None): ] return self.turns + async def latest_turn_per_session(self, *, project_id, session_ids): + by_session: dict = {} + for turn in self.turns: + if turn.session_id not in session_ids: + continue + current = by_session.get(turn.session_id) + if current is None or turn.turn_index > current.turn_index: + by_session[turn.session_id] = turn + return by_session + async def delete_by_session_id(self, *, project_id, session_id): self.delete_calls.append({"project_id": project_id, "session_id": session_id}) return len(self.turns) @@ -286,7 +296,12 @@ async def test_query_sessions_no_filter_returns_all_streams(): result = await svc.query_sessions(project_id=_PROJECT) - assert result == [stream] + # `SessionListItem` (stream fields + `references`), not the bare `SessionStream` -- + # equality is type-sensitive in pydantic, so compare the inherited fields as a dict + # (full-field pinning) and assert the added field separately. + assert len(result) == 1 + assert result[0].model_dump(exclude={"references"}) == stream.model_dump() + assert result[0].references is None assert streams.query_calls[0]["session_ids"] is None diff --git a/web/oss/src/components/AgentChatSlice/state/projectSessions.ts b/web/oss/src/components/AgentChatSlice/state/projectSessions.ts index 804e9c4a8d..d7b6b5bd09 100644 --- a/web/oss/src/components/AgentChatSlice/state/projectSessions.ts +++ b/web/oss/src/components/AgentChatSlice/state/projectSessions.ts @@ -60,7 +60,9 @@ export const projectSessionsAtomFamily = atomFamily((appId: string) => }), ) -/** Last-activity epoch for ordering/dedup: heartbeat `updated_at`, falling back to `created_at`. */ +/** Last-activity epoch for ordering/dedup: heartbeat `updated_at`, falling back to `created_at`. + * The server (`/sessions/query`) now orders by `updated_at` too (WP0-R1) — this client-side sort + * is belt-and-suspenders (dedup still needs it to pick the fresher of two rows for one session_id). */ const activity = (s: SessionStream): number => { const ts = s.updated_at ?? s.created_at const ms = ts ? Date.parse(ts) : NaN diff --git a/web/packages/agenta-entities/src/session/api/api.ts b/web/packages/agenta-entities/src/session/api/api.ts index 105e0c4299..68561c3d12 100644 --- a/web/packages/agenta-entities/src/session/api/api.ts +++ b/web/packages/agenta-entities/src/session/api/api.ts @@ -8,6 +8,8 @@ * const events = await querySessionRecords({sessionId, projectId}) * ``` */ +import type {AgentaApi} from "@agentaai/api-client" + import {safeParseWithLogging} from "../../shared/utils/zodSchema" import { mountFileContentResponseSchema, @@ -246,32 +248,61 @@ export interface QuerySessionsParams { * hide them by display filter, rather than mistake an archived row for a hard-delete and prune * it. Set false only for a view that wants strictly non-archived rows. */ includeArchived?: boolean + /** Case-insensitive substring match over the session title (`session_streams.name`). */ + search?: string appId?: string abortSignal?: AbortSignal lowPriority?: boolean + /** Page size — omit for the server default (no `windowing` sent at all, preserving prior + * unpaginated behavior). */ + limit?: number + /** Cursor: the `id` of the last row from the previous page. */ + next?: string + /** Cursor: the activity value (`updated_at ?? created_at`, server-coalesced) of the last + * row from the previous page (pairs with `next`). */ + newest?: string } /** * The durable session list for the project: merged stream rows (id, `name` title, flags, * `created_at`, `deleted_at`=ended), filtered by the turns' workflow `references`. This is the - * server source the reconciling sidebar merges over its localStorage cache. Returns `null` on - * failure / missing project scope. + * server source the reconciling sidebar merges over its localStorage cache. Ordered by last + * activity (`updated_at`) server-side. Returns `null` on failure / missing project scope. */ export async function querySessions({ projectId, references, includeEnded = true, includeArchived = true, + search, appId, abortSignal, lowPriority, + limit, + next, + newest, }: QuerySessionsParams): Promise { if (!projectId) return null + // Only attach `windowing` when a caller actually opts into pagination — an absent + // field preserves the prior unwindowed (server-default-ordered) query shape. + const windowing = + limit !== undefined || next !== undefined || newest !== undefined + ? {limit, next, newest} + : undefined + const client = lowPriority ? getLowPrioritySessionsClient() : getSessionsClient() const data = await callFern("[querySessions]", () => client.querySessions( - {references, include_ended: includeEnded, include_archived: includeArchived}, + { + references, + include_ended: includeEnded, + include_archived: includeArchived, + windowing, + // TODO(fern-regen): `search` isn't in the generated SessionQueryRequest yet + // (regen out of scope) — widen the type until the client picks it up. + search, + } as AgentaApi.SessionQueryRequest & {search?: string}, projectScopedRequest(projectId, appId, abortSignal), ), ) diff --git a/web/packages/agenta-entities/src/session/core/schema.ts b/web/packages/agenta-entities/src/session/core/schema.ts index 5c5489963d..cc2a0c0c2f 100644 --- a/web/packages/agenta-entities/src/session/core/schema.ts +++ b/web/packages/agenta-entities/src/session/core/schema.ts @@ -82,6 +82,14 @@ export type SessionInteraction = z.infer export type SessionInteractionStatusCode = "pending" | "responded" | "resolved" | "cancelled" export type SessionInteractionKind = "user_approval" | "user_input" | "client_tool" +/** A `{id, slug, version}` workflow/agent reference — mirrors `QuerySessionsParams.references` + * on the request side. Every field is optional: a turn's reference may carry only a subset. */ +export const sessionReferenceSchema = z.object({ + id: z.string().nullish(), + slug: z.string().nullish(), + version: z.string().nullish(), +}) + /** * A live stream handle. Liveness rides `flags` (nested: alive ⊇ running ⊇ attached); * `resumable` (alive & !running) and `reattachable` (running & !attached) are derived @@ -109,6 +117,9 @@ export const sessionStreamSchema = z.object({ deleted_at: z.string().nullish(), // `archived_at` set = hidden-but-recoverable (distinct from `deleted_at`=ended, still resumable). archived_at: z.string().nullish(), + // `/sessions/query` only (WP0-R3): the session's latest turn's workflow/agent references — + // absent for a session with no turns yet, and for a plain stream fetch (not query'd). + references: z.array(sessionReferenceSchema).nullish(), }) export const sessionStreamsResponseSchema = z.object({ @@ -137,6 +148,7 @@ export const sessionStreamCommandResponseSchema = z.object({ }) export type SessionStream = z.infer +export type SessionReference = z.infer export type SessionStreamCommandResponse = z.infer /** One entry in a mount's durable file listing. `path` is relative to the mount root; folders diff --git a/web/packages/agenta-entities/tests/unit/session-query-schema.test.ts b/web/packages/agenta-entities/tests/unit/session-query-schema.test.ts new file mode 100644 index 0000000000..ec4bd54459 --- /dev/null +++ b/web/packages/agenta-entities/tests/unit/session-query-schema.test.ts @@ -0,0 +1,135 @@ +/** + * Pins the `/sessions/query` wire shape for `sessionStreamSchema`/`sessionsQueryResponseSchema`. + * + * Fern's compile-time types under-declare backend `extra="allow"` fields and don't catch + * server-side field renames, and zod silently STRIPS unknown wire keys to `undefined` on a + * `.nullish()` field — so a renamed backend key (e.g. `name`, `references`) tsc-passes and + * parse-succeeds while the FE session list silently loses the data (this class of drift has + * bitten the session schemas twice: see `session-record-schema.test.ts`). These tests assert + * a realistic wire payload survives parsing with its values intact, and document — via a + * deliberately-renamed fixture — that a real rename would slip past zod undetected unless + * this fixture is kept in sync with an actual backend payload. + */ +import {describe, expect, it} from "vitest" + +import {sessionsQueryResponseSchema, sessionStreamSchema} from "../../src/session/core/schema" + +/** A fully-populated `/sessions/query` row exactly as the backend serializes it today: + * `SessionListItem` (`SessionStream` + `Identifier`/`Header`/`Lifecycle` + `references`), + * `response_model_exclude_none=True` on the route so nulled optionals are simply absent. */ +const wireRow = { + id: "22222222-2222-2222-2222-222222222222", + project_id: "11111111-1111-1111-1111-111111111111", + session_id: "sess-1", + name: "Refactor the auth flow", + description: "First user message becomes the session title", + flags: {is_alive: true, is_running: false, is_attached: false}, + tags: {priority: "high"}, + meta: {source: "web"}, + turn_id: "turn-7", + created_at: "2026-07-20T10:00:00Z", + updated_at: "2026-07-24T09:30:00Z", + references: [ + {id: "33333333-3333-3333-3333-333333333333", slug: "support-router", version: "v3"}, + ], +} + +const wireEnvelope = {count: 1, sessions: [wireRow]} + +describe("sessionStreamSchema (/sessions/query rows)", () => { + it("parses a fully-populated stamped row and keeps the fields the session list reads", () => { + const out = sessionStreamSchema.parse(wireRow) + expect(out.session_id).toBe("sess-1") + expect(out.name).toBe("Refactor the auth flow") + expect(out.flags).toEqual({is_alive: true, is_running: false, is_attached: false}) + expect(out.updated_at).toBe("2026-07-24T09:30:00Z") + expect(out.references?.[0]?.id).toBe("33333333-3333-3333-3333-333333333333") + expect(out.references?.[0]?.slug).toBe("support-router") + expect(out.references?.[0]?.version).toBe("v3") + }) + + it("parses the minimum row (only the three non-nullish fields) with everything else undefined", () => { + // `id`/`project_id`/`session_id` are the schema's only required fields — every other + // field is `.nullish()`. This is the legacy/never-turned session shape. + const minimal = { + id: "22222222-2222-2222-2222-222222222222", + project_id: "11111111-1111-1111-1111-111111111111", + session_id: "sess-legacy", + } + const out = sessionStreamSchema.parse(minimal) + expect(out.session_id).toBe("sess-legacy") + expect(out.name).toBeUndefined() + expect(out.description).toBeUndefined() + expect(out.flags).toBeUndefined() + expect(out.turn_id).toBeUndefined() + expect(out.created_at).toBeUndefined() + expect(out.updated_at).toBeUndefined() + expect(out.deleted_at).toBeUndefined() + expect(out.archived_at).toBeUndefined() + expect(out.references).toBeUndefined() + }) + + it("parses an include_ended (soft-deleted) row and keeps deleted_at + flags", () => { + const deleted = { + ...wireRow, + session_id: "sess-ended", + deleted_at: "2026-07-23T12:00:00Z", + flags: {is_alive: false, is_running: false, is_attached: false}, + } + const out = sessionStreamSchema.parse(deleted) + expect(out.deleted_at).toBe("2026-07-23T12:00:00Z") + expect(out.flags).toEqual({is_alive: false, is_running: false, is_attached: false}) + }) + + it("parses an archived row and keeps archived_at distinct from deleted_at", () => { + const archived = { + ...wireRow, + session_id: "sess-archived", + archived_at: "2026-07-22T08:15:00Z", + } + const out = sessionStreamSchema.parse(archived) + expect(out.archived_at).toBe("2026-07-22T08:15:00Z") + expect(out.deleted_at).toBeUndefined() + }) + + it("parses a row with no references key at all (exclude_none: no turns yet)", () => { + const {references: _references, ...noRefs} = wireRow + const out = sessionStreamSchema.parse(noRefs) + expect(out.references).toBeUndefined() + }) +}) + +describe("sessionsQueryResponseSchema (envelope)", () => { + it("parses {count, sessions} and each row through sessionStreamSchema", () => { + const out = sessionsQueryResponseSchema.parse(wireEnvelope) + expect(out.count).toBe(1) + expect(out.sessions).toHaveLength(1) + expect(out.sessions[0].session_id).toBe("sess-1") + expect(out.sessions[0].references?.[0]?.id).toBe("33333333-3333-3333-3333-333333333333") + }) +}) + +describe("drift guard: a renamed wire key is silently dropped, not rejected", () => { + // This is the failure mode the whole file exists to catch early: zod schemas built from + // `.nullish()` fields don't fail on an unknown key, they just parse it away. If the backend + // ever renames `name` or `references` the way it once renamed the record envelope (see + // `session-record-schema.test.ts`), THIS test's fixture (`wireRow` above) must be updated + // from a real payload — at which point the assertions in the "fully-populated" case above + // (`out.name`, `out.references?.[0]?.id`) start failing and catch the drift. The cases below + // do not test correct behavior; they document what a silent break would look like today. + it("dropping `name` in favor of a renamed key parses clean but loses the title", () => { + const {name: _name, ...rest} = wireRow + const renamed = {...rest, header_name: "Refactor the auth flow"} + const out = sessionStreamSchema.parse(renamed) + expect(out.name).toBeUndefined() + expect((out as Record).header_name).toBeUndefined() + }) + + it("dropping `references` in favor of a renamed key parses clean but loses the reference list", () => { + const {references: _references, ...rest} = wireRow + const renamed = {...rest, refs: wireRow.references} + const out = sessionStreamSchema.parse(renamed) + expect(out.references).toBeUndefined() + expect((out as Record).refs).toBeUndefined() + }) +})