From 514d82241f5e7b7dfaf20883ac5c2cc47caab1ff Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 2 Jul 2026 10:12:20 -0700 Subject: [PATCH 01/10] graph::goals: add ThreadGoal data model + status/budget helpers Introduce the per-thread goal primitive under graph::goals: a single durable objective per thread with a small lifecycle (Active/Paused/BudgetLimited/ Complete), an optional token budget, and the pure GoalProgress accounting shape plus active_goal_context_block prompt renderer. Ported from OpenHuman's thread_goals, minus app-specific coupling. Claude-Session: https://claude.ai/code/session_01NRL2bim9Gz3UN3SRtpeXEX --- src/graph/goals/mod.rs | 22 ++++++ src/graph/goals/test.rs | 77 +++++++++++++++++++ src/graph/goals/types.rs | 161 +++++++++++++++++++++++++++++++++++++++ src/graph/mod.rs | 4 + 4 files changed, 264 insertions(+) create mode 100644 src/graph/goals/mod.rs create mode 100644 src/graph/goals/test.rs create mode 100644 src/graph/goals/types.rs diff --git a/src/graph/goals/mod.rs b/src/graph/goals/mod.rs new file mode 100644 index 0000000..1b208ed --- /dev/null +++ b/src/graph/goals/mod.rs @@ -0,0 +1,22 @@ +//! Per-thread durable **goal**: one objective per thread, carried across +//! supersteps, interrupts, and resumes. +//! +//! A thread goal is a single "completion contract" — a durable objective a +//! graph keeps pursuing until the model marks it [`Complete`], a token budget +//! is exhausted, or a host pauses it. This module owns the data model +//! ([`types`]), harness-[`Store`](crate::harness::store::Store)-backed +//! persistence ([`store`]), the model-facing controls exposed as harness tools +//! ([`tool`]), and the graph-native continuation surface ([`continuation`]). +//! +//! It is the graph analogue of OpenHuman's `thread_goals`, minus the +//! app-specific coupling (event bus, RPC envelopes, heartbeat scheduler): the +//! primitive is provider-neutral and drives off the graph runtime. + +mod types; + +pub use types::{ + GoalProgress, ThreadGoal, ThreadGoalStatus, TurnOutcome, active_goal_context_block, +}; + +#[cfg(test)] +mod test; diff --git a/src/graph/goals/test.rs b/src/graph/goals/test.rs new file mode 100644 index 0000000..91ccbcf --- /dev/null +++ b/src/graph/goals/test.rs @@ -0,0 +1,77 @@ +//! Unit tests for the thread-goal domain types. + +use super::types::*; + +fn goal(status: ThreadGoalStatus, token_budget: Option, tokens_used: u64) -> ThreadGoal { + ThreadGoal { + thread_id: "t".into(), + goal_id: "goal-0".into(), + objective: "ship the release".into(), + status, + token_budget, + tokens_used, + time_used_seconds: 0, + created_at_ms: 0, + updated_at_ms: 0, + continuation_suppressed: false, + } +} + +#[test] +fn status_strings_match_serialized() { + assert_eq!(ThreadGoalStatus::Active.as_str(), "active"); + assert_eq!(ThreadGoalStatus::Paused.as_str(), "paused"); + assert_eq!(ThreadGoalStatus::BudgetLimited.as_str(), "budget_limited"); + assert_eq!(ThreadGoalStatus::Complete.as_str(), "complete"); +} + +#[test] +fn active_and_terminal_predicates() { + assert!(ThreadGoalStatus::Active.is_active()); + assert!(!ThreadGoalStatus::Paused.is_active()); + assert!(ThreadGoalStatus::Complete.is_terminal()); + assert!(ThreadGoalStatus::BudgetLimited.is_terminal()); + assert!(!ThreadGoalStatus::Active.is_terminal()); + assert!(!ThreadGoalStatus::Paused.is_terminal()); +} + +#[test] +fn budget_helpers() { + let mut g = goal(ThreadGoalStatus::Active, Some(100), 40); + assert_eq!(g.budget_remaining(), Some(60)); + assert!(!g.over_budget()); + g.tokens_used = 120; + assert_eq!(g.budget_remaining(), Some(0)); + assert!(g.over_budget()); + g.token_budget = None; + assert_eq!(g.budget_remaining(), None); + assert!(!g.over_budget()); +} + +#[test] +fn context_block_present_only_for_active_and_budget_limited() { + let active = goal(ThreadGoalStatus::Active, Some(500), 100); + let block = active_goal_context_block(&active).expect("active goal has a context block"); + assert!(block.contains("ship the release")); + assert!(block.contains("goal_complete")); + assert!(block.contains("400"), "should name remaining budget"); + + let limited = goal(ThreadGoalStatus::BudgetLimited, Some(100), 100); + let block = active_goal_context_block(&limited).expect("budget-limited has a stop block"); + assert!(block.contains("budget")); + + assert!(active_goal_context_block(&goal(ThreadGoalStatus::Paused, None, 0)).is_none()); + assert!(active_goal_context_block(&goal(ThreadGoalStatus::Complete, None, 0)).is_none()); +} + +#[test] +fn thread_goal_round_trips_through_json() { + let g = goal(ThreadGoalStatus::Active, Some(1000), 250); + let json = serde_json::to_value(&g).unwrap(); + // camelCase field names on the wire. + assert_eq!(json["threadId"], "t"); + assert_eq!(json["goalId"], "goal-0"); + assert_eq!(json["tokenBudget"], 1000); + let back: ThreadGoal = serde_json::from_value(json).unwrap(); + assert_eq!(back, g); +} diff --git a/src/graph/goals/types.rs b/src/graph/goals/types.rs new file mode 100644 index 0000000..25daaef --- /dev/null +++ b/src/graph/goals/types.rs @@ -0,0 +1,161 @@ +//! Domain types for the thread-level goal. +//! +//! A **thread goal** is a single, thread-scoped "completion contract" — a +//! durable objective a graph keeps pursuing across supersteps, interrupts, +//! resumes, and budget boundaries. There is **exactly one** goal per thread, +//! with a small lifecycle and an optional token budget. +//! +//! The shape is ported from OpenHuman's `thread_goals`, re-hosted on the +//! harness [`Store`](crate::harness::store::Store) (see [`super::store`]) and +//! driven by the graph runtime rather than an out-of-band heartbeat (see +//! [`super::continuation`]). + +use serde::{Deserialize, Serialize}; + +/// Lifecycle state of a thread goal. +/// +/// Ownership is **asymmetric**: a model may create/replace a goal and mark it +/// [`Complete`](ThreadGoalStatus::Complete); [`Paused`](ThreadGoalStatus::Paused) +/// / [`BudgetLimited`](ThreadGoalStatus::BudgetLimited) are system-driven +/// (host control and accounting respectively), and clearing deletes the row +/// entirely rather than being a status. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ThreadGoalStatus { + /// The graph may make progress and (when driven) auto-continue. + Active, + /// Work is suspended (host/user control); the objective persists and is + /// reactivated on resume. + Paused, + /// The token budget has been reached; substantive work halts until the + /// budget is raised or the goal is cleared. + BudgetLimited, + /// Evidence confirms the objective is satisfied. + Complete, +} + +impl ThreadGoalStatus { + /// The stable lower-snake-case status label. + pub fn as_str(&self) -> &'static str { + match self { + Self::Active => "active", + Self::Paused => "paused", + Self::BudgetLimited => "budget_limited", + Self::Complete => "complete", + } + } + + /// Whether the goal is in a state where the graph should keep working it + /// (and a continuation loop may fire). + pub fn is_active(&self) -> bool { + matches!(self, Self::Active) + } + + /// Whether the goal is in a terminal state for continuation purposes — + /// [`Complete`](Self::Complete) or [`BudgetLimited`](Self::BudgetLimited) + /// never auto-continue. + pub fn is_terminal(&self) -> bool { + matches!(self, Self::Complete | Self::BudgetLimited) + } +} + +/// A single thread-scoped goal. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ThreadGoal { + /// The thread this goal belongs to (one goal per thread). + pub thread_id: String, + /// Version identifier, re-minted on **every objective replacement**. Stale + /// accounting writes that pass a non-matching `expected_goal_id` are + /// silently ignored — see [`super::store::account_usage`]. + pub goal_id: String, + /// The durable objective, one or more sentences. + pub objective: String, + /// Lifecycle state. + pub status: ThreadGoalStatus, + /// Optional token ceiling. When set and `tokens_used >= token_budget`, the + /// goal transitions to [`ThreadGoalStatus::BudgetLimited`]. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub token_budget: Option, + /// Cumulative tokens accounted against this goal. + #[serde(default)] + pub tokens_used: u64, + /// Cumulative wall-clock seconds accounted against this goal. + #[serde(default)] + pub time_used_seconds: u64, + /// Creation time (unix epoch milliseconds). + pub created_at_ms: u64, + /// Last-mutation time (unix epoch milliseconds). + pub updated_at_ms: u64, + /// Set when a continuation iteration produced **no progress**, to stop a + /// continuation loop. Cleared on any user-initiated run or external + /// mutation (e.g. a fresh `goal_set` or [`super::note_user_turn`]). + #[serde(default)] + pub continuation_suppressed: bool, +} + +impl ThreadGoal { + /// Tokens remaining before the budget cap, if a budget is set. + pub fn budget_remaining(&self) -> Option { + self.token_budget + .map(|b| b.saturating_sub(self.tokens_used)) + } + + /// Whether accounting has reached or exceeded the configured budget. + pub fn over_budget(&self) -> bool { + matches!(self.token_budget, Some(b) if self.tokens_used >= b) + } +} + +/// Usage accounted against a goal for one work iteration. +/// +/// The graph runtime is provider-neutral and does not meter tokens per node, so +/// accounting is **explicit** at the work-node boundary: a work node records +/// what it spent (and whether it advanced the objective) and the continuation +/// gate folds it in. `made_progress` is the graph analogue of OpenHuman's +/// "the turn produced tool calls" — the caller defines it (e.g. the iteration +/// produced tool calls or a non-empty state delta). +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct GoalProgress { + /// Tokens spent by the work iteration that just finished. + pub tokens_used: u64, + /// Wall-clock seconds spent by the work iteration that just finished. + pub elapsed_secs: u64, + /// Whether the iteration advanced the objective. A `false` here stops the + /// self-driving loop (one-shot suppression) so it cannot spin uselessly. + pub made_progress: bool, +} + +/// Outcome of running one continuation turn through the external-scheduler +/// driver ([`super::run_continuation_tick`]). Identical in shape to +/// [`GoalProgress`]; named distinctly at the driver boundary for clarity. +pub type TurnOutcome = GoalProgress; + +/// Renders the per-iteration context block a caller can prepend to a work +/// node's prompt so the model knows the active objective and how to close it +/// out. Returns `None` for statuses that should not drive further work +/// ([`Paused`](ThreadGoalStatus::Paused) / [`Complete`](ThreadGoalStatus::Complete)). +pub fn active_goal_context_block(goal: &ThreadGoal) -> Option { + match goal.status { + ThreadGoalStatus::Active => { + let budget = match goal.budget_remaining() { + Some(remaining) => format!(" (~{remaining} tokens of budget remain)"), + None => String::new(), + }; + Some(format!( + "[thread goal] You are working toward this thread's durable goal{budget}.\n\n\ + Goal: {objective}\n\n\ + Assess progress against concrete evidence, then take the next useful step. \ + If the goal is already satisfied, call `goal_complete` now.", + objective = goal.objective, + )) + } + ThreadGoalStatus::BudgetLimited => Some(format!( + "[thread goal] The token budget for this goal is exhausted. Summarise progress \ + and stop; do not start new substantive work.\n\nGoal: {}", + goal.objective, + )), + ThreadGoalStatus::Paused | ThreadGoalStatus::Complete => None, + } +} diff --git a/src/graph/mod.rs b/src/graph/mod.rs index abc1530..489fbf0 100644 --- a/src/graph/mod.rs +++ b/src/graph/mod.rs @@ -24,6 +24,7 @@ pub mod checkpoint; pub mod command; pub mod compiled; pub mod export; +pub mod goals; pub mod observability; pub mod orchestration; pub mod parallel; @@ -57,6 +58,9 @@ pub use export::{ NodePolicySummary, RouteInfo, ValidationReport, WaitingEdgeInfo, blueprint_to_json, blueprint_to_mermaid, blueprint_to_topology, from_json, to_json, to_mermaid, }; +pub use goals::{ + GoalProgress, ThreadGoal, ThreadGoalStatus, TurnOutcome, active_goal_context_block, +}; pub use observability::{ GraphEventJournal, GraphHealthSummary, GraphLangfuseExporter, GraphLatencyMetrics, GraphNodeHealth, GraphNodeLatency, GraphObservation, GraphStatusStore, GraphStepLatency, From 5377b910c319dbdaa8c6a04229aeb1ce2221c171 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 2 Jul 2026 10:14:44 -0700 Subject: [PATCH 02/10] graph::goals: persist goals on harness Store with per-thread RMW lock Add graph::goals::store: namespaced (graph.goals) Store-backed CRUD keyed by hex(thread_id), serialised per thread by an async mutex map since the Store trait offers no CAS. Preserves OpenHuman's goal_id compare-and-set stale-write guard (account_usage / set_continuation_suppressed_if) and the budget cap that flips an active goal to BudgetLimited. Uses next_seq() for goal ids and a local SystemTime clock (no uuid/chrono deps). Documents the single-process RMW caveat. Claude-Session: https://claude.ai/code/session_01NRL2bim9Gz3UN3SRtpeXEX --- src/graph/goals/mod.rs | 1 + src/graph/goals/store.rs | 338 +++++++++++++++++++++++++++++++++++++++ src/graph/goals/test.rs | 175 ++++++++++++++++++++ 3 files changed, 514 insertions(+) create mode 100644 src/graph/goals/store.rs diff --git a/src/graph/goals/mod.rs b/src/graph/goals/mod.rs index 1b208ed..3cad02d 100644 --- a/src/graph/goals/mod.rs +++ b/src/graph/goals/mod.rs @@ -12,6 +12,7 @@ //! app-specific coupling (event bus, RPC envelopes, heartbeat scheduler): the //! primitive is provider-neutral and drives off the graph runtime. +pub mod store; mod types; pub use types::{ diff --git a/src/graph/goals/store.rs b/src/graph/goals/store.rs new file mode 100644 index 0000000..82233fa --- /dev/null +++ b/src/graph/goals/store.rs @@ -0,0 +1,338 @@ +//! Persistence for the thread-level goal, on the harness +//! [`Store`](crate::harness::store::Store). +//! +//! Each thread's goal is a single serialized [`ThreadGoal`] value under the +//! [`GOALS_NAMESPACE`] namespace, keyed by the hex-encoded thread id. There is +//! at most one goal per thread. +//! +//! The [`Store`] trait offers no compare-and-set and no cross-key transaction, +//! so every mutation runs `load → mutate → put` under a **per-thread async +//! mutex** ([`thread_lock`]) — the process-local analogue of OpenHuman's +//! file-rename atomicity. Inside that lock the `goal_id` compare-and-set guard +//! (see [`account_usage`] / [`set_continuation_suppressed_if`]) still rejects +//! stale accounting from a replaced goal. +//! +//! # Single-process only +//! +//! `thread_lock` serializes writers **within one process**. Across multiple +//! processes sharing the same [`FileStore`](crate::harness::store::FileStore) +//! there is no atomic CAS, so two concurrent read-modify-writes can lose an +//! update. The `goal_id` guard still prevents *logical corruption* from stale +//! accounting, but not lost updates. For multi-writer deployments funnel goal +//! mutations through a single process (a clean fix would be a +//! `Store::compare_and_swap` extension, out of scope here). + +use std::collections::HashMap; +use std::sync::{Arc, Mutex as StdMutex, OnceLock}; +use std::time::{SystemTime, UNIX_EPOCH}; + +use tokio::sync::Mutex; + +use super::types::{ThreadGoal, ThreadGoalStatus}; +use crate::error::{Result, TinyAgentsError}; +use crate::harness::ids::next_seq; +use crate::harness::store::Store; + +/// The [`Store`] namespace holding one [`ThreadGoal`] per thread. +pub const GOALS_NAMESPACE: &str = "graph.goals"; + +/// Serialises `load → mutate → put` per thread so a read-modify-write is atomic +/// within the process. Returns the thread's dedicated async mutex, creating it +/// on first use. +fn thread_lock(thread_id: &str) -> Arc> { + static LOCKS: OnceLock>>>> = OnceLock::new(); + let map = LOCKS.get_or_init(|| StdMutex::new(HashMap::new())); + let mut guard = map.lock().expect("goal lock map poisoned"); + guard.entry(thread_id.to_string()).or_default().clone() +} + +/// Current unix time in milliseconds. Dependency-free (no `chrono`). +fn now_ms() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0) +} + +/// Hex-encodes the thread id into a [`Store`]-safe key. Required because +/// [`FileStore`](crate::harness::store::FileStore) rejects key bytes outside +/// `[A-Za-z0-9._-]`; hex is uniform across every backend. +fn key(thread_id: &str) -> String { + thread_id + .as_bytes() + .iter() + .map(|b| format!("{b:02x}")) + .collect() +} + +/// Mints a fresh, process-unique goal id (`goal-`). +fn new_goal_id() -> String { + format!("goal-{}", next_seq()) +} + +fn validate_thread_id(thread_id: &str) -> Result { + let trimmed = thread_id.trim(); + if trimmed.is_empty() { + return Err(TinyAgentsError::Validation( + "thread goal thread_id must not be empty or whitespace".to_string(), + )); + } + Ok(trimmed.to_string()) +} + +/// Reads the raw goal for `thread_id` (no validation, no lock). Returns `None` +/// when the thread has no goal. +async fn load(store: &Arc, thread_id: &str) -> Result> { + match store.get(GOALS_NAMESPACE, &key(thread_id)).await? { + Some(value) => Ok(Some(serde_json::from_value(value)?)), + None => Ok(None), + } +} + +/// Serializes and persists `goal`. +async fn save(store: &Arc, goal: &ThreadGoal) -> Result<()> { + let value = serde_json::to_value(goal)?; + store + .put(GOALS_NAMESPACE, &key(&goal.thread_id), value) + .await +} + +/// Load the goal for `thread_id` (read-only), or `None`. +pub async fn get(store: &Arc, thread_id: &str) -> Result> { + let thread_id = validate_thread_id(thread_id)?; + load(store, &thread_id).await +} + +/// Load every persisted thread goal (read-only). Skips values that fail to +/// deserialize so one corrupt entry can't hide the rest. +pub async fn list_all(store: &Arc) -> Result> { + let mut goals = Vec::new(); + for k in store.list(GOALS_NAMESPACE).await? { + if let Some(value) = store.get(GOALS_NAMESPACE, &k).await? + && let Ok(goal) = serde_json::from_value::(value) + { + goals.push(goal); + } + } + Ok(goals) +} + +/// Build + persist the goal for a `set`. The caller MUST hold the thread lock so +/// the read-modify-write is atomic. +/// +/// If `objective` **differs** from the current one a fresh `goal_id` is minted +/// and counters reset (status → `Active`). If the objective is **unchanged**, +/// counters and `goal_id` are preserved and only the budget / `updated_at` are +/// refreshed; the goal re-opens to `Active` unless it is still over budget. +async fn compute_and_put_set( + store: &Arc, + thread_id: &str, + objective: &str, + token_budget: Option, +) -> Result { + let now = now_ms(); + let goal = match load(store, thread_id).await? { + Some(mut existing) if existing.objective == objective => { + existing.token_budget = token_budget; + existing.continuation_suppressed = false; + existing.updated_at_ms = now; + existing.status = if existing.over_budget() { + ThreadGoalStatus::BudgetLimited + } else { + ThreadGoalStatus::Active + }; + existing + } + existing => { + let created_at_ms = existing.as_ref().map(|g| g.created_at_ms).unwrap_or(now); + ThreadGoal { + thread_id: thread_id.to_string(), + goal_id: new_goal_id(), + objective: objective.to_string(), + status: ThreadGoalStatus::Active, + token_budget, + tokens_used: 0, + time_used_seconds: 0, + created_at_ms, + updated_at_ms: now, + continuation_suppressed: false, + } + } + }; + save(store, &goal).await?; + Ok(goal) +} + +/// Create or replace the goal for `thread_id`. +pub async fn set( + store: &Arc, + thread_id: &str, + objective: &str, + token_budget: Option, +) -> Result { + let objective = objective.trim(); + if objective.is_empty() { + return Err(TinyAgentsError::Validation( + "thread goal objective must not be empty".to_string(), + )); + } + let thread_id = validate_thread_id(thread_id)?; + let lock = thread_lock(&thread_id); + let _guard = lock.lock().await; + compute_and_put_set(store, &thread_id, objective, token_budget).await +} + +/// Set the goal **only if the thread has none yet**. Returns `Some(goal)` when a +/// new goal was created, or `None` when a goal already existed (left untouched). +/// The check and the write run under one lock so a concurrent writer can't slip +/// into the gap. +pub async fn set_if_absent( + store: &Arc, + thread_id: &str, + objective: &str, + token_budget: Option, +) -> Result> { + let objective = objective.trim(); + if objective.is_empty() { + return Err(TinyAgentsError::Validation( + "thread goal objective must not be empty".to_string(), + )); + } + let thread_id = validate_thread_id(thread_id)?; + let lock = thread_lock(&thread_id); + let _guard = lock.lock().await; + if load(store, &thread_id).await?.is_some() { + return Ok(None); + } + Ok(Some( + compute_and_put_set(store, &thread_id, objective, token_budget).await?, + )) +} + +/// Delete the goal for `thread_id`. Returns whether one existed. +pub async fn clear(store: &Arc, thread_id: &str) -> Result { + let thread_id = validate_thread_id(thread_id)?; + let lock = thread_lock(&thread_id); + let _guard = lock.lock().await; + let existed = load(store, &thread_id).await?.is_some(); + store.delete(GOALS_NAMESPACE, &key(&thread_id)).await?; + Ok(existed) +} + +/// Generic guarded mutator: load, apply `f`, persist. Returns the updated goal, +/// or a [`Validation`](TinyAgentsError::Validation) error if the thread has none. +async fn mutate(store: &Arc, thread_id: &str, f: F) -> Result +where + F: FnOnce(&mut ThreadGoal), +{ + let thread_id = validate_thread_id(thread_id)?; + let lock = thread_lock(&thread_id); + let _guard = lock.lock().await; + let mut goal = load(store, &thread_id).await?.ok_or_else(|| { + TinyAgentsError::Validation(format!("no thread goal for thread '{thread_id}'")) + })?; + f(&mut goal); + goal.updated_at_ms = now_ms(); + save(store, &goal).await?; + Ok(goal) +} + +/// Mark the goal `Complete` (model-driven success). Suppresses further +/// continuation so a completed goal never re-drives. +pub async fn complete(store: &Arc, thread_id: &str) -> Result { + mutate(store, thread_id, |g| { + g.status = ThreadGoalStatus::Complete; + g.continuation_suppressed = true; + }) + .await +} + +/// Pause an `Active` goal (host-driven). A no-op for goals that aren't active. +pub async fn pause(store: &Arc, thread_id: &str) -> Result { + mutate(store, thread_id, |g| { + if g.status.is_active() { + g.status = ThreadGoalStatus::Paused; + } + }) + .await +} + +/// Resume a `Paused` goal (host-driven). A no-op for goals that aren't paused — +/// a completed/budget-limited goal is not reactivated. +pub async fn resume(store: &Arc, thread_id: &str) -> Result { + mutate(store, thread_id, |g| { + if matches!(g.status, ThreadGoalStatus::Paused) { + g.status = ThreadGoalStatus::Active; + g.continuation_suppressed = false; + } + }) + .await +} + +/// Set `continuation_suppressed` only when the thread's current goal still +/// matches `expected_goal_id` and is still active (compare-and-set). Returns the +/// goal as it stands after the (possibly skipped) write, or `None` when the +/// thread has no goal. +/// +/// The guard means a goal completed or replaced during a continuation iteration +/// is never suppressed by a late post-iteration write. +pub async fn set_continuation_suppressed_if( + store: &Arc, + thread_id: &str, + expected_goal_id: &str, + suppressed: bool, +) -> Result> { + let thread_id = validate_thread_id(thread_id)?; + let lock = thread_lock(&thread_id); + let _guard = lock.lock().await; + let Some(mut goal) = load(store, &thread_id).await? else { + return Ok(None); + }; + if goal.goal_id != expected_goal_id + || !goal.status.is_active() + || goal.continuation_suppressed == suppressed + { + return Ok(Some(goal)); + } + goal.continuation_suppressed = suppressed; + goal.updated_at_ms = now_ms(); + save(store, &goal).await?; + Ok(Some(goal)) +} + +/// Account token + time usage against the goal, applying the budget constraint. +/// +/// **Stale-write guard:** the delta is **silently ignored** when +/// `expected_goal_id` doesn't match the current goal — an in-flight accounting +/// call from a now-replaced goal must not corrupt the new one. An active goal +/// that crosses its budget becomes [`BudgetLimited`](ThreadGoalStatus::BudgetLimited). +/// Returns the goal as it stands after the (possibly skipped) update, or `None` +/// if there is no goal for the thread. +pub async fn account_usage( + store: &Arc, + thread_id: &str, + expected_goal_id: &str, + token_delta: u64, + secs_delta: u64, +) -> Result> { + let thread_id = validate_thread_id(thread_id)?; + let lock = thread_lock(&thread_id); + let _guard = lock.lock().await; + let Some(mut goal) = load(store, &thread_id).await? else { + return Ok(None); + }; + if goal.goal_id != expected_goal_id { + return Ok(Some(goal)); + } + if token_delta == 0 && secs_delta == 0 { + return Ok(Some(goal)); + } + goal.tokens_used = goal.tokens_used.saturating_add(token_delta); + goal.time_used_seconds = goal.time_used_seconds.saturating_add(secs_delta); + if goal.status.is_active() && goal.over_budget() { + goal.status = ThreadGoalStatus::BudgetLimited; + } + goal.updated_at_ms = now_ms(); + save(store, &goal).await?; + Ok(Some(goal)) +} diff --git a/src/graph/goals/test.rs b/src/graph/goals/test.rs index 91ccbcf..6ad4db7 100644 --- a/src/graph/goals/test.rs +++ b/src/graph/goals/test.rs @@ -75,3 +75,178 @@ fn thread_goal_round_trips_through_json() { let back: ThreadGoal = serde_json::from_value(json).unwrap(); assert_eq!(back, g); } + +mod store_tests { + use std::sync::Arc; + + use super::super::store; + use super::ThreadGoalStatus; + use crate::harness::store::{InMemoryStore, Store}; + + fn store() -> Arc { + Arc::new(InMemoryStore::default()) + } + + #[tokio::test] + async fn set_get_clear_round_trip() { + let s = store(); + assert!(store::get(&s, "t1").await.unwrap().is_none()); + + let g = store::set(&s, "t1", "ship the feature", None) + .await + .unwrap(); + assert_eq!(g.objective, "ship the feature"); + assert_eq!(g.status, ThreadGoalStatus::Active); + assert_eq!(g.tokens_used, 0); + + let loaded = store::get(&s, "t1").await.unwrap().unwrap(); + assert_eq!(loaded.goal_id, g.goal_id); + + assert!(store::clear(&s, "t1").await.unwrap()); + assert!(store::get(&s, "t1").await.unwrap().is_none()); + assert!(!store::clear(&s, "t1").await.unwrap()); + } + + #[tokio::test] + async fn set_same_objective_preserves_goal_id_and_counters() { + let s = store(); + let g1 = store::set(&s, "t", "objective A", Some(100)).await.unwrap(); + store::account_usage(&s, "t", &g1.goal_id, 30, 5) + .await + .unwrap() + .unwrap(); + let g2 = store::set(&s, "t", "objective A", Some(200)).await.unwrap(); + assert_eq!(g1.goal_id, g2.goal_id, "same objective keeps goal_id"); + assert_eq!(g2.tokens_used, 30, "counters preserved"); + assert_eq!(g2.token_budget, Some(200), "budget refreshed"); + } + + #[tokio::test] + async fn set_same_objective_stays_budget_limited_when_over_budget() { + let s = store(); + let g = store::set(&s, "t", "obj", Some(100)).await.unwrap(); + let limited = store::account_usage(&s, "t", &g.goal_id, 120, 0) + .await + .unwrap() + .unwrap(); + assert_eq!(limited.status, ThreadGoalStatus::BudgetLimited); + let resed = store::set(&s, "t", "obj", Some(100)).await.unwrap(); + assert_eq!(resed.tokens_used, 120, "same-objective preserves counters"); + assert_eq!( + resed.status, + ThreadGoalStatus::BudgetLimited, + "still over budget → must stay budget_limited, not active" + ); + let raised = store::set(&s, "t", "obj", Some(1000)).await.unwrap(); + assert_eq!(raised.status, ThreadGoalStatus::Active); + } + + #[tokio::test] + async fn set_changed_objective_mints_new_goal_id_and_resets() { + let s = store(); + let g1 = store::set(&s, "t", "objective A", None).await.unwrap(); + store::account_usage(&s, "t", &g1.goal_id, 30, 5) + .await + .unwrap() + .unwrap(); + let g2 = store::set(&s, "t", "objective B", None).await.unwrap(); + assert_ne!(g1.goal_id, g2.goal_id); + assert_eq!(g2.tokens_used, 0, "counters reset on new objective"); + assert_eq!(g2.created_at_ms, g1.created_at_ms, "created_at preserved"); + } + + #[tokio::test] + async fn set_if_absent_only_bootstraps_when_empty() { + let s = store(); + let created = store::set_if_absent(&s, "t", "scout goal", None) + .await + .unwrap(); + assert!(created.is_some()); + let again = store::set_if_absent(&s, "t", "different goal", None) + .await + .unwrap(); + assert!(again.is_none()); + assert_eq!( + store::get(&s, "t").await.unwrap().unwrap().objective, + "scout goal" + ); + } + + #[tokio::test] + async fn account_usage_ignores_stale_goal_id() { + let s = store(); + let g = store::set(&s, "t", "obj", None).await.unwrap(); + let after = store::account_usage(&s, "t", "not-the-goal-id", 50, 1) + .await + .unwrap() + .unwrap(); + assert_eq!(after.tokens_used, 0); + let after = store::account_usage(&s, "t", &g.goal_id, 50, 1) + .await + .unwrap() + .unwrap(); + assert_eq!(after.tokens_used, 50); + } + + #[tokio::test] + async fn account_usage_trips_budget_limited() { + let s = store(); + let g = store::set(&s, "t", "obj", Some(100)).await.unwrap(); + let after = store::account_usage(&s, "t", &g.goal_id, 120, 2) + .await + .unwrap() + .unwrap(); + assert_eq!(after.status, ThreadGoalStatus::BudgetLimited); + assert_eq!(after.budget_remaining(), Some(0)); + } + + #[tokio::test] + async fn pause_resume_complete_transitions() { + let s = store(); + store::set(&s, "t", "obj", None).await.unwrap(); + assert_eq!( + store::pause(&s, "t").await.unwrap().status, + ThreadGoalStatus::Paused + ); + assert_eq!( + store::resume(&s, "t").await.unwrap().status, + ThreadGoalStatus::Active + ); + let done = store::complete(&s, "t").await.unwrap(); + assert_eq!(done.status, ThreadGoalStatus::Complete); + // Resume does not reactivate a completed goal. + assert_eq!( + store::resume(&s, "t").await.unwrap().status, + ThreadGoalStatus::Complete + ); + } + + #[tokio::test] + async fn mutators_error_without_a_goal() { + let s = store(); + assert!(store::complete(&s, "missing").await.is_err()); + assert!(store::pause(&s, "missing").await.is_err()); + } + + #[tokio::test] + async fn empty_objective_and_blank_thread_id_rejected() { + let s = store(); + assert!(store::set(&s, "t", " ", None).await.is_err()); + assert!(store::set(&s, " ", "obj", None).await.is_err()); + } + + #[tokio::test] + async fn list_all_returns_every_thread_goal() { + let s = store(); + store::set(&s, "alpha", "a", None).await.unwrap(); + store::set(&s, "beta", "b", None).await.unwrap(); + let mut ids: Vec = store::list_all(&s) + .await + .unwrap() + .into_iter() + .map(|g| g.thread_id) + .collect(); + ids.sort(); + assert_eq!(ids, vec!["alpha".to_string(), "beta".to_string()]); + } +} From 982bc46c9f1aaa390d586bfcc0e2b671ed52959e Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 2 Jul 2026 10:16:48 -0700 Subject: [PATCH 03/10] graph::goals: expose goal_get/goal_set/goal_complete as harness tools Add graph::goals::tool: GoalTool dispatching on GoalToolKind, mirroring OrchestrationTool. The default model-facing set is goal_get/goal_set/ goal_complete (asymmetric ownership); pause/resume/clear are constructible host controls. thread_id is resolved from ToolExecutionContext (never a tool arg), so a model cannot address another thread's goal; the bare call() entry point errors without an active thread. Store is captured as Arc at construction. Claude-Session: https://claude.ai/code/session_01NRL2bim9Gz3UN3SRtpeXEX --- src/graph/goals/mod.rs | 2 + src/graph/goals/test.rs | 139 +++++++++++++++++++ src/graph/goals/tool.rs | 298 ++++++++++++++++++++++++++++++++++++++++ 3 files changed, 439 insertions(+) create mode 100644 src/graph/goals/tool.rs diff --git a/src/graph/goals/mod.rs b/src/graph/goals/mod.rs index 3cad02d..e00f32f 100644 --- a/src/graph/goals/mod.rs +++ b/src/graph/goals/mod.rs @@ -13,8 +13,10 @@ //! primitive is provider-neutral and drives off the graph runtime. pub mod store; +mod tool; mod types; +pub use tool::{GoalTool, GoalToolKind, goal_tools, register_goal_tools}; pub use types::{ GoalProgress, ThreadGoal, ThreadGoalStatus, TurnOutcome, active_goal_context_block, }; diff --git a/src/graph/goals/test.rs b/src/graph/goals/test.rs index 6ad4db7..b9c8f7f 100644 --- a/src/graph/goals/test.rs +++ b/src/graph/goals/test.rs @@ -250,3 +250,142 @@ mod store_tests { assert_eq!(ids, vec!["alpha".to_string(), "beta".to_string()]); } } + +mod tool_tests { + use std::sync::Arc; + + use serde_json::json; + + use super::super::tool::{GoalTool, GoalToolKind, goal_tools}; + use crate::harness::events::EventSink; + use crate::harness::ids::{RunId, ThreadId}; + use crate::harness::store::{InMemoryStore, Store}; + use crate::harness::tool::{Tool, ToolCall, ToolExecutionContext}; + + fn store() -> Arc { + Arc::new(InMemoryStore::default()) + } + + fn ctx(thread_id: Option<&str>) -> ToolExecutionContext { + ToolExecutionContext { + run_id: RunId::new("run-1"), + thread_id: thread_id.map(ThreadId::new), + depth: 0, + max_turn_output_tokens: None, + events: EventSink::new(), + workspace: None, + } + } + + fn call(id: &str, name: &str, args: serde_json::Value) -> ToolCall { + ToolCall { + id: id.to_string(), + name: name.to_string(), + arguments: args, + } + } + + #[test] + fn goal_tools_builds_the_model_facing_set() { + let tools = goal_tools(store()); + let names: Vec<&str> = tools.iter().map(|t| Tool::<()>::name(t.as_ref())).collect(); + assert_eq!(names, vec!["goal_get", "goal_set", "goal_complete"]); + } + + #[tokio::test] + async fn set_get_complete_via_tools_in_thread_scope() { + let s = store(); + let set = GoalTool::new(GoalToolKind::Set, s.clone()); + let res = Tool::<()>::call_with_context( + &set, + &(), + call( + "c1", + "goal_set", + json!({ "objective": "land the PR", "token_budget": 5000 }), + ), + ctx(Some("thread-tools")), + ) + .await + .unwrap(); + assert!(res.error.is_none(), "{res:?}"); + assert!(res.content.contains("land the PR")); + + let get = GoalTool::new(GoalToolKind::Get, s.clone()); + let res = Tool::<()>::call_with_context( + &get, + &(), + call("c2", "goal_get", json!({})), + ctx(Some("thread-tools")), + ) + .await + .unwrap(); + assert!(res.content.contains("status: active")); + + let done = GoalTool::new(GoalToolKind::Complete, s.clone()); + let res = Tool::<()>::call_with_context( + &done, + &(), + call("c3", "goal_complete", json!({})), + ctx(Some("thread-tools")), + ) + .await + .unwrap(); + assert!(res.content.contains("status: complete")); + } + + #[tokio::test] + async fn tools_error_without_thread_scope() { + let s = store(); + let set = GoalTool::new(GoalToolKind::Set, s); + // Bare call (no context) errors. + let res = Tool::<()>::call( + &set, + &(), + call("c1", "goal_set", json!({ "objective": "x" })), + ) + .await + .unwrap(); + assert!(res.error.is_some()); + assert!(res.error.unwrap().contains("active thread")); + // Context with no thread id also errors. + let set = GoalTool::new(GoalToolKind::Set, store()); + let res = Tool::<()>::call_with_context( + &set, + &(), + call("c2", "goal_set", json!({ "objective": "x" })), + ctx(None), + ) + .await + .unwrap(); + assert!(res.error.is_some()); + } + + #[tokio::test] + async fn get_reports_absent_goal() { + let get = GoalTool::new(GoalToolKind::Get, store()); + let res = Tool::<()>::call_with_context( + &get, + &(), + call("c1", "goal_get", json!({})), + ctx(Some("empty-thread")), + ) + .await + .unwrap(); + assert!(res.content.contains("no goal set")); + } + + #[tokio::test] + async fn set_missing_objective_is_a_soft_error() { + let set = GoalTool::new(GoalToolKind::Set, store()); + let res = Tool::<()>::call_with_context( + &set, + &(), + call("c1", "goal_set", json!({})), + ctx(Some("t")), + ) + .await + .unwrap(); + assert!(res.content.contains("missing 'objective'")); + } +} diff --git a/src/graph/goals/tool.rs b/src/graph/goals/tool.rs new file mode 100644 index 0000000..8030df0 --- /dev/null +++ b/src/graph/goals/tool.rs @@ -0,0 +1,298 @@ +//! Model-facing controls for the thread goal, exposed as ordinary harness +//! [`Tool`]s. +//! +//! Ownership is **asymmetric**: a model may read the goal (`goal_get`), create +//! or replace it (`goal_set`), and mark it done (`goal_complete`). Pause / +//! resume / clear are host-driven controls — constructible as tools for a host +//! that wants to expose them, but not part of the default model-facing set +//! returned by [`goal_tools`]. +//! +//! The target thread is resolved from +//! [`ToolExecutionContext::thread_id`](crate::harness::tool::ToolExecutionContext), +//! the harness analogue of an ambient thread id: a tool never takes a +//! `thread_id` argument, so a model can't address another thread's goal. The +//! bare [`Tool::call`] entry point (no context) errors, matching the "tools +//! require an active thread" contract. + +use std::sync::Arc; + +use async_trait::async_trait; +use serde_json::{Value, json}; + +use super::store; +use super::types::ThreadGoal; +use crate::error::Result; +use crate::harness::store::Store; +use crate::harness::tool::{ + Tool, ToolCall, ToolExecutionContext, ToolPolicy, ToolRegistry, ToolResult, ToolSchema, + ToolSideEffects, +}; + +/// Which thread-goal control a [`GoalTool`] implements. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] +pub enum GoalToolKind { + /// Read the current thread goal. + Get, + /// Create or replace the current thread goal. + Set, + /// Mark the current thread goal complete. + Complete, + /// Pause the current thread goal (host control). + Pause, + /// Resume a paused thread goal (host control). + Resume, + /// Delete the current thread goal (host control). + Clear, +} + +impl GoalToolKind { + /// The default model-facing controls (asymmetric ownership). + pub const MODEL_FACING: [Self; 3] = [Self::Get, Self::Set, Self::Complete]; + + /// Every control, including the host-driven pause/resume/clear. + pub const ALL: [Self; 6] = [ + Self::Get, + Self::Set, + Self::Complete, + Self::Pause, + Self::Resume, + Self::Clear, + ]; + + /// Stable model-visible tool name. + pub fn name(self) -> &'static str { + match self { + Self::Get => "goal_get", + Self::Set => "goal_set", + Self::Complete => "goal_complete", + Self::Pause => "goal_pause", + Self::Resume => "goal_resume", + Self::Clear => "goal_clear", + } + } + + /// Short model-visible description. + pub fn description(self) -> &'static str { + match self { + Self::Get => { + "Read this thread's goal — the durable objective you're pursuing across \ + turns — with its status (active/paused/budget_limited/complete) and token \ + usage. Returns 'no goal set' when the thread has none." + } + Self::Set => { + "Set (or replace) this thread's goal — the durable objective you should keep \ + pursuing across turns until it's complete. Changing the objective resets \ + usage counters. Optionally set a token_budget; when reached, work halts." + } + Self::Complete => { + "Mark this thread's goal complete. Only call this when concrete evidence \ + confirms the objective is satisfied — completing stops autonomous \ + continuation." + } + Self::Pause => "Pause this thread's active goal (host control).", + Self::Resume => "Resume this thread's paused goal (host control).", + Self::Clear => "Delete this thread's goal (host control).", + } + } + + /// Whether the control only reads state. + fn read_only(self) -> bool { + matches!(self, Self::Get) + } + + /// The model-visible JSON-schema parameters for the control. + fn parameters(self) -> Value { + match self { + Self::Set => json!({ + "type": "object", + "required": ["objective"], + "properties": { + "objective": { + "type": "string", + "description": "The durable objective — what 'done' looks like for this thread." + }, + "token_budget": { + "type": "integer", + "minimum": 1, + "description": "Optional token ceiling for the goal. Omit for no limit." + } + } + }), + _ => json!({ "type": "object", "properties": {} }), + } + } +} + +/// A harness [`Tool`] for one thread-goal control, backed by a +/// [`Store`](crate::harness::store::Store). +pub struct GoalTool { + kind: GoalToolKind, + store: Arc, +} + +impl GoalTool { + /// Creates one goal tool of `kind` backed by `store`. + pub fn new(kind: GoalToolKind, store: Arc) -> Self { + Self { kind, store } + } + + /// The control kind this tool implements. + pub fn kind(&self) -> GoalToolKind { + self.kind + } + + /// Dispatches the control against `thread_id`, returning the model-facing + /// content and a structured `raw` payload. + async fn dispatch(&self, thread_id: &str, args: &Value) -> Result<(String, Option)> { + match self.kind { + GoalToolKind::Get => match store::get(&self.store, thread_id).await? { + Some(goal) => Ok((render_goal(&goal), Some(serde_json::to_value(&goal)?))), + None => Ok(("no goal set for this thread".to_string(), None)), + }, + GoalToolKind::Set => { + let Some(objective) = args.get("objective").and_then(Value::as_str) else { + return Ok(("error: missing 'objective' parameter".to_string(), None)); + }; + let token_budget = args.get("token_budget").and_then(Value::as_u64); + let goal = store::set(&self.store, thread_id, objective, token_budget).await?; + Ok(( + format!("Goal set.\n{}", render_goal(&goal)), + Some(serde_json::to_value(&goal)?), + )) + } + GoalToolKind::Complete => { + let goal = store::complete(&self.store, thread_id).await?; + Ok(( + format!("Goal marked complete.\n{}", render_goal(&goal)), + Some(serde_json::to_value(&goal)?), + )) + } + GoalToolKind::Pause => { + let goal = store::pause(&self.store, thread_id).await?; + Ok((render_goal(&goal), Some(serde_json::to_value(&goal)?))) + } + GoalToolKind::Resume => { + let goal = store::resume(&self.store, thread_id).await?; + Ok((render_goal(&goal), Some(serde_json::to_value(&goal)?))) + } + GoalToolKind::Clear => { + let removed = store::clear(&self.store, thread_id).await?; + Ok(( + format!("Goal cleared (removed={removed})."), + Some(json!({ "removed": removed })), + )) + } + } + } +} + +/// Renders a goal as a compact, model-readable block. +fn render_goal(goal: &ThreadGoal) -> String { + let budget = match goal.token_budget { + Some(b) => format!( + "{} used / {b} budget ({} left)", + goal.tokens_used, + goal.budget_remaining().unwrap_or(0) + ), + None => format!("{} used / no budget", goal.tokens_used), + }; + format!( + "[thread_goal]\nstatus: {}\nobjective: {}\ntokens: {budget}\n[/thread_goal]", + goal.status.as_str(), + goal.objective + ) +} + +fn error_result(call_id: String, name: &str, message: impl Into) -> ToolResult { + ToolResult { + call_id, + name: name.to_string(), + content: String::new(), + raw: None, + error: Some(message.into()), + elapsed_ms: 0, + } +} + +/// Builds the default **model-facing** goal controls (`goal_get`, `goal_set`, +/// `goal_complete`). +pub fn goal_tools(store: Arc) -> Vec> { + GoalToolKind::MODEL_FACING + .into_iter() + .map(|kind| Arc::new(GoalTool::new(kind, store.clone()))) + .collect() +} + +/// Registers the default model-facing goal controls into a tool registry. +pub fn register_goal_tools( + registry: &mut ToolRegistry, + store: Arc, +) -> &mut ToolRegistry { + for tool in goal_tools(store) { + registry.register(tool); + } + registry +} + +#[async_trait] +impl Tool for GoalTool { + fn name(&self) -> &str { + self.kind.name() + } + + fn description(&self) -> &str { + self.kind.description() + } + + fn schema(&self) -> ToolSchema { + ToolSchema { + name: self.kind.name().to_string(), + description: self.kind.description().to_string(), + parameters: self.kind.parameters(), + format: Default::default(), + } + } + + fn policy(&self) -> ToolPolicy { + ToolPolicy { + classified: true, + side_effects: ToolSideEffects { + read_only: self.kind.read_only(), + ..Default::default() + }, + ..Default::default() + } + } + + async fn call(&self, _state: &State, call: ToolCall) -> Result { + Ok(error_result( + call.id, + self.kind.name(), + "goal tools require an active thread (no thread_id in tool context)", + )) + } + + async fn call_with_context( + &self, + _state: &State, + call: ToolCall, + context: ToolExecutionContext, + ) -> Result { + let Some(thread_id) = context.thread_id.as_ref() else { + return Ok(error_result( + call.id, + self.kind.name(), + "goal tools require an active thread (no thread_id in tool context)", + )); + }; + let (content, raw) = self.dispatch(thread_id.as_str(), &call.arguments).await?; + Ok(ToolResult { + call_id: call.id, + name: self.kind.name().to_string(), + content, + raw, + error: None, + elapsed_ms: 0, + }) + } +} From 606ee87b998061d30dd7d0e194b0ab617e46fc08 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 2 Jul 2026 10:22:14 -0700 Subject: [PATCH 04/10] graph::goals: add graph-native continuation (goal_gate_node + driver) Re-express OpenHuman's heartbeat auto-continuation on the graph runtime: - goal_gate_node: a command-routing node forming a self-driving bounded loop. After each work iteration it folds the iteration's GoalProgress via account_usage and routes back to the work node while the goal is Active and under budget, else to END. recursion_limit is the hard backstop; a zero- progress iteration sets the one-shot suppression and stops. - run_continuation_tick: a faithful port of the heartbeat for callers with an external scheduler, generic over a run-turn closure (idle filter, oldest-first, max_per_tick, one-shot suppression). - note_user_turn: clears one-shot suppression and reactivates a paused goal on a user-initiated run. Token accounting is explicit at the work-node boundary since the graph runtime does not meter tokens per node. Tests build real graphs on InMemoryStore. Claude-Session: https://claude.ai/code/session_01NRL2bim9Gz3UN3SRtpeXEX --- src/graph/goals/continuation.rs | 189 +++++++++++++++++++++++++ src/graph/goals/mod.rs | 2 + src/graph/goals/store.rs | 2 +- src/graph/goals/test.rs | 239 ++++++++++++++++++++++++++++++++ src/graph/mod.rs | 4 +- 5 files changed, 434 insertions(+), 2 deletions(-) create mode 100644 src/graph/goals/continuation.rs diff --git a/src/graph/goals/continuation.rs b/src/graph/goals/continuation.rs new file mode 100644 index 0000000..417a08b --- /dev/null +++ b/src/graph/goals/continuation.rs @@ -0,0 +1,189 @@ +//! Graph-native continuation for the thread goal. +//! +//! OpenHuman drives a goal from an out-of-band heartbeat: when a thread carries +//! an active goal and goes idle, the heartbeat injects one continuation turn, +//! marks the goal one-shot `continuation_suppressed`, and a later user turn +//! clears it. TinyAgents has no heartbeat and no ambient agent, so the same +//! behaviour is re-expressed on the graph runtime three ways: +//! +//! - [`goal_gate_node`] (**primary**) — a self-driving bounded loop. A +//! command-routing node that, after each work iteration, folds the iteration's +//! usage into the goal and routes back to the work node while the goal is +//! Active and under budget, else routes to [`END`]. The graph's +//! `recursion_limit` is the hard backstop. +//! - [`run_continuation_tick`] (**driver**) — a faithful port of the heartbeat +//! for callers that *do* have an external scheduler: it selects idle active +//! goals and runs one turn each through a caller-supplied closure. +//! - [`note_user_turn`] — call at the start of a user-initiated run to clear the +//! one-shot suppression (and reactivate a paused goal), the analogue of +//! OpenHuman's post-turn `account_turn_against_goal` suppression reset. +//! +//! # Token accounting boundary +//! +//! The graph runtime is provider-neutral and does not meter tokens per node. +//! Accounting is therefore **explicit**: the work node records what it spent +//! into `State` and the caller's `progress`/`run_turn` closure reports it as a +//! [`GoalProgress`]/[`TurnOutcome`]. `made_progress == false` stops the loop +//! (one-shot suppression) exactly as "the turn produced no tool calls" does in +//! OpenHuman. + +use std::sync::Arc; +use std::time::Duration; + +use super::store; +use super::types::{GoalProgress, ThreadGoal, ThreadGoalStatus, TurnOutcome}; +use crate::error::Result; +use crate::graph::builder::{END, NodeContext, NodeFuture}; +use crate::graph::command::{Command, NodeResult}; +use crate::harness::store::Store; + +/// Builds a **goal gate** node handler: a self-driving bounded loop that keeps +/// re-running `work_node` while the thread's goal is Active and under budget. +/// +/// Wire it as `work_node -> gate` (a static edge) and register `gate` as a +/// command node whose destinations are `[work_node, END]` +/// (`with_command_destinations`). On each activation the gate: +/// +/// 1. reads the thread id from the [`NodeContext`] (absent → routes to [`END`]); +/// 2. loads the goal (`None` → routes to [`END`]); +/// 3. folds the just-finished iteration's [`GoalProgress`] via +/// [`store::account_usage`], which flips an over-budget goal to +/// [`BudgetLimited`](ThreadGoalStatus::BudgetLimited); +/// 4. routes a single `goto`: +/// - not Active, or `continuation_suppressed` → [`END`]; +/// - the iteration made no progress → set one-shot suppression, then [`END`]; +/// - otherwise → back to `work_node` (loop). +/// +/// The handler only routes (it never updates state), so `Update` carries no +/// bound beyond `Send`. Pass the returned closure straight to +/// [`GraphBuilder::add_node`](crate::graph::GraphBuilder::add_node). +pub fn goal_gate_node( + store: Arc, + work_node: impl Into, + progress: impl Fn(&State) -> GoalProgress + Send + Sync + 'static, +) -> impl Fn(State, NodeContext) -> NodeFuture + Send + Sync + 'static +where + State: Send + 'static, + Update: Send + 'static, +{ + let work_node = work_node.into(); + let progress = Arc::new(progress); + move |state, ctx| { + let store = store.clone(); + let work_node = work_node.clone(); + let progress = progress.clone(); + Box::pin(async move { + let Some(thread_id) = ctx.thread_id.as_ref().map(|t| t.as_str().to_string()) else { + return Ok(NodeResult::Command(Command::goto([END]))); + }; + let Some(goal) = store::get(&store, &thread_id).await? else { + return Ok(NodeResult::Command(Command::goto([END]))); + }; + + let p = progress(&state); + let goal = match store::account_usage( + &store, + &thread_id, + &goal.goal_id, + p.tokens_used, + p.elapsed_secs, + ) + .await? + { + Some(g) => g, + None => return Ok(NodeResult::Command(Command::goto([END]))), + }; + + if !goal.status.is_active() || goal.continuation_suppressed { + return Ok(NodeResult::Command(Command::goto([END]))); + } + if !p.made_progress { + // One-shot: a zero-progress iteration stops the loop until a + // user-initiated run clears the flag (see `note_user_turn`). + store::set_continuation_suppressed_if(&store, &thread_id, &goal.goal_id, true) + .await?; + return Ok(NodeResult::Command(Command::goto([END]))); + } + Ok(NodeResult::Command(Command::goto([work_node]))) + }) + } +} + +/// One continuation pass for callers with an external scheduler: run at most +/// `max_per_tick` idle active goals, oldest-idle first, one turn each. +/// +/// A goal is a candidate when it is Active, not already suppressed, and has been +/// idle (no `updated_at` change) for at least `idle`. For each selected goal the +/// closure `run_turn` runs one turn; its [`TurnOutcome`] is folded via +/// [`store::account_usage`], and a no-progress turn sets the one-shot +/// suppression. Returns the number of turns actually run. +/// +/// A `run_turn` that errors is skipped (best-effort — a failed turn must not +/// stop the whole tick); [`Store`] errors from accounting propagate. This is a +/// single sequential async fn, so it already serializes; concurrent scheduling +/// is the caller's responsibility. +pub async fn run_continuation_tick( + store: &Arc, + idle: Duration, + max_per_tick: usize, + run_turn: F, +) -> Result +where + F: Fn(ThreadGoal) -> Fut, + Fut: Future>, +{ + let now = store::now_ms(); + let idle_ms = idle.as_millis() as u64; + let mut candidates: Vec = store::list_all(store) + .await? + .into_iter() + .filter(|g| g.status.is_active()) + .filter(|g| !g.continuation_suppressed) + .filter(|g| now.saturating_sub(g.updated_at_ms) >= idle_ms) + .collect(); + // Oldest-idle first so the most-neglected goal advances under the cap. + candidates.sort_by_key(|g| g.updated_at_ms); + candidates.truncate(max_per_tick); + + let mut ran = 0; + for goal in candidates { + let outcome = match run_turn(goal.clone()).await { + Ok(o) => o, + Err(_) => continue, + }; + store::account_usage( + store, + &goal.thread_id, + &goal.goal_id, + outcome.tokens_used, + outcome.elapsed_secs, + ) + .await?; + if !outcome.made_progress { + store::set_continuation_suppressed_if(store, &goal.thread_id, &goal.goal_id, true) + .await?; + } + ran += 1; + } + Ok(ran) +} + +/// Notes the start of a **user-initiated** run for `thread_id`: clears the +/// one-shot continuation suppression on an active goal and reactivates a paused +/// one. Returns the goal as it stands afterward, or `None` when the thread has +/// none. +/// +/// This is what distinguishes a user turn from a self-driving loop iteration — +/// the loop never calls it, so it can never clear its own suppression. +pub async fn note_user_turn(store: &Arc, thread_id: &str) -> Result> { + let Some(goal) = store::get(store, thread_id).await? else { + return Ok(None); + }; + match goal.status { + ThreadGoalStatus::Paused => store::resume(store, thread_id).await.map(Some), + ThreadGoalStatus::Active if goal.continuation_suppressed => { + store::set_continuation_suppressed_if(store, thread_id, &goal.goal_id, false).await + } + _ => Ok(Some(goal)), + } +} diff --git a/src/graph/goals/mod.rs b/src/graph/goals/mod.rs index e00f32f..b614879 100644 --- a/src/graph/goals/mod.rs +++ b/src/graph/goals/mod.rs @@ -12,10 +12,12 @@ //! app-specific coupling (event bus, RPC envelopes, heartbeat scheduler): the //! primitive is provider-neutral and drives off the graph runtime. +mod continuation; pub mod store; mod tool; mod types; +pub use continuation::{goal_gate_node, note_user_turn, run_continuation_tick}; pub use tool::{GoalTool, GoalToolKind, goal_tools, register_goal_tools}; pub use types::{ GoalProgress, ThreadGoal, ThreadGoalStatus, TurnOutcome, active_goal_context_block, diff --git a/src/graph/goals/store.rs b/src/graph/goals/store.rs index 82233fa..39d833d 100644 --- a/src/graph/goals/store.rs +++ b/src/graph/goals/store.rs @@ -47,7 +47,7 @@ fn thread_lock(thread_id: &str) -> Arc> { } /// Current unix time in milliseconds. Dependency-free (no `chrono`). -fn now_ms() -> u64 { +pub(crate) fn now_ms() -> u64 { SystemTime::now() .duration_since(UNIX_EPOCH) .map(|d| d.as_millis() as u64) diff --git a/src/graph/goals/test.rs b/src/graph/goals/test.rs index b9c8f7f..e40dcd2 100644 --- a/src/graph/goals/test.rs +++ b/src/graph/goals/test.rs @@ -389,3 +389,242 @@ mod tool_tests { assert!(res.content.contains("missing 'objective'")); } } + +mod continuation_tests { + use std::sync::Arc; + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::time::Duration; + + use super::super::store; + use super::super::types::{GoalProgress, ThreadGoalStatus, TurnOutcome}; + use super::super::{goal_gate_node, note_user_turn, run_continuation_tick}; + use crate::error::TinyAgentsError; + use crate::graph::GraphBuilder; + use crate::graph::command::NodeResult; + use crate::harness::store::{InMemoryStore, Store}; + + fn store() -> Arc { + Arc::new(InMemoryStore::default()) + } + + /// State overwritten by each work iteration: the tokens it "spent" and + /// whether it made progress. The gate's `progress` closure reads these. + #[derive(Clone, Debug, Default, PartialEq)] + struct GateState { + iters: usize, + tokens: u64, + progress: bool, + } + + /// Builds and runs `work_node -> gate` under `thread_id`, looping until the + /// gate routes to END or the recursion limit trips. `work` is the per-iteration + /// behavior producing the next [`GateState`]. + async fn run_gate( + s: &Arc, + thread_id: &str, + recursion_limit: usize, + per_iter_tokens: u64, + made_progress: bool, + work: W, + ) -> crate::error::Result + where + W: Fn(usize) -> Fut + Send + Sync + 'static, + Fut: std::future::Future + Send + 'static, + { + let counter = Arc::new(AtomicUsize::new(0)); + let work = Arc::new(work); + let work_node = { + let counter = counter.clone(); + let work = work.clone(); + move |_state: GateState, _ctx| { + let counter = counter.clone(); + let work = work.clone(); + Box::pin(async move { + let n = counter.fetch_add(1, Ordering::SeqCst) + 1; + work(n).await; + Ok(NodeResult::Update(GateState { + iters: n, + tokens: per_iter_tokens, + progress: made_progress, + })) + }) as crate::graph::NodeFuture + } + }; + + let gate = goal_gate_node::(s.clone(), "work", |st: &GateState| { + GoalProgress { + tokens_used: st.tokens, + elapsed_secs: 0, + made_progress: st.progress, + } + }); + + let graph = GraphBuilder::::overwrite() + .with_recursion_limit(recursion_limit) + .add_node("work", work_node) + .add_node("gate", gate) + .set_entry("work") + .add_edge("work", "gate") + .with_command_destinations("gate", ["work", crate::graph::END]) + .compile()?; + + let exec = graph + .run_with_thread(thread_id, GateState::default()) + .await?; + Ok(exec.state) + } + + #[tokio::test] + async fn gate_loops_while_active_then_stops_on_complete() { + let s = store(); + store::set(&s, "t", "obj", None).await.unwrap(); + let s2 = s.clone(); + let final_state = run_gate(&s, "t", 100, 10, true, move |n| { + let s2 = s2.clone(); + async move { + if n >= 3 { + store::complete(&s2, "t").await.unwrap(); + } + } + }) + .await + .unwrap(); + assert_eq!( + final_state.iters, 3, + "runs the work node until goal completes" + ); + let goal = store::get(&s, "t").await.unwrap().unwrap(); + assert_eq!(goal.status, ThreadGoalStatus::Complete); + } + + #[tokio::test] + async fn gate_stops_on_budget_cap() { + let s = store(); + store::set(&s, "t", "obj", Some(100)).await.unwrap(); + let final_state = run_gate(&s, "t", 100, 40, true, |_n| async {}) + .await + .unwrap(); + assert_eq!(final_state.iters, 3, "40*3 = 120 crosses the 100 budget"); + let goal = store::get(&s, "t").await.unwrap().unwrap(); + assert_eq!(goal.status, ThreadGoalStatus::BudgetLimited); + assert_eq!(goal.tokens_used, 120); + } + + #[tokio::test] + async fn gate_stops_and_suppresses_on_zero_progress() { + let s = store(); + store::set(&s, "t", "obj", None).await.unwrap(); + let final_state = run_gate(&s, "t", 100, 0, false, |_n| async {}) + .await + .unwrap(); + assert_eq!( + final_state.iters, 1, + "a no-progress iteration stops the loop" + ); + let goal = store::get(&s, "t").await.unwrap().unwrap(); + assert!(goal.continuation_suppressed, "one-shot suppression set"); + } + + #[tokio::test] + async fn gate_recursion_limit_is_the_backstop() { + let s = store(); + store::set(&s, "t", "obj", None).await.unwrap(); + // Never completes, never stalls, no budget → the loop only stops at the + // recursion limit (work+gate = 2 supersteps per iteration). + let err = run_gate(&s, "t", 6, 5, true, |_n| async {}) + .await + .unwrap_err(); + assert!(matches!(err, TinyAgentsError::RecursionLimit(_))); + } + + #[tokio::test] + async fn gate_stops_when_thread_has_no_goal() { + let s = store(); + // No goal set for the thread → gate routes straight to END after one work run. + let final_state = run_gate(&s, "t", 100, 5, true, |_n| async {}) + .await + .unwrap(); + assert_eq!(final_state.iters, 1); + } + + #[tokio::test] + async fn driver_runs_up_to_max_per_tick() { + let s = store(); + store::set(&s, "a", "a", None).await.unwrap(); + store::set(&s, "b", "b", None).await.unwrap(); + store::set(&s, "c", "c", None).await.unwrap(); + let ran = run_continuation_tick(&s, Duration::ZERO, 2, |_g| async { + Ok(TurnOutcome { + tokens_used: 5, + elapsed_secs: 0, + made_progress: true, + }) + }) + .await + .unwrap(); + assert_eq!(ran, 2, "capped at max_per_tick"); + } + + #[tokio::test] + async fn driver_skips_non_idle_goals() { + let s = store(); + store::set(&s, "fresh", "obj", None).await.unwrap(); + // A 1-hour idle window over a just-created goal → nothing runs. + let ran = run_continuation_tick(&s, Duration::from_secs(3600), 5, |_g| async { + Ok(TurnOutcome::default()) + }) + .await + .unwrap(); + assert_eq!(ran, 0); + } + + #[tokio::test] + async fn driver_suppresses_after_a_no_progress_turn() { + let s = store(); + store::set(&s, "t", "obj", None).await.unwrap(); + let ran = run_continuation_tick(&s, Duration::ZERO, 5, |_g| async { + Ok(TurnOutcome { + tokens_used: 3, + elapsed_secs: 0, + made_progress: false, + }) + }) + .await + .unwrap(); + assert_eq!(ran, 1); + assert!( + store::get(&s, "t") + .await + .unwrap() + .unwrap() + .continuation_suppressed + ); + // A second tick finds the goal suppressed and runs nothing. + let ran2 = run_continuation_tick(&s, Duration::ZERO, 5, |_g| async { + Ok(TurnOutcome::default()) + }) + .await + .unwrap(); + assert_eq!(ran2, 0); + } + + #[tokio::test] + async fn note_user_turn_clears_suppression_and_resumes() { + let s = store(); + let g = store::set(&s, "t", "obj", None).await.unwrap(); + // Suppress, then a user turn clears it. + store::set_continuation_suppressed_if(&s, "t", &g.goal_id, true) + .await + .unwrap(); + let after = note_user_turn(&s, "t").await.unwrap().unwrap(); + assert!(!after.continuation_suppressed); + + // A paused goal is reactivated by a user turn. + store::pause(&s, "t").await.unwrap(); + let after = note_user_turn(&s, "t").await.unwrap().unwrap(); + assert_eq!(after.status, ThreadGoalStatus::Active); + + // No goal → None. + assert!(note_user_turn(&s, "missing").await.unwrap().is_none()); + } +} diff --git a/src/graph/mod.rs b/src/graph/mod.rs index 489fbf0..685a687 100644 --- a/src/graph/mod.rs +++ b/src/graph/mod.rs @@ -59,7 +59,9 @@ pub use export::{ blueprint_to_mermaid, blueprint_to_topology, from_json, to_json, to_mermaid, }; pub use goals::{ - GoalProgress, ThreadGoal, ThreadGoalStatus, TurnOutcome, active_goal_context_block, + GoalProgress, GoalTool, GoalToolKind, ThreadGoal, ThreadGoalStatus, TurnOutcome, + active_goal_context_block, goal_gate_node, goal_tools, note_user_turn, register_goal_tools, + run_continuation_tick, }; pub use observability::{ GraphEventJournal, GraphHealthSummary, GraphLangfuseExporter, GraphLatencyMetrics, From e59aba4bf7706da46f834759dedc1baa4b60d674 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 2 Jul 2026 10:23:55 -0700 Subject: [PATCH 05/10] graph::goals: wire crate-root re-exports + module README Re-export the goal surface at the crate root (goal_store CRUD alias, GoalTool, goal_gate_node, run_continuation_tick, note_user_turn, ThreadGoal, ...) and note the primitive in the graph-surface doc. Add src/graph/goals/README.md covering the data model, Store key scheme, the single-process RMW/CAS caveat, the tool surface, and the continuation mapping. Claude-Session: https://claude.ai/code/session_01NRL2bim9Gz3UN3SRtpeXEX --- src/graph/goals/README.md | 123 ++++++++++++++++++++++++++++++++++++++ src/lib.rs | 13 +++- 2 files changed, 135 insertions(+), 1 deletion(-) create mode 100644 src/graph/goals/README.md diff --git a/src/graph/goals/README.md b/src/graph/goals/README.md new file mode 100644 index 0000000..3504c94 --- /dev/null +++ b/src/graph/goals/README.md @@ -0,0 +1,123 @@ +# graph::goals + +A per-thread **goal**: exactly one durable objective per thread, carried across +supersteps, interrupts, and resumes. Ported from OpenHuman's `thread_goals` and +re-hosted on the TinyAgents graph runtime — provider-neutral, offline-testable, +with no app-specific coupling (no event bus, RPC envelopes, or heartbeat). + +Distinct from [`graph::todos`](../todos) (a *list* of task cards per thread): a +goal is the single "completion contract" the graph pursues; the board holds the +concrete work items. + +## Data model (`types.rs`) + +- `ThreadGoalStatus { Active, Paused, BudgetLimited, Complete }` — `is_active` / + `is_terminal` predicates. Ownership is asymmetric: a model creates/replaces a + goal and marks it `Complete`; `Paused` / `BudgetLimited` are system-driven. +- `ThreadGoal { thread_id, goal_id, objective, status, token_budget, tokens_used, + time_used_seconds, created_at_ms, updated_at_ms, continuation_suppressed }` + (serde `camelCase`) with `budget_remaining` / `over_budget`. +- `GoalProgress { tokens_used, elapsed_secs, made_progress }` — usage a work + iteration reports to the continuation gate. `TurnOutcome` is an alias used at + the driver boundary. +- `active_goal_context_block(&ThreadGoal) -> Option` — a prompt block to + prepend to a work node's input (steer text for Active / stop text for + BudgetLimited; `None` for Paused/Complete). + +## Persistence (`store.rs`) + +One serialized `ThreadGoal` per thread under the `graph.goals` namespace of a +`crate::harness::store::Store`, keyed by `hex(thread_id)` (hex keeps arbitrary +thread ids valid across `InMemoryStore` / `FileStore` / Sqlite). CRUD: +`get` / `set` / `set_if_absent` / `complete` / `pause` / `resume` / `clear` / +`list_all` / `account_usage` / `set_continuation_suppressed_if`. + +Semantics preserved from OpenHuman: + +- `set` mints a fresh `goal_id` and resets counters when the objective changes; + a same-objective re-set preserves counters and re-opens to `Active` unless + still over budget. +- `account_usage` folds token/time usage and flips an active goal to + `BudgetLimited` at the cap. Its `expected_goal_id` **compare-and-set** guard + silently drops stale accounting from a replaced goal. +- `set_continuation_suppressed_if` writes only when the current goal still + matches `expected_goal_id` and is active. + +**Concurrency / single-process caveat.** The `Store` trait offers no CAS and no +cross-key transaction, so each mutation runs `load → mutate → put` under a +per-thread async mutex — atomic **within one process**. Across processes sharing +a `FileStore`, two concurrent read-modify-writes can lose an update; the +`goal_id` guard still prevents logical corruption from stale accounting but not +lost updates. Funnel goal mutations through one process, or wait for a future +`Store::compare_and_swap`. + +## Tools (`tool.rs`) + +`GoalTool` dispatches on `GoalToolKind`. The default model-facing set +(`goal_tools` / `register_goal_tools`) is `goal_get`, `goal_set`, +`goal_complete`; `goal_pause` / `goal_resume` / `goal_clear` are constructible +host controls. The target thread comes from `ToolExecutionContext::thread_id` +(never a tool argument), so a model can't address another thread's goal; the +bare `Tool::call` entry point (no context) errors. + +## Continuation (`continuation.rs`) + +OpenHuman's idle heartbeat becomes graph-native, three ways: + +- **`goal_gate_node`** (primary) — a command-routing node forming a self-driving + bounded loop. Wire `work_node -> gate` and register `gate` with + `with_command_destinations([work_node, END])`. Each pass reads the thread id + from `NodeContext`, folds the iteration's `GoalProgress` via `account_usage`, + and routes back to `work_node` while the goal is Active and under budget, else + to `END`. `recursion_limit` is the hard backstop; a zero-progress iteration + sets the one-shot suppression and stops. +- **`run_continuation_tick`** (driver) — for callers with an external scheduler: + selects idle, active, non-suppressed goals (oldest-first, `max_per_tick`) and + runs one turn each through a `run_turn` closure, then accounts + one-shot + suppresses on no progress. +- **`note_user_turn`** — call at the start of a user-initiated run to clear the + one-shot suppression and reactivate a paused goal. A loop iteration never + clears its own suppression. + +**Token accounting boundary.** The graph runtime does not meter tokens per node, +so accounting is explicit: a work node writes what it spent into `State` and the +`progress` / `run_turn` closure reports it. `made_progress == false` is the +graph analogue of OpenHuman's "the turn produced no tool calls". + +## Example: a self-driving goal loop + +```rust,ignore +use std::sync::Arc; +use tinyagents::{GraphBuilder, END, GoalProgress, goal_gate_node, goal_store}; +use tinyagents::harness::store::{InMemoryStore, Store}; + +let store: Arc = Arc::new(InMemoryStore::default()); +goal_store::set(&store, "thread-1", "summarise the repo", Some(50_000)).await?; + +let gate = goal_gate_node::(store.clone(), "work", |s: &St| GoalProgress { + tokens_used: s.last_turn_tokens, + elapsed_secs: 0, + made_progress: s.made_progress, +}); + +let graph = GraphBuilder::::overwrite() + .with_recursion_limit(64) + .add_node("work", work_node) // a subagent_node that calls goal_complete when done + .add_node("gate", gate) + .set_entry("work") + .add_edge("work", "gate") + .with_command_destinations("gate", ["work", END]) + .compile()?; + +let exec = graph.run_with_thread("thread-1", St::default()).await?; +``` + +## Files + +| File | Role | +| --- | --- | +| `types.rs` | Data model + `GoalProgress` + `active_goal_context_block`. | +| `store.rs` | `Store`-backed CRUD, per-thread RMW lock, budget + CAS guards. | +| `tool.rs` | `GoalTool` / `GoalToolKind` harness tools. | +| `continuation.rs` | `goal_gate_node`, `run_continuation_tick`, `note_user_turn`. | +| `test.rs` | Unit tests (types, store, tools, continuation loop). | diff --git a/src/lib.rs b/src/lib.rs index b389c22..e8178e8 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -26,7 +26,8 @@ //! 2. **Graph runtime** ([`graph`]) — LangGraph-style durable typed state //! graphs: [`START`]/[`END`], nodes, conditional routing, [`Command`]s, //! fan-out, reducers/channels, [`Checkpoint`]s, [`Interrupt`]s, subgraphs, -//! streaming, and topology export. +//! streaming, topology export, and a per-thread durable [`ThreadGoal`] with +//! graph-native continuation, exposed as harness tools. //! 3. **Registry** ([`registry`]) — a named capability catalog (models, tools, //! agents, graphs, stores, middleware, policy) that `.rag`/`.ragsh` bind by //! name. @@ -184,6 +185,16 @@ pub use graph::{ orchestration_tools_with_steering, register_orchestration_tools, }; +// --- Graph: per-thread goal (durable objective + graph-native continuation) --- +// `goal_store` is the programmatic CRUD surface (get/set/complete/account_usage); +// the tools and continuation helpers are re-exported flat for discoverability. +pub use graph::goals::store as goal_store; +pub use graph::{ + GoalProgress, GoalTool, GoalToolKind, ThreadGoal, ThreadGoalStatus, TurnOutcome, + active_goal_context_block, goal_gate_node, goal_tools, note_user_turn, register_goal_tools, + run_continuation_tick, +}; + // --- Graph: parallel map/reduce helper --- pub use graph::parallel::{ FailurePolicy, ItemOutcome, ParallelOptions, ParallelOutcome, map_reduce, From f43022e400da5e39446423cebc0ba51ec50f090d Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 2 Jul 2026 10:26:45 -0700 Subject: [PATCH 06/10] graph::todos: add task-board card model, normalise, markdown render Introduce the per-thread kanban board under graph::todos: TaskCardStatus, TaskApprovalMode, TaskBoardCard, TaskBoard, plus the pure helpers parse_status, render_markdown (GitHub-flavored markers [ ]/[x]/[~]/[!]/[?]/[-] + indented metadata), normalise_board (id generation, order recompute, empty-title drop, blocker-from-notes), CardPatch, and TodosSnapshot. Ported from OpenHuman's task board; uses next_seq() ids and a dependency-free millis timestamp. Claude-Session: https://claude.ai/code/session_01NRL2bim9Gz3UN3SRtpeXEX --- src/graph/mod.rs | 5 + src/graph/todos/mod.rs | 23 +++ src/graph/todos/test.rs | 122 +++++++++++++ src/graph/todos/types.rs | 382 +++++++++++++++++++++++++++++++++++++++ 4 files changed, 532 insertions(+) create mode 100644 src/graph/todos/mod.rs create mode 100644 src/graph/todos/test.rs create mode 100644 src/graph/todos/types.rs diff --git a/src/graph/mod.rs b/src/graph/mod.rs index 685a687..3ef39d9 100644 --- a/src/graph/mod.rs +++ b/src/graph/mod.rs @@ -35,6 +35,7 @@ pub mod stream; pub mod subagent_node; pub mod subgraph; pub mod testkit; +pub mod todos; // --- Durable execution model --- pub use builder::{ @@ -94,3 +95,7 @@ pub use testkit::{ assert_graph, failing_node, fanout_node, interrupting_node, noop_node, run_recorded, scripted_route_node, scripted_update_node, subagent_fake_node, subgraph_test_node, }; +pub use todos::{ + CardPatch, TaskApprovalMode, TaskBoard, TaskBoardCard, TaskCardStatus, TodosSnapshot, + normalise_board, parse_status, render_markdown, +}; diff --git a/src/graph/todos/mod.rs b/src/graph/todos/mod.rs new file mode 100644 index 0000000..8f248d4 --- /dev/null +++ b/src/graph/todos/mod.rs @@ -0,0 +1,23 @@ +//! Per-thread **task board** (kanban todos): a list of task cards per thread. +//! +//! Where [`graph::goals`](crate::graph::goals) holds a single durable objective +//! per thread, a task board holds the concrete work items: an ordered list of +//! [`TaskBoardCard`]s with a small kanban lifecycle. This module owns the data +//! model and markdown rendering ([`types`]), +//! harness-[`Store`](crate::harness::store::Store)-backed CRUD with the +//! single-`InProgress` invariant ([`store`]), and the model-facing multiplexer +//! tool ([`tool`]). +//! +//! Ported from OpenHuman's task board / `todos` modules, minus the app-specific +//! coupling (progress events, RPC envelopes, in-memory scratch fallback): a +//! board is always `(Store, thread_id)`. + +mod types; + +pub use types::{ + CardPatch, TaskApprovalMode, TaskBoard, TaskBoardCard, TaskCardStatus, TodosSnapshot, + normalise_board, parse_status, render_markdown, +}; + +#[cfg(test)] +mod test; diff --git a/src/graph/todos/test.rs b/src/graph/todos/test.rs new file mode 100644 index 0000000..d6e326b --- /dev/null +++ b/src/graph/todos/test.rs @@ -0,0 +1,122 @@ +//! Unit tests for the task-board domain types. + +use super::types::*; + +fn card(id: &str, title: &str, status: TaskCardStatus) -> TaskBoardCard { + TaskBoardCard { + id: id.to_string(), + title: title.to_string(), + status, + ..TaskBoardCard::new(title) + } +} + +#[test] +fn status_strings_match_serialized() { + assert_eq!(TaskCardStatus::Todo.as_str(), "todo"); + assert_eq!( + TaskCardStatus::AwaitingApproval.as_str(), + "awaiting_approval" + ); + assert_eq!(TaskCardStatus::Ready.as_str(), "ready"); + assert_eq!(TaskCardStatus::InProgress.as_str(), "in_progress"); + assert_eq!(TaskCardStatus::Blocked.as_str(), "blocked"); + assert_eq!(TaskCardStatus::Done.as_str(), "done"); + assert_eq!(TaskCardStatus::Rejected.as_str(), "rejected"); + assert_eq!(TaskApprovalMode::Required.as_str(), "required"); + assert_eq!(TaskApprovalMode::NotRequired.as_str(), "not_required"); +} + +#[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 + ); + assert_eq!(parse_status("approved").unwrap(), TaskCardStatus::Ready); + assert_eq!(parse_status("done").unwrap(), TaskCardStatus::Done); + assert_eq!(parse_status("denied").unwrap(), TaskCardStatus::Rejected); + assert!(parse_status("nope").is_err()); +} + +#[test] +fn card_and_board_round_trip_through_json() { + let mut c = card("task-1", "Draft plan", TaskCardStatus::AwaitingApproval); + c.approval_mode = Some(TaskApprovalMode::Required); + c.plan = vec!["step one".into()]; + let board = TaskBoard { + thread_id: "t".into(), + cards: vec![c.clone()], + updated_at: "0".into(), + }; + let json = serde_json::to_value(&board).unwrap(); + assert_eq!(json["threadId"], "t"); + assert_eq!(json["cards"][0]["approvalMode"], "required"); + let back: TaskBoard = serde_json::from_value(json).unwrap(); + assert_eq!(back, board); +} + +#[test] +fn render_markdown_uses_status_markers_and_sub_lines() { + let mut done = card("task-1", "Ship it", TaskCardStatus::Done); + done.objective = Some("release the crate".into()); + let mut blocked = card("task-2", "Wait on CI", TaskCardStatus::Blocked); + blocked.blocker = Some("CI is red".into()); + let in_progress = card("task-3", "Write docs", TaskCardStatus::InProgress); + let awaiting = card("task-4", "Approve plan", TaskCardStatus::AwaitingApproval); + let rejected = card("task-5", "Nope", TaskCardStatus::Rejected); + let todo = card("task-6", "Later", TaskCardStatus::Todo); + + let md = render_markdown(&[done, blocked, in_progress, awaiting, rejected, todo]); + assert!(md.contains("- [x] Ship it `(task-1)`")); + assert!(md.contains(" - objective: release the crate")); + assert!(md.contains("- [!] Wait on CI")); + assert!(md.contains(" - _blocked:_ CI is red")); + assert!(md.contains("- [~] Write docs")); + assert!(md.contains("- [?] Approve plan")); + assert!(md.contains("- [-] Nope")); + assert!(md.contains("- [ ] Later")); +} + +#[test] +fn render_markdown_empty_is_placeholder() { + assert_eq!(render_markdown(&[]), "_No todos yet._"); +} + +#[test] +fn normalise_trims_generates_ids_and_recomputes_order() { + let mut board = TaskBoard { + thread_id: " t ".into(), + cards: vec![ + TaskBoardCard { + order: 99, + objective: Some(" ship briefs ".into()), + plan: vec![" extend schema ".into(), " ".into()], + allowed_tools: vec![" todo ".into(), "".into()], + ..card("", " Draft plan ", TaskCardStatus::Todo) + }, + // Empty title → dropped. + card("empty", " ", TaskCardStatus::Todo), + // Blocked without a blocker → backfilled from notes. + TaskBoardCard { + notes: Some("waiting on user".into()), + ..card("blocked", "Need approval", TaskCardStatus::Blocked) + }, + ], + updated_at: String::new(), + }; + normalise_board(&mut board); + + assert_eq!(board.thread_id, "t"); + assert_eq!(board.cards.len(), 2, "empty-title card dropped"); + assert_eq!(board.cards[0].title, "Draft plan"); + assert_eq!(board.cards[0].objective.as_deref(), Some("ship briefs")); + assert_eq!(board.cards[0].plan, vec!["extend schema"]); + assert_eq!(board.cards[0].allowed_tools, vec!["todo"]); + assert!(board.cards[0].id.starts_with("task-"), "blank id generated"); + assert_eq!(board.cards[0].order, 0); + assert_eq!(board.cards[1].order, 1); + assert_eq!(board.cards[1].blocker.as_deref(), Some("waiting on user")); +} diff --git a/src/graph/todos/types.rs b/src/graph/todos/types.rs new file mode 100644 index 0000000..a02d695 --- /dev/null +++ b/src/graph/todos/types.rs @@ -0,0 +1,382 @@ +//! Domain types for the per-thread task board (kanban todos). +//! +//! A **task board** is a per-thread list of [`TaskBoardCard`]s — the concrete +//! work items a graph tracks, distinct from the single per-thread +//! [`ThreadGoal`](crate::graph::goals::ThreadGoal). Ported from OpenHuman's +//! task board / `todos` modules, minus the app-specific coupling (progress +//! events, RPC envelopes, scratch fallback). + +use serde::{Deserialize, Serialize}; + +use crate::harness::ids::next_seq; + +/// Lifecycle state of a task card. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum TaskCardStatus { + /// Not started. + Todo, + /// Plan approval required and pending; will not run until approved + /// (→ `Ready`) or rejected (→ `Rejected`). + AwaitingApproval, + /// Approved for execution — runnable without a further approval check. + Ready, + /// Currently being worked. At most one card may be `InProgress` at a time. + InProgress, + /// Blocked on an external dependency; carries a `blocker` reason. + Blocked, + /// Finished. + Done, + /// Plan approval was denied; the card is not executed. + Rejected, +} + +impl TaskCardStatus { + /// The stable lower-snake-case status label. + 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", + } + } +} + +/// Whether a card requires human plan approval before it runs. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum TaskApprovalMode { + /// A human must approve the plan before execution. + Required, + /// No approval gate. + NotRequired, +} + +impl TaskApprovalMode { + /// The stable lower-snake-case mode label. + pub fn as_str(&self) -> &'static str { + match self { + Self::Required => "required", + Self::NotRequired => "not_required", + } + } +} + +/// A single task card on a thread's board. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct TaskBoardCard { + /// Stable card id (`task-`); server-generated when blank. + pub id: String, + /// One-line title / summary. + pub title: String, + /// Lifecycle state. + pub status: TaskCardStatus, + /// Optional richer objective for the card. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub objective: Option, + /// Ordered plan steps. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub plan: Vec, + /// The agent assigned to run this card, if any. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub assigned_agent: Option, + /// Tools the assigned agent is allowed to use. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub allowed_tools: Vec, + /// Plan-approval mode, if the card is gated. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub approval_mode: Option, + /// Acceptance criteria that define "done" for this card. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub acceptance_criteria: Vec, + /// Evidence gathered toward completion. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub evidence: Vec, + /// Free-form notes. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub notes: Option, + /// Blocker reason (populated from `notes` on normalise when a `Blocked` card + /// has none). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub blocker: Option, + /// Conversation thread id of the card's live/last run, for UI cross-linking. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub session_thread_id: Option, + /// Provider/source identifiers for a card ingested from an external source. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub source_metadata: Option, + /// Position on the board (recomputed on normalise). + #[serde(default)] + pub order: u32, + /// Last-mutation timestamp (unix-epoch milliseconds, as a string). + #[serde(default)] + pub updated_at: String, +} + +impl TaskBoardCard { + /// Creates a minimal `Todo` card with `title`, a fresh id, and no metadata. + pub fn new(title: impl Into) -> Self { + Self { + id: new_card_id(), + title: title.into(), + status: TaskCardStatus::Todo, + 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: now_stamp(), + } + } +} + +/// A per-thread board: an ordered list of cards. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct TaskBoard { + /// The thread this board belongs to. + pub thread_id: String, + /// The cards, in board order. + pub cards: Vec, + /// Last-mutation timestamp (unix-epoch milliseconds, as a string). + pub updated_at: String, +} + +impl TaskBoard { + /// An empty board for `thread_id`. + pub fn empty(thread_id: impl Into) -> Self { + Self { + thread_id: thread_id.into(), + cards: Vec::new(), + updated_at: now_stamp(), + } + } +} + +/// A single CRUD outcome: the post-mutation cards plus a markdown rendering. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct TodosSnapshot { + /// The thread the board belongs to. + pub thread_id: String, + /// The cards after the mutation. + pub cards: Vec, + /// GitHub-flavored markdown rendering of the cards. + pub markdown: String, +} + +/// Optional fields supplied by `add` / `edit`. +/// +/// `approval_mode` is doubly-optional: `None` leaves it untouched, `Some(None)` +/// clears it, `Some(Some(_))` sets it. +#[derive(Debug, Default, Clone)] +pub struct CardPatch { + /// New title (the model-facing "content"). + pub content: Option, + /// New status. + pub status: Option, + /// New objective (empty clears). + pub objective: Option, + /// New plan steps. + pub plan: Option>, + /// New assigned agent (empty clears). + pub assigned_agent: Option, + /// New allowed-tools list. + pub allowed_tools: Option>, + /// New approval mode (`Some(None)` clears). + pub approval_mode: Option>, + /// New acceptance criteria. + pub acceptance_criteria: Option>, + /// New evidence list. + pub evidence: Option>, + /// New notes (empty clears). + pub notes: Option, + /// New blocker (empty clears). + pub blocker: Option, + /// New source metadata (`Some` sets; `None` leaves untouched). + pub source_metadata: Option, +} + +/// Parses a stable string status alias into a [`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)" + )), + } +} + +/// Renders a card list as GitHub-flavored markdown: one `- [marker] title` +/// line per card (`[ ]` todo/ready, `[?]` awaiting approval, `[~]` in progress, +/// `[!]` blocked, `[x]` done, `[-]` rejected) followed by indented metadata. +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() +} + +/// Normalises a board in place: trims fields, generates missing card ids, drops +/// empty-title cards, backfills a `Blocked` card's blocker from its notes, and +/// recomputes `order` / `updated_at`. +pub fn normalise_board(board: &mut TaskBoard) { + let now = now_stamp(); + board.thread_id = board.thread_id.trim().to_string(); + board.updated_at = now.clone(); + + for card in board.cards.iter_mut() { + card.title = card.title.trim().to_string(); + if card.id.trim().is_empty() { + card.id = new_card_id(); + } else { + card.id = card.id.trim().to_string(); + } + card.notes = trim_opt(card.notes.take()); + card.objective = trim_opt(card.objective.take()); + card.assigned_agent = trim_opt(card.assigned_agent.take()); + trim_string_vec(&mut card.plan); + trim_string_vec(&mut card.allowed_tools); + trim_string_vec(&mut card.acceptance_criteria); + trim_string_vec(&mut card.evidence); + card.blocker = trim_opt(card.blocker.take()); + card.session_thread_id = trim_opt(card.session_thread_id.take()); + if card.status == TaskCardStatus::Blocked && card.blocker.is_none() { + card.blocker = card.notes.clone(); + } + } + + board.cards.retain(|card| !card.title.is_empty()); + + for (idx, card) in board.cards.iter_mut().enumerate() { + card.order = idx as u32; + card.updated_at = now.clone(); + } +} + +/// Trims a string, returning `None` when the result is empty. +pub(crate) fn non_empty(s: String) -> Option { + let trimmed = s.trim(); + if trimmed.is_empty() { + None + } else { + Some(trimmed.to_string()) + } +} + +fn trim_opt(value: Option) -> Option { + value.and_then(non_empty) +} + +fn trim_string_vec(values: &mut Vec) { + values.retain_mut(|value| { + *value = value.trim().to_string(); + !value.is_empty() + }); +} + +/// Mints a fresh, process-unique card id (`task-`). +pub(crate) fn new_card_id() -> String { + format!("task-{}", next_seq()) +} + +/// Current unix time in milliseconds, as a string. Dependency-free (no `chrono`). +pub(crate) fn now_stamp() -> String { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0) + .to_string() +} From a383bd75f57e43ff306f9188fee50df2623056c2 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 2 Jul 2026 10:28:56 -0700 Subject: [PATCH 07/10] graph::todos: persist boards on harness Store with CRUD + single-in-progress Add graph::todos::store: namespaced (graph.todos) Store-backed board CRUD keyed by hex(thread_id), serialised per thread by an async mutex. Ports add/edit/ update_status/decide_plan/revise_plan/remove/replace/clear/list/claim_card/ set_session_thread, each returning a TodosSnapshot (normalised cards + markdown). Enforces the single-in-progress invariant (errors, never auto-fixes), guards decide_plan to AwaitingApproval cards, and does an atomic CAS claim_card. Claude-Session: https://claude.ai/code/session_01NRL2bim9Gz3UN3SRtpeXEX --- src/graph/todos/mod.rs | 1 + src/graph/todos/store.rs | 396 +++++++++++++++++++++++++++++++++++++++ src/graph/todos/test.rs | 232 +++++++++++++++++++++++ 3 files changed, 629 insertions(+) create mode 100644 src/graph/todos/store.rs diff --git a/src/graph/todos/mod.rs b/src/graph/todos/mod.rs index 8f248d4..704550d 100644 --- a/src/graph/todos/mod.rs +++ b/src/graph/todos/mod.rs @@ -12,6 +12,7 @@ //! coupling (progress events, RPC envelopes, in-memory scratch fallback): a //! board is always `(Store, thread_id)`. +pub mod store; mod types; pub use types::{ diff --git a/src/graph/todos/store.rs b/src/graph/todos/store.rs new file mode 100644 index 0000000..cad263d --- /dev/null +++ b/src/graph/todos/store.rs @@ -0,0 +1,396 @@ +//! CRUD for the per-thread task board, on the harness +//! [`Store`](crate::harness::store::Store). +//! +//! Each thread's board is a single serialized [`TaskBoard`] value under the +//! [`TODOS_NAMESPACE`] namespace, keyed by the hex-encoded thread id. Every +//! mutation runs `load → mutate → normalise → put` under a **per-thread async +//! mutex** ([`thread_lock`]) so the read-modify-write is atomic within the +//! process (the same single-process caveat as +//! [`graph::goals::store`](crate::graph::goals::store)). +//! +//! Each mutator returns a [`TodosSnapshot`] — the normalised cards plus a +//! markdown rendering — so an agent transcript and a UI stay in lock-step. + +use std::collections::HashMap; +use std::sync::{Arc, Mutex as StdMutex, OnceLock}; + +use tokio::sync::Mutex; + +use super::types::{ + CardPatch, TaskBoard, TaskBoardCard, TaskCardStatus, TodosSnapshot, non_empty, normalise_board, + now_stamp, render_markdown, +}; +use crate::error::{Result, TinyAgentsError}; +use crate::harness::store::Store; + +/// The [`Store`] namespace holding one [`TaskBoard`] per thread. +pub const TODOS_NAMESPACE: &str = "graph.todos"; + +/// Serialises `load → mutate → put` per thread so a read-modify-write is atomic +/// within the process. +fn thread_lock(thread_id: &str) -> Arc> { + static LOCKS: OnceLock>>>> = OnceLock::new(); + let map = LOCKS.get_or_init(|| StdMutex::new(HashMap::new())); + let mut guard = map.lock().expect("todo lock map poisoned"); + guard.entry(thread_id.to_string()).or_default().clone() +} + +/// Hex-encodes the thread id into a [`Store`]-safe key. +fn key(thread_id: &str) -> String { + thread_id + .as_bytes() + .iter() + .map(|b| format!("{b:02x}")) + .collect() +} + +fn validate_thread_id(thread_id: &str) -> Result { + let trimmed = thread_id.trim(); + if trimmed.is_empty() { + return Err(TinyAgentsError::Validation( + "task board thread_id must not be empty or whitespace".to_string(), + )); + } + Ok(trimmed.to_string()) +} + +/// Loads the raw cards for `thread_id` (empty when the thread has no board). +async fn load_cards(store: &Arc, thread_id: &str) -> Result> { + match store.get(TODOS_NAMESPACE, &key(thread_id)).await? { + Some(value) => { + let board: TaskBoard = serde_json::from_value(value)?; + Ok(board.cards) + } + None => Ok(Vec::new()), + } +} + +/// Normalises and persists `cards` for `thread_id`, returning the normalised set. +async fn save_cards( + store: &Arc, + thread_id: &str, + cards: Vec, +) -> Result> { + let mut board = TaskBoard { + thread_id: thread_id.to_string(), + cards, + updated_at: now_stamp(), + }; + normalise_board(&mut board); + let value = serde_json::to_value(&board)?; + store.put(TODOS_NAMESPACE, &key(thread_id), value).await?; + Ok(board.cards) +} + +fn snapshot(thread_id: &str, cards: Vec) -> TodosSnapshot { + let markdown = render_markdown(&cards); + TodosSnapshot { + thread_id: thread_id.to_string(), + cards, + markdown, + } +} + +/// At most one card may be `InProgress` at a time. Returns a +/// [`Validation`](TinyAgentsError::Validation) error otherwise (never silently +/// fixes it). +fn enforce_single_in_progress(cards: &[TaskBoardCard]) -> Result<()> { + let in_progress = cards + .iter() + .filter(|c| matches!(c.status, TaskCardStatus::InProgress)) + .count(); + if in_progress > 1 { + return Err(TinyAgentsError::Validation(format!( + "only one todo may be `in_progress` at a time (got {in_progress})" + ))); + } + Ok(()) +} + +/// Snapshot the current board without mutating. +pub async fn list(store: &Arc, thread_id: &str) -> Result { + let thread_id = validate_thread_id(thread_id)?; + let lock = thread_lock(&thread_id); + let _guard = lock.lock().await; + let cards = load_cards(store, &thread_id).await?; + Ok(snapshot(&thread_id, cards)) +} + +/// Append a new card. `content` is the required title; `patch` supplies the rest. +pub async fn add( + store: &Arc, + thread_id: &str, + content: &str, + patch: CardPatch, +) -> Result { + let thread_id = validate_thread_id(thread_id)?; + let content = content.trim(); + if content.is_empty() { + return Err(TinyAgentsError::Validation( + "todo content must not be empty".to_string(), + )); + } + let lock = thread_lock(&thread_id); + let _guard = lock.lock().await; + let mut cards = load_cards(store, &thread_id).await?; + let order = cards.len() as u32; + cards.push(TaskBoardCard { + 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), + source_metadata: patch.source_metadata, + order, + ..TaskBoardCard::new(content) + }); + enforce_single_in_progress(&cards)?; + let cards = save_cards(store, &thread_id, cards).await?; + Ok(snapshot(&thread_id, cards)) +} + +/// Edit an existing card. Fields left `None` in `patch` are untouched. Errors if +/// `id` is unknown. +pub async fn edit( + store: &Arc, + thread_id: &str, + id: &str, + patch: CardPatch, +) -> Result { + let thread_id = validate_thread_id(thread_id)?; + let lock = thread_lock(&thread_id); + let _guard = lock.lock().await; + let mut cards = load_cards(store, &thread_id).await?; + let card = cards + .iter_mut() + .find(|c| c.id == id) + .ok_or_else(|| TinyAgentsError::Validation(format!("todo id '{id}' not found")))?; + if let Some(content) = patch.content { + let trimmed = content.trim().to_string(); + if trimmed.is_empty() { + return Err(TinyAgentsError::Validation( + "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 = now_stamp(); + enforce_single_in_progress(&cards)?; + let cards = save_cards(store, &thread_id, cards).await?; + Ok(snapshot(&thread_id, cards)) +} + +/// Update only the status of a card. +pub async fn update_status( + store: &Arc, + thread_id: &str, + id: &str, + status: TaskCardStatus, +) -> Result { + edit( + store, + thread_id, + id, + CardPatch { + status: Some(status), + ..Default::default() + }, + ) + .await +} + +/// Stamp (or clear, with a blank id) a card's `session_thread_id` — the +/// conversation thread of its live/last run. Pure session-link bookkeeping, +/// orthogonal to the lifecycle (does not touch status or the invariant). +pub async fn set_session_thread( + store: &Arc, + thread_id: &str, + id: &str, + session_thread_id: Option, +) -> Result { + let thread_id = validate_thread_id(thread_id)?; + let lock = thread_lock(&thread_id); + let _guard = lock.lock().await; + let mut cards = load_cards(store, &thread_id).await?; + let card = cards + .iter_mut() + .find(|c| c.id == id) + .ok_or_else(|| TinyAgentsError::Validation(format!("todo id '{id}' not found")))?; + card.session_thread_id = session_thread_id.and_then(non_empty); + card.updated_at = now_stamp(); + let cards = save_cards(store, &thread_id, cards).await?; + Ok(snapshot(&thread_id, cards)) +} + +/// Resolve a plan-approval decision: approve (→ `Ready`) 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( + store: &Arc, + thread_id: &str, + id: &str, + approve: bool, +) -> Result { + let thread_id = validate_thread_id(thread_id)?; + let lock = thread_lock(&thread_id); + let _guard = lock.lock().await; + let mut cards = load_cards(store, &thread_id).await?; + let card = cards + .iter_mut() + .find(|c| c.id == id) + .ok_or_else(|| TinyAgentsError::Validation(format!("todo id '{id}' not found")))?; + if card.status != TaskCardStatus::AwaitingApproval { + return Err(TinyAgentsError::Validation(format!( + "card '{id}' is not awaiting approval (status: {})", + card.status.as_str() + ))); + } + card.status = if approve { + TaskCardStatus::Ready + } else { + TaskCardStatus::Rejected + }; + card.updated_at = now_stamp(); + let cards = save_cards(store, &thread_id, cards).await?; + Ok(snapshot(&thread_id, cards)) +} + +/// Reject **every** `AwaitingApproval` card so none stays runnable, clearing a +/// parked plan for re-planning. Lenient when nothing is awaiting (a benign +/// no-op rather than an error). +pub async fn revise_plan(store: &Arc, thread_id: &str) -> Result { + let thread_id = validate_thread_id(thread_id)?; + let lock = thread_lock(&thread_id); + let _guard = lock.lock().await; + let mut cards = load_cards(store, &thread_id).await?; + for card in cards.iter_mut() { + if card.status == TaskCardStatus::AwaitingApproval { + card.status = TaskCardStatus::Rejected; + card.updated_at = now_stamp(); + } + } + let cards = save_cards(store, &thread_id, cards).await?; + Ok(snapshot(&thread_id, cards)) +} + +/// Remove a card by id. Errors if `id` is unknown. +pub async fn remove(store: &Arc, thread_id: &str, id: &str) -> Result { + let thread_id = validate_thread_id(thread_id)?; + let lock = thread_lock(&thread_id); + let _guard = lock.lock().await; + let mut cards = load_cards(store, &thread_id).await?; + let before = cards.len(); + cards.retain(|c| c.id != id); + if cards.len() == before { + return Err(TinyAgentsError::Validation(format!( + "todo id '{id}' not found" + ))); + } + let cards = save_cards(store, &thread_id, cards).await?; + Ok(snapshot(&thread_id, cards)) +} + +/// Wholesale-replace the board's cards. Cards missing ids get server-generated +/// ones on normalise. +pub async fn replace( + store: &Arc, + thread_id: &str, + cards: Vec, +) -> Result { + let thread_id = validate_thread_id(thread_id)?; + let lock = thread_lock(&thread_id); + let _guard = lock.lock().await; + enforce_single_in_progress(&cards)?; + let cards = save_cards(store, &thread_id, cards).await?; + Ok(snapshot(&thread_id, cards)) +} + +/// Empty the board. +pub async fn clear(store: &Arc, thread_id: &str) -> Result { + let thread_id = validate_thread_id(thread_id)?; + let lock = thread_lock(&thread_id); + let _guard = lock.lock().await; + let cards = save_cards(store, &thread_id, Vec::new()).await?; + Ok(snapshot(&thread_id, cards)) +} + +/// Atomic compare-and-set claim: transition a card from one of `expected` to +/// `target` under the per-thread lock, returning the fresh card on success. If +/// the card's current status is not in `expected`, the claim is rejected — the +/// caller lost the race or the card already moved on. +pub async fn claim_card( + store: &Arc, + thread_id: &str, + card_id: &str, + expected: &[TaskCardStatus], + target: TaskCardStatus, +) -> Result { + let thread_id = validate_thread_id(thread_id)?; + let lock = thread_lock(&thread_id); + let _guard = lock.lock().await; + let mut cards = load_cards(store, &thread_id).await?; + let card = cards.iter_mut().find(|c| c.id == card_id).ok_or_else(|| { + TinyAgentsError::Validation(format!("claim_card: card '{card_id}' not found on board")) + })?; + if !expected.contains(&card.status) { + return Err(TinyAgentsError::Validation(format!( + "claim_card: card '{card_id}' status is '{}', expected one of [{}]; claim rejected", + card.status.as_str(), + expected + .iter() + .map(TaskCardStatus::as_str) + .collect::>() + .join(", ") + ))); + } + card.status = target; + card.updated_at = now_stamp(); + let claimed_id = card.id.clone(); + enforce_single_in_progress(&cards)?; + let cards = save_cards(store, &thread_id, cards).await?; + cards + .into_iter() + .find(|c| c.id == claimed_id) + .ok_or_else(|| { + TinyAgentsError::Graph(format!("claim_card: card '{claimed_id}' lost after save")) + }) +} diff --git a/src/graph/todos/test.rs b/src/graph/todos/test.rs index d6e326b..66ccb59 100644 --- a/src/graph/todos/test.rs +++ b/src/graph/todos/test.rs @@ -120,3 +120,235 @@ fn normalise_trims_generates_ids_and_recomputes_order() { assert_eq!(board.cards[1].order, 1); assert_eq!(board.cards[1].blocker.as_deref(), Some("waiting on user")); } + +mod store_tests { + use std::sync::Arc; + + use super::super::store; + use super::super::types::{CardPatch, TaskBoardCard, TaskCardStatus}; + use crate::harness::store::{InMemoryStore, Store}; + + fn store() -> Arc { + Arc::new(InMemoryStore::default()) + } + + #[tokio::test] + async fn add_list_remove_round_trip() { + let s = store(); + assert!(store::list(&s, "t").await.unwrap().cards.is_empty()); + + let snap = store::add(&s, "t", "Write the RFC", CardPatch::default()) + .await + .unwrap(); + assert_eq!(snap.thread_id, "t"); + assert_eq!(snap.cards.len(), 1); + assert_eq!(snap.cards[0].title, "Write the RFC"); + assert!(snap.markdown.contains("Write the RFC")); + let id = snap.cards[0].id.clone(); + + let listed = store::list(&s, "t").await.unwrap(); + assert_eq!(listed.cards.len(), 1); + + let after = store::remove(&s, "t", &id).await.unwrap(); + assert!(after.cards.is_empty()); + assert!( + store::remove(&s, "t", &id).await.is_err(), + "unknown id errors" + ); + } + + #[tokio::test] + async fn add_rejects_empty_content_and_blank_thread() { + let s = store(); + assert!( + store::add(&s, "t", " ", CardPatch::default()) + .await + .is_err() + ); + assert!( + store::add(&s, " ", "x", CardPatch::default()) + .await + .is_err() + ); + } + + #[tokio::test] + async fn single_in_progress_invariant_is_enforced() { + let s = store(); + let a = store::add(&s, "t", "A", CardPatch::default()) + .await + .unwrap(); + let a_id = a.cards[0].id.clone(); + let b = store::add(&s, "t", "B", CardPatch::default()) + .await + .unwrap(); + let b_id = b.cards[1].id.clone(); + + store::update_status(&s, "t", &a_id, TaskCardStatus::InProgress) + .await + .unwrap(); + // A second in-progress card is rejected, not silently fixed. + let err = store::update_status(&s, "t", &b_id, TaskCardStatus::InProgress) + .await + .unwrap_err(); + assert!(format!("{err}").contains("in_progress")); + // The board still has exactly one in-progress card. + let listed = store::list(&s, "t").await.unwrap(); + let in_progress = listed + .cards + .iter() + .filter(|c| c.status == TaskCardStatus::InProgress) + .count(); + assert_eq!(in_progress, 1); + } + + #[tokio::test] + async fn replace_enforces_invariant() { + let s = store(); + let two_in_progress = vec![ + TaskBoardCard { + status: TaskCardStatus::InProgress, + ..TaskBoardCard::new("A") + }, + TaskBoardCard { + status: TaskCardStatus::InProgress, + ..TaskBoardCard::new("B") + }, + ]; + assert!(store::replace(&s, "t", two_in_progress).await.is_err()); + } + + #[tokio::test] + async fn decide_plan_only_from_awaiting_approval() { + let s = store(); + let snap = store::add( + &s, + "t", + "Gated work", + CardPatch { + status: Some(TaskCardStatus::AwaitingApproval), + ..Default::default() + }, + ) + .await + .unwrap(); + let id = snap.cards[0].id.clone(); + + let approved = store::decide_plan(&s, "t", &id, true).await.unwrap(); + assert_eq!(approved.cards[0].status, TaskCardStatus::Ready); + // A second decision on a now-Ready card errors (can't resurrect). + assert!(store::decide_plan(&s, "t", &id, false).await.is_err()); + } + + #[tokio::test] + async fn revise_plan_rejects_all_awaiting_and_is_lenient_when_empty() { + let s = store(); + store::add( + &s, + "t", + "Gated", + CardPatch { + status: Some(TaskCardStatus::AwaitingApproval), + ..Default::default() + }, + ) + .await + .unwrap(); + let after = store::revise_plan(&s, "t").await.unwrap(); + assert_eq!(after.cards[0].status, TaskCardStatus::Rejected); + // Nothing awaiting now → benign no-op. + let again = store::revise_plan(&s, "t").await.unwrap(); + assert_eq!(again.cards[0].status, TaskCardStatus::Rejected); + } + + #[tokio::test] + async fn claim_card_cas_accepts_then_rejects() { + let s = store(); + let snap = store::add( + &s, + "t", + "Runnable", + CardPatch { + status: Some(TaskCardStatus::Ready), + ..Default::default() + }, + ) + .await + .unwrap(); + let id = snap.cards[0].id.clone(); + + let claimed = store::claim_card( + &s, + "t", + &id, + &[TaskCardStatus::Ready], + TaskCardStatus::InProgress, + ) + .await + .unwrap(); + assert_eq!(claimed.status, TaskCardStatus::InProgress); + // A second claim expecting Ready now fails (already in progress). + assert!( + store::claim_card( + &s, + "t", + &id, + &[TaskCardStatus::Ready], + TaskCardStatus::InProgress + ) + .await + .is_err() + ); + } + + #[tokio::test] + async fn edit_leaves_unset_fields_untouched() { + let s = store(); + let snap = store::add( + &s, + "t", + "Task", + CardPatch { + objective: Some("keep me".into()), + ..Default::default() + }, + ) + .await + .unwrap(); + let id = snap.cards[0].id.clone(); + // Edit only the notes; objective is preserved. + let edited = store::edit( + &s, + "t", + &id, + CardPatch { + notes: Some("a note".into()), + ..Default::default() + }, + ) + .await + .unwrap(); + assert_eq!(edited.cards[0].objective.as_deref(), Some("keep me")); + assert_eq!(edited.cards[0].notes.as_deref(), Some("a note")); + } + + #[tokio::test] + async fn set_session_thread_links_then_clears() { + let s = store(); + let snap = store::add(&s, "t", "Task", CardPatch::default()) + .await + .unwrap(); + let id = snap.cards[0].id.clone(); + let linked = store::set_session_thread(&s, "t", &id, Some("thread-xyz".into())) + .await + .unwrap(); + assert_eq!( + linked.cards[0].session_thread_id.as_deref(), + Some("thread-xyz") + ); + let cleared = store::set_session_thread(&s, "t", &id, Some(" ".into())) + .await + .unwrap(); + assert_eq!(cleared.cards[0].session_thread_id, None); + } +} From acc2f54aac1af792dcfb46922c22b89ec08c2b9e Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 2 Jul 2026 10:30:42 -0700 Subject: [PATCH 08/10] graph::todos: add the todo multiplexer tool Add graph::todos::tool: a single `todo` harness Tool dispatching on the `op` field over add/edit/update_status/decide_plan/revise_plan/remove/replace/clear/ list, mirroring OpenHuman's todo tool. The board is bound to ToolExecutionContext.thread_id (never a tool arg); the bare call() entry point errors without a thread. Domain errors (unknown id, invariant violation) are surfaced to the model as tool errors rather than failing the run. Captures Arc at construction. Claude-Session: https://claude.ai/code/session_01NRL2bim9Gz3UN3SRtpeXEX --- src/graph/todos/mod.rs | 2 + src/graph/todos/test.rs | 165 +++++++++++++++++++ src/graph/todos/tool.rs | 345 ++++++++++++++++++++++++++++++++++++++++ 3 files changed, 512 insertions(+) create mode 100644 src/graph/todos/tool.rs diff --git a/src/graph/todos/mod.rs b/src/graph/todos/mod.rs index 704550d..04f7c0d 100644 --- a/src/graph/todos/mod.rs +++ b/src/graph/todos/mod.rs @@ -13,8 +13,10 @@ //! board is always `(Store, thread_id)`. pub mod store; +mod tool; mod types; +pub use tool::{TodoTool, register_todo_tools, todo_tools}; pub use types::{ CardPatch, TaskApprovalMode, TaskBoard, TaskBoardCard, TaskCardStatus, TodosSnapshot, normalise_board, parse_status, render_markdown, diff --git a/src/graph/todos/test.rs b/src/graph/todos/test.rs index 66ccb59..d3ecfce 100644 --- a/src/graph/todos/test.rs +++ b/src/graph/todos/test.rs @@ -352,3 +352,168 @@ mod store_tests { assert_eq!(cleared.cards[0].session_thread_id, None); } } + +mod tool_tests { + use std::sync::Arc; + + use serde_json::json; + + use super::super::tool::{TodoTool, todo_tools}; + use super::super::types::TaskCardStatus; + use crate::harness::events::EventSink; + use crate::harness::ids::{RunId, ThreadId}; + use crate::harness::store::{InMemoryStore, Store}; + use crate::harness::tool::{Tool, ToolCall, ToolExecutionContext}; + + fn store() -> Arc { + Arc::new(InMemoryStore::default()) + } + + fn ctx(thread_id: Option<&str>) -> ToolExecutionContext { + ToolExecutionContext { + run_id: RunId::new("run-1"), + thread_id: thread_id.map(ThreadId::new), + depth: 0, + max_turn_output_tokens: None, + events: EventSink::new(), + workspace: None, + } + } + + fn call(args: serde_json::Value) -> ToolCall { + ToolCall { + id: "c1".to_string(), + name: "todo".to_string(), + arguments: args, + } + } + + async fn run( + tool: &TodoTool, + thread: Option<&str>, + args: serde_json::Value, + ) -> crate::harness::tool::ToolResult { + Tool::<()>::call_with_context(tool, &(), call(args), ctx(thread)) + .await + .unwrap() + } + + #[test] + fn todo_tools_builds_a_single_tool() { + let tools = todo_tools(store()); + assert_eq!(tools.len(), 1); + assert_eq!(Tool::<()>::name(tools[0].as_ref()), "todo"); + } + + #[tokio::test] + async fn add_list_via_tool_persists_to_the_thread() { + let tool = TodoTool::new(store()); + let res = run( + &tool, + Some("t"), + json!({ "op": "add", "content": "Write tests" }), + ) + .await; + assert!(res.error.is_none(), "{res:?}"); + let raw = res.raw.unwrap(); + assert_eq!(raw["threadId"], "t"); + assert!(raw["markdown"].as_str().unwrap().contains("Write tests")); + + let res = run(&tool, Some("t"), json!({ "op": "list" })).await; + assert_eq!(res.raw.unwrap()["cards"].as_array().unwrap().len(), 1); + } + + #[tokio::test] + async fn add_then_update_status() { + let tool = TodoTool::new(store()); + let res = run(&tool, Some("t"), json!({ "op": "add", "content": "Task" })).await; + let id = res.raw.unwrap()["cards"][0]["id"] + .as_str() + .unwrap() + .to_string(); + let res = run( + &tool, + Some("t"), + json!({ "op": "update_status", "id": id, "status": "in_progress" }), + ) + .await; + assert_eq!(res.raw.unwrap()["cards"][0]["status"], "in_progress"); + } + + #[tokio::test] + async fn tool_requires_a_thread() { + let tool = TodoTool::new(store()); + // Bare call (no context). + let res = Tool::<()>::call(&tool, &(), call(json!({ "op": "list" }))) + .await + .unwrap(); + assert!(res.error.unwrap().contains("active thread")); + // Context without a thread id. + let res = run(&tool, None, json!({ "op": "list" })).await; + assert!(res.error.is_some()); + } + + #[tokio::test] + async fn unknown_op_and_missing_field_are_soft_errors() { + let tool = TodoTool::new(store()); + let res = run(&tool, Some("t"), json!({ "op": "frobnicate" })).await; + assert!(res.error.unwrap().contains("unknown op")); + let res = run(&tool, Some("t"), json!({ "op": "add" })).await; + assert!(res.error.unwrap().contains("content")); + } + + #[tokio::test] + async fn invariant_violation_is_a_soft_error() { + let tool = TodoTool::new(store()); + let a = run(&tool, Some("t"), json!({ "op": "add", "content": "A" })).await; + let a_id = a.raw.unwrap()["cards"][0]["id"] + .as_str() + .unwrap() + .to_string(); + run(&tool, Some("t"), json!({ "op": "add", "content": "B" })).await; + run( + &tool, + Some("t"), + json!({ "op": "update_status", "id": a_id, "status": "in_progress" }), + ) + .await; + // Second in-progress via replace/update is surfaced as an error, run continues. + let b = run(&tool, Some("t"), json!({ "op": "list" })).await; + let b_id = b.raw.unwrap()["cards"][1]["id"] + .as_str() + .unwrap() + .to_string(); + let res = run( + &tool, + Some("t"), + json!({ "op": "update_status", "id": b_id, "status": "in_progress" }), + ) + .await; + assert!(res.error.unwrap().contains("in_progress")); + } + + #[tokio::test] + async fn decide_plan_via_tool() { + let tool = TodoTool::new(store()); + let res = run( + &tool, + Some("t"), + json!({ "op": "add", "content": "Gated", "status": "awaiting_approval" }), + ) + .await; + let id = res.raw.unwrap()["cards"][0]["id"] + .as_str() + .unwrap() + .to_string(); + let res = run( + &tool, + Some("t"), + json!({ "op": "decide_plan", "id": id, "approve": true }), + ) + .await; + assert_eq!( + res.raw.unwrap()["cards"][0]["status"], + TaskCardStatus::Ready.as_str() + ); + } +} diff --git a/src/graph/todos/tool.rs b/src/graph/todos/tool.rs new file mode 100644 index 0000000..00d9839 --- /dev/null +++ b/src/graph/todos/tool.rs @@ -0,0 +1,345 @@ +//! `todo` — a single multiplexer harness [`Tool`] over the per-thread task +//! board. +//! +//! Dispatches on the `op` field so one tool exposes `add` / `edit` / +//! `update_status` / `decide_plan` / `revise_plan` / `remove` / `replace` / +//! `clear` / `list`. The board is bound to the caller's +//! [`ToolExecutionContext::thread_id`] (never a tool argument), so a model can't +//! address another thread's board; the bare [`Tool::call`] entry point (no +//! context) errors. Returns the updated cards plus a markdown rendering. + +use std::sync::Arc; + +use async_trait::async_trait; +use serde_json::{Value, json}; + +use super::store; +use super::types::{CardPatch, TaskApprovalMode, TaskBoardCard, parse_status}; +use crate::error::Result; +use crate::harness::store::Store; +use crate::harness::tool::{ + Tool, ToolCall, ToolExecutionContext, ToolPolicy, ToolRegistry, ToolResult, ToolSchema, + ToolSideEffects, +}; + +const TODO_TOOL_NAME: &str = "todo"; + +const TODO_DESCRIPTION: &str = "Maintain a visible plan for THIS thread: an ordered kanban board of \ + task cards that survives across turns. Use it for any request with several distinct steps: at \ + the start, `add` one card per step; keep exactly ONE card `in_progress` at a time; mark a card \ + `done` the moment it finishes; if a step is blocked, set it `blocked` with a `blocker`. `list` \ + to re-read the plan. The board is bound automatically to the current thread — do not pass a \ + thread id. Dispatch via `op`: `add` (content, status?, objective?, plan?, assignedAgent?, \ + allowedTools?, approvalMode?, acceptanceCriteria?, evidence?, notes?, blocker?), `edit` (id, \ + same optional fields), `update_status` (id, status), `decide_plan` (id, approve), \ + `revise_plan`, `remove` (id), `replace` (cards), `clear`, or `list`. Returns the updated cards \ + plus a markdown rendering."; + +/// A single harness [`Tool`] exposing the whole task-board CRUD surface, backed +/// by a [`Store`](crate::harness::store::Store). +pub struct TodoTool { + store: Arc, +} + +impl TodoTool { + /// Creates the `todo` tool backed by `store`. + pub fn new(store: Arc) -> Self { + Self { store } + } + + async fn dispatch(&self, thread_id: &str, args: &Value) -> Result { + let Some(op) = args.get("op").and_then(Value::as_str).map(str::trim) else { + return Ok(TodoOutcome::Error( + "missing required field `op`".to_string(), + )); + }; + let s = &self.store; + let snap = match op { + "add" => { + let Some(content) = required_str(args, "content") else { + return Ok(TodoOutcome::Error( + "missing required field `content`".to_string(), + )); + }; + match patch_from_args(args) { + Ok(patch) => store::add(s, thread_id, &content, patch).await, + Err(e) => return Ok(TodoOutcome::Error(e)), + } + } + "edit" => { + let Some(id) = required_str(args, "id") else { + return Ok(TodoOutcome::Error( + "missing required field `id`".to_string(), + )); + }; + match patch_from_args(args) { + Ok(mut patch) => { + patch.content = args + .get("content") + .and_then(Value::as_str) + .map(str::to_string); + store::edit(s, thread_id, &id, patch).await + } + Err(e) => return Ok(TodoOutcome::Error(e)), + } + } + "update_status" => { + let (Some(id), Some(status)) = + (required_str(args, "id"), required_str(args, "status")) + else { + return Ok(TodoOutcome::Error( + "update_status requires `id` and `status`".to_string(), + )); + }; + match parse_status(&status) { + Ok(status) => store::update_status(s, thread_id, &id, status).await, + Err(e) => return Ok(TodoOutcome::Error(e)), + } + } + "decide_plan" => { + let Some(id) = required_str(args, "id") else { + return Ok(TodoOutcome::Error( + "missing required field `id`".to_string(), + )); + }; + let Some(approve) = args.get("approve").and_then(Value::as_bool) else { + return Ok(TodoOutcome::Error( + "decide_plan requires a boolean `approve`".to_string(), + )); + }; + store::decide_plan(s, thread_id, &id, approve).await + } + "revise_plan" => store::revise_plan(s, thread_id).await, + "remove" => { + let Some(id) = required_str(args, "id") else { + return Ok(TodoOutcome::Error( + "missing required field `id`".to_string(), + )); + }; + store::remove(s, thread_id, &id).await + } + "replace" => { + let Some(cards_value) = args.get("cards") else { + return Ok(TodoOutcome::Error( + "missing `cards` for op=replace".to_string(), + )); + }; + match serde_json::from_value::>(cards_value.clone()) { + Ok(cards) => store::replace(s, thread_id, cards).await, + Err(e) => return Ok(TodoOutcome::Error(format!("invalid `cards`: {e}"))), + } + } + "clear" => store::clear(s, thread_id).await, + "list" => store::list(s, thread_id).await, + other => { + return Ok(TodoOutcome::Error(format!( + "unknown op '{other}' (expected add|edit|update_status|decide_plan|revise_plan|remove|replace|clear|list)" + ))); + } + }; + // A domain error (unknown id, invariant violation) is surfaced to the + // model rather than failing the whole run. + match snap { + Ok(snap) => Ok(TodoOutcome::Ok(json!({ + "threadId": snap.thread_id, + "cards": snap.cards, + "markdown": snap.markdown, + }))), + Err(e) => Ok(TodoOutcome::Error(e.to_string())), + } + } +} + +/// Internal dispatch outcome: a structured payload or a model-facing error. +enum TodoOutcome { + Ok(Value), + Error(String), +} + +fn required_str(args: &Value, key: &str) -> Option { + args.get(key) + .and_then(Value::as_str) + .map(str::trim) + .filter(|s| !s.is_empty()) + .map(str::to_string) +} + +fn optional_string(args: &Value, key: &str) -> Option { + args.get(key).and_then(Value::as_str).map(str::to_string) +} + +fn optional_string_array( + args: &Value, + key: &str, +) -> std::result::Result>, String> { + match args.get(key) { + None | Some(Value::Null) => Ok(None), + Some(Value::Array(items)) => items + .iter() + .map(|v| { + v.as_str() + .map(str::to_string) + .ok_or_else(|| format!("`{key}` items must be strings")) + }) + .collect::, _>>() + .map(Some), + Some(_) => Err(format!("`{key}` must be an array of strings")), + } +} + +fn patch_from_args(args: &Value) -> std::result::Result { + let status = match args.get("status").and_then(Value::as_str) { + Some(s) => Some(parse_status(s)?), + None => None, + }; + let approval_mode = match args.get("approvalMode") { + None => None, + Some(Value::Null) => Some(None), + Some(Value::String(s)) => match s.as_str() { + "required" => Some(Some(TaskApprovalMode::Required)), + "not_required" => Some(Some(TaskApprovalMode::NotRequired)), + other => { + return Err(format!( + "invalid approvalMode '{other}' (expected required|not_required|null)" + )); + } + }, + Some(_) => { + return Err( + "invalid approvalMode type (expected required|not_required|null)".to_string(), + ); + } + }; + Ok(CardPatch { + content: None, + status, + objective: optional_string(args, "objective"), + plan: optional_string_array(args, "plan")?, + assigned_agent: optional_string(args, "assignedAgent"), + allowed_tools: optional_string_array(args, "allowedTools")?, + approval_mode, + acceptance_criteria: optional_string_array(args, "acceptanceCriteria")?, + evidence: optional_string_array(args, "evidence")?, + notes: optional_string(args, "notes"), + blocker: optional_string(args, "blocker"), + source_metadata: None, + }) +} + +fn parameters_schema() -> Value { + json!({ + "type": "object", + "required": ["op"], + "properties": { + "op": { + "type": "string", + "enum": ["add", "edit", "update_status", "decide_plan", "revise_plan", "remove", "replace", "clear", "list"] + }, + "id": { "type": "string", "description": "Card id (required for edit/update_status/decide_plan/remove)." }, + "content": { "type": "string", "description": "Card title (required for add; optional for edit)." }, + "status": { + "type": "string", + "enum": ["todo", "awaiting_approval", "ready", "in_progress", "blocked", "done", "rejected"] + }, + "approve": { "type": "boolean", "description": "For op=decide_plan: approve (true) or reject (false)." }, + "notes": { "type": "string" }, + "blocker": { "type": "string" }, + "objective": { "type": "string", "description": "Desired outcome for this task." }, + "plan": { "type": "array", "items": { "type": "string" }, "description": "Ordered execution steps." }, + "assignedAgent": { "type": "string" }, + "allowedTools": { "type": "array", "items": { "type": "string" } }, + "approvalMode": { "type": ["string", "null"], "enum": ["required", "not_required", null] }, + "acceptanceCriteria": { "type": "array", "items": { "type": "string" } }, + "evidence": { "type": "array", "items": { "type": "string" } }, + "cards": { "type": "array", "items": { "type": "object" }, "description": "Full card list for op=replace." } + } + }) +} + +fn error_result(call_id: String, message: impl Into) -> ToolResult { + ToolResult { + call_id, + name: TODO_TOOL_NAME.to_string(), + content: String::new(), + raw: None, + error: Some(message.into()), + elapsed_ms: 0, + } +} + +/// Builds the `todo` tool backed by `store`. +pub fn todo_tools(store: Arc) -> Vec> { + vec![Arc::new(TodoTool::new(store))] +} + +/// Registers the `todo` tool into a tool registry. +pub fn register_todo_tools( + registry: &mut ToolRegistry, + store: Arc, +) -> &mut ToolRegistry { + registry.register(Arc::new(TodoTool::new(store))); + registry +} + +#[async_trait] +impl Tool for TodoTool { + fn name(&self) -> &str { + TODO_TOOL_NAME + } + + fn description(&self) -> &str { + TODO_DESCRIPTION + } + + fn schema(&self) -> ToolSchema { + ToolSchema { + name: TODO_TOOL_NAME.to_string(), + description: TODO_DESCRIPTION.to_string(), + parameters: parameters_schema(), + format: Default::default(), + } + } + + fn policy(&self) -> ToolPolicy { + ToolPolicy { + classified: true, + side_effects: ToolSideEffects { + read_only: false, + ..Default::default() + }, + ..Default::default() + } + } + + async fn call(&self, _state: &State, call: ToolCall) -> Result { + Ok(error_result( + call.id, + "todo tool requires an active thread (no thread_id in tool context)", + )) + } + + async fn call_with_context( + &self, + _state: &State, + call: ToolCall, + context: ToolExecutionContext, + ) -> Result { + let Some(thread_id) = context.thread_id.as_ref() else { + return Ok(error_result( + call.id, + "todo tool requires an active thread (no thread_id in tool context)", + )); + }; + match self.dispatch(thread_id.as_str(), &call.arguments).await? { + TodoOutcome::Ok(payload) => Ok(ToolResult { + call_id: call.id, + name: TODO_TOOL_NAME.to_string(), + content: payload.to_string(), + raw: Some(payload), + error: None, + elapsed_ms: 0, + }), + TodoOutcome::Error(message) => Ok(error_result(call.id, message)), + } + } +} From b9a00d883e9cf6dd1345a785855f265a4f53eafc Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 2 Jul 2026 10:31:52 -0700 Subject: [PATCH 09/10] graph::todos: wire crate-root re-exports + module README Re-export the task-board surface at the crate root (todo_store CRUD alias, TodoTool, todo_tools, register_todo_tools, TaskBoard, TaskBoardCard, ...) and note the primitive in the graph-surface docs. Add src/graph/todos/README.md covering the data model, Store key scheme, invariants (single-in-progress, decide_plan guard, CAS claim_card), and the tool surface. Claude-Session: https://claude.ai/code/session_01NRL2bim9Gz3UN3SRtpeXEX --- src/graph/mod.rs | 8 ++-- src/graph/todos/README.md | 86 +++++++++++++++++++++++++++++++++++++++ src/lib.rs | 14 ++++++- 3 files changed, 103 insertions(+), 5 deletions(-) create mode 100644 src/graph/todos/README.md diff --git a/src/graph/mod.rs b/src/graph/mod.rs index 3ef39d9..f369847 100644 --- a/src/graph/mod.rs +++ b/src/graph/mod.rs @@ -13,7 +13,9 @@ //! interrupts ([`command`]), a builder/compile contract ([`builder`]), a //! superstep executor ([`compiled`]), checkpointing ([`checkpoint`]), //! streaming/events ([`stream`]), run-status snapshots ([`status`]), graph -//! export/visualization ([`export`]), and subgraph embedding ([`subgraph`]). +//! export/visualization ([`export`]), subgraph embedding ([`subgraph`]), and +//! per-thread productivity primitives — a durable goal ([`goals`]) and a kanban +//! task board ([`todos`]) — exposed as harness tools. //! //! Each concern lives in its own submodule with `types.rs` (definitions), //! `mod.rs` (implementations), and `test.rs` (unit tests). @@ -96,6 +98,6 @@ pub use testkit::{ scripted_route_node, scripted_update_node, subagent_fake_node, subgraph_test_node, }; pub use todos::{ - CardPatch, TaskApprovalMode, TaskBoard, TaskBoardCard, TaskCardStatus, TodosSnapshot, - normalise_board, parse_status, render_markdown, + CardPatch, TaskApprovalMode, TaskBoard, TaskBoardCard, TaskCardStatus, TodoTool, TodosSnapshot, + normalise_board, parse_status, register_todo_tools, render_markdown, todo_tools, }; diff --git a/src/graph/todos/README.md b/src/graph/todos/README.md new file mode 100644 index 0000000..d6e789e --- /dev/null +++ b/src/graph/todos/README.md @@ -0,0 +1,86 @@ +# graph::todos + +A per-thread **task board** (kanban todos): an ordered *list* of task cards per +thread. Ported from OpenHuman's task board / `todos` modules and re-hosted on +the TinyAgents graph runtime — provider-neutral, offline-testable, with no +app-specific coupling (no progress events, RPC envelopes, or scratch fallback). + +Distinct from [`graph::goals`](../goals) (a *single* durable objective per +thread): a goal is the completion contract the graph pursues; the board holds +the concrete work items. + +## Data model (`types.rs`) + +- `TaskCardStatus { Todo, AwaitingApproval, Ready, InProgress, Blocked, Done, + Rejected }` and `TaskApprovalMode { Required, NotRequired }` (each `as_str`). +- `TaskBoardCard { id, title, status, objective, plan, assigned_agent, + allowed_tools, approval_mode, acceptance_criteria, evidence, notes, blocker, + session_thread_id, source_metadata, order, updated_at }` (serde `camelCase`). +- `TaskBoard { thread_id, cards, updated_at }`. +- `CardPatch` — optional `add`/`edit` fields; `approval_mode` is doubly-optional + (`None` untouched, `Some(None)` clears, `Some(Some(_))` sets). +- `TodosSnapshot { thread_id, cards, markdown }` — every CRUD op returns one. +- `parse_status` (accepts aliases like `pending`→`Todo`, `approved`→`Ready`), + `render_markdown` (`[ ]`/`[x]`/`[~]`/`[!]`/`[?]`/`[-]` markers + indented + metadata), `normalise_board` (id generation, trimming, empty-title drop, + blocker-from-notes, order recompute). + +`id`s are `task-` (via `next_seq()`); `updated_at` is unix-epoch millis as a +string (dependency-free, no `chrono`). + +## Persistence (`store.rs`) + +One serialized `TaskBoard` per thread under the `graph.todos` namespace of a +`crate::harness::store::Store`, keyed by `hex(thread_id)`. Every mutation runs +`load → mutate → normalise → put` under a per-thread async mutex — atomic within +one process (same single-process caveat as `graph::goals::store`). Ops: +`add` / `edit` / `update_status` / `decide_plan` / `revise_plan` / `remove` / +`replace` / `clear` / `list` / `claim_card` / `set_session_thread`. + +Invariants preserved from OpenHuman: + +- **Single in-progress:** at most one card may be `InProgress`; a violation is a + `Validation` error on `add` / `edit` / `replace` / `claim_card` — never + silently fixed. +- `decide_plan` only transitions an `AwaitingApproval` card (approve → `Ready`, + reject → `Rejected`); a stale decision errors. +- `revise_plan` rejects every `AwaitingApproval` card and is a lenient no-op + when none is awaiting. +- `claim_card` is an atomic compare-and-set: it transitions a card from one of + `expected` to `target` under the lock, rejecting the claim otherwise. + +## Tool (`tool.rs`) + +`TodoTool` is a single multiplexer harness `Tool` dispatching on an `op` field +(`add`/`edit`/`update_status`/`decide_plan`/`revise_plan`/`remove`/`replace`/ +`clear`/`list`). Build it with `todo_tools(store)` or `register_todo_tools`. The +target thread comes from `ToolExecutionContext::thread_id` (never a tool +argument); the bare `Tool::call` entry point errors without a thread. Domain +errors (unknown id, invariant violation) are surfaced to the model as tool +errors rather than failing the run. + +## Example + +```rust,ignore +use std::sync::Arc; +use tinyagents::{TodoTool, todo_store}; +use tinyagents::harness::store::{InMemoryStore, Store}; + +let store: Arc = Arc::new(InMemoryStore::default()); + +// Programmatic: +let snap = todo_store::add(&store, "thread-1", "Write the RFC", Default::default()).await?; +println!("{}", snap.markdown); + +// Or register the `todo` tool for a model to drive: +let tool = TodoTool::new(store.clone()); +``` + +## Files + +| File | Role | +| --- | --- | +| `types.rs` | Card/board model, `parse_status`, `render_markdown`, `normalise_board`, `CardPatch`, `TodosSnapshot`. | +| `store.rs` | `Store`-backed CRUD, per-thread RMW lock, single-in-progress invariant, CAS `claim_card`. | +| `tool.rs` | The `todo` multiplexer tool. | +| `test.rs` | Unit tests (types, store, tool). | diff --git a/src/lib.rs b/src/lib.rs index e8178e8..00906a9 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -26,8 +26,9 @@ //! 2. **Graph runtime** ([`graph`]) — LangGraph-style durable typed state //! graphs: [`START`]/[`END`], nodes, conditional routing, [`Command`]s, //! fan-out, reducers/channels, [`Checkpoint`]s, [`Interrupt`]s, subgraphs, -//! streaming, topology export, and a per-thread durable [`ThreadGoal`] with -//! graph-native continuation, exposed as harness tools. +//! streaming, topology export, and per-thread productivity primitives — a +//! durable [`ThreadGoal`] with graph-native continuation and a +//! [`TaskBoard`] kanban — exposed as harness tools. //! 3. **Registry** ([`registry`]) — a named capability catalog (models, tools, //! agents, graphs, stores, middleware, policy) that `.rag`/`.ragsh` bind by //! name. @@ -195,6 +196,15 @@ pub use graph::{ run_continuation_tick, }; +// --- Graph: per-thread task board (kanban todos) --- +// `todo_store` is the programmatic CRUD surface (add/edit/claim_card/...); the +// tool and data model are re-exported flat for discoverability. +pub use graph::todos::store as todo_store; +pub use graph::{ + CardPatch, TaskApprovalMode, TaskBoard, TaskBoardCard, TaskCardStatus, TodoTool, TodosSnapshot, + normalise_board, parse_status, register_todo_tools, render_markdown, todo_tools, +}; + // --- Graph: parallel map/reduce helper --- pub use graph::parallel::{ FailurePolicy, ItemOutcome, ParallelOptions, ParallelOutcome, map_reduce, From a891732356350c55b568db9cc0ab80896f65847e Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 2 Jul 2026 10:38:43 -0700 Subject: [PATCH 10/10] tests+docs: e2e goal loop + todo tool; graph docs/modules Add integration tests exercising the public crate surface end-to-end: tests/e2e_graph_goals.rs drives a self-driving goal loop (goal_gate_node + goal_store on a shared InMemoryStore) until the work node completes the goal; tests/e2e_graph_todos.rs drives the `todo` tool through a MockModel agent loop and asserts the board persists under the run's thread id. Add docs/modules/graph/{goals,todos}.md and link them from the graph module index. Claude-Session: https://claude.ai/code/session_01NRL2bim9Gz3UN3SRtpeXEX --- docs/modules/graph/README.md | 2 + docs/modules/graph/goals.md | 68 +++++++++++++++++++++++ docs/modules/graph/todos.md | 55 +++++++++++++++++++ tests/e2e_graph_goals.rs | 83 ++++++++++++++++++++++++++++ tests/e2e_graph_todos.rs | 103 +++++++++++++++++++++++++++++++++++ 5 files changed, 311 insertions(+) create mode 100644 docs/modules/graph/goals.md create mode 100644 docs/modules/graph/todos.md create mode 100644 tests/e2e_graph_goals.rs create mode 100644 tests/e2e_graph_todos.rs diff --git a/docs/modules/graph/README.md b/docs/modules/graph/README.md index eb1df04..6dc4623 100644 --- a/docs/modules/graph/README.md +++ b/docs/modules/graph/README.md @@ -146,4 +146,6 @@ topology or executable code. - [Sub-agents, recursion, and depth tracking](subagents-recursion.md) - [Memory and stores boundary](memory-boundary.md) - [Visualization, introspection, and testkit](visualization-testkit.md) +- [Per-thread goal and graph-native continuation](goals.md) +- [Per-thread task board (kanban todos)](todos.md) - [Implementation milestones](milestones.md) diff --git a/docs/modules/graph/goals.md b/docs/modules/graph/goals.md new file mode 100644 index 0000000..53afdec --- /dev/null +++ b/docs/modules/graph/goals.md @@ -0,0 +1,68 @@ +# Per-thread goal and graph-native continuation + +`graph::goals` gives a graph a single durable **objective per thread** — a +"completion contract" carried across supersteps, interrupts, and resumes — plus +a graph-native way to keep working it. It is a provider-neutral port of +OpenHuman's `thread_goals`, re-hosted on the harness `Store` and driven off the +graph runtime rather than an out-of-band heartbeat. + +See the source module README at `src/graph/goals/README.md` for the full public +surface; this spec captures the design contract. + +## Model + +- Exactly **one** `ThreadGoal` per thread, keyed by `thread_id`. +- `ThreadGoalStatus`: `Active` → the graph may work it and auto-continue; + `Paused` (host control); `BudgetLimited` (accounting reached the token cap); + `Complete` (model-confirmed success). Ownership is asymmetric: a model + creates/replaces and completes a goal; pause/budget-limit are system-driven. +- Optional `token_budget`; `account_usage` folds usage and flips an active goal + to `BudgetLimited` at the cap. + +## Persistence + +One serialized `ThreadGoal` per thread in the `graph.goals` namespace of a +`harness::store::Store`, keyed by `hex(thread_id)`. The `Store` trait has no CAS, +so each mutation runs `load → mutate → put` under a per-thread async mutex — +atomic within one process. A `goal_id` compare-and-set guard drops stale +accounting from a replaced goal. Cross-process lost-update is a documented +limitation (a future `Store::compare_and_swap` is the clean fix). + +## Tools + +`GoalTool` / `GoalToolKind` expose `goal_get`, `goal_set`, `goal_complete` as +the default model-facing set (`goal_tools` / `register_goal_tools`); +`goal_pause` / `goal_resume` / `goal_clear` are host controls. The target thread +comes from `ToolExecutionContext::thread_id` — never a tool argument — so a +model can't address another thread's goal. + +## Continuation (heartbeat → graph) + +OpenHuman's idle heartbeat becomes three graph-native primitives: + +1. **`goal_gate_node`** (primary) — a command-routing node forming a self-driving + bounded loop. Wired `work_node -> gate` with the gate a command node whose + destinations are `[work_node, END]`, it folds each iteration's `GoalProgress` + via `account_usage` and routes back to `work_node` while the goal is Active + and under budget, else to `END`. The graph `recursion_limit` is the hard + backstop; a zero-progress iteration sets a one-shot suppression and stops. +2. **`run_continuation_tick`** — a faithful heartbeat port for callers that have + an external scheduler: selects idle, active, non-suppressed goals (oldest + first, `max_per_tick`) and runs one turn each through a caller closure. +3. **`note_user_turn`** — clears the one-shot suppression and reactivates a + paused goal on a user-initiated run. A loop iteration never clears its own + suppression, so user-vs-continuation is distinguished structurally. + +### Token accounting boundary + +The graph runtime is provider-neutral and does not meter tokens per node, so +accounting is **explicit**: a work node (typically a `subagent_node`) writes what +it spent into `State`, and the caller's `progress` / `run_turn` closure reports +it. `made_progress == false` is the graph analogue of OpenHuman's "the turn +produced no tool calls". + +## Testing + +Unit tests in `src/graph/goals/test.rs` (types, store, tools, and the gate loop +on `InMemoryStore`); an end-to-end self-driving loop in +`tests/e2e_graph_goals.rs`. diff --git a/docs/modules/graph/todos.md b/docs/modules/graph/todos.md new file mode 100644 index 0000000..b921bdf --- /dev/null +++ b/docs/modules/graph/todos.md @@ -0,0 +1,55 @@ +# Per-thread task board (kanban todos) + +`graph::todos` gives a graph a per-thread **task board**: an ordered *list* of +task cards with a small kanban lifecycle. It is the concrete-work-items +counterpart to the single-objective [`graph::goals`](goals.md), a +provider-neutral port of OpenHuman's task board / `todos` modules. + +See the source module README at `src/graph/todos/README.md` for the full public +surface; this spec captures the design contract. + +## Model + +- `TaskBoardCard { id, title, status, objective, plan, assigned_agent, + allowed_tools, approval_mode, acceptance_criteria, evidence, notes, blocker, + session_thread_id, source_metadata, order, updated_at }`. +- `TaskCardStatus`: `Todo`, `AwaitingApproval`, `Ready`, `InProgress`, + `Blocked`, `Done`, `Rejected`. `TaskApprovalMode`: `Required`, `NotRequired`. +- `TaskBoard { thread_id, cards, updated_at }`; `TodosSnapshot` (cards + + markdown) is returned by every CRUD op. +- `render_markdown` renders GitHub-flavored markers + (`[ ]`/`[x]`/`[~]`/`[!]`/`[?]`/`[-]`) with indented metadata; `parse_status` + accepts aliases; `normalise_board` generates ids, trims, drops empty-title + cards, backfills a blocker from notes, and recomputes order. + +## Persistence + +One serialized `TaskBoard` per thread in the `graph.todos` namespace of a +`harness::store::Store`, keyed by `hex(thread_id)`. Each mutation runs +`load → mutate → normalise → put` under a per-thread async mutex (atomic within +one process, same caveat as `graph::goals`). + +### Invariants + +- **Single in-progress:** at most one card may be `InProgress`; a violation is a + `Validation` error on `add` / `edit` / `replace` / `claim_card` — never + silently fixed. +- `decide_plan` only transitions an `AwaitingApproval` card; a stale decision + errors. `revise_plan` rejects all awaiting cards and is a lenient no-op when + none awaits. +- `claim_card` is an atomic compare-and-set: transition from one of `expected` + to `target` under the lock, else reject. + +## Tool + +`TodoTool` is a single multiplexer harness `Tool` dispatching on an `op` field +(`add`/`edit`/`update_status`/`decide_plan`/`revise_plan`/`remove`/`replace`/ +`clear`/`list`), built with `todo_tools` / `register_todo_tools`. The board is +bound to `ToolExecutionContext::thread_id` (never a tool argument). Domain errors +(unknown id, invariant violation) are surfaced to the model as tool errors +rather than failing the run. + +## Testing + +Unit tests in `src/graph/todos/test.rs` (types, store invariants, tool); an +end-to-end model-driven tool run in `tests/e2e_graph_todos.rs`. diff --git a/tests/e2e_graph_goals.rs b/tests/e2e_graph_goals.rs new file mode 100644 index 0000000..66fb2ee --- /dev/null +++ b/tests/e2e_graph_goals.rs @@ -0,0 +1,83 @@ +//! End-to-end coverage for the per-thread goal (`graph::goals`) exercised +//! through the public crate surface: a self-driving graph loop wired with +//! `goal_gate_node` keeps re-running a work node while the thread's goal is +//! active, and stops when the work node marks the goal complete. + +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; + +use tinyagents::graph::command::NodeResult; +use tinyagents::graph::{NodeContext, NodeFuture}; +use tinyagents::harness::store::{InMemoryStore, Store}; +use tinyagents::{GoalProgress, GraphBuilder, ThreadGoalStatus, goal_gate_node, goal_store}; + +/// State overwritten by each work iteration. +#[derive(Clone, Debug, Default, PartialEq)] +struct LoopState { + iters: usize, +} + +#[tokio::test] +async fn self_driving_goal_loop_runs_until_complete() { + let store: Arc = Arc::new(InMemoryStore::default()); + goal_store::set(&store, "thread-goal", "process every item", None) + .await + .expect("set goal"); + + // The work node advances a counter and, on the third iteration, marks the + // goal complete (the graph analogue of a model calling `goal_complete`). + let counter = Arc::new(AtomicUsize::new(0)); + let work_store = store.clone(); + let work_node = move |_state: LoopState, _ctx: NodeContext| { + let counter = counter.clone(); + let work_store = work_store.clone(); + Box::pin(async move { + let n = counter.fetch_add(1, Ordering::SeqCst) + 1; + if n >= 3 { + goal_store::complete(&work_store, "thread-goal") + .await + .expect("complete goal"); + } + Ok(NodeResult::Update(LoopState { iters: n })) + }) as NodeFuture + }; + + // The gate accounts a fixed cost per iteration and reports progress so the + // loop is only stopped by the goal reaching `Complete`. + let gate = goal_gate_node::(store.clone(), "work", |_s: &LoopState| { + GoalProgress { + tokens_used: 10, + elapsed_secs: 1, + made_progress: true, + } + }); + + let graph = GraphBuilder::::overwrite() + .with_recursion_limit(64) + .add_node("work", work_node) + .add_node("gate", gate) + .set_entry("work") + .add_edge("work", "gate") + .with_command_destinations("gate", ["work", tinyagents::END]) + .compile() + .expect("graph compiles"); + + let exec = graph + .run_with_thread("thread-goal", LoopState::default()) + .await + .expect("graph runs to completion"); + + assert_eq!(exec.state.iters, 3, "loops until the goal is completed"); + + let goal = goal_store::get(&store, "thread-goal") + .await + .expect("load goal") + .expect("goal exists"); + assert_eq!(goal.status, ThreadGoalStatus::Complete); + // Usage was accounted across the iterations that ran through the gate. + assert!( + goal.tokens_used >= 20, + "usage accrued: {}", + goal.tokens_used + ); +} diff --git a/tests/e2e_graph_todos.rs b/tests/e2e_graph_todos.rs new file mode 100644 index 0000000..59971da --- /dev/null +++ b/tests/e2e_graph_todos.rs @@ -0,0 +1,103 @@ +//! End-to-end coverage for the per-thread task board (`graph::todos`) exercised +//! through the public crate surface: a `MockModel`-driven agent loop calls the +//! `todo` tool, and the board persists on a shared `Store` addressed by the +//! run's thread id. + +use std::sync::Arc; + +use serde_json::json; + +use tinyagents::harness::context::RunConfig; +use tinyagents::harness::message::{AssistantMessage, ContentBlock, Message}; +use tinyagents::harness::model::ModelResponse; +use tinyagents::harness::providers::MockModel; +use tinyagents::harness::runtime::AgentHarness; +use tinyagents::harness::store::{InMemoryStore, Store}; +use tinyagents::harness::tool::ToolCall; +use tinyagents::harness::usage::Usage; +use tinyagents::{TodoTool, todo_store}; + +fn tool_call_response(id: &str, name: &str, arguments: serde_json::Value) -> ModelResponse { + ModelResponse { + message: AssistantMessage { + id: Some(format!("msg-{id}")), + content: Vec::new(), + tool_calls: vec![ToolCall::new(id, name, arguments)], + usage: Some(Usage::new(7, 3)), + }, + usage: Some(Usage::new(7, 3)), + finish_reason: Some("tool_calls".to_string()), + raw: None, + resolved_model: None, + } +} + +fn text_response(text: &str) -> ModelResponse { + ModelResponse { + message: AssistantMessage { + id: None, + content: vec![ContentBlock::Text(text.to_string())], + tool_calls: Vec::new(), + usage: Some(Usage::new(4, 2)), + }, + usage: Some(Usage::new(4, 2)), + finish_reason: Some("stop".to_string()), + raw: None, + resolved_model: None, + } +} + +#[tokio::test] +async fn model_drives_the_todo_tool_and_board_persists_to_the_thread() { + let store: Arc = Arc::new(InMemoryStore::default()); + + let mut harness: AgentHarness<()> = AgentHarness::new(); + harness + .register_model( + "mock", + Arc::new(MockModel::with_responses(vec![ + // First turn: add a card via the `todo` tool. + tool_call_response( + "call-1", + "todo", + json!({ "op": "add", "content": "Write the integration test" }), + ), + // Second turn: mark it in progress. + tool_call_response("call-2", "todo", json!({ "op": "list" })), + text_response("done"), + ])), + ) + .set_default_model("mock") + .register_tool(Arc::new(TodoTool::new(store.clone()))); + + let run = harness + .invoke( + &(), + (), + RunConfig::new("run").with_thread("thread-e2e"), + vec![Message::user("track this work")], + ) + .await + .expect("agent run succeeds"); + + assert_eq!(run.tool_calls, 2, "both todo tool calls executed"); + + // The board persisted under the run's thread id, reachable via the public + // programmatic surface. + let snapshot = todo_store::list(&store, "thread-e2e") + .await + .expect("board lists"); + assert_eq!(snapshot.cards.len(), 1); + assert_eq!(snapshot.cards[0].title, "Write the integration test"); + assert!( + snapshot.markdown.contains("Write the integration test"), + "markdown renders the card: {}", + snapshot.markdown + ); + + // A different thread has its own (empty) board. + let other = todo_store::list(&store, "other-thread") + .await + .expect("other board lists"); + assert!(other.cards.is_empty()); +}