[feat] Order and search the session list by last activity (2/12)

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

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,
         )

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

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,14 +45,18 @@ 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
         turns' references (WP1's GIN `.contains()`) to resolve the matching
         `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,
         *,

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

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,

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,
         *,

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]

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,
         *,

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)