fix(manager-core): F-P-005 keyset cursor on execution_logs list_for_script

list_for_script used ORDER BY created_at DESC LIMIT $2 OFFSET $3.
Postgres has to scan + discard OFFSET rows on every page, so deep
pagination on a script with many log rows gets linearly slower. Every
sister-table list endpoint in the repo already uses cursor pagination
— this is the outlier.

Add ExecutionLogCursor { created_at, id } with `<rfc3339>_<uuid>`
encode/decode. New trait signature:
  list_for_script(script_id, limit, cursor: Option<ExecutionLogCursor>)

Predicate: WHERE script_id=$1 AND (created_at, id) < ($cursor_ts, $cursor_id).
ORDER BY also gains `id DESC` for deterministic ordering at equal-time
boundaries.

API surface:
- /api/v1/admin/scripts/{id}/logs?cursor=<token>&limit=50 is the new
  shape (dashboards adopt later — separate finding F-U-012).
- Legacy `offset` query param accepted-and-ignored to keep older
  dashboards from 400'ing during rollout.

AUDIT.md anchor: F-P-005.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

crates/manager-core/src/api.rs

@@ -375,8 +375,14 @@ async fn delete_script<R: ScriptRepository, L: ExecutionLogRepository>(
 pub struct LogsQuery {
     #[serde(default = "default_limit")]
     pub limit: i64,
+    /// F-P-005: keyset cursor (`<rfc3339>_<uuid>`). Replaces the OFFSET
+    /// path that scanned + discarded N rows per page. Absent for the
+    /// first page. The legacy `offset` query param is accepted but
+    /// silently ignored to keep older dashboards from 400'ing.
     #[serde(default)]
-    pub offset: i64,
+    pub cursor: Option<String>,
+    #[serde(default, rename = "offset")]
+    pub _legacy_offset: Option<i64>,
 }
 
 const fn default_limit() -> i64 {
@@ -399,8 +405,11 @@ async fn list_logs<R: ScriptRepository, L: ExecutionLogRepository>(
     // Cap to keep the dashboard responsive; the data plane writes are
     // unbounded over time so a paged read is the only sane default.
     let limit = q.limit.clamp(1, 200);
-    let offset = q.offset.max(0);
-    let logs = state.logs.list_for_script(id, limit, offset).await?;
+    let cursor = q
+        .cursor
+        .as_deref()
+        .and_then(crate::repo::ExecutionLogCursor::decode);
+    let logs = state.logs.list_for_script(id, limit, cursor).await?;
     Ok(Json(logs))
 }
 

crates/manager-core/src/repo.rs

@@ -485,13 +485,43 @@ impl<R: ScriptRepository> ScriptResolver for RepoResolver<R> {
 /// Read-side access to the `execution_logs` table. Writes go through
 /// `PostgresExecutionLogSink` so the read and write paths can diverge
 /// in cluster mode without disturbing this trait.
+/// Opaque cursor encoding `(created_at, id)` for keyset pagination.
+/// Format: `<rfc3339>_<uuid>` — readable in URLs and stable across
+/// server restarts (no opaque internal IDs).
+#[derive(Debug, Clone)]
+pub struct ExecutionLogCursor {
+    pub created_at: chrono::DateTime<chrono::Utc>,
+    pub id: uuid::Uuid,
+}
+
+impl ExecutionLogCursor {
+    /// Encode as `<rfc3339>_<uuid>`.
+    #[must_use]
+    pub fn encode(&self) -> String {
+        format!("{}_{}", self.created_at.to_rfc3339(), self.id)
+    }
+
+    /// Decode from the wire format. Returns `None` on any parse error;
+    /// the caller treats that as "no cursor" (start of the page).
+    #[must_use]
+    pub fn decode(s: &str) -> Option<Self> {
+        let (ts, id) = s.rsplit_once('_')?;
+        let created_at = chrono::DateTime::parse_from_rfc3339(ts).ok()?.to_utc();
+        let id = uuid::Uuid::parse_str(id).ok()?;
+        Some(Self { created_at, id })
+    }
+}
+
 #[async_trait]
 pub trait ExecutionLogRepository: Send + Sync {
+    /// F-P-005: keyset cursor on `(created_at, id)` — `cursor` is the
+    /// boundary of the previous page (or `None` for the first page).
+    /// Avoids the OFFSET tail-latency cliff on deep pagination.
     async fn list_for_script(
         &self,
         script_id: ScriptId,
         limit: i64,
-        offset: i64,
+        cursor: Option<ExecutionLogCursor>,
     ) -> Result<Vec<ExecutionLog>, ScriptRepositoryError>;
 }
 
@@ -512,23 +542,45 @@ impl ExecutionLogRepository for PostgresExecutionLogRepository {
         &self,
         script_id: ScriptId,
         limit: i64,
-        offset: i64,
+        cursor: Option<ExecutionLogCursor>,
     ) -> Result<Vec<ExecutionLog>, ScriptRepositoryError> {
-        let rows = sqlx::query_as::<_, ExecutionLogRow>(
-            "SELECT id, app_id, script_id, request_id, \
-                    request_path, request_headers, request_body, \
-                    response_code, response_body, \
-                    logs, duration_ms, status, created_at \
-             FROM execution_logs \
-             WHERE script_id = $1 \
-             ORDER BY created_at DESC \
-             LIMIT $2 OFFSET $3",
-        )
-        .bind(script_id.into_inner())
-        .bind(limit)
-        .bind(offset)
-        .fetch_all(&self.pool)
-        .await?;
+        let rows = match cursor {
+            Some(c) => {
+                sqlx::query_as::<_, ExecutionLogRow>(
+                    "SELECT id, app_id, script_id, request_id, \
+                            request_path, request_headers, request_body, \
+                            response_code, response_body, \
+                            logs, duration_ms, status, created_at \
+                     FROM execution_logs \
+                     WHERE script_id = $1 \
+                       AND (created_at, id) < ($2, $3) \
+                     ORDER BY created_at DESC, id DESC \
+                     LIMIT $4",
+                )
+                .bind(script_id.into_inner())
+                .bind(c.created_at)
+                .bind(c.id)
+                .bind(limit)
+                .fetch_all(&self.pool)
+                .await?
+            }
+            None => {
+                sqlx::query_as::<_, ExecutionLogRow>(
+                    "SELECT id, app_id, script_id, request_id, \
+                            request_path, request_headers, request_body, \
+                            response_code, response_body, \
+                            logs, duration_ms, status, created_at \
+                     FROM execution_logs \
+                     WHERE script_id = $1 \
+                     ORDER BY created_at DESC, id DESC \
+                     LIMIT $2",
+                )
+                .bind(script_id.into_inner())
+                .bind(limit)
+                .fetch_all(&self.pool)
+                .await?
+            }
+        };
 
         Ok(rows.into_iter().map(Into::into).collect())
     }