diff --git a/Cargo.lock b/Cargo.lock index 3313c5266..5a36a5dcd 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -156,7 +156,7 @@ dependencies = [ "maybe-owned", "rustix", "rustix-linux-procfs", - "windows-sys 0.60.2", + "windows-sys 0.61.2", "winx", ] @@ -351,7 +351,7 @@ dependencies = [ "libc", "option-ext", "redox_users", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -371,6 +371,12 @@ version = "0.15.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1aaf95b3e5c8f23aa320147307562d361db0ae0d51242340f558153b4eb2439b" +[[package]] +name = "dyn-clone" +version = "1.0.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d0881ea181b1df73ff77ffaaf9c7544ecc11e82fba9b5f27b262a3c73a332555" + [[package]] name = "equivalent" version = "1.0.3" @@ -384,7 +390,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -1327,7 +1333,7 @@ dependencies = [ "once_cell", "socket2", "tracing", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -1431,6 +1437,26 @@ dependencies = [ "thiserror", ] +[[package]] +name = "ref-cast" +version = "1.0.27" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7e440fb4e4b4147295338efb76001ab9e4efc0e5839df2c47fc5ac2381d365c3" +dependencies = [ + "ref-cast-impl", +] + +[[package]] +name = "ref-cast-impl" +version = "1.0.27" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92ecd8964f8453721699a1ed72037b0db49ce2f5a5138486ee89bed6f67cdf3a" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.9", +] + [[package]] name = "regex" version = "1.13.1" @@ -1567,7 +1593,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -1637,6 +1663,32 @@ dependencies = [ "winapi-util", ] +[[package]] +name = "schemars" +version = "1.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ab508826f74a77ca9d5aba6ff19b522583ee3eaf28a19384ff3d0e5835fadf6e" +dependencies = [ + "chrono", + "dyn-clone", + "ref-cast", + "schemars_derive", + "serde", + "serde_json", +] + +[[package]] +name = "schemars_derive" +version = "1.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e1c3a92094fa7d61aa124645844facb6b554dfc797136d0f5fd1f890e2bffc69" +dependencies = [ + "proc-macro2", + "quote", + "serde_derive_internals", + "syn 3.0.9", +] + [[package]] name = "scopeguard" version = "1.2.0" @@ -1673,6 +1725,17 @@ dependencies = [ "syn 3.0.9", ] +[[package]] +name = "serde_derive_internals" +version = "0.30.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f852137cce035d6a4df67ccce505ff6b3e9fd3a10e3e52b24dc71e650bb1a9bd" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.9", +] + [[package]] name = "serde_json" version = "1.0.154" @@ -1796,7 +1859,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4" dependencies = [ "libc", - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -1892,7 +1955,7 @@ dependencies = [ "getrandom 0.4.3", "once_cell", "rustix", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -1948,6 +2011,7 @@ dependencies = [ "regex", "reqwest", "rustix", + "schemars", "serde", "serde_json", "sha2 0.11.1", @@ -1993,6 +2057,7 @@ dependencies = [ "regex", "reqwest", "rustix", + "schemars", "serde", "serde_json", "sha2 0.11.1", @@ -2156,6 +2221,7 @@ dependencies = [ "chrono", "hex", "parking_lot", + "schemars", "serde", "serde_json", "sha2 0.11.1", @@ -3092,7 +3158,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] diff --git a/crates/tinyagents-harness-tools/Cargo.toml b/crates/tinyagents-harness-tools/Cargo.toml index 73828e783..162f5910c 100644 --- a/crates/tinyagents-harness-tools/Cargo.toml +++ b/crates/tinyagents-harness-tools/Cargo.toml @@ -24,6 +24,7 @@ reqwest = { workspace = true, features = ["stream", "multipart"], optional = tru sha2 = { version = "0.11", optional = true } serde = { workspace = true } serde_json = { workspace = true } +schemars = { version = "1", features = ["derive"], optional = true } tinyagents-harness = { path = "../tinyagents-harness", version = "3.0.0", default-features = false } tinyagents-tasks = { path = "../tinyagents-tasks", version = "3.0.0", default-features = false, optional = true } tinyinference-llm = { path = "../../vendor/tinyinference/crates/tinyinference-llm", version = "0.5.0", optional = true } @@ -47,6 +48,7 @@ goals = [] skills = [] state-block = ["goals", "todos", "dep:tinyinference-llm"] subagents = ["dep:tinyagents-orchestration", "dep:tinyagents-tasks", "dep:tinyinference-llm"] +json-schema = ["dep:schemars"] fs = ["dep:base64", "dep:fs2", "dep:glob", "dep:libc", "dep:log", "dep:regex", "dep:walkdir", "tokio/fs", "tokio/io-util"] net = ["dep:futures-util", "dep:log", "dep:reqwest", "dep:sha2", "tokio/fs", "tokio/io-util"] exec = ["dep:rustix", "tokio/fs", "tokio/io-util", "tokio/process"] diff --git a/crates/tinyagents-harness-tools/README.md b/crates/tinyagents-harness-tools/README.md index c8bbc1000..1da857750 100644 --- a/crates/tinyagents-harness-tools/README.md +++ b/crates/tinyagents-harness-tools/README.md @@ -11,7 +11,8 @@ feature enables the generic job and message tools and brings in `tinyagents-orchestration` and its task registry dependencies. Time, clarification, plan-exit, per-call binding, and thread-lock support are available without optional features. The `state-block` feature enables the -run-state middleware and its required `goals` and `todos` modules. +run-state middleware and its required `goals` and `todos` modules. The +`json-schema` feature enables schema derives on tool-owned data types. The opt-in `fs`, `net`, and `exec` features expose the standard tool families under `tinyagents_harness_tools::{fs,net,exec}`. The `media` feature exposes @@ -50,6 +51,10 @@ default set. `AskClarificationTool`, `PlanExitTool`, `WaitTool`, `WaitLoopTool`, and the `per_call` binding adapter. +`WaitTool` and `WaitLoopTool` provide bounded delayed callback ticks for +orchestrator-controlled polling. They return the caller's callback message +without owning scheduling state. + Subagent registry policy stays in `tinyagents-orchestration`; this crate owns the reusable tool declarations. The harness owns tool registration, admission, schema preparation, discovery, and dispatch infrastructure. diff --git a/crates/tinyagents-harness-tools/src/goals/mod_tests.rs b/crates/tinyagents-harness-tools/src/goals/mod_tests.rs index d491e6898..a2371f38d 100644 --- a/crates/tinyagents-harness-tools/src/goals/mod_tests.rs +++ b/crates/tinyagents-harness-tools/src/goals/mod_tests.rs @@ -3,6 +3,40 @@ use super::prompt::*; use super::types::*; +#[cfg(feature = "json-schema")] +fn schema_string_values<'a>( + root: &'a serde_json::Value, + mut schema: &'a serde_json::Value, +) -> Vec { + while let Some(reference) = schema.get("$ref").and_then(serde_json::Value::as_str) { + let pointer = reference.strip_prefix('#').expect("local schema reference"); + schema = root.pointer(pointer).expect("schema reference resolves"); + } + + let mut values: Vec = + if let Some(values) = schema.get("enum").and_then(serde_json::Value::as_array) { + values + .iter() + .map(|value| value.as_str().expect("string enum value").to_owned()) + .collect() + } else { + schema + .get("oneOf") + .and_then(serde_json::Value::as_array) + .expect("string enum or oneOf schema") + .iter() + .map(|variant| { + variant["const"] + .as_str() + .expect("string oneOf const") + .to_owned() + }) + .collect() + }; + values.sort(); + values +} + fn goal(status: ThreadGoalStatus, token_budget: Option, tokens_used: u64) -> ThreadGoal { ThreadGoal { thread_id: "t".into(), @@ -26,6 +60,50 @@ fn status_strings_match_serialized() { assert_eq!(ThreadGoalStatus::Complete.as_str(), "complete"); } +#[cfg(feature = "json-schema")] +#[test] +fn public_goal_schema_uses_the_serialized_camel_case_record() { + let schema = serde_json::to_value( + schemars::generate::SchemaSettings::draft2020_12() + .for_serialize() + .into_generator() + .into_root_schema_for::(), + ) + .unwrap(); + let properties = &schema["properties"]; + for field in [ + "threadId", + "goalId", + "objective", + "status", + "tokenBudget", + "tokensUsed", + "timeUsedSeconds", + "timeUsedSubsecNanos", + "createdAtMs", + "updatedAtMs", + ] { + assert!(properties.get(field).is_some(), "missing {field}: {schema}"); + } + let status_values = schema_string_values(&schema, &properties["status"]); + let mut expected_status_values = vec![ + "active".to_owned(), + "paused".to_owned(), + "budget_limited".to_owned(), + "complete".to_owned(), + ]; + expected_status_values.sort(); + assert_eq!(status_values, expected_status_values); + for (status, serialized) in [ + (ThreadGoalStatus::Active, "active"), + (ThreadGoalStatus::Paused, "paused"), + (ThreadGoalStatus::BudgetLimited, "budget_limited"), + (ThreadGoalStatus::Complete, "complete"), + ] { + assert_eq!(serde_json::to_value(status).unwrap(), serialized); + } +} + #[test] fn active_and_terminal_predicates() { assert!(ThreadGoalStatus::Active.is_active()); diff --git a/crates/tinyagents-harness-tools/src/goals/types.rs b/crates/tinyagents-harness-tools/src/goals/types.rs index a660e8643..f7dc02084 100644 --- a/crates/tinyagents-harness-tools/src/goals/types.rs +++ b/crates/tinyagents-harness-tools/src/goals/types.rs @@ -20,6 +20,7 @@ use serde::{Deserialize, Serialize}; /// system (host control and accounting respectively), and clearing deletes the /// row entirely rather than being a status. #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))] #[serde(rename_all = "snake_case")] pub enum ThreadGoalStatus { /// The graph may make progress and (when driven) auto-continue. @@ -61,6 +62,7 @@ impl ThreadGoalStatus { /// A single thread-scoped goal. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))] #[serde(rename_all = "camelCase")] pub struct ThreadGoal { /// The thread this goal belongs to (one goal per thread). diff --git a/crates/tinyagents-harness-tools/src/todos/mod_tests.rs b/crates/tinyagents-harness-tools/src/todos/mod_tests.rs index dbeb456e3..bd032435b 100644 --- a/crates/tinyagents-harness-tools/src/todos/mod_tests.rs +++ b/crates/tinyagents-harness-tools/src/todos/mod_tests.rs @@ -2,6 +2,40 @@ use super::types::*; +#[cfg(feature = "json-schema")] +fn schema_string_values<'a>( + root: &'a serde_json::Value, + mut schema: &'a serde_json::Value, +) -> Vec { + while let Some(reference) = schema.get("$ref").and_then(serde_json::Value::as_str) { + let pointer = reference.strip_prefix('#').expect("local schema reference"); + schema = root.pointer(pointer).expect("schema reference resolves"); + } + + let mut values: Vec = + if let Some(values) = schema.get("enum").and_then(serde_json::Value::as_array) { + values + .iter() + .map(|value| value.as_str().expect("string enum value").to_owned()) + .collect() + } else { + schema + .get("oneOf") + .and_then(serde_json::Value::as_array) + .expect("string enum or oneOf schema") + .iter() + .map(|variant| { + variant["const"] + .as_str() + .expect("string oneOf const") + .to_owned() + }) + .collect() + }; + values.sort(); + values +} + fn item(content: &str, status: TodoStatus) -> TodoItem { TodoItem::with_status(content, status) } @@ -25,6 +59,35 @@ fn status_strings_match_serialized() { } } +#[cfg(feature = "json-schema")] +#[test] +fn public_todo_schema_matches_the_serialized_item_shape() { + let schema = serde_json::to_value( + schemars::generate::SchemaSettings::draft2020_12() + .for_serialize() + .into_generator() + .into_root_schema_for::(), + ) + .unwrap(); + let properties = &schema["properties"]; + assert!(properties.get("content").is_some()); + assert!(properties.get("status").is_some()); + assert!(properties.get("reason").is_some()); + let status_values = schema_string_values(&schema, &properties["status"]); + let mut expected_status_values = vec![ + "pending".to_owned(), + "in_progress".to_owned(), + "completed".to_owned(), + "cancelled".to_owned(), + ]; + expected_status_values.sort(); + assert_eq!(status_values, expected_status_values); + assert_eq!( + serde_json::to_value(TodoItem::new("ship it")).unwrap(), + serde_json::json!({"content": "ship it", "status": "pending"}) + ); +} + #[test] fn parse_status_accepts_aliases() { assert_eq!(parse_status("pending").unwrap(), TodoStatus::Pending); diff --git a/crates/tinyagents-harness-tools/src/todos/types.rs b/crates/tinyagents-harness-tools/src/todos/types.rs index 642734bc3..edf5e5411 100644 --- a/crates/tinyagents-harness-tools/src/todos/types.rs +++ b/crates/tinyagents-harness-tools/src/todos/types.rs @@ -10,6 +10,7 @@ use serde::{Deserialize, Serialize}; /// Lifecycle state of a todo item. #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))] #[serde(rename_all = "snake_case")] pub enum TodoStatus { /// Not started. @@ -41,6 +42,7 @@ impl TodoStatus { /// One item on a thread's todo list. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))] #[serde(rename_all = "camelCase")] pub struct TodoItem { /// The step, as the model wrote it. diff --git a/crates/tinyagents-harness-tools/src/wait_tests.rs b/crates/tinyagents-harness-tools/src/wait_tests.rs index 2352b2c18..9fdbe9dbf 100644 --- a/crates/tinyagents-harness-tools/src/wait_tests.rs +++ b/crates/tinyagents-harness-tools/src/wait_tests.rs @@ -84,6 +84,17 @@ fn wait_loop_tick_omits_an_absent_loop_key() { ); } +#[test] +fn timeout_policy_covers_the_requested_wait_with_grace() { + use tinytools::{Tool, ToolTimeout}; + + let args = json!({"message": "poll", "duration_secs": 4}); + assert_eq!( + WaitTool::new().timeout_policy(&args), + ToolTimeout::Millis(5_000) + ); +} + #[tokio::test] async fn wait_execute_returns_callback_message() { let res = WaitTool::new() diff --git a/crates/tinyagents-harness/Cargo.toml b/crates/tinyagents-harness/Cargo.toml index f707c2f29..a739aaea1 100644 --- a/crates/tinyagents-harness/Cargo.toml +++ b/crates/tinyagents-harness/Cargo.toml @@ -26,6 +26,7 @@ regex = "1" reqwest = { workspace = true, features = ["stream", "http2"], optional = true } serde = { workspace = true } serde_json = { workspace = true } +schemars = { version = "1", features = ["derive"], optional = true } sha2 = "0.11" thiserror = "2" tracing = { workspace = true } @@ -77,6 +78,8 @@ media = ["dep:tinyinference-image", "dep:tinyinference-video"] claude-code = ["dep:wait-timeout", "dep:dirs"] # Gates the Langfuse observability exporter and its `reqwest` transport. langfuse = ["dep:reqwest"] +# Optional schema support for public harness wire types such as `RunMode`. +json-schema = ["dep:schemars"] # Tracing instrumentation is now always compiled in (via the `tracing` crate # dependency above). This feature is retained as a no-op so downstream # feature forwards keep compiling. diff --git a/crates/tinyagents-harness/src/middleware/library/mod_tests.rs b/crates/tinyagents-harness/src/middleware/library/mod_tests.rs index 0e8e8fe60..106fd39c4 100644 --- a/crates/tinyagents-harness/src/middleware/library/mod_tests.rs +++ b/crates/tinyagents-harness/src/middleware/library/mod_tests.rs @@ -18,6 +18,40 @@ use tinyinference_llm::model::{ModelRequest, ModelResponse, ResponseFormat}; use tinyinference_llm::tool::{ToolCall, ToolFormat, ToolSchema}; use tinytools::{SandboxMode, ToolAccess, ToolPolicy, ToolResult, ToolRuntime, ToolSideEffects}; +#[cfg(feature = "json-schema")] +fn schema_string_values<'a>( + root: &'a serde_json::Value, + mut schema: &'a serde_json::Value, +) -> Vec { + while let Some(reference) = schema.get("$ref").and_then(serde_json::Value::as_str) { + let pointer = reference.strip_prefix('#').expect("local schema reference"); + schema = root.pointer(pointer).expect("schema reference resolves"); + } + + let mut values: Vec = + if let Some(values) = schema.get("enum").and_then(serde_json::Value::as_array) { + values + .iter() + .map(|value| value.as_str().expect("string enum value").to_owned()) + .collect() + } else { + schema + .get("oneOf") + .and_then(serde_json::Value::as_array) + .expect("string enum or oneOf schema") + .iter() + .map(|variant| { + variant["const"] + .as_str() + .expect("string oneOf const") + .to_owned() + }) + .collect() + }; + values.sort(); + values +} + // ── helpers ─────────────────────────────────────────────────────────────────── fn ctx_with_recorder() -> (RunContext, Arc) { @@ -2084,3 +2118,26 @@ async fn run_mode_handle_set_takes_effect_immediately_on_shared_clones() { .await .expect_err("plan mode now denies the same call"); } + +#[cfg(feature = "json-schema")] +#[test] +fn run_mode_schema_and_serde_use_the_wire_discriminants() { + let schema = serde_json::to_value( + schemars::generate::SchemaSettings::draft2020_12() + .for_serialize() + .into_generator() + .into_root_schema_for::(), + ) + .unwrap(); + let mode_values = schema_string_values(&schema, &schema); + assert_eq!(mode_values, vec!["build".to_owned(), "plan".to_owned()]); + assert_eq!( + serde_json::to_value(RunMode::Build).unwrap(), + json!("build") + ); + assert_eq!(serde_json::to_value(RunMode::Plan).unwrap(), json!("plan")); + assert_eq!( + serde_json::from_value::(json!("plan")).unwrap(), + RunMode::Plan + ); +} diff --git a/crates/tinyagents-harness/src/middleware/library/types.rs b/crates/tinyagents-harness/src/middleware/library/types.rs index 9352e3d42..f27f8e03e 100644 --- a/crates/tinyagents-harness/src/middleware/library/types.rs +++ b/crates/tinyagents-harness/src/middleware/library/types.rs @@ -637,7 +637,9 @@ pub struct TraceCounts { /// the middleware's allowlist, which stay available in either mode so a host /// can keep read-only tools and a handful of plan-mode-specific tools (e.g. /// `plan_exit`, `request_plan_review`, `todo`) reachable while planning. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, serde::Serialize, serde::Deserialize)] +#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))] +#[serde(rename_all = "snake_case")] pub enum RunMode { /// Every registered tool may be exposed and executed. The default. #[default] diff --git a/crates/tinyagents-harness/src/progress.rs b/crates/tinyagents-harness/src/progress.rs index 82970ecf6..75f7fae36 100644 --- a/crates/tinyagents-harness/src/progress.rs +++ b/crates/tinyagents-harness/src/progress.rs @@ -463,6 +463,17 @@ pub struct SubagentUsage { pub cost_source: crate::cost::CostSource, } +impl SubagentUsage { + /// The run's cost as a tally, for folding into a parent's. + #[must_use] + pub fn cost(&self) -> crate::cost::CostTally { + crate::cost::CostTally { + known_usd: self.charged_amount_usd, + source: self.cost_source, + } + } +} + fn unknown_cost_source() -> crate::cost::CostSource { crate::cost::CostSource::Unknown } diff --git a/crates/tinyagents-harness/src/progress_tests.rs b/crates/tinyagents-harness/src/progress_tests.rs index c629102b4..bd5d0d6c3 100644 --- a/crates/tinyagents-harness/src/progress_tests.rs +++ b/crates/tinyagents-harness/src/progress_tests.rs @@ -1,5 +1,46 @@ +use super::*; +use crate::cost::CostSource; use crate::progress::AgentProgress; +#[test] +fn subagent_usage_cost_carries_the_known_spend_and_its_certainty() { + let usage = SubagentUsage { + input_tokens: 10, + output_tokens: 5, + cached_input_tokens: 2, + charged_amount_usd: 0.25, + cost_source: CostSource::Charged, + }; + let tally = usage.cost(); + assert_eq!(tally.known_usd, 0.25); + assert_eq!(tally.source, CostSource::Charged); + assert_eq!(tally.usd(), Some(0.25)); +} + +#[test] +fn subagent_usage_with_an_unknown_cost_has_no_usd() { + let usage = SubagentUsage { + cost_source: CostSource::Unknown, + ..SubagentUsage::default() + }; + let tally = usage.cost(); + assert_eq!(tally.source, CostSource::Unknown); + assert_eq!(tally.usd(), None); +} + +#[test] +fn subagent_usage_omits_an_unknown_cost_source_on_the_wire() { + let usage = SubagentUsage { + cost_source: CostSource::Unknown, + ..SubagentUsage::default() + }; + let json = serde_json::to_value(usage).unwrap(); + assert!(json.get("cost_source").is_none()); + // Absent on the wire reads back as an unknown cost, never a charged one. + let round_trip: SubagentUsage = serde_json::from_value(json).unwrap(); + assert_eq!(round_trip, usage); +} + #[test] fn context_compacting_is_a_progress_event() { let event = AgentProgress::ContextCompacting; diff --git a/crates/tinyagents-harness/src/turn_runner/README.md b/crates/tinyagents-harness/src/turn_runner/README.md index e4780b9d8..8311c58fc 100644 --- a/crates/tinyagents-harness/src/turn_runner/README.md +++ b/crates/tinyagents-harness/src/turn_runner/README.md @@ -27,16 +27,26 @@ including fork isolation for parallel runs, and must check a session back in after every successful checkout. The runner persists a deterministic reply id before emitting exactly one terminal event. -The idle guard defaults to 900 seconds and resets on progress. Hosts signal -`WorkStarted` and `WorkFinished` around model or tool work so a quiet but -healthy request is not treated as idle. The optional wall deadline remains a -separate total ceiling. Either guard cancels child work and asks the -`Summarizer` for a bounded grounded close before classifying the turn as failed. +The idle guard defaults to 900 seconds and resets on progress. Hosts can use +`WorkTracker` to fold `AgentProgress` into `WorkStarted` and `WorkFinished` +signals around model or tool work, so a quiet but healthy request is not +treated as idle. The optional wall deadline remains a separate total ceiling. +Either guard cancels child work and asks the `Summarizer` for a bounded +grounded close before classifying the turn as failed. `AgentProgress` is the canonical channel-independent progress vocabulary. Hosts translate it to their own event wire format; this module does not define channel event names or transport payloads. +`TurnRequest

` is itself a [`QueuedMessage`](crate::run_queue::QueuedMessage) +for every payload that implements `RequeuePayload`, so a turn's run queue can be +handed straight to the steering forwarder. A steer the forwarder recovers and +pushes back is rebuilt from a bare id, text and thread label; the runner +restores the scope and thread when it dispatches that item as a fresh turn and +lets the payload adopt the metadata of the turn that was running. +`TurnRunnerConfig::cancel_grace` keeps a cancelled host run polled, with its own +cancellation token already fired, for a bounded time so the host can hand back +and persist the rounds it completed. It defaults to zero (drop at once). ## Host progress and failure adapters Use one `WorkTracker` for each active turn when converting `AgentProgress` to diff --git a/crates/tinyagents-harness/src/turn_runner/mod.rs b/crates/tinyagents-harness/src/turn_runner/mod.rs index a9e00e90b..436e26328 100644 --- a/crates/tinyagents-harness/src/turn_runner/mod.rs +++ b/crates/tinyagents-harness/src/turn_runner/mod.rs @@ -7,6 +7,7 @@ //! owned. mod types; +mod work_tracker; #[cfg(test)] pub(crate) use types::ActiveTurn; @@ -15,8 +16,9 @@ pub use types::{ EventSink, FailureClassification, FailureClassifier, IngressDecision, IngressPolicy, IngressStage, LiveTurn, QueueDisposition, ReplyStore, RequeuePayload, ScopeKey, Summarizer, TerminalEvent, TurnHost, TurnQueueItem, TurnRequest, TurnResult, TurnRunner, TurnRunnerConfig, - TurnRunnerError, TurnSession, TurnSignal, WorkTracker, deterministic_reply_id, + TurnRunnerError, TurnSession, TurnSignal, deterministic_reply_id, }; +pub use work_tracker::WorkTracker; #[cfg(test)] #[path = "mod_tests.rs"] diff --git a/crates/tinyagents-harness/src/turn_runner/mod_tests.rs b/crates/tinyagents-harness/src/turn_runner/mod_tests.rs index 080fce43c..16cb3e6cf 100644 --- a/crates/tinyagents-harness/src/turn_runner/mod_tests.rs +++ b/crates/tinyagents-harness/src/turn_runner/mod_tests.rs @@ -2242,6 +2242,7 @@ async fn idle_guard_restarts_when_quiet_work_finishes() { idle_timeout: std::time::Duration::from_secs(5), deadline: None, close_timeout: std::time::Duration::from_secs(1), + ..TurnRunnerConfig::default() }, )); let mut turn = request(QueueDisposition::Interrupt, false); @@ -2278,6 +2279,7 @@ async fn continuous_progress_cannot_starve_the_wall_deadline() { idle_timeout: std::time::Duration::from_secs(60), deadline: Some(std::time::Duration::from_secs(5)), close_timeout: std::time::Duration::from_secs(1), + ..TurnRunnerConfig::default() }, )); let run = { @@ -2314,6 +2316,7 @@ async fn wall_deadline_is_polled_while_progress_delivery_is_pending() { idle_timeout: std::time::Duration::from_secs(60), deadline: Some(std::time::Duration::from_secs(5)), close_timeout: std::time::Duration::from_secs(1), + ..TurnRunnerConfig::default() }, )); let run = { @@ -2407,6 +2410,7 @@ async fn cancellation_during_grounded_close_prevents_persisting_completion() { idle_timeout: std::time::Duration::from_secs(5), deadline: None, close_timeout: std::time::Duration::from_secs(60), + ..TurnRunnerConfig::default() }, )); let mut turn = request(QueueDisposition::Interrupt, false); @@ -2625,3 +2629,86 @@ async fn queue_snapshot_preserves_lane_fifo_and_does_not_drain_work() { Err(TurnRunnerError::Cancelled) )); } + +#[derive(Default)] +struct UnwindingHost { + started: tokio::sync::Notify, + persisted: std::sync::atomic::AtomicBool, +} + +#[async_trait::async_trait] +impl TurnHost for UnwindingHost { + async fn checkout(&self, request: &TurnRequest) -> Result { + Ok(TurnSession { + session_id: request.thread_id.clone(), + fingerprint: None, + }) + } + async fn checkout_fork(&self, request: &TurnRequest) -> Result { + self.checkout(request).await + } + fn run_origin(&self, _request: &TurnRequest) -> serde_json::Value { + serde_json::Value::Null + } + async fn run( + &self, + _request: &TurnRequest, + _session: &TurnSession, + _origin: &serde_json::Value, + cancellation: crate::CancellationToken, + _queue: Arc>, + _progress: tokio::sync::mpsc::Sender, + ) -> Result { + self.started.notify_one(); + cancellation.cancelled().await; + // Hand back and persist the rounds completed so far. + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + self.persisted + .store(true, std::sync::atomic::Ordering::SeqCst); + Err("stopped".into()) + } + async fn checkin(&self, _request: &TurnRequest) {} + async fn cancel_children(&self, _request: &TurnRequest) {} +} + +async fn cancel_unwinding_turn(grace: std::time::Duration) -> bool { + let host = Arc::new(UnwindingHost::default()); + let runner = Arc::new(TurnRunner::new( + host.clone(), + Arc::new(Ingress::default()), + Arc::new(Replies::default()), + Arc::new(Events::default()), + Arc::new(Failures), + Arc::new(Summary), + TurnRunnerConfig { + cancel_grace: grace, + ..TurnRunnerConfig::default() + }, + )); + let primary = { + let runner = runner.clone(); + tokio::spawn(async move { + runner + .run(request(QueueDisposition::Interrupt, false)) + .await + }) + }; + host.started.notified().await; + let req = request(QueueDisposition::Interrupt, false); + assert!(runner.cancel(&req.scope, &req.thread_id, None).await); + assert!(matches!( + primary.await.unwrap(), + Err(TurnRunnerError::Cancelled) + )); + host.persisted.load(std::sync::atomic::Ordering::SeqCst) +} + +#[tokio::test] +async fn cancelled_run_unwinds_within_the_cancel_grace() { + assert!(cancel_unwinding_turn(std::time::Duration::from_secs(2)).await); +} + +#[tokio::test] +async fn cancelled_run_is_dropped_at_once_without_a_cancel_grace() { + assert!(!cancel_unwinding_turn(std::time::Duration::ZERO).await); +} diff --git a/crates/tinyagents-harness/src/turn_runner/types.rs b/crates/tinyagents-harness/src/turn_runner/types.rs index 00efbe77b..b737b930b 100644 --- a/crates/tinyagents-harness/src/turn_runner/types.rs +++ b/crates/tinyagents-harness/src/turn_runner/types.rs @@ -1,5 +1,5 @@ use std::{ - collections::{HashMap, HashSet}, + collections::HashMap, future::Future, hash::Hash, sync::{Arc, Weak}, @@ -257,49 +257,6 @@ pub trait RequeuePayload: Sized { fn restore_thread_id(&mut self, _thread_id: &str) {} } -/// Folds progress events into balanced work-liveness signals for the runner. -/// -/// A turn is considered busy while a model iteration or tool call is active. -/// Overlapping work emits one `WorkStarted` transition and one `WorkFinished` -/// transition when the last item settles. Duplicate start/completion events -/// do not produce unbalanced signals. -#[derive(Debug, Default)] -pub struct WorkTracker { - active: HashSet, -} - -impl WorkTracker { - /// Record an event and return a signal only when busy/idle state changes. - pub fn observe(&mut self, event: &AgentProgress) -> Option { - let was_active = !self.active.is_empty(); - match event { - AgentProgress::IterationStarted { iteration, .. } => { - self.active.insert(format!("model:{iteration}")); - } - AgentProgress::ModelCallCompleted { - iteration, - subagent_task_id: None, - .. - } => { - self.active.remove(&format!("model:{iteration}")); - } - AgentProgress::ToolCallStarted { call_id, .. } => { - self.active.insert(format!("tool:{call_id}")); - } - AgentProgress::ToolCallCompleted { call_id, .. } => { - self.active.remove(&format!("tool:{call_id}")); - } - AgentProgress::TurnCompleted { .. } => self.active.clear(), - _ => {} - } - match (was_active, !self.active.is_empty()) { - (false, true) => Some(TurnSignal::WorkStarted), - (true, false) => Some(TurnSignal::WorkFinished), - _ => None, - } - } -} - #[async_trait] pub trait TurnHost

>: Send + Sync { /// Check out or create the session for this request and verify its fingerprint. @@ -407,6 +364,11 @@ pub struct TurnRunnerConfig { pub idle_timeout: Duration, pub deadline: Option, pub close_timeout: Duration, + /// How long a cancelled host run keeps being polled, with its own + /// cancellation token already fired, before it is dropped. A host uses the + /// window to hand back and persist the rounds it completed. Zero drops the + /// run immediately. + pub cancel_grace: Duration, } impl Default for TurnRunnerConfig { @@ -415,6 +377,7 @@ impl Default for TurnRunnerConfig { idle_timeout: Duration::from_secs(900), deadline: None, close_timeout: Duration::from_secs(60), + cancel_grace: Duration::ZERO, } } } @@ -1344,6 +1307,17 @@ where _ = cancellation.cancelled() => { task_cancellation.cancel(); self.host.cancel_children(request).await; + if !self.config.cancel_grace.is_zero() { + // Let the host unwind with its token fired; its result + // is discarded either way. + let unwound = tokio::time::timeout(self.config.cancel_grace, &mut run).await; + tracing::debug!( + thread_id = %request.thread_id, + request_id = %request.request_id, + unwound = unwound.is_ok(), + "turn runner: cancelled run left its grace window" + ); + } return Err("turn cancelled".into()); } _ = async { if let Some(deadline) = deadline.at { tokio::time::sleep_until(deadline).await } else { std::future::pending().await } }, if deadline.at.is_some() => { diff --git a/crates/tinyagents-harness/src/turn_runner/work_tracker.rs b/crates/tinyagents-harness/src/turn_runner/work_tracker.rs new file mode 100644 index 000000000..6789191b5 --- /dev/null +++ b/crates/tinyagents-harness/src/turn_runner/work_tracker.rs @@ -0,0 +1,65 @@ +//! Folds generic agent progress into balanced work-liveness signals. + +use std::collections::HashSet; + +use crate::progress::AgentProgress; + +use super::TurnSignal; + +/// Tracks whether model or tool work is currently open for a turn. +/// +/// The tracker emits [`TurnSignal::WorkStarted`] when progress first opens +/// work and [`TurnSignal::WorkFinished`] when the last open operation closes. +/// This lets a turn runner distinguish a quiet but healthy provider/tool call +/// from an idle turn without depending on a channel-specific event format. +#[derive(Debug, Default)] +pub struct WorkTracker { + model_call_open: bool, + open_tool_calls: HashSet, +} + +impl WorkTracker { + fn open(&self) -> bool { + self.model_call_open || !self.open_tool_calls.is_empty() + } + + /// Folds one progress event into the open-work state. + /// + /// `IterationStarted` closes tool calls whose completion may have been + /// dropped, while a parent `ToolCallStarted` closes the model call that + /// requested it. Subagent model completions leave parent work unchanged. + pub fn observe(&mut self, event: &AgentProgress) -> Option { + let was_open = self.open(); + match event { + AgentProgress::IterationStarted { .. } => { + self.open_tool_calls.clear(); + self.model_call_open = true; + } + AgentProgress::ModelCallCompleted { + subagent_task_id: None, + .. + } => self.model_call_open = false, + AgentProgress::ToolCallStarted { call_id, .. } => { + self.model_call_open = false; + self.open_tool_calls.insert(call_id.clone()); + } + AgentProgress::ToolCallCompleted { call_id, .. } => { + self.open_tool_calls.remove(call_id); + } + AgentProgress::TurnCompleted { .. } => { + self.open_tool_calls.clear(); + self.model_call_open = false; + } + _ => {} + } + match (was_open, self.open()) { + (false, true) => Some(TurnSignal::WorkStarted), + (true, false) => Some(TurnSignal::WorkFinished), + _ => None, + } + } +} + +#[cfg(test)] +#[path = "work_tracker_tests.rs"] +mod tests; diff --git a/crates/tinyagents-harness/src/turn_runner/work_tracker_tests.rs b/crates/tinyagents-harness/src/turn_runner/work_tracker_tests.rs new file mode 100644 index 000000000..4bf4602b5 --- /dev/null +++ b/crates/tinyagents-harness/src/turn_runner/work_tracker_tests.rs @@ -0,0 +1,152 @@ +use super::*; + +fn iteration_started() -> AgentProgress { + AgentProgress::IterationStarted { + iteration: 1, + max_iterations: 50, + } +} + +fn model_call_completed(subagent_task_id: Option<&str>) -> AgentProgress { + AgentProgress::ModelCallCompleted { + model: "m".into(), + provider_id: "managed".into(), + subagent_task_id: subagent_task_id.map(str::to_string), + input: None, + output: None, + iteration: 1, + input_tokens: 0, + output_tokens: 0, + cached_input_tokens: 0, + cache_creation_tokens: 0, + reasoning_tokens: 0, + cost_usd: None, + } +} + +fn tool_started(call_id: &str) -> AgentProgress { + AgentProgress::ToolCallStarted { + call_id: call_id.into(), + tool_name: "shell".into(), + arguments: serde_json::json!({}), + iteration: 1, + display_label: None, + display_detail: None, + } +} + +fn tool_completed(call_id: &str) -> AgentProgress { + AgentProgress::ToolCallCompleted { + call_id: call_id.into(), + tool_name: "shell".into(), + success: true, + output_chars: 0, + output: String::new(), + arguments: None, + elapsed_ms: 0, + iteration: 1, + failure: None, + display_label: None, + display_detail: None, + structured: None, + } +} + +fn signal(tracker: &mut WorkTracker, event: &AgentProgress) -> Option<&'static str> { + tracker.observe(event).map(|signal| match signal { + TurnSignal::WorkStarted => "started", + TurnSignal::WorkFinished => "finished", + TurnSignal::Progress(_) => "progress", + }) +} + +/// A provider retrying a failed call remains work until the parent call ends. +#[test] +fn an_open_model_call_is_work_until_it_completes() { + let mut tracker = WorkTracker::default(); + assert_eq!(signal(&mut tracker, &iteration_started()), Some("started")); + assert_eq!( + signal( + &mut tracker, + &AgentProgress::TextDelta { + delta: "hi".into(), + iteration: 1, + } + ), + None + ); + assert_eq!( + signal(&mut tracker, &model_call_completed(None)), + Some("finished") + ); +} + +#[test] +fn a_subagent_model_call_does_not_close_parent_work() { + let mut tracker = WorkTracker::default(); + signal(&mut tracker, &iteration_started()); + assert_eq!( + signal(&mut tracker, &model_call_completed(Some("task-1"))), + None + ); + assert_eq!( + signal(&mut tracker, &model_call_completed(None)), + Some("finished") + ); +} + +/// Parallel tool calls stay open until the last completion arrives. +#[test] +fn open_tool_calls_finish_only_after_the_last_call_completes() { + let mut tracker = WorkTracker::default(); + assert_eq!(signal(&mut tracker, &tool_started("a")), Some("started")); + assert_eq!(signal(&mut tracker, &tool_started("b")), None); + assert_eq!(signal(&mut tracker, &tool_completed("a")), None); + assert_eq!(signal(&mut tracker, &tool_completed("b")), Some("finished")); +} + +/// A tool start hands work over from the model call without an idle gap. +#[test] +fn tool_start_closes_its_model_call_without_finishing_work() { + let mut tracker = WorkTracker::default(); + signal(&mut tracker, &iteration_started()); + assert_eq!(signal(&mut tracker, &tool_started("a")), None); + assert_eq!(signal(&mut tracker, &tool_completed("a")), Some("finished")); +} + +/// The next iteration closes a tool whose completion was lost under backpressure. +#[test] +fn the_next_iteration_closes_a_tool_with_missing_completion() { + let mut tracker = WorkTracker::default(); + signal(&mut tracker, &tool_started("a")); + assert_eq!(signal(&mut tracker, &iteration_started()), None); + assert_eq!( + signal(&mut tracker, &model_call_completed(None)), + Some("finished") + ); +} + +#[test] +fn turn_completion_closes_all_open_work() { + let mut tracker = WorkTracker::default(); + signal(&mut tracker, &iteration_started()); + assert_eq!( + signal(&mut tracker, &AgentProgress::turn_completed(1)), + Some("finished") + ); + + signal(&mut tracker, &tool_started("a")); + assert_eq!( + signal(&mut tracker, &AgentProgress::turn_completed(1)), + Some("finished") + ); +} + +#[test] +fn duplicate_starts_and_unknown_completions_do_not_emit_spurious_edges() { + let mut tracker = WorkTracker::default(); + assert_eq!(signal(&mut tracker, &tool_started("a")), Some("started")); + assert_eq!(signal(&mut tracker, &tool_started("a")), None); + assert_eq!(signal(&mut tracker, &tool_completed("missing")), None); + assert_eq!(signal(&mut tracker, &tool_completed("a")), Some("finished")); +} diff --git a/crates/tinyagents-integration-tests/tests/feature_session_transcript.rs b/crates/tinyagents-integration-tests/tests/feature_session_transcript.rs index ad7e0be99..b8d5dcca0 100644 --- a/crates/tinyagents-integration-tests/tests/feature_session_transcript.rs +++ b/crates/tinyagents-integration-tests/tests/feature_session_transcript.rs @@ -26,6 +26,7 @@ fn meta(turn_count: usize, input_tokens: u64, output_tokens: u64) -> TranscriptM updated: format!("2026-09-19T10:00:0{turn_count}Z"), turn_count, prefix_message_count: None, + prefix_reset: false, input_tokens, output_tokens, cached_input_tokens: 3, diff --git a/crates/tinyagents-integration-tests/tests/session_conformance.rs b/crates/tinyagents-integration-tests/tests/session_conformance.rs index 3bcb4c0ae..0a24fd5c5 100644 --- a/crates/tinyagents-integration-tests/tests/session_conformance.rs +++ b/crates/tinyagents-integration-tests/tests/session_conformance.rs @@ -47,6 +47,7 @@ fn contract_meta() -> TranscriptMeta { updated: String::new(), turn_count: 0, prefix_message_count: None, + prefix_reset: false, input_tokens: 0, output_tokens: 0, cached_input_tokens: 0, diff --git a/crates/tinyagents-orchestration/src/lib.rs b/crates/tinyagents-orchestration/src/lib.rs index ae0cc682f..a4b446a1e 100644 --- a/crates/tinyagents-orchestration/src/lib.rs +++ b/crates/tinyagents-orchestration/src/lib.rs @@ -13,6 +13,7 @@ //! `orchestration -> {harness, session, tasks, storage ports}`. Lower-level //! crates never depend on this composition layer. +pub mod run_progress_ledger; pub mod steering; pub mod subagent; diff --git a/crates/tinyagents-runtime/src/hooks_tests.rs b/crates/tinyagents-runtime/src/hooks_tests.rs index 95cd4e6d5..71dbfbaaf 100644 --- a/crates/tinyagents-runtime/src/hooks_tests.rs +++ b/crates/tinyagents-runtime/src/hooks_tests.rs @@ -67,6 +67,7 @@ async fn callback_adapter_implements_runtime_session_hooks() { history: &[], raw_history: &raw_history, prefix: &prefix, + prefix_is_frozen: false, transcript_target: None, committed_turns: 4, resumed: true, @@ -136,6 +137,7 @@ async fn callback_adapter_propagates_hook_errors_through_runtime_trait() { history: &[], raw_history: &raw_history, prefix: &prefix, + prefix_is_frozen: false, transcript_target: None, committed_turns: 0, resumed: false, diff --git a/crates/tinyagents-runtime/src/lib_tests.rs b/crates/tinyagents-runtime/src/lib_tests.rs index 595773769..aee817595 100644 --- a/crates/tinyagents-runtime/src/lib_tests.rs +++ b/crates/tinyagents-runtime/src/lib_tests.rs @@ -193,6 +193,7 @@ fn meta() -> TranscriptMeta { updated: "then".into(), turn_count: 0, prefix_message_count: None, + prefix_reset: false, input_tokens: 0, output_tokens: 0, cached_input_tokens: 0, diff --git a/crates/tinyagents-runtime/src/session.rs b/crates/tinyagents-runtime/src/session.rs index 0c820bacd..f5effbc96 100644 --- a/crates/tinyagents-runtime/src/session.rs +++ b/crates/tinyagents-runtime/src/session.rs @@ -26,6 +26,10 @@ pub struct Session { driver: Arc>, codec: Option>>, hooks: Arc>, + /// Prefix configured when this session instance was constructed. A reset + /// discovered through another live instance clears the old frozen prefix + /// but keeps this current host configuration available to its hooks. + configured_prefix: PrefixSnapshot, post_commit_sinks: Vec>>, identity_config: Option>, established_identity: Option, @@ -67,6 +71,7 @@ impl Session { target: Option, identity_config: Option>, ) -> Self { + let configured_prefix = prefix.frozen(); Self { driver, codec, @@ -75,8 +80,9 @@ impl Session { identity_config, established_identity: None, identity: None, - history: prefix.messages().to_vec(), - prefix: prefix.frozen(), + history: configured_prefix.messages().to_vec(), + prefix: configured_prefix.clone(), + configured_prefix, default_tools, persisted: Vec::new(), reconcile_prior: Vec::new(), @@ -381,8 +387,39 @@ impl Session { .as_ref() .is_some_and(|session| session.generation > 0) || transcript.meta.parent_session_id.is_some(); + // An explicit history reset suppresses prompt-boundary recovery from + // the sealed parent. The marker distinguishes it from an ordinary + // empty successor whose legacy prompt still needs to be inferred. + let reset_head = transcript.meta.prefix_reset && recorded_boundary.is_none(); + let changed_head = self + .transcript + .as_ref() + .is_some_and(|bound| bound.path() != read.path()); + if changed_head { + // Another live Session instance may have advanced this identity to + // a new generation. Rebuild cached transcript state from that head + // instead of retaining the old handle's turn count or tool belt. + // An explicit reset head is unfrozen and starts with this instance's + // configured prefix; a committed successor below restores its + // authoritative persisted prefix (including an explicit Some(0)). + if reset_head || recorded_boundary.is_some() { + self.prefix = if reset_head { + self.configured_prefix.clone() + } else { + PrefixSnapshot::default() + }; + } + self.history = self.prefix.messages().to_vec(); + self.persisted.clear(); + self.persisted_prefix_len = None; + self.recorded_tools = None; + self.recorded_tools_json = None; + self.committed_turns = 0; + self.transcript = None; + } let mut recovered_boundary = None; - if recorded_boundary.is_none() && cached_boundary.is_none() && compacted_head { + if recorded_boundary.is_none() && cached_boundary.is_none() && compacted_head && !reset_head + { // Without the sealed root there is no safe boundary in a head // containing a System summary. If a replacement prefix was // supplied, fail rather than replaying unverifiable old System @@ -650,6 +687,77 @@ impl Session { }) } + /// Starts a fresh model context for the bound durable session. + /// + /// The current generation is sealed and retained as the parent of an + /// empty successor. Repeating this operation on an empty or absent + /// session is a no-op, so it never manufactures a timestamped identity or + /// an unbounded chain of empty generations. + pub async fn reset_history(&mut self) -> Result { + let Some(target) = self.target.as_ref() else { + return Ok(false); + }; + let Some(bound) = target.session.as_ref() else { + return Ok(false); + }; + + // Generation locks are keyed by the first generation, so resetting a + // current head serializes with turns bound to either the root or head. + let root = bound.first_generation(); + let _turn = lock_session_turn(target.locator.as_ref(), &root).await; + let head = target.locator.head_generation(bound); + let Some(read) = target.locator.read_session_transcript(&head) else { + return Ok(false); + }; + let transcript = read + .read_session() + .map_err(|error| RuntimeError::Persistence(error.to_string()))?; + let Some(transcript) = transcript else { + return Ok(false); + }; + if transcript.messages.is_empty() + && transcript.meta.prefix_message_count.is_none() + && transcript.tools.is_none() + && transcript.meta.turn_count == 0 + { + return Ok(false); + } + + let mut seed = transcript.meta.clone(); + seed.turn_count = 0; + seed.prefix_message_count = None; + seed.prefix_reset = true; + seed.session_id = None; + seed.parent_session_id = None; + let (successor, handle) = target + .locator + .begin_generation_from_baseline(&head, seed, &transcript.messages) + .map_err(|error| RuntimeError::Persistence(error.to_string()))?; + // Generation allocation reserves the successor but leaves it + // unwritten. Persist an explicit empty context now so a process restart + // before the next user turn still resolves the successor as the head. + handle + .replace(&[]) + .map_err(|error| RuntimeError::Persistence(error.to_string()))?; + + let target = self.target.as_mut().expect("target was checked above"); + target.session = Some(successor.clone()); + target.meta.session_id = Some(successor.session_id()); + target.meta.parent_session_id = successor.parent_session_id(); + target.meta.turn_count = 0; + target.meta.prefix_message_count = None; + target.meta.prefix_reset = true; + self.transcript = Some(handle); + self.prefix = PrefixSnapshot::default(); + self.history.clear(); + self.persisted.clear(); + self.persisted_prefix_len = None; + self.recorded_tools = None; + self.recorded_tools_json = None; + self.committed_turns = 0; + Ok(true) + } + /// Runs one user turn: wraps `input` in a [`SessionTurnRequest`] and drives /// [`Session::turn`], the single entry hosts call instead of owning a turn /// loop. Host policy rides in `options` and the session's hooks. @@ -1042,6 +1150,8 @@ impl Session { history: &self.history, raw_history: &self.persisted, prefix: &self.prefix, + prefix_is_frozen: self.persisted_prefix_len.is_some() + || !self.prefix.messages().is_empty(), transcript_target: self.target.as_ref(), committed_turns: self.committed_turns, resumed, @@ -1179,6 +1289,7 @@ impl Session { }; meta.turn_count += 1; meta.prefix_message_count = Some(prefix_len); + meta.prefix_reset = false; meta.updated = chrono::Utc::now().to_rfc3339(); // Record every ordinary turn's declarations. Comparing against this // session's cached snapshot is unsafe when another live Session has diff --git a/crates/tinyagents-runtime/src/session_tests.rs b/crates/tinyagents-runtime/src/session_tests.rs index ffada9309..00aba6546 100644 --- a/crates/tinyagents-runtime/src/session_tests.rs +++ b/crates/tinyagents-runtime/src/session_tests.rs @@ -814,6 +814,7 @@ fn meta() -> TranscriptMeta { updated: "then".into(), turn_count: 0, prefix_message_count: None, + prefix_reset: false, input_tokens: 0, output_tokens: 0, cached_input_tokens: 0, @@ -832,6 +833,7 @@ impl TranscriptCodec for RoleCodec { .messages .iter() .map(|row| match row.role.as_str() { + "system" => Message::system(&row.content), "user" => Message::user(&row.content), _ => Message::assistant(&row.content), }) @@ -841,15 +843,31 @@ impl TranscriptCodec for RoleCodec { fn reconcile( &self, prior: &[TranscriptMessage], - _: &[Message], + previous: &[Message], next: &[Message], _: &crate::TranscriptTurnOptions, ) -> Result, RuntimeError> { let mut rows = prior.to_vec(); + let prior_system_count = prior.iter().take_while(|row| row.role == "system").count(); + let next_system_count = next + .iter() + .take_while(|message| matches!(message, Message::System(_))) + .count(); + if prior_system_count == 0 && next_system_count > 0 { + rows.splice( + 0..0, + next[..next_system_count] + .iter() + .map(|message| TranscriptMessage::new("system", message.text())), + ); + } rows.extend( - next[prior.len().min(next.len())..] + next[next_system_count + .saturating_add(previous.len()) + .min(next.len())..] .iter() .map(|message| match message { + Message::System(_) => TranscriptMessage::new("system", message.text()), Message::User(_) => TranscriptMessage::user(message.text()), _ => TranscriptMessage::assistant(message.text()), }), @@ -858,6 +876,709 @@ impl TranscriptCodec for RoleCodec { } } +struct CapturingDriver(Arc>>>); + +#[async_trait] +impl SessionDriver for CapturingDriver { + async fn execute(&self, request: DriverRequest) -> Result { + self.0.lock().unwrap().push(request.history.clone()); + let mut history = request.history; + history.push(Message::assistant("reply")); + Ok(DriverOutcome { + outcome: None, + history, + output: Some("reply".into()), + partial: None, + interrupted: false, + tool_outcomes: Vec::new(), + }) + } +} + +struct PrefixWhenUnfrozen(&'static str); + +#[async_trait] +impl SessionHooks for PrefixWhenUnfrozen { + async fn before_turn( + &self, + _: &mut SessionTurnRequest, + _: &mut TurnOptions, + state: SessionStateView<'_>, + ) -> Result { + Ok(TurnPreparation { + prefix: (!state.prefix_is_frozen) + .then(|| PrefixSnapshot::new(vec![Message::system(self.0)])), + ..TurnPreparation::default() + }) + } + + async fn on_terminal(&self, _: SessionTerminal) -> Result<(), RuntimeError> { + Ok(()) + } + + async fn before_commit( + &self, + _: &SessionTurnOutcome, + _: &TranscriptTurnOptions, + ) -> Result<(), RuntimeError> { + Ok(()) + } +} + +struct MutablePrefix(Arc>); + +#[async_trait] +impl SessionHooks for MutablePrefix { + async fn before_turn( + &self, + _: &mut SessionTurnRequest, + _: &mut TurnOptions, + state: SessionStateView<'_>, + ) -> Result { + let prompt = self.0.lock().unwrap().clone(); + Ok(TurnPreparation { + prefix: (!state.prefix_is_frozen) + .then(|| PrefixSnapshot::new(vec![Message::system(prompt)])), + ..TurnPreparation::default() + }) + } + + async fn on_terminal(&self, _: SessionTerminal) -> Result<(), RuntimeError> { + Ok(()) + } + + async fn before_commit( + &self, + _: &SessionTurnOutcome, + _: &TranscriptTurnOptions, + ) -> Result<(), RuntimeError> { + Ok(()) + } +} + +#[tokio::test] +async fn reset_history_seals_old_turns_and_starts_a_fresh_resumable_context() { + let locator = memory(); + let session_ref = SessionRef::scoped("reset-thread", "agent-id"); + let first_requests = Arc::new(Mutex::new(Vec::new())); + let mut session = SessionBuilder::new(Arc::new(CapturingDriver(first_requests))) + .codec(Arc::new(RoleCodec)) + .hooks(Arc::new(PrefixWhenUnfrozen("original prompt"))) + .session(locator.clone(), session_ref.clone(), meta()) + .build() + .unwrap(); + session + .turn( + SessionTurnRequest::new(Message::user("before reset")), + TurnOptions { + resume: ResumeMode::Session, + session: Some(session_ref.clone()), + ..TurnOptions::default() + }, + ) + .await + .unwrap(); + + let old_head = locator.head_generation(&session_ref); + let old_before = locator + .read_session_transcript(&old_head) + .unwrap() + .read_session() + .unwrap() + .unwrap(); + assert_eq!(old_before.messages[0].content, "original prompt"); + assert!(session.reset_history().await.unwrap()); + + let new_head = locator.head_generation(&session_ref); + assert_eq!(new_head.generation, old_head.generation + 1); + assert_eq!(new_head.parent_session_id(), Some(old_head.session_id())); + let old_after = locator + .read_session_transcript(&old_head) + .unwrap() + .read_session() + .unwrap() + .unwrap(); + assert_eq!( + serde_json::to_value((&old_after.meta, &old_after.messages, &old_after.tools,)).unwrap(), + serde_json::to_value((&old_before.meta, &old_before.messages, &old_before.tools,)).unwrap(), + ); + assert!( + locator + .read_session_transcript(&new_head) + .unwrap() + .read_session() + .unwrap() + .unwrap() + .messages + .is_empty() + ); + + assert!(!session.reset_history().await.unwrap()); + assert_eq!(locator.head_generation(&session_ref), new_head); + + let reset_requests = Arc::new(Mutex::new(Vec::new())); + let mut after_reset = SessionBuilder::new(Arc::new(CapturingDriver(reset_requests.clone()))) + .codec(Arc::new(RoleCodec)) + .hooks(Arc::new(PrefixWhenUnfrozen("fresh prompt"))) + .session(locator.clone(), session_ref.clone(), meta()) + .build() + .unwrap(); + after_reset + .turn( + SessionTurnRequest::new(Message::user("after reset")), + TurnOptions { + resume: ResumeMode::Session, + session: Some(session_ref.clone()), + ..TurnOptions::default() + }, + ) + .await + .unwrap(); + let reset_request = reset_requests.lock().unwrap()[0].clone(); + assert_eq!(reset_request[0].text(), "fresh prompt"); + assert_eq!(reset_request[1].text(), "after reset"); + assert!( + !reset_request + .iter() + .any(|message| message.text() == "before reset") + ); + + let resumed_requests = Arc::new(Mutex::new(Vec::new())); + let mut resumed = SessionBuilder::new(Arc::new(CapturingDriver(resumed_requests.clone()))) + .codec(Arc::new(RoleCodec)) + .hooks(Arc::new(PrefixWhenUnfrozen("replacement prompt"))) + .session(locator.clone(), session_ref.clone(), meta()) + .build() + .unwrap(); + resumed + .turn( + SessionTurnRequest::new(Message::user("cold resume")), + TurnOptions { + resume: ResumeMode::Session, + session: Some(session_ref.clone()), + ..TurnOptions::default() + }, + ) + .await + .unwrap(); + let cold_request = resumed_requests.lock().unwrap()[0].clone(); + assert_eq!(cold_request[0].text(), "fresh prompt"); + assert!( + !cold_request + .iter() + .any(|message| message.text() == "replacement prompt") + ); + assert!( + !cold_request + .iter() + .any(|message| message.text() == "before reset") + ); +} + +#[tokio::test] +async fn reset_by_another_live_session_rebinds_cached_prompt_and_turn_state() { + let locator = memory(); + let session_ref = SessionRef::scoped("live-reset", "agent-id"); + let prompt = Arc::new(Mutex::new("original prompt".to_owned())); + let requests = Arc::new(Mutex::new(Vec::new())); + let mut live = SessionBuilder::new(Arc::new(CapturingDriver(requests.clone()))) + .codec(Arc::new(RoleCodec)) + .hooks(Arc::new(MutablePrefix(prompt.clone()))) + .session(locator.clone(), session_ref.clone(), meta()) + .build() + .unwrap(); + live.turn( + SessionTurnRequest::new(Message::user("before reset")), + TurnOptions { + resume: ResumeMode::Session, + session: Some(session_ref.clone()), + ..TurnOptions::default() + }, + ) + .await + .unwrap(); + + let mut resetter = + SessionBuilder::new(Arc::new(CapturingDriver(Arc::new(Mutex::new(Vec::new()))))) + .codec(Arc::new(RoleCodec)) + .session(locator.clone(), session_ref.clone(), meta()) + .build() + .unwrap(); + assert!(resetter.reset_history().await.unwrap()); + *prompt.lock().unwrap() = "replacement prompt".to_owned(); + + live.turn( + SessionTurnRequest::new(Message::user("after reset")), + TurnOptions { + resume: ResumeMode::Session, + session: Some(session_ref.clone()), + ..TurnOptions::default() + }, + ) + .await + .unwrap(); + let history = requests.lock().unwrap()[1].clone(); + assert_eq!(history[0].text(), "replacement prompt"); + assert_eq!(history[1].text(), "after reset"); + assert!( + !history + .iter() + .any(|message| message.text() == "original prompt") + ); + assert!( + !history + .iter() + .any(|message| message.text() == "before reset") + ); + let head = locator.head_generation(&session_ref); + let transcript = locator + .read_session_transcript(&head) + .unwrap() + .read_session() + .unwrap() + .unwrap(); + assert_eq!(transcript.meta.turn_count, 1); + assert_eq!(transcript.meta.prefix_message_count, Some(1)); +} + +#[tokio::test] +async fn a_live_session_rebinds_after_another_session_commits_the_reset_head() { + let locator = memory(); + let session_ref = SessionRef::scoped("live-reset-commit", "agent-id"); + let prompt = Arc::new(Mutex::new("original prompt".to_owned())); + let requests = Arc::new(Mutex::new(Vec::new())); + let mut live = SessionBuilder::new(Arc::new(CapturingDriver(requests.clone()))) + .codec(Arc::new(RoleCodec)) + .hooks(Arc::new(MutablePrefix(prompt.clone()))) + .session(locator.clone(), session_ref.clone(), meta()) + .build() + .unwrap(); + live.turn( + SessionTurnRequest::new(Message::user("before reset")), + TurnOptions { + resume: ResumeMode::Session, + session: Some(session_ref.clone()), + ..TurnOptions::default() + }, + ) + .await + .unwrap(); + + let mut resetter = + SessionBuilder::new(Arc::new(CapturingDriver(Arc::new(Mutex::new(Vec::new()))))) + .codec(Arc::new(RoleCodec)) + .session(locator.clone(), session_ref.clone(), meta()) + .build() + .unwrap(); + assert!(resetter.reset_history().await.unwrap()); + + let mut first_after_reset = + SessionBuilder::new(Arc::new(CapturingDriver(Arc::new(Mutex::new(Vec::new()))))) + .codec(Arc::new(RoleCodec)) + .hooks(Arc::new(PrefixWhenUnfrozen("committed reset prompt"))) + .session(locator.clone(), session_ref.clone(), meta()) + .build() + .unwrap(); + first_after_reset + .turn( + SessionTurnRequest::new(Message::user("first after reset")), + TurnOptions { + resume: ResumeMode::Session, + session: Some(session_ref.clone()), + ..TurnOptions::default() + }, + ) + .await + .unwrap(); + + *prompt.lock().unwrap() = "stale replacement prompt".to_owned(); + live.turn( + SessionTurnRequest::new(Message::user("after concurrent commit")), + TurnOptions { + resume: ResumeMode::Session, + session: Some(session_ref.clone()), + ..TurnOptions::default() + }, + ) + .await + .unwrap(); + let history = requests.lock().unwrap()[1].clone(); + assert_eq!(history[0].text(), "committed reset prompt"); + assert_eq!(history[1].text(), "first after reset"); + assert_eq!(history[2].text(), "reply"); + assert_eq!(history[3].text(), "after concurrent commit"); + assert!(!history.iter().any(|message| { + matches!(message, Message::System(_)) + && (message.text() == "original prompt" || message.text() == "stale replacement prompt") + })); + let head = locator.head_generation(&session_ref); + let transcript = locator + .read_session_transcript(&head) + .unwrap() + .read_session() + .unwrap() + .unwrap(); + assert_eq!(transcript.meta.turn_count, 2); +} + +#[tokio::test] +async fn reset_history_advances_an_explicit_frozen_empty_prefix() { + let locator = memory(); + let session_ref = SessionRef::scoped("empty-prefix-reset", "agent-id"); + let mut empty_prefix_meta = meta(); + empty_prefix_meta.prefix_message_count = Some(0); + let empty_head = locator + .open_stem(&session_stem(&session_ref), empty_prefix_meta) + .unwrap(); + empty_head.replace(&[]).unwrap(); + let mut session = + SessionBuilder::new(Arc::new(CapturingDriver(Arc::new(Mutex::new(Vec::new()))))) + .codec(Arc::new(RoleCodec)) + .session(locator.clone(), session_ref.clone(), meta()) + .build() + .unwrap(); + + assert!(session.reset_history().await.unwrap()); + let reset_head = locator.head_generation(&session_ref); + assert_eq!(reset_head.generation, 1); + assert_eq!( + locator + .read_session_transcript(&reset_head) + .unwrap() + .read_session() + .unwrap() + .unwrap() + .meta + .prefix_message_count, + None, + ); + assert!( + locator + .read_session_transcript(&reset_head) + .unwrap() + .read_session() + .unwrap() + .unwrap() + .meta + .prefix_reset + ); + + let requests = Arc::new(Mutex::new(Vec::new())); + let mut cold = SessionBuilder::new(Arc::new(CapturingDriver(requests.clone()))) + .codec(Arc::new(RoleCodec)) + .hooks(Arc::new(PrefixWhenUnfrozen("after empty reset"))) + .session(locator, session_ref.clone(), meta()) + .build() + .unwrap(); + cold.turn( + SessionTurnRequest::new(Message::user("new turn")), + TurnOptions { + resume: ResumeMode::Session, + session: Some(session_ref), + ..TurnOptions::default() + }, + ) + .await + .unwrap(); + assert_eq!(requests.lock().unwrap()[0][0].text(), "after empty reset"); +} + +#[tokio::test] +async fn an_empty_compaction_without_reset_marker_recovers_explicit_parent_boundary() { + let locator = memory(); + let session_ref = SessionRef::scoped("empty-compaction", "agent-id"); + let mut root_meta = meta(); + root_meta.prefix_message_count = Some(0); + let root = locator + .open_stem(&session_stem(&session_ref), root_meta) + .unwrap(); + root.replace(&[]).unwrap(); + let (successor, empty_head) = locator + .begin_generation_from_baseline(&session_ref, meta(), &[]) + .unwrap(); + empty_head.replace(&[]).unwrap(); + + let requests = Arc::new(Mutex::new(Vec::new())); + let mut resumed = SessionBuilder::new(Arc::new(CapturingDriver(requests.clone()))) + .codec(Arc::new(RoleCodec)) + .hooks(Arc::new(PrefixWhenUnfrozen( + "must not replace frozen empty", + ))) + .session(locator, session_ref.clone(), meta()) + .build() + .unwrap(); + resumed + .turn( + SessionTurnRequest::new(Message::user("after compaction")), + TurnOptions { + resume: ResumeMode::Session, + session: Some(session_ref), + ..TurnOptions::default() + }, + ) + .await + .unwrap(); + let request = requests.lock().unwrap()[0].clone(); + assert_eq!(request.len(), 1); + assert_eq!(request[0].text(), "after compaction"); + assert_eq!(successor.generation, 1); +} + +#[tokio::test] +async fn reset_waits_for_an_in_flight_append_before_sealing_the_head() { + let locator = memory(); + let session_ref = SessionRef::scoped("reset-race", "agent-id"); + seed( + locator.as_ref(), + &session_stem(&session_ref), + &[ + TranscriptMessage::user("before concurrent turn"), + TranscriptMessage::assistant("old answer"), + ], + &meta(), + ); + let started = Arc::new(Notify::new()); + let release = Arc::new(Notify::new()); + let mut turning = SessionBuilder::new(Arc::new(GatedDriver { + started: started.clone(), + release: release.clone(), + })) + .codec(Arc::new(RoleCodec)) + .session(locator.clone(), session_ref.clone(), meta()) + .build() + .unwrap(); + let turn_ref = session_ref.clone(); + let turn = tokio::spawn(async move { + turning + .turn( + SessionTurnRequest::new(Message::user("commits before reset")), + TurnOptions { + resume: ResumeMode::Session, + session: Some(turn_ref), + ..TurnOptions::default() + }, + ) + .await + }); + started.notified().await; + + let reset_ref = SessionRef::scoped("reset-race", "agent-id"); + let mut resetting = + SessionBuilder::new(Arc::new(CapturingDriver(Arc::new(Mutex::new(Vec::new()))))) + .codec(Arc::new(RoleCodec)) + .session(locator.clone(), reset_ref, meta()) + .build() + .unwrap(); + let reset = tokio::spawn(async move { resetting.reset_history().await }); + tokio::task::yield_now().await; + assert!( + !reset.is_finished(), + "reset bypassed the active session turn lock" + ); + + release.notify_one(); + turn.await.unwrap().unwrap(); + assert!(reset.await.unwrap().unwrap()); + let old = locator + .read_session_transcript(&session_ref) + .unwrap() + .read_session() + .unwrap() + .unwrap(); + assert!( + old.messages + .iter() + .any(|message| message.content == "commits before reset") + ); + assert_eq!(locator.head_generation(&session_ref).generation, 1); +} + +#[tokio::test] +async fn reset_does_not_create_an_absent_session_or_cross_agent_boundary() { + let locator = memory(); + let absent_ref = SessionRef::scoped("absent-reset", "agent-a"); + let mut absent = + SessionBuilder::new(Arc::new(CapturingDriver(Arc::new(Mutex::new(Vec::new()))))) + .codec(Arc::new(RoleCodec)) + .session(locator.clone(), absent_ref.clone(), meta()) + .build() + .unwrap(); + assert!(!absent.reset_history().await.unwrap()); + assert!(locator.read_session_transcript(&absent_ref).is_none()); + + let first_ref = SessionRef::scoped("same-thread", "agent-a"); + let other_agent_ref = SessionRef::scoped("same-thread", "agent-b"); + seed( + locator.as_ref(), + &session_stem(&first_ref), + &[ + TranscriptMessage::user("first agent"), + TranscriptMessage::assistant("answer"), + ], + &meta(), + ); + seed( + locator.as_ref(), + &session_stem(&other_agent_ref), + &[ + TranscriptMessage::user("other agent"), + TranscriptMessage::assistant("answer"), + ], + &meta(), + ); + let mut first = + SessionBuilder::new(Arc::new(CapturingDriver(Arc::new(Mutex::new(Vec::new()))))) + .codec(Arc::new(RoleCodec)) + .session(locator.clone(), first_ref.clone(), meta()) + .build() + .unwrap(); + assert!(first.reset_history().await.unwrap()); + assert_eq!(locator.head_generation(&first_ref).generation, 1); + assert_eq!(locator.head_generation(&other_agent_ref).generation, 0); + assert!( + locator + .read_session_transcript(&other_agent_ref) + .unwrap() + .read_session() + .unwrap() + .unwrap() + .messages + .iter() + .any(|message| message.content == "other agent") + ); +} + +/// An atomic import may create a durable session before its first model turn. +/// The resumed history must still receive the host's initial prefix, while a +/// later process must reuse that committed prefix rather than its new prompt. +#[tokio::test] +async fn imported_history_gets_initial_prefix_before_the_first_turn() { + let locator = memory(); + let session_ref = SessionRef::scoped("relay-import", "agent"); + let prior = vec![ + TranscriptMessage::user("legacy question"), + TranscriptMessage::assistant("legacy answer"), + ]; + assert!( + locator + .adopt_messages_if_absent(&session_ref, meta(), &prior) + .unwrap() + ); + + let first_requests = Arc::new(Mutex::new(Vec::new())); + let mut first = SessionBuilder::new(Arc::new(CapturingDriver(first_requests.clone()))) + .codec(Arc::new(RoleCodec)) + .hooks(Arc::new(PrefixWhenUnfrozen("configured initial prompt"))) + .session(locator.clone(), session_ref.clone(), meta()) + .build() + .unwrap(); + first + .turn( + SessionTurnRequest::new(Message::user("current question")), + TurnOptions { + resume: ResumeMode::Session, + session: Some(session_ref.clone()), + thread_id: Some("relay-import".into()), + ..TurnOptions::default() + }, + ) + .await + .unwrap(); + + let first_history = first_requests.lock().unwrap()[0].clone(); + assert!(matches!(&first_history[0], Message::System(_))); + assert_eq!(first_history[0].text(), "configured initial prompt"); + assert!(matches!(&first_history[1], Message::User(_))); + assert_eq!(first_history[1].text(), "legacy question"); + assert!(matches!(&first_history[2], Message::Assistant(_))); + assert_eq!(first_history[2].text(), "legacy answer"); + assert!(matches!(&first_history[3], Message::User(_))); + assert_eq!(first_history[3].text(), "current question"); + let head = locator.head_generation(&session_ref); + let persisted = locator + .read_session_transcript(&head) + .unwrap() + .read_session() + .unwrap() + .unwrap(); + assert_eq!(persisted.messages[0].role, "system"); + assert_eq!(persisted.messages[0].content, "configured initial prompt"); + + let second_requests = Arc::new(Mutex::new(Vec::new())); + let mut second = SessionBuilder::new(Arc::new(CapturingDriver(second_requests.clone()))) + .codec(Arc::new(RoleCodec)) + .hooks(Arc::new(PrefixWhenUnfrozen("replacement prompt"))) + .session(locator, session_ref.clone(), meta()) + .build() + .unwrap(); + second + .turn( + SessionTurnRequest::new(Message::user("after restart")), + TurnOptions { + resume: ResumeMode::Session, + session: Some(session_ref), + thread_id: Some("relay-import".into()), + ..TurnOptions::default() + }, + ) + .await + .unwrap(); + let second_history = second_requests.lock().unwrap()[0].clone(); + assert!(matches!(&second_history[0], Message::System(_))); + assert_eq!(second_history[0].text(), "configured initial prompt"); + assert!(matches!(&second_history[1], Message::User(_))); + assert_eq!(second_history[1].text(), "legacy question"); + assert!(matches!(&second_history[2], Message::Assistant(_))); + assert_eq!(second_history[2].text(), "legacy answer"); + assert!( + matches!(&second_history[3], Message::User(_)), + "resumed request history: {:?}", + second_history.iter().map(Message::text).collect::>() + ); + assert_eq!(second_history[3].text(), "current question"); + assert!(matches!(second_history.last(), Some(Message::User(_)))); + assert_eq!(second_history.last().unwrap().text(), "after restart"); +} + +#[tokio::test] +async fn an_explicit_empty_frozen_prefix_stays_empty() { + let locator = memory(); + let session_ref = SessionRef::scoped("empty-prefix", "agent"); + let mut explicit_empty_prefix = meta(); + explicit_empty_prefix.prefix_message_count = Some(0); + seed( + locator.as_ref(), + &session_stem(&session_ref), + &[TranscriptMessage::user("existing message")], + &explicit_empty_prefix, + ); + + let requests = Arc::new(Mutex::new(Vec::new())); + let mut session = SessionBuilder::new(Arc::new(CapturingDriver(requests.clone()))) + .codec(Arc::new(RoleCodec)) + .hooks(Arc::new(PrefixWhenUnfrozen("must remain unused"))) + .session(locator, session_ref.clone(), explicit_empty_prefix) + .build() + .unwrap(); + session + .turn( + SessionTurnRequest::new(Message::user("next message")), + TurnOptions { + resume: ResumeMode::Session, + session: Some(session_ref), + thread_id: Some("empty-prefix".into()), + ..TurnOptions::default() + }, + ) + .await + .unwrap(); + assert!( + requests.lock().unwrap()[0] + .iter() + .all(|message| !matches!(message, Message::System(_))) + ); +} + /// Answers each turn with `reply`, but only after the test releases it. struct GatedDriver { started: Arc, diff --git a/crates/tinyagents-runtime/src/test_store.rs b/crates/tinyagents-runtime/src/test_store.rs index 3578ad746..e994e9a60 100644 --- a/crates/tinyagents-runtime/src/test_store.rs +++ b/crates/tinyagents-runtime/src/test_store.rs @@ -147,6 +147,7 @@ fn placeholder_meta() -> TranscriptMeta { updated: String::new(), turn_count: 0, prefix_message_count: None, + prefix_reset: false, input_tokens: 0, output_tokens: 0, cached_input_tokens: 0, diff --git a/crates/tinyagents-runtime/src/types.rs b/crates/tinyagents-runtime/src/types.rs index 62421184e..39cd4fe53 100644 --- a/crates/tinyagents-runtime/src/types.rs +++ b/crates/tinyagents-runtime/src/types.rs @@ -236,6 +236,11 @@ pub struct SessionStateView<'a> { pub history: &'a [Message], pub raw_history: &'a [TranscriptMessage], pub prefix: &'a PrefixSnapshot, + /// Whether the loaded transcript already has an authoritative prefix. + /// A durable transcript can contain imported conversation rows before its + /// first model turn; such a session is resumed for history purposes but + /// still needs the host to prepare and commit its initial prompt. + pub prefix_is_frozen: bool, pub transcript_target: Option<&'a TranscriptTarget>, pub committed_turns: usize, /// `true` only when this call loaded and decoded a durable transcript diff --git a/crates/tinyagents-session/Cargo.toml b/crates/tinyagents-session/Cargo.toml index 8e16dd90a..91b2ce925 100644 --- a/crates/tinyagents-session/Cargo.toml +++ b/crates/tinyagents-session/Cargo.toml @@ -15,6 +15,7 @@ anyhow = { workspace = true } thiserror = "2" serde = { workspace = true } serde_json = { workspace = true } +schemars = { version = "1", features = ["derive", "chrono04"] } tinyagents-harness = { path = "../tinyagents-harness", version = "3.0.0", default-features = false, features = ["sqlite"] } # `OrchestrationTaskStatus` is the canonical lifecycle status the run-ledger and # transcript-view statuses convert to and from. diff --git a/crates/tinyagents-session/examples/turn_state_schema.rs b/crates/tinyagents-session/examples/turn_state_schema.rs new file mode 100644 index 000000000..87ca20fdc --- /dev/null +++ b/crates/tinyagents-session/examples/turn_state_schema.rs @@ -0,0 +1,26 @@ +//! Print the authoritative Rust JSON Schema for persisted session UI records. + +use schemars::JsonSchema; +use tinyagents_session::{ + run_ledger::types::{AgentRun, RunEvent, RunTelemetry}, + turn_state::types::TurnState, +}; + +#[derive(JsonSchema)] +#[schemars(rename_all = "camelCase")] +#[allow(dead_code)] +struct SessionUiWireTypes { + turn_state: TurnState, + agent_run: AgentRun, + run_event: RunEvent, + run_telemetry: RunTelemetry, +} + +fn main() -> Result<(), Box> { + let schema = schemars::generate::SchemaSettings::draft2020_12() + .for_serialize() + .into_generator() + .into_root_schema_for::(); + println!("{}", serde_json::to_string_pretty(&schema)?); + Ok(()) +} diff --git a/crates/tinyagents-session/src/port/drivers/transcripts.rs b/crates/tinyagents-session/src/port/drivers/transcripts.rs index 4ecca743e..a30ab3500 100644 --- a/crates/tinyagents-session/src/port/drivers/transcripts.rs +++ b/crates/tinyagents-session/src/port/drivers/transcripts.rs @@ -746,6 +746,25 @@ impl TranscriptHistory for DriverTranscriptHistory { .map(|_| ()) } + fn adopt_if_empty(&self, messages: &[TranscriptMessage]) -> anyhow::Result { + let messages = normalized(messages.to_vec()); + if messages.is_empty() { + return Ok(false); + } + self.commit(move |replay, seed| { + if replay.written { + return Ok(None); + } + anyhow::ensure!(!replay.sealed, "transcript generation is sealed"); + Ok(Some(Entry { + written: true, + meta: Some(seed.clone()), + set: Some(messages.clone()), + ..Entry::default() + })) + }) + } + fn replace(&self, messages: &[TranscriptMessage]) -> anyhow::Result<()> { let messages = normalized(messages.to_vec()); self.commit(move |replay, seed| { @@ -982,6 +1001,7 @@ fn discovered_seed(stem: &str) -> TranscriptMeta { updated: String::new(), turn_count: 0, prefix_message_count: None, + prefix_reset: false, input_tokens: 0, output_tokens: 0, cached_input_tokens: 0, diff --git a/crates/tinyagents-session/src/port/drivers/transcripts_adoption_tests.rs b/crates/tinyagents-session/src/port/drivers/transcripts_adoption_tests.rs index 7d2ea6211..4591ecf02 100644 --- a/crates/tinyagents-session/src/port/drivers/transcripts_adoption_tests.rs +++ b/crates/tinyagents-session/src/port/drivers/transcripts_adoption_tests.rs @@ -3,10 +3,11 @@ use std::sync::Arc; use super::super::{leases_spec, reservation::RESERVATION_TTL_MS}; use crate::port::drivers::transcripts::tests::meta; +use crate::testkit::MemoryStorage; use crate::transcript::TranscriptMessage; use crate::transcript::test_locator::conditioned_driver_locator; -use tinyagents_storage_kit::Blocking; use tinyagents_storage_kit::primitives::Leases; +use tinyagents_storage_kit::{Blocking, DocumentStore, Scope}; const THREAD: &str = "t"; @@ -58,6 +59,117 @@ fn contents(locator: &dyn TranscriptLocator, session: &SessionRef) -> Vec = Arc::clone( + storage + .for_scope(&Scope::new("local").unwrap()) + .unwrap() + .documents(), + ); + let locators = [ + Arc::new(DriverTranscriptLocator::new( + Arc::clone(&docs), + Blocking::new().unwrap(), + "memory://adoption-a", + )), + Arc::new(DriverTranscriptLocator::new( + Arc::clone(&docs), + Blocking::new().unwrap(), + "memory://adoption-b", + )), + ]; + let session = SessionRef::scoped("relay-race", "orchestrator"); + let barrier = Arc::new(std::sync::Barrier::new(2)); + let histories = [ + vec![TranscriptMessage::user("first import")], + vec![ + TranscriptMessage::user("second import"), + TranscriptMessage::assistant("second answer"), + ], + ]; + let results = std::thread::scope(|scope| { + let handles = histories + .into_iter() + .zip(locators) + .map(|(messages, locator)| { + let barrier = Arc::clone(&barrier); + let session = session.clone(); + scope.spawn(move || { + barrier.wait(); + locator + .adopt_messages_if_absent(&session, meta("relay-race"), &messages) + .unwrap() + }) + }) + .collect::>(); + handles + .into_iter() + .map(|handle| handle.join().unwrap()) + .collect::>() + }); + assert_eq!(results.iter().filter(|adopted| **adopted).count(), 1); + + let locator = + DriverTranscriptLocator::new(docs, Blocking::new().unwrap(), "memory://adoption-read"); + let persisted = contents(&locator, &session); + assert!( + persisted == vec!["first import"] || persisted == vec!["second import", "second answer"], + "a competing import must not leave a partial or mixed transcript: {persisted:?}" + ); +} + #[test] fn legacy_roots_are_folded_oldest_first_once() { let locator = locator(); diff --git a/crates/tinyagents-session/src/port/memory/mod_tests.rs b/crates/tinyagents-session/src/port/memory/mod_tests.rs index a8733ea08..8b60fac65 100644 --- a/crates/tinyagents-session/src/port/memory/mod_tests.rs +++ b/crates/tinyagents-session/src/port/memory/mod_tests.rs @@ -16,6 +16,7 @@ fn test_meta(thread_id: &str) -> TranscriptMeta { updated: "2026-01-01T00:00:00Z".to_string(), turn_count: 0, prefix_message_count: None, + prefix_reset: false, input_tokens: 0, output_tokens: 0, cached_input_tokens: 0, @@ -169,6 +170,7 @@ fn the_newest_root_wins_and_sub_agents_are_never_roots() { updated: "2026-01-01T00:00:00Z".to_string(), turn_count: 0, prefix_message_count: None, + prefix_reset: false, input_tokens: 0, output_tokens: 0, cached_input_tokens: 0, diff --git a/crates/tinyagents-session/src/run_ledger/types.rs b/crates/tinyagents-session/src/run_ledger/types.rs index 97e030c51..a939b076b 100644 --- a/crates/tinyagents-session/src/run_ledger/types.rs +++ b/crates/tinyagents-session/src/run_ledger/types.rs @@ -20,7 +20,7 @@ use serde_json::Value; /// /// Distinguishes the several call sites that create a run row so listings and /// UI can filter by origin without inferring it from other fields. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)] #[serde(rename_all = "snake_case")] pub enum AgentRunKind { /// A synchronous sub-agent invocation nested inside a parent run. @@ -68,7 +68,7 @@ impl AgentRunKind { /// /// Convertible to and from the canonical `tinyagents_tasks::OrchestrationTaskStatus` /// (see `status_map.rs`); the serialized form is unchanged. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)] #[serde(rename_all = "snake_case")] pub enum AgentRunStatus { /// Created but not yet started. @@ -186,7 +186,7 @@ impl WorkflowRunStatus { /// A persisted background agent run: a subagent, worker thread, background /// agent, team member, or workflow child, as classified by `kind`. -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, schemars::JsonSchema)] #[serde(rename_all = "camelCase")] pub struct AgentRun { /// Caller-supplied stable identifier. @@ -278,7 +278,7 @@ pub enum WorkflowLeaseClaim { /// /// `sequence` is allocated by the insert itself (see `super::ops`) rather /// than computed by the caller, so concurrent appenders cannot collide. -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, schemars::JsonSchema)] #[serde(rename_all = "camelCase")] pub struct RunEvent { pub run_id: String, @@ -296,7 +296,7 @@ pub struct RunEvent { /// /// One row per `run_id`, updated in place as a run progresses (see /// `super::ops::upsert_run_telemetry` for how partial updates are coalesced). -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Default)] +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Default, schemars::JsonSchema)] #[serde(rename_all = "camelCase")] pub struct RunTelemetry { pub run_id: String, @@ -406,7 +406,7 @@ pub struct AgentRunListRequest { } /// Response shape for `super::ops::list_agent_runs`. -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone, Serialize, Deserialize, schemars::JsonSchema)] #[serde(rename_all = "camelCase")] pub struct AgentRunListResponse { pub runs: Vec, @@ -455,7 +455,7 @@ pub struct RunEventListRequest { } /// Response shape for `super::ops::list_recent_run_events`. -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone, Serialize, Deserialize, schemars::JsonSchema)] #[serde(rename_all = "camelCase")] pub struct RunEventListResponse { pub events: Vec, @@ -795,3 +795,7 @@ pub enum RunTransition { /// No run has this id. NotFound, } + +#[cfg(test)] +#[path = "types_tests.rs"] +mod tests; diff --git a/crates/tinyagents-session/src/run_ledger/types_tests.rs b/crates/tinyagents-session/src/run_ledger/types_tests.rs new file mode 100644 index 000000000..ddfc928ed --- /dev/null +++ b/crates/tinyagents-session/src/run_ledger/types_tests.rs @@ -0,0 +1,36 @@ +use super::*; + +#[test] +fn run_ledger_schemas_use_the_persisted_camel_case_fields_and_enum_names() { + let run_schema = serde_json::to_value(schemars::schema_for!(AgentRun)).unwrap(); + let run_properties = run_schema["properties"].as_object().unwrap(); + for field in [ + "parentThreadId", + "workerThreadId", + "startedAt", + "completedAt", + ] { + assert!( + run_properties.contains_key(field), + "AgentRun schema is missing {field}" + ); + } + + let event_schema = serde_json::to_value(schemars::schema_for!(RunEvent)).unwrap(); + let event_properties = event_schema["properties"].as_object().unwrap(); + for field in ["runId", "eventType", "timestamp"] { + assert!( + event_properties.contains_key(field), + "RunEvent schema is missing {field}" + ); + } + + assert_eq!( + serde_json::to_value(AgentRunKind::WorkerThread).unwrap(), + "worker_thread" + ); + assert_eq!( + serde_json::to_value(AgentRunStatus::AwaitingUser).unwrap(), + "awaiting_user" + ); +} diff --git a/crates/tinyagents-session/src/testkit/conformance.rs b/crates/tinyagents-session/src/testkit/conformance.rs index cd096b856..3b0b534c8 100644 --- a/crates/tinyagents-session/src/testkit/conformance.rs +++ b/crates/tinyagents-session/src/testkit/conformance.rs @@ -330,6 +330,7 @@ fn contract_meta() -> TranscriptMeta { updated: "2026-01-01T00:00:00Z".to_string(), turn_count: 0, prefix_message_count: None, + prefix_reset: false, input_tokens: 0, output_tokens: 0, cached_input_tokens: 0, diff --git a/crates/tinyagents-session/src/testkit/in_memory_transcript.rs b/crates/tinyagents-session/src/testkit/in_memory_transcript.rs index a6548a040..16318431b 100644 --- a/crates/tinyagents-session/src/testkit/in_memory_transcript.rs +++ b/crates/tinyagents-session/src/testkit/in_memory_transcript.rs @@ -271,6 +271,24 @@ impl TranscriptHistory for InMemoryTranscriptHistory { Ok(()) } + fn adopt_if_empty(&self, messages: &[TranscriptMessage]) -> anyhow::Result { + if messages.is_empty() { + return Ok(false); + } + let _gate = self + .generation_gate + .lock() + .unwrap_or_else(|e| e.into_inner()); + let mut state = self.state.lock().unwrap_or_else(|e| e.into_inner()); + anyhow::ensure!(!state.sealed, "transcript generation is sealed"); + if state.written { + return Ok(false); + } + state.messages = normalized_rows(messages); + state.written = true; + Ok(true) + } + fn replace(&self, messages: &[TranscriptMessage]) -> anyhow::Result<()> { let _gate = self .generation_gate diff --git a/crates/tinyagents-session/src/testkit/in_memory_transcript_tests.rs b/crates/tinyagents-session/src/testkit/in_memory_transcript_tests.rs index 8ef7b37d5..0e6309481 100644 --- a/crates/tinyagents-session/src/testkit/in_memory_transcript_tests.rs +++ b/crates/tinyagents-session/src/testkit/in_memory_transcript_tests.rs @@ -22,6 +22,7 @@ fn in_memory_transcript_starts_empty() { task_id: None, session_id: None, parent_session_id: None, + prefix_reset: false, }; let transcript = locator.open_stem("fixture", meta).expect("bind transcript"); diff --git a/crates/tinyagents-session/src/threads/driver/append.rs b/crates/tinyagents-session/src/threads/driver/append.rs index 5841e413e..721191af9 100644 --- a/crates/tinyagents-session/src/threads/driver/append.rs +++ b/crates/tinyagents-session/src/threads/driver/append.rs @@ -101,7 +101,7 @@ impl DriverConversationStore { thread_id: &str, message: &ThreadMessage, ) -> ConversationResult> { - if !is_deterministic_message_id(&message.id) + if !is_idempotent_message_id(&message.id) && message .extra_metadata .get("scope") @@ -174,6 +174,11 @@ impl DriverConversationStore { } let stream = thread.stream()?; if thread.visible_end != offset { + // A concurrent retry with the same idempotency key may have + // committed this exact intent after the check above. + if self.committed(intent).await? { + return intent.message(); + } return Err(StorageError::serialization( "conversation append prefix is inconsistent", ) diff --git a/crates/tinyagents-session/src/threads/driver/rewrite.rs b/crates/tinyagents-session/src/threads/driver/rewrite.rs index 4a77748d0..d9ae5e6b7 100644 --- a/crates/tinyagents-session/src/threads/driver/rewrite.rs +++ b/crates/tinyagents-session/src/threads/driver/rewrite.rs @@ -136,6 +136,9 @@ impl DriverConversationStore { if let Some(extra) = patch.extra_metadata.clone() { message.extra_metadata = extra; } + if let Some(merge) = patch.merge_extra_metadata.clone() { + Self::merge_json_patch(&mut message.extra_metadata, merge); + } let updated = message.clone(); Ok(Some((messages, updated))) }) @@ -143,6 +146,30 @@ impl DriverConversationStore { updated.ok_or_else(|| StorageError::not_found("conversation message not found").into()) } + /// Apply an RFC 7396-style JSON merge patch to a message metadata value. + /// Called inside `rewrite`'s CAS retry so each retry uses the current row. + fn merge_json_patch(target: &mut serde_json::Value, patch: serde_json::Value) { + match patch { + serde_json::Value::Object(patch) => { + if !target.is_object() { + *target = serde_json::json!({}); + } + let target = target.as_object_mut().expect("object established"); + for (key, value) in patch { + if value.is_null() { + target.remove(&key); + } else { + Self::merge_json_patch( + target.entry(key).or_insert(serde_json::Value::Null), + value, + ); + } + } + } + replacement => *target = replacement, + } + } + pub(super) async fn delete_messages_from_native( &self, thread_id: &str, diff --git a/crates/tinyagents-session/src/threads/driver_tests.rs b/crates/tinyagents-session/src/threads/driver_tests.rs index 4fafdc251..c6b417ee7 100644 --- a/crates/tinyagents-session/src/threads/driver_tests.rs +++ b/crates/tinyagents-session/src/threads/driver_tests.rs @@ -299,6 +299,82 @@ async fn conflicting_metadata_edits_reread_version_and_preserve_both_fields() { assert_eq!(row.labels, vec!["tasks"]); } +#[tokio::test(flavor = "current_thread")] +async fn concurrent_message_metadata_merges_retry_against_latest_row() { + use std::sync::atomic::{AtomicUsize, Ordering}; + let backend = MemoryStorage::new(); + let base = scoped(&backend, "profile"); + seed(&base, "one", &["hello"]).await; + base.update_message( + "one", + "message:0", + ThreadMessagePatch { + extra_metadata: Some(json!({"voice": {"transcript": "spoken"}})), + ..Default::default() + }, + ) + .await + .unwrap(); + + let scope = Scope::new("profile").unwrap(); + let ports = backend.for_scope(&scope).unwrap(); + let docs = Arc::new(fixtures::ConflictingDocs { + inner: ports.documents().clone(), + attempts: AtomicUsize::new(0), + barrier: tokio::sync::Barrier::new(2), + refuse_get: false, + }); + let storage = tinystoragedrivers_api::Storage::new( + docs.clone(), + ports.streams().clone(), + ports.kv().clone(), + ports.blobs().clone(), + ports.queues().clone(), + ports.secrets().clone(), + ); + let store = DriverConversationStore::new(ScopedStorage::new(storage, scope)); + let (reaction, receipt) = tokio::join!( + store.update_message( + "one", + "message:0", + ThreadMessagePatch { + merge_extra_metadata: Some(json!({ + "channelActivity": {"reactions": {"reaction-key": {"value": "👍"}}} + })), + ..Default::default() + }, + ), + store.update_message( + "one", + "message:0", + ThreadMessagePatch { + merge_extra_metadata: Some(json!({ + "channelActivity": {"readReceipts": {"receipt-key": {"actorId": "reader"}}} + })), + ..Default::default() + }, + ), + ); + reaction.unwrap(); + receipt.unwrap(); + assert!( + docs.attempts.load(Ordering::SeqCst) >= 3, + "fixture must force one metadata rewrite to retry after a CAS conflict" + ); + + let messages = store.get_messages("one").await.unwrap(); + let message = &messages[0]; + assert_eq!(message.extra_metadata["voice"]["transcript"], "spoken"); + assert_eq!( + message.extra_metadata["channelActivity"]["reactions"]["reaction-key"]["value"], + "👍" + ); + assert_eq!( + message.extra_metadata["channelActivity"]["readReceipts"]["receipt-key"]["actorId"], + "reader" + ); +} + fn instrumented_streams( store: &DriverConversationStore, pause_first: bool, diff --git a/crates/tinyagents-session/src/threads/mod.rs b/crates/tinyagents-session/src/threads/mod.rs index a31520fc6..2256c8952 100644 --- a/crates/tinyagents-session/src/threads/mod.rs +++ b/crates/tinyagents-session/src/threads/mod.rs @@ -40,8 +40,8 @@ pub use schema::conversations_storage_schema; pub use types::{ ConversationThread, CreateConversationThread, CrossThreadHit, DETERMINISTIC_MESSAGE_ID_PREFIX, - ThreadMessage, ThreadMessagePatch, is_deterministic_message_id, reply_run_id, - run_reply_message_id, + ThreadMessage, ThreadMessagePatch, is_deterministic_message_id, is_idempotent_message_id, + reply_run_id, run_reply_message_id, }; #[cfg(test)] diff --git a/crates/tinyagents-session/src/threads/persistence_tests.rs b/crates/tinyagents-session/src/threads/persistence_tests.rs index f39823e0d..380083f7f 100644 --- a/crates/tinyagents-session/src/threads/persistence_tests.rs +++ b/crates/tinyagents-session/src/threads/persistence_tests.rs @@ -70,6 +70,46 @@ async fn memory_store_preserves_deterministic_append_and_missing_thread_behavior ); } +#[tokio::test(flavor = "multi_thread", worker_threads = 8)] +async fn concurrent_user_request_id_appends_are_idempotent() { + let store = memory_store().await; + for round in 0..32 { + let thread_id = format!("thread-{round}"); + store.ensure_thread(thread(&thread_id)).await.unwrap(); + let barrier = std::sync::Arc::new(tokio::sync::Barrier::new(8)); + let mut writers = Vec::new(); + for _ in 0..8 { + let store = store.clone(); + let barrier = barrier.clone(); + let thread_id = thread_id.clone(); + writers.push(tokio::spawn(async move { + barrier.wait().await; + store + .append_message( + &thread_id, + ThreadMessage { + sender: "user".into(), + ..message("user:request-1", "queued prompt") + }, + ) + .await + })); + } + let mut result = None; + for writer in writers { + let row = writer.await.unwrap().unwrap(); + assert_eq!(row.id, "user:request-1"); + assert_eq!(row.content, "queued prompt"); + assert_eq!(result.get_or_insert(row.clone()), &row); + } + let messages = store.get_messages(&thread_id).await.unwrap(); + assert_eq!(messages.len(), 1); + assert_eq!(messages[0].id, "user:request-1"); + assert_eq!(messages[0].content, "queued prompt"); + assert_eq!(messages[0].sender, "user"); + } +} + #[tokio::test] async fn memory_store_preserves_title_labels_and_metadata_patch() { let store = memory_store().await; @@ -94,6 +134,7 @@ async fn memory_store_preserves_title_labels_and_metadata_patch() { "agent:x", ThreadMessagePatch { extra_metadata: Some(json!({"success": true})), + ..Default::default() }, ) .await @@ -129,6 +170,92 @@ async fn memory_store_preserves_title_labels_and_metadata_patch() { ); } +#[tokio::test] +async fn concurrent_metadata_merge_patches_preserve_each_other_and_existing_metadata() { + let store = memory_store().await; + store.ensure_thread(thread("one")).await.unwrap(); + store + .append_message( + "one", + ThreadMessage { + extra_metadata: json!({"voice": {"transcript": "spoken"}}), + ..message("user:receipt", "hello") + }, + ) + .await + .unwrap(); + + let (reaction, receipt) = tokio::join!( + store.update_message( + "one", + "user:receipt", + ThreadMessagePatch { + merge_extra_metadata: Some(json!({ + "channelActivity": {"reactions": {"actor-reaction": {"value": "👍"}}} + })), + ..Default::default() + }, + ), + store.update_message( + "one", + "user:receipt", + ThreadMessagePatch { + merge_extra_metadata: Some(json!({ + "channelActivity": {"readReceipts": {"reader": {"timestamp": "2026-10-20T00:00:00Z"}}} + })), + ..Default::default() + }, + ), + ); + reaction.unwrap(); + receipt.unwrap(); + + let stored = store.get_messages("one").await.unwrap(); + let metadata = &stored[0].extra_metadata; + assert_eq!(metadata["voice"]["transcript"], "spoken"); + assert_eq!( + metadata["channelActivity"]["reactions"]["actor-reaction"]["value"], + "👍" + ); + assert_eq!( + metadata["channelActivity"]["readReceipts"]["reader"]["timestamp"], + "2026-10-20T00:00:00Z" + ); +} + +#[tokio::test] +async fn metadata_merge_patch_null_removes_only_the_named_member() { + let store = memory_store().await; + store.ensure_thread(thread("one")).await.unwrap(); + store + .append_message( + "one", + ThreadMessage { + extra_metadata: json!({"channelActivity": {"reactions": {"gone": {}, "kept": {}}}}), + ..message("user:receipt", "hello") + }, + ) + .await + .unwrap(); + store + .update_message( + "one", + "user:receipt", + ThreadMessagePatch { + merge_extra_metadata: Some(json!({ + "channelActivity": {"reactions": {"gone": null}} + })), + ..Default::default() + }, + ) + .await + .unwrap(); + let messages = store.get_messages("one").await.unwrap(); + let reactions = &messages[0].extra_metadata["channelActivity"]["reactions"]; + assert!(reactions.get("gone").is_none()); + assert!(reactions.get("kept").is_some()); +} + #[tokio::test] async fn memory_store_tail_delete_recreate_and_purge() { let store = memory_store().await; @@ -285,6 +412,7 @@ async fn update_message_rewrites_only_the_patched_message_and_keeps_appends_work "b", ThreadMessagePatch { extra_metadata: Some(json!({"k": 1})), + ..Default::default() }, ) .await diff --git a/crates/tinyagents-session/src/threads/schema_tests.rs b/crates/tinyagents-session/src/threads/schema_tests.rs index 891e43b44..a8b1233ad 100644 --- a/crates/tinyagents-session/src/threads/schema_tests.rs +++ b/crates/tinyagents-session/src/threads/schema_tests.rs @@ -334,6 +334,7 @@ async fn a_migrated_conversation_reads_recovers_its_intent_and_moves_to_canonica // A rewrite opens a fresh incarnation whose stream carries the thread id. let patch = ThreadMessagePatch { extra_metadata: Some(json!({"edited": true})), + merge_extra_metadata: None, }; store.update_message(id, "m1", patch).await.unwrap(); let streams = storage diff --git a/crates/tinyagents-session/src/threads/types.rs b/crates/tinyagents-session/src/threads/types.rs index 1037a521b..0da7f8088 100644 --- a/crates/tinyagents-session/src/threads/types.rs +++ b/crates/tinyagents-session/src/threads/types.rs @@ -102,6 +102,12 @@ pub struct ThreadMessagePatch { /// untouched. #[serde(default)] pub extra_metadata: Option, + /// Recursively merges a JSON Merge Patch into the latest `extraMetadata`. + /// Object members are merged, arrays/scalars are replaced, and `null` + /// removes an object member. Applied inside the driver's compare-and-swap + /// retry so concurrent patches are not based on a stale host-side read. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub merge_extra_metadata: Option, } /// A single match returned by @@ -124,14 +130,23 @@ pub struct CrossThreadHit { pub score: f64, } -/// Prefix of the message ids the core mints **deterministically** rather than -/// from a fresh UUID. +/// Prefix of the assistant reply ids the core mints from a request id. /// -/// Only such an id can be presented to the store twice by two different -/// writers, so it is also the marker -/// [`is_deterministic_message_id`] keys the store's idempotency lookup on. +/// [`is_deterministic_message_id`] identifies this assistant-only format for +/// transcript correlation. The append retry contract is broader and is +/// defined by [`is_idempotent_message_id`]. pub const DETERMINISTIC_MESSAGE_ID_PREFIX: &str = "agent:"; +/// Whether a message id marks an append that may be retried safely. +/// +/// Agent replies use `agent:` and user rows submitted through a +/// queued/hosted path use `user:`. Both identifiers are stable +/// across retries; ordinary UI ids and provider-specific sender ids remain +/// outside this contract unless their message carries `scope: channel`. +pub fn is_idempotent_message_id(id: &str) -> bool { + id.starts_with(DETERMINISTIC_MESSAGE_ID_PREFIX) || id.starts_with("user:") +} + /// The id an autonomous run's closing reply is stored under. /// /// Two writers legitimately persist that one reply — the core's @@ -143,15 +158,11 @@ pub fn run_reply_message_id(run_id: &str) -> String { format!("{DETERMINISTIC_MESSAGE_ID_PREFIX}{run_id}") } -/// Whether `id` is one the core mints deterministically, i.e. one a second -/// writer can legitimately present again. +/// Whether `id` is an assistant reply id minted from a run/request id. /// -/// This is what buys back the constant-time append path: every other id in the -/// store is UUID-fresh by construction (`user:`, `:`), can -/// never be re-presented, and so must not pay for a duplicate lookup. The -/// `agent:` ids the subagent/worker-thread writers mint do match — they -/// pay for a lookup they can never hit, which is one cheap scan of a -/// two-message worker transcript. +/// This predicate is used for transcript correlation by +/// [`reply_run_id`]. Use [`is_idempotent_message_id`] to check the broader +/// append retry contract. pub fn is_deterministic_message_id(id: &str) -> bool { id.starts_with(DETERMINISTIC_MESSAGE_ID_PREFIX) } diff --git a/crates/tinyagents-session/src/threads/types_tests.rs b/crates/tinyagents-session/src/threads/types_tests.rs index 23e13b6e0..7d29b1c70 100644 --- a/crates/tinyagents-session/src/threads/types_tests.rs +++ b/crates/tinyagents-session/src/threads/types_tests.rs @@ -32,6 +32,7 @@ fn conversation_thread_serde_uses_camel_case_and_defaults_labels() { fn conversation_message_patch_defaults_to_no_changes() { let patch: ThreadMessagePatch = serde_json::from_value(json!({})).unwrap(); assert!(patch.extra_metadata.is_none()); + assert!(patch.merge_extra_metadata.is_none()); let patch_with_metadata: ThreadMessagePatch = serde_json::from_value(json!({"extraMetadata": {"source": "mock"}})).unwrap(); @@ -39,6 +40,14 @@ fn conversation_message_patch_defaults_to_no_changes() { patch_with_metadata.extra_metadata, Some(json!({"source": "mock"})) ); + assert!(patch_with_metadata.merge_extra_metadata.is_none()); + + let merge_patch: ThreadMessagePatch = + serde_json::from_value(json!({"mergeExtraMetadata": {"channelActivity": {}}})).unwrap(); + assert_eq!( + merge_patch.merge_extra_metadata, + Some(json!({"channelActivity": {}})) + ); } #[test] @@ -76,8 +85,8 @@ fn run_reply_id_is_deterministic_and_recognised_as_such() { #[test] fn client_generated_ids_are_not_deterministic() { - // These are UUID-fresh per message, so they can never be re-presented and - // must keep the constant-time append path. + // User ids are not transcript-correlated assistant replies. The user: + // prefix is a separate append-idempotency contract. assert!(!is_deterministic_message_id( "user:5f1d0c3e-1f8b-4c1a-9c2e-2a7b6d4e8f90" )); @@ -85,3 +94,10 @@ fn client_generated_ids_are_not_deterministic() { "assistant:5f1d0c3e-1f8b-4c1a-9c2e-2a7b6d4e8f90" )); } + +#[test] +fn queued_user_request_ids_are_append_idempotency_keys() { + assert!(is_idempotent_message_id("agent:run-7")); + assert!(is_idempotent_message_id("user:request-7")); + assert!(!is_idempotent_message_id("ui:message-7")); +} diff --git a/crates/tinyagents-session/src/transcript/background_tests.rs b/crates/tinyagents-session/src/transcript/background_tests.rs index 45c4528cb..e433706c4 100644 --- a/crates/tinyagents-session/src/transcript/background_tests.rs +++ b/crates/tinyagents-session/src/transcript/background_tests.rs @@ -20,6 +20,7 @@ fn meta() -> TranscriptMeta { updated: "2026-10-06T00:00:00Z".into(), turn_count: 1, prefix_message_count: None, + prefix_reset: false, input_tokens: 0, output_tokens: 0, cached_input_tokens: 0, diff --git a/crates/tinyagents-session/src/transcript/history.rs b/crates/tinyagents-session/src/transcript/history.rs index c8c5f558b..37d2d5ac7 100644 --- a/crates/tinyagents-session/src/transcript/history.rs +++ b/crates/tinyagents-session/src/transcript/history.rs @@ -134,6 +134,20 @@ pub trait TranscriptHistory: TranscriptRead { /// Appends one durable message while preserving all of its fields. fn append(&self, message: TranscriptMessage) -> anyhow::Result<()>; + /// Adopt imported model-context messages only while this transcript has + /// never been written. Returns `true` when this call installed the import + /// and `false` when a competing or earlier turn already owns the session. + /// + /// Store implementations must make the empty check and write atomic. + /// The default refuses the operation so a storage backend cannot silently + /// provide a racy import. + fn adopt_if_empty(&self, messages: &[TranscriptMessage]) -> anyhow::Result { + if messages.is_empty() { + return Ok(false); + } + anyhow::bail!("transcript store does not support atomic history adoption") + } + /// Replaces the logical model context by appending a compaction record. fn replace(&self, messages: &[TranscriptMessage]) -> anyhow::Result<()>; @@ -344,6 +358,21 @@ pub trait TranscriptLocator: Send + Sync { self.open_stem(&session_stem(session), seed) } + /// Atomically install imported history for a not-yet-written session. + /// A previously written session wins unchanged; concurrent importers + /// serialize through the bound transcript's append contract. + fn adopt_messages_if_absent( + &self, + session: &SessionRef, + seed: TranscriptMeta, + messages: &[TranscriptMessage], + ) -> anyhow::Result { + if messages.is_empty() { + return Ok(false); + } + self.open_session(session, seed)?.adopt_if_empty(messages) + } + /// Folds any pre-identity transcripts of `thread_id` into `session`, once. /// /// A conversation written before session identity existed is spread across @@ -596,6 +625,7 @@ pub(super) fn seed_meta_for_discovered(agent_name: &str) -> TranscriptMeta { updated: String::new(), turn_count: 0, prefix_message_count: None, + prefix_reset: false, input_tokens: 0, output_tokens: 0, cached_input_tokens: 0, diff --git a/crates/tinyagents-session/src/transcript/jsonl.rs b/crates/tinyagents-session/src/transcript/jsonl.rs index d26103645..91d5920d4 100644 --- a/crates/tinyagents-session/src/transcript/jsonl.rs +++ b/crates/tinyagents-session/src/transcript/jsonl.rs @@ -104,6 +104,8 @@ pub(super) struct MetaPayload { pub(super) turn_count: usize, #[serde(default, skip_serializing_if = "Option::is_none")] pub(super) prefix_message_count: Option, + #[serde(default, skip_serializing_if = "is_false")] + pub(super) prefix_reset: bool, pub(super) input_tokens: u64, pub(super) output_tokens: u64, pub(super) cached_input_tokens: u64, @@ -295,6 +297,7 @@ fn meta_payload_from(meta: &TranscriptMeta) -> MetaPayload { updated: meta.updated.clone(), turn_count: meta.turn_count, prefix_message_count: meta.prefix_message_count, + prefix_reset: meta.prefix_reset, input_tokens: meta.input_tokens, output_tokens: meta.output_tokens, cached_input_tokens: meta.cached_input_tokens, @@ -751,6 +754,7 @@ pub(super) fn meta_from_payload(mp: MetaPayload) -> TranscriptMeta { updated: mp.updated, turn_count: mp.turn_count, prefix_message_count: mp.prefix_message_count, + prefix_reset: mp.prefix_reset, input_tokens: mp.input_tokens, output_tokens: mp.output_tokens, cached_input_tokens: mp.cached_input_tokens, diff --git a/crates/tinyagents-session/src/transcript/legacy_md.rs b/crates/tinyagents-session/src/transcript/legacy_md.rs index 6f6dd3660..fe7c24167 100644 --- a/crates/tinyagents-session/src/transcript/legacy_md.rs +++ b/crates/tinyagents-session/src/transcript/legacy_md.rs @@ -73,6 +73,7 @@ fn parse_legacy_meta(raw: &str) -> Result { updated: get("updated").unwrap_or_default(), turn_count: get("turn_count").and_then(|s| s.parse().ok()).unwrap_or(0), prefix_message_count: None, + prefix_reset: false, input_tokens: get("input_tokens") .and_then(|s| s.parse().ok()) .unwrap_or(0), diff --git a/crates/tinyagents-session/src/transcript/spend_tests.rs b/crates/tinyagents-session/src/transcript/spend_tests.rs index 308f114f7..90e1ba82c 100644 --- a/crates/tinyagents-session/src/transcript/spend_tests.rs +++ b/crates/tinyagents-session/src/transcript/spend_tests.rs @@ -26,6 +26,7 @@ pub(super) fn meta(agent: &str, agent_type: &str, thread_id: Option<&str>) -> Tr updated: "2026-09-22T01:00:00Z".into(), turn_count: 0, prefix_message_count: None, + prefix_reset: false, // Every fixture leaves the denormalised `_meta` rollups at zero: that is // exactly what is on disk for every root transcript written since // `33566d382`, and the aggregate must not depend on them. diff --git a/crates/tinyagents-session/src/transcript/transcript_tests.rs b/crates/tinyagents-session/src/transcript/transcript_tests.rs index 7636b41ef..2d0343972 100644 --- a/crates/tinyagents-session/src/transcript/transcript_tests.rs +++ b/crates/tinyagents-session/src/transcript/transcript_tests.rs @@ -22,6 +22,7 @@ fn meta() -> TranscriptMeta { updated: "2026-01-01T00:00:00Z".into(), turn_count: 1, prefix_message_count: None, + prefix_reset: false, input_tokens: 1, output_tokens: 2, cached_input_tokens: 0, @@ -51,6 +52,20 @@ fn frozen_prefix_count_round_trips_in_meta() { ); } +#[test] +fn reset_context_marker_round_trips_in_meta() { + let dir = tempdir().unwrap(); + let path = resolve_keyed_transcript_path(dir.path(), "reset-context").unwrap(); + let mut seed = meta(); + seed.prefix_message_count = None; + seed.prefix_reset = true; + write_transcript(&path, &[], &seed, None).unwrap(); + + let restored = read_transcript(&path).unwrap().meta; + assert!(restored.prefix_reset); + assert_eq!(restored.prefix_message_count, None); +} + #[test] fn jsonl_round_trip_keeps_raw_tool_arguments_and_provider_extension() { let dir = tempdir().unwrap(); diff --git a/crates/tinyagents-session/src/transcript/typed_rows_tests.rs b/crates/tinyagents-session/src/transcript/typed_rows_tests.rs index a0909670d..ab8db0b7a 100644 --- a/crates/tinyagents-session/src/transcript/typed_rows_tests.rs +++ b/crates/tinyagents-session/src/transcript/typed_rows_tests.rs @@ -24,6 +24,7 @@ fn meta() -> TranscriptMeta { updated: "2026-01-01T00:00:00Z".into(), turn_count: 1, prefix_message_count: None, + prefix_reset: false, input_tokens: 0, output_tokens: 0, cached_input_tokens: 0, diff --git a/crates/tinyagents-session/src/transcript/types.rs b/crates/tinyagents-session/src/transcript/types.rs index 35eafcbc9..8e1242488 100644 --- a/crates/tinyagents-session/src/transcript/types.rs +++ b/crates/tinyagents-session/src/transcript/types.rs @@ -689,6 +689,11 @@ pub struct TranscriptMeta { /// generation. `None` on older transcripts that predate this boundary. #[serde(default, skip_serializing_if = "Option::is_none")] pub prefix_message_count: Option, + /// Explicitly marks a successor generation as an intentionally reset + /// model context. Unlike `None` on legacy transcripts, this suppresses + /// prompt-boundary recovery from the sealed parent. + #[serde(default)] + pub prefix_reset: bool, /// Cumulative input tokens across all provider calls this session. pub input_tokens: u64, /// Cumulative output tokens across all provider calls this session. diff --git a/crates/tinyagents-session/src/transcript/view/project.rs b/crates/tinyagents-session/src/transcript/view/project.rs index bc278e863..9c4c277de 100644 --- a/crates/tinyagents-session/src/transcript/view/project.rs +++ b/crates/tinyagents-session/src/transcript/view/project.rs @@ -289,6 +289,7 @@ impl Projector { display_content: sanitized, request_id: msg.request_id.clone(), ts: msg.ts.clone(), + extra_metadata: msg.message.extra_metadata.clone(), }); } "assistant" => self.assistant(msg), @@ -304,6 +305,7 @@ impl Projector { model: msg.turn_usage.as_ref().map(|tu| tu.model.clone()), iteration: msg.iteration, ts: msg.ts.clone(), + extra_metadata: msg.message.extra_metadata.clone(), }); } } @@ -389,6 +391,7 @@ impl Projector { model: msg.turn_usage.as_ref().map(|tu| tu.model.clone()), iteration: Some(iteration), ts: msg.ts.clone(), + extra_metadata: msg.message.extra_metadata.clone(), }); } diff --git a/crates/tinyagents-session/src/transcript/view/transcript_ordering_tests.rs b/crates/tinyagents-session/src/transcript/view/transcript_ordering_tests.rs index f5b44a50c..8c4070610 100644 --- a/crates/tinyagents-session/src/transcript/view/transcript_ordering_tests.rs +++ b/crates/tinyagents-session/src/transcript/view/transcript_ordering_tests.rs @@ -26,6 +26,7 @@ fn meta(thread_id: &str, session_id: Option, parent: Option) -> updated: "2023-11-14T22:00:00+00:00".into(), turn_count: 1, prefix_message_count: None, + prefix_reset: false, input_tokens: 0, output_tokens: 0, cached_input_tokens: 0, diff --git a/crates/tinyagents-session/src/transcript/view/transcript_view_tests.rs b/crates/tinyagents-session/src/transcript/view/transcript_view_tests.rs index 4154c668f..516a7d692 100644 --- a/crates/tinyagents-session/src/transcript/view/transcript_view_tests.rs +++ b/crates/tinyagents-session/src/transcript/view/transcript_view_tests.rs @@ -147,6 +147,27 @@ fn projects_turn_with_tools_reasoning_and_sanitization() { } } +#[test] +fn user_message_display_item_preserves_host_metadata_for_voice_note_duration() { + let dir = TempDir::new().unwrap(); + let path = write_raw( + dir.path(), + "200_orchestrator", + "thr_voice_note", + &[ + r#"{"role":"user","content":"Remember the dentist","extra_metadata":{"voiceNote":{"noteId":"vn-1","durationMs":12000}}}"#, + ], + ); + let display = read_transcript_display(&path).unwrap(); + let items = project_records(&display.records); + + let value = serde_json::to_value(&items[0]).unwrap(); + assert_eq!( + value["extraMetadata"]["voiceNote"]["durationMs"], 12000, + "the host needs the recorded duration to render its voice-note row after reload" + ); +} + #[test] fn reuses_synthetic_tool_call_ids_in_a_later_turn() { let dir = TempDir::new().unwrap(); @@ -474,6 +495,7 @@ fn append_transcript_turn_projects_full_display_shape() { updated: now, turn_count: 1, prefix_message_count: None, + prefix_reset: false, input_tokens: 30, output_tokens: 13, cached_input_tokens: 0, diff --git a/crates/tinyagents-session/src/transcript/view/types.rs b/crates/tinyagents-session/src/transcript/view/types.rs index 5a794f121..10d78dec8 100644 --- a/crates/tinyagents-session/src/transcript/view/types.rs +++ b/crates/tinyagents-session/src/transcript/view/types.rs @@ -94,6 +94,9 @@ pub enum DisplayItem { /// before timestamps were persisted — never backfilled. #[serde(skip_serializing_if = "Option::is_none")] ts: Option, + /// Opaque metadata persisted on the host's user-message row. + #[serde(skip_serializing_if = "Option::is_none")] + extra_metadata: Option, }, /// An assistant answer. `interim: true` marks a non-terminal tool-calling /// step within a multi-iteration turn (not the final answer bubble). @@ -109,6 +112,9 @@ pub enum DisplayItem { iteration: Option, #[serde(skip_serializing_if = "Option::is_none")] ts: Option, + /// Opaque metadata persisted on the host's assistant-message row. + #[serde(skip_serializing_if = "Option::is_none")] + extra_metadata: Option, }, /// The model's reasoning/thinking that preceded an assistant message. /// `iteration` is the model call it belongs to — the same value as the diff --git a/crates/tinyagents-session/src/turn_state/mirror/mod_tests.rs b/crates/tinyagents-session/src/turn_state/mirror/mod_tests.rs index 7a9b746ee..2cc9fe2c8 100644 --- a/crates/tinyagents-session/src/turn_state/mirror/mod_tests.rs +++ b/crates/tinyagents-session/src/turn_state/mirror/mod_tests.rs @@ -103,6 +103,7 @@ fn seed_root_transcript( updated: "2026-07-21T00:00:00Z".into(), turn_count: 1, prefix_message_count: None, + prefix_reset: false, input_tokens: 0, output_tokens: 0, cached_input_tokens: 0, diff --git a/crates/tinyagents-session/src/turn_state/shape_tests.rs b/crates/tinyagents-session/src/turn_state/shape_tests.rs index d72146d98..409aa430b 100644 --- a/crates/tinyagents-session/src/turn_state/shape_tests.rs +++ b/crates/tinyagents-session/src/turn_state/shape_tests.rs @@ -36,3 +36,35 @@ fn snapshot_written_before_optional_fields_still_loads() { assert_eq!(state.lifecycle, TurnLifecycle::Interrupted); assert!(state.transcript.is_empty()); } + +#[test] +fn turn_state_json_schema_tracks_the_serialized_wire_names() { + let schema = serde_json::to_value( + schemars::generate::SchemaSettings::draft2020_12() + .for_serialize() + .into_generator() + .into_root_schema_for::(), + ) + .unwrap(); + let properties = schema["properties"].as_object().unwrap(); + for field in [ + "threadId", + "requestId", + "maxIterations", + "toolTimeline", + "transcript", + "startedAt", + "updatedAt", + ] { + assert!(properties.contains_key(field), "schema is missing {field}"); + } + let required = schema["required"].as_array().unwrap(); + for field in ["threadId", "streamingText", "thinking", "toolTimeline"] { + assert!(required.iter().any(|required| required == field)); + } + assert!(!required.iter().any(|required| required == "transcript")); + let subagent = &schema["$defs"]["SubagentActivity"]; + let subagent_required = subagent["required"].as_array().unwrap(); + assert!(subagent_required.iter().any(|field| field == "toolCalls")); + assert!(!subagent_required.iter().any(|field| field == "transcript")); +} diff --git a/crates/tinyagents-session/src/turn_state/types.rs b/crates/tinyagents-session/src/turn_state/types.rs index 841eae0f1..6d86dafff 100644 --- a/crates/tinyagents-session/src/turn_state/types.rs +++ b/crates/tinyagents-session/src/turn_state/types.rs @@ -15,7 +15,7 @@ use serde::{Deserialize, Serialize}; /// progress signal arrives. `Interrupted` is stamped at startup on /// any snapshot that survived a process restart — there is no live /// driver to resume it, so the UI should surface a retry affordance. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)] #[serde(rename_all = "snake_case")] pub enum TurnLifecycle { Started, @@ -29,7 +29,7 @@ pub enum TurnLifecycle { } /// High-level phase the agent is in within an iteration. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)] #[serde(rename_all = "snake_case")] pub enum TurnPhase { Thinking, @@ -38,7 +38,7 @@ pub enum TurnPhase { } /// Per-tool entry shown in the live timeline UI. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)] #[serde(rename_all = "snake_case")] pub enum ToolTimelineStatus { Running, @@ -54,7 +54,7 @@ pub enum ToolTimelineStatus { /// settled/reloaded turn keeps its "why + what to do next" copy across a thread /// switch or a cold boot. Absent on successful rows and on snapshots written /// before this field. -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)] #[serde(rename_all = "camelCase")] pub struct PersistedToolFailure { /// Stable failure-class variant name, e.g. `"Timeout"`, `"Denied"`. @@ -74,7 +74,7 @@ pub struct PersistedToolFailure { /// Field names use camelCase on the wire so a snapshot can be applied /// directly to `chatRuntimeSlice.toolTimelineByThread` without a /// translation layer in the UI. -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, schemars::JsonSchema)] #[serde(rename_all = "camelCase")] pub struct ToolTimelineEntry { pub id: String, @@ -112,7 +112,7 @@ pub struct ToolTimelineEntry { } /// Live sub-agent activity nested under a `subagent:*` timeline row. -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, schemars::JsonSchema)] #[serde(rename_all = "camelCase")] pub struct SubagentActivity { pub task_id: String, @@ -164,7 +164,7 @@ pub struct SubagentActivity { } /// One child tool call performed by a running sub-agent. -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, schemars::JsonSchema)] #[serde(rename_all = "camelCase")] pub struct SubagentToolCall { pub call_id: String, @@ -221,7 +221,7 @@ pub struct SubagentToolCall { /// union 1:1 (order = push order; no `seq` is needed because each sub-agent's /// transcript is built as a single ordered list). Persisting these lets the /// inline "Agentic task insights" thoughts survive a settled turn / reload. -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, schemars::JsonSchema)] // `rename_all` renames the variant tags; `rename_all_fields` renames the // fields *inside* the struct variants (call_id → callId, …) — without the // latter the FE would read `undefined` for camelCase fields. @@ -272,7 +272,7 @@ pub enum SubagentTranscriptItem { /// (round alone can't order narration vs thinking within one round). Tool /// items hold only a `call_id` pointer into [`TurnState::tool_timeline`] so /// the row's status/label live in exactly one place. -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, schemars::JsonSchema)] // `rename_all_fields` is required so the `ToolCall.call_id` field serializes as // `callId` (the FE reads camelCase) — `rename_all` alone only renames variants. #[serde( @@ -314,7 +314,7 @@ pub enum TranscriptItem { /// "View processing" panel can replay the finished turn's transcript); a /// non-terminal snapshot surviving startup is marked /// [`TurnLifecycle::Interrupted`]. -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, schemars::JsonSchema)] #[serde(rename_all = "camelCase")] pub struct TurnState { pub thread_id: String, diff --git a/integration-proofs/run-ledger-mongodb/Cargo.lock b/integration-proofs/run-ledger-mongodb/Cargo.lock index 7d7eaa5d9..f80ac737b 100644 --- a/integration-proofs/run-ledger-mongodb/Cargo.lock +++ b/integration-proofs/run-ledger-mongodb/Cargo.lock @@ -503,6 +503,12 @@ dependencies = [ "syn 3.0.9", ] +[[package]] +name = "dyn-clone" +version = "1.0.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d0881ea181b1df73ff77ffaaf9c7544ecc11e82fba9b5f27b262a3c73a332555" + [[package]] name = "either" version = "1.19.0" @@ -1711,6 +1717,26 @@ dependencies = [ "bitflags", ] +[[package]] +name = "ref-cast" +version = "1.0.27" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7e440fb4e4b4147295338efb76001ab9e4efc0e5839df2c47fc5ac2381d365c3" +dependencies = [ + "ref-cast-impl", +] + +[[package]] +name = "ref-cast-impl" +version = "1.0.27" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92ecd8964f8453721699a1ed72037b0db49ce2f5a5138486ee89bed6f67cdf3a" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.9", +] + [[package]] name = "regex" version = "1.13.1" @@ -1931,6 +1957,32 @@ dependencies = [ "winapi-util", ] +[[package]] +name = "schemars" +version = "1.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ab508826f74a77ca9d5aba6ff19b522583ee3eaf28a19384ff3d0e5835fadf6e" +dependencies = [ + "chrono", + "dyn-clone", + "ref-cast", + "schemars_derive", + "serde", + "serde_json", +] + +[[package]] +name = "schemars_derive" +version = "1.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e1c3a92094fa7d61aa124645844facb6b554dfc797136d0f5fd1f890e2bffc69" +dependencies = [ + "proc-macro2", + "quote", + "serde_derive_internals", + "syn 3.0.9", +] + [[package]] name = "scopeguard" version = "1.2.0" @@ -1983,6 +2035,17 @@ dependencies = [ "syn 3.0.9", ] +[[package]] +name = "serde_derive_internals" +version = "0.30.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f852137cce035d6a4df67ccce505ff6b3e9fd3a10e3e52b24dc71e650bb1a9bd" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.9", +] + [[package]] name = "serde_json" version = "1.0.154" @@ -2394,6 +2457,7 @@ dependencies = [ "chrono", "hex", "parking_lot", + "schemars", "serde", "serde_json", "sha2 0.11.1",