//! The provider-independent agent loop (`SMH-SPEC-SPEC0001`, Tools and //! Compaction cost and secrets). //! //! One runtime drives every interface mode: it sends provider-neutral //! requests, streams normalized events, executes tool calls through the //! recorded [`ToolSession`](crate::tools::ToolSession) boundary, and settles //! turns with ordered, paired durable events. Vendor details stop at the //! [`StreamFn`](smith::provider::StreamFn) it consumes. use crate::session::{COMPACTION_KIND, EntryContent, EntryKind, PROVIDER_HTTP_KIND}; use crate::tools::ToolSession; use smith::error::{Result, SmithError}; use smith::id::ToolCallId; use smith::message::{ContentBlock, Message, Role}; use smith::provider::{ProviderRequest, StreamFn}; use smith::stream::{StopReason, StreamEvent, Usage}; use smith::tool::CancelHandle; use std::collections::{HashMap, HashSet, VecDeque}; /// Observable agent lifecycle event, in emission order. #[derive(Clone, Debug, PartialEq, Eq)] pub enum AgentEvent { /// A turn started for this input. TurnStarted { /// The user input that began the turn. input: String, }, /// A normalized provider event surfaced during the turn. Streamed { /// The provider event, unchanged. event: StreamEvent, }, /// A tool call began executing. ToolStarted { /// Tool name. name: String, /// Paired call identifier. call_id: ToolCallId, }, /// A tool call finished. ToolFinished { /// Tool name. name: String, /// Paired call identifier. call_id: ToolCallId, /// Whether the recorded effect succeeded. ok: bool, }, /// The turn settled; no further events follow for it. TurnSettled { /// Why the turn ended. reason: StopReason, /// Accumulated usage across the turn's provider requests. cost: Usage, }, } /// How queued input resumes: immediately between provider requests, or once /// the current turn settles. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum Delivery { /// Delivered before the next provider request. BetweenRequests, /// Delivered after the current turn settles. AfterTurn, } /// Queued steering or follow-up input. #[derive(Clone, Debug, PartialEq, Eq)] pub struct QueuedInput { /// Text to deliver as a user message. pub text: String, /// When the input may be delivered. pub delivery: Delivery, } /// The agent runtime. pub struct Agent { stream: StreamFn, tools: ToolSession, model: String, cancel: CancelHandle, queue: VecDeque, /// Secret placeholder prefix used by [`SecretProxy`]-masked content. secrets: SecretProxy, /// Redacted provider effect records spooled by the transport. provider_records: Option, } /// Replacement and masking of registered secrets. /// /// Context and provider requests keep `{{secret:name}}` placeholders. /// Placeholders are restored only for authorized local effects; their inputs, /// outputs, events, and errors are masked before recording or exposure. #[derive(Clone, Default)] pub struct SecretProxy { registered: Vec<(String, String)>, } impl SecretProxy { /// An empty proxy. #[must_use] pub fn new() -> Self { Self::default() } /// Register or replace a secret under a name. pub fn register(&mut self, name: impl Into, value: impl Into) { let name = name.into(); let value = value.into(); if let Some((_, registered)) = self .registered .iter_mut() .find(|(registered, _)| registered == &name) { *registered = value; } else { self.registered.push((name, value)); } // Replace longer values first so overlapping secrets cannot leave a // plaintext suffix behind. self.registered .sort_by_key(|entry| std::cmp::Reverse(entry.1.len())); } fn placeholder(name: &str) -> String { format!("{{{{secret:{name}}}}}") } /// Restore placeholders immediately before an authorized local effect. #[must_use] pub fn unmask(&self, text: &str) -> String { let mut out = text.to_string(); for (name, value) in &self.registered { let placeholder = Self::placeholder(name); if out.contains(&placeholder) { out = out.replace(&placeholder, value); } } out } /// Replace registered values with placeholders before content crosses a /// persistence, provider, event, or diagnostic boundary. #[must_use] pub fn mask(&self, text: &str) -> String { let mut out = text.to_string(); self.mask_in_place(&mut out); out } /// [`SecretProxy::mask`] without a copy when `text` holds no secret. fn mask_in_place(&self, text: &mut String) { for (name, value) in &self.registered { if !value.is_empty() && text.contains(value.as_str()) { *text = text.replace(value.as_str(), &Self::placeholder(name)); } } } fn transform_json( value: &serde_json::Value, transform: &impl Fn(&str) -> String, ) -> serde_json::Value { match value { serde_json::Value::String(text) => serde_json::Value::String(transform(text)), serde_json::Value::Array(values) => serde_json::Value::Array( values .iter() .map(|value| Self::transform_json(value, transform)) .collect(), ), serde_json::Value::Object(values) => serde_json::Value::Object( values .iter() .map(|(key, value)| (transform(key), Self::transform_json(value, transform))) .collect(), ), scalar => scalar.clone(), } } /// Mask a decoded `message` in place; texts without a secret are not /// copied. fn mask_message(&self, message: &mut Message) { if self.registered.is_empty() { return; } for block in &mut message.blocks { match block { ContentBlock::Text(text) | ContentBlock::Thinking(text) => { self.mask_in_place(text); } ContentBlock::ToolUse { name, input, .. } => { self.mask_in_place(name); *input = self.mask_json(input); } ContentBlock::ToolResult { output, .. } => self.mask_in_place(output), } } } fn mask_json(&self, value: &serde_json::Value) -> serde_json::Value { Self::transform_json(value, &|text| self.mask(text)) } fn unmask_json(&self, value: &serde_json::Value) -> serde_json::Value { Self::transform_json(value, &|text| self.unmask(text)) } fn mask_stream_events(&self, events: Vec) -> Vec { let mut text = HashMap::new(); let mut thinking = HashMap::new(); for event in &events { match event { StreamEvent::TextDelta { message_id, delta } => { text.entry(*message_id) .or_insert_with(String::new) .push_str(delta); } StreamEvent::ThinkingDelta { message_id, delta } => { thinking .entry(*message_id) .or_insert_with(String::new) .push_str(delta); } _ => {} } } let mut emitted_text = HashSet::new(); let mut emitted_thinking = HashSet::new(); events .into_iter() .filter_map(|event| match event { StreamEvent::TextDelta { message_id, .. } => { emitted_text .insert(message_id) .then(|| StreamEvent::TextDelta { message_id, delta: self.mask(&text[&message_id]), }) } StreamEvent::ThinkingDelta { message_id, .. } => emitted_thinking .insert(message_id) .then(|| StreamEvent::ThinkingDelta { message_id, delta: self.mask(&thinking[&message_id]), }), StreamEvent::ToolUse { message_id, name, input, call_id, } => Some(StreamEvent::ToolUse { message_id, name: self.mask(&name), input: self.mask_json(&input), call_id, }), StreamEvent::ToolResult { message_id, call_id, ok, output, } => Some(StreamEvent::ToolResult { message_id, call_id, ok, output: self.mask(&output), }), StreamEvent::Stop { reason, usage } => Some(StreamEvent::Stop { reason, usage }), StreamEvent::Error { error } => Some(StreamEvent::Error { error: self.mask(&error), }), }) .collect() } fn mask_error(&self, error: SmithError) -> SmithError { match error { SmithError::Session { code, message } => SmithError::Session { code, message: self.mask(&message), }, SmithError::Provider { fault } => SmithError::Provider { fault: match fault { smith::error::ProviderFault::Transient { message } => { smith::error::ProviderFault::Transient { message: self.mask(&message), } } smith::error::ProviderFault::Invalid { field, message } => { smith::error::ProviderFault::Invalid { field, message: self.mask(&message), } } smith::error::ProviderFault::Authentication { message } => { smith::error::ProviderFault::Authentication { message: self.mask(&message), } } smith::error::ProviderFault::Overloaded { message } => { smith::error::ProviderFault::Overloaded { message: self.mask(&message), } } smith::error::ProviderFault::Protocol { message } => { smith::error::ProviderFault::Protocol { message: self.mask(&message), } } other => other, }, }, SmithError::Tool { code, message } => SmithError::Tool { code, message: self.mask(&message), }, SmithError::RecordFailed { tool, mutation_applied, message, } => SmithError::RecordFailed { tool, mutation_applied, message: self.mask(&message), }, SmithError::Config { field, message } => SmithError::Config { field, message: self.mask(&message), }, other => other, } } } /// Result of one settled turn. #[derive(Clone, Debug, PartialEq, Eq)] pub struct TurnOutcome { /// Final assistant text of the turn. pub text: String, /// Why the turn ended. pub reason: StopReason, /// Accumulated usage. pub cost: Usage, /// Events emitted during the turn. pub events: Vec, } /// A compaction summary; incomplete summaries are never persisted. #[derive(Clone, Debug, PartialEq, Eq)] pub struct CompactionSummary { /// Summary text. pub text: String, /// Whether the summarizer considers the summary complete. pub complete: bool, } /// One consumed provider round. struct Round { /// Accumulated assistant text. text: String, /// Complete tool uses, in emission order. tools: Vec<(ToolCallId, String, serde_json::Value)>, /// Terminal stop with usage, when the round ended cleanly. terminal: Option<(StopReason, Usage)>, } impl Agent { /// An agent over a stream function and a recorded tool session. #[must_use] pub fn new(stream: StreamFn, tools: ToolSession, model: impl Into) -> Self { Self::with_provider_records(stream, tools, model, None) } /// An agent whose provider rounds drain redacted effect records into /// the session after each round. #[must_use] pub fn with_provider_records( stream: StreamFn, tools: ToolSession, model: impl Into, records: Option, ) -> Self { let cancel = tools.cancel_handle(); Self { stream, tools, model: model.into(), cancel, queue: VecDeque::new(), secrets: SecretProxy::new(), provider_records: records, } } /// A handle that aborts the running turn. #[must_use] pub fn cancel_handle(&self) -> CancelHandle { self.cancel.clone() } /// Replace a settled turn's cancellation handle. /// /// Call only between turns. The returned handle controls the next turn's /// provider requests and local tool effects. pub fn renew_cancel_handle(&mut self) -> CancelHandle { let cancel = CancelHandle::new(); self.cancel = cancel.clone(); self.tools.set_cancel_handle(cancel.clone()); cancel } /// Mutable access to the secret proxy before turns run. pub const fn secrets_mut(&mut self) -> &mut SecretProxy { &mut self.secrets } /// The live session, for inspection and persistence by the caller. #[must_use] pub const fn session(&self) -> &crate::session::Session { &self.tools.session } /// Queue steering or follow-up input. /// /// Queued input is delivered only between provider requests, never /// mid-request, so an injected message can never split a tool call from /// its result. pub fn queue(&mut self, input: QueuedInput) { self.queue.push_back(input); } /// Persist a compaction boundary and reject incomplete summaries. /// /// # Errors /// /// Returns a [`SmithError::Session`] fault when `summary` is incomplete; /// nothing is persisted in that case. pub fn compact(&mut self, summary: &CompactionSummary) -> Result { if !summary.complete { return Err(SmithError::Session { code: "COMPACTION_INCOMPLETE".to_string(), message: "refusing to persist a summary known to be incomplete".to_string(), }); } let masked = self.secrets.mask(&summary.text); let entry = self.tools.record(&EntryContent::Meta { kind: COMPACTION_KIND.to_string(), detail: serde_json::json!({ "summary": masked, "complete": true, }), })?; Ok(entry.id) } /// Run one user turn to settlement. /// /// # Errors /// /// Returns [`SmithError::Cancelled`] when aborted, or the first /// provider, tool, or recording failure; durable state stays consistent /// with what was recorded before the failure. pub fn run_turn(&mut self, input: impl Into) -> Result { let input = input.into(); let masked_input = self.secrets.mask(&input); let mut events = vec![AgentEvent::TurnStarted { input: masked_input.clone(), }]; self.record_user_message(&masked_input)?; let outcome = self.drive_settled(&mut events)?; Ok(outcome) } /// Consume queued after-turn input as follow-up turns. /// /// # Errors /// /// Same as [`Agent::run_turn`]. pub fn run_queued(&mut self) -> Result> { let mut outcomes = Vec::new(); while let Some(input) = self.queue.pop_front() { outcomes.push(self.run_turn(input.text)?); } Ok(outcomes) } fn drive_settled(&mut self, events: &mut Vec) -> Result { let mut cost = Usage::new(); loop { self.check_cancelled()?; self.deliver_between_request_queue()?; let request = self.build_request()?; let stream = (self.stream)(request, self.cancel.clone()); let Round { text, tools: pending_tools, terminal, } = self.consume(stream, events)?; self.drain_provider_records()?; let Some((reason, usage)) = terminal else { return Err(SmithError::Provider { fault: smith::error::ProviderFault::Incomplete, }); }; cost = Usage { input_tokens: cost.input_tokens + usage.input_tokens, output_tokens: cost.output_tokens + usage.output_tokens, input_cache_hit_tokens: cost.input_cache_hit_tokens + usage.input_cache_hit_tokens, input_cache_write_tokens: cost.input_cache_write_tokens + usage.input_cache_write_tokens, }; // The assistant message carries its text and every tool use it // issued, so the provider sees the same pairing the session // recorded. let masked_text = self.secrets.mask(&text); if !masked_text.is_empty() || !pending_tools.is_empty() { let mut message = Message::new(Role::Assistant); if !masked_text.is_empty() { message.add_block(ContentBlock::text(masked_text.clone())); } for (call_id, name, tool_input) in &pending_tools { message.add_block(ContentBlock::tool_use( self.secrets.mask(name), self.secrets.mask_json(tool_input), *call_id, )); } self.record_message(message)?; } let mut tool_results = Vec::new(); for (call_id, name, tool_input) in pending_tools { let recorded_name = self.secrets.mask(&name); let recorded_input = self.secrets.mask_json(&tool_input); let effect_input = self.secrets.unmask_json(&recorded_input); events.push(AgentEvent::ToolStarted { name: recorded_name.clone(), call_id, }); let secrets = &self.secrets; let outcome = self .tools .invoke_local( call_id, &recorded_name, &recorded_input, &effect_input, |text| secrets.mask(text), ) .map_err(|error| self.secrets.mask_error(error))?; events.push(AgentEvent::ToolFinished { name: recorded_name, call_id, ok: outcome.ok, }); tool_results.push(ContentBlock::tool_result( call_id, outcome.ok, self.secrets.mask(&outcome.output), )); } // Tool results return as one Tool-role message, paired and // ordered exactly as the calls were issued. if !tool_results.is_empty() { let mut message = Message::new(Role::Tool); for block in tool_results { message.add_block(block); } self.record_message(message)?; } self.check_cancelled()?; match reason { StopReason::ToolUse => {} settled => { events.push(AgentEvent::TurnSettled { reason: settled, cost, }); return Ok(TurnOutcome { text: masked_text, reason: settled, cost, events: events.clone(), }); } } } } /// Poll one provider stream to its single terminal outcome. fn consume( &self, stream: smith::provider::ProviderStream, events: &mut Vec, ) -> Result { let mut round = Round { text: String::new(), tools: Vec::new(), terminal: None, }; let mut streamed = Vec::new(); for item in futures::executor::block_on_stream(stream) { self.check_cancelled()?; let event = item.map_err(|error| self.secrets.mask_error(error))?; if let StreamEvent::Error { error } = &event { return Err(SmithError::Provider { fault: smith::error::ProviderFault::Protocol { message: self.secrets.mask(error), }, }); } match &event { StreamEvent::TextDelta { delta, .. } => round.text.push_str(delta), StreamEvent::ThinkingDelta { .. } | StreamEvent::ToolResult { .. } => {} StreamEvent::ToolUse { name, input, call_id, .. } => round.tools.push((*call_id, name.clone(), input.clone())), StreamEvent::Stop { reason, usage } => { round.terminal = Some((*reason, usage.unwrap_or_default())); // Terminal is emitted once; the stream ends here. break; } StreamEvent::Error { .. } => unreachable!("errors return before event handling"), } streamed.push(event); } events.extend( self.secrets .mask_stream_events(streamed) .into_iter() .map(|event| AgentEvent::Streamed { event }), ); Ok(round) } /// Decode the provider context into this request: the active branch's /// messages since the compaction boundary. This is the one per-request /// reader of entry content. fn build_request(&self) -> Result { let session = &self.tools.session; let since = session.compaction_boundary(); let mut messages = Vec::new(); let mut include = since.is_none(); for entry in session.active_branch().entries() { if since == Some(entry.id) { include = true; continue; } if !include || !matches!(entry.kind, EntryKind::Message { .. }) { continue; } // `content` verifies the record kind, so a message record // decodes to a message or fails with `ENTRY_KIND_MISMATCH`. if let EntryContent::Message(mut message) = session.content(entry)? { // Provider-bound context never receives registered plaintext. self.secrets.mask_message(&mut message); messages.push(message); } } let mut request = ProviderRequest::new(self.model.clone(), messages); request.tools = self.tools.registered_tools(); Ok(request) } fn record_user_message(&mut self, text: &str) -> Result<()> { self.record_message(Message::with_text(Role::User, text)) } fn record_message(&mut self, message: Message) -> Result<()> { self.tools.record(&EntryContent::Message(message))?; Ok(()) } /// Drain spooled provider effect records into the session as metadata /// entries; every round's records land before its assistant message. fn drain_provider_records(&mut self) -> Result<()> { let Some(drain) = &self.provider_records else { return Ok(()); }; // Records are drained in bulk before any session entry is built. let drained: Vec = drain.try_iter().collect(); for detail in drained { self.tools.record(&EntryContent::Meta { kind: PROVIDER_HTTP_KIND.to_string(), detail, })?; } Ok(()) } fn deliver_between_request_queue(&mut self) -> Result<()> { loop { let next = self .queue .front() .is_some_and(|input| input.delivery == Delivery::BetweenRequests); if !next { return Ok(()); } let input = self.queue.pop_front(); if let Some(input) = input { self.record_user_message(&self.secrets.mask(&input.text))?; } } } fn check_cancelled(&self) -> Result<()> { if self.cancel.is_cancelled() { return Err(SmithError::Cancelled); } Ok(()) } } #[cfg(test)] #[expect( clippy::unwrap_used, reason = "agent tests assert on scripted mock streams" )] mod tests { use super::*; use crate::session::Session; use crate::store::open_session; use futures::StreamExt; use smith::config::Config; use smith::message::{ContentBlock, Role}; use smith::stream::StopReason; /// Mock stream that records the requests it saw. fn recording( responses: Vec>, ) -> ( StreamFn, std::sync::Arc>>, ) { let seen = std::sync::Arc::new(std::sync::Mutex::new(Vec::new())); let responses = std::sync::Arc::new(std::sync::Mutex::new(responses)); let seen_for_stream = std::sync::Arc::clone(&seen); let stream: StreamFn = std::sync::Arc::new(move |request, _cancel| { let index = seen_for_stream.lock().unwrap().len(); seen_for_stream.lock().unwrap().push(request); let response = responses.lock().unwrap()[index].clone(); futures::stream::iter(response.into_iter().map(Ok)).boxed() }); (stream, seen) } /// Every decoded entry content, rendered for secret checks. fn decoded(session: &Session) -> String { let contents: Vec = session .active_branch() .entries() .iter() .map(|entry| session.content(entry).unwrap()) .collect(); format!("{contents:?}") } fn temp_tools(tag: &str) -> ToolSession { let dir = std::env::temp_dir().join(format!("smith_agent_{tag}_{}", std::process::id())); let _ = std::fs::remove_dir_all(&dir); std::fs::create_dir_all(&dir).unwrap(); ToolSession::new(&dir) } fn tool_use_turn_response() -> Vec { vec![ StreamEvent::text_delta(smith::id::MessageId::new(), "Writing the file."), StreamEvent::tool_use( smith::id::MessageId::new(), "write", serde_json::json!({"path": "note.txt", "content": "hi"}), smith::id::ToolCallId::new(), ), StreamEvent::stop(StopReason::ToolUse, Some(Usage::new())), ] } fn final_response() -> Vec { vec![ StreamEvent::text_delta(smith::id::MessageId::new(), "Done."), StreamEvent::stop(StopReason::EndTurn, Some(Usage::new())), ] } #[test] fn secret_registration_handles_rotation_and_overlapping_values() { let mut secrets = SecretProxy::new(); secrets.register("short", "token"); secrets.register("long", "token-suffix"); assert_eq!(secrets.mask("token-suffix"), "{{secret:long}}"); secrets.register("long", "replacement"); assert_eq!(secrets.unmask("{{secret:long}}"), "replacement"); assert_eq!(secrets.mask("replacement"), "{{secret:long}}"); } #[test] fn mock_provider_completes_a_tool_using_turn_with_paired_durable_events() { let (stream, _seen) = recording(vec![tool_use_turn_response(), final_response()]); let tools = temp_tools("turn"); let mut agent = Agent::new(stream, tools, "mock-smith"); let outcome = agent.run_turn("write hi to note.txt").unwrap(); assert_eq!(outcome.reason, StopReason::EndTurn); assert_eq!(outcome.text, "Done."); // Durable order: user, assistant, call, result, assistant. let kinds: Vec<&str> = agent .session() .active_branch() .entries() .iter() .map(|entry| match entry.kind { EntryKind::Message { role, .. } => match role { Role::User => "user", Role::Assistant => "assistant", Role::Tool => "tool", Role::System => "system", }, EntryKind::ToolCall { .. } => "call", EntryKind::ToolResult { .. } => "result", EntryKind::Meta { .. } => "meta", }) .collect(); assert_eq!( kinds, vec!["user", "assistant", "call", "result", "tool", "assistant"] ); // The tool effect happened. assert_eq!( std::fs::read_to_string( std::env::temp_dir() .join(format!("smith_agent_turn_{}", std::process::id())) .join("note.txt") ) .unwrap_or_default(), "hi" ); // Tool lifecycle events bracket the recorded pair. let names: Vec<&str> = outcome .events .iter() .filter_map(|event| match event { AgentEvent::ToolStarted { name, .. } | AgentEvent::ToolFinished { name, .. } => { Some(name.as_str()) } _ => None, }) .collect(); assert_eq!(names, vec!["write", "write"]); } #[test] fn executed_tool_failure_returns_to_provider_and_turn_continues() { let call_id = smith::id::ToolCallId::new(); let failed_read = vec![ StreamEvent::tool_use( smith::id::MessageId::new(), "read", serde_json::json!({"path": "missing.txt"}), call_id, ), StreamEvent::stop(StopReason::ToolUse, None), ]; let (stream, seen) = recording(vec![failed_read, final_response()]); let tools = temp_tools("tool_failure"); let mut agent = Agent::new(stream, tools, "mock-smith"); let outcome = agent.run_turn("read the missing file").unwrap(); assert_eq!(outcome.reason, StopReason::EndTurn); assert!(outcome.events.iter().any(|event| matches!( event, AgentEvent::ToolFinished { call_id: finished, ok: false, .. } if *finished == call_id ))); { let requests = seen.lock().unwrap(); assert_eq!(requests.len(), 2); assert!(requests[1].messages.iter().any(|message| { message.blocks.iter().any(|block| { matches!( block, ContentBlock::ToolResult { call_id: result, ok: false, output, } if *result == call_id && output.contains("READ_ERROR") ) }) })); drop(requests); } let durable_ids: Vec = agent .session() .active_branch() .entries() .iter() .filter_map(|entry| match entry.kind { EntryKind::ToolCall { call_id } | EntryKind::ToolResult { call_id, .. } => { Some(call_id) } _ => None, }) .collect(); assert_eq!(durable_ids, vec![call_id, call_id]); } #[test] fn steering_is_delivered_between_requests_and_never_splits_pairs() { let third: Vec = vec![StreamEvent::stop(StopReason::EndTurn, None)]; let (stream, seen) = recording(vec![tool_use_turn_response(), final_response(), third]); let tools = temp_tools("steering"); let mut agent = Agent::new(stream, tools, "mock-smith"); agent.queue(QueuedInput { text: "also uppercase it".to_string(), delivery: Delivery::BetweenRequests, }); agent.run_turn("write hi").unwrap(); // Second request already carries the steering message: user, tool // exchange, steering. The recorded pair stays adjacent. let second = { let requests = seen.lock().unwrap(); requests[1].clone() }; assert_eq!(second.messages.len(), 4); assert!( second.messages.iter().any(|message| matches!( message.blocks.first(), Some(ContentBlock::Text(text)) if text.contains("uppercase") )), "steering message missing from {second:?}" ); let kinds: Vec<&str> = agent .session() .active_branch() .entries() .iter() .map(|entry| match entry.kind { EntryKind::ToolCall { .. } => "call", EntryKind::ToolResult { .. } => "result", _ => "other", }) .collect(); let pair_position = kinds .windows(2) .position(|window| window == ["call", "result"]) .unwrap(); // Steering was recorded after the pair, never inside it. assert!(pair_position + 2 <= kinds.len()); } #[test] fn turn_cost_sums_the_usage_of_every_round() { let usage = |input, output, hit, write| Usage { input_tokens: input, output_tokens: output, input_cache_hit_tokens: hit, input_cache_write_tokens: write, }; let mut tool_round = tool_use_turn_response(); tool_round.pop(); tool_round.push(StreamEvent::stop( StopReason::ToolUse, Some(usage(3, 5, 7, 11)), )); let final_round = vec![StreamEvent::stop( StopReason::EndTurn, Some(usage(2, 4, 6, 8)), )]; let (stream, _seen) = recording(vec![tool_round, final_round]); let mut agent = Agent::new(stream, temp_tools("cost"), "mock-smith"); let outcome = agent.run_turn("write hi").unwrap(); assert_eq!(outcome.cost, usage(5, 9, 13, 19)); assert!(matches!( outcome.events.last(), Some(AgentEvent::TurnSettled { cost, .. }) if *cost == outcome.cost )); } #[test] fn queued_input_is_delivered_in_order_within_a_deadline() { let (stream, seen) = recording(vec![final_response(), final_response()]); let mut agent = Agent::new(stream, temp_tools("queued"), "mock-smith"); agent.queue(QueuedInput { text: "steer".to_string(), delivery: Delivery::BetweenRequests, }); agent.queue(QueuedInput { text: "follow up".to_string(), delivery: Delivery::AfterTurn, }); // The agent runs off the test thread so a delivery loop that never // settles fails the deadline instead of hanging the test. let (done, settled) = std::sync::mpsc::channel(); std::thread::spawn(move || { let texts = |outcomes: Vec| -> Vec { outcomes.into_iter().map(|outcome| outcome.text).collect() }; let first = agent.run_turn("start").map(|outcome| texts(vec![outcome])); let _ = done.send((first, agent.run_queued().map(texts))); }); let (first, queued) = settled .recv_timeout(std::time::Duration::from_secs(10)) .expect("queued input settled within the deadline"); assert_eq!(first.unwrap(), vec!["Done."]); assert_eq!(queued.unwrap(), vec!["Done."]); let user_texts: Vec> = seen .lock() .unwrap() .iter() .map(|request| { request .messages .iter() .filter(|message| message.role == Role::User) .filter_map(|message| match message.blocks.first() { Some(ContentBlock::Text(text)) => Some(text.clone()), _ => None, }) .collect() }) .collect(); assert_eq!( user_texts, vec![vec!["start", "steer"], vec!["start", "steer", "follow up"]] ); } #[test] fn cancellation_aborts_without_partial_durable_claims() { let (stream, _seen) = recording(vec![tool_use_turn_response()]); let tools = temp_tools("cancel"); let mut agent = Agent::new(stream, tools, "mock-smith"); let cancel = agent.cancel_handle(); cancel.cancel(); let err = agent.run_turn("hello").unwrap_err(); assert_eq!(err.code(), "CANCELLED"); // Only the user message was recorded; no assistant claims exist. assert_eq!(agent.session().active_branch().entries().len(), 1); } #[test] fn request_build_rejects_a_message_record_holding_other_content() { let mut forged = crate::session::EntryFrame::new( None, &EntryContent::ToolResult { ok: true, output: "not a message".to_string(), call_id: ToolCallId::new(), }, ) .unwrap(); forged.kind = EntryKind::Message { id: smith::id::MessageId::new(), role: Role::User, }; let (stream, seen) = recording(vec![final_response()]); let mut tools = temp_tools("kind_mismatch"); tools.session = Session::from_frames(&[crate::frame::Frame::Known { entry: forged }]).unwrap(); let mut agent = Agent::new(stream, tools, "mock-smith"); let err = agent.run_turn("hello").unwrap_err(); assert_eq!(err.code(), "ENTRY_KIND_MISMATCH"); assert!(seen.lock().unwrap().is_empty()); } #[test] fn cancelled_tool_is_paired_before_the_next_turn() { let call_id = smith::id::ToolCallId::new(); let signal = crate::process_signal::ProcessSignal::new(); let running_tool = vec![ StreamEvent::tool_use( smith::id::MessageId::new(), "bash", serde_json::json!({"command": format!("{}; sleep 30", signal.raise()), "timeout_ms": 30_000}), call_id, ), StreamEvent::stop(StopReason::ToolUse, None), ]; let (stream, seen) = recording(vec![running_tool, final_response()]); let tools = temp_tools("cancelled_pair"); let mut agent = Agent::new(stream, tools, "mock-smith"); let cancel = agent.cancel_handle(); // synchronize: cancel once the tool's command reports it runs, never after a guessed delay. let canceller = std::thread::spawn(move || { let mut socket = signal.wait(); cancel.cancel(); crate::process_signal::closed(&mut socket) }); let error = agent.run_turn("start the command").unwrap_err(); assert!(canceller.join().unwrap(), "tool command kept running"); assert_eq!(error.code(), "CANCELLED"); agent.renew_cancel_handle(); agent.run_turn("continue").unwrap(); let requests = seen.lock().unwrap(); let restarted = &requests[1]; let uses = restarted .messages .iter() .flat_map(|message| &message.blocks) .filter(|block| matches!(block, ContentBlock::ToolUse { call_id: id, .. } if *id == call_id)) .count(); let results = restarted .messages .iter() .flat_map(|message| &message.blocks) .filter(|block| matches!(block, ContentBlock::ToolResult { call_id: id, ok: false, .. } if *id == call_id)) .count(); assert_eq!((uses, results), (1, 1)); drop(requests); } #[test] fn secrets_cross_only_the_authorized_local_effect_boundary() { let call_id = smith::id::ToolCallId::new(); let tool_response = vec![ StreamEvent::tool_use( smith::id::MessageId::new(), "write", serde_json::json!({ "path": "secret.txt", "content": "{{secret:api_key}}", }), call_id, ), StreamEvent::stop(StopReason::ToolUse, Some(Usage::new())), ]; let echo_id = smith::id::MessageId::new(); let secret_echo = vec![ StreamEvent::text_delta(echo_id, "s3cr3t-"), StreamEvent::text_delta(echo_id, "value"), StreamEvent::stop(StopReason::EndTurn, Some(Usage::new())), ]; let (stream, seen) = recording(vec![tool_response, secret_echo]); let tools = temp_tools("secrets"); let secret_path = tools.working_dir.join("secret.txt"); let mut agent = Agent::new(stream, tools, "mock-smith"); agent.secrets_mut().register("api_key", "s3cr3t-value"); let outcome = agent.run_turn("use {{secret:api_key}} locally").unwrap(); let requests = seen.lock().unwrap(); let dispatched = format!("{requests:?}"); assert!(!dispatched.contains("s3cr3t-value")); drop(requests); let recorded = decoded(agent.session()); assert!(recorded.contains("{{secret:api_key}}")); assert!(!recorded.contains("s3cr3t-value")); assert_eq!( std::fs::read_to_string(secret_path).unwrap(), "s3cr3t-value" ); let surfaced: String = outcome .events .iter() .filter_map(|event| match event { AgentEvent::Streamed { event: StreamEvent::TextDelta { delta, .. }, } => Some(delta.as_str()), _ => None, }) .collect(); assert!(!surfaced.contains("s3cr3t-value")); assert!(surfaced.contains("{{secret:api_key}}")); } #[test] fn secrets_registered_later_are_masked_in_earlier_recorded_context() { let (stream, seen) = recording(vec![final_response(), final_response()]); let mut agent = Agent::new(stream, temp_tools("late_secret"), "mock-smith"); agent.run_turn("my key is s3cr3t-value").unwrap(); agent.secrets_mut().register("api_key", "s3cr3t-value"); agent.run_turn("again").unwrap(); let requests = seen.lock().unwrap(); let resent = format!("{:?}", requests[1].messages); drop(requests); assert!(resent.contains("my key is {{secret:api_key}}"), "{resent}"); assert!(!resent.contains("s3cr3t-value"), "{resent}"); } #[test] fn secret_bearing_tool_output_stays_out_of_persisted_session() { let dir = std::env::temp_dir().join(format!("smith_agent_secret_output_{}", std::process::id())); let _ = std::fs::remove_dir_all(&dir); std::fs::create_dir_all(&dir).unwrap(); let path = dir.join("session.smh"); let opened = open_session(&path, Config::default_valid()).unwrap(); let tools = ToolSession::with_session(&dir, opened.session, opened.writer); let response = vec![ StreamEvent::tool_use( smith::id::MessageId::new(), "bash", serde_json::json!({"command": "printf '{{secret:api_key}}'"}), smith::id::ToolCallId::new(), ), StreamEvent::stop(StopReason::ToolUse, None), ]; let (stream, seen) = recording(vec![response, final_response()]); let mut agent = Agent::new(stream, tools, "mock-smith"); agent.secrets_mut().register("api_key", "s3cr3t-value"); let outcome = agent.run_turn("run it locally").unwrap(); assert!(!format!("{outcome:?}").contains("s3cr3t-value")); assert!(!format!("{:?}", *seen.lock().unwrap()).contains("s3cr3t-value")); drop(agent); assert!( !std::fs::read(&path) .unwrap() .windows("s3cr3t-value".len()) .any(|window| window == b"s3cr3t-value") ); let reopened = open_session(&path, Config::default_valid()).unwrap(); let restored = decoded(&reopened.session); assert!(restored.contains("{{secret:api_key}}")); assert!(!restored.contains("s3cr3t-value")); let _ = std::fs::remove_dir_all(&dir); } #[test] fn secret_bearing_tool_errors_are_recorded_and_returned_redacted() { let response = vec![ StreamEvent::tool_use( smith::id::MessageId::new(), "bash", serde_json::json!({ "command": "printf '{{secret:api_key}}' >&2; exit 7", }), smith::id::ToolCallId::new(), ), StreamEvent::stop(StopReason::ToolUse, None), ]; let (stream, seen) = recording(vec![response, final_response()]); let tools = temp_tools("secret_error"); let mut agent = Agent::new(stream, tools, "mock-smith"); agent.secrets_mut().register("api_key", "s3cr3t-value"); let outcome = agent.run_turn("run it locally").unwrap(); assert!(!format!("{outcome:?}").contains("s3cr3t-value")); { let requests = seen.lock().unwrap(); let tool_error = requests[1] .messages .iter() .flat_map(|message| &message.blocks) .find_map(|block| match block { ContentBlock::ToolResult { ok: false, output, .. } => Some(output), _ => None, }) .unwrap(); assert!(tool_error.contains("{{secret:api_key}}")); assert!(!tool_error.contains("s3cr3t-value")); drop(requests); } let recorded = decoded(agent.session()); assert!(recorded.contains("{{secret:api_key}}")); assert!(!recorded.contains("s3cr3t-value")); } #[test] fn compaction_rejects_incomplete_summaries_and_resets_context() { let (stream, seen) = recording(vec![final_response(), final_response()]); let tools = temp_tools("compaction"); let mut agent = Agent::new(stream, tools, "mock-smith"); agent.run_turn("first").unwrap(); let err = agent .compact(&CompactionSummary { text: "partial".to_string(), complete: false, }) .unwrap_err(); assert_eq!(err.code(), "COMPACTION_INCOMPLETE"); agent .compact(&CompactionSummary { text: "summary of first".to_string(), complete: true, }) .unwrap(); // Requests after compaction start from the boundary. agent.run_turn("second").unwrap(); let last = { let requests = seen.lock().unwrap(); requests.last().cloned() }; assert!(last.is_some_and(|request| request.messages.len() <= 2)); } #[test] fn durable_turn_survives_reload_with_same_identity() { let dir = std::env::temp_dir().join(format!("smith_agent_durable_{}", std::process::id())); let _ = std::fs::remove_dir_all(&dir); std::fs::create_dir_all(&dir).unwrap(); let path = dir.join("session.smh"); let config = Config::default_valid(); let opened = open_session(&path, config).unwrap(); let tools = ToolSession::with_session(&dir, opened.session, opened.writer); let (stream, _seen) = recording(vec![final_response()]); let mut agent = Agent::new(stream, tools, "mock-smith"); agent.run_turn("persisted").unwrap(); let session_id = agent.session().id(); drop(agent); let reopened = open_session(&path, Config::default_valid()).unwrap(); assert_eq!(reopened.session.id(), session_id); assert_eq!(reopened.session.active_branch().entries().len(), 2); let _ = std::fs::remove_dir_all(&dir); } } /// Secret masking laws (`SMH-SPEC-SPEC0001`, Compaction, cost, and secrets). /// /// Secret values use `[A-Z0-9]`, possibly empty, names `[a-z]`, and filler text never holds /// `{`: no value can match inside a placeholder and no input holds one, the /// precondition under which masking is reversible. #[cfg(test)] mod properties { use super::SecretProxy; use proptest::collection::vec; use proptest::prelude::*; /// Registered secrets and a text interleaving filler with their values. fn secrets_and_text() -> impl Strategy, String)> { vec(("[a-z]{1,6}", "[A-Z0-9]{0,8}"), 1..5).prop_flat_map(|secrets| { let count = secrets.len(); let pieces = vec(("[^{]{0,12}", proptest::option::of(0..count)), 0..6); (Just(secrets), pieces).prop_map(|(secrets, pieces)| { let mut text = String::new(); for (filler, secret) in pieces { text.push_str(&filler); if let Some(index) = secret { text.push_str(&secrets[index].1); } } (secrets, text) }) }) } fn proxy(secrets: &[(String, String)]) -> SecretProxy { let mut proxy = SecretProxy::new(); for (name, value) in secrets { proxy.register(name.clone(), value.clone()); } proxy } proptest! { #[test] fn unmask_restores_masked_text((secrets, text) in secrets_and_text()) { let proxy = proxy(&secrets); prop_assert_eq!(proxy.unmask(&proxy.mask(&text)), text); } #[test] fn masked_text_holds_no_registered_value((secrets, text) in secrets_and_text()) { let masked = proxy(&secrets).mask(&text); // A later registration under the same name replaces the value. let registered: std::collections::HashMap<&str, &str> = secrets .iter() .map(|(name, value)| (name.as_str(), value.as_str())) .collect(); for value in registered.values().filter(|value| !value.is_empty()) { prop_assert!(!masked.contains(value), "{value} survived in {masked}"); } } } }