From 2d326566184e3680d214028cd07da5c916fdff62 Mon Sep 17 00:00:00 2001 From: Steven Enamakel <31011319+senamakel@users.noreply.github.com> Date: Tue, 28 Jul 2026 11:09:42 +0300 Subject: [PATCH] refactor(todos): cut over task boards to TinyAgents (#5239) --- src/core/runtime/services.rs | 6 +- src/openhuman/agent/message_convert.rs | 4 + src/openhuman/agent/task_board.rs | 308 ++---- src/openhuman/agent/tools/todo.rs | 19 +- src/openhuman/tinyagents/mod.rs | 1 + src/openhuman/tinyagents/todos.rs | 148 +++ src/openhuman/todos/README.md | 98 +- src/openhuman/todos/crate_adapter.rs | 555 ---------- src/openhuman/todos/invariant_tests.rs | 750 -------------- src/openhuman/todos/mod.rs | 6 - src/openhuman/todos/ops.rs | 1283 +++--------------------- src/openhuman/todos/runs.rs | 2 +- src/openhuman/todos/store.rs | 44 - vendor/tinyagents | 2 +- 14 files changed, 417 insertions(+), 2809 deletions(-) create mode 100644 src/openhuman/tinyagents/todos.rs delete mode 100644 src/openhuman/todos/crate_adapter.rs delete mode 100644 src/openhuman/todos/invariant_tests.rs delete mode 100644 src/openhuman/todos/store.rs diff --git a/src/core/runtime/services.rs b/src/core/runtime/services.rs index ea746fcbc..2db9d80e7 100644 --- a/src/core/runtime/services.rs +++ b/src/core/runtime/services.rs @@ -387,10 +387,8 @@ async fn run_legacy_migrations(config: &Config) { // `graph.todos` store, which is now authoritative. Idempotent and returns // fast on an empty/absent legacy dir (the `*.runs.json` ledger stays local). // As above, each core boot must inspect its own workspace. - match crate::openhuman::todos::crate_adapter::migrate_legacy_task_boards_into_crate_store( - &config.workspace_dir, - ) - .await + match crate::openhuman::tinyagents::todos::migrate_legacy_task_boards(&config.workspace_dir) + .await { Ok(report) if report.total > 0 => { log::info!( diff --git a/src/openhuman/agent/message_convert.rs b/src/openhuman/agent/message_convert.rs index d7e17b91f..2e717afaa 100644 --- a/src/openhuman/agent/message_convert.rs +++ b/src/openhuman/agent/message_convert.rs @@ -114,6 +114,7 @@ pub(crate) fn chat_message_to_message(msg: &ChatMessage) -> Message { Message::Tool(ToolMessage { tool_call_id, content: vec![ContentBlock::Text(content)], + trusted_verbatim: false, }) } // "user" and any unrecognized role default to a user turn — the safest @@ -446,6 +447,7 @@ mod tests { panic!("expected Tool, got {t:?}"); }; assert_eq!(tm.tool_call_id, "call-1"); + assert!(!tm.trusted_verbatim); assert_eq!(t.text(), "echoed:hi"); // Outbound: re-serialized to a well-formed native tool round (assistant @@ -554,6 +556,7 @@ mod tests { let messages = vec![Message::Tool(ToolMessage { tool_call_id: "call-7".into(), content: vec![ContentBlock::Text("done".into())], + trusted_verbatim: false, })]; let back = messages_to_history(&messages); assert_eq!(back[0].role, "tool"); @@ -581,6 +584,7 @@ mod tests { Message::Tool(ToolMessage { tool_call_id: "c1".into(), content: vec![ContentBlock::Text("echoed:hi".into())], + trusted_verbatim: false, }), Message::Assistant(AssistantMessage { id: None, diff --git a/src/openhuman/agent/task_board.rs b/src/openhuman/agent/task_board.rs index f317d20d1..e8e368c39 100644 --- a/src/openhuman/agent/task_board.rs +++ b/src/openhuman/agent/task_board.rs @@ -5,132 +5,21 @@ //! not the retired `/agent_task_boards/.json` //! file-JSON tree. [`TaskBoardStore`] is a thin adapter that preserves the //! historical `get`/`put`/`delete` surface every consumer uses and forwards each -//! operation to the crate store via [`crate::openhuman::todos::crate_adapter`], -//! converting the crate `TaskBoard` back to the local [`TaskBoard`] and its -//! `TinyAgentsError` to a `String`. +//! operation directly to `tinyagents::graph::todos`. //! //! The agent updates boards through the `todo` tool; the UI can fetch or replace //! them through the `threads.task_board_*` and granular `openhuman.todos_*` RPC -//! surfaces. The single-writer constraint (the core process is the only writer) -//! is documented in the adapter module. +//! surfaces. The core process remains the single writer. -use chrono::Utc; -use serde::{Deserialize, Serialize}; use std::path::{Path, PathBuf}; +use chrono::{TimeZone, Utc}; use tinyagents::graph::todos::store as crate_todos; +pub use tinyagents::graph::todos::{ + normalise_board, TaskApprovalMode, TaskBoard, TaskBoardCard, TaskCardStatus, +}; -use crate::openhuman::todos::crate_adapter; - -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -#[serde(rename_all = "snake_case")] -pub enum TaskCardStatus { - Todo, - /// Plan approval required and pending — the dispatcher parked the card here - /// and emitted `TaskPlanAwaitingApproval`; it will not run until a human - /// approves (→ `Ready`) or rejects (→ `Rejected`). - AwaitingApproval, - /// Approved for execution — the dispatcher runs `Ready` cards without a - /// further approval check (distinguishes "approved" from the initial - /// `Todo`, which the approval gate would otherwise re-park). - Ready, - InProgress, - Blocked, - Done, - /// Plan approval was denied; the card is not executed. - Rejected, -} - -impl TaskCardStatus { - pub fn as_str(&self) -> &'static str { - match self { - Self::Todo => "todo", - Self::AwaitingApproval => "awaiting_approval", - Self::Ready => "ready", - Self::InProgress => "in_progress", - Self::Blocked => "blocked", - Self::Done => "done", - Self::Rejected => "rejected", - } - } -} - -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -#[serde(rename_all = "snake_case")] -pub enum TaskApprovalMode { - Required, - NotRequired, -} - -impl TaskApprovalMode { - pub fn as_str(&self) -> &'static str { - match self { - Self::Required => "required", - Self::NotRequired => "not_required", - } - } -} - -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -#[serde(rename_all = "camelCase")] -pub struct TaskBoardCard { - pub id: String, - pub title: String, - pub status: TaskCardStatus, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub objective: Option, - #[serde(default, skip_serializing_if = "Vec::is_empty")] - pub plan: Vec, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub assigned_agent: Option, - #[serde(default, skip_serializing_if = "Vec::is_empty")] - pub allowed_tools: Vec, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub approval_mode: Option, - #[serde(default, skip_serializing_if = "Vec::is_empty")] - pub acceptance_criteria: Vec, - #[serde(default, skip_serializing_if = "Vec::is_empty")] - pub evidence: Vec, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub notes: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub blocker: Option, - /// Conversation thread id of the card's live/last agent session, when one - /// exists. Set by the autonomous dispatcher (`task_session`) and the manual - /// "Work" path so the UI can offer a "View session" jump into Conversations. - /// `None` for a card that has never been run. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub session_thread_id: Option, - /// Provider/source identifiers for a card ingested from a task source - /// (`{provider, source_id, external_id, url, repo?, urgency}`). Set by - /// the `task_sources` route; consumed downstream for prioritisation and - /// external write-back. `None` for agent/UI-authored cards. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub source_metadata: Option, - #[serde(default)] - pub order: u32, - #[serde(default)] - pub updated_at: String, -} - -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -#[serde(rename_all = "camelCase")] -pub struct TaskBoard { - pub thread_id: String, - pub cards: Vec, - pub updated_at: String, -} - -impl TaskBoard { - pub fn empty(thread_id: impl Into) -> Self { - let now = Utc::now().to_rfc3339(); - Self { - thread_id: thread_id.into(), - cards: Vec::new(), - updated_at: now, - } - } -} +use crate::openhuman::tinyagents::todos::todos_store; #[derive(Debug, Clone)] pub struct TaskBoardStore { @@ -148,10 +37,13 @@ impl TaskBoardStore { pub async fn get(&self, thread_id: &str) -> Result, String> { let thread_id = validate_thread_id(thread_id)?; tracing::debug!(thread_id = %thread_id, "[agent:task_board] get entry"); - let store = crate_adapter::crate_todos_store(&self.workspace_dir); - match crate_adapter::get_crate_board_raw(&store, &thread_id).await? { - Some(crate_board) => { - let board = crate_adapter::from_crate_board(&crate_board); + let store = todos_store(&self.workspace_dir); + match crate_todos::get(&store, &thread_id) + .await + .map_err(|error| format!("decode crate task board for thread {thread_id}: {error}"))? + { + Some(board) => { + let board = normalize_board_for_wire(board); tracing::debug!( thread_id = %thread_id, card_count = board.cards.len(), @@ -166,34 +58,23 @@ impl TaskBoardStore { } } - /// Persist `board`. Locally normalises first (minting `task-` ids for - /// blank cards, recomputing order), then hands the cards to the crate - /// `replace` op — which re-normalises and **enforces the single-`InProgress` + /// Persist `board`. Hands the cards directly to the crate `replace` op, + /// which owns normalisation and **enforces the single-`InProgress` /// invariant** (an invalid board now errors here rather than being silently /// saved). Returns the saved board with crate-normalised cards. - pub async fn put(&self, mut board: TaskBoard) -> Result { + pub async fn put(&self, board: TaskBoard) -> Result { tracing::debug!( thread_id = %board.thread_id, card_count = board.cards.len(), "[agent:task_board] put entry" ); - normalise_board(&mut board); let thread_id = validate_thread_id(&board.thread_id)?; - board.thread_id = thread_id.clone(); - let store = crate_adapter::crate_todos_store(&self.workspace_dir); - let crate_cards: Vec<_> = board - .cards - .iter() - .map(crate_adapter::to_crate_card) - .collect(); - let snap = crate_todos::replace(&store, &thread_id, crate_cards) + let store = todos_store(&self.workspace_dir); + let snap = crate_todos::replace(&store, &thread_id, board.cards) .await .map_err(|e| e.to_string())?; - let cards: Vec = snap - .cards - .iter() - .map(crate_adapter::from_crate_card) - .collect(); + let mut cards = snap.cards; + normalize_cards_for_wire(&mut cards); tracing::debug!( thread_id = %thread_id, card_count = cards.len(), @@ -212,13 +93,10 @@ impl TaskBoardStore { pub async fn delete(&self, thread_id: &str) -> Result { let thread_id = validate_thread_id(thread_id)?; tracing::debug!(thread_id = %thread_id, "[agent:task_board] delete entry"); - let store = crate_adapter::crate_todos_store(&self.workspace_dir); - let existed = crate_adapter::get_crate_board_raw(&store, &thread_id) - .await? - .is_some(); - if existed { - crate_adapter::delete_crate_board_raw(&store, &thread_id).await?; - } + let store = todos_store(&self.workspace_dir); + let existed = crate_todos::delete(&store, &thread_id) + .await + .map_err(|e| e.to_string())?; tracing::debug!(thread_id = %thread_id, existed, "[agent:task_board] delete ok"); Ok(existed) } @@ -230,91 +108,32 @@ pub async fn board_for_thread(workspace_dir: &Path, thread_id: &str) -> Result String { + if chrono::DateTime::parse_from_rfc3339(value).is_ok() { + return value.to_owned(); + } + if let Ok(updated_at_ms) = value.parse::() { + if let Some(updated_at) = Utc.timestamp_millis_opt(updated_at_ms).single() { + return updated_at.to_rfc3339(); } } + tracing::warn!(updated_at = %value, "invalid task-board timestamp; using current time"); + Utc::now().to_rfc3339() +} - board.cards.retain(|card| !card.title.is_empty()); - let removed = before_count.saturating_sub(board.cards.len()); - if removed > 0 { - tracing::debug!( - thread_id = %board.thread_id, - removed, - "[agent:task_board] normalise removed_empty_title_cards" - ); +pub(crate) fn normalize_cards_for_wire(cards: &mut [TaskBoardCard]) { + for card in cards { + card.updated_at = normalize_timestamp_for_wire(&card.updated_at); } +} - for (idx, card) in board.cards.iter_mut().enumerate() { - card.order = idx as u32; - card.updated_at = now.clone(); - } - - tracing::trace!( - thread_id = %board.thread_id, - card_count = board.cards.len(), - "[agent:task_board] normalise exit" - ); +fn normalize_board_for_wire(mut board: TaskBoard) -> TaskBoard { + board.updated_at = normalize_timestamp_for_wire(&board.updated_at); + normalize_cards_for_wire(&mut board.cards); + board } fn validate_thread_id(thread_id: &str) -> Result { @@ -325,13 +144,6 @@ fn validate_thread_id(thread_id: &str) -> Result { Ok(trimmed.to_string()) } -fn trim_string_vec(values: &mut Vec) { - values.retain_mut(|value| { - *value = value.trim().to_string(); - !value.is_empty() - }); -} - #[cfg(test)] mod tests { use super::*; @@ -421,10 +233,25 @@ mod tests { assert_eq!(saved.cards[0].order, 0); assert!(saved.cards[0].id.starts_with("task-")); assert_eq!(saved.cards[1].blocker.as_deref(), Some("waiting on user")); + assert!( + chrono::DateTime::parse_from_rfc3339(&saved.updated_at).is_ok(), + "board updated_at must preserve its RFC 3339 wire format" + ); let loaded = store.get("thread-1").await.expect("get").expect("present"); assert_eq!(loaded.cards.len(), 2); assert_eq!(loaded.cards[1].status, TaskCardStatus::Blocked); + assert!( + chrono::DateTime::parse_from_rfc3339(&loaded.updated_at).is_ok(), + "persisted board reads must preserve the RFC 3339 wire format" + ); + assert!( + loaded + .cards + .iter() + .all(|card| { chrono::DateTime::parse_from_rfc3339(&card.updated_at).is_ok() }), + "persisted card reads must preserve the RFC 3339 wire format" + ); assert!(store.delete("thread-1").await.expect("delete existing")); assert!(store.get("thread-1").await.expect("get deleted").is_none()); @@ -438,6 +265,15 @@ mod tests { assert!(store.get("missing").await.expect("get").is_none()); } + #[tokio::test] + async fn missing_board_fallback_uses_rfc3339_timestamp() { + let dir = tempdir().expect("tempdir"); + let board = board_for_thread(dir.path(), "missing") + .await + .expect("board"); + assert!(chrono::DateTime::parse_from_rfc3339(&board.updated_at).is_ok()); + } + #[tokio::test] async fn blank_thread_id_is_rejected() { let dir = tempdir().expect("tempdir"); @@ -518,8 +354,12 @@ mod tests { use tinyagents::graph::todos::store::TODOS_NAMESPACE; let dir = tempdir().expect("tempdir"); - let store = crate_adapter::crate_todos_store(dir.path()); - let key = crate_adapter::todo_key("thread-corrupt"); + let store = todos_store(dir.path()); + let key = "thread-corrupt" + .as_bytes() + .iter() + .map(|byte| format!("{byte:02x}")) + .collect::(); let undecodable = serde_json::json!({ "not": "a board" }); store .put(TODOS_NAMESPACE, &key, undecodable.clone()) diff --git a/src/openhuman/agent/tools/todo.rs b/src/openhuman/agent/tools/todo.rs index 81fbc988a..e42c7bc56 100644 --- a/src/openhuman/agent/tools/todo.rs +++ b/src/openhuman/agent/tools/todo.rs @@ -307,7 +307,6 @@ fn optional_string_array( #[cfg(test)] mod tests { use super::*; - use crate::openhuman::todos::global_scratch_store; use serde_json::Value; /// Serialize tests that share the process-global scratch store with @@ -317,14 +316,16 @@ mod tests { crate::openhuman::todos::ops::scratch_test_lock() } - fn reset_scratch() { - global_scratch_store().replace(Vec::new()); + async fn reset_scratch() { + crate::openhuman::todos::ops::clear(&BoardLocation::Scratch) + .await + .expect("clear scratch"); } #[tokio::test] async fn add_then_list_round_trips_via_scratch() { let _guard = scratch_lock(); - reset_scratch(); + reset_scratch().await; let tool = TodoTool::new(); let added = tool .execute(json!({ "op": "add", "content": "Write tests" })) @@ -353,7 +354,7 @@ mod tests { .as_str() .unwrap() .contains("[x] Write tests")); - reset_scratch(); + reset_scratch().await; } #[tokio::test] @@ -394,7 +395,7 @@ mod tests { #[tokio::test] async fn edit_rejects_unknown_id() { let _guard = scratch_lock(); - reset_scratch(); + reset_scratch().await; let tool = TodoTool::new(); let result = tool .execute(json!({ "op": "edit", "id": "task-missing", "content": "x" })) @@ -402,13 +403,13 @@ mod tests { .unwrap(); assert!(result.is_error); assert!(result.output().contains("not found")); - reset_scratch(); + reset_scratch().await; } #[tokio::test] async fn replace_accepts_full_card_list() { let _guard = scratch_lock(); - reset_scratch(); + reset_scratch().await; let tool = TodoTool::new(); let result = tool .execute(json!({ @@ -435,6 +436,6 @@ mod tests { assert!(!result.is_error, "{}", result.output()); let payload: Value = serde_json::from_str(&result.output()).unwrap(); assert_eq!(payload["cards"].as_array().unwrap().len(), 2); - reset_scratch(); + reset_scratch().await; } } diff --git a/src/openhuman/tinyagents/mod.rs b/src/openhuman/tinyagents/mod.rs index 363aea279..bc6824b09 100644 --- a/src/openhuman/tinyagents/mod.rs +++ b/src/openhuman/tinyagents/mod.rs @@ -40,6 +40,7 @@ pub(crate) mod stop_hooks; pub(crate) mod subagent_graph; mod summarize; pub mod thread_context; +pub mod todos; pub(crate) mod tools; mod topology; diff --git a/src/openhuman/tinyagents/todos.rs b/src/openhuman/tinyagents/todos.rs new file mode 100644 index 000000000..7d0fe535d --- /dev/null +++ b/src/openhuman/tinyagents/todos.rs @@ -0,0 +1,148 @@ +//! OpenHuman integration for the TinyAgents task-board implementation. + +use std::path::Path; +use std::sync::{Arc, OnceLock}; + +use tinyagents::graph::todos::{store as todos, TaskBoard}; +use tinyagents::harness::store::{InMemoryStore, Store}; + +use crate::openhuman::session_import::ops::open_session_stores; + +/// Open the durable TinyAgents store used by per-thread task boards. +pub fn todos_store(workspace_dir: &Path) -> Arc { + Arc::new(open_session_stores(workspace_dir).kv) +} + +/// Shared ephemeral TinyAgents store used when a tool has no thread context. +pub fn scratch_todos_store() -> Arc { + static STORE: OnceLock> = OnceLock::new(); + STORE.get_or_init(|| Arc::new(InMemoryStore::new())).clone() +} + +/// Synthetic thread key for the process-global scratch board. +pub const SCRATCH_THREAD_ID: &str = "_scratch_"; + +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct TaskBoardMigrationReport { + pub total: usize, + pub copied: usize, + pub skipped: usize, +} + +async fn read_legacy_boards(workspace_dir: &Path) -> Result, String> { + let dir = workspace_dir.join("agent_task_boards"); + let mut entries = match tokio::fs::read_dir(&dir).await { + Ok(entries) => entries, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()), + Err(error) => { + return Err(format!( + "read legacy task boards dir {}: {error}", + dir.display() + )) + } + }; + let mut boards = Vec::new(); + while let Some(entry) = entries + .next_entry() + .await + .map_err(|error| format!("iterate legacy task boards dir: {error}"))? + { + let path = entry.path(); + let is_board = path.extension().and_then(|value| value.to_str()) == Some("json") + && !path + .file_name() + .and_then(|value| value.to_str()) + .is_some_and(|name| name.ends_with(".runs.json")); + if !is_board { + continue; + } + match tokio::fs::read_to_string(&path).await { + Ok(body) => match serde_json::from_str(&body) { + Ok(board) => boards.push(board), + Err(error) => { + tracing::debug!(path = %path.display(), %error, "skip invalid legacy task board") + } + }, + Err(error) => { + tracing::debug!(path = %path.display(), %error, "skip unreadable legacy task board") + } + } + } + Ok(boards) +} + +/// Copy boards from the retired file store without replacing TinyAgents data. +pub async fn migrate_legacy_task_boards( + workspace_dir: &Path, +) -> Result { + let legacy = read_legacy_boards(workspace_dir).await?; + let store = todos_store(workspace_dir); + let mut report = TaskBoardMigrationReport { + total: legacy.len(), + ..Default::default() + }; + for board in legacy { + if todos::import_if_absent(&store, board) + .await + .map_err(|error| error.to_string())? + { + report.copied += 1; + } else { + report.skipped += 1; + } + } + Ok(report) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn legacy_migration_copies_once_without_replacing_tinyagents_data() { + let workspace = tempfile::tempdir().expect("workspace"); + let legacy_dir = workspace.path().join("agent_task_boards"); + tokio::fs::create_dir_all(&legacy_dir) + .await + .expect("legacy dir"); + let mut legacy = TaskBoard::empty("thread-1"); + legacy + .cards + .push(tinyagents::graph::todos::TaskBoardCard::new("legacy")); + tokio::fs::write( + legacy_dir.join("thread-1.json"), + serde_json::to_vec(&legacy).expect("encode legacy"), + ) + .await + .expect("write legacy"); + + let first = migrate_legacy_task_boards(workspace.path()) + .await + .expect("first migration"); + assert_eq!( + first, + TaskBoardMigrationReport { + total: 1, + copied: 1, + skipped: 0, + } + ); + + let store = todos_store(workspace.path()); + todos::clear(&store, "thread-1").await.expect("edit board"); + let second = migrate_legacy_task_boards(workspace.path()) + .await + .expect("second migration"); + assert_eq!(second.copied, 0); + assert_eq!(second.skipped, 1); + assert!( + todos::get(&store, "thread-1") + .await + .expect("get board") + .expect("present") + .cards + .is_empty(), + "existing TinyAgents board must remain authoritative" + ); + } +} diff --git a/src/openhuman/todos/README.md b/src/openhuman/todos/README.md index 79a3d17a8..4f53eda92 100644 --- a/src/openhuman/todos/README.md +++ b/src/openhuman/todos/README.md @@ -1,87 +1,23 @@ # todos -Per-thread todo list (a.k.a. the **agent task board**): CRUD operations plus a markdown renderer over a thread's task cards. Backed by `crate::openhuman::agent::task_board` for persistence, so agent-side edits via the `todo` tool and user-side edits via the `openhuman.todos_*` RPCs share one source of truth. Each operation loads the current cards for a thread (or a process-global scratch store when there's no thread context), applies a mutation, persists, and returns a `TodosSnapshot` containing both the updated cards and a markdown rendering for the chat UI / agent transcript. +Compatibility surface for OpenHuman task-board callers. -## Responsibilities +The task-board model and behavior are owned by +`tinyagents::graph::todos`: types, normalization, markdown rendering, CRUD, +plan decisions, session links, the single-`in_progress` invariant, atomic +claims, durable storage, and the in-memory scratch board. -- CRUD over a thread's task cards: `list`, `add`, `edit`, `update_status`, `decide_plan`, `remove`, `replace`, `clear`. -- Resolve the working set location: a per-thread `TaskBoardStore` file, or the in-memory process-global scratch store as a fallback when no thread id is available. -- Render a card list as GitHub-flavored markdown (`- [ ]` / `- [x]` / `[~]` in-progress / `[!]` blocked / `[?]` awaiting-approval / `[-]` rejected, plus indented objective/agent/tools/plan/acceptance-criteria/evidence/notes/blocker lines). -- Parse stable string status aliases (`pending`→`Todo`, `approved`→`Ready`, etc.) and approval modes (`required` / `not_required`). -- Enforce the invariant that at most one card is `in_progress` at a time. -- Plan-approval transitions via `decide_plan` (approve → `Ready`/runnable, reject → `Rejected`), guarded so only `AwaitingApproval` cards can be decided. -- Emit `AgentProgress::TaskBoardUpdated` to the forked agent's progress channel after a thread-scoped mutation. +OpenHuman keeps this module to preserve app-specific integration: -## Key files +- `ops.rs` maps `BoardLocation` onto TinyAgents stores, preserves the optional + `threadId` snapshot shape used by scratch callers, and emits + `AgentProgress::TaskBoardUpdated`. +- `schemas.rs` preserves the `openhuman.todos_*` JSON-RPC API. +- `tools.rs` preserves the granular `todo_*` agent tools. +- `runs.rs` owns the OpenHuman autonomous-run ledger, which is separate from + task-board storage. -| File | Role | -| --- | --- | -| `src/openhuman/todos/mod.rs` | Export-focused: declares `ops`/`schemas`/`store`, re-exports the controller-schema pair and the scratch store. | -| `src/openhuman/todos/ops.rs` | Core CRUD business logic. Defines `TodosSnapshot`, `CardPatch`, `BoardLocation`; load/save/markdown helpers; `add`/`edit`/`update_status`/`decide_plan`/`remove`/`replace`/`clear`/`list`; `parse_status`; single-in-progress enforcement; progress emission. Inline test suite. | -| `src/openhuman/todos/schemas.rs` | JSON-RPC controller surface: `all_controller_schemas` / `all_registered_controllers`, per-function `ControllerSchema`s, `handle_*` async handlers, param structs, approval-mode parsing, and `thread_location` resolution. | -| `src/openhuman/todos/store.rs` | `ScratchTodoStore` + `global_scratch_store()` — the process-global in-memory fallback list. | - -## Public surface - -From `mod.rs` re-exports: - -- `all_todos_controller_schemas` / `all_todos_registered_controllers` (aliases of `schemas::all_controller_schemas` / `all_registered_controllers`). -- `ScratchTodoStore`, `global_scratch_store` (from `store`). - -Heavily used by callers directly via `todos::ops`: `BoardLocation` (`Thread { workspace_dir, thread_id }` | `Scratch`), `CardPatch`, `TodosSnapshot`, `parse_status`, and the CRUD fns `add`/`edit`/`update_status`/`decide_plan`/`remove`/`replace`/`clear`/`list`/`render_markdown`. - -## RPC / controllers - -Namespace `todos` (wire methods `openhuman.todos_*`), wired into the registry via `src/core/all.rs`: - -| Function | Description | -| --- | --- | -| `list` | Return the thread's cards + markdown. | -| `add` | Append a new card (`content` required; optional status/objective/plan/assignedAgent/allowedTools/approvalMode/acceptanceCriteria/evidence/notes/blocker). | -| `edit` | Edit a card by `id`; omitted fields untouched; `approvalMode: null` clears the mode. | -| `update_status` | Change only the status. | -| `decide_plan` | Approve/reject a card `awaiting_approval` (approve → `ready`, reject → `rejected`). | -| `remove` | Delete a card by `id`. | -| `replace` | Wholesale-replace the list (`cards` JSON array; empty ids are server-generated). | -| `clear` | Empty the list. | - -All take a required `thread_id` (same id as `threads.task_board_*`) and return a `snapshot` JSON object (`threadId`, `cards`, `markdown`). `thread_location` loads `Config::load_or_init()` to resolve `workspace_dir`. - -## Agent tools - -This module owns no `tools.rs`. The agent-facing `todo` tool lives in `src/openhuman/agent/tools/todo.rs` and delegates into `todos::ops`, sharing the same persistence/rendering. It uses `BoardLocation::Scratch` when no thread context is present, else `BoardLocation::Thread`. - -## Events - -No `bus.rs`. Instead, thread-scoped mutations emit `AgentProgress::TaskBoardUpdated { board }` on the forked agent's progress channel (`agent::harness::fork_context::current_parent().on_progress`). This is an in-process agent-progress channel send, not a `DomainEvent` publish. - -## Persistence - -- **Per-thread (authoritative):** `TaskBoardStore` writes to `/agent_task_boards/.json` (atomic file rename via `TaskBoardStore::put`; cards normalised with `normalise_board`). -- **Scratch (fallback):** `ScratchTodoStore` — a process-global `Arc>>` (`global_scratch_store()`), in-memory only, used when no thread id is available. The same `Arc` is returned across calls so tool re-registration keeps state. - -## Dependencies - -- `crate::openhuman::agent::task_board` — `TaskBoard`, `TaskBoardCard`, `TaskBoardStore`, `TaskCardStatus`, `TaskApprovalMode`, `normalise_board`. The actual card model and per-thread persistence. -- `crate::openhuman::agent::progress::AgentProgress` — variant emitted on board changes. -- `crate::openhuman::agent::harness::fork_context` — `current_parent()` to reach the parent agent's progress sender. -- `crate::openhuman::config::Config` — `load_or_init()` resolves the `workspace_dir` for thread-scoped boards (in `schemas.rs`). -- `crate::core::all` — `ControllerFuture`, `RegisteredController` for the RPC registry. -- `crate::core::{ControllerSchema, FieldSchema, TypeSchema}` — controller schema types. - -## Used by - -- `src/core/all.rs` — registers the controllers/schemas into the RPC registry. -- `src/openhuman/agent/tools/todo.rs` — the agent `todo` tool delegates to `todos::ops`. -- `src/openhuman/agent/task_dispatcher.rs` — dispatcher reads/transitions cards. -- `src/openhuman/agent/triage/{escalation.rs, envelope.rs}` — triage flows. -- `src/openhuman/task_sources/route.rs` — task-source ingestion (uses `CardPatch.source_metadata`). - -## Notes / gotchas - -- **Scratch CRUD serialization:** `ScratchTodoStore::snapshot`/`replace` each take the inner lock independently; `ops.rs` wraps scratch load→mutate→save in a coarser process-global `scratch_serial_lock` so the pair runs in one critical section. Thread ops rely on `TaskBoardStore::put`'s file-rename atomicity instead. -- **Single in-progress invariant** is enforced on `add`/`edit`/`replace` (`enforce_single_in_progress`); violating it returns an error rather than silently fixing it. -- **`approval_mode` is doubly-optional** (`Option>`): `None` = leave untouched, `Some(None)` = clear, `Some(Some(_))` = set. The `edit` RPC distinguishes "absent" from explicit `null` via `approval_mode_patch_from_params` reading the raw params map before deserialization. -- **`source_metadata`** is only settable via `ops` directly (e.g. task-source route); the `add`/`edit` RPC handlers always pass `source_metadata: None`, and an edit with `None` preserves existing metadata. -- `decide_plan` errors on a non-`AwaitingApproval` card so a stale/duplicate decision can't resurrect a card that already moved on. -- `#[cfg(test)] scratch_test_lock()` is `pub(crate)` so the agent `todo` tool tests can serialize shared-scratch access under cargo's parallel runner. +`agent::task_board` re-exports the TinyAgents board types and keeps the legacy +`TaskBoardStore` facade for existing callers. Legacy +`agent_task_boards/*.json` values are imported at startup through +`openhuman::tinyagents::todos`; existing TinyAgents values are never replaced. diff --git a/src/openhuman/todos/crate_adapter.rs b/src/openhuman/todos/crate_adapter.rs deleted file mode 100644 index ed6634761..000000000 --- a/src/openhuman/todos/crate_adapter.rs +++ /dev/null @@ -1,555 +0,0 @@ -//! Adapter seam onto the vendored `tinyagents::graph::todos` crate store. -//! -//! The crate store is now **authoritative** for per-thread task boards — the -//! OpenHuman [`TaskBoardStore`](crate::openhuman::agent::task_board::TaskBoardStore) -//! and [`todos::ops`](crate::openhuman::todos::ops) `Thread` path delegate every -//! read/write to it. This module supplies the conversion helpers (OpenHuman ↔ -//! crate `TaskBoardCard`/`TaskBoard`/`TaskCardStatus`/`TaskApprovalMode`), the -//! store-handle opener ([`crate_todos_store`]), the raw crate-board read/write -//! helpers used for the existence-preserving `get`/`delete` semantics, and the -//! idempotent [`migrate_legacy_task_boards_into_crate_store`] boot helper that -//! copies boards left in the retired `/agent_task_boards/` -//! file-JSON tree only when the crate store has no value for that thread. -//! -//! Persistence target: the crate [`Store`] rooted at the shared workspace KV -//! tree (`/tinyagents_store/kv`), namespace [`TODOS_NAMESPACE`] -//! (`graph.todos`), keyed by `hex(thread_id)` — byte-for-byte the key the -//! crate's own `graph::todos::store` computes. -//! -//! # Scratch has no crate target -//! OpenHuman keeps an in-memory, thread-less -//! [`BoardLocation::Scratch`](crate::openhuman::todos::ops::BoardLocation) -//! fallback (tool calls outside a chat thread). The crate board is always -//! `(Store, thread_id)`, so scratch stays entirely on the local -//! [`ScratchTodoStore`](crate::openhuman::todos::store::ScratchTodoStore) path; -//! only `BoardLocation::Thread` routes here. -//! -//! # Single-writer constraint -//! The crate `Store` has no cross-process compare-and-set; its per-thread -//! atomicity is a process-local async mutex. This is acceptable because the -//! OpenHuman core is the single writer of task boards. - -use std::path::Path; -use std::sync::Arc; - -use tinyagents::graph::todos::store::TODOS_NAMESPACE; -use tinyagents::graph::todos::{ - TaskApprovalMode as CrateApprovalMode, TaskBoard as CrateBoard, TaskBoardCard as CrateCard, - TaskCardStatus as CrateStatus, -}; -use tinyagents::harness::store::Store; - -use crate::openhuman::agent::task_board::{ - TaskApprovalMode as OhApprovalMode, TaskBoard as OhBoard, TaskBoardCard as OhCard, - TaskCardStatus as OhStatus, -}; -use crate::openhuman::session_import::ops::open_session_stores; - -/// Open the crate [`Store`] handle backing the `graph.todos` namespace, rooted -/// at the shared workspace KV tree (`/tinyagents_store/kv`). Same -/// layout the 04-sessions journal + thread goals use, so everything lives under -/// one tree. -pub(crate) fn crate_todos_store(workspace_dir: &Path) -> Arc { - Arc::new(open_session_stores(workspace_dir).kv) -} - -/// The crate store key for a thread's board: lowercase hex of the (trimmed) -/// thread-id bytes. MUST match the crate's private `graph::todos::store` key -/// function exactly so the crate reader resolves our value. -pub(crate) fn todo_key(thread_id: &str) -> String { - thread_id - .trim() - .as_bytes() - .iter() - .map(|b| format!("{b:02x}")) - .collect() -} - -/// Maps an OpenHuman [`OhStatus`] to the crate [`CrateStatus`]. Total (the two -/// enums share the same seven variants). -pub(crate) fn map_status_to_crate(status: &OhStatus) -> CrateStatus { - match status { - OhStatus::Todo => CrateStatus::Todo, - OhStatus::AwaitingApproval => CrateStatus::AwaitingApproval, - OhStatus::Ready => CrateStatus::Ready, - OhStatus::InProgress => CrateStatus::InProgress, - OhStatus::Blocked => CrateStatus::Blocked, - OhStatus::Done => CrateStatus::Done, - OhStatus::Rejected => CrateStatus::Rejected, - } -} - -/// Maps a crate [`CrateStatus`] back to an OpenHuman [`OhStatus`]. Total; the -/// inverse of [`map_status_to_crate`]. -pub(crate) fn map_status_from_crate(status: CrateStatus) -> OhStatus { - match status { - CrateStatus::Todo => OhStatus::Todo, - CrateStatus::AwaitingApproval => OhStatus::AwaitingApproval, - CrateStatus::Ready => OhStatus::Ready, - CrateStatus::InProgress => OhStatus::InProgress, - CrateStatus::Blocked => OhStatus::Blocked, - CrateStatus::Done => OhStatus::Done, - CrateStatus::Rejected => OhStatus::Rejected, - } -} - -fn map_approval_mode(mode: &OhApprovalMode) -> CrateApprovalMode { - match mode { - OhApprovalMode::Required => CrateApprovalMode::Required, - OhApprovalMode::NotRequired => CrateApprovalMode::NotRequired, - } -} - -fn map_approval_mode_from_crate(mode: CrateApprovalMode) -> OhApprovalMode { - match mode { - CrateApprovalMode::Required => OhApprovalMode::Required, - CrateApprovalMode::NotRequired => OhApprovalMode::NotRequired, - } -} - -/// Converts an OpenHuman [`OhCard`] into the crate [`CrateCard`], preserving the -/// id, status, and all optional metadata so a persisted board round-trips. -pub(crate) fn to_crate_card(card: &OhCard) -> CrateCard { - CrateCard { - id: card.id.clone(), - title: card.title.clone(), - status: map_status_to_crate(&card.status), - objective: card.objective.clone(), - plan: card.plan.clone(), - assigned_agent: card.assigned_agent.clone(), - allowed_tools: card.allowed_tools.clone(), - approval_mode: card.approval_mode.as_ref().map(map_approval_mode), - acceptance_criteria: card.acceptance_criteria.clone(), - evidence: card.evidence.clone(), - notes: card.notes.clone(), - blocker: card.blocker.clone(), - session_thread_id: card.session_thread_id.clone(), - source_metadata: card.source_metadata.clone(), - order: card.order, - updated_at: card.updated_at.clone(), - } -} - -/// Converts a crate [`CrateCard`] back into an OpenHuman [`OhCard`] (the inverse -/// of [`to_crate_card`]), used by the read path to return local cards from the -/// crate store. -pub(crate) fn from_crate_card(card: &CrateCard) -> OhCard { - OhCard { - id: card.id.clone(), - title: card.title.clone(), - status: map_status_from_crate(card.status), - objective: card.objective.clone(), - plan: card.plan.clone(), - assigned_agent: card.assigned_agent.clone(), - allowed_tools: card.allowed_tools.clone(), - approval_mode: card.approval_mode.map(map_approval_mode_from_crate), - acceptance_criteria: card.acceptance_criteria.clone(), - evidence: card.evidence.clone(), - notes: card.notes.clone(), - blocker: card.blocker.clone(), - session_thread_id: card.session_thread_id.clone(), - source_metadata: card.source_metadata.clone(), - order: card.order, - updated_at: card.updated_at.clone(), - } -} - -/// Convert a whole OpenHuman [`OhBoard`] into a crate [`CrateBoard`] (faithful -/// projection — no re-mint, no re-normalise). -pub(crate) fn to_crate_board(board: &OhBoard) -> CrateBoard { - CrateBoard { - thread_id: board.thread_id.clone(), - cards: board.cards.iter().map(to_crate_card).collect(), - updated_at: board.updated_at.clone(), - } -} - -/// Convert a crate [`CrateBoard`] back into an OpenHuman [`OhBoard`]. -pub(crate) fn from_crate_board(board: &CrateBoard) -> OhBoard { - OhBoard { - thread_id: board.thread_id.clone(), - cards: board.cards.iter().map(from_crate_card).collect(), - updated_at: board.updated_at.clone(), - } -} - -/// Read the raw crate [`CrateBoard`] stored for `thread_id` (ns `graph.todos`, -/// key `hex(thread_id)`), or `None` when the thread has no board value. Does -/// **not** normalise. An undecodable value returns an error so live -/// read/modify/write callers fail closed instead of replacing a present row. -pub(crate) async fn get_crate_board_raw( - store: &Arc, - thread_id: &str, -) -> Result, String> { - let value = store - .get(TODOS_NAMESPACE, &todo_key(thread_id)) - .await - .map_err(|e| format!("read crate task board: {e}"))?; - match value { - Some(v) => serde_json::from_value::(v) - .map(Some) - .map_err(|e| format!("decode crate task board for thread {thread_id}: {e}")), - None => Ok(None), - } -} - -/// Whether the crate store contains any value for `thread_id`, without -/// attempting to decode it. Migration uses this stricter existence check so a -/// corrupt or forward-schema value remains authoritative and is never -/// overwritten by a stale legacy file. -async fn crate_board_value_exists(store: &Arc, thread_id: &str) -> Result { - store - .get(TODOS_NAMESPACE, &todo_key(thread_id)) - .await - .map(|value| value.is_some()) - .map_err(|e| format!("check crate task board existence: {e}")) -} - -/// Delete the crate board value for `thread_id` (ns `graph.todos`). No-op when -/// absent. Unlike the crate `clear` op (which writes an empty board), this -/// removes the key so a subsequent read reports `None` — matching the legacy -/// file-delete semantics. -pub(crate) async fn delete_crate_board_raw( - store: &Arc, - thread_id: &str, -) -> Result<(), String> { - store - .delete(TODOS_NAMESPACE, &todo_key(thread_id)) - .await - .map_err(|e| format!("delete crate task board: {e}")) -} - -/// Write a faithful raw crate mirror of `board` (ns `graph.todos`). Used only by -/// the boot migration; the live path goes through the crate `replace` op. -async fn put_crate_board_raw(store: &Arc, board: &CrateBoard) -> Result<(), String> { - let value = serde_json::to_value(board).map_err(|e| format!("serialize crate board: {e}"))?; - store - .put(TODOS_NAMESPACE, &todo_key(&board.thread_id), value) - .await - .map_err(|e| format!("mirror task board into {TODOS_NAMESPACE}: {e}")) -} - -// ── Idempotent legacy→crate migration helper (run on each core boot) ────────── - -/// Outcome of a [`migrate_legacy_task_boards_into_crate_store`] run. -#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] -pub struct TaskBoardMigrationReport { - /// Legacy board files examined. - pub total: usize, - /// Boards written into the crate store because no crate value existed. - pub copied: usize, - /// Boards already present in the crate store and left authoritative. - pub skipped: usize, -} - -/// Read any boards left in the retired legacy `/agent_task_boards/` -/// file-JSON tree. Returns an empty vec when the directory is absent (the common -/// case after the first migration). The per-thread run ledger (`*.runs.json`) -/// stays local and is skipped. Undecodable/unreadable files are skipped so a -/// stray file can't wedge the idempotent copy. -async fn read_legacy_file_boards(workspace_dir: &Path) -> Result, String> { - const LEGACY_DIR: &str = "agent_task_boards"; - const LEGACY_EXT: &str = "json"; - const RUNS_SUFFIX: &str = ".runs.json"; - let dir = workspace_dir.join(LEGACY_DIR); - let mut entries = match tokio::fs::read_dir(&dir).await { - Ok(rd) => rd, - Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()), - Err(e) => { - return Err(format!( - "read legacy task boards dir {}: {e}", - dir.display() - )) - } - }; - let mut boards = Vec::new(); - while let Some(entry) = entries - .next_entry() - .await - .map_err(|e| format!("iterate legacy task boards dir: {e}"))? - { - let path = entry.path(); - if path.extension().and_then(|e| e.to_str()) != Some(LEGACY_EXT) { - continue; - } - // Skip the run ledger sidecar files, which stay on the local fs path. - if path - .file_name() - .and_then(|n| n.to_str()) - .map(|n| n.ends_with(RUNS_SUFFIX)) - .unwrap_or(false) - { - continue; - } - match tokio::fs::read_to_string(&path).await { - Ok(body) => match serde_json::from_str::(&body) { - Ok(board) => boards.push(board), - Err(e) => { - tracing::debug!(path = %path.display(), error = %e, "[todos][crate-migrate] skip parse error"); - } - }, - Err(e) => { - tracing::debug!(path = %path.display(), error = %e, "[todos][crate-migrate] skip read error"); - } - } - } - Ok(boards) -} - -/// Copy legacy task boards missing from the crate `graph.todos` store. -/// -/// **Idempotent and non-destructive**: any existing crate value is -/// authoritative and skipped, even when it differs from the stale legacy -/// file. This prevents a later process restart from reverting post-migration -/// edits while the retired files remain on disk. -/// -/// Wired into `core::runtime::services` on every core boot. Honors the -/// single-writer constraint: run it only inside the core process. -pub async fn migrate_legacy_task_boards_into_crate_store( - workspace_dir: &Path, -) -> Result { - let legacy = read_legacy_file_boards(workspace_dir).await?; - let store = crate_todos_store(workspace_dir); - let mut report = TaskBoardMigrationReport { - total: legacy.len(), - ..Default::default() - }; - tracing::info!( - workspace = %workspace_dir.display(), - total = report.total, - "[todos][crate-migrate] start copy legacy task boards → graph.todos" - ); - for board in &legacy { - let crate_board = to_crate_board(board); - match crate_board_value_exists(&store, &board.thread_id).await { - Ok(true) => { - report.skipped += 1; - tracing::debug!( - thread_id = %board.thread_id, - "[todos][crate-migrate] skip (crate board already authoritative)" - ); - continue; - } - Ok(false) => {} - Err(e) => { - return Err(format!( - "check crate task board before migrating thread {}: {e}", - board.thread_id - )) - } - } - put_crate_board_raw(&store, &crate_board).await?; - report.copied += 1; - tracing::debug!( - thread_id = %board.thread_id, - card_count = crate_board.cards.len(), - "[todos][crate-migrate] copied" - ); - } - tracing::info!( - total = report.total, - copied = report.copied, - skipped = report.skipped, - "[todos][crate-migrate] done" - ); - Ok(report) -} - -#[cfg(test)] -mod tests { - use super::*; - - fn oh_card(id: &str, status: OhStatus) -> OhCard { - OhCard { - id: id.to_string(), - title: format!("card {id}"), - status, - objective: Some("obj".to_string()), - plan: vec!["step-1".to_string()], - assigned_agent: Some("planner".to_string()), - allowed_tools: vec!["todo".to_string()], - approval_mode: Some(OhApprovalMode::Required), - acceptance_criteria: vec!["tests pass".to_string()], - evidence: vec!["cargo test".to_string()], - notes: Some("note".to_string()), - blocker: None, - session_thread_id: Some("thread-x".to_string()), - source_metadata: Some(serde_json::json!({ "urgency": 0.5 })), - order: 3, - updated_at: "2026-07-03T00:00:00Z".to_string(), - } - } - - #[test] - fn status_mapping_is_total_and_round_trips() { - let all = [ - OhStatus::Todo, - OhStatus::AwaitingApproval, - OhStatus::Ready, - OhStatus::InProgress, - OhStatus::Blocked, - OhStatus::Done, - OhStatus::Rejected, - ]; - for oh in all { - let crate_status = map_status_to_crate(&oh); - assert_eq!(oh.as_str(), crate_status.as_str()); - assert_eq!(map_status_from_crate(crate_status), oh); - } - } - - #[test] - fn card_conversion_round_trips_all_fields() { - let oh = oh_card("task-1", OhStatus::InProgress); - let c = to_crate_card(&oh); - assert_eq!(c.id, "task-1"); - assert_eq!(c.status, CrateStatus::InProgress); - assert_eq!(c.approval_mode, Some(CrateApprovalMode::Required)); - assert_eq!(c.updated_at, "2026-07-03T00:00:00Z"); - // Full round-trip identity. - assert_eq!(from_crate_card(&c), oh); - } - - #[test] - fn approval_mode_maps_both_directions() { - assert_eq!( - map_approval_mode(&OhApprovalMode::Required), - CrateApprovalMode::Required - ); - assert_eq!( - map_approval_mode_from_crate(CrateApprovalMode::NotRequired), - OhApprovalMode::NotRequired - ); - } - - #[tokio::test] - async fn raw_get_delete_round_trip_and_crate_reader_agrees() { - let dir = tempfile::tempdir().expect("tempdir"); - let store = crate_todos_store(dir.path()); - let board = OhBoard { - thread_id: "user-tasks".to_string(), - cards: vec![ - oh_card("task-a", OhStatus::Todo), - oh_card("task-b", OhStatus::InProgress), - ], - updated_at: "2026-07-03T00:00:00Z".to_string(), - }; - assert!(get_crate_board_raw(&store, "user-tasks") - .await - .unwrap() - .is_none()); - put_crate_board_raw(&store, &to_crate_board(&board)) - .await - .unwrap(); - let read = get_crate_board_raw(&store, "user-tasks") - .await - .unwrap() - .unwrap(); - assert_eq!(from_crate_board(&read), board, "raw board round-trips"); - - // The crate's own reader resolves the same key/namespace we wrote. - let via_crate = tinyagents::graph::todos::store::list(&store, "user-tasks") - .await - .unwrap(); - assert_eq!(via_crate.cards.len(), 2); - assert_eq!(via_crate.cards[1].status, CrateStatus::InProgress); - - delete_crate_board_raw(&store, "user-tasks").await.unwrap(); - assert!(get_crate_board_raw(&store, "user-tasks") - .await - .unwrap() - .is_none()); - // Delete is idempotent (no-op when absent). - delete_crate_board_raw(&store, "user-tasks").await.unwrap(); - } - - /// Write a board into the retired legacy `/agent_task_boards/` - /// file-JSON tree (the shape `read_legacy_file_boards` reads). - fn write_legacy_board(dir: &std::path::Path, board: &OhBoard) { - let legacy_dir = dir.join("agent_task_boards"); - std::fs::create_dir_all(&legacy_dir).unwrap(); - let path = legacy_dir.join(format!("{}.json", hex::encode(board.thread_id.as_bytes()))); - std::fs::write(&path, serde_json::to_string(board).unwrap()).unwrap(); - } - - fn legacy_board(thread_id: &str, card_id: &str) -> OhBoard { - OhBoard { - thread_id: thread_id.to_string(), - cards: vec![oh_card(card_id, OhStatus::Todo)], - updated_at: "2026-07-03T00:00:00Z".to_string(), - } - } - - #[tokio::test] - async fn migration_copies_then_is_idempotent_and_skips_runs_sidecar() { - let tmp = tempfile::tempdir().unwrap(); - let dir = tmp.path(); - write_legacy_board(dir, &legacy_board("t1", "task-1")); - write_legacy_board(dir, &legacy_board("t2", "task-2")); - // A runs sidecar must be ignored, not parsed as a board. - let legacy_dir = dir.join("agent_task_boards"); - std::fs::write( - legacy_dir.join(format!("{}.runs.json", hex::encode("t1".as_bytes()))), - "[]", - ) - .unwrap(); - - let r1 = migrate_legacy_task_boards_into_crate_store(dir) - .await - .unwrap(); - assert_eq!(r1.total, 2, "runs sidecar not counted as a board"); - assert_eq!(r1.copied, 2); - assert_eq!(r1.skipped, 0); - - let store = crate_todos_store(dir); - let m2 = get_crate_board_raw(&store, "t2").await.unwrap().unwrap(); - assert_eq!(m2.cards[0].id, "task-2"); - - // Simulate a live post-migration edit while the legacy files remain. - let mut edited = m2; - edited.cards[0].title = "newer crate title".to_string(); - edited.updated_at = "2026-07-04T00:00:00Z".to_string(); - put_crate_board_raw(&store, &edited).await.unwrap(); - - // Second run is a no-op (idempotent) and must not restore stale data. - let r2 = migrate_legacy_task_boards_into_crate_store(dir) - .await - .unwrap(); - assert_eq!(r2.total, 2); - assert_eq!(r2.copied, 0, "idempotent: nothing re-copied"); - assert_eq!(r2.skipped, 2); - let preserved = get_crate_board_raw(&store, "t2").await.unwrap().unwrap(); - assert_eq!(preserved.cards[0].title, "newer crate title"); - assert_eq!(preserved.updated_at, "2026-07-04T00:00:00Z"); - } - - #[tokio::test] - async fn migration_preserves_undecodable_but_present_crate_board() { - let tmp = tempfile::tempdir().unwrap(); - let dir = tmp.path(); - write_legacy_board(dir, &legacy_board("t-corrupt", "stale-task")); - - let store = crate_todos_store(dir); - let key = todo_key("t-corrupt"); - let undecodable = serde_json::json!({ - "thread_id": "t-corrupt", - "cards": "future-schema", - }); - store - .put(TODOS_NAMESPACE, &key, undecodable.clone()) - .await - .unwrap(); - - let report = migrate_legacy_task_boards_into_crate_store(dir) - .await - .unwrap(); - assert_eq!(report.total, 1); - assert_eq!(report.copied, 0); - assert_eq!(report.skipped, 1); - assert_eq!( - store.get(TODOS_NAMESPACE, &key).await.unwrap(), - Some(undecodable), - "migration must not overwrite a present row it cannot decode" - ); - } -} diff --git a/src/openhuman/todos/invariant_tests.rs b/src/openhuman/todos/invariant_tests.rs deleted file mode 100644 index 0eab78c0b..000000000 --- a/src/openhuman/todos/invariant_tests.rs +++ /dev/null @@ -1,750 +0,0 @@ -//! Race and invariant tests for the task-board scheduler kernel. -//! -//! These tests exercise concurrent claim races, run-pointer consistency, -//! status-transition validity, double-completion rejection, and randomised -//! operation sequences over small task graphs. Each test is deterministic -//! (seeded PRNG where randomness is needed). -//! -//! The board CRUD is now crate-backed and **async** (see -//! [`crate::openhuman::todos::ops`]), so the concurrency tests use tokio tasks -//! and the per-board async claim lock rather than OS threads. - -use std::collections::HashSet; -use std::sync::Arc; -use std::time::Duration; -use tempfile::tempdir; - -use crate::openhuman::agent::task_board::{TaskBoardCard, TaskCardStatus}; -use crate::openhuman::todos::ops::{ - add, claim_card, list, update_status, BoardLocation, CardPatch, -}; -use crate::openhuman::todos::runs::{ - complete_run, create_run, get_run, list_runs, reclaim_stale, update_heartbeat, RunLimits, - RunOutcome, -}; - -fn thread_loc(dir: &std::path::Path, id: &str) -> BoardLocation { - BoardLocation::Thread { - workspace_dir: dir.to_path_buf(), - thread_id: id.to_string(), - } -} - -async fn add_card(loc: &BoardLocation, title: &str) -> String { - add(loc, title, CardPatch::default()) - .await - .unwrap() - .cards - .last() - .unwrap() - .id - .clone() -} - -/// Race `n` concurrent claimers of `card_id` (Todo/Ready → InProgress) on tokio -/// tasks gated by a shared barrier, returning each claim's result. Exercises the -/// per-board async CAS lock in [`claim_card`]. -async fn race_claims( - loc: &BoardLocation, - card_id: &str, - n: usize, - expected: Vec, -) -> Vec> { - let barrier = Arc::new(tokio::sync::Barrier::new(n)); - let mut handles = Vec::new(); - for _ in 0..n { - let loc = loc.clone(); - let id = card_id.to_string(); - let barrier = barrier.clone(); - let expected = expected.clone(); - handles.push(tokio::spawn(async move { - barrier.wait().await; - claim_card(&loc, &id, &expected, TaskCardStatus::InProgress).await - })); - } - let mut results = Vec::new(); - for h in handles { - results.push(h.await.unwrap()); - } - results -} - -// ── Invariant checkers ───────────────────────────────────────────────── - -async fn assert_board_invariants(loc: &BoardLocation) { - let snap = list(loc).await.unwrap(); - let cards = &snap.cards; - - // At most one InProgress card. - let in_progress: Vec<_> = cards - .iter() - .filter(|c| c.status == TaskCardStatus::InProgress) - .collect(); - assert!( - in_progress.len() <= 1, - "invariant violated: {} cards InProgress (max 1): {:?}", - in_progress.len(), - in_progress.iter().map(|c| &c.id).collect::>() - ); - - // All card IDs are unique. - let mut ids = HashSet::new(); - for card in cards { - assert!( - ids.insert(&card.id), - "invariant violated: duplicate card id '{}'", - card.id - ); - } - - // Order indices are contiguous 0..n. - let mut orders: Vec = cards.iter().map(|c| c.order).collect(); - orders.sort(); - let expected: Vec = (0..cards.len() as u32).collect(); - assert_eq!( - orders, expected, - "invariant violated: order indices not contiguous" - ); -} - -async fn assert_run_pointer_invariants(loc: &BoardLocation) { - let snap = list(loc).await.unwrap(); - let runs = list_runs(loc, None).unwrap_or_default(); - - // Every active run's card_id references a card that exists on the board. - let card_ids: HashSet<&str> = snap.cards.iter().map(|c| c.id.as_str()).collect(); - for run in &runs { - if run.is_active() { - assert!( - card_ids.contains(run.card_id.as_str()), - "invariant violated: active run '{}' references missing card '{}'", - run.run_id, - run.card_id - ); - } - } - - // Every InProgress card has at least one active run. - for card in &snap.cards { - if card.status == TaskCardStatus::InProgress { - let active_runs: Vec<_> = runs - .iter() - .filter(|r| r.card_id == card.id && r.is_active()) - .collect(); - assert!( - !active_runs.is_empty(), - "invariant violated: InProgress card '{}' has no active run", - card.id - ); - } - } - - // No card has more than one active run. - let mut active_by_card: std::collections::HashMap<&str, usize> = - std::collections::HashMap::new(); - for run in &runs { - if run.is_active() { - *active_by_card.entry(&run.card_id).or_default() += 1; - } - } - for (card_id, count) in &active_by_card { - assert!( - *count <= 1, - "invariant violated: card '{}' has {} active runs (max 1)", - card_id, - count - ); - } -} - -/// Wait just long enough for `check_staleness` (which truncates to whole -/// seconds via `num_seconds()`) to see the run as older than 0s. -async fn sleep_for_staleness() { - tokio::time::sleep(Duration::from_millis(1100)).await; -} - -// ── 1. Double-claim invariant (stress) ──────────────────────────────── - -#[tokio::test(flavor = "multi_thread", worker_threads = 4)] -async fn stress_concurrent_claims_n_threads() { - let dir = tempdir().unwrap(); - let loc = thread_loc(dir.path(), "stress-claim-1"); - let card_id = add_card(&loc, "race target").await; - - let n = 8; - let results = race_claims( - &loc, - &card_id, - n, - vec![TaskCardStatus::Todo, TaskCardStatus::Ready], - ) - .await; - - let wins = results.iter().filter(|r| r.is_ok()).count(); - assert_eq!(wins, 1, "exactly one of {n} concurrent claimers must win"); - - let snap = list(&loc).await.unwrap(); - assert_eq!(snap.cards[0].status, TaskCardStatus::InProgress); - assert_board_invariants(&loc).await; -} - -#[tokio::test(flavor = "multi_thread", worker_threads = 4)] -async fn stress_concurrent_claims_multiple_cards() { - let dir = tempdir().unwrap(); - let loc = thread_loc(dir.path(), "stress-claim-multi"); - - let id_a = add_card(&loc, "card A").await; - let id_b = add_card(&loc, "card B").await; - - // 4 tasks race to claim card A. - let results_a = race_claims(&loc, &id_a, 4, vec![TaskCardStatus::Todo]).await; - - assert_eq!( - results_a.iter().filter(|r| r.is_ok()).count(), - 1, - "exactly one claimer wins card A" - ); - - // Card B cannot go InProgress because enforce_single_in_progress blocks it. - let claim_b = claim_card( - &loc, - &id_b, - &[TaskCardStatus::Todo], - TaskCardStatus::InProgress, - ) - .await; - assert!( - claim_b.is_err(), - "claiming card B as InProgress must fail while card A is InProgress" - ); - - assert_board_invariants(&loc).await; -} - -// ── 2. Run pointer invariant ────────────────────────────────────────── - -#[tokio::test] -async fn run_pointer_active_run_references_existing_card() { - let dir = tempdir().unwrap(); - let loc = thread_loc(dir.path(), "run-ptr-1"); - let card_id = add_card(&loc, "task with run").await; - - claim_card( - &loc, - &card_id, - &[TaskCardStatus::Todo], - TaskCardStatus::InProgress, - ) - .await - .unwrap(); - - let run = create_run(&loc, "run-1", &card_id, "dispatcher").unwrap(); - assert!(run.is_active()); - assert_run_pointer_invariants(&loc).await; -} - -#[tokio::test] -async fn run_pointer_completed_run_consistency() { - let dir = tempdir().unwrap(); - let loc = thread_loc(dir.path(), "run-ptr-2"); - let card_id = add_card(&loc, "task to complete").await; - - claim_card( - &loc, - &card_id, - &[TaskCardStatus::Todo], - TaskCardStatus::InProgress, - ) - .await - .unwrap(); - - create_run(&loc, "run-2", &card_id, "dispatcher").unwrap(); - complete_run( - &loc, - "run-2", - RunOutcome::Success, - None, - vec!["evidence-1".into()], - ) - .unwrap(); - - update_status(&loc, &card_id, TaskCardStatus::Done) - .await - .unwrap(); - - let snap = list(&loc).await.unwrap(); - assert_eq!(snap.cards[0].status, TaskCardStatus::Done); - - // No active runs remain. - let runs = list_runs(&loc, Some(&card_id)).unwrap(); - assert!( - runs.iter().all(|r| !r.is_active()), - "all runs should be completed after card is Done" - ); - assert_board_invariants(&loc).await; -} - -#[tokio::test] -async fn run_pointer_no_orphaned_active_runs_after_reclaim() { - let dir = tempdir().unwrap(); - let loc = thread_loc(dir.path(), "run-ptr-3"); - let card_id = add_card(&loc, "stale task").await; - - claim_card( - &loc, - &card_id, - &[TaskCardStatus::Todo], - TaskCardStatus::InProgress, - ) - .await - .unwrap(); - - create_run(&loc, "run-stale", &card_id, "dispatcher").unwrap(); - - // check_staleness uses strict `>`, so we need the run to age past 0s. - sleep_for_staleness().await; - - let limits = RunLimits { - heartbeat_stale_secs: 0, - claim_ttl_secs: 0, - max_reclaim_count: 3, - }; - let result = reclaim_stale(&loc, &limits).await.unwrap(); - assert_eq!(result.reclaimed_count, 1); - - // After reclaim, the card should be back to Todo and no active runs. - let snap = list(&loc).await.unwrap(); - assert_eq!(snap.cards[0].status, TaskCardStatus::Todo); - - let runs = list_runs(&loc, Some(&card_id)).unwrap(); - assert!( - runs.iter().all(|r| !r.is_active()), - "no active runs should remain after reclaim" - ); - assert_board_invariants(&loc).await; -} - -// ── 3. Status invariant ─────────────────────────────────────────────── - -#[tokio::test] -async fn status_claim_rejects_wrong_expected_status() { - let dir = tempdir().unwrap(); - let loc = thread_loc(dir.path(), "status-inv-1"); - let card_id = add_card(&loc, "wrong status").await; - - // Card is Todo; claiming with expected=[InProgress] must fail. - let err = claim_card( - &loc, - &card_id, - &[TaskCardStatus::InProgress], - TaskCardStatus::Done, - ) - .await; - assert!(err.is_err(), "claim with wrong expected status must fail"); - - // Card should still be Todo. - let snap = list(&loc).await.unwrap(); - assert_eq!(snap.cards[0].status, TaskCardStatus::Todo); - assert_board_invariants(&loc).await; -} - -#[tokio::test] -async fn status_all_valid_statuses_accepted() { - let dir = tempdir().unwrap(); - let loc = thread_loc(dir.path(), "status-inv-2"); - - let statuses = [ - TaskCardStatus::Todo, - TaskCardStatus::AwaitingApproval, - TaskCardStatus::Ready, - TaskCardStatus::Blocked, - TaskCardStatus::Rejected, - TaskCardStatus::Done, - ]; - - for status in &statuses { - let card_id = add_card(&loc, &format!("card-{:?}", status)).await; - update_status(&loc, &card_id, status.clone()).await.unwrap(); - let snap = list(&loc).await.unwrap(); - let card = snap.cards.iter().find(|c| c.id == card_id).unwrap(); - assert_eq!(&card.status, status); - } - assert_board_invariants(&loc).await; -} - -#[tokio::test] -async fn status_claim_metadata_consistent_with_status() { - let dir = tempdir().unwrap(); - let loc = thread_loc(dir.path(), "status-inv-3"); - let card_id = add_card(&loc, "claim consistency").await; - - // Claim Todo → InProgress via claim_card. - let claimed = claim_card( - &loc, - &card_id, - &[TaskCardStatus::Todo], - TaskCardStatus::InProgress, - ) - .await - .unwrap(); - assert_eq!(claimed.status, TaskCardStatus::InProgress); - assert!( - !claimed.updated_at.is_empty(), - "updated_at must be set after claim" - ); - - // The board snapshot agrees with the returned card's status. - // (normalise_board may re-stamp updated_at, so only compare status.) - let snap = list(&loc).await.unwrap(); - let board_card = snap.cards.iter().find(|c| c.id == card_id).unwrap(); - assert_eq!(board_card.status, TaskCardStatus::InProgress); - assert!(!board_card.updated_at.is_empty()); - assert_board_invariants(&loc).await; -} - -#[tokio::test] -async fn status_blocked_card_has_blocker_after_reclaim() { - let dir = tempdir().unwrap(); - let loc = thread_loc(dir.path(), "status-inv-4"); - let card_id = add_card(&loc, "reclaim-to-blocked").await; - - let limits = RunLimits { - heartbeat_stale_secs: 0, - claim_ttl_secs: 0, - max_reclaim_count: 2, - }; - - // First reclaim (count=1 < max=2): card goes back to Todo. - claim_card( - &loc, - &card_id, - &[TaskCardStatus::Todo], - TaskCardStatus::InProgress, - ) - .await - .unwrap(); - create_run(&loc, "run-r1", &card_id, "d").unwrap(); - sleep_for_staleness().await; - let r1 = reclaim_stale(&loc, &limits).await.unwrap(); - assert_eq!(r1.reclaimed_count, 1); - - // Second reclaim (count=2 >= max=2): card goes to Blocked. - claim_card( - &loc, - &card_id, - &[TaskCardStatus::Todo], - TaskCardStatus::InProgress, - ) - .await - .unwrap(); - create_run(&loc, "run-r2", &card_id, "d").unwrap(); - sleep_for_staleness().await; - let r2 = reclaim_stale(&loc, &limits).await.unwrap(); - assert_eq!(r2.blocked_count, 1); - - let snap = list(&loc).await.unwrap(); - let card = snap.cards.iter().find(|c| c.id == card_id).unwrap(); - assert_eq!(card.status, TaskCardStatus::Blocked); - assert!( - card.blocker.is_some(), - "blocked card must have a blocker message" - ); - assert_board_invariants(&loc).await; -} - -// ── 4. Completion invariant ─────────────────────────────────────────── - -#[tokio::test] -async fn completion_run_cannot_be_completed_twice() { - let dir = tempdir().unwrap(); - let loc = thread_loc(dir.path(), "complete-inv-1"); - let card_id = add_card(&loc, "double complete").await; - - claim_card( - &loc, - &card_id, - &[TaskCardStatus::Todo], - TaskCardStatus::InProgress, - ) - .await - .unwrap(); - - create_run(&loc, "run-dc", &card_id, "dispatcher").unwrap(); - complete_run(&loc, "run-dc", RunOutcome::Success, None, vec![]).unwrap(); - - // Second completion must fail (run is no longer active). - let err = complete_run( - &loc, - "run-dc", - RunOutcome::Failed, - Some("retry".into()), - vec![], - ); - assert!( - err.is_err(), - "completing an already-completed run must fail" - ); -} - -#[tokio::test] -async fn completion_distinct_runs_same_card_only_one_active() { - let dir = tempdir().unwrap(); - let loc = thread_loc(dir.path(), "complete-inv-2"); - let card_id = add_card(&loc, "multi-run card").await; - - // First claim + run cycle. - claim_card( - &loc, - &card_id, - &[TaskCardStatus::Todo], - TaskCardStatus::InProgress, - ) - .await - .unwrap(); - create_run(&loc, "run-a", &card_id, "d1").unwrap(); - complete_run( - &loc, - "run-a", - RunOutcome::Failed, - Some("oops".into()), - vec![], - ) - .unwrap(); - - // Move card back to Todo for re-dispatch. - update_status(&loc, &card_id, TaskCardStatus::Todo) - .await - .unwrap(); - - // Second claim + run cycle. - claim_card( - &loc, - &card_id, - &[TaskCardStatus::Todo], - TaskCardStatus::InProgress, - ) - .await - .unwrap(); - create_run(&loc, "run-b", &card_id, "d2").unwrap(); - - // run-a is already completed, run-b is active. - let run_a = get_run(&loc, "run-a").unwrap().unwrap(); - let run_b = get_run(&loc, "run-b").unwrap().unwrap(); - assert!(!run_a.is_active()); - assert!(run_b.is_active()); - - assert_run_pointer_invariants(&loc).await; -} - -#[tokio::test] -async fn completion_heartbeat_after_completion_fails() { - let dir = tempdir().unwrap(); - let loc = thread_loc(dir.path(), "complete-inv-3"); - let card_id = add_card(&loc, "hb after complete").await; - - claim_card( - &loc, - &card_id, - &[TaskCardStatus::Todo], - TaskCardStatus::InProgress, - ) - .await - .unwrap(); - create_run(&loc, "run-hb", &card_id, "d").unwrap(); - complete_run(&loc, "run-hb", RunOutcome::Success, None, vec![]).unwrap(); - - let err = update_heartbeat(&loc, "run-hb"); - assert!(err.is_err(), "heartbeat on a completed run must fail"); -} - -// ── 5. Randomised operation sequence ────────────────────────────────── - -#[tokio::test] -async fn randomised_operation_sequence_maintains_invariants() { - // Deterministic PRNG via simple LCG. - struct Lcg(u64); - impl Lcg { - fn next(&mut self) -> u64 { - self.0 = self - .0 - .wrapping_mul(6364136223846793005) - .wrapping_add(1442695040888963407); - self.0 >> 33 - } - fn range(&mut self, max: u64) -> u64 { - if max == 0 { - return 0; - } - self.next() % max - } - } - - let dir = tempdir().unwrap(); - let loc = thread_loc(dir.path(), "rand-seq-1"); - let mut rng = Lcg(42); - let mut card_ids: Vec = Vec::new(); - let mut run_counter = 0u32; - - for _step in 0..200 { - let op = rng.range(6); - match op { - // Add a new card. - 0 => { - let title = format!("task-{}", card_ids.len()); - let id = add_card(&loc, &title).await; - card_ids.push(id); - } - // Claim a random card Todo → InProgress. - 1 if !card_ids.is_empty() => { - let idx = rng.range(card_ids.len() as u64) as usize; - let _ = claim_card( - &loc, - &card_ids[idx], - &[TaskCardStatus::Todo, TaskCardStatus::Ready], - TaskCardStatus::InProgress, - ) - .await; - } - // Complete the in-progress card's run (if any) then mark Done. - 2 => { - let snap = list(&loc).await.unwrap(); - if let Some(ip) = snap - .cards - .iter() - .find(|c| c.status == TaskCardStatus::InProgress) - { - let runs = list_runs(&loc, Some(&ip.id)).unwrap_or_default(); - if let Some(active) = runs.iter().find(|r| r.is_active()) { - let outcome = if rng.range(2) == 0 { - RunOutcome::Success - } else { - RunOutcome::Failed - }; - let _ = complete_run(&loc, &active.run_id, outcome, None, vec![]); - } - let _ = update_status(&loc, &ip.id, TaskCardStatus::Done).await; - } - } - // Create a run for an InProgress card that lacks one. - 3 => { - let snap = list(&loc).await.unwrap(); - if let Some(ip) = snap - .cards - .iter() - .find(|c| c.status == TaskCardStatus::InProgress) - { - let runs = list_runs(&loc, Some(&ip.id)).unwrap_or_default(); - if !runs.iter().any(|r| r.is_active()) { - run_counter += 1; - let _ = create_run(&loc, &format!("rnd-run-{run_counter}"), &ip.id, "d"); - } - } - } - // Move a random card back to Todo (simulating retry). - 4 if !card_ids.is_empty() => { - let idx = rng.range(card_ids.len() as u64) as usize; - let snap = list(&loc).await.unwrap(); - if let Some(card) = snap.cards.iter().find(|c| c.id == card_ids[idx]) { - if matches!( - card.status, - TaskCardStatus::Done | TaskCardStatus::Blocked | TaskCardStatus::Rejected - ) { - let _ = update_status(&loc, &card_ids[idx], TaskCardStatus::Todo).await; - } - } - } - // Heartbeat a random active run. - 5 => { - let runs = list_runs(&loc, None).unwrap_or_default(); - let active: Vec<_> = runs.iter().filter(|r| r.is_active()).collect(); - if !active.is_empty() { - let idx = rng.range(active.len() as u64) as usize; - let _ = update_heartbeat(&loc, &active[idx].run_id); - } - } - _ => {} - } - - // After every operation, all invariants must hold. - assert_board_invariants(&loc).await; - } -} - -// ── 6. Concurrent claim + run lifecycle stress ──────────────────────── - -#[tokio::test(flavor = "multi_thread", worker_threads = 4)] -async fn stress_claim_create_run_complete_cycle() { - let dir = tempdir().unwrap(); - let loc = thread_loc(dir.path(), "stress-lifecycle"); - let card_id = add_card(&loc, "lifecycle target").await; - - for round in 0..10 { - // N tasks race to claim. - let results = race_claims( - &loc, - &card_id, - 4, - vec![TaskCardStatus::Todo, TaskCardStatus::Ready], - ) - .await; - - let wins = results.iter().filter(|r| r.is_ok()).count(); - assert_eq!(wins, 1, "round {round}: exactly one claimer wins"); - - assert_board_invariants(&loc).await; - - // Winner creates a run, completes it, moves card back. - let run_id = format!("run-cycle-{round}"); - create_run(&loc, &run_id, &card_id, "d").unwrap(); - assert_run_pointer_invariants(&loc).await; - - complete_run(&loc, &run_id, RunOutcome::Success, None, vec![]).unwrap(); - update_status(&loc, &card_id, TaskCardStatus::Todo) - .await - .unwrap(); - assert_board_invariants(&loc).await; - } -} - -// ── 7. Claim after reclaim race ─────────────────────────────────────── - -#[tokio::test(flavor = "multi_thread", worker_threads = 4)] -async fn reclaim_then_concurrent_re_claim() { - let dir = tempdir().unwrap(); - let loc = thread_loc(dir.path(), "reclaim-race-1"); - let card_id = add_card(&loc, "reclaim-race").await; - - // Claim and create a run, then let it age and reclaim. - claim_card( - &loc, - &card_id, - &[TaskCardStatus::Todo], - TaskCardStatus::InProgress, - ) - .await - .unwrap(); - create_run(&loc, "run-pre-reclaim", &card_id, "d").unwrap(); - - sleep_for_staleness().await; - - let limits = RunLimits { - heartbeat_stale_secs: 0, - claim_ttl_secs: 0, - max_reclaim_count: 10, - }; - let result = reclaim_stale(&loc, &limits).await.unwrap(); - assert_eq!( - result.reclaimed_count, 1, - "run must be reclaimed after aging" - ); - - // Card is back to Todo. Now N tasks race to re-claim. - let results = race_claims(&loc, &card_id, 6, vec![TaskCardStatus::Todo]).await; - - assert_eq!( - results.iter().filter(|r| r.is_ok()).count(), - 1, - "exactly one re-claimer wins after reclaim" - ); - assert_board_invariants(&loc).await; -} diff --git a/src/openhuman/todos/mod.rs b/src/openhuman/todos/mod.rs index 1e05fda29..892e462a8 100644 --- a/src/openhuman/todos/mod.rs +++ b/src/openhuman/todos/mod.rs @@ -13,18 +13,12 @@ //! `markdown` string so the chat UI / agent transcript can render the //! list directly without re-formatting. -pub mod crate_adapter; pub mod ops; pub mod runs; pub mod schemas; -pub mod store; pub mod tools; -#[cfg(test)] -mod invariant_tests; - pub use schemas::{ all_controller_schemas as all_todos_controller_schemas, all_registered_controllers as all_todos_registered_controllers, }; -pub use store::{global_scratch_store, ScratchTodoStore}; diff --git a/src/openhuman/todos/ops.rs b/src/openhuman/todos/ops.rs index 2ca4546ce..1e3defc5c 100644 --- a/src/openhuman/todos/ops.rs +++ b/src/openhuman/todos/ops.rs @@ -1,79 +1,27 @@ -//! Core todo CRUD operations. +//! Compatibility facade over [`tinyagents::graph::todos`]. //! -//! Each operation loads the current cards for a thread (or the -//! process-global scratch store when no thread id is given), applies the -//! mutation, persists the result, and returns both the updated cards and -//! a markdown rendering. The agent-facing `todo` tool and the -//! `openhuman.todos_*` RPC handlers both call into this module so behavior -//! stays in lock-step across surfaces. +//! OpenHuman keeps its historical board-location and optional-thread snapshot +//! shapes for RPC and tool callers. All task-board data, normalization, CRUD, +//! and compare-and-set behavior is owned by TinyAgents. + +use std::path::PathBuf; +use std::sync::Arc; -use crate::openhuman::agent::progress::AgentProgress; -use crate::openhuman::agent::task_board::{ - normalise_board, TaskApprovalMode, TaskBoard, TaskBoardCard, TaskBoardStore, TaskCardStatus, -}; use chrono::Utc; use serde::{Deserialize, Serialize}; -use std::collections::HashMap; -use std::path::PathBuf; -use std::sync::{Arc, Mutex, OnceLock}; -use tokio::sync::Mutex as AsyncMutex; -use uuid::Uuid; +use tinyagents::graph::todos::store as todos; +use tinyagents::harness::store::Store; + +use crate::openhuman::agent::progress::AgentProgress; +use crate::openhuman::agent::task_board::normalize_cards_for_wire; +pub use crate::openhuman::agent::task_board::{TaskApprovalMode, TaskBoardCard, TaskCardStatus}; +use crate::openhuman::tinyagents::todos::{scratch_todos_store, todos_store, SCRATCH_THREAD_ID}; -/// Thread id backing the personal "user tasks" kanban board. -/// -/// Must match the frontend constant in `app/src/services/api/todosApi.ts`. -/// This is the board the kanban UI renders as the user's working lane and that -/// the task dispatcher's board poller executes **agent-assigned** cards on -/// (tasks approved out of the proactive `task-sources` inbox). Manually-created -/// human cards on this board carry no `assigned_agent` and are never auto-run. pub const USER_TASKS_THREAD_ID: &str = "user-tasks"; - -/// The orchestrator's single, app-wide task board. The orchestrator's `todo` -/// tool always targets this fixed board (not a per-thread one) so it owns one -/// global Kanban across every delegation — surfaced in the UI by -/// `OrchestratorTaskBoard` under the same id. pub const ORCHESTRATOR_TASKS_THREAD_ID: &str = "orchestrator-tasks"; -use super::store::{global_scratch_store, ScratchTodoStore}; +pub use tinyagents::graph::todos::{parse_status, render_markdown, CardPatch}; -/// Per-board **async** mutex map for serialising every read/modify/write -/// operation. Keyed by a -/// canonical board key (thread_id for `Thread`, `"_scratch_"` for `Scratch`). -/// The outer sync `Mutex` only guards the map itself (never held across an -/// await); the inner `Arc>` is held across load → mutate → save, -/// including its `.await` points. -fn board_lock(location: &BoardLocation) -> Arc> { - static MAP: OnceLock>>>> = OnceLock::new(); - let map_mu = MAP.get_or_init(|| Mutex::new(HashMap::new())); - let key = match location { - BoardLocation::Thread { thread_id, .. } => thread_id.clone(), - BoardLocation::Scratch => "_scratch_".to_string(), - }; - map_mu - .lock() - .expect("todos board lock map poisoned") - .entry(key) - .or_default() - .clone() -} - -/// Stable string aliases accepted on the wire for [`TaskCardStatus`]. -pub fn parse_status(raw: &str) -> Result { - match raw.trim().to_ascii_lowercase().as_str() { - "todo" | "pending" => Ok(TaskCardStatus::Todo), - "awaiting_approval" | "awaiting-approval" => Ok(TaskCardStatus::AwaitingApproval), - "ready" | "approved" => Ok(TaskCardStatus::Ready), - "in_progress" | "in-progress" | "inprogress" | "started" => Ok(TaskCardStatus::InProgress), - "blocked" => Ok(TaskCardStatus::Blocked), - "done" | "completed" | "complete" => Ok(TaskCardStatus::Done), - "rejected" | "denied" => Ok(TaskCardStatus::Rejected), - other => Err(format!( - "invalid status '{other}' (expected todo|awaiting_approval|ready|in_progress|blocked|done|rejected)" - )), - } -} - -/// A single CRUD outcome: the post-mutation cards plus a markdown rendering. #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct TodosSnapshot { @@ -82,562 +30,47 @@ pub struct TodosSnapshot { pub markdown: String, } -/// Optional fields supplied by `add` / `edit` callers. -#[derive(Debug, Default, Clone)] -pub struct CardPatch { - pub content: Option, - pub status: Option, - pub objective: Option, - pub plan: Option>, - pub assigned_agent: Option, - pub allowed_tools: Option>, - pub approval_mode: Option>, - pub acceptance_criteria: Option>, - pub evidence: Option>, - pub notes: Option, - pub blocker: Option, - /// Provider/source identifiers for a task-source-ingested card. `Some` - /// sets the card's `source_metadata`; `None` leaves it untouched. - pub source_metadata: Option, -} - -/// Where to load/save the working set of cards. #[derive(Debug, Clone)] pub enum BoardLocation { - /// Persisted to `/agent_task_boards/.json`. Thread { workspace_dir: PathBuf, thread_id: String, }, - /// In-memory only, shared across the process. Scratch, } impl BoardLocation { pub fn thread_id(&self) -> Option<&str> { match self { - Self::Thread { thread_id, .. } => Some(thread_id.as_str()), + Self::Thread { thread_id, .. } => Some(thread_id), Self::Scratch => None, } } } -async fn load_cards(location: &BoardLocation) -> Result, String> { +fn target(location: &BoardLocation) -> (Arc, &str) { match location { BoardLocation::Thread { workspace_dir, thread_id, - } => { - let store = TaskBoardStore::new(workspace_dir.clone()); - Ok(store - .get(thread_id) - .await? - .map(|board| board.cards) - .unwrap_or_default()) - } - BoardLocation::Scratch => Ok(global_scratch_store().snapshot()), + } => (todos_store(workspace_dir), thread_id), + BoardLocation::Scratch => (scratch_todos_store(), SCRATCH_THREAD_ID), } } -async fn save_cards( +fn snapshot( location: &BoardLocation, - cards: Vec, -) -> Result, String> { - match location { - BoardLocation::Thread { - workspace_dir, - thread_id, - } => { - let board = TaskBoard { - thread_id: thread_id.clone(), - cards, - updated_at: Utc::now().to_rfc3339(), - }; - // `TaskBoardStore::put` normalises (minting `task-` ids for - // blank cards), enforces the single-`InProgress` invariant, and - // persists into the crate `graph.todos` store. - let store = TaskBoardStore::new(workspace_dir.clone()); - let saved = store.put(board).await?.cards; - Ok(saved) - } - BoardLocation::Scratch => { - let mut board = TaskBoard { - thread_id: "_scratch_".to_string(), - cards, - updated_at: Utc::now().to_rfc3339(), - }; - normalise_board(&mut board); - let scratch: std::sync::Arc = global_scratch_store(); - scratch.replace(board.cards.clone()); - Ok(board.cards) - } - } -} - -fn into_snapshot(location: &BoardLocation, cards: Vec) -> TodosSnapshot { - let markdown = render_markdown(&cards); + value: tinyagents::graph::todos::TodosSnapshot, +) -> TodosSnapshot { TodosSnapshot { - thread_id: location.thread_id().map(|s| s.to_string()), - cards, - markdown, + thread_id: location.thread_id().map(str::to_owned), + cards: value.cards, + markdown: value.markdown, } } -/// Render a card list as GitHub-flavored markdown. Each card becomes a -/// `- [ ]` / `- [x]` line (with `[~]` for in-progress and `[!]` for -/// blocked) followed by indented notes / blocker reasons. -pub fn render_markdown(cards: &[TaskBoardCard]) -> String { - if cards.is_empty() { - return "_No todos yet._".to_string(); - } - let mut out = String::new(); - for card in cards { - let marker = match card.status { - TaskCardStatus::Todo | TaskCardStatus::Ready => "[ ]", - TaskCardStatus::AwaitingApproval => "[?]", - TaskCardStatus::InProgress => "[~]", - TaskCardStatus::Blocked => "[!]", - TaskCardStatus::Done => "[x]", - TaskCardStatus::Rejected => "[-]", - }; - out.push_str("- "); - out.push_str(marker); - out.push(' '); - out.push_str(&card.title); - out.push_str(&format!(" `({})`", card.id)); - out.push('\n'); - - if let Some(objective) = card.objective.as_deref() { - out.push_str(" - objective: "); - out.push_str(objective); - out.push('\n'); - } - if let Some(agent) = card.assigned_agent.as_deref() { - out.push_str(" - agent: "); - out.push_str(agent); - out.push('\n'); - } - if !card.allowed_tools.is_empty() { - out.push_str(" - tools: "); - out.push_str(&card.allowed_tools.join(", ")); - out.push('\n'); - } - if let Some(mode) = card.approval_mode.as_ref() { - out.push_str(" - approval: "); - out.push_str(mode.as_str()); - out.push('\n'); - } - if !card.plan.is_empty() { - out.push_str(" - plan:\n"); - for step in &card.plan { - out.push_str(" - "); - out.push_str(step); - out.push('\n'); - } - } - if !card.acceptance_criteria.is_empty() { - out.push_str(" - acceptance criteria:\n"); - for criterion in &card.acceptance_criteria { - out.push_str(" - "); - out.push_str(criterion); - out.push('\n'); - } - } - if !card.evidence.is_empty() { - out.push_str(" - evidence:\n"); - for item in &card.evidence { - out.push_str(" - "); - out.push_str(item); - out.push('\n'); - } - } - - if matches!(card.status, TaskCardStatus::Blocked) { - if let Some(reason) = card.blocker.as_deref().or(card.notes.as_deref()) { - out.push_str(" - _blocked:_ "); - out.push_str(reason); - out.push('\n'); - } - } else if let Some(notes) = card.notes.as_deref() { - out.push_str(" - "); - out.push_str(notes); - out.push('\n'); - } - } - out.trim_end().to_string() -} - -/// Append a new card. `content` is required; missing status defaults to -/// `todo`. -pub async fn add( - location: &BoardLocation, - content: &str, - patch: CardPatch, -) -> Result { - tracing::debug!( - thread_id = ?location.thread_id(), - content_len = content.len(), - "[todos][ops] add entry" - ); - let lock = board_lock(location); - let _guard = lock.lock().await; - let content = content.trim(); - if content.is_empty() { - return Err("todo content must not be empty".to_string()); - } - let mut cards = load_cards(location).await?; - let new_card = TaskBoardCard { - id: format!("task-{}", Uuid::new_v4()), - title: content.to_string(), - status: patch.status.unwrap_or(TaskCardStatus::Todo), - objective: patch.objective.and_then(non_empty), - plan: patch.plan.unwrap_or_default(), - assigned_agent: patch.assigned_agent.and_then(non_empty), - allowed_tools: patch.allowed_tools.unwrap_or_default(), - approval_mode: patch.approval_mode.flatten(), - acceptance_criteria: patch.acceptance_criteria.unwrap_or_default(), - evidence: patch.evidence.unwrap_or_default(), - notes: patch.notes.and_then(non_empty), - blocker: patch.blocker.and_then(non_empty), - session_thread_id: None, - source_metadata: patch.source_metadata, - order: cards.len() as u32, - updated_at: Utc::now().to_rfc3339(), - }; - cards.push(new_card); - enforce_single_in_progress(&cards)?; - let cards = save_cards(location, cards).await?; - emit_progress(location, &cards); - Ok(into_snapshot(location, cards)) -} - -/// Edit an existing card's content / notes / blocker / status. Any field -/// left as `None` in `patch` is left untouched. Errors if `id` is unknown. -pub async fn edit( - location: &BoardLocation, - id: &str, - patch: CardPatch, -) -> Result { - tracing::debug!( - thread_id = ?location.thread_id(), - id, - "[todos][ops] edit entry" - ); - let lock = board_lock(location); - let _guard = lock.lock().await; - let mut cards = load_cards(location).await?; - let card = cards - .iter_mut() - .find(|c| c.id == id) - .ok_or_else(|| format!("todo id '{id}' not found"))?; - if let Some(content) = patch.content { - let trimmed = content.trim().to_string(); - if trimmed.is_empty() { - return Err("todo content must not be empty".to_string()); - } - card.title = trimmed; - } - if let Some(status) = patch.status { - card.status = status; - } - if let Some(objective) = patch.objective { - card.objective = non_empty(objective); - } - if let Some(plan) = patch.plan { - card.plan = plan; - } - if let Some(assigned_agent) = patch.assigned_agent { - card.assigned_agent = non_empty(assigned_agent); - } - if let Some(allowed_tools) = patch.allowed_tools { - card.allowed_tools = allowed_tools; - } - if let Some(approval_mode) = patch.approval_mode { - card.approval_mode = approval_mode; - } - if let Some(acceptance_criteria) = patch.acceptance_criteria { - card.acceptance_criteria = acceptance_criteria; - } - if let Some(evidence) = patch.evidence { - card.evidence = evidence; - } - if let Some(notes) = patch.notes { - card.notes = non_empty(notes); - } - if let Some(blocker) = patch.blocker { - card.blocker = non_empty(blocker); - } - if let Some(source_metadata) = patch.source_metadata { - card.source_metadata = Some(source_metadata); - } - card.updated_at = Utc::now().to_rfc3339(); - enforce_single_in_progress(&cards)?; - let cards = save_cards(location, cards).await?; - emit_progress(location, &cards); - Ok(into_snapshot(location, cards)) -} - -/// Stamp (or clear) a card's `session_thread_id` — the conversation thread of -/// its live/last agent run — so the UI can offer a "View session" jump into -/// Conversations. Used by the autonomous dispatcher (`task_session`, direct -/// call) and the manual "Work" path (via the `todos_set_session_thread` RPC). -/// A blank id clears the link. Does NOT touch status or `enforce_single_in_progress` -/// — this is pure session-link bookkeeping, orthogonal to the card lifecycle. -pub async fn set_session_thread( - location: &BoardLocation, - id: &str, - session_thread_id: Option, -) -> Result { - let lock = board_lock(location); - let _guard = lock.lock().await; - let mut cards = load_cards(location).await?; - let card = cards - .iter_mut() - .find(|c| c.id == id) - .ok_or_else(|| format!("todo id '{id}' not found"))?; - card.session_thread_id = session_thread_id.and_then(non_empty); - card.updated_at = Utc::now().to_rfc3339(); - let cards = save_cards(location, cards).await?; - emit_progress(location, &cards); - Ok(into_snapshot(location, cards)) -} - -/// Update only the status of a card. -pub async fn update_status( - location: &BoardLocation, - id: &str, - status: TaskCardStatus, -) -> Result { - edit( - location, - id, - CardPatch { - status: Some(status), - ..Default::default() - }, - ) - .await -} - -/// Resolve a plan-approval decision: approve (→`Ready`, so the dispatcher runs -/// it) or reject (→`Rejected`). Errors unless the card is currently -/// `AwaitingApproval`, so a stale/duplicate decision can't resurrect a card -/// that already moved on. -pub async fn decide_plan( - location: &BoardLocation, - id: &str, - approve: bool, -) -> Result { - let lock = board_lock(location); - let _guard = lock.lock().await; - let mut cards = load_cards(location).await?; - let current = cards - .iter_mut() - .find(|c| c.id == id) - .ok_or_else(|| format!("todo id '{id}' not found"))?; - if current.status != TaskCardStatus::AwaitingApproval { - return Err(format!( - "card '{id}' is not awaiting approval (status: {})", - current.status.as_str() - )); - } - current.status = if approve { - TaskCardStatus::Ready - } else { - TaskCardStatus::Rejected - }; - current.updated_at = Utc::now().to_rfc3339(); - let cards = save_cards(location, cards).await?; - emit_progress(location, &cards); - Ok(into_snapshot(location, cards)) -} - -/// Clear a parked plan for re-planning. Transitions **every** -/// `AwaitingApproval` card on the board to `Rejected` so none stays runnable, -/// then returns the fresh snapshot. The caller (the plan-review surface) sends -/// the user's `feedback` back into the thread as a normal message so the -/// orchestrator re-plans and re-parks a new plan. Lenient when nothing is -/// awaiting — a benign no-op (returns the snapshot unchanged) rather than an -/// error, so a racing decision can't strand the feedback message. -pub async fn revise_plan( - location: &BoardLocation, - feedback: &str, -) -> Result { - let lock = board_lock(location); - let _guard = lock.lock().await; - let mut cards = load_cards(location).await?; - let mut revised = 0usize; - for card in cards.iter_mut() { - if card.status == TaskCardStatus::AwaitingApproval { - card.status = TaskCardStatus::Rejected; - card.updated_at = Utc::now().to_rfc3339(); - revised += 1; - } - } - tracing::info!( - thread_id = ?location.thread_id(), - revised, - feedback_len = feedback.len(), - "[todos][ops] revise_plan rejected awaiting cards for re-plan" - ); - let cards = save_cards(location, cards).await?; - emit_progress(location, &cards); - Ok(into_snapshot(location, cards)) -} - -/// Remove a card by id. Errors if `id` is unknown. -pub async fn remove(location: &BoardLocation, id: &str) -> Result { - tracing::debug!( - thread_id = ?location.thread_id(), - id, - "[todos][ops] remove entry" - ); - let lock = board_lock(location); - let _guard = lock.lock().await; - let mut cards = load_cards(location).await?; - let before = cards.len(); - cards.retain(|c| c.id != id); - if cards.len() == before { - return Err(format!("todo id '{id}' not found")); - } - let cards = save_cards(location, cards).await?; - emit_progress(location, &cards); - Ok(into_snapshot(location, cards)) -} - -/// Wholesale replace the list. Generates ids for cards missing them. -pub async fn replace( - location: &BoardLocation, - cards: Vec, -) -> Result { - tracing::debug!( - thread_id = ?location.thread_id(), - card_count = cards.len(), - "[todos][ops] replace entry" - ); - let lock = board_lock(location); - let _guard = lock.lock().await; - enforce_single_in_progress(&cards)?; - let cards = save_cards(location, cards).await?; - emit_progress(location, &cards); - Ok(into_snapshot(location, cards)) -} - -/// Empty the list. -pub async fn clear(location: &BoardLocation) -> Result { - tracing::debug!(thread_id = ?location.thread_id(), "[todos][ops] clear entry"); - let lock = board_lock(location); - let _guard = lock.lock().await; - let cards = save_cards(location, Vec::new()).await?; - emit_progress(location, &cards); - Ok(into_snapshot(location, cards)) -} - -/// Snapshot the current list without mutating. -pub async fn list(location: &BoardLocation) -> Result { - let lock = board_lock(location); - let _guard = lock.lock().await; - let cards = load_cards(location).await?; - Ok(into_snapshot(location, cards)) -} - -/// Atomic compare-and-set claim: transition a card from one of the -/// `expected` statuses to `target` under a per-board lock, returning the -/// **fresh** card snapshot on success. If the card's current status is not -/// in `expected`, the claim is rejected with `Err` — the caller lost the -/// race or the card already moved on. -/// -/// This is the single safe entry-point for the dispatcher to claim a card; -/// callers must **not** do a manual load→check→write outside this lock. -pub async fn claim_card( - location: &BoardLocation, - card_id: &str, - expected: &[TaskCardStatus], - target: TaskCardStatus, -) -> Result { - let lock = board_lock(location); - let _guard = lock.lock().await; - - tracing::debug!( - card_id = %card_id, - expected = ?expected.iter().map(TaskCardStatus::as_str).collect::>(), - target = %target.as_str(), - "[todos][ops] claim_card entry" - ); - - let mut cards = load_cards(location).await?; - - // The whole load → check → write runs under the per-board async lock above, - // so the compare-and-set stays atomic within the process (the single-writer - // core). A rejected claim (card missing / wrong status) short-circuits here. - let claimed_card = apply_claim(&mut cards, card_id, expected, target)?; - let saved = save_cards(location, cards).await?; - emit_progress(location, &saved); - tracing::info!( - card_id = %card_id, - new_status = %claimed_card.status.as_str(), - "[todos][ops] claim_card ok" - ); - Ok(claimed_card) -} - -/// Applies a claim to an in-memory card set: find `card_id`, verify its status -/// is in `expected`, transition it to `target`, and enforce the single- -/// `InProgress` invariant. Returns the claimed card (cloned) on success. Does -/// **not** persist — the caller saves the mutated `cards`. Extracted so -/// [`claim_card`] can capture a single ok/err verdict for its crate shadow. -fn apply_claim( - cards: &mut [TaskBoardCard], - card_id: &str, - expected: &[TaskCardStatus], - target: TaskCardStatus, -) -> Result { - let card = cards - .iter_mut() - .find(|c| c.id == card_id) - .ok_or_else(|| format!("[todos][ops] claim_card: card '{card_id}' not found on board"))?; - - if !expected.contains(&card.status) { - let current = card.status.as_str(); - return Err(format!( - "[todos][ops] claim_card: card '{card_id}' status is '{current}', \ - expected one of [{}]; claim rejected", - expected - .iter() - .map(TaskCardStatus::as_str) - .collect::>() - .join(", ") - )); - } - - card.status = target; - card.updated_at = Utc::now().to_rfc3339(); - let claimed_card = card.clone(); - - enforce_single_in_progress(cards)?; - Ok(claimed_card) -} - -fn non_empty(s: String) -> Option { - let trimmed = s.trim(); - if trimmed.is_empty() { - None - } else { - Some(trimmed.to_string()) - } -} - -fn enforce_single_in_progress(cards: &[TaskBoardCard]) -> Result<(), String> { - let in_progress = cards - .iter() - .filter(|c| matches!(c.status, TaskCardStatus::InProgress)) - .count(); - if in_progress > 1 { - return Err(format!( - "only one todo may be `in_progress` at a time (got {in_progress})" - )); - } - Ok(()) +fn progress_updated_at() -> String { + Utc::now().to_rfc3339() } fn emit_progress(location: &BoardLocation, cards: &[TaskBoardCard]) { @@ -650,27 +83,141 @@ fn emit_progress(location: &BoardLocation, cards: &[TaskBoardCard]) { let Some(tx) = parent.on_progress else { return; }; - let board = TaskBoard { + let board = tinyagents::graph::todos::TaskBoard { thread_id: thread_id.clone(), cards: cards.to_vec(), - updated_at: Utc::now().to_rfc3339(), + updated_at: progress_updated_at(), }; - if let Err(err) = tx.try_send(AgentProgress::TaskBoardUpdated { board }) { - tracing::debug!( - thread_id = %thread_id, - error = %err, - "[todos][ops] task board progress dropped" - ); + if let Err(error) = tx.try_send(AgentProgress::TaskBoardUpdated { board }) { + tracing::debug!(thread_id, %error, "task board progress dropped"); } } -/// Process-global lock that test code (here and in -/// `agent::tools::todo`) uses to serialize access to the shared -/// scratch store under `cargo test`'s parallel runner. +fn finish( + location: &BoardLocation, + result: tinyagents::error::Result, +) -> Result { + let mut value = result.map_err(|error| error.to_string())?; + normalize_cards_for_wire(&mut value.cards); + emit_progress(location, &value.cards); + Ok(snapshot(location, value)) +} + +pub async fn add( + location: &BoardLocation, + content: &str, + patch: CardPatch, +) -> Result { + let (store, thread_id) = target(location); + finish( + location, + todos::add(&store, thread_id, content, patch).await, + ) +} + +pub async fn edit( + location: &BoardLocation, + id: &str, + patch: CardPatch, +) -> Result { + let (store, thread_id) = target(location); + finish(location, todos::edit(&store, thread_id, id, patch).await) +} + +pub async fn update_status( + location: &BoardLocation, + id: &str, + status: TaskCardStatus, +) -> Result { + let (store, thread_id) = target(location); + finish( + location, + todos::update_status(&store, thread_id, id, status).await, + ) +} + +pub async fn set_session_thread( + location: &BoardLocation, + id: &str, + session_thread_id: Option, +) -> Result { + let (store, thread_id) = target(location); + finish( + location, + todos::set_session_thread(&store, thread_id, id, session_thread_id).await, + ) +} + +pub async fn decide_plan( + location: &BoardLocation, + id: &str, + approve: bool, +) -> Result { + let (store, thread_id) = target(location); + finish( + location, + todos::decide_plan(&store, thread_id, id, approve).await, + ) +} + +pub async fn revise_plan( + location: &BoardLocation, + feedback: &str, +) -> Result { + let (store, thread_id) = target(location); + tracing::info!( + thread_id, + feedback_len = feedback.len(), + "[todos][ops] revise_plan requested re-plan" + ); + finish(location, todos::revise_plan(&store, thread_id).await) +} + +pub async fn remove(location: &BoardLocation, id: &str) -> Result { + let (store, thread_id) = target(location); + finish(location, todos::remove(&store, thread_id, id).await) +} + +pub async fn replace( + location: &BoardLocation, + cards: Vec, +) -> Result { + let (store, thread_id) = target(location); + finish(location, todos::replace(&store, thread_id, cards).await) +} + +pub async fn clear(location: &BoardLocation) -> Result { + let (store, thread_id) = target(location); + finish(location, todos::clear(&store, thread_id).await) +} + +pub async fn list(location: &BoardLocation) -> Result { + let (store, thread_id) = target(location); + todos::list(&store, thread_id) + .await + .map(|value| snapshot(location, value)) + .map_err(|error| error.to_string()) +} + +pub async fn claim_card( + location: &BoardLocation, + card_id: &str, + expected: &[TaskCardStatus], + target_status: TaskCardStatus, +) -> Result { + let (store, thread_id) = target(location); + let card = todos::claim_card(&store, thread_id, card_id, expected, target_status) + .await + .map_err(|error| error.to_string())?; + if let Ok(value) = todos::list(&store, thread_id).await { + emit_progress(location, &value.cards); + } + Ok(card) +} + #[cfg(test)] pub(crate) fn scratch_test_lock() -> std::sync::MutexGuard<'static, ()> { - use std::sync::Mutex; - use std::sync::OnceLock; + use std::sync::{Mutex, OnceLock}; static LOCK: OnceLock> = OnceLock::new(); LOCK.get_or_init(|| Mutex::new(())) .lock() @@ -679,524 +226,12 @@ pub(crate) fn scratch_test_lock() -> std::sync::MutexGuard<'static, ()> { #[cfg(test)] mod tests { - use super::*; - use tempfile::tempdir; - - fn thread_loc(dir: &std::path::Path, id: &str) -> BoardLocation { - BoardLocation::Thread { - workspace_dir: dir.to_path_buf(), - thread_id: id.to_string(), - } - } - #[test] - fn parse_status_accepts_aliases() { - assert_eq!(parse_status("todo").unwrap(), TaskCardStatus::Todo); - assert_eq!(parse_status("PENDING").unwrap(), TaskCardStatus::Todo); - assert_eq!( - parse_status("in_progress").unwrap(), - TaskCardStatus::InProgress + fn progress_timestamp_preserves_rfc3339_wire_format() { + let timestamp = super::progress_updated_at(); + assert!( + chrono::DateTime::parse_from_rfc3339(×tamp).is_ok(), + "progress-event updated_at must remain RFC 3339" ); - assert_eq!(parse_status("blocked").unwrap(), TaskCardStatus::Blocked); - assert_eq!(parse_status("done").unwrap(), TaskCardStatus::Done); - assert_eq!( - parse_status("awaiting_approval").unwrap(), - TaskCardStatus::AwaitingApproval - ); - assert_eq!(parse_status("ready").unwrap(), TaskCardStatus::Ready); - assert_eq!(parse_status("approved").unwrap(), TaskCardStatus::Ready); - assert_eq!(parse_status("rejected").unwrap(), TaskCardStatus::Rejected); - assert!(parse_status("nope").is_err()); - } - - #[tokio::test] - async fn set_session_thread_links_then_clears() { - let dir = tempdir().unwrap(); - let loc = thread_loc(dir.path(), "t1"); - let snap = add(&loc, "Do the thing", CardPatch::default()) - .await - .unwrap(); - let card_id = snap.cards[0].id.clone(); - - // Link a session thread → exposed on the card for the UI "View session". - let linked = set_session_thread(&loc, &card_id, Some("thread-xyz".into())) - .await - .unwrap(); - assert_eq!( - linked.cards[0].session_thread_id.as_deref(), - Some("thread-xyz") - ); - - // A blank id clears the link (non_empty trims to None). - let cleared = set_session_thread(&loc, &card_id, Some(" ".into())) - .await - .unwrap(); - assert!(cleared.cards[0].session_thread_id.is_none()); - - // Unknown card id is an error, not a silent no-op. - assert!(set_session_thread(&loc, "missing", Some("t".into())) - .await - .is_err()); - } - - #[tokio::test] - async fn add_appends_and_returns_markdown() { - let dir = tempdir().unwrap(); - let loc = thread_loc(dir.path(), "t1"); - let snap = add( - &loc, - "First task", - CardPatch { - objective: Some("Ship a richer handoff".into()), - plan: Some(vec![ - "Inspect existing board".into(), - "Update schema".into(), - ]), - assigned_agent: Some("planner".into()), - allowed_tools: Some(vec!["todo".into(), "spawn_subagent".into()]), - approval_mode: Some(Some(TaskApprovalMode::Required)), - acceptance_criteria: Some(vec!["Tests pass".into()]), - evidence: Some(vec!["cargo test".into()]), - ..Default::default() - }, - ) - .await - .unwrap(); - assert_eq!(snap.cards.len(), 1); - assert!(snap.markdown.contains("[ ] First task")); - assert!(snap.markdown.contains("objective: Ship a richer handoff")); - assert!(snap.markdown.contains("agent: planner")); - assert!(snap.markdown.contains("tools: todo, spawn_subagent")); - assert!(snap.markdown.contains("approval: required")); - assert!(snap.markdown.contains("Inspect existing board")); - assert!(snap.markdown.contains("Tests pass")); - assert!(snap.markdown.contains("cargo test")); - assert!(snap.markdown.contains(&snap.cards[0].id)); - } - - #[tokio::test(flavor = "multi_thread", worker_threads = 4)] - async fn concurrent_adds_preserve_every_thread_board_mutation() { - const WRITERS: usize = 12; - let dir = tempdir().unwrap(); - let loc = thread_loc(dir.path(), "concurrent-adds"); - let barrier = Arc::new(tokio::sync::Barrier::new(WRITERS)); - let mut handles = Vec::with_capacity(WRITERS); - - for index in 0..WRITERS { - let loc = loc.clone(); - let barrier = barrier.clone(); - handles.push(tokio::spawn(async move { - barrier.wait().await; - add(&loc, &format!("task {index}"), CardPatch::default()).await - })); - } - - for handle in handles { - handle.await.unwrap().unwrap(); - } - - let snapshot = list(&loc).await.unwrap(); - assert_eq!(snapshot.cards.len(), WRITERS); - for index in 0..WRITERS { - assert!( - snapshot - .cards - .iter() - .any(|card| card.title == format!("task {index}")), - "missing concurrent mutation {index}" - ); - } - } - - #[tokio::test] - async fn edit_updates_fields_by_id() { - let dir = tempdir().unwrap(); - let loc = thread_loc(dir.path(), "t1"); - let added = add(&loc, "Draft plan", CardPatch::default()).await.unwrap(); - let id = added.cards[0].id.clone(); - let snap = edit( - &loc, - &id, - CardPatch { - content: Some("Refined plan".into()), - ..Default::default() - }, - ) - .await - .unwrap(); - assert_eq!(snap.cards[0].title, "Refined plan"); - } - - #[tokio::test] - async fn source_metadata_round_trips_through_add_and_edit() { - let dir = tempdir().unwrap(); - let loc = thread_loc(dir.path(), "t1"); - let added = add( - &loc, - "ingested task", - CardPatch { - source_metadata: Some(serde_json::json!({ - "provider": "github", - "external_id": "7", - })), - ..Default::default() - }, - ) - .await - .unwrap(); - let id = added.cards[0].id.clone(); - assert_eq!( - added.cards[0].source_metadata.as_ref().unwrap()["external_id"], - serde_json::json!("7") - ); - - // A subsequent edit with `Some(..)` replaces the stamped metadata. - let snap = edit( - &loc, - &id, - CardPatch { - source_metadata: Some(serde_json::json!({ - "provider": "github", - "external_id": "8", - })), - ..Default::default() - }, - ) - .await - .unwrap(); - assert_eq!( - snap.cards[0].source_metadata.as_ref().unwrap()["external_id"], - serde_json::json!("8") - ); - - // An edit that leaves `source_metadata: None` preserves the value. - let snap2 = edit( - &loc, - &id, - CardPatch { - notes: Some("touch".into()), - ..Default::default() - }, - ) - .await - .unwrap(); - assert_eq!( - snap2.cards[0].source_metadata.as_ref().unwrap()["external_id"], - serde_json::json!("8") - ); - } - - #[tokio::test] - async fn decide_plan_approves_and_rejects_only_when_awaiting() { - let dir = tempdir().unwrap(); - let loc = thread_loc(dir.path(), "t1"); - let added = add(&loc, "task", CardPatch::default()).await.unwrap(); - let id = added.cards[0].id.clone(); - - // A todo card isn't awaiting approval yet → decision rejected. - assert!(decide_plan(&loc, &id, true).await.is_err()); - - // Park it, then approve → Ready. - update_status(&loc, &id, TaskCardStatus::AwaitingApproval) - .await - .unwrap(); - let approved = decide_plan(&loc, &id, true).await.unwrap(); - assert_eq!(approved.cards[0].status, TaskCardStatus::Ready); - - // Re-park, then reject → Rejected. - update_status(&loc, &id, TaskCardStatus::AwaitingApproval) - .await - .unwrap(); - let rejected = decide_plan(&loc, &id, false).await.unwrap(); - assert_eq!(rejected.cards[0].status, TaskCardStatus::Rejected); - } - - #[tokio::test] - async fn revise_plan_rejects_only_awaiting_cards() { - let dir = tempdir().unwrap(); - let loc = thread_loc(dir.path(), "t1"); - // Each `add` returns the whole board; the new card is the last one. - let a = add(&loc, "A", CardPatch::default()).await.unwrap(); - let b = add(&loc, "B", CardPatch::default()).await.unwrap(); - let c = add(&loc, "C", CardPatch::default()).await.unwrap(); - let a_id = a.cards.last().unwrap().id.clone(); - let b_id = b.cards.last().unwrap().id.clone(); - let c_id = c.cards.last().unwrap().id.clone(); - - // Two cards parked for review, one left as a plain todo. - update_status(&loc, &a_id, TaskCardStatus::AwaitingApproval) - .await - .unwrap(); - update_status(&loc, &b_id, TaskCardStatus::AwaitingApproval) - .await - .unwrap(); - - let snap = revise_plan(&loc, "please add a verification step") - .await - .unwrap(); - let by_id = |id: &str| { - snap.cards - .iter() - .find(|c| c.id == id) - .unwrap() - .status - .clone() - }; - assert_eq!(by_id(&a_id), TaskCardStatus::Rejected); - assert_eq!(by_id(&b_id), TaskCardStatus::Rejected); - // The non-awaiting card is untouched. - assert_eq!(by_id(&c_id), TaskCardStatus::Todo); - } - - #[tokio::test] - async fn revise_plan_is_noop_when_nothing_awaiting() { - let dir = tempdir().unwrap(); - let loc = thread_loc(dir.path(), "t1"); - let a = add(&loc, "A", CardPatch::default()).await.unwrap(); - let a_id = a.cards[0].id.clone(); - let snap = revise_plan(&loc, "tweak it").await.unwrap(); - assert_eq!(snap.cards.len(), 1); - assert_eq!(snap.cards[0].id, a_id); - assert_eq!(snap.cards[0].status, TaskCardStatus::Todo); - } - - #[tokio::test] - async fn edit_can_clear_approval_mode() { - let dir = tempdir().unwrap(); - let loc = thread_loc(dir.path(), "t1"); - let added = add( - &loc, - "Draft plan", - CardPatch { - approval_mode: Some(Some(TaskApprovalMode::Required)), - ..Default::default() - }, - ) - .await - .unwrap(); - let id = added.cards[0].id.clone(); - - let snap = edit( - &loc, - &id, - CardPatch { - approval_mode: Some(None), - ..Default::default() - }, - ) - .await - .unwrap(); - - assert_eq!(snap.cards[0].approval_mode, None); - } - - #[tokio::test] - async fn edit_unknown_id_errors() { - let dir = tempdir().unwrap(); - let loc = thread_loc(dir.path(), "t1"); - let err = edit(&loc, "task-missing", CardPatch::default()) - .await - .unwrap_err(); - assert!(err.contains("not found")); - } - - #[tokio::test] - async fn update_status_changes_only_status() { - let dir = tempdir().unwrap(); - let loc = thread_loc(dir.path(), "t1"); - let added = add(&loc, "Write tests", CardPatch::default()) - .await - .unwrap(); - let id = added.cards[0].id.clone(); - let snap = update_status(&loc, &id, TaskCardStatus::Done) - .await - .unwrap(); - assert_eq!(snap.cards[0].status, TaskCardStatus::Done); - assert!(snap.markdown.contains("[x] Write tests")); - } - - #[tokio::test] - async fn remove_drops_card_by_id() { - let dir = tempdir().unwrap(); - let loc = thread_loc(dir.path(), "t1"); - let a = add(&loc, "A", CardPatch::default()).await.unwrap(); - let _ = add(&loc, "B", CardPatch::default()).await.unwrap(); - let snap = remove(&loc, &a.cards[0].id).await.unwrap(); - assert_eq!(snap.cards.len(), 1); - assert_eq!(snap.cards[0].title, "B"); - } - - #[tokio::test] - async fn replace_enforces_single_in_progress() { - let dir = tempdir().unwrap(); - let loc = thread_loc(dir.path(), "t1"); - let cards = vec![ - TaskBoardCard { - id: "a".into(), - title: "A".into(), - status: TaskCardStatus::InProgress, - objective: None, - plan: Vec::new(), - assigned_agent: None, - allowed_tools: Vec::new(), - approval_mode: None, - acceptance_criteria: Vec::new(), - evidence: Vec::new(), - notes: None, - blocker: None, - session_thread_id: None, - source_metadata: None, - order: 0, - updated_at: String::new(), - }, - TaskBoardCard { - id: "b".into(), - title: "B".into(), - status: TaskCardStatus::InProgress, - objective: None, - plan: Vec::new(), - assigned_agent: None, - allowed_tools: Vec::new(), - approval_mode: None, - acceptance_criteria: Vec::new(), - evidence: Vec::new(), - notes: None, - blocker: None, - session_thread_id: None, - source_metadata: None, - order: 1, - updated_at: String::new(), - }, - ]; - let err = replace(&loc, cards).await.unwrap_err(); - assert!(err.contains("in_progress")); - } - - #[tokio::test] - async fn clear_empties_the_list() { - let dir = tempdir().unwrap(); - let loc = thread_loc(dir.path(), "t1"); - let _ = add(&loc, "A", CardPatch::default()).await.unwrap(); - let snap = clear(&loc).await.unwrap(); - assert!(snap.cards.is_empty()); - assert!(snap.markdown.contains("No todos")); - } - - #[tokio::test] - async fn claim_card_transitions_todo_to_in_progress() { - let dir = tempdir().unwrap(); - let loc = thread_loc(dir.path(), "claim-1"); - let added = add(&loc, "claimable task", CardPatch::default()) - .await - .unwrap(); - let id = added.cards[0].id.clone(); - - let claimed = claim_card( - &loc, - &id, - &[TaskCardStatus::Todo, TaskCardStatus::Ready], - TaskCardStatus::InProgress, - ) - .await - .unwrap(); - assert_eq!(claimed.status, TaskCardStatus::InProgress); - assert_eq!(claimed.id, id); - - let snap = list(&loc).await.unwrap(); - assert_eq!(snap.cards[0].status, TaskCardStatus::InProgress); - } - - #[tokio::test] - async fn claim_card_rejects_when_status_does_not_match() { - let dir = tempdir().unwrap(); - let loc = thread_loc(dir.path(), "claim-2"); - let added = add(&loc, "done task", CardPatch::default()).await.unwrap(); - let id = added.cards[0].id.clone(); - update_status(&loc, &id, TaskCardStatus::Done) - .await - .unwrap(); - - let err = claim_card( - &loc, - &id, - &[TaskCardStatus::Todo, TaskCardStatus::Ready], - TaskCardStatus::InProgress, - ) - .await - .unwrap_err(); - assert!(err.contains("claim rejected"), "err: {err}"); - } - - #[tokio::test] - async fn claim_card_returns_not_found_for_missing_id() { - let dir = tempdir().unwrap(); - let loc = thread_loc(dir.path(), "claim-3"); - let err = claim_card( - &loc, - "task-nonexistent", - &[TaskCardStatus::Todo], - TaskCardStatus::InProgress, - ) - .await - .unwrap_err(); - assert!(err.contains("not found"), "err: {err}"); - } - - #[tokio::test(flavor = "multi_thread", worker_threads = 2)] - async fn concurrent_claims_only_one_wins() { - let dir = tempdir().unwrap(); - let loc = thread_loc(dir.path(), "race-1"); - let added = add(&loc, "race target", CardPatch::default()) - .await - .unwrap(); - let id = added.cards[0].id.clone(); - - // Two concurrent claimers gated on a shared barrier so both hit the - // per-board async CAS lock at once. Exactly one wins; the other observes - // the card already `InProgress` and is rejected. - let barrier = Arc::new(tokio::sync::Barrier::new(2)); - let mut handles = Vec::new(); - for _ in 0..2 { - let loc = loc.clone(); - let id = id.clone(); - let barrier = barrier.clone(); - handles.push(tokio::spawn(async move { - barrier.wait().await; - claim_card( - &loc, - &id, - &[TaskCardStatus::Todo, TaskCardStatus::Ready], - TaskCardStatus::InProgress, - ) - .await - })); - } - let mut results = Vec::new(); - for h in handles { - results.push(h.await.unwrap()); - } - - let wins = results.iter().filter(|r| r.is_ok()).count(); - let losses = results.iter().filter(|r| r.is_err()).count(); - assert_eq!(wins, 1, "exactly one claimer wins"); - assert_eq!(losses, 1, "exactly one claimer is rejected"); - - let snap = list(&loc).await.unwrap(); - assert_eq!(snap.cards[0].status, TaskCardStatus::InProgress); - } - - #[tokio::test] - async fn scratch_store_works_without_thread_context() { - let _guard = super::scratch_test_lock(); - global_scratch_store().replace(Vec::new()); - let loc = BoardLocation::Scratch; - let snap = add(&loc, "Scratch task", CardPatch::default()) - .await - .unwrap(); - assert_eq!(snap.cards.len(), 1); - assert!(snap.thread_id.is_none()); - let listed = list(&loc).await.unwrap(); - assert_eq!(listed.cards.len(), 1); - global_scratch_store().replace(Vec::new()); } } diff --git a/src/openhuman/todos/runs.rs b/src/openhuman/todos/runs.rs index 1d0f72f36..9db109586 100644 --- a/src/openhuman/todos/runs.rs +++ b/src/openhuman/todos/runs.rs @@ -398,7 +398,7 @@ pub async fn reclaim_stale( }; let patch = CardPatch { - status: Some(new_status.clone()), + status: Some(new_status), blocker: blocker_msg, ..Default::default() }; diff --git a/src/openhuman/todos/store.rs b/src/openhuman/todos/store.rs deleted file mode 100644 index 61d53ed08..000000000 --- a/src/openhuman/todos/store.rs +++ /dev/null @@ -1,44 +0,0 @@ -//! Process-global scratch store used as a fallback when no thread context -//! is present (e.g. agent-tool invocations outside a chat thread). The -//! authoritative per-thread state lives in -//! [`crate::openhuman::agent::task_board::TaskBoardStore`]. - -use crate::openhuman::agent::task_board::TaskBoardCard; -use once_cell::sync::OnceCell; -use parking_lot::Mutex; -use std::sync::Arc; - -/// Process-global scratch list of cards used when there is no current -/// thread id. Replaced wholesale by callers. -#[derive(Default)] -pub struct ScratchTodoStore { - inner: Mutex>, -} - -impl ScratchTodoStore { - pub fn new() -> Self { - Self::default() - } - - pub fn snapshot(&self) -> Vec { - self.inner.lock().clone() - } - - pub fn replace(&self, cards: Vec) { - *self.inner.lock() = cards; - } -} - -// NOTE: scratch CRUD calls in [`super::ops`] (load → mutate → save) are -// serialised by a coarser process-global mutex on the ops side, so the -// pair runs in one critical section even though [`snapshot`] and -// [`replace`] each take this inner lock independently. - -/// Process-global scratch store handle. The same `Arc` is returned across -/// calls so tool re-registration doesn't lose in-memory state. -pub fn global_scratch_store() -> Arc { - static STORE: OnceCell> = OnceCell::new(); - STORE - .get_or_init(|| Arc::new(ScratchTodoStore::new())) - .clone() -} diff --git a/vendor/tinyagents b/vendor/tinyagents index 4358efe61..b6a4120e8 160000 --- a/vendor/tinyagents +++ b/vendor/tinyagents @@ -1 +1 @@ -Subproject commit 4358efe618c65dd72d30948172f4faa403c7fb27 +Subproject commit b6a4120e822ef13c423d45b5b6b2bbc2a48a24cc