diff --git a/crates/cli/src/main.rs b/crates/cli/src/main.rs index a8d1dee..075bbaa 100644 --- a/crates/cli/src/main.rs +++ b/crates/cli/src/main.rs @@ -3151,7 +3151,9 @@ fn build_mcp_server() -> anyhow::Result { "assignee": {"type": "string", "description": "Filter by assignee name"}, "priority_gte": {"type": "integer", "description": "Minimum priority (inclusive)"}, "priority_lte": {"type": "integer", "description": "Maximum priority (inclusive)"}, - "limit": {"type": "integer", "description": "Max results (default 50)"}, + "limit": {"type": "integer", "description": "Max results (default 50). Rows are returned NEWEST FIRST, so a limit keeps the most recent work rather than an arbitrary slice."}, + "offset": {"type": "integer", "description": "Skip this many rows before the limit, for paging through a long board."}, + "full": {"type": "boolean", "description": "Include body, result and metadata on every row. Off by default: a few hundred full rows exceed the tool-result token limit. Use task_get for the detail of a specific task."}, "cql_host": {"type": "string", "description": "CQL host:port (default: 127.0.0.1:9042)"}, "debug_stop": {"type": "boolean", "description": "When true, attach a board-health alert (or fail on critical board degradation) so you stop and investigate instead of trusting a degraded board. Off by default."} } @@ -3183,8 +3185,26 @@ fn build_mcp_server() -> anyhow::Result { .and_then(|v| v.as_u64()) .map(|i| i as usize), }; - let tasks = store.list_tasks(filter).map_err(|e| e.to_string())?; - finish_task(&store, &args, &tasks) + let offset = args.get("offset").and_then(|v| v.as_u64()).unwrap_or(0) as usize; + let page = store + .list_tasks_paged(filter, offset) + .map_err(|e| e.to_string())?; + let full = args.get("full").and_then(|v| v.as_bool()).unwrap_or(false); + let rows: Vec<_> = if full { + page.tasks + } else { + page.tasks.iter().map(forge_tasks::Task::slim).collect() + }; + // Report the window explicitly. A capped read that says so is usable; + // one that does not is worse than an error -- the previous behaviour + // silently omitted recent tasks and looked like a complete answer. + let payload = serde_json::json!({ + "tasks": rows, + "total": page.total, + "offset": offset, + "truncated": page.truncated, + }); + finish_task(&store, &args, &payload) } ); @@ -3313,6 +3333,7 @@ fn build_mcp_server() -> anyhow::Result { "type": "object", "properties": { "cql_host": {"type": "string", "description": "CQL host:port (default: 127.0.0.1:9042)"}, + "full": {"type": "boolean", "description": "Include body, result and metadata on every row. Off by default: the full board exceeds the tool-result token limit. Use task_get for the detail of a specific task."}, "debug_stop": {"type": "boolean", "description": "When true, attach a board-health alert (or fail on critical board degradation) so you stop and investigate instead of trusting a degraded board. Off by default."} } }), @@ -3321,7 +3342,16 @@ fn build_mcp_server() -> anyhow::Result { forge_tasks::resolve_cql_hosts(args.get("cql_host").and_then(|v| v.as_str())); let store = forge_tasks::TaskStore::connect(&cql_hosts, None).map_err(|e| e.to_string())?; let board = store.board().map_err(|e| e.to_string())?; - finish_task(&store, &args,&board) + // Slim rows by default. The full board was 619,574 characters for 382 + // tasks and exceeded the tool-result token limit outright, so an agent + // could not read it without spilling to a file. task_get still + // returns everything for the one task you pick. + let full = args + .get("full") + .and_then(|v| v.as_bool()) + .unwrap_or(false); + let board = if full { board } else { board.slim() }; + finish_task(&store, &args, &board) } ); @@ -6202,8 +6232,18 @@ fn handle_task(action: TaskAction, pretty: bool) -> anyhow::Result<()> { priority_lte, limit: Some(limit), }; - let tasks = store.list_tasks(filter)?; - println!("{}", forge_shared::emit_json(&tasks, pretty)?); + // stdout keeps the plain array: `frg task list | jq` depends on it, and + // rows are now newest-first so a --limit keeps the most recent work. + // Truncation goes to STDERR rather than being folded into the payload, + // so a capped read still says so without breaking the pipe. + let page = store.list_tasks_paged(filter, 0)?; + if page.truncated { + eprintln!( + "warning: read hit the fetch bound; {} rows scanned, so older tasks may be missing", + page.total + ); + } + println!("{}", forge_shared::emit_json(&page.tasks, pretty)?); } TaskAction::Link { diff --git a/crates/tasks/src/store.rs b/crates/tasks/src/store.rs index 9c5097d..dd4a32c 100644 --- a/crates/tasks/src/store.rs +++ b/crates/tasks/src/store.rs @@ -73,6 +73,45 @@ pub struct TaskStore { tenant_id: String, } +/// How many rows a board or list read will pull before it stops. +/// +/// The reads used to pass the caller's `limit` straight to CQL. The tasks table +/// is `PRIMARY KEY (tenant_id, task_id)`, so rows come back in task_id order -- +/// effectively arbitrary -- and a LIMIT therefore took an ARBITRARY SLICE, not +/// the newest rows. With a few hundred open tasks the window stopped including +/// anything recent: tasks created hours earlier were absent from both +/// `task_board` and `task_list` while `task_get` returned them fine. An agent +/// that captured a deferral and read the board back could not see its own write. +/// +/// CQL cannot fix this with ORDER BY: `created_at` is not a clustering column, so +/// ordering has to happen after the fetch, which means fetching enough to order +/// meaningfully. This is that bound -- high enough to cover the whole board in +/// practice, finite so a runaway table cannot be read into memory unbounded. When +/// it is reached the caller is TOLD, rather than being handed a silent window. +const MAX_FETCH_ROWS: usize = 10_000; + +/// Newest first, with task_id breaking ties so the order is total and a page +/// boundary cannot shuffle between reads. +pub(crate) fn sort_newest_first(tasks: &mut [Task]) { + tasks.sort_by(|a, b| { + b.created_at + .cmp(&a.created_at) + .then_with(|| a.task_id.cmp(&b.task_id)) + }); +} + +/// A read that may have been cut short, and says so. +#[derive(Debug, Clone)] +pub struct FetchedTasks { + pub tasks: Vec, + /// Rows matching before the caller's limit/offset was applied. + pub total: usize, + /// True when MAX_FETCH_ROWS was hit, so `total` is a floor rather than the + /// count. A capped read that reports itself is usable; one that does not is + /// worse than an error. + pub truncated: bool, +} + impl TaskStore { /// Connect to the CQL cluster, create schema, and return a `TaskStore`. /// @@ -376,12 +415,15 @@ impl TaskStore { let where_clause = conditions.join(" AND "); let limit = filter.limit.unwrap_or(100); + // Fetch to the bound, THEN order, THEN apply the caller's limit. Passing + // the limit to CQL took an arbitrary slice, because rows arrive in + // task_id order and created_at is not a clustering column. let cql = format!( "SELECT task_id, title, body, status, assignee, reviewer, priority, \ workspace_kind, workspace_path, created_by, block_reason, result, summary, \ metadata, skills, related_entity_ids, created_at, updated_at \ FROM agent_memory.tasks WHERE {} LIMIT {} ALLOW FILTERING", - where_clause, limit + where_clause, MAX_FETCH_ROWS ); let result = cql_exec!(self.rt, &self.session, cql) @@ -395,9 +437,33 @@ impl TaskStore { tasks.push(task); } } + sort_newest_first(&mut tasks); + tasks.truncate(limit); Ok(tasks) } + /// As `list_tasks`, but reporting how many matched and whether the read was + /// cut short, so a caller can tell a complete answer from a window. + pub fn list_tasks_paged(&self, filter: TaskFilter, offset: usize) -> Result { + let limit = filter.limit.unwrap_or(100); + // Ask for everything the bound allows; the window is applied here. + let mut unbounded = filter; + unbounded.limit = Some(MAX_FETCH_ROWS); + let mut tasks = self.list_tasks(unbounded)?; + let truncated = tasks.len() >= MAX_FETCH_ROWS; + let total = tasks.len(); + let tasks = if offset >= tasks.len() { + Vec::new() + } else { + tasks.split_off(offset).into_iter().take(limit).collect() + }; + Ok(FetchedTasks { + tasks, + total, + truncated, + }) + } + /// Create a parent→child link (stored in both directions). pub fn link_tasks(&self, parent_id: &str, child_id: &str, link_type: &str) -> Result<()> { let now = now_ms(); @@ -485,8 +551,9 @@ impl TaskStore { metadata, skills, related_entity_ids, created_at, updated_at \ FROM agent_memory.tasks \ WHERE tenant_id={tenant} \ - LIMIT 500 ALLOW FILTERING", + LIMIT {limit} ALLOW FILTERING", tenant = self.tenant_id, + limit = MAX_FETCH_ROWS, ); let result = cql_exec!(self.rt, &self.session, cql) @@ -513,6 +580,19 @@ impl TaskStore { } } + // Newest first in every column. Without this the board's order was + // task_id order -- arbitrary -- so the most recently captured work sat + // wherever its id happened to fall. + for column in [ + &mut triage, + &mut ready, + &mut in_progress, + &mut blocked, + &mut complete, + ] { + sort_newest_first(column); + } + Ok(KanbanBoard { columns: KanbanColumns { triage, diff --git a/crates/tasks/src/types.rs b/crates/tasks/src/types.rs index ca68b3b..99db2b6 100644 --- a/crates/tasks/src/types.rs +++ b/crates/tasks/src/types.rs @@ -147,6 +147,30 @@ pub struct TaskFilter { // KanbanBoard // --------------------------------------------------------------------------- +impl Task { + /// A listing-sized copy: the fields you scan a board with, without the prose. + /// + /// `task_board` returned 619,574 characters for 382 rows and exceeded the + /// tool-result token limit outright, so an agent could not read the board at + /// all without spilling it to a file first. Almost all of that is `body`, + /// `result` and `metadata` -- the long-form fields that matter when you are + /// working a single task and are noise when you are choosing between eighty. + /// + /// `task_get` still returns everything, so nothing is lost: the detail is one + /// call away for the task you actually pick. + pub fn slim(&self) -> Task { + Task { + body: None, + result: None, + metadata: None, + // Kept: a one-line summary is what makes a listing decidable, and it + // is bounded by construction. + summary: self.summary.clone(), + ..self.clone() + } + } +} + #[derive(Debug, Serialize, Deserialize)] pub struct KanbanBoard { pub columns: KanbanColumns, @@ -161,6 +185,22 @@ pub struct KanbanColumns { pub complete: Vec, } +impl KanbanBoard { + /// Every column slimmed. See `Task::slim`. + pub fn slim(&self) -> KanbanBoard { + let slim = |column: &Vec| column.iter().map(Task::slim).collect(); + KanbanBoard { + columns: KanbanColumns { + triage: slim(&self.columns.triage), + ready: slim(&self.columns.ready), + in_progress: slim(&self.columns.in_progress), + blocked: slim(&self.columns.blocked), + complete: slim(&self.columns.complete), + }, + } + } +} + #[cfg(test)] mod tests { use super::*; @@ -171,6 +211,92 @@ mod tests { /// serializes the in-memory `Task` via serde_json. Before this was locked /// in, a body with raw newlines could surface as unescaped control chars and /// break `frg task list | jq`. + fn task_at(id: &str, created_at: i64) -> Task { + Task { + task_id: id.to_string(), + title: format!("task {id}"), + body: Some("a long body that a listing does not need".repeat(40)), + status: TaskStatus::Triage, + assignee: None, + reviewer: None, + priority: 50, + workspace_kind: None, + workspace_path: None, + created_by: "agent".to_string(), + block_reason: None, + result: Some("a long result".repeat(40)), + summary: Some("one line that makes a listing decidable".to_string()), + metadata: None, + skills: Vec::new(), + related_entity_ids: Vec::new(), + created_at, + updated_at: created_at, + } + } + + #[test] + fn slim_drops_the_prose_and_keeps_what_a_listing_needs() { + // task_board returned 619,574 characters for 382 tasks and exceeded the + // tool-result token limit outright, so the board could not be read at all + // without spilling it to a file. Nearly all of it is body/result. + let full = task_at("t_1", 10); + let slim = full.slim(); + + assert_eq!(slim.body, None, "body is the bulk and belongs in task_get"); + assert_eq!(slim.result, None); + assert_eq!(slim.metadata, None); + + // Kept, because these are what make a row decidable without opening it. + assert_eq!(slim.task_id, "t_1"); + assert_eq!(slim.title, full.title); + assert_eq!(slim.status, full.status); + assert_eq!(slim.priority, 50); + assert_eq!(slim.summary, full.summary); + assert_eq!(slim.created_at, 10); + + let big = serde_json::to_string(&full).unwrap().len(); + let small = serde_json::to_string(&slim).unwrap().len(); + assert!( + small * 4 < big, + "slim should be far smaller: {small} vs {big}" + ); + } + + #[test] + fn a_board_column_is_ordered_newest_first() { + // The defect: rows arrive in task_id order because the table is + // PRIMARY KEY (tenant_id, task_id), so a LIMIT took an arbitrary slice + // rather than the newest rows. Tasks created hours earlier were absent + // from task_board and task_list while task_get returned them fine. + let mut column = vec![ + task_at("t_aaa", 100), + task_at("t_zzz", 300), + task_at("t_mmm", 200), + ]; + crate::store::sort_newest_first(&mut column); + assert_eq!( + column.iter().map(|t| t.created_at).collect::>(), + vec![300, 200, 100], + "newest first, regardless of how task_id sorts" + ); + } + + #[test] + fn ties_break_on_task_id_so_paging_cannot_shuffle() { + // Two tasks created in the same millisecond must have a stable order, or a + // page boundary could return one twice and the other never. + let mut column = vec![task_at("t_bbb", 100), task_at("t_aaa", 100)]; + crate::store::sort_newest_first(&mut column); + assert_eq!( + column + .iter() + .map(|t| t.task_id.as_str()) + .collect::>(), + vec!["t_aaa", "t_bbb"], + "equal timestamps order by id, giving a total order" + ); + } + #[test] fn task_with_multiline_body_serializes_to_valid_json() { let task = Task {