From f5a079f172580808aa436b2757a8ec429b334779 Mon Sep 17 00:00:00 2001 From: Ben Kearns <35475+bkearns@users.noreply.github.com> Date: Thu, 20 Aug 2026 20:13:50 -0700 Subject: [PATCH] fix(tasks): never answer a board read with an empty result you could not obtain `forge task list` printed `[]` and exited 0 against a board it had not read. A *query* answered "there is nothing" when the truth was "I could not look", and every consumer -- `/whats-next`, `/roadmap`, the defer-capture hook -- was confidently wrong in the same direction. The deferred-work rule's whole premise is that captured work is durable and queryable; this is the failure mode it exists to prevent. Three states are now distinguishable at every read path: cannot reach the contact point -> error naming the host:port actually tried reached it, keyspace unusable -> error saying which keyspace, on which host reached it, board is empty -> [] Connection: `TaskStore::connect` names every contact point it tried, and separates "cannot reach" from "reached, but the board keyspace is not usable there". An unreachable board previously reported `connect to CQL: Connection refused` with no indication which of --cql-host / FORGE_CQL_HOST / project config / global config had supplied the address. Row decoding: `list_tasks` and `board` dropped rows that would not parse (`if let Ok(task) = parse_task_row(row)`), so a drifted schema simply reported fewer tasks with no error -- the same silent emptiness, one row at a time. Rows now fail the read. `parse_task_row` is strict about schema drift and lenient about data: a NULL in a nullable column keeps its documented default, while a missing column, a wrong CQL type, an unknown status, or an absent `created_at` is an error. An unknown status used to be filed under `triage`, which put a `complete` task back into the outstanding pile. Comment rows no longer blank their author and body when they fail to decode. Config: a `.forge/config.toml` or `~/.config/forge.toml` that exists but cannot be read or parsed was discarded by `.ok()`, so forge queried `127.0.0.1:9042` while the user was looking at a file naming a different port -- the config half of the same symptom. Malformed or unreadable config is now an error naming the file; a file that is simply absent still falls through. `FORGE_DEBUG_STOP` with an unrecognised value is an error rather than a silent "off", which is the state the operator was trying to leave. sheet-sync: `BoardSink::existing_status` returned `get_task(..).ok()`, so an unreadable board looked like "no such task" -- exactly the answer that disarms the never-move-backward rule and lets a pull reset a `complete` or `archived` task to the sheet's status. It returns `Result>` now, backed by a new `TaskStore::find_task` that keeps absence and failure apart. Verified against the live 3-node board (2556 tasks across all columns) so the stricter parser reads real data, and over MCP for both a live and a dead host. --- crates/cli/src/main.rs | 75 ++-- crates/cli/tests/task_reads_fail_loud.rs | 92 +++++ crates/sheet-sync/src/board.rs | 31 +- crates/sheet-sync/src/board_exec.rs | 12 +- crates/sheet-sync/src/board_plan.rs | 92 +++-- crates/sheet-sync/src/sync.rs | 64 +++- crates/sheet-sync/tests/board_exec_live.rs | 2 +- crates/tasks/src/config.rs | 259 +++++++++++--- crates/tasks/src/schema.rs | 4 + crates/tasks/src/store.rs | 378 ++++++++++++++++---- crates/tasks/tests/board_reads_fail_loud.rs | 119 ++++++ 11 files changed, 932 insertions(+), 196 deletions(-) create mode 100644 crates/cli/tests/task_reads_fail_loud.rs create mode 100644 crates/tasks/tests/board_reads_fail_loud.rs diff --git a/crates/cli/src/main.rs b/crates/cli/src/main.rs index 075bbaa..2f4e2fd 100644 --- a/crates/cli/src/main.rs +++ b/crates/cli/src/main.rs @@ -1750,7 +1750,8 @@ fn finish_task( args: &serde_json::Value, value: &T, ) -> Result { - let on = forge_tasks::resolve_debug_stop(args.get("debug_stop").and_then(|v| v.as_bool())); + let on = forge_tasks::resolve_debug_stop(args.get("debug_stop").and_then(|v| v.as_bool())) + .map_err(|e| format!("{e:#}"))?; let v = serde_json::to_value(value).map_err(|e| e.to_string())?; let out = forge_tasks::apply_debug_stop(Ok(v), &store.board_health(), on).map_err(|(_, m)| m)?; @@ -3045,7 +3046,8 @@ fn build_mcp_server() -> anyhow::Result { }), |args| { let cql_hosts = - forge_tasks::resolve_cql_hosts(args.get("cql_host").and_then(|v| v.as_str())); + forge_tasks::resolve_cql_hosts(args.get("cql_host").and_then(|v| v.as_str())) + .map_err(|e| format!("{e:#}"))?; let store = forge_tasks::TaskStore::connect(&cql_hosts, None).map_err(|e| e.to_string())?; let req = forge_tasks::CreateTaskRequest { title: args.get("title").and_then(|v| v.as_str()).ok_or("title is required")?.to_string(), @@ -3093,7 +3095,8 @@ fn build_mcp_server() -> anyhow::Result { |args| { let task_id = args.get("task_id").and_then(|v| v.as_str()).ok_or("task_id is required")?; let cql_hosts = - forge_tasks::resolve_cql_hosts(args.get("cql_host").and_then(|v| v.as_str())); + forge_tasks::resolve_cql_hosts(args.get("cql_host").and_then(|v| v.as_str())) + .map_err(|e| format!("{e:#}"))?; let store = forge_tasks::TaskStore::connect(&cql_hosts, None).map_err(|e| e.to_string())?; let patch = forge_tasks::UpdateTaskPatch { status: args.get("status").and_then(|v| v.as_str()).map(str::to_string), @@ -3131,7 +3134,8 @@ fn build_mcp_server() -> anyhow::Result { .and_then(|v| v.as_str()) .ok_or("task_id is required")?; let cql_hosts = - forge_tasks::resolve_cql_hosts(args.get("cql_host").and_then(|v| v.as_str())); + forge_tasks::resolve_cql_hosts(args.get("cql_host").and_then(|v| v.as_str())) + .map_err(|e| format!("{e:#}"))?; let store = forge_tasks::TaskStore::connect(&cql_hosts, None).map_err(|e| e.to_string())?; let task = store.get_task(task_id).map_err(|e| e.to_string())?; @@ -3160,7 +3164,8 @@ fn build_mcp_server() -> anyhow::Result { }), |args| { let cql_hosts = - forge_tasks::resolve_cql_hosts(args.get("cql_host").and_then(|v| v.as_str())); + forge_tasks::resolve_cql_hosts(args.get("cql_host").and_then(|v| v.as_str())) + .map_err(|e| format!("{e:#}"))?; let store = forge_tasks::TaskStore::connect(&cql_hosts, None).map_err(|e| e.to_string())?; let filter = forge_tasks::TaskFilter { @@ -3233,7 +3238,8 @@ fn build_mcp_server() -> anyhow::Result { .and_then(|v| v.as_str()) .ok_or("child_id is required")?; let cql_hosts = - forge_tasks::resolve_cql_hosts(args.get("cql_host").and_then(|v| v.as_str())); + forge_tasks::resolve_cql_hosts(args.get("cql_host").and_then(|v| v.as_str())) + .map_err(|e| format!("{e:#}"))?; let store = forge_tasks::TaskStore::connect(&cql_hosts, None).map_err(|e| e.to_string())?; store @@ -3272,7 +3278,8 @@ fn build_mcp_server() -> anyhow::Result { .and_then(|v| v.as_str()) .ok_or("child_id is required")?; let cql_hosts = - forge_tasks::resolve_cql_hosts(args.get("cql_host").and_then(|v| v.as_str())); + forge_tasks::resolve_cql_hosts(args.get("cql_host").and_then(|v| v.as_str())) + .map_err(|e| format!("{e:#}"))?; let store = forge_tasks::TaskStore::connect(&cql_hosts, None).map_err(|e| e.to_string())?; store @@ -3316,7 +3323,8 @@ fn build_mcp_server() -> anyhow::Result { .and_then(|v| v.as_str()) .unwrap_or("agent"); let cql_hosts = - forge_tasks::resolve_cql_hosts(args.get("cql_host").and_then(|v| v.as_str())); + forge_tasks::resolve_cql_hosts(args.get("cql_host").and_then(|v| v.as_str())) + .map_err(|e| format!("{e:#}"))?; let store = forge_tasks::TaskStore::connect(&cql_hosts, None).map_err(|e| e.to_string())?; let comment = store @@ -3339,7 +3347,8 @@ fn build_mcp_server() -> anyhow::Result { }), |args| { let cql_hosts = - forge_tasks::resolve_cql_hosts(args.get("cql_host").and_then(|v| v.as_str())); + forge_tasks::resolve_cql_hosts(args.get("cql_host").and_then(|v| v.as_str())) + .map_err(|e| format!("{e:#}"))?; 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())?; // Slim rows by default. The full board was 619,574 characters for 382 @@ -6152,10 +6161,8 @@ fn handle_task(action: TaskAction, pretty: bool) -> anyhow::Result<()> { created_by, cql_host, } => { - let store = forge_tasks::TaskStore::connect( - &forge_tasks::resolve_cql_hosts(cql_host.as_deref()), - None, - )?; + let cql_hosts = forge_tasks::resolve_cql_hosts(cql_host.as_deref())?; + let store = forge_tasks::TaskStore::connect(&cql_hosts, None)?; let req = forge_tasks::CreateTaskRequest { title, body, @@ -6185,10 +6192,8 @@ fn handle_task(action: TaskAction, pretty: bool) -> anyhow::Result<()> { summary, cql_host, } => { - let store = forge_tasks::TaskStore::connect( - &forge_tasks::resolve_cql_hosts(cql_host.as_deref()), - None, - )?; + let cql_hosts = forge_tasks::resolve_cql_hosts(cql_host.as_deref())?; + let store = forge_tasks::TaskStore::connect(&cql_hosts, None)?; let patch = forge_tasks::UpdateTaskPatch { status, assignee, @@ -6205,10 +6210,8 @@ fn handle_task(action: TaskAction, pretty: bool) -> anyhow::Result<()> { } TaskAction::Get { task_id, cql_host } => { - let store = forge_tasks::TaskStore::connect( - &forge_tasks::resolve_cql_hosts(cql_host.as_deref()), - None, - )?; + let cql_hosts = forge_tasks::resolve_cql_hosts(cql_host.as_deref())?; + let store = forge_tasks::TaskStore::connect(&cql_hosts, None)?; let task = store.get_task(&task_id)?; println!("{}", forge_shared::emit_json(&task, pretty)?); } @@ -6221,10 +6224,8 @@ fn handle_task(action: TaskAction, pretty: bool) -> anyhow::Result<()> { limit, cql_host, } => { - let store = forge_tasks::TaskStore::connect( - &forge_tasks::resolve_cql_hosts(cql_host.as_deref()), - None, - )?; + let cql_hosts = forge_tasks::resolve_cql_hosts(cql_host.as_deref())?; + let store = forge_tasks::TaskStore::connect(&cql_hosts, None)?; let filter = forge_tasks::TaskFilter { status, assignee, @@ -6251,10 +6252,8 @@ fn handle_task(action: TaskAction, pretty: bool) -> anyhow::Result<()> { child_id, cql_host, } => { - let store = forge_tasks::TaskStore::connect( - &forge_tasks::resolve_cql_hosts(cql_host.as_deref()), - None, - )?; + let cql_hosts = forge_tasks::resolve_cql_hosts(cql_host.as_deref())?; + let store = forge_tasks::TaskStore::connect(&cql_hosts, None)?; store.link_tasks(&parent_id, &child_id, "child")?; println!("Linked {} \u{2192} {}", parent_id, child_id); } @@ -6264,10 +6263,8 @@ fn handle_task(action: TaskAction, pretty: bool) -> anyhow::Result<()> { child_id, cql_host, } => { - let store = forge_tasks::TaskStore::connect( - &forge_tasks::resolve_cql_hosts(cql_host.as_deref()), - None, - )?; + let cql_hosts = forge_tasks::resolve_cql_hosts(cql_host.as_deref())?; + let store = forge_tasks::TaskStore::connect(&cql_hosts, None)?; store.unlink_tasks(&parent_id, &child_id)?; println!("Unlinked {} \u{2194} {}", parent_id, child_id); } @@ -6278,19 +6275,15 @@ fn handle_task(action: TaskAction, pretty: bool) -> anyhow::Result<()> { author, cql_host, } => { - let store = forge_tasks::TaskStore::connect( - &forge_tasks::resolve_cql_hosts(cql_host.as_deref()), - None, - )?; + let cql_hosts = forge_tasks::resolve_cql_hosts(cql_host.as_deref())?; + let store = forge_tasks::TaskStore::connect(&cql_hosts, None)?; let comment = store.add_comment(&task_id, &author, &body)?; println!("{}", forge_shared::emit_json(&comment, pretty)?); } TaskAction::Board { cql_host } => { - let store = forge_tasks::TaskStore::connect( - &forge_tasks::resolve_cql_hosts(cql_host.as_deref()), - None, - )?; + let cql_hosts = forge_tasks::resolve_cql_hosts(cql_host.as_deref())?; + let store = forge_tasks::TaskStore::connect(&cql_hosts, None)?; let board = store.board()?; println!("{}", forge_shared::emit_json(&board, pretty)?); } diff --git a/crates/cli/tests/task_reads_fail_loud.rs b/crates/cli/tests/task_reads_fail_loud.rs new file mode 100644 index 0000000..e2cb54c --- /dev/null +++ b/crates/cli/tests/task_reads_fail_loud.rs @@ -0,0 +1,92 @@ +//! End-to-end guard for the exact shape of the reported bug: +//! +//! $ forge task list --limit 1 # nothing listening on the CQL port +//! [] # exit 0, nothing on stderr +//! +//! A read against an unreachable board must exit non-zero, must not print an +//! empty JSON result on stdout, and must name the host and port it tried. +//! `/whats-next` and `/roadmap` read this board; an empty array they cannot +//! distinguish from a dead one turns "I could not look" into "there is no work". + +use std::net::TcpListener; +use std::process::Command; + +/// A `127.0.0.1:` that nothing is listening on. +fn dead_contact_point() -> String { + let listener = TcpListener::bind("127.0.0.1:0").expect("bind ephemeral port"); + let addr = listener.local_addr().expect("local_addr").to_string(); + drop(listener); + addr +} + +/// Run `frg task --cql-host ` with the ambient config layers +/// neutralised, so the test asserts on the flag it passes and nothing else. +fn run_task_read(args: &[&str], host: &str) -> (bool, String, String) { + let out = Command::new(env!("CARGO_BIN_EXE_frg")) + .arg("task") + .args(args) + .arg("--cql-host") + .arg(host) + .env_remove("FORGE_CQL_HOST") + .env_remove("FORGE_DEBUG_STOP") + .output() + .expect("run frg"); + ( + out.status.success(), + String::from_utf8_lossy(&out.stdout).into_owned(), + String::from_utf8_lossy(&out.stderr).into_owned(), + ) +} + +#[test] +fn task_list_against_a_dead_board_exits_non_zero_and_prints_no_empty_array() { + let host = dead_contact_point(); + let (ok, stdout, stderr) = run_task_read(&["list", "--limit", "1"], &host); + + assert!( + !ok, + "an unreachable board must not exit 0; stdout was {stdout:?}" + ); + assert!( + stdout.trim().is_empty(), + "an unreachable board must print nothing on stdout, not a result; got {stdout:?}" + ); + assert!( + stderr.contains(&host), + "stderr must name the host and port tried, got: {stderr}" + ); +} + +#[test] +fn task_board_against_a_dead_board_exits_non_zero_and_names_the_host() { + let host = dead_contact_point(); + let (ok, stdout, stderr) = run_task_read(&["board"], &host); + + assert!( + !ok, + "an unreachable board must not exit 0; stdout was {stdout:?}" + ); + assert!( + stdout.trim().is_empty(), + "an unreachable board must print nothing on stdout; got {stdout:?}" + ); + assert!( + stderr.contains(&host), + "stderr must name the host and port tried, got: {stderr}" + ); +} + +#[test] +fn task_get_against_a_dead_board_exits_non_zero_and_names_the_host() { + let host = dead_contact_point(); + let (ok, stdout, stderr) = run_task_read(&["get", "t_deadbeef"], &host); + + assert!( + !ok, + "an unreachable board must not exit 0; stdout was {stdout:?}" + ); + assert!( + stderr.contains(&host), + "stderr must name the host and port tried, got: {stderr}" + ); +} diff --git a/crates/sheet-sync/src/board.rs b/crates/sheet-sync/src/board.rs index 03666f1..e80f7e4 100644 --- a/crates/sheet-sync/src/board.rs +++ b/crates/sheet-sync/src/board.rs @@ -8,10 +8,15 @@ use forge_tasks::TaskStatus; /// Board read/write boundary for pull. pub trait BoardSink { - /// The board's *current* status for `task_id`, or `None` if no such task - /// is known to the board. Feeds [`crate::board_plan::plan_pull`]'s + /// The board's *current* status for `task_id`, or `Ok(None)` if no such + /// task is known to the board. Feeds [`crate::board_plan::plan_pull`]'s /// never-move-backward rule — see that module's doc. - fn existing_status(&self, task_id: &str) -> Option; + /// + /// A read that FAILS is an `Err`, never `Ok(None)`. Collapsing the two + /// disarmed the never-move-backward rule exactly when the board was + /// unreachable: an unreadable `complete` task looked like a task with no + /// status, and the pull happily reset it to the sheet's value. + fn existing_status(&self, task_id: &str) -> anyhow::Result>; /// Applies one planned op to the board. Returns the task id it created /// or updated; `None` for [`BoardOp::Skip`], which never touches the @@ -35,6 +40,9 @@ pub(crate) struct FakeBoard { /// was (and wasn't) persisted before the failure. fail_on_call: Option, call_count: usize, + /// When true, `existing_status` returns `Err` — simulates the board being + /// unreadable while a pull is deciding whether a task's status is protected. + fail_status_read: bool, } #[cfg(test)] @@ -46,6 +54,7 @@ impl FakeBoard { next_id: 0, fail_on_call: None, call_count: 0, + fail_status_read: false, } } @@ -58,6 +67,15 @@ impl FakeBoard { } } + /// A [`FakeBoard`] whose `existing_status` fails — the board is + /// reachable enough to be asked and cannot answer. + pub(crate) fn new_failing_status_read() -> Self { + Self { + fail_status_read: true, + ..Self::new() + } + } + fn mint_task_id(&mut self) -> String { self.next_id += 1; format!("t_{}", self.next_id) @@ -66,8 +84,11 @@ impl FakeBoard { #[cfg(test)] impl BoardSink for FakeBoard { - fn existing_status(&self, task_id: &str) -> Option { - self.statuses.get(task_id).cloned() + fn existing_status(&self, task_id: &str) -> anyhow::Result> { + if self.fail_status_read { + anyhow::bail!("FakeBoard: simulated status-read failure for {task_id}"); + } + Ok(self.statuses.get(task_id).cloned()) } fn apply(&mut self, op: &BoardOp) -> anyhow::Result> { diff --git a/crates/sheet-sync/src/board_exec.rs b/crates/sheet-sync/src/board_exec.rs index 40ea91f..b29c0d5 100644 --- a/crates/sheet-sync/src/board_exec.rs +++ b/crates/sheet-sync/src/board_exec.rs @@ -26,7 +26,7 @@ impl BoardExec { /// that function's doc) and wraps the resulting store. No tenant /// scoping: sheet-sync operates on the default tenant. pub fn connect(cql_host: Option<&str>) -> anyhow::Result { - let hosts = resolve_cql_hosts(cql_host); + let hosts = resolve_cql_hosts(cql_host)?; let store = TaskStore::connect(&hosts, None)?; Ok(Self { store }) } @@ -39,8 +39,14 @@ impl BoardExec { } impl BoardSink for BoardExec { - fn existing_status(&self, task_id: &str) -> Option { - self.store.get_task(task_id).ok().map(|t| t.task.status) + fn existing_status(&self, task_id: &str) -> anyhow::Result> { + // `get_task(..).ok()` turned "the board is unreachable" into "no such + // task", which is the answer that lets a pull overwrite a protected + // status. `find_task` keeps absence and failure apart. + Ok(self + .store + .find_task(task_id)? + .map(|found| found.task.status)) } fn apply(&mut self, op: &BoardOp) -> anyhow::Result> { diff --git a/crates/sheet-sync/src/board_plan.rs b/crates/sheet-sync/src/board_plan.rs index 4b106fb..bcef048 100644 --- a/crates/sheet-sync/src/board_plan.rs +++ b/crates/sheet-sync/src/board_plan.rs @@ -29,6 +29,7 @@ use crate::config::SheetMapping; use crate::model::{CanonicalField, CanonicalRow}; use crate::state::State; +use anyhow::Context; use forge_tasks::{CreateTaskRequest, TaskStatus, UpdateTaskPatch}; /// One planned board operation for a single sheet row. @@ -77,8 +78,8 @@ pub fn plan_pull( rows: &[CanonicalRow], mapping: &SheetMapping, state: &State, - existing_status: &dyn Fn(&str) -> Option, -) -> Vec { + existing_status: &dyn Fn(&str) -> anyhow::Result>, +) -> anyhow::Result> { rows.iter() .map(|row| plan_row(row, mapping, state, existing_status)) .collect() @@ -89,26 +90,29 @@ fn plan_row( row: &CanonicalRow, mapping: &SheetMapping, state: &State, - existing_status: &dyn Fn(&str) -> Option, -) -> BoardOp { + existing_status: &dyn Fn(&str) -> anyhow::Result>, +) -> anyhow::Result { let Some(entry) = state.rows.get(&row.id) else { - return BoardOp::Create { + return Ok(BoardOp::Create { row_id: row.id.clone(), req: build_create_request(row, mapping), target_status: target_status(row, mapping), - }; + }); }; let hash = content_hash(row); if entry.content_hash == hash { - return BoardOp::Skip { + return Ok(BoardOp::Skip { row_id: row.id.clone(), reason: "unchanged".to_string(), - }; + }); } - let protected = - existing_status(&entry.task_id).is_some_and(|status| PROTECTED_STATUSES.contains(&status)); + // Propagate: a status we could not read must stop the plan, not be treated + // as an unprotected task whose status the sheet may overwrite. + let protected = existing_status(&entry.task_id) + .with_context(|| format!("read current status of {}", entry.task_id))? + .is_some_and(|status| PROTECTED_STATUSES.contains(&status)); let status = if protected { None } else { @@ -127,11 +131,11 @@ fn plan_row( summary: None, }; - BoardOp::Update { + Ok(BoardOp::Update { row_id: row.id.clone(), task_id: entry.task_id.clone(), patch, - } + }) } /// Builds the `CreateTaskRequest` for a brand-new row. All fields not @@ -379,8 +383,8 @@ terminal_status = ["Verified/Closed", "Won't Fix", "Duplicate"] } } - fn no_task_exists(_task_id: &str) -> Option { - None + fn no_task_exists(_task_id: &str) -> anyhow::Result> { + Ok(None) } // -- build_create_request / target_status ------------------------------ @@ -396,7 +400,8 @@ terminal_status = ["Verified/Closed", "Won't Fix", "Duplicate"] &mapping, &state, &no_task_exists, - ); + ) + .unwrap(); assert_eq!(ops.len(), 1); match &ops[0] { @@ -474,7 +479,7 @@ terminal_status = ["Verified/Closed", "Won't Fix", "Duplicate"] }, ); - let ops = plan_pull(&[row], &mapping, &state, &no_task_exists); + let ops = plan_pull(&[row], &mapping, &state, &no_task_exists).unwrap(); assert_eq!(ops.len(), 1); match &ops[0] { BoardOp::Skip { row_id, reason } => { @@ -505,12 +510,12 @@ terminal_status = ["Verified/Closed", "Won't Fix", "Duplicate"] "Updated description".to_string(), ); - let existing_status = |task_id: &str| -> Option { + let existing_status = |task_id: &str| -> anyhow::Result> { assert_eq!(task_id, "t_existing01"); - Some(TaskStatus::InProgress) + Ok(Some(TaskStatus::InProgress)) }; - let ops = plan_pull(&[row], &mapping, &state, &existing_status); + let ops = plan_pull(&[row], &mapping, &state, &existing_status).unwrap(); assert_eq!(ops.len(), 1); match &ops[0] { BoardOp::Update { @@ -553,12 +558,12 @@ terminal_status = ["Verified/Closed", "Won't Fix", "Duplicate"] "Updated description".to_string(), ); - let existing_status = |task_id: &str| -> Option { + let existing_status = |task_id: &str| -> anyhow::Result> { assert_eq!(task_id, "t_existing03"); - Some(TaskStatus::Archived) + Ok(Some(TaskStatus::Archived)) }; - let ops = plan_pull(&[row], &mapping, &state, &existing_status); + let ops = plan_pull(&[row], &mapping, &state, &existing_status).unwrap(); assert_eq!(ops.len(), 1); match &ops[0] { BoardOp::Update { @@ -577,6 +582,44 @@ terminal_status = ["Verified/Closed", "Won't Fix", "Duplicate"] } } + /// The never-move-backward rule is only as good as the status read behind + /// it. When that read failed, `get_task(..).ok()` reported `None`, the task + /// looked unprotected, and the plan overwrote a `complete` or `archived` + /// status with the sheet's value. A read we could not perform must stop the + /// plan. + #[test] + fn a_status_read_that_fails_stops_the_plan_instead_of_overwriting_the_status() { + let mapping = qa_mapping(); + let mut row = qa_016(); + let mut state = State::default(); + state.upsert( + "QA-016".to_string(), + StateEntry { + task_id: "t_existing04".to_string(), + content_hash: content_hash(&row), + last_push_status: None, + }, + ); + row.fields.insert( + CanonicalField::Description, + "Updated description".to_string(), + ); + + let existing_status = |_task_id: &str| -> anyhow::Result> { + Err(anyhow::anyhow!( + "cannot reach the task board at 127.0.0.1:9042" + )) + }; + + let err = plan_pull(&[row], &mapping, &state, &existing_status) + .expect_err("an unreadable status must fail the plan, not be treated as unprotected"); + let rendered = format!("{err:#}"); + assert!( + rendered.contains("t_existing04"), + "the error should name the task whose status could not be read, got: {rendered}" + ); + } + #[test] fn changed_row_with_non_dev_owned_status_advances_status_from_map() { let mapping = qa_mapping(); @@ -595,9 +638,10 @@ terminal_status = ["Verified/Closed", "Won't Fix", "Duplicate"] "Updated description".to_string(), ); - let existing_status = |_task_id: &str| -> Option { Some(TaskStatus::Triage) }; + let existing_status = + |_task_id: &str| -> anyhow::Result> { Ok(Some(TaskStatus::Triage)) }; - let ops = plan_pull(&[row], &mapping, &state, &existing_status); + let ops = plan_pull(&[row], &mapping, &state, &existing_status).unwrap(); assert_eq!(ops.len(), 1); match &ops[0] { BoardOp::Update { patch, .. } => { diff --git a/crates/sheet-sync/src/sync.rs b/crates/sheet-sync/src/sync.rs index 8721ce0..548e0ce 100644 --- a/crates/sheet-sync/src/sync.rs +++ b/crates/sheet-sync/src/sync.rs @@ -104,7 +104,7 @@ pub fn pull( // before the apply loop takes a mutable borrow of `board`. let ops = plan_pull(&mapped.rows, mapping, &state, &|task_id| { board.existing_status(task_id) - }); + })?; let mut created = 0usize; let mut updated = 0usize; @@ -369,6 +369,20 @@ terminal_status = ["Verified/Closed", "Won't Fix", "Duplicate"] ]) } + /// `pull_fixture_grid` with every row's Description set to `desc`, so a + /// test can change sheet-side content between two pulls. + fn sheets_with_description(desc: &str) -> FakeSheets { + let mut grid = pull_fixture_grid(); + let col = qa_headers() + .iter() + .position(|h| h == "Description") + .expect("Description column"); + for row in &mut grid.rows { + row[col] = desc.to_string(); + } + FakeSheets::new(grid) + } + /// Fixture grid used by both pull tests: two fresh, importable rows /// (QA-010 "New", QA-011 "Triaged"), one terminal row (QA-012 /// "Verified/Closed"), and a QA-005 id collision across two rows. @@ -431,6 +445,54 @@ terminal_status = ["Verified/Closed", "Won't Fix", "Duplicate"] // -- pull: real ------------------------------------------------------- + /// End-to-end counterpart of + /// `board_plan::tests::a_status_read_that_fails_stops_the_plan_instead_of_overwriting_the_status`: + /// a pull over rows that are already joined to tasks must FAIL when the + /// board cannot report their current status, rather than plan status + /// overwrites from a board it could not read. + #[test] + fn pull_fails_when_the_board_cannot_report_current_status() { + let mapping = qa_mapping(); + let tmpdir = tempfile::tempdir().expect("tempdir creation"); + let state_path = tmpdir + .path() + .join(".forge") + .join("sheets") + .join("qa.state.toml"); + + // First pull joins QA-010/QA-011 to tasks and records their hashes. + let mut board = FakeBoard::new(); + pull( + &sheets_with_description("desc"), + &mut board, + &mapping, + &state_path, + &PullOptions { dry_run: false }, + ) + .expect("seed pull should succeed"); + + // Now the sheet content changes, so the plan must consult the board for + // each joined task's current status -- and the board cannot answer. + let mut broken = FakeBoard::new_failing_status_read(); + let err = pull( + &sheets_with_description("edited description"), + &mut broken, + &mapping, + &state_path, + &PullOptions { dry_run: false }, + ) + .expect_err("a board that cannot report status must fail the pull"); + + assert!( + format!("{err:#}").contains("status"), + "the error should say the status read failed, got: {err:#}" + ); + assert!( + broken.applied.is_empty(), + "nothing may be applied to the board once the plan failed" + ); + } + #[test] fn pull_real_applies_creates_and_persists_state() { let mapping = qa_mapping(); diff --git a/crates/sheet-sync/tests/board_exec_live.rs b/crates/sheet-sync/tests/board_exec_live.rs index 893f38a..80bd779 100644 --- a/crates/sheet-sync/tests/board_exec_live.rs +++ b/crates/sheet-sync/tests/board_exec_live.rs @@ -40,7 +40,7 @@ fn create_with_non_triage_target_status_lands_via_follow_up_update() { .expect("create returns a task id"); assert_eq!( - exec.existing_status(&task_id), + exec.existing_status(&task_id).expect("read status"), Some(TaskStatus::InProgress), "create + follow-up update should land the task at the mapped target status" ); diff --git a/crates/tasks/src/config.rs b/crates/tasks/src/config.rs index 77ca75b..37eb4f0 100644 --- a/crates/tasks/src/config.rs +++ b/crates/tasks/src/config.rs @@ -21,9 +21,18 @@ //! file. Project still beats global, so a repo can pin its own board. //! //! Blank/whitespace values at any layer are ignored and fall through. +//! +//! Every layer fails loud. A config file that exists but cannot be read or +//! parsed is an ERROR, not a silent fall-through to the next layer: falling +//! through sends the board query to `127.0.0.1:9042` -- a different database, or +//! none at all -- while the user is looking at a file that says otherwise. The +//! symptom is an empty or wrong board with no indication that the config they +//! wrote was never applied. A file that is simply absent is not an error; that +//! is what "not configured at this layer" means. use std::path::Path; +use anyhow::{Context, Result}; use serde::Deserialize; /// Built-in fallback when nothing else is configured. @@ -62,22 +71,46 @@ fn pick( .unwrap_or_else(|| DEFAULT_CQL_HOST.to_string()) } +/// Parse a config body. A syntax error is reported, never swallowed. +fn parse_config_toml(body: &str) -> Result { + toml::from_str::(body).context("invalid TOML") +} + /// Parse `cql_host` out of a `.forge/config.toml` body. Pure; testable. -fn parse_cql_host_toml(body: &str) -> Option { - toml::from_str::(body) - .ok() - .and_then(|c| c.cql_host) +fn parse_cql_host_toml(body: &str) -> Result> { + Ok(parse_config_toml(body)?.cql_host) +} + +/// Read a config file. `Ok(None)` when it does not exist; `Err` when it exists +/// and cannot be read (permissions, a directory in its place, an I/O fault) -- +/// those mean the user's settings were not applied, which they need to know. +fn read_config_file(path: &Path) -> Result> { + match std::fs::read_to_string(path) { + Ok(body) => Ok(Some(body)), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None), + Err(e) => Err(e).with_context(|| format!("read forge config {}", path.display())), + } } /// Walk up from `start` looking for `.forge/config.toml`; return its `cql_host`. -fn read_config_cql_host(start: &Path) -> Option { +fn read_config_cql_host(start: &Path) -> Result> { + match find_project_config(start)? { + Some((path, body)) => { + parse_cql_host_toml(&body).with_context(|| format!("in {}", path.display())) + } + None => Ok(None), + } +} + +/// The nearest `.forge/config.toml` walking up from `start`, with its body. +fn find_project_config(start: &Path) -> Result> { for dir in start.ancestors() { let candidate = dir.join(".forge").join("config.toml"); - if let Ok(body) = std::fs::read_to_string(&candidate) { - return parse_cql_host_toml(&body); + if let Some(body) = read_config_file(&candidate)? { + return Ok(Some((candidate, body))); } } - None + Ok(None) } /// The machine-wide config file: `~/.config/forge.toml`. @@ -89,19 +122,41 @@ pub fn global_config_path() -> Option { } /// Read the global config body, if there is one. -fn read_global_config() -> Option { - let path = global_config_path()?; - std::fs::read_to_string(path).ok() +fn read_global_config() -> Result> { + let Some(path) = global_config_path() else { + return Ok(None); + }; + Ok(read_config_file(&path)?.map(|body| (path, body))) +} + +/// Read an environment variable. Absent is `Ok(None)`; present but not UTF-8 +/// is an error, because the user set it and it is not being honoured. +fn env_var(name: &str) -> Result> { + match std::env::var(name) { + Ok(v) => Ok(Some(v)), + Err(std::env::VarError::NotPresent) => Ok(None), + Err(e) => Err(e).with_context(|| format!("read {name}")), + } } /// Resolve the effective CQL `host:port` for the task store. -pub fn resolve_cql_host(explicit: Option<&str>) -> String { - let env = std::env::var("FORGE_CQL_HOST").ok(); - let file = std::env::current_dir() - .ok() - .and_then(|cwd| read_config_cql_host(&cwd)); - let global = read_global_config().and_then(|body| parse_cql_host_toml(&body)); - pick(explicit, env, file, global) +pub fn resolve_cql_host(explicit: Option<&str>) -> Result { + let env = env_var("FORGE_CQL_HOST")?; + let file = match std::env::current_dir() { + Ok(cwd) => read_config_cql_host(&cwd)?, + // A cwd we cannot read is not "no project config" -- it is a question we + // failed to ask, and the answer decides which database is queried. + Err(e) => { + return Err(e).context("resolve the project forge config: read current directory") + } + }; + let global = match read_global_config()? { + Some((path, body)) => { + parse_cql_host_toml(&body).with_context(|| format!("in {}", path.display()))? + } + None => None, + }; + Ok(pick(explicit, env, file, global)) } /// Split a (possibly comma-separated) contact-point string into individual @@ -119,60 +174,71 @@ fn split_hosts(s: &str) -> Vec { /// bootstrap from whichever is up and fail over for queries, so the board /// survives a single node loss instead of dying with one fixed contact point. /// Always returns at least one entry (the resolved value, or [`DEFAULT_CQL_HOST`]). -pub fn resolve_cql_hosts(explicit: Option<&str>) -> Vec { - let hosts = split_hosts(&resolve_cql_host(explicit)); - if hosts.is_empty() { +pub fn resolve_cql_hosts(explicit: Option<&str>) -> Result> { + let hosts = split_hosts(&resolve_cql_host(explicit)?); + Ok(if hosts.is_empty() { vec![DEFAULT_CQL_HOST.to_string()] } else { hosts + }) +} + +/// Parse the `FORGE_DEBUG_STOP` value. Pure; testable without touching the +/// process environment. +/// +/// An unrecognised value is an error. Falling through to `false` left the +/// degraded-board alerting the operator had just switched on silently OFF -- +/// the state they were trying to leave. +fn parse_debug_stop_env(raw: &str) -> Result { + match raw.trim().to_ascii_lowercase().as_str() { + "1" | "true" | "yes" | "on" => Ok(true), + "0" | "false" | "no" | "off" => Ok(false), + other => anyhow::bail!( + "FORGE_DEBUG_STOP={other:?} is not a boolean (use 1/true/yes/on or 0/false/no/off)" + ), } } /// Parse `debug_stop` from a `.forge/config.toml` body. Pure; testable. -fn parse_debug_stop_toml(body: &str) -> Option { - toml::from_str::(body) - .ok() - .and_then(|c| c.debug_stop) +fn parse_debug_stop_toml(body: &str) -> Result> { + Ok(parse_config_toml(body)?.debug_stop) } /// Walk up from `start` for `.forge/config.toml`; return its `debug_stop`. -fn read_config_debug_stop(start: &Path) -> Option { - for dir in start.ancestors() { - let candidate = dir.join(".forge").join("config.toml"); - if let Ok(body) = std::fs::read_to_string(&candidate) { - return parse_debug_stop_toml(&body); +fn read_config_debug_stop(start: &Path) -> Result> { + match find_project_config(start)? { + Some((path, body)) => { + parse_debug_stop_toml(&body).with_context(|| format!("in {}", path.display())) } + None => Ok(None), } - None } /// Resolve whether `debug_stop` board alerting is on. Precedence: explicit tool /// arg → `FORGE_DEBUG_STOP` (1/true/yes) → `.forge/config.toml` `debug_stop` → /// `~/.config/forge.toml` `debug_stop` → false. The explicit arg lets the LLM flip it on per call when it suspects the /// board is degraded. -pub fn resolve_debug_stop(explicit: Option) -> bool { +pub fn resolve_debug_stop(explicit: Option) -> Result { if let Some(b) = explicit { - return b; + return Ok(b); } - if let Ok(raw) = std::env::var("FORGE_DEBUG_STOP") { - match raw.trim().to_ascii_lowercase().as_str() { - "1" | "true" | "yes" | "on" => return true, - "0" | "false" | "no" | "off" => return false, - _ => {} - } + if let Some(raw) = env_var("FORGE_DEBUG_STOP")? { + return parse_debug_stop_env(&raw); } - if let Some(project) = std::env::current_dir() - .ok() - .and_then(|cwd| read_config_debug_stop(&cwd)) - { - return project; + let cwd = std::env::current_dir() + .context("resolve the project forge config: read current directory")?; + if let Some(project) = read_config_debug_stop(&cwd)? { + return Ok(project); } // Same file, same schema, same precedence as cql_host -- a setting that // honoured the global config for one key and ignored it for the other would // be the more surprising design. - read_global_config() - .and_then(|body| parse_debug_stop_toml(&body)) - .unwrap_or(false) + match read_global_config()? { + Some((path, body)) => Ok(parse_debug_stop_toml(&body) + .with_context(|| format!("in {}", path.display()))? + .unwrap_or(false)), + None => Ok(false), + } } #[cfg(test)] @@ -291,10 +357,10 @@ mod tests { // mean two formats to document and keep in step. let body = "cql_host = \"127.0.0.1:47017\"\ndebug_stop = true\n"; assert_eq!( - parse_cql_host_toml(body).as_deref(), + parse_cql_host_toml(body).unwrap().as_deref(), Some("127.0.0.1:47017") ); - assert_eq!(parse_debug_stop_toml(body), Some(true)); + assert_eq!(parse_debug_stop_toml(body).unwrap(), Some(true)); } #[test] @@ -311,34 +377,55 @@ mod tests { #[test] fn resolve_hosts_always_nonempty() { // A single explicit host yields one contact point; the list form yields many. - assert_eq!(resolve_cql_hosts(Some("h:1")), vec!["h:1"]); + assert_eq!(resolve_cql_hosts(Some("h:1")).unwrap(), vec!["h:1"]); assert_eq!( - resolve_cql_hosts(Some("n1:19042,n2:19042,n3:19042")), + resolve_cql_hosts(Some("n1:19042,n2:19042,n3:19042")).unwrap(), vec!["n1:19042", "n2:19042", "n3:19042"] ); } #[test] fn parses_debug_stop_and_explicit_wins() { - assert_eq!(parse_debug_stop_toml("debug_stop = true\n"), Some(true)); - assert_eq!(parse_debug_stop_toml("debug_stop = false\n"), Some(false)); - assert_eq!(parse_debug_stop_toml("cql_host = \"h:1\"\n"), None); + assert_eq!( + parse_debug_stop_toml("debug_stop = true\n").unwrap(), + Some(true) + ); + assert_eq!( + parse_debug_stop_toml("debug_stop = false\n").unwrap(), + Some(false) + ); + assert_eq!(parse_debug_stop_toml("cql_host = \"h:1\"\n").unwrap(), None); // explicit arg short-circuits env/file - assert!(resolve_debug_stop(Some(true))); - assert!(!resolve_debug_stop(Some(false))); + assert!(resolve_debug_stop(Some(true)).unwrap()); + assert!(!resolve_debug_stop(Some(false)).unwrap()); + } + + /// A typo in `FORGE_DEBUG_STOP` used to fall through to `false`, silently + /// disabling the degraded-board alerting the operator had just switched on. + /// Asserted on the pure parser: mutating the process environment would race + /// the other tests in this binary. + #[test] + fn an_unparseable_debug_stop_value_is_an_error_not_a_silent_off() { + assert!(parse_debug_stop_env("1").unwrap()); + assert!(!parse_debug_stop_env(" OFF ").unwrap()); + let err = + parse_debug_stop_env("sure").expect_err("a non-boolean value must not be read as off"); + assert!(format!("{err:#}").contains("FORGE_DEBUG_STOP")); } #[test] fn parses_cql_host_from_toml() { assert_eq!( - parse_cql_host_toml("cql_host = \"127.0.0.1:19042\"\n").as_deref(), + parse_cql_host_toml("cql_host = \"127.0.0.1:19042\"\n") + .unwrap() + .as_deref(), Some("127.0.0.1:19042") ); } #[test] fn toml_without_cql_host_is_none() { - assert_eq!(parse_cql_host_toml("other = 1\n"), None); + assert_eq!(parse_cql_host_toml("other = 1\n").unwrap(), None); } #[test] @@ -354,6 +441,60 @@ mod tests { .unwrap(); let got = read_config_cql_host(&sub); std::fs::remove_dir_all(&base).ok(); - assert_eq!(got.as_deref(), Some("10.0.0.1:9999")); + assert_eq!(got.unwrap().as_deref(), Some("10.0.0.1:9999")); + } + + /// A `.forge/config.toml` with a syntax error used to be discarded by + /// `toml::from_str(..).ok()`, so forge queried `127.0.0.1:9042` while the + /// user was looking at a file naming a different port. The board then read + /// as empty (or as somebody else's), with nothing to say the file had been + /// ignored -- the same "I could not look" reported as "there is nothing". + #[test] + fn a_malformed_config_is_an_error_not_a_silent_fallback_to_the_default_host() { + let err = parse_cql_host_toml("cql_host = \n") + .expect_err("a malformed config must not be silently discarded"); + assert!( + format!("{err:#}").contains("TOML"), + "the error should say the file is not valid TOML, got: {err:#}" + ); + assert!(parse_debug_stop_toml("debug_stop = maybe\n").is_err()); + } + + #[test] + fn a_config_file_that_is_absent_is_not_an_error() { + // Absent means "not configured at this layer", which is the normal case + // and must keep falling through to the next one. + let missing = std::env::temp_dir().join("forge_no_such_config_file.toml"); + assert!(read_config_file(&missing).unwrap().is_none()); + } + + #[test] + fn a_config_file_that_cannot_be_read_is_an_error() { + // A directory where the file should be: it exists, so "not configured" + // is the wrong conclusion, and reading it fails with something other + // than NotFound. + let base = std::env::temp_dir().join(format!("forge_unreadable_{}", std::process::id())); + let path = base.join("config.toml"); + std::fs::create_dir_all(&path).unwrap(); + let got = read_config_file(&path); + std::fs::remove_dir_all(&base).ok(); + assert!( + got.is_err(), + "a config path that exists but cannot be read must be an error" + ); + } + + #[test] + fn a_malformed_project_config_names_the_file_it_came_from() { + let base = std::env::temp_dir().join(format!("forge_badcfg_{}", std::process::id())); + std::fs::create_dir_all(base.join(".forge")).unwrap(); + std::fs::write(base.join(".forge").join("config.toml"), "cql_host = \n").unwrap(); + let got = read_config_cql_host(&base); + std::fs::remove_dir_all(&base).ok(); + let err = got.expect_err("a malformed project config must be an error"); + assert!( + format!("{err:#}").contains("config.toml"), + "the error must name the offending file, got: {err:#}" + ); } } diff --git a/crates/tasks/src/schema.rs b/crates/tasks/src/schema.rs index 6c5cc83..98d14fd 100644 --- a/crates/tasks/src/schema.rs +++ b/crates/tasks/src/schema.rs @@ -1,5 +1,9 @@ //! CQL schema for the task tables (idempotent CREATE TABLE IF NOT EXISTS). +/// The keyspace the board lives in. Named in errors so "I connected but could +/// not read the board" says *which* keyspace was missing. +pub const BOARD_KEYSPACE: &str = "agent_memory"; + pub const CREATE_TASKS_TABLE: &str = " CREATE TABLE IF NOT EXISTS agent_memory.tasks ( tenant_id uuid, diff --git a/crates/tasks/src/store.rs b/crates/tasks/src/store.rs index dd4a32c..b2db08b 100644 --- a/crates/tasks/src/store.rs +++ b/crates/tasks/src/store.rs @@ -15,7 +15,9 @@ use std::sync::Arc; use anyhow::{anyhow, Context, Result}; use uuid::Uuid; -use crate::schema::{CREATE_TASKS_TABLE, CREATE_TASK_COMMENTS_TABLE, CREATE_TASK_LINKS_TABLE}; +use crate::schema::{ + BOARD_KEYSPACE, CREATE_TASKS_TABLE, CREATE_TASK_COMMENTS_TABLE, CREATE_TASK_LINKS_TABLE, +}; use crate::types::{ Comment, CreateTaskRequest, KanbanBoard, KanbanColumns, Task, TaskFilter, TaskStatus, TaskWithLinks, UpdateTaskPatch, @@ -24,6 +26,14 @@ use crate::types::{ /// Fixed tenant UUID for the single-user forge setup. const TENANT_ID: &str = "00000000-0000-0000-0000-000000000001"; +/// Applied when a row's `priority` is NULL (a value the writer never stores, but +/// which an externally-inserted row may carry). Mirrors `create_task`'s default. +const DEFAULT_PRIORITY: i32 = 50; + +/// Applied when a row's `created_by` is NULL. A label, not a decision input, so +/// a documented placeholder is better than failing an otherwise-readable board. +const DEFAULT_CREATED_BY: &str = "agent"; + /// Escape a string for inline CQL (double any single quotes). fn esc(s: &str) -> String { s.replace('\'', "''") @@ -124,8 +134,18 @@ impl TaskStore { .build() .context("build tokio runtime")?; - anyhow::ensure!(!cql_hosts.is_empty(), "no CQL contact points provided"); + anyhow::ensure!( + !cql_hosts.is_empty(), + "no CQL contact point configured for the task board -- set --cql-host, \ + FORGE_CQL_HOST, or cql_host in .forge/config.toml / ~/.config/forge.toml" + ); let hosts = cql_hosts.to_vec(); + // Every failure below names the contact points actually tried. Without + // it the operator is told "connect to CQL: Connection refused" and has + // no way to know WHICH host that was -- the flag, the environment + // variable, the project config or the global config may each have + // supplied it. + let contacted = hosts.join(", "); let session: scylla::Session = rt .block_on(async { scylla::SessionBuilder::new() @@ -134,7 +154,9 @@ impl TaskStore { .build() .await }) - .context("connect to CQL")?; + .with_context(|| { + format!("cannot reach the task board at {contacted} (CQL contact points tried)") + })?; let session = Arc::new(session); let store = Self { @@ -142,7 +164,13 @@ impl TaskStore { session, tenant_id: tenant_id.unwrap_or(TENANT_ID).to_string(), }; - store.ensure_schema()?; + store.ensure_schema().with_context(|| { + format!( + "connected to {contacted}, but the task board keyspace '{BOARD_KEYSPACE}' is not \ + usable there -- create the keyspace, or point the board at the database that \ + holds it" + ) + })?; Ok(store) } @@ -173,7 +201,7 @@ impl TaskStore { CREATE_TASK_COMMENTS_TABLE, ] { cql_exec!(self.rt, &self.session, stmt.to_string()) - .with_context(|| format!("ensure_schema: {}", &stmt[..50]))?; + .with_context(|| format!("ensure_schema: {}", first_line(stmt)))?; } Ok(()) } @@ -313,9 +341,25 @@ impl TaskStore { Ok(task) } - /// Fetch a task with its links and recent comments. + /// Fetch a task with its links and recent comments. Errors when no such + /// task exists; see [`TaskStore::find_task`] when absence is a legitimate + /// answer rather than a failure. pub fn get_task(&self, task_id: &str) -> Result { - let task = self.fetch_task_row(task_id)?; + self.find_task(task_id)? + .ok_or_else(|| anyhow!("Task not found: {}", task_id)) + } + + /// Fetch a task, distinguishing "no such task" (`Ok(None)`) from "I could + /// not read the board" (`Err`). + /// + /// `get_task(..).ok()` collapsed both into `None`, so a caller asking + /// "does this task already exist?" during an outage was told "no" and acted + /// on it -- creating a duplicate, or moving a finished task backwards + /// because its protected status could not be read. + pub fn find_task(&self, task_id: &str) -> Result> { + let Some(task) = self.fetch_task_row(task_id)? else { + return Ok(None); + }; // Fetch links where this task is the source let links_cql = format!( @@ -332,7 +376,10 @@ impl TaskStore { let mut parents = Vec::new(); let mut children = Vec::new(); - for row in links_result.rows_or_empty() { + let link_rows = links_result + .rows() + .context("get_task: expected link rows")?; + for row in link_rows { let (_, link_type, dst): (String, String, String) = row.into_typed().context("get_task: parse link row")?; if link_type == "parent" { @@ -364,20 +411,17 @@ impl TaskStore { let mut recent_comments = Vec::new(); let mut comment_seq: u32 = 0; - for row in comments_result.rows_or_empty() { - let vals = row.columns; - let author = match vals.first().and_then(|v| v.as_ref()) { - Some(scylla::frame::response::result::CqlValue::Text(s)) => s.clone(), - _ => String::new(), - }; - let body = match vals.get(1).and_then(|v| v.as_ref()) { - Some(scylla::frame::response::result::CqlValue::Text(s)) => s.clone(), - _ => String::new(), - }; - let created_at = match vals.get(2).and_then(|v| v.as_ref()) { - Some(scylla::frame::response::result::CqlValue::BigInt(i)) => *i, - _ => 0, - }; + let comment_rows = comments_result + .rows() + .context("get_task: expected comment rows")?; + for row in comment_rows { + // A comment whose columns do not decode is schema drift, not an + // empty comment: blanking the author and body used to publish a + // row that says nothing and looks deliberate. + let mut vals = row.columns.into_iter(); + let author = col_str(vals.next(), "task_comments.author")?; + let body = col_str(vals.next(), "task_comments.body")?; + let created_at = col_i64(vals.next(), "task_comments.created_at")?; comment_seq += 1; recent_comments.push(Comment { comment_id: format!("c{}", comment_seq), @@ -387,12 +431,12 @@ impl TaskStore { }); } - Ok(TaskWithLinks { + Ok(Some(TaskWithLinks { task, parents, children, recent_comments, - }) + })) } /// List tasks, optionally filtered by status, assignee, and priority range. @@ -431,12 +475,8 @@ impl TaskStore { .into_legacy_result() .context("list_tasks: legacy result")?; - let mut tasks = Vec::new(); - for row in result.rows_or_empty() { - if let Ok(task) = parse_task_row(row) { - tasks.push(task); - } - } + let rows = result.rows().context("list_tasks: expected rows")?; + let mut tasks = parse_task_rows(rows).context("list_tasks: parse rows")?; sort_newest_first(&mut tasks); tasks.truncate(limit); Ok(tasks) @@ -561,22 +601,23 @@ impl TaskStore { .into_legacy_result() .context("board: legacy result")?; + let rows = result.rows().context("board: expected rows")?; + let tasks = parse_task_rows(rows).context("board: parse rows")?; + let mut triage = Vec::new(); let mut ready = Vec::new(); let mut in_progress = Vec::new(); let mut blocked = Vec::new(); let mut complete = Vec::new(); - for row in result.rows_or_empty() { - if let Ok(task) = parse_task_row(row) { - match task.status { - TaskStatus::Triage => triage.push(task), - TaskStatus::Ready => ready.push(task), - TaskStatus::InProgress => in_progress.push(task), - TaskStatus::Blocked => blocked.push(task), - TaskStatus::Complete => complete.push(task), - TaskStatus::Archived => {} - } + for task in tasks { + match task.status { + TaskStatus::Triage => triage.push(task), + TaskStatus::Ready => ready.push(task), + TaskStatus::InProgress => in_progress.push(task), + TaskStatus::Blocked => blocked.push(task), + TaskStatus::Complete => complete.push(task), + TaskStatus::Archived => {} } } @@ -608,7 +649,8 @@ impl TaskStore { // Internals // ----------------------------------------------------------------------- - fn fetch_task_row(&self, task_id: &str) -> Result { + /// The task's own row, or `Ok(None)` when the board holds no such task. + fn fetch_task_row(&self, task_id: &str) -> Result> { let cql = format!( "SELECT task_id, title, body, status, assignee, reviewer, priority, \ workspace_kind, workspace_path, created_by, block_reason, result, summary, \ @@ -625,11 +667,12 @@ impl TaskStore { .context("fetch_task_row: legacy result")?; let rows = result.rows().context("fetch_task_row: expected rows")?; - let row = rows - .into_iter() - .next() - .ok_or_else(|| anyhow!("Task not found: {}", task_id))?; - parse_task_row(row) + match rows.into_iter().next() { + None => Ok(None), + Some(row) => parse_task_row(row) + .with_context(|| format!("task {task_id}")) + .map(Some), + } } } @@ -637,28 +680,50 @@ impl TaskStore { // Row parsing // --------------------------------------------------------------------------- +/// Parse every row of a board read, or fail the whole read. +/// +/// Dropping the rows that would not parse (`if let Ok(task) = ...`) meant a +/// board that had drifted from this schema simply reported fewer tasks, with no +/// error and no count to compare against. That is the same failure as an empty +/// answer from a dead host, one row at a time: the caller cannot tell "this task +/// does not exist" from "I could not decode it". +fn parse_task_rows(rows: Vec) -> Result> { + rows.into_iter() + .enumerate() + .map(|(i, row)| parse_task_row(row).with_context(|| format!("row {i}"))) + .collect() +} + /// Parse a CQL row into a `Task`. /// /// Column order must match SELECT list in `fetch_task_row` / `list_tasks` / `board`. +/// +/// Strict on purpose. A NULL in a nullable column is *data* and gets the +/// documented default; a column that is missing or of the wrong CQL type is +/// schema drift, and is an error rather than a plausible-looking task. fn parse_task_row(row: scylla::frame::response::result::Row) -> Result { - use scylla::frame::response::result::CqlValue; - let mut cols = row.columns.into_iter(); let task_id = col_str(cols.next(), "task_id")?; let title = col_str(cols.next(), "title")?; let body = col_opt_str(cols.next()); let status_str = col_str(cols.next(), "status")?; - let status = TaskStatus::parse(&status_str).unwrap_or(TaskStatus::Triage); + // An unrecognised status used to be silently filed under `triage`, which + // put the task in the wrong kanban column and, for a `complete` task, + // resurrected it as outstanding work. + let status = TaskStatus::parse(&status_str).ok_or_else(|| { + anyhow!( + "task {task_id}: unknown status {status_str:?} -- the board schema is \ + ahead of this build of forge" + ) + })?; let assignee = col_opt_str(cols.next()); let reviewer = col_opt_str(cols.next()); - let priority = match cols.next().flatten() { - Some(CqlValue::Int(i)) => i, - _ => 50, - }; + let priority = col_i32_or(cols.next(), "priority", DEFAULT_PRIORITY)?; let workspace_kind = col_opt_str(cols.next()); let workspace_path = col_opt_str(cols.next()); - let created_by = col_str(cols.next(), "created_by").unwrap_or_else(|_| "agent".to_string()); + let created_by = col_opt_str_strict(cols.next(), "created_by")? + .unwrap_or_else(|| DEFAULT_CREATED_BY.to_string()); let block_reason = col_opt_str(cols.next()); let result_val = col_opt_str(cols.next()); let summary = col_opt_str(cols.next()); @@ -666,14 +731,10 @@ fn parse_task_row(row: scylla::frame::response::result::Row) -> Result { .map(|s| serde_json::from_str(&s).unwrap_or(serde_json::Value::String(s))); let _skills = cols.next(); // set — skipped for now let _related = cols.next(); // set — skipped for now - let created_at = match cols.next().flatten() { - Some(CqlValue::BigInt(i)) => i, - _ => 0, - }; - let updated_at = match cols.next().flatten() { - Some(CqlValue::BigInt(i)) => i, - _ => 0, - }; + // Timestamps decide the newest-first order every read depends on. A row we + // cannot time would sort to the bottom and quietly fall out of any window. + let created_at = col_i64(cols.next(), "created_at")?; + let updated_at = col_i64(cols.next(), "updated_at")?; Ok(Task { task_id, @@ -716,6 +777,61 @@ fn col_opt_str(v: Option>) -> } } +/// `Ok(None)` for a CQL NULL, `Err` for a missing column or a non-text type. +/// [`col_opt_str`] cannot tell those apart, which is fine where the column is +/// genuinely optional and wrong where a decode failure would masquerade as one. +fn col_opt_str_strict( + v: Option>, + name: &str, +) -> Result> { + use scylla::frame::response::result::CqlValue; + match v { + None => Err(anyhow!("Missing column: {}", name)), + Some(None) => Ok(None), + Some(Some(CqlValue::Text(s) | CqlValue::Ascii(s))) => Ok(Some(s)), + Some(Some(other)) => Err(anyhow!( + "Column {} is not text: {:?}", + name, + std::mem::discriminant(&other) + )), + } +} + +/// A CQL `int`, defaulting when NULL but failing on a missing or mistyped column. +fn col_i32_or( + v: Option>, + name: &str, + default: i32, +) -> Result { + use scylla::frame::response::result::CqlValue; + match v { + None => Err(anyhow!("Missing column: {}", name)), + Some(None) => Ok(default), + Some(Some(CqlValue::Int(i))) => Ok(i), + Some(Some(_)) => Err(anyhow!("Column {} is not an int", name)), + } +} + +/// A CQL `bigint`. Missing, NULL, or mistyped are all errors: every caller uses +/// these for ordering, and a fabricated 0 sorts the row out of sight. +fn col_i64( + v: Option>, + name: &str, +) -> Result { + use scylla::frame::response::result::CqlValue; + match v.flatten() { + Some(CqlValue::BigInt(i)) => Ok(i), + Some(_) => Err(anyhow!("Column {} is not a bigint", name)), + None => Err(anyhow!("Missing or null column: {}", name)), + } +} + +/// First non-blank line of a CQL statement, for error context. Slicing the raw +/// first 50 bytes panics on a multi-byte boundary and reads as noise. +fn first_line(stmt: &str) -> &str { + stmt.trim().lines().next().unwrap_or(stmt).trim() +} + // --------------------------------------------------------------------------- // Helpers // --------------------------------------------------------------------------- @@ -728,3 +844,141 @@ fn format_set_text(items: &[String]) -> String { let inner: Vec = items.iter().map(|s| format!("'{}'", esc(s))).collect(); format!("{{{}}}", inner.join(", ")) } + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +#[cfg(test)] +mod tests { + use super::*; + use scylla::frame::response::result::{CqlValue, Row}; + + /// A well-formed `tasks` row in the column order every SELECT uses. + fn valid_row() -> Row { + Row { + columns: vec![ + Some(CqlValue::Text("t_00000001".into())), // task_id + Some(CqlValue::Text("a title".into())), // title + Some(CqlValue::Text("a body".into())), // body + Some(CqlValue::Text("ready".into())), // status + None, // assignee + None, // reviewer + Some(CqlValue::Int(70)), // priority + None, // workspace_kind + None, // workspace_path + Some(CqlValue::Text("agent".into())), // created_by + None, // block_reason + None, // result + None, // summary + None, // metadata + None, // skills + None, // related_entity_ids + Some(CqlValue::BigInt(1_700_000_000_000)), // created_at + Some(CqlValue::BigInt(1_700_000_000_001)), // updated_at + ], + } + } + + fn row_with(index: usize, value: Option) -> Row { + let mut row = valid_row(); + row.columns[index] = value; + row + } + + const STATUS: usize = 3; + const PRIORITY: usize = 6; + const CREATED_BY: usize = 9; + const CREATED_AT: usize = 16; + + #[test] + fn a_well_formed_row_parses() { + let task = parse_task_row(valid_row()).expect("valid row"); + assert_eq!(task.task_id, "t_00000001"); + assert_eq!(task.status, TaskStatus::Ready); + assert_eq!(task.priority, 70); + assert_eq!(task.created_at, 1_700_000_000_000); + } + + /// The board used to file an unrecognised status under `triage`. A task the + /// cluster considers `complete` then reappeared as outstanding work, and a + /// board written by a newer forge was silently mis-columned by an older one. + #[test] + fn an_unknown_status_is_an_error_not_a_task_filed_under_triage() { + let err = parse_task_row(row_with(STATUS, Some(CqlValue::Text("in_review".into())))) + .expect_err("an unknown status must not be guessed at"); + let rendered = format!("{err:#}"); + assert!( + rendered.contains("in_review") && rendered.contains("t_00000001"), + "the error should name the status and the task, got: {rendered}" + ); + } + + #[test] + fn a_mistyped_column_is_an_error_not_a_default_value() { + // Schema drift, not data: `priority` is an int, and reading a text there + // as 50 invents a value the board never stored. + assert!(parse_task_row(row_with(PRIORITY, Some(CqlValue::Text("high".into())))).is_err()); + assert!(parse_task_row(row_with(CREATED_BY, Some(CqlValue::Int(1)))).is_err()); + } + + #[test] + fn a_missing_timestamp_is_an_error_because_it_decides_the_read_order() { + // created_at = 0 sorts to the bottom of every newest-first read, so a row + // we cannot time silently drops out of any limited window. + let err = parse_task_row(row_with(CREATED_AT, None)) + .expect_err("a row without a created_at must not be fabricated as epoch 0"); + assert!(format!("{err:#}").contains("created_at")); + } + + #[test] + fn a_short_row_is_an_error_not_a_partially_filled_task() { + let mut row = valid_row(); + row.columns.truncate(4); + assert!(parse_task_row(row).is_err()); + } + + #[test] + fn a_null_in_a_nullable_column_keeps_its_documented_default() { + // NULL is data, not a decode failure: the row is readable and the + // documented default applies. + let task = parse_task_row(row_with(PRIORITY, None)).expect("null priority is readable"); + assert_eq!(task.priority, DEFAULT_PRIORITY); + let task = parse_task_row(row_with(CREATED_BY, None)).expect("null created_by is readable"); + assert_eq!(task.created_by, DEFAULT_CREATED_BY); + } + + /// The board read used to be `if let Ok(task) = parse_task_row(row)`, so a + /// row it could not decode simply did not appear. The caller got a shorter + /// list and no error -- the same "I could not look" reported as "there is + /// nothing", one task at a time. + #[test] + fn one_unreadable_row_fails_the_whole_read_instead_of_vanishing_from_it() { + let rows = vec![ + valid_row(), + row_with(STATUS, Some(CqlValue::Text("nonsense".into()))), + valid_row(), + ]; + let err = parse_task_rows(rows).expect_err("an undecodable row must fail the read"); + assert!( + format!("{err:#}").contains("row 1"), + "the error should say which row failed, got: {err:#}" + ); + } + + #[test] + fn all_readable_rows_are_returned() { + let tasks = parse_task_rows(vec![valid_row(), valid_row()]).expect("both rows readable"); + assert_eq!(tasks.len(), 2); + } + + #[test] + fn first_line_of_a_statement_is_used_for_error_context() { + // The old context sliced 50 raw bytes, which panics on a multi-byte + // boundary and reads as truncated noise. + assert_eq!( + first_line(CREATE_TASKS_TABLE), + "CREATE TABLE IF NOT EXISTS agent_memory.tasks (" + ); + } +} diff --git a/crates/tasks/tests/board_reads_fail_loud.rs b/crates/tasks/tests/board_reads_fail_loud.rs new file mode 100644 index 0000000..b74538e --- /dev/null +++ b/crates/tasks/tests/board_reads_fail_loud.rs @@ -0,0 +1,119 @@ +//! A read against a board that cannot be reached must be an **error**, never an +//! empty result. +//! +//! The bug these tests lock out: `forge task list` printed `[]` and exited 0 +//! when nothing was listening on the CQL port. A *query* answered "there is +//! nothing" when the truth was "I could not look", and every consumer of the +//! board (`/whats-next`, `/roadmap`, the defer-capture hook) was confidently +//! wrong in the same direction. +//! +//! No live cluster is needed: every case here is a socket the test owns, so +//! these run in CI alongside the unit tests. + +use std::io::Read; +use std::net::TcpListener; +use std::time::{Duration, Instant}; + +use forge_tasks::TaskStore; + +/// A `127.0.0.1:` that nothing is listening on: bind to get a +/// kernel-assigned port, then drop the listener so a connect is refused. +fn dead_contact_point() -> String { + let listener = TcpListener::bind("127.0.0.1:0").expect("bind ephemeral port"); + let addr = listener.local_addr().expect("local_addr").to_string(); + drop(listener); + addr +} + +/// A live TCP port that is not a CQL server: it accepts, reads nothing useful, +/// and hangs up. This is the shape of a stale port-forward (podman/gvproxy +/// keeps the port open after the container behind it dies), which is worse than +/// a refused connection precisely because the socket looks healthy. +fn non_cql_listener() -> (String, std::thread::JoinHandle<()>) { + let listener = TcpListener::bind("127.0.0.1:0").expect("bind ephemeral port"); + let addr = listener.local_addr().expect("local_addr").to_string(); + let handle = std::thread::spawn(move || { + // Serve a bounded number of connections and then stop: the driver + // retries, and an unbounded accept loop would never join. + for _ in 0..64 { + match listener.accept() { + Ok((mut sock, _)) => { + let mut scratch = [0u8; 64]; + let _ = sock.read(&mut scratch); + drop(sock); + } + Err(_) => return, + } + } + }); + (addr, handle) +} + +/// Connect and require failure, returning the fully rendered error chain. +/// `TaskStore` is not `Debug`, so `expect_err` is unavailable here. +fn connect_error(hosts: &[String], why: &str) -> String { + match TaskStore::connect(hosts, None) { + Ok(_) => panic!("{why}"), + Err(e) => format!("{e:#}"), + } +} + +#[test] +fn connecting_to_a_dead_port_is_an_error_that_names_the_contact_point() { + let addr = dead_contact_point(); + let rendered = connect_error( + std::slice::from_ref(&addr), + "an unreachable contact point must be an error, not a store whose reads look empty", + ); + assert!( + rendered.contains(&addr), + "the error must name the host and port actually tried, got: {rendered}" + ); +} + +#[test] +fn connecting_to_a_live_port_that_is_not_cql_is_an_error_that_names_the_contact_point() { + let (addr, handle) = non_cql_listener(); + let started = Instant::now(); + let rendered = connect_error( + std::slice::from_ref(&addr), + "a port that is open but not speaking CQL must be an error", + ); + assert!( + rendered.contains(&addr), + "the error must name the host and port actually tried, got: {rendered}" + ); + assert!( + started.elapsed() < Duration::from_secs(60), + "a non-CQL endpoint must fail within a bounded time, not hang" + ); + drop(handle); +} + +#[test] +fn every_dead_contact_point_is_named_so_the_operator_knows_what_was_tried() { + // With several bootstrap contact points the message has to say which set was + // attempted; naming only the first would send the reader to the wrong host. + let a = dead_contact_point(); + let b = dead_contact_point(); + let rendered = connect_error( + &[a.clone(), b.clone()], + "all contact points unreachable must be an error", + ); + assert!( + rendered.contains(&a) && rendered.contains(&b), + "the error must name every contact point tried, got: {rendered}" + ); +} + +#[test] +fn an_empty_contact_point_list_is_an_error_not_a_silent_default() { + let rendered = connect_error( + &[], + "no contact points is a configuration error, not an empty board", + ); + assert!( + rendered.contains("contact point"), + "the error should say what is missing, got: {rendered}" + ); +}