admin/openhuman
1
0
Fork 0
mirror of https://github.com/tinyhumansai/openhuman.git synced 2026-07-30 23:14:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(goals): consolidate thread goals under tinyagents (#5235)

Browse Source
This commit is contained in:
Steven Enamakel
2026-07-28 11:09:19 +03:00
committed by GitHub
parent 4ed193b060
commit 149137a71f
12 changed files with 219 additions and 852 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/core/runtime/services.rs
+2 -4
View File
@@ -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!(
src/openhuman/agent/harness/session/turn/core.rs
+1 -3
View File
@@ -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(),
src/openhuman/thread_goals/continuation.rs
+2 -2
View File
@@ -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;
src/openhuman/thread_goals/crate_adapter.rs
-538
View File
@@ -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<dyn Store> {
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<std::path::PathBuf, String> {
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<bool, String> {
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<dyn Store>, 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<dyn Store>,
thread_id: &str,
) -> Result<Option<ThreadGoal>, 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::<CrateThreadGoal>(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<dyn Store>, 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<Vec<LegacyGoalRow>, 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::<ThreadGoal>(&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<GoalMigrationReport, String> {
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, &current).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());
}
}
src/openhuman/thread_goals/migration.rs
+177
View File
@@ -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<dyn Store> {
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<std::path::PathBuf, String> {
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<bool, String> {
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<Vec<LegacyGoalRow>, 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::<ThreadGoal>(&body) {
goals.push(LegacyGoalRow { path, goal });
}
}
}
Ok(goals)
}
pub async fn migrate_legacy_goals(workspace_dir: &Path) -> Result<GoalMigrationReport, String> {
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());
}
}
src/openhuman/thread_goals/mod.rs
+9 -25
View File
@@ -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 (`<workspace>/tinyagents_store/kv/graph.goals/`); [`store`] is a thin
//! adapter over it (see [`crate_adapter`]). Goals from the retired legacy
//! `<workspace>/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};
src/openhuman/thread_goals/ops.rs
+1 -1
View File
@@ -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;
src/openhuman/thread_goals/runtime.rs
+3 -102
View File
@@ -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<String> {
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();
src/openhuman/thread_goals/schemas.rs
+1 -1
View File
@@ -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_<function>`:
//! `get`, `set`, `complete`, `pause`, `resume`, `clear`. Handlers load the
src/openhuman/thread_goals/store.rs
+21 -23
View File
@@ -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<u64>,
) -> Result<ThreadGoal, String> {
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<u64>,
) -> Result<Option<ThreadGoal>, 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<Option<ThreadGoal>, 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<Vec<ThreadGoal>, 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<bool, String> {
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<bool, String
/// Mark the goal `Complete` (model-driven success).
pub async fn complete(workspace_dir: &Path, thread_id: &str) -> Result<ThreadGoal, String> {
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<ThreadGoal, String> {
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<ThreadGoal, String> {
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<Option<ThreadGoal>, 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<Option<ThreadGoal>, 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)
}
src/openhuman/thread_goals/tools.rs
+2 -2
View File
@@ -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};
src/openhuman/thread_goals/types.rs
-151
View File
@@ -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<u64>,
/// 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<u64> {
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());
}
}