feat(master): paginate task view and add TTL-based history cleanup

crates/lance-context-api/src/lib.rs

@@ -1070,10 +1070,14 @@ pub struct EnqueueTaskRequest {
     pub depends_on: Vec<String>,
 }
 
-/// Response for `GET /api/v1/tasks`.
+/// Response for `GET /api/v1/tasks`. Paginated newest-first.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct TaskListResponse {
     pub tasks: Vec<TaskRecord>,
+    /// Total number of tasks (queue + retained history) before paging.
+    pub total: usize,
+    pub limit: usize,
+    pub offset: usize,
 }
 
 #[cfg(test)]

crates/lance-context-master/src/config.rs

@@ -110,6 +110,14 @@ pub struct MasterConfig {
     #[arg(long, env = "TASK_HISTORY_LIMIT", default_value_t = 1_000)]
     pub task_history_limit: usize,
 
+    /// Maximum age of terminal (Done/Failed) scheduler tasks before they are
+    /// pruned, in seconds. `0` disables age-based pruning (the count cap in
+    /// `TASK_HISTORY_LIMIT` still applies). Whichever of the two removes a task
+    /// first wins. Queued/running tasks and terminal tasks that a live task
+    /// still depends on are never pruned regardless of age.
+    #[arg(long, env = "TASK_HISTORY_TTL_SECS", default_value_t = 86_400)]
+    pub task_history_ttl_secs: u64,
+
     /// Directory of built UI assets to serve. When unset, only the JSON API is
     /// exposed.
     #[arg(long, env = "UI_DIR")]

crates/lance-context-master/src/routes.rs

@@ -42,6 +42,15 @@ fn default_records_limit() -> usize {
     25
 }
 
+/// Query params for the paginated task list.
+#[derive(Debug, Deserialize)]
+pub struct TaskListParams {
+    #[serde(default = "default_limit")]
+    pub limit: usize,
+    #[serde(default)]
+    pub offset: usize,
+}
+
 /// Query params for the detail endpoint.
 #[derive(Debug, Deserialize)]
 pub struct DetailParams {
@@ -410,17 +419,26 @@ pub async fn enqueue_task(
     Ok((StatusCode::ACCEPTED, Json(task)))
 }
 
-/// `GET /api/v1/tasks` — all tasks (queue + recent history), newest first.
+/// `GET /api/v1/tasks` — paginated tasks (queue + recent history), newest first.
 pub async fn list_tasks(
     State(state): State<Arc<MasterState>>,
+    Query(params): Query<TaskListParams>,
 ) -> Result<Json<TaskListResponse>, MasterError> {
     let mut tasks = state
         .task_store
         .list()
         .await
         .map_err(MasterError::from_lance)?;
     tasks.sort_by_key(|t| std::cmp::Reverse(t.enqueued_at));
-    Ok(Json(TaskListResponse { tasks }))
+    let total = tasks.len();
+    let limit = params.limit.clamp(1, 200);
+    let page = tasks.into_iter().skip(params.offset).take(limit).collect();
+    Ok(Json(TaskListResponse {
+        tasks: page,
+        total,
+        limit,
+        offset: params.offset,
+    }))
 }
 
 /// `GET /api/v1/tasks/{id}` — a single task by id.
@@ -536,6 +554,7 @@ mod tests {
             etcd_client_key: None,
             etcd_lease_ttl_secs: 5,
             task_history_limit: 1_000,
+            task_history_ttl_secs: 86_400,
             ui_dir: None,
         }
     }

crates/lance-context-master/src/scheduler.rs

@@ -449,6 +449,7 @@ mod tests {
             etcd_client_key: None,
             etcd_lease_ttl_secs: 5,
             task_history_limit: 1_000,
+            task_history_ttl_secs: 86_400,
             ui_dir: None,
         }
     }

crates/lance-context-master/src/state.rs

@@ -149,6 +149,7 @@ mod tests {
             etcd_client_key: None,
             etcd_lease_ttl_secs: 5,
             task_history_limit: 1_000,
+            task_history_ttl_secs: 86_400,
             ui_dir: None,
         }
     }

crates/lance-context-master/src/task_store.rs

@@ -29,6 +29,7 @@ const TASK_POLL_BATCH: usize = 256;
 pub struct TaskStore {
     inner: Arc<EtcdTaskStore>,
     history_limit: usize,
+    history_ttl_secs: u64,
 }
 
 struct EtcdTaskStore {
@@ -80,6 +81,7 @@ impl TaskStore {
         let store = Self {
             inner: Arc::new(EtcdTaskStore::connect(config).await?),
             history_limit: config.task_history_limit.max(1),
+            history_ttl_secs: config.task_history_ttl_secs,
         };
         store.recover_orphaned().await?;
         store.prune_terminal_history().await?;
@@ -171,23 +173,12 @@ impl TaskStore {
 
     async fn prune_terminal_history(&self) -> lance::Result<usize> {
         let tasks = self.list().await?;
-        let protected = tasks
-            .iter()
-            .filter(|task| matches!(task.state, TaskState::Queued | TaskState::Running))
-            .flat_map(|task| task.depends_on.iter().cloned())
-            .collect::<HashSet<_>>();
-        let mut terminal = tasks
-            .into_iter()
-            .filter(|task| matches!(task.state, TaskState::Done | TaskState::Failed))
-            .filter(|task| !protected.contains(&task.id))
-            .collect::<Vec<_>>();
-        terminal
-            .sort_by_key(|task| std::cmp::Reverse(task.finished_at.unwrap_or(task.enqueued_at)));
-        let ids = terminal
-            .into_iter()
-            .skip(self.history_limit)
-            .map(|task| task.id)
-            .collect::<Vec<_>>();
+        let ttl_cutoff = if self.history_ttl_secs > 0 {
+            Some(now_ms() - (self.history_ttl_secs as i64) * 1_000)
+        } else {
+            None
+        };
+        let ids = prunable_terminal_ids(tasks, self.history_limit, ttl_cutoff);
         if ids.is_empty() {
             return Ok(0);
         }
@@ -196,6 +187,45 @@ impl TaskStore {
     }
 }
 
+/// Select terminal (Done/Failed) task ids to prune under two independent
+/// policies, whichever removes a task first:
+/// - **count**: keep only the newest `history_limit` terminal tasks;
+/// - **age (TTL)**: when `ttl_cutoff` is `Some`, drop terminal tasks whose
+///   `finished_at` (fallback `enqueued_at`) is at or before the cutoff.
+///
+/// Queued/Running tasks, and any terminal task a live (Queued/Running) task
+/// still lists in `depends_on`, are never selected regardless of age or count.
+/// Pure and etcd-free so the policy is unit-testable.
+fn prunable_terminal_ids(
+    tasks: Vec<TaskRecord>,
+    history_limit: usize,
+    ttl_cutoff: Option<i64>,
+) -> Vec<String> {
+    let protected = tasks
+        .iter()
+        .filter(|task| matches!(task.state, TaskState::Queued | TaskState::Running))
+        .flat_map(|task| task.depends_on.iter().cloned())
+        .collect::<HashSet<_>>();
+    let mut terminal = tasks
+        .into_iter()
+        .filter(|task| matches!(task.state, TaskState::Done | TaskState::Failed))
+        .filter(|task| !protected.contains(&task.id))
+        .collect::<Vec<_>>();
+    // Newest first, so rank >= history_limit are the ones over the count cap.
+    terminal.sort_by_key(|task| std::cmp::Reverse(task.finished_at.unwrap_or(task.enqueued_at)));
+    terminal
+        .into_iter()
+        .enumerate()
+        .filter(|(rank, task)| {
+            let over_count = *rank >= history_limit;
+            let expired = ttl_cutoff
+                .is_some_and(|cutoff| task.finished_at.unwrap_or(task.enqueued_at) <= cutoff);
+            over_count || expired
+        })
+        .map(|(_, task)| task.id)
+        .collect()
+}
+
 impl EtcdTaskStore {
     async fn connect(config: &MasterConfig) -> lance::Result<Self> {
         if config.etcd_endpoints.is_empty() {
@@ -896,6 +926,7 @@ mod tests {
             etcd_client_key: None,
             etcd_lease_ttl_secs: 30,
             task_history_limit: 1_000,
+            task_history_ttl_secs: 86_400,
             ui_dir: None,
         }
     }
@@ -910,6 +941,73 @@ mod tests {
         assert!(error.to_string().contains("ETCD_ENDPOINTS is required"));
     }
 
+    fn terminal_task(id: &str, state: TaskState, finished_at: i64) -> TaskRecord {
+        TaskRecord {
+            id: id.to_string(),
+            kind: TaskKind::Compact,
+            target: "exp".to_string(),
+            state,
+            error: None,
+            detail: None,
+            enqueued_at: finished_at,
+            started_at: Some(finished_at),
+            finished_at: Some(finished_at),
+            depends_on: Vec::new(),
+        }
+    }
+
+    #[test]
+    fn prune_keeps_newest_over_count_cap() {
+        // 5 terminal tasks, cap 2, TTL disabled → oldest 3 pruned.
+        let tasks = (0..5)
+            .map(|i| terminal_task(&format!("t{i}"), TaskState::Done, 1_000 + i * 10))
+            .collect::<Vec<_>>();
+        let mut pruned = prunable_terminal_ids(tasks, 2, None);
+        pruned.sort();
+        assert_eq!(pruned, vec!["t0", "t1", "t2"]);
+    }
+
+    #[test]
+    fn prune_expires_by_ttl_cutoff() {
+        // High count cap so only the TTL policy fires.
+        let tasks = vec![
+            terminal_task("old-1", TaskState::Done, 100),
+            terminal_task("old-2", TaskState::Failed, 200),
+            terminal_task("fresh", TaskState::Done, 5_000),
+        ];
+        let mut pruned = prunable_terminal_ids(tasks, 1_000, Some(1_000));
+        pruned.sort();
+        assert_eq!(pruned, vec!["old-1", "old-2"]);
+    }
+
+    #[test]
+    fn prune_never_touches_active_or_depended_on() {
+        // A queued task depends on a terminal one that is otherwise TTL-expired;
+        // that dependency must be protected. Queued/Running are never terminal
+        // candidates in the first place.
+        let mut dep = terminal_task("dep", TaskState::Done, 100);
+        dep.id = "dep".to_string();
+        let mut queued = terminal_task("live", TaskState::Queued, 100);
+        queued.depends_on = vec!["dep".to_string()];
+        let expired = terminal_task("expired", TaskState::Done, 100);
+        let tasks = vec![dep, queued, expired];
+
+        let pruned = prunable_terminal_ids(tasks, 0, Some(1_000));
+        // `dep` protected by the live task; `live` is Queued (not terminal);
+        // only the unreferenced expired terminal task is pruned.
+        assert_eq!(pruned, vec!["expired"]);
+    }
+
+    #[test]
+    fn prune_ttl_disabled_uses_count_only() {
+        let tasks = vec![
+            terminal_task("a", TaskState::Done, 1),
+            terminal_task("b", TaskState::Done, 2),
+        ];
+        // TTL None + generous cap → nothing pruned even though timestamps are old.
+        assert!(prunable_terminal_ids(tasks, 10, None).is_empty());
+    }
+
     #[tokio::test]
     #[ignore = "requires ETCD_TEST_ENDPOINTS"]
     async fn etcd_coordinates_dedupe_claims_and_target_locks() {

crates/lance-context-master/ui/src/App.tsx

@@ -931,19 +931,24 @@ function IndexIdButton({ name }: { name: string }) {
 }
 
 function TaskQueue() {
+  const [page, setPage] = useState(0);
+  const pageSize = 50;
   const tasks = useQuery({
-    queryKey: ["tasks"],
-    queryFn: () => listTasks(),
+    queryKey: ["tasks", page],
+    queryFn: () => listTasks(pageSize, page * pageSize),
     refetchInterval: 1000,
+    placeholderData: (prev) => prev,
   });
   const rows = tasks.data?.tasks ?? [];
+  const total = tasks.data?.total ?? 0;
   const active = rows.filter((t) => t.state === "queued" || t.state === "running").length;
+  const hasMore = (page + 1) * pageSize < total;
 
   return (
     <>
       <div className="stats">
-        <StatCard label="Tasks" value={fmtInt(rows.length)} />
-        <StatCard label="Active" value={fmtInt(active)} />
+        <StatCard label="Tasks" value={fmtInt(total)} />
+        <StatCard label="Active (page)" value={fmtInt(active)} />
       </div>
       <div className="table-wrap">
         {tasks.isError && <div className="error">{String(tasks.error)}</div>}
@@ -979,6 +984,17 @@ function TaskQueue() {
           <div className="empty">No tasks in the queue.</div>
         )}
       </div>
+      {(page > 0 || hasMore) && (
+        <div className="pager records-pager">
+          <span className="pager__info">page {page + 1}</span>
+          <button className="btn btn--ghost" disabled={page === 0} onClick={() => setPage(page - 1)}>
+            ← Prev
+          </button>
+          <button className="btn btn--ghost" disabled={!hasMore} onClick={() => setPage(page + 1)}>
+            Next →
+          </button>
+        </div>
+      )}
     </>
   );
 }

crates/lance-context-master/ui/src/api.ts

@@ -104,6 +104,9 @@ export interface TaskRecord {
 
 export interface TaskListResponse {
   tasks: TaskRecord[];
+  total: number;
+  limit: number;
+  offset: number;
 }
 
 export interface EnqueueTaskRequest {
@@ -181,8 +184,14 @@ export async function compactionStatus(name: string): Promise<CompactJobStatus>
   );
 }
 
-export async function listTasks(): Promise<TaskListResponse> {
-  return json(await fetch(`${API}/tasks`));
+export async function listTasks(
+  limit = 50,
+  offset = 0,
+): Promise<TaskListResponse> {
+  const params = new URLSearchParams();
+  params.set("limit", String(limit));
+  params.set("offset", String(offset));
+  return json(await fetch(`${API}/tasks?${params.toString()}`));
 }
 
 export async function getTask(id: string): Promise<TaskRecord> {