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(todos): cut over task boards to TinyAgents (#5239)

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