feat(cortex): page session inventory metadata

docs/api.md

@@ -85,7 +85,7 @@ compatibility routes.
 
 | Method | Path | Scope | Request | Response (top-level) | Status codes | Idempotent | Notes |
 | --- | --- | --- | --- | --- | --- | --- | --- |
-| GET | `/api/sessions` | read | query: `project?`, `tool?`, `hostname?`, `from?`, `to?`, `limit?` | `ListSessionsResponse { count, sessions: [AiSessionEntry] }` | 200, 400, 401, 503, 500 | Y | Inventory of indexed AI transcripts. |
+| GET | `/api/sessions` | read | query: `project?`, `tool?`, `host?`, `since?`, `until?`, `limit?`, `offset?` | `ListSessionsResponse { count, sessions: [AiSessionEntry], rollup_as_of? }` | 200, 400, 401, 503, 500 | Y | Bounded pages of indexed AI transcript metadata; up to 1,000 sessions per response and offset capped at 10,000. No transcript text is returned. |
 | GET | `/api/sessions/rendered` | read | query: `project`, `tool`, `session_id`, `host` (all REQUIRED), `cursor?`, `limit?` | `RenderedSessionPageResponse` (`delivery=polling`, semantic events, durable next cursor, high-water mark, truncation and retry metadata) | 200, 400, 401, 503, 500 | Y | Keyset pagination by persisted `logs.id`, ascending. Maximum 200 events and 256 KiB per page; oversized event text is UTF-8-safely truncated with a parse warning. Schema: `contracts/rendered-session-page.schema.json`. Clients may hand its committed cursor to the native session stream. |
 | GET | `/api/sessions/search` | read | query: `query` (REQUIRED), `project?`, `tool?`, `from?`, `to?`, `limit?` (u32) | `SearchSessionsResponse { total_candidates, candidate_rows, candidate_cap, candidate_window_truncated, truncated, sessions: [SearchedSessionEntry], limit_clamped_to? }` | 200, 400, 401, 503, 500 | Y | `limit` clamped at **100** — see Response size caps. |
 | GET | `/api/sessions/abuse` | read | query: `project?`, `tool?`, `from?`, `to?`, `limit?`, `before?` (u32), `after?` (u32), **`terms?`** (repeated key: `?terms=foo&terms=bar`) | `AbuseSearchResponse { terms, candidate_rows, candidate_cap, candidate_window_truncated, truncated, matches: [AbuseMatch], limit_clamped_to? }` | 200, 400, 401, 503, 500 | Y | `limit` clamped at **500**. Decoded via `serde_qs::axum::QsQuery`, so `Vec<String>` is supported through repeated `terms=` keys (the CLI's `HttpClient` serializes the shared request type the same way). |

src/api_tests.rs

@@ -1305,6 +1305,41 @@ async fn sessions_route_lists_ingested_session() {
     assert_eq!(value["sessions"][0]["tool"], "claude");
 }
 
+#[tokio::test]
+async fn sessions_route_pages_metadata_with_offset() {
+    let (state, pool, _dir) = test_state(Some("secret".into()));
+    db::insert_logs_batch(
+        &pool,
+        &[
+            ai_entry(
+                "2026-01-01T00:00:00Z",
+                "host-a",
+                "info",
+                "first",
+                "cortex",
+                "codex",
+                "s1",
+            ),
+            ai_entry(
+                "2026-01-02T00:00:00Z",
+                "host-a",
+                "info",
+                "second",
+                "cortex",
+                "codex",
+                "s2",
+            ),
+        ],
+    )
+    .unwrap();
+    let app = router(state).unwrap();
+    let (status, value) = get_json(app, "/api/sessions?limit=1&offset=1", Some("secret")).await;
+    assert_eq!(status, axum::http::StatusCode::OK);
+    assert_eq!(value["count"], 1);
+    assert_eq!(value["sessions"][0]["session_id"], "s1");
+    assert!(value["sessions"][0].get("message").is_none());
+}
+
 #[tokio::test]
 async fn ai_search_requires_query_param() {
     let (state, _pool, _dir) = test_state(Some("secret".into()));

src/app/models/ai_sessions.rs

@@ -9,6 +9,8 @@ pub struct ListSessionsRequest {
     pub since: Option<String>,
     pub until: Option<String>,
     pub limit: Option<u32>,
+    /// Bounded page offset over the session rollup or a time-filtered query.
+    pub offset: Option<u32>,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]

src/app/services/ai.rs

@@ -81,6 +81,7 @@ impl CortexService {
             since: from,
             until: to,
             limit: req.limit,
+            offset: req.offset,
         };
         let (rows, rollup_as_of) = self
             .run_db("list_sessions", move |pool| {

src/cli/dispatch.rs

@@ -153,6 +153,7 @@ impl SessionsArgs {
             since: self.since,
             until: self.until,
             limit: self.limit,
+            offset: None,
         }
     }
 }

src/cli/dispatch_tests.rs

@@ -247,7 +247,7 @@ fn sessions_args_into_request_snapshot() {
     let req = args.into_request();
     assert_eq!(
         format!("{req:?}"),
-        "ListSessionsRequest { project: Some(\"/home/me/proj\"), tool: Some(\"claude\"), host: None, since: None, until: None, limit: Some(20) }"
+        "ListSessionsRequest { project: Some(\"/home/me/proj\"), tool: Some(\"claude\"), host: None, since: None, until: None, limit: Some(20), offset: None }"
     );
 }
 

src/db/models.rs

@@ -69,6 +69,7 @@ pub struct ListAiSessionsParams {
     pub since: Option<String>,
     pub until: Option<String>,
     pub limit: Option<u32>,
+    pub offset: Option<u32>,
 }
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]

src/db/queries.rs

@@ -729,6 +729,7 @@ pub fn list_ai_sessions_live(
 ) -> Result<Vec<AiSessionEntry>> {
     let conn = pool.get()?;
     let limit = params.limit.unwrap_or(100).min(1000);
+    let offset = params.offset.unwrap_or(0).min(10_000);
     let mut sql = String::from(
         "WITH filtered_logs AS MATERIALIZED (
             SELECT id, ai_project, ai_tool, ai_session_id, ai_transcript_path,
@@ -807,7 +808,7 @@ pub fn list_ai_sessions_live(
           GROUP BY f.ai_project, f.ai_tool, f.ai_session_id, f.hostname,
                    t.session_title, t.session_title_provenance
           ORDER BY last_seen DESC
-          LIMIT {limit}"
+          LIMIT {limit} OFFSET {offset}"
     ));
 
     let mut stmt = conn.prepare(&sql)?;
@@ -835,6 +836,7 @@ fn list_ai_sessions_from_rollup(
 ) -> Result<Vec<AiSessionEntry>> {
     let conn = pool.get()?;
     let limit = params.limit.unwrap_or(100).min(1000);
+    let offset = params.offset.unwrap_or(0).min(10_000);
     let mut sql = String::from(
         "SELECT ai_project, ai_tool, ai_session_id, ai_transcript_path,
                 hostname, first_seen, last_seen, event_count,
@@ -864,7 +866,9 @@ fn list_ai_sessions_from_rollup(
     // (the cost that made the live aggregation slow). Adding tiebreak columns
     // would reintroduce a temp b-tree, so ties stay engine-arbitrary here just
     // as they are in the live query.
-    sql.push_str(&format!(" ORDER BY last_seen DESC LIMIT {limit}"));
+    sql.push_str(&format!(
+        " ORDER BY last_seen DESC LIMIT {limit} OFFSET {offset}"
+    ));
 
     let mut stmt = conn.prepare(&sql)?;
     let rows = stmt.query_map(rusqlite::params_from_iter(bindings.iter()), |row| {

src/db/queries_tests.rs

@@ -2012,6 +2012,7 @@ fn ai_session_queries_respect_filters() {
             since: Some("2026-01-01T00:00:00Z".into()),
             until: Some("2026-01-01T23:59:59Z".into()),
             limit: Some(10),
+            offset: None,
         },
     )
     .unwrap();
@@ -2347,6 +2348,7 @@ fn default_session_params() -> ListAiSessionsParams {
         since: None,
         until: None,
         limit: Some(100),
+        offset: None,
     }
 }
 
@@ -2386,6 +2388,40 @@ fn rollup_result_equals_live_aggregation() {
     }
 }
 
+#[test]
+fn session_inventory_offset_pages_match_live_and_rollup() {
+    let (pool, _dir) = test_pool();
+    seed_ai_sessions(&pool);
+    let all = list_ai_sessions(&pool, &default_session_params()).unwrap();
+    let page = ListAiSessionsParams {
+        limit: Some(5),
+        offset: Some(5),
+        ..default_session_params()
+    };
+    let live = list_ai_sessions(&pool, &page).unwrap();
+    assert_eq!(live.len(), 5);
+    assert_eq!(
+        live.iter()
+            .map(|row| &row.ai_session_id)
+            .collect::<Vec<_>>(),
+        all[5..10]
+            .iter()
+            .map(|row| &row.ai_session_id)
+            .collect::<Vec<_>>()
+    );
+    refresh_ai_session_rollup(&pool).unwrap();
+    let rolled = list_ai_sessions(&pool, &page).unwrap();
+    assert_eq!(
+        rolled
+            .iter()
+            .map(|row| &row.ai_session_id)
+            .collect::<Vec<_>>(),
+        live.iter()
+            .map(|row| &row.ai_session_id)
+            .collect::<Vec<_>>()
+    );
+}
+
 #[test]
 fn rollup_status_reports_refresh_time() {
     let (pool, _dir) = test_pool();
@@ -3413,6 +3449,7 @@ fn bench_stats_and_sessions() {
         since: None,
         until: None,
         limit: Some(100),
+        offset: None,
     };
     let mut live_rows = 0usize;
     let sessions_live_latency = bench_percentiles_ms(20, || {