From 149137a71f6780512261dca7fad9b6c1e494020e Mon Sep 17 00:00:00 2001 From: Steven Enamakel <31011319+senamakel@users.noreply.github.com> Date: Tue, 28 Jul 2026 11:09:19 +0300 Subject: [PATCH] refactor(goals): consolidate thread goals under tinyagents (#5235) --- src/core/runtime/services.rs | 6 +- .../agent/harness/session/turn/core.rs | 4 +- src/openhuman/thread_goals/continuation.rs | 4 +- src/openhuman/thread_goals/crate_adapter.rs | 538 ------------------ src/openhuman/thread_goals/migration.rs | 177 ++++++ src/openhuman/thread_goals/mod.rs | 34 +- src/openhuman/thread_goals/ops.rs | 2 +- src/openhuman/thread_goals/runtime.rs | 105 +--- src/openhuman/thread_goals/schemas.rs | 2 +- src/openhuman/thread_goals/store.rs | 44 +- src/openhuman/thread_goals/tools.rs | 4 +- src/openhuman/thread_goals/types.rs | 151 ----- 12 files changed, 219 insertions(+), 852 deletions(-) delete mode 100644 src/openhuman/thread_goals/crate_adapter.rs create mode 100644 src/openhuman/thread_goals/migration.rs delete mode 100644 src/openhuman/thread_goals/types.rs diff --git a/src/core/runtime/services.rs b/src/core/runtime/services.rs index fbc303fa4..ea746fcbc 100644 --- a/src/core/runtime/services.rs +++ b/src/core/runtime/services.rs @@ -367,10 +367,8 @@ async fn run_legacy_migrations(config: &Config) { // // Both copies are idempotent and must run for each workspace so an // in-process restart with a different workspace migrates that workspace. - match crate::openhuman::thread_goals::crate_adapter::migrate_legacy_goals_into_crate_store( - &config.workspace_dir, - ) - .await + match crate::openhuman::thread_goals::migration::migrate_legacy_goals(&config.workspace_dir) + .await { Ok(report) if report.total > 0 => { log::info!( diff --git a/src/openhuman/agent/harness/session/turn/core.rs b/src/openhuman/agent/harness/session/turn/core.rs index a09b7776f..f8b0b503e 100644 --- a/src/openhuman/agent/harness/session/turn/core.rs +++ b/src/openhuman/agent/harness/session/turn/core.rs @@ -792,9 +792,7 @@ impl Agent { } }; if let Some(ref goal) = active_goal { - if let Some(block) = - crate::openhuman::thread_goals::runtime::active_goal_context_block(goal) - { + if let Some(block) = tinyagents::graph::goals::active_goal_context_block(goal) { log::info!( "[thread_goals] injecting active_goal block status={} budget={:?} ({} chars)", goal.status.as_str(), diff --git a/src/openhuman/thread_goals/continuation.rs b/src/openhuman/thread_goals/continuation.rs index 6b302599e..299e437d3 100644 --- a/src/openhuman/thread_goals/continuation.rs +++ b/src/openhuman/thread_goals/continuation.rs @@ -1,4 +1,4 @@ -//! Heartbeat-driven autonomous continuation of idle thread goals (Codex's +//! OpenHuman heartbeat adapter for tinyagents thread-goal continuation. //! `MaybeContinueIfIdle`). //! //! When a thread carries an **active** goal and goes idle — no in-flight turn @@ -28,7 +28,7 @@ use std::sync::OnceLock; use tokio::sync::Semaphore; use super::store; -use super::types::{ThreadGoal, ThreadGoalStatus}; +use super::{ThreadGoal, ThreadGoalStatus}; use crate::openhuman::agent::turn_origin::{with_origin, AgentTurnOrigin, TrustedAutomationSource}; use crate::openhuman::agent::Agent; use crate::openhuman::config::Config; diff --git a/src/openhuman/thread_goals/crate_adapter.rs b/src/openhuman/thread_goals/crate_adapter.rs deleted file mode 100644 index e88c40409..000000000 --- a/src/openhuman/thread_goals/crate_adapter.rs +++ /dev/null @@ -1,538 +0,0 @@ -//! Adapter seam onto the tinyagents `graph::goals` crate store (issue #4249). -//! -//! The crate store is now **authoritative** for thread goals — [`super::store`] -//! delegates every operation to it. This module supplies the conversion helpers -//! (local ↔ crate [`ThreadGoal`]/[`ThreadGoalStatus`]), the store-handle opener -//! ([`crate_goals_store`]), raw mirror helpers used by migration tests, and the idempotent -//! [`migrate_legacy_goals_into_crate_store`] boot helper that copies goals left -//! in the retired `{workspace}/thread_goals/` file-JSON tree only when the crate -//! store has no value for that thread, then removes the retired row. -//! -//! Persistence target: the crate [`Store`] rooted at the shared workspace KV -//! tree (`{workspace}/tinyagents_store/kv`), namespace [`GOALS_NAMESPACE`] -//! (`graph.goals`), keyed by `hex(thread_id)` — byte-for-byte the key the -//! crate's own `graph::goals::store` computes. -//! -//! # Single-writer constraint -//! -//! The crate `Store` has **no compare-and-set and no cross-key transaction**; -//! its per-thread atomicity is a *process-local* async mutex. This is acceptable -//! because **the OpenHuman core is the single writer** of thread goals — RPC -//! handlers, agent tools, and the heartbeat continuation runtime all run inside -//! one core process. Do not add a second mutating writer (a sidecar, a second -//! core, a cron in another process) without introducing a real CAS first. - -use std::path::Path; -use std::sync::Arc; - -use tinyagents::graph::goals::store::GOALS_NAMESPACE; -use tinyagents::graph::goals::{ThreadGoal as CrateThreadGoal, ThreadGoalStatus as CrateStatus}; -use tinyagents::harness::store::Store; - -use super::types::{ThreadGoal, ThreadGoalStatus}; -use crate::openhuman::session_import::ops::open_session_stores; - -const LEGACY_GOALS_DIR: &str = "thread_goals"; -const LEGACY_GOALS_EXTENSION: &str = "json"; - -/// Open the crate [`Store`] handle used for the goals mirror, rooted at the -/// shared workspace KV tree (`{workspace}/tinyagents_store/kv`). Same layout the -/// 04-sessions journal + status store use, so everything lives under one tree. -pub(crate) fn crate_goals_store(workspace_dir: &Path) -> Arc { - Arc::new(open_session_stores(workspace_dir).kv) -} - -/// The crate store key for a thread's goal: lowercase hex of the (trimmed) -/// thread-id bytes. This MUST match the crate's private `graph::goals::store` -/// key function exactly so the crate reader resolves our mirrored value. -fn goal_key(thread_id: &str) -> String { - thread_id - .trim() - .as_bytes() - .iter() - .map(|b| format!("{b:02x}")) - .collect() -} - -fn legacy_goal_path(workspace_dir: &Path, thread_id: &str) -> Result { - let thread_id = thread_id.trim(); - if thread_id.is_empty() { - return Err("invalid thread goal thread_id: empty or whitespace".to_string()); - } - Ok(workspace_dir.join(LEGACY_GOALS_DIR).join(format!( - "{}.{LEGACY_GOALS_EXTENSION}", - hex::encode(thread_id.as_bytes()) - ))) -} - -/// Remove a retired file-store row for `thread_id`, if present. -/// -/// Clears call this before deleting the authoritative crate row so a failed -/// legacy cleanup can never leave an absent crate row that migration would -/// resurrect on the next boot. -pub(crate) async fn delete_legacy_goal_file( - workspace_dir: &Path, - thread_id: &str, -) -> Result { - let path = legacy_goal_path(workspace_dir, thread_id)?; - match tokio::fs::remove_file(&path).await { - Ok(()) => Ok(true), - Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(false), - Err(e) => Err(format!("delete legacy thread goal {}: {e}", path.display())), - } -} - -/// Map a legacy [`ThreadGoalStatus`] onto the crate [`CrateStatus`]. The two -/// enums are 1:1 (Active/Paused/BudgetLimited/Complete) — this is the mapping -/// the parity tests pin. -pub(crate) fn to_crate_status(status: ThreadGoalStatus) -> CrateStatus { - match status { - ThreadGoalStatus::Active => CrateStatus::Active, - ThreadGoalStatus::Paused => CrateStatus::Paused, - ThreadGoalStatus::BudgetLimited => CrateStatus::BudgetLimited, - ThreadGoalStatus::Complete => CrateStatus::Complete, - } -} - -/// Map a crate [`CrateStatus`] back onto the legacy [`ThreadGoalStatus`] (the -/// inverse of [`to_crate_status`]). -pub(crate) fn from_crate_status(status: CrateStatus) -> ThreadGoalStatus { - match status { - CrateStatus::Active => ThreadGoalStatus::Active, - CrateStatus::Paused => ThreadGoalStatus::Paused, - CrateStatus::BudgetLimited => ThreadGoalStatus::BudgetLimited, - CrateStatus::Complete => ThreadGoalStatus::Complete, - } -} - -/// Convert a legacy [`ThreadGoal`] into the crate [`CrateThreadGoal`], -/// preserving every field verbatim (id, objective, status, budget/usage -/// counters, timestamps, continuation flag). A **faithful** projection — no -/// re-minting, no counter reset. -pub(crate) fn to_crate_goal(goal: &ThreadGoal) -> CrateThreadGoal { - CrateThreadGoal { - thread_id: goal.thread_id.clone(), - goal_id: goal.goal_id.clone(), - objective: goal.objective.clone(), - status: to_crate_status(goal.status), - token_budget: goal.token_budget, - tokens_used: goal.tokens_used, - time_used_seconds: goal.time_used_seconds, - created_at_ms: goal.created_at_ms, - updated_at_ms: goal.updated_at_ms, - continuation_suppressed: goal.continuation_suppressed, - } -} - -/// Convert a crate [`CrateThreadGoal`] back into a legacy [`ThreadGoal`] (the -/// inverse of [`to_crate_goal`]), used by the store adapter to return -/// local goals from the crate store. -pub(crate) fn from_crate_goal(goal: &CrateThreadGoal) -> ThreadGoal { - ThreadGoal { - thread_id: goal.thread_id.clone(), - goal_id: goal.goal_id.clone(), - objective: goal.objective.clone(), - status: from_crate_status(goal.status), - token_budget: goal.token_budget, - tokens_used: goal.tokens_used, - time_used_seconds: goal.time_used_seconds, - created_at_ms: goal.created_at_ms, - updated_at_ms: goal.updated_at_ms, - continuation_suppressed: goal.continuation_suppressed, - } -} - -/// Write the faithful crate mirror of `goal` into `store` (ns `graph.goals`, -/// key `hex(thread_id)`). Overwrites any prior mirror; idempotent for an -/// unchanged value. -pub(crate) async fn put_mirror(store: &Arc, goal: &ThreadGoal) -> Result<(), String> { - let crate_goal = to_crate_goal(goal); - let value = - serde_json::to_value(&crate_goal).map_err(|e| format!("serialize crate goal: {e}"))?; - store - .put(GOALS_NAMESPACE, &goal_key(&goal.thread_id), value) - .await - .map_err(|e| format!("mirror thread goal into {GOALS_NAMESPACE}: {e}")) -} - -/// Read the current crate mirror for `thread_id`, or `None`. Skips a mirror that -/// fails to decode (treated as absent) so a legacy/corrupt row can't wedge the -/// shadow path. -pub(crate) async fn get_mirror( - store: &Arc, - thread_id: &str, -) -> Result, String> { - let value = store - .get(GOALS_NAMESPACE, &goal_key(thread_id)) - .await - .map_err(|e| format!("read crate goal mirror: {e}"))?; - match value { - Some(v) => match serde_json::from_value::(v) { - Ok(crate_goal) => Ok(Some(from_crate_goal(&crate_goal))), - Err(e) => { - tracing::debug!( - thread_id = %thread_id, - error = %e, - "[thread_goals][crate-shadow] undecodable crate mirror; treating as absent" - ); - Ok(None) - } - }, - None => Ok(None), - } -} - -/// Delete the crate mirror for `thread_id`. No-op when absent (matches the -/// crate/legacy clear contract). -pub(crate) async fn delete_mirror(store: &Arc, thread_id: &str) -> Result<(), String> { - store - .delete(GOALS_NAMESPACE, &goal_key(thread_id)) - .await - .map_err(|e| format!("delete crate goal mirror: {e}")) -} - -// ── Idempotent legacy→crate migration helper (run on each core boot) ────────── - -/// Outcome of a [`migrate_legacy_goals_into_crate_store`] run. -#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] -pub struct GoalMigrationReport { - /// Legacy goal rows examined. - pub total: usize, - /// Rows written into the crate store because no crate value existed. - pub copied: usize, - /// Rows already present in the crate store and left authoritative. - pub skipped: usize, -} - -/// Copy legacy thread-goal rows missing from the crate `graph.goals` store. -/// -/// **Idempotent**: any existing crate value is authoritative and skipped, even -/// when it differs from the stale legacy file. After a row is copied or -/// skipped, its retired file is removed so clearing the crate row later cannot -/// resurrect stale state on the next boot. -/// -/// Wired into `core::runtime::services::start_boot_once_jobs` on every core -/// boot. Honors the single-writer constraint: run it only inside the core -/// process. -/// Read any goals left in the retired legacy `{workspace}/thread_goals/` file- -/// JSON tree. Returns an empty vec when the directory is absent (the common -/// case after the first migration). Undecodable/unreadable files are skipped — -/// a stray file can't wedge the idempotent copy. -struct LegacyGoalRow { - path: std::path::PathBuf, - goal: ThreadGoal, -} - -async fn read_legacy_file_goals(workspace_dir: &Path) -> Result, String> { - let dir = workspace_dir.join(LEGACY_GOALS_DIR); - let mut entries = match tokio::fs::read_dir(&dir).await { - Ok(rd) => rd, - Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()), - Err(e) => { - return Err(format!( - "read legacy thread goals dir {}: {e}", - dir.display() - )) - } - }; - let mut goals = Vec::new(); - while let Some(entry) = entries - .next_entry() - .await - .map_err(|e| format!("iterate legacy thread goals dir: {e}"))? - { - let path = entry.path(); - if path.extension().and_then(|e| e.to_str()) != Some(LEGACY_GOALS_EXTENSION) { - continue; - } - match tokio::fs::read_to_string(&path).await { - Ok(body) => match serde_json::from_str::(&body) { - Ok(goal) => goals.push(LegacyGoalRow { path, goal }), - Err(e) => { - tracing::debug!(path = %path.display(), error = %e, "[thread_goals][crate-migrate] skip parse error"); - } - }, - Err(e) => { - tracing::debug!(path = %path.display(), error = %e, "[thread_goals][crate-migrate] skip read error"); - } - } - } - Ok(goals) -} - -pub async fn migrate_legacy_goals_into_crate_store( - workspace_dir: &Path, -) -> Result { - let legacy = read_legacy_file_goals(workspace_dir).await?; - let store = crate_goals_store(workspace_dir); - let mut report = GoalMigrationReport { - total: legacy.len(), - ..Default::default() - }; - tracing::info!( - workspace = %workspace_dir.display(), - total = report.total, - "[thread_goals][crate-migrate] start copy legacy goals → graph.goals" - ); - for row in &legacy { - let goal = &row.goal; - match store.get(GOALS_NAMESPACE, &goal_key(&goal.thread_id)).await { - Ok(Some(_)) => { - report.skipped += 1; - tracing::debug!( - thread_id = %goal.thread_id, - goal_id = %goal.goal_id, - "[thread_goals][crate-migrate] skip (crate goal already authoritative)" - ); - } - Ok(None) => { - put_mirror(&store, goal).await?; - report.copied += 1; - tracing::debug!( - thread_id = %goal.thread_id, - goal_id = %goal.goal_id, - "[thread_goals][crate-migrate] copied" - ); - } - Err(e) => { - return Err(format!( - "check crate goal before migrating thread {}: {e}", - goal.thread_id - )) - } - } - tokio::fs::remove_file(&row.path).await.map_err(|e| { - format!( - "remove migrated legacy thread goal {}: {e}", - row.path.display() - ) - })?; - } - tracing::info!( - total = report.total, - copied = report.copied, - skipped = report.skipped, - "[thread_goals][crate-migrate] done" - ); - Ok(report) -} - -#[cfg(test)] -mod tests { - use super::*; - - fn sample_goal(status: ThreadGoalStatus) -> ThreadGoal { - ThreadGoal { - thread_id: "thread-α".into(), - goal_id: "goal-uuid-1".into(), - objective: "ship the migration".into(), - status, - token_budget: Some(5_000), - tokens_used: 1_234, - time_used_seconds: 42, - created_at_ms: 1_000, - updated_at_ms: 2_000, - continuation_suppressed: true, - } - } - - #[test] - fn status_mapping_is_bijective_across_all_variants() { - for status in [ - ThreadGoalStatus::Active, - ThreadGoalStatus::Paused, - ThreadGoalStatus::BudgetLimited, - ThreadGoalStatus::Complete, - ] { - let round = from_crate_status(to_crate_status(status)); - assert_eq!(round, status, "status round-trip must be identity"); - } - // Pin the exact crate labels the mapping produces. - assert_eq!(to_crate_status(ThreadGoalStatus::Active).as_str(), "active"); - assert_eq!(to_crate_status(ThreadGoalStatus::Paused).as_str(), "paused"); - assert_eq!( - to_crate_status(ThreadGoalStatus::BudgetLimited).as_str(), - "budget_limited" - ); - assert_eq!( - to_crate_status(ThreadGoalStatus::Complete).as_str(), - "complete" - ); - } - - #[test] - fn goal_mapping_preserves_every_field_and_completion_contract() { - // Completion contract: Complete + continuation_suppressed carries through. - let g = sample_goal(ThreadGoalStatus::Complete); - let crate_goal = to_crate_goal(&g); - assert_eq!(crate_goal.thread_id, g.thread_id); - assert_eq!( - crate_goal.goal_id, g.goal_id, - "goal_id preserved (no re-mint)" - ); - assert_eq!(crate_goal.objective, g.objective); - assert_eq!(crate_goal.status, CrateStatus::Complete); - assert_eq!(crate_goal.token_budget, g.token_budget, "budget preserved"); - assert_eq!(crate_goal.tokens_used, g.tokens_used, "usage preserved"); - assert_eq!(crate_goal.time_used_seconds, g.time_used_seconds); - assert_eq!(crate_goal.created_at_ms, g.created_at_ms); - assert_eq!(crate_goal.updated_at_ms, g.updated_at_ms); - assert!( - crate_goal.continuation_suppressed, - "completion suppresses continuation" - ); - // Full round-trip identity. - assert_eq!(from_crate_goal(&crate_goal), g); - } - - #[test] - fn budget_limited_maps_and_over_budget_carries() { - let mut g = sample_goal(ThreadGoalStatus::BudgetLimited); - g.tokens_used = 6_000; // over the 5_000 budget - let crate_goal = to_crate_goal(&g); - assert_eq!(crate_goal.status, CrateStatus::BudgetLimited); - assert!(crate_goal.over_budget(), "over-budget invariant carries"); - assert_eq!(crate_goal.budget_remaining(), Some(0)); - } - - #[tokio::test] - async fn put_get_delete_mirror_round_trip() { - let tmp = tempfile::tempdir().unwrap(); - let store = crate_goals_store(tmp.path()); - let g = sample_goal(ThreadGoalStatus::Active); - - assert!(get_mirror(&store, &g.thread_id).await.unwrap().is_none()); - put_mirror(&store, &g).await.unwrap(); - let read = get_mirror(&store, &g.thread_id).await.unwrap().unwrap(); - assert_eq!(read, g, "mirror round-trips the exact legacy value"); - - delete_mirror(&store, &g.thread_id).await.unwrap(); - assert!(get_mirror(&store, &g.thread_id).await.unwrap().is_none()); - // Delete is idempotent (no-op when absent). - delete_mirror(&store, &g.thread_id).await.unwrap(); - } - - #[tokio::test] - async fn crate_reader_resolves_the_mirrored_key() { - // Proves the key/namespace we write matches what the crate's own - // `graph::goals::store` reader computes — the whole point of the mirror. - let tmp = tempfile::tempdir().unwrap(); - let store = crate_goals_store(tmp.path()); - let g = sample_goal(ThreadGoalStatus::Paused); - put_mirror(&store, &g).await.unwrap(); - - let via_crate = tinyagents::graph::goals::store::get(&store, &g.thread_id) - .await - .unwrap() - .expect("crate reader finds the mirrored row"); - assert_eq!(via_crate.goal_id, g.goal_id); - assert_eq!(via_crate.status, CrateStatus::Paused); - assert_eq!(via_crate.tokens_used, g.tokens_used); - } - - /// Write a goal into the retired legacy `{workspace}/thread_goals/` file-JSON - /// tree (the shape `read_legacy_file_goals` reads), so the migration has - /// something to copy. - fn legacy_goal_file(dir: &std::path::Path, thread_id: &str) -> std::path::PathBuf { - legacy_goal_path(dir, thread_id).unwrap() - } - - fn write_legacy_goal(dir: &std::path::Path, goal: &ThreadGoal) { - let legacy_dir = dir.join("thread_goals"); - std::fs::create_dir_all(&legacy_dir).unwrap(); - let path = legacy_goal_file(dir, &goal.thread_id); - std::fs::write(&path, serde_json::to_string(goal).unwrap()).unwrap(); - } - - fn legacy_goal(thread_id: &str, objective: &str, tokens_used: u64) -> ThreadGoal { - ThreadGoal { - thread_id: thread_id.into(), - goal_id: format!("goal-{thread_id}"), - objective: objective.into(), - status: ThreadGoalStatus::Active, - token_budget: None, - tokens_used, - time_used_seconds: 0, - created_at_ms: 1_000, - updated_at_ms: 2_000, - continuation_suppressed: false, - } - } - - #[tokio::test] - async fn migration_copies_then_is_idempotent() { - let tmp = tempfile::tempdir().unwrap(); - let dir = tmp.path(); - // Seed two goals into the legacy file-JSON tree. - write_legacy_goal(dir, &legacy_goal("t1", "objective one", 0)); - write_legacy_goal(dir, &legacy_goal("t2", "objective two", 50)); - - // First run copies both. - let r1 = migrate_legacy_goals_into_crate_store(dir).await.unwrap(); - assert_eq!(r1.total, 2); - assert_eq!(r1.copied, 2); - assert_eq!(r1.skipped, 0); - assert!(!legacy_goal_file(dir, "t1").exists()); - assert!(!legacy_goal_file(dir, "t2").exists()); - - // Crate rows now match legacy rows exactly. - let store = crate_goals_store(dir); - let m2 = get_mirror(&store, "t2").await.unwrap().unwrap(); - assert_eq!(m2.tokens_used, 50); - assert_eq!(m2.objective, "objective two"); - - // Simulate live usage/status updates after migration. - let mut advanced = m2; - advanced.tokens_used = 999; - advanced.status = ThreadGoalStatus::Complete; - advanced.updated_at_ms = 3_000; - put_mirror(&store, &advanced).await.unwrap(); - - // Second run has no retired rows left and preserves the newer state. - let r2 = migrate_legacy_goals_into_crate_store(dir).await.unwrap(); - assert_eq!(r2.total, 0); - assert_eq!(r2.copied, 0, "idempotent: nothing re-copied"); - assert_eq!(r2.skipped, 0); - let preserved = get_mirror(&store, "t2").await.unwrap().unwrap(); - assert_eq!(preserved.tokens_used, 999); - assert_eq!(preserved.status, ThreadGoalStatus::Complete); - assert_eq!(preserved.updated_at_ms, 3_000); - } - - #[tokio::test] - async fn migration_discards_stale_legacy_row_when_crate_is_authoritative() { - let tmp = tempfile::tempdir().unwrap(); - let dir = tmp.path(); - let stale = legacy_goal("t1", "stale objective", 1); - write_legacy_goal(dir, &stale); - - let store = crate_goals_store(dir); - let current = legacy_goal("t1", "current objective", 99); - put_mirror(&store, ¤t).await.unwrap(); - - let report = migrate_legacy_goals_into_crate_store(dir).await.unwrap(); - assert_eq!(report.total, 1); - assert_eq!(report.copied, 0); - assert_eq!(report.skipped, 1); - assert!(!legacy_goal_file(dir, "t1").exists()); - assert_eq!(get_mirror(&store, "t1").await.unwrap(), Some(current)); - } - - #[tokio::test] - async fn clear_does_not_allow_migrated_goal_to_resurrect() { - let tmp = tempfile::tempdir().unwrap(); - let dir = tmp.path(); - let goal = legacy_goal("t1", "retired objective", 5); - write_legacy_goal(dir, &goal); - let store = crate_goals_store(dir); - put_mirror(&store, &goal).await.unwrap(); - - assert!(super::super::store::clear(dir, "t1").await.unwrap()); - assert!(!legacy_goal_file(dir, "t1").exists()); - assert!(get_mirror(&store, "t1").await.unwrap().is_none()); - - let report = migrate_legacy_goals_into_crate_store(dir).await.unwrap(); - assert_eq!(report.total, 0); - assert!(get_mirror(&store, "t1").await.unwrap().is_none()); - } -} diff --git a/src/openhuman/thread_goals/migration.rs b/src/openhuman/thread_goals/migration.rs new file mode 100644 index 000000000..103b9753a --- /dev/null +++ b/src/openhuman/thread_goals/migration.rs @@ -0,0 +1,177 @@ +//! One-time migration from OpenHuman's retired file-backed thread-goal store +//! into tinyagents' authoritative `graph.goals` namespace. + +use std::path::Path; +use std::sync::Arc; + +use ::tinyagents::graph::goals::store::GOALS_NAMESPACE; +use ::tinyagents::harness::store::Store; + +use super::ThreadGoal; +use crate::openhuman::session_import::ops::open_session_stores; + +const LEGACY_GOALS_DIR: &str = "thread_goals"; +const LEGACY_GOALS_EXTENSION: &str = "json"; + +pub(crate) fn goals_store(workspace_dir: &Path) -> Arc { + Arc::new(open_session_stores(workspace_dir).kv) +} + +fn goal_key(thread_id: &str) -> String { + thread_id + .trim() + .as_bytes() + .iter() + .map(|byte| format!("{byte:02x}")) + .collect() +} + +fn legacy_goal_path(workspace_dir: &Path, thread_id: &str) -> Result { + let thread_id = thread_id.trim(); + if thread_id.is_empty() { + return Err("invalid thread goal thread_id: empty or whitespace".to_string()); + } + Ok(workspace_dir.join(LEGACY_GOALS_DIR).join(format!( + "{}.{LEGACY_GOALS_EXTENSION}", + hex::encode(thread_id.as_bytes()) + ))) +} + +pub(crate) async fn delete_legacy_goal_file( + workspace_dir: &Path, + thread_id: &str, +) -> Result { + let path = legacy_goal_path(workspace_dir, thread_id)?; + match tokio::fs::remove_file(&path).await { + Ok(()) => Ok(true), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(false), + Err(error) => Err(format!( + "delete legacy thread goal {}: {error}", + path.display() + )), + } +} + +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct GoalMigrationReport { + pub total: usize, + pub copied: usize, + pub skipped: usize, +} + +struct LegacyGoalRow { + path: std::path::PathBuf, + goal: ThreadGoal, +} + +async fn read_legacy_goals(workspace_dir: &Path) -> Result, String> { + let dir = workspace_dir.join(LEGACY_GOALS_DIR); + let mut entries = match tokio::fs::read_dir(&dir).await { + Ok(entries) => entries, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()), + Err(error) => { + return Err(format!( + "read legacy thread goals dir {}: {error}", + dir.display() + )) + } + }; + let mut goals = Vec::new(); + while let Some(entry) = entries + .next_entry() + .await + .map_err(|error| format!("iterate legacy thread goals dir: {error}"))? + { + let path = entry.path(); + if path.extension().and_then(|extension| extension.to_str()) != Some(LEGACY_GOALS_EXTENSION) + { + continue; + } + if let Ok(body) = tokio::fs::read_to_string(&path).await { + if let Ok(goal) = serde_json::from_str::(&body) { + goals.push(LegacyGoalRow { path, goal }); + } + } + } + Ok(goals) +} + +pub async fn migrate_legacy_goals(workspace_dir: &Path) -> Result { + let legacy = read_legacy_goals(workspace_dir).await?; + let store = goals_store(workspace_dir); + let mut report = GoalMigrationReport { + total: legacy.len(), + ..Default::default() + }; + + for row in legacy { + let key = goal_key(&row.goal.thread_id); + if store + .get(GOALS_NAMESPACE, &key) + .await + .map_err(|error| format!("read tinyagents goal during migration: {error}"))? + .is_some() + { + report.skipped += 1; + } else { + let value = serde_json::to_value(&row.goal) + .map_err(|error| format!("serialize legacy thread goal: {error}"))?; + store + .put(GOALS_NAMESPACE, &key, value) + .await + .map_err(|error| format!("write tinyagents goal during migration: {error}"))?; + report.copied += 1; + } + tokio::fs::remove_file(&row.path) + .await + .map_err(|error| format!("remove migrated goal {}: {error}", row.path.display()))?; + } + Ok(report) +} + +#[cfg(test)] +mod tests { + use super::*; + use ::tinyagents::graph::goals::{store, ThreadGoalStatus}; + + #[tokio::test] + async fn migrates_legacy_goal_into_tinyagents_store() { + let temp = tempfile::tempdir().unwrap(); + let legacy_dir = temp.path().join(LEGACY_GOALS_DIR); + tokio::fs::create_dir_all(&legacy_dir).await.unwrap(); + let goal = ThreadGoal { + thread_id: "thread-1".into(), + goal_id: "legacy-id".into(), + objective: "legacy objective".into(), + status: ThreadGoalStatus::Active, + token_budget: Some(100), + tokens_used: 10, + time_used_seconds: 2, + created_at_ms: 1, + updated_at_ms: 2, + continuation_suppressed: false, + }; + let path = legacy_goal_path(temp.path(), &goal.thread_id).unwrap(); + tokio::fs::write(&path, serde_json::to_vec(&goal).unwrap()) + .await + .unwrap(); + + let report = migrate_legacy_goals(temp.path()).await.unwrap(); + assert_eq!( + report, + GoalMigrationReport { + total: 1, + copied: 1, + skipped: 0 + } + ); + assert_eq!( + store::get(&goals_store(temp.path()), "thread-1") + .await + .unwrap() + .unwrap(), + goal + ); + assert!(!path.exists()); + } +} diff --git a/src/openhuman/thread_goals/mod.rs b/src/openhuman/thread_goals/mod.rs index dd27d30b2..1eb3ca3f1 100644 --- a/src/openhuman/thread_goals/mod.rs +++ b/src/openhuman/thread_goals/mod.rs @@ -1,36 +1,20 @@ -//! `thread_goals` — the agent's single, thread-scoped goal. +//! Thin OpenHuman host adapters for [`tinyagents::graph::goals`]. //! -//! A **thread goal** is a durable "completion contract" the agent keeps -//! pursuing across turns, interrupts, resumes, and budget boundaries — modelled -//! on OpenAI Codex's `/goal`. It is deliberately distinct from the two existing -//! "goals" concepts: -//! -//! - [`memory_goals`](crate::openhuman::memory_goals) — a *global*, long-term -//! list of the user's durable goals (`MEMORY_GOALS.md`). -//! - the per-thread kanban [task board](crate::openhuman::agent::task_board) — -//! a list of work cards. -//! -//! There is **exactly one** thread goal per thread, with a small lifecycle -//! (active / paused / budget_limited / complete), an optional token budget, and -//! support for autonomous idle continuation. -//! -//! Persistence lives in the vendored `tinyagents` crate's `graph::goals` KV -//! store (`/tinyagents_store/kv/graph.goals/`); [`store`] is a thin -//! adapter over it (see [`crate_adapter`]). Goals from the retired legacy -//! `/thread_goals/` file-JSON tree are copied into the crate store -//! once at boot. Two writers may set the goal when a chat begins: the -//! orchestrator (authoritative, via `goal_set`) and the context-gathering path -//! (proposes only if absent, via [`store::set_if_absent`]). +//! Tinyagents owns the goal types, lifecycle, persistence, prompt rendering, +//! graph continuation, and native harness tools. This compatibility module +//! contains only OpenHuman-specific concerns: workspace-store resolution, +//! JSON-RPC schemas, domain events, the legacy file migration, heartbeat +//! dispatch, and adapters for OpenHuman's pre-tinyagents `Tool`/`StopHook` +//! traits. The external `thread_goals.*` RPC namespace remains stable. pub mod continuation; -pub mod crate_adapter; +pub mod migration; pub mod ops; pub mod runtime; mod schemas; pub mod store; pub mod tools; -pub mod types; +pub use ::tinyagents::graph::goals::{ThreadGoal, ThreadGoalStatus}; pub use schemas::{all_thread_goals_controller_schemas, all_thread_goals_registered_controllers}; pub use tools::{GoalCompleteTool, GoalGetTool, GoalSetTool}; -pub use types::{ThreadGoal, ThreadGoalStatus}; diff --git a/src/openhuman/thread_goals/ops.rs b/src/openhuman/thread_goals/ops.rs index 9a2b0bba7..5b46e3670 100644 --- a/src/openhuman/thread_goals/ops.rs +++ b/src/openhuman/thread_goals/ops.rs @@ -8,7 +8,7 @@ use std::path::Path; use serde::Serialize; use super::store; -use super::types::ThreadGoal; +use super::ThreadGoal; use crate::core::event_bus::{publish_global, DomainEvent}; use crate::rpc::RpcOutcome; diff --git a/src/openhuman/thread_goals/runtime.rs b/src/openhuman/thread_goals/runtime.rs index 71f4016fa..317bb715c 100644 --- a/src/openhuman/thread_goals/runtime.rs +++ b/src/openhuman/thread_goals/runtime.rs @@ -1,13 +1,9 @@ -//! Harness-level runtime for the thread goal: per-turn context injection, -//! token-budget accounting, and the mid-turn budget stop hook. +//! OpenHuman runtime adapters for tinyagents thread-goal accounting and the +//! host-specific mid-turn budget stop hook. //! //! These are the pieces that make a stored goal actually steer the agent //! (Codex parity): //! -//! - [`active_goal_context_block`] renders a compact `[active_goal]` block that -//! the turn loop prepends to the user message **fresh each turn** (never the -//! cached system-prompt prefix), so the objective stays visible and the model -//! sees live budget/status. //! - [`account_turn_against_goal`] folds a completed turn's token + time usage //! into the active goal, flipping it to `budget_limited` when the cap is //! crossed. @@ -26,7 +22,7 @@ use std::path::{Path, PathBuf}; use async_trait::async_trait; use super::store; -use super::types::{ThreadGoal, ThreadGoalStatus}; +use super::{ThreadGoal, ThreadGoalStatus}; use crate::core::event_bus::{publish_global, DomainEvent}; use crate::openhuman::agent::stop_hooks::{StopDecision, StopHook, TurnState}; use crate::openhuman::tinyagents::thread_context::current_thread_id; @@ -89,43 +85,6 @@ pub async fn pause_for_current_thread(workspace_dir: &Path) { } } -/// Render the per-turn `[active_goal]` context block for `goal`, or `None` when -/// the goal is in a state that needs no steering text. -/// -/// The block is intentionally tiny and source-attributed so it reads as harness -/// state, not user instruction. -pub fn active_goal_context_block(goal: &ThreadGoal) -> Option { - let directive = match goal.status { - ThreadGoalStatus::Active => { - "Keep working toward this goal. Before responding, verify whether the \ - objective is satisfied. If confirmed, call `goal_complete` now. \ - If the objective has changed, call `goal_set` to update it." - } - ThreadGoalStatus::BudgetLimited => { - "This goal has reached its token budget. Stop substantive work: summarise \ - progress and blockers, and name the next useful step. Do not continue \ - until the user raises the budget or clears the goal." - } - // A paused goal isn't being worked right now; a completed goal needs no - // steering. Surfacing them would only add noise to the turn. - ThreadGoalStatus::Paused | ThreadGoalStatus::Complete => return None, - }; - let budget = match (goal.token_budget, goal.budget_remaining()) { - (Some(b), Some(rem)) => format!( - "\nbudget: {} used / {b} ({rem} remaining)", - goal.tokens_used - ), - _ => String::new(), - }; - Some(format!( - "[active_goal]\nstatus: {}\nobjective: {}{}\n{}\n[/active_goal]\n\n", - goal.status.as_str(), - goal.objective, - budget, - directive - )) -} - /// The per-turn token total used for budget accounting (prompt + completion). fn turn_tokens(input: u64, output: u64) -> u64 { input.saturating_add(output) @@ -299,64 +258,6 @@ mod tests { tc } - #[test] - fn active_block_includes_objective_and_budget() { - let goal = ThreadGoal { - thread_id: "t".into(), - goal_id: "g".into(), - objective: "ship the feature".into(), - status: ThreadGoalStatus::Active, - token_budget: Some(1000), - tokens_used: 250, - time_used_seconds: 0, - created_at_ms: 0, - updated_at_ms: 0, - continuation_suppressed: false, - }; - let block = active_goal_context_block(&goal).unwrap(); - assert!(block.contains("[active_goal]")); - assert!(block.contains("ship the feature")); - assert!(block.contains("250 used / 1000")); - assert!(block.contains("goal_complete")); - } - - #[test] - fn budget_limited_block_steers_to_summarise() { - let goal = ThreadGoal { - thread_id: "t".into(), - goal_id: "g".into(), - objective: "obj".into(), - status: ThreadGoalStatus::BudgetLimited, - token_budget: Some(100), - tokens_used: 100, - time_used_seconds: 0, - created_at_ms: 0, - updated_at_ms: 0, - continuation_suppressed: false, - }; - let block = active_goal_context_block(&goal).unwrap(); - assert!(block.contains("reached its token budget")); - } - - #[test] - fn paused_and_complete_render_no_block() { - let mut goal = ThreadGoal { - thread_id: "t".into(), - goal_id: "g".into(), - objective: "obj".into(), - status: ThreadGoalStatus::Paused, - token_budget: None, - tokens_used: 0, - time_used_seconds: 0, - created_at_ms: 0, - updated_at_ms: 0, - continuation_suppressed: false, - }; - assert!(active_goal_context_block(&goal).is_none()); - goal.status = ThreadGoalStatus::Complete; - assert!(active_goal_context_block(&goal).is_none()); - } - #[tokio::test] async fn account_turn_charges_active_goal_and_trips_budget() { let tmp = tempfile::tempdir().unwrap(); diff --git a/src/openhuman/thread_goals/schemas.rs b/src/openhuman/thread_goals/schemas.rs index 2fb0efbc9..d6f92309a 100644 --- a/src/openhuman/thread_goals/schemas.rs +++ b/src/openhuman/thread_goals/schemas.rs @@ -1,4 +1,4 @@ -//! Controller schemas + JSON-RPC handlers for the `thread_goals` namespace. +//! OpenHuman JSON-RPC adapters for tinyagents' `graph::goals` domain. //! //! Methods are exposed as `openhuman.thread_goals_`: //! `get`, `set`, `complete`, `pause`, `resume`, `clear`. Handlers load the diff --git a/src/openhuman/thread_goals/store.rs b/src/openhuman/thread_goals/store.rs index 78ad50fe0..8cf54d78d 100644 --- a/src/openhuman/thread_goals/store.rs +++ b/src/openhuman/thread_goals/store.rs @@ -1,4 +1,4 @@ -//! Persistence for the thread-level goal. +//! Workspace-path adapter onto tinyagents' authoritative goal store. //! //! **Crate-backed.** The per-thread goal now lives in the vendored `tinyagents` //! crate's `graph::goals` KV store @@ -17,9 +17,9 @@ use std::path::Path; -use super::crate_adapter::{crate_goals_store, delete_legacy_goal_file, from_crate_goal}; -use super::types::ThreadGoal; -use tinyagents::graph::goals::store as crate_store; +use super::migration::{delete_legacy_goal_file, goals_store}; +use super::ThreadGoal; +use ::tinyagents::graph::goals::store as crate_store; /// Set (create or replace) the thread's goal. A changed objective mints a fresh /// goal and resets counters; an unchanged objective preserves counters and @@ -30,11 +30,10 @@ pub async fn set( objective: &str, token_budget: Option, ) -> Result { - let store = crate_goals_store(workspace_dir); + let store = goals_store(workspace_dir); let goal = crate_store::set(&store, thread_id, objective, token_budget) .await .map_err(|e| e.to_string())?; - let goal = from_crate_goal(&goal); tracing::info!( thread_id = %goal.thread_id, goal_id = %goal.goal_id, @@ -53,35 +52,35 @@ pub async fn set_if_absent( objective: &str, token_budget: Option, ) -> Result, String> { - let store = crate_goals_store(workspace_dir); + let store = goals_store(workspace_dir); let goal = crate_store::set_if_absent(&store, thread_id, objective, token_budget) .await .map_err(|e| e.to_string())?; - Ok(goal.as_ref().map(from_crate_goal)) + Ok(goal) } /// The thread's current goal, or `None`. pub async fn get(workspace_dir: &Path, thread_id: &str) -> Result, String> { - let store = crate_goals_store(workspace_dir); + let store = goals_store(workspace_dir); let goal = crate_store::get(&store, thread_id) .await .map_err(|e| e.to_string())?; - Ok(goal.as_ref().map(from_crate_goal)) + Ok(goal) } /// Every stored thread goal (used by the heartbeat continuation sweep). pub async fn list_all(workspace_dir: &Path) -> Result, String> { - let store = crate_goals_store(workspace_dir); + let store = goals_store(workspace_dir); let goals = crate_store::list_all(&store) .await .map_err(|e| e.to_string())?; - Ok(goals.iter().map(from_crate_goal).collect()) + Ok(goals) } /// Delete the thread's goal. Returns whether a goal was present. pub async fn clear(workspace_dir: &Path, thread_id: &str) -> Result { delete_legacy_goal_file(workspace_dir, thread_id).await?; - let store = crate_goals_store(workspace_dir); + let store = goals_store(workspace_dir); let existed = crate_store::clear(&store, thread_id) .await .map_err(|e| e.to_string())?; @@ -91,31 +90,30 @@ pub async fn clear(workspace_dir: &Path, thread_id: &str) -> Result Result { - let store = crate_goals_store(workspace_dir); + let store = goals_store(workspace_dir); let goal = crate_store::complete(&store, thread_id) .await .map_err(|e| e.to_string())?; - let goal = from_crate_goal(&goal); tracing::info!(thread_id = %goal.thread_id, goal_id = %goal.goal_id, "[thread_goals] complete"); Ok(goal) } /// Mark the goal `Paused`. pub async fn pause(workspace_dir: &Path, thread_id: &str) -> Result { - let store = crate_goals_store(workspace_dir); + let store = goals_store(workspace_dir); let goal = crate_store::pause(&store, thread_id) .await .map_err(|e| e.to_string())?; - Ok(from_crate_goal(&goal)) + Ok(goal) } /// Resume a `Paused` goal back to `Active`. pub async fn resume(workspace_dir: &Path, thread_id: &str) -> Result { - let store = crate_goals_store(workspace_dir); + let store = goals_store(workspace_dir); let goal = crate_store::resume(&store, thread_id) .await .map_err(|e| e.to_string())?; - Ok(from_crate_goal(&goal)) + Ok(goal) } /// Set `continuation_suppressed` only when the thread's current goal still @@ -128,7 +126,7 @@ pub async fn set_continuation_suppressed_if( expected_goal_id: &str, suppressed: bool, ) -> Result, String> { - let store = crate_goals_store(workspace_dir); + let store = goals_store(workspace_dir); let goal = crate_store::set_continuation_suppressed_if( &store, thread_id, @@ -137,7 +135,7 @@ pub async fn set_continuation_suppressed_if( ) .await .map_err(|e| e.to_string())?; - Ok(goal.as_ref().map(from_crate_goal)) + Ok(goal) } /// Account token + time usage against the goal, applying the budget constraint. @@ -151,10 +149,10 @@ pub async fn account_usage( token_delta: u64, secs_delta: u64, ) -> Result, String> { - let store = crate_goals_store(workspace_dir); + let store = goals_store(workspace_dir); let goal = crate_store::account_usage(&store, thread_id, expected_goal_id, token_delta, secs_delta) .await .map_err(|e| e.to_string())?; - Ok(goal.as_ref().map(from_crate_goal)) + Ok(goal) } diff --git a/src/openhuman/thread_goals/tools.rs b/src/openhuman/thread_goals/tools.rs index 076fcf632..f050326d0 100644 --- a/src/openhuman/thread_goals/tools.rs +++ b/src/openhuman/thread_goals/tools.rs @@ -1,4 +1,4 @@ -//! Agent-facing tools for the thread-level goal. +//! OpenHuman `Tool` adapters for tinyagents' model-facing goal controls. //! //! These let the orchestrator (and any agent that allowlists them) read and //! drive the current thread's goal. Ownership is **asymmetric** (Codex parity): @@ -18,7 +18,7 @@ use async_trait::async_trait; use serde_json::json; use super::store; -use super::types::ThreadGoal; +use super::ThreadGoal; use crate::openhuman::tinyagents::thread_context::current_thread_id; use crate::openhuman::tools::traits::{PermissionLevel, Tool, ToolResult}; diff --git a/src/openhuman/thread_goals/types.rs b/src/openhuman/thread_goals/types.rs deleted file mode 100644 index b828ef7ba..000000000 --- a/src/openhuman/thread_goals/types.rs +++ /dev/null @@ -1,151 +0,0 @@ -//! Domain types for the thread-level goal. -//! -//! A **thread goal** is a single, thread-scoped "completion contract" — a -//! durable objective the agent keeps pursuing across turns, interrupts, -//! resumes, and budget boundaries. It is distinct from the global -//! [`memory_goals`](crate::openhuman::memory_goals) list (long-term, workspace -//! wide) and from the per-thread kanban task board: there is **exactly one** -//! goal per thread, with a small lifecycle and optional token budget. -//! -//! The shape mirrors OpenAI Codex's `thread_goals` row, adapted to OpenHuman's -//! per-thread file-JSON persistence (see [`super::store`]). - -use serde::{Deserialize, Serialize}; - -/// Lifecycle state of a thread goal. -/// -/// Ownership is **asymmetric** (Codex parity): the model may create/replace a -/// goal and mark it `Complete`; `Paused` / `BudgetLimited` are system-driven -/// (interrupt/abort 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 agent may make progress and (when idle) auto-continue. - Active, - /// Work is suspended (user interrupt/abort); the objective persists and is - /// reactivated on thread resume. - Paused, - /// The token budget has been reached; substantive work halts until the user - /// raises the budget or clears the goal. - BudgetLimited, - /// Evidence confirms the objective is satisfied. - Complete, -} - -impl ThreadGoalStatus { - 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 agent should keep working it - /// (and idle auto-continuation may fire). - pub fn is_active(&self) -> bool { - matches!(self, Self::Active) - } - - /// Whether the goal is in a terminal state for continuation purposes — - /// `Complete` or `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 an idle auto-continuation turn produced **zero tool calls**, to - /// stop a continuation loop. Cleared on any user action, tool execution, or - /// external mutation (e.g. `goal_set`). - #[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) - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[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()); - } - - #[test] - fn budget_helpers() { - let mut g = ThreadGoal { - thread_id: "t".into(), - goal_id: "g".into(), - objective: "do it".into(), - status: ThreadGoalStatus::Active, - token_budget: Some(100), - tokens_used: 40, - time_used_seconds: 0, - created_at_ms: 0, - updated_at_ms: 0, - continuation_suppressed: false, - }; - 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()); - } -}