repositories / smith
smith
There are many coding harnesses - but this one is fast
owned by admin
smith-core/src/agent.rs
Raw//! 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<QueuedInput>,
/// Secret placeholder prefix used by [`SecretProxy`]-masked content.
secrets: SecretProxy,
/// Redacted provider effect records spooled by the transport.
provider_records: Option<smith::provider::RecordDrain>,
}
/// 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<String>, value: impl Into<String>) {
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<StreamEvent>) -> Vec<StreamEvent> {
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<AgentEvent>,
}
/// 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<String>) -> 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<String>,
records: Option<smith::provider::RecordDrain>,
) -> 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<smith::id::EntryId> {
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<String>) -> Result<TurnOutcome> {
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<Vec<TurnOutcome>> {
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<AgentEvent>) -> Result<TurnOutcome> {
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<AgentEvent>,
) -> Result<Round> {
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<ProviderRequest> {
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<serde_json::Value> = 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<Vec<StreamEvent>>,
) -> (
StreamFn,
std::sync::Arc<std::sync::Mutex<Vec<ProviderRequest>>>,
) {
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<EntryContent> = 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<StreamEvent> {
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<StreamEvent> {
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<ToolCallId> = 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<StreamEvent> = 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<TurnOutcome>| -> Vec<String> {
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<Vec<String>> = 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<Value = (Vec<(String, String)>, 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}");
}
}
}
}