use super::*;
use crate::{
context::project_provider_request_input_tokens,
output::{ActivityKind, ActivityStatus},
providers::{
AnthropicProvider, HttpRequest, HttpTransport, ProviderConversationItem, ProviderEvent,
ProviderToolResult,
},
sessions::SessionEvent,
};
use serde_json::json;
use std::sync::{
Arc, Mutex,
atomic::{AtomicBool, Ordering},
};
#[derive(Default)]
struct CapturingOutputSink {
text: String,
blocks: Vec<String>,
outputs: Vec<OutputEvent>,
activities: Vec<ActivityEvent>,
}
impl AgentOutputSink for CapturingOutputSink {
fn assistant_delta(&mut self, text: &str) -> anyhow::Result<()> {
self.text.push_str(text);
Ok(())
}
fn output_event(&mut self, event: OutputEvent) -> anyhow::Result<()> {
match &event {
OutputEvent::AssistantDelta { text } => self.text.push_str(text),
OutputEvent::BashCommand { .. }
| OutputEvent::AssistantComplete { .. }
| OutputEvent::ContextUsage { .. }
| OutputEvent::ThinkingSummaryDelta { .. }
| OutputEvent::ThinkingSummaryComplete { .. }
| OutputEvent::ThinkingSummaryCompleteIdentified { .. }
| OutputEvent::Diagnostic { .. }
| OutputEvent::HookDiagnostic { .. }
| OutputEvent::ProviderContextInjection { .. }
| OutputEvent::SubdirInstructionInjection { .. }
| OutputEvent::ToolStarted { .. } => {}
OutputEvent::ToolResult { call, result, .. } => {
self.blocks.push(format_tool_block(call, result));
}
OutputEvent::SessionHeader { .. } | OutputEvent::UserPrompt { .. } => {}
}
self.outputs.push(event);
Ok(())
}
fn activity_event(&mut self, event: ActivityEvent) -> anyhow::Result<()> {
self.activities.push(event);
Ok(())
}
fn tool_block(&mut self, block: &str) -> anyhow::Result<()> {
self.blocks.push(block.to_string());
Ok(())
}
}
struct ProbeActivitySink {
outputs: Vec<OutputEvent>,
activities: Vec<ActivityEvent>,
sender_events: Arc<Mutex<Vec<ActivityEvent>>>,
probe: Arc<dyn Fn(&ActivityEvent) + Send + Sync>,
}
impl ProbeActivitySink {
fn new(probe: impl Fn(&ActivityEvent) + Send + Sync + 'static) -> Self {
Self {
outputs: Vec::new(),
activities: Vec::new(),
sender_events: Arc::new(Mutex::new(Vec::new())),
probe: Arc::new(probe),
}
}
}
impl AgentOutputSink for ProbeActivitySink {
fn assistant_delta(&mut self, _text: &str) -> anyhow::Result<()> {
Ok(())
}
fn output_event(&mut self, event: OutputEvent) -> anyhow::Result<()> {
self.outputs.push(event);
Ok(())
}
fn activity_event(&mut self, event: ActivityEvent) -> anyhow::Result<()> {
self.activities.push(event);
Ok(())
}
fn activity_sender(&self) -> Option<ActivitySender> {
let events = Arc::clone(&self.sender_events);
let probe = Arc::clone(&self.probe);
Some(Arc::new(move |event| {
probe(&event);
events.lock().unwrap().push(event);
}))
}
fn tool_block(&mut self, _block: &str) -> anyhow::Result<()> {
Ok(())
}
}
#[derive(Clone, Copy)]
enum SinkFailurePoint {
ActivityStarted,
ActivityFinished,
AssistantComplete,
ToolResult,
}
struct FailingOutputSink {
failure_point: SinkFailurePoint,
activities: Vec<ActivityEvent>,
outputs: Vec<OutputEvent>,
}
impl FailingOutputSink {
fn new(failure_point: SinkFailurePoint) -> Self {
Self {
failure_point,
activities: Vec::new(),
outputs: Vec::new(),
}
}
}
impl AgentOutputSink for FailingOutputSink {
fn assistant_delta(&mut self, _text: &str) -> anyhow::Result<()> {
Ok(())
}
fn output_event(&mut self, event: OutputEvent) -> anyhow::Result<()> {
if matches!(
(&self.failure_point, &event),
(
SinkFailurePoint::AssistantComplete,
OutputEvent::AssistantComplete { .. }
) | (SinkFailurePoint::ToolResult, OutputEvent::ToolResult { .. })
) {
anyhow::bail!(match self.failure_point {
SinkFailurePoint::AssistantComplete => "sink assistant complete broke",
SinkFailurePoint::ToolResult => "sink tool result broke",
SinkFailurePoint::ActivityStarted | SinkFailurePoint::ActivityFinished => {
"sink output broke"
}
});
}
self.outputs.push(event);
Ok(())
}
fn activity_event(&mut self, event: ActivityEvent) -> anyhow::Result<()> {
if matches!(
(&self.failure_point, &event),
(
SinkFailurePoint::ActivityStarted,
ActivityEvent::Started { .. }
) | (
SinkFailurePoint::ActivityFinished,
ActivityEvent::Finished { .. }
)
) {
anyhow::bail!("sink activity broke");
}
self.activities.push(event);
Ok(())
}
fn tool_block(&mut self, _block: &str) -> anyhow::Result<()> {
Ok(())
}
}
struct SessionOrderProbeSink<'a> {
session: &'a crate::sessions::Session,
events_at_assistant_complete: Vec<String>,
outputs: Vec<OutputEvent>,
}
impl<'a> SessionOrderProbeSink<'a> {
fn new(session: &'a crate::sessions::Session) -> Self {
Self {
session,
events_at_assistant_complete: Vec::new(),
outputs: Vec::new(),
}
}
}
impl AgentOutputSink for SessionOrderProbeSink<'_> {
fn assistant_delta(&mut self, _text: &str) -> anyhow::Result<()> {
Ok(())
}
fn output_event(&mut self, event: OutputEvent) -> anyhow::Result<()> {
if matches!(event, OutputEvent::AssistantComplete { .. })
&& self.events_at_assistant_complete.is_empty()
{
self.events_at_assistant_complete = self
.session
.read_events()?
.into_iter()
.map(|event| event.event_type)
.collect();
}
self.outputs.push(event);
Ok(())
}
fn tool_block(&mut self, _block: &str) -> anyhow::Result<()> {
Ok(())
}
}
struct RecordedRequests(Mutex<Vec<ProviderRequest>>);
impl RecordedRequests {
fn push(&self, request: &ProviderRequest) {
self.0.lock().unwrap().push(request.to_owned_request());
}
fn snapshot(&self) -> Vec<ProviderRequest> {
self.0.lock().unwrap().clone()
}
}
impl Default for RecordedRequests {
fn default() -> Self {
Self(Mutex::new(Vec::new()))
}
}
struct ScriptedProvider {
responses: Mutex<Vec<Vec<ProviderEvent>>>,
requests: RecordedRequests,
}
impl ScriptedProvider {
fn new(responses: Vec<Vec<ProviderEvent>>) -> Self {
Self {
responses: Mutex::new(responses.into_iter().rev().collect()),
requests: RecordedRequests::default(),
}
}
fn requests(&self) -> Vec<ProviderRequest> {
self.requests.snapshot()
}
}
impl Provider for ScriptedProvider {
fn stream_cancellable(
&self,
request: ProviderRequest,
_cancellation: &AgentCancellation,
on_event: &mut dyn FnMut(ProviderEvent) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
self.requests.push(&request);
let events = self.responses.lock().unwrap().pop().unwrap_or_default();
for event in events {
on_event(event)?;
}
Ok(())
}
}
struct CancelingProvider {
cancel: Arc<AtomicBool>,
requests: RecordedRequests,
}
impl CancelingProvider {
fn new(cancel: Arc<AtomicBool>) -> Self {
Self {
cancel,
requests: RecordedRequests::default(),
}
}
}
impl Provider for CancelingProvider {
fn stream_cancellable(
&self,
request: ProviderRequest,
_cancellation: &AgentCancellation,
on_event: &mut dyn FnMut(ProviderEvent) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
self.requests.push(&request);
on_event(read_call("cancel_read"))?;
self.cancel.store(true, Ordering::SeqCst);
on_event(text("ignored"))?;
Ok(())
}
}
struct ActiveCancelingProvider {
cancel: Arc<AtomicBool>,
sent_events: Arc<Mutex<usize>>,
total_events: usize,
}
impl ActiveCancelingProvider {
fn new(cancel: Arc<AtomicBool>, total_events: usize) -> Self {
Self {
cancel,
sent_events: Arc::new(Mutex::new(0)),
total_events,
}
}
fn sent_events(&self) -> usize {
*self.sent_events.lock().unwrap()
}
}
impl Provider for ActiveCancelingProvider {
fn stream_cancellable(
&self,
_request: ProviderRequest,
_cancellation: &AgentCancellation,
on_event: &mut dyn FnMut(ProviderEvent) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
for index in 0..self.total_events {
*self.sent_events.lock().unwrap() += 1;
on_event(text(format!("chunk-{index}")))?;
if index == 0 {
self.cancel.store(true, Ordering::SeqCst);
}
}
Ok(())
}
}
struct FailingProvider {
events: Vec<ProviderEvent>,
}
struct AutoContinueProvider {
steps: Mutex<Vec<AutoContinueStep>>,
requests: RecordedRequests,
}
struct AnthropicPendingToolMessageStopTransport {
requests: Arc<Mutex<Vec<HttpRequest>>>,
}
impl AnthropicPendingToolMessageStopTransport {
fn new() -> Self {
Self {
requests: Arc::new(Mutex::new(Vec::new())),
}
}
fn requests_handle(&self) -> Arc<Mutex<Vec<HttpRequest>>> {
Arc::clone(&self.requests)
}
}
impl HttpTransport for AnthropicPendingToolMessageStopTransport {
fn stream_json(
&self,
request: HttpRequest,
on_chunk: &mut dyn FnMut(&str) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
self.stream_json_cancellable(request, &AgentCancellation::default(), on_chunk)
}
fn stream_json_cancellable(
&self,
request: HttpRequest,
cancellation: &AgentCancellation,
on_chunk: &mut dyn FnMut(&str) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
self.requests.lock().unwrap().push(request);
cancellation.check()?;
on_chunk(
"data: {\"type\":\"content_block_start\",\"index\":1,\"content_block\":{\"type\":\"tool_use\",\"id\":\"toolu_trace_full_id\",\"name\":\"read\"}}\n\n",
)?;
on_chunk(
"data: {\"type\":\"content_block_delta\",\"index\":1,\"delta\":{\"type\":\"input_json_delta\",\"partial_json\":\"{\\\"path\\\":\\\"/tmp\"}}\n\n",
)?;
on_chunk(
"data: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"tool_use\"},\"usage\":{\"output_tokens\":7}}\n\n",
)?;
on_chunk("data: {\"type\":\"message_stop\"}\n\n")
}
fn stream_json_cancellable_with_semantic_deadline(
&self,
request: HttpRequest,
cancellation: &AgentCancellation,
_semantic_deadline: &std::sync::atomic::AtomicU64,
on_chunk: &mut dyn FnMut(&str) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
self.stream_json_cancellable(request, cancellation, on_chunk)
}
}
struct AnthropicHiddenToolProgressTransport {
requests: Arc<Mutex<Vec<HttpRequest>>>,
}
impl AnthropicHiddenToolProgressTransport {
fn new() -> Self {
Self {
requests: Arc::new(Mutex::new(Vec::new())),
}
}
fn requests_handle(&self) -> Arc<Mutex<Vec<HttpRequest>>> {
Arc::clone(&self.requests)
}
}
impl HttpTransport for AnthropicHiddenToolProgressTransport {
fn stream_json(
&self,
request: HttpRequest,
on_chunk: &mut dyn FnMut(&str) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
self.stream_json_cancellable(request, &AgentCancellation::default(), on_chunk)
}
fn stream_json_cancellable(
&self,
request: HttpRequest,
cancellation: &AgentCancellation,
on_chunk: &mut dyn FnMut(&str) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
self.requests.lock().unwrap().push(request);
cancellation.check()?;
on_chunk(
"data: {\"type\":\"content_block_delta\",\"delta\":{\"type\":\"text_delta\",\"text\":\"partial\"}}\n\n",
)?;
on_chunk(
"data: {\"type\":\"content_block_start\",\"index\":1,\"content_block\":{\"type\":\"tool_use\",\"id\":\"call_1\",\"name\":\"read\"}}\n\n",
)?;
Err(crate::providers::ProviderError::stream_terminal(
"provider stream no semantic progress before timeout",
)
.into())
}
fn stream_json_cancellable_with_semantic_deadline(
&self,
request: HttpRequest,
cancellation: &AgentCancellation,
_semantic_deadline: &std::sync::atomic::AtomicU64,
on_chunk: &mut dyn FnMut(&str) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
self.stream_json_cancellable(request, cancellation, on_chunk)
}
}
#[allow(clippy::enum_variant_names)]
enum AutoContinueStep {
TextThenEligibleTimeout(&'static str),
TextThenToolThenEligibleTimeout,
TextThenFunctionItemThenEligibleTimeout,
TextThenHiddenToolProgressTimeout,
TextThenDone(&'static str),
}
impl AutoContinueProvider {
fn new(steps: Vec<AutoContinueStep>) -> Self {
Self {
steps: Mutex::new(steps.into_iter().rev().collect()),
requests: RecordedRequests::default(),
}
}
fn requests(&self) -> Vec<ProviderRequest> {
self.requests.snapshot()
}
}
impl Provider for AutoContinueProvider {
fn stream_cancellable(
&self,
request: ProviderRequest,
_cancellation: &AgentCancellation,
on_event: &mut dyn FnMut(ProviderEvent) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
self.requests.push(&request);
match self.steps.lock().unwrap().pop().unwrap() {
AutoContinueStep::TextThenEligibleTimeout(text) => {
on_event(crate::providers::ProviderEvent::TextDelta(text.to_string()))?;
Err(crate::providers::ProviderError::stream_failed_incomplete(
"provider stream ended prematurely after partial response; response is incomplete: provider stream no semantic progress before timeout",
)
.into())
}
AutoContinueStep::TextThenToolThenEligibleTimeout => {
on_event(text("partial"))?;
on_event(read_call("auto_continue_blocked_tool"))?;
Err(crate::providers::ProviderError::stream_failed_incomplete(
"provider stream ended prematurely after partial response; response is incomplete: provider stream no semantic progress before timeout",
)
.into())
}
AutoContinueStep::TextThenFunctionItemThenEligibleTimeout => {
on_event(text("partial"))?;
on_event(ProviderEvent::ResponseItem(response_function_call_item(
"auto_continue_blocked_item",
"read",
)))?;
Err(crate::providers::ProviderError::stream_failed_incomplete(
"provider stream ended prematurely after partial response; response is incomplete: provider stream no semantic progress before timeout",
)
.into())
}
AutoContinueStep::TextThenHiddenToolProgressTimeout => {
on_event(text("partial"))?;
Err(crate::providers::ProviderError::stream_failed_incomplete(
"provider stream ended prematurely after partial response; response is incomplete: provider stream no semantic progress before timeout; unsafe tool-call progress observed",
)
.into())
}
AutoContinueStep::TextThenDone(text) => {
on_event(crate::providers::ProviderEvent::TextDelta(text.to_string()))?;
on_event(done())
}
}
}
}
struct PostToolContinuationFailingProvider {
requests: RecordedRequests,
}
impl PostToolContinuationFailingProvider {
fn requests(&self) -> Vec<ProviderRequest> {
self.requests.snapshot()
}
}
impl Provider for PostToolContinuationFailingProvider {
fn stream_cancellable(
&self,
request: ProviderRequest,
_cancellation: &AgentCancellation,
on_event: &mut dyn FnMut(ProviderEvent) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
let is_continuation = !request.tool_results().is_empty();
self.requests.push(&request);
if is_continuation {
anyhow::bail!("provider stream no semantic progress before timeout");
}
on_event(read_call("call_final"))?;
on_event(done())?;
Ok(())
}
}
struct ParentSubagentProvider {
requests: RecordedRequests,
}
impl ParentSubagentProvider {
fn requests(&self) -> Vec<ProviderRequest> {
self.requests.snapshot()
}
}
impl Provider for ParentSubagentProvider {
fn stream_cancellable(
&self,
request: ProviderRequest,
_cancellation: &AgentCancellation,
on_event: &mut dyn FnMut(ProviderEvent) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
let is_tool_continuation = !request.tool_results().is_empty();
let is_child = request
.messages()
.last()
.is_some_and(|message| message.content.contains("Subagent g1 task intent"));
self.requests.push(&request);
if is_tool_continuation {
on_event(text("parent done"))?;
} else if is_child {
on_event(text("child raw output"))?;
} else {
on_event(ProviderEvent::ToolCall(ToolCall {
id: "parent_subagents".to_string(),
name: "subagents".to_string(),
arguments: json!({"tasks":[{"intent":"child intent"}],"concurrency":1}),
}))?;
}
on_event(done())?;
Ok(())
}
}
struct ParentSubagentMixedToolProvider {
requests: RecordedRequests,
}
impl ParentSubagentMixedToolProvider {
fn requests(&self) -> Vec<ProviderRequest> {
self.requests.snapshot()
}
}
impl Provider for ParentSubagentMixedToolProvider {
fn stream_cancellable(
&self,
request: ProviderRequest,
_cancellation: &AgentCancellation,
on_event: &mut dyn FnMut(ProviderEvent) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
let tool_results = request.tool_results();
let is_child_prompt = request
.messages()
.iter()
.any(|message| message.content.contains("Subagent g1 task intent"));
self.requests.push(&request);
if tool_results
.iter()
.any(|result| result.call_id == "parent_subagents")
{
on_event(text("parent done"))?;
} else if tool_results
.iter()
.any(|result| result.call_id == "child_read")
{
on_event(text(" child done"))?;
} else if is_child_prompt {
on_event(text("child will read"))?;
on_event(ProviderEvent::ToolCall(ToolCall {
id: "child_read".to_string(),
name: "read".to_string(),
arguments: json!({"path":"file.txt"}),
}))?;
} else {
on_event(ProviderEvent::ToolCall(ToolCall {
id: "parent_subagents".to_string(),
name: "subagents".to_string(),
arguments: json!({"tasks":[{"intent":"child intent"}],"concurrency":1}),
}))?;
}
on_event(done())?;
Ok(())
}
}
struct ParentSubagentHookFailureProvider {
requests: RecordedRequests,
}
impl ParentSubagentHookFailureProvider {
fn requests(&self) -> Vec<ProviderRequest> {
self.requests.snapshot()
}
}
impl Provider for ParentSubagentHookFailureProvider {
fn stream_cancellable(
&self,
request: ProviderRequest,
_cancellation: &AgentCancellation,
on_event: &mut dyn FnMut(ProviderEvent) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
let is_tool_continuation = !request.tool_results().is_empty();
let is_child = request
.messages()
.last()
.is_some_and(|message| message.content.contains("Subagent g1 task intent"));
self.requests.push(&request);
if is_tool_continuation {
on_event(text("parent done"))?;
} else if is_child {
on_event(ProviderEvent::ToolCall(ToolCall {
id: "child_write".to_string(),
name: "write".to_string(),
arguments: json!({"path":"child.txt","content":"child write"}),
}))?;
} else {
on_event(ProviderEvent::ToolCall(ToolCall {
id: "parent_subagents".to_string(),
name: "subagents".to_string(),
arguments: json!({"tasks":[{"intent":"child hook fail"}],"concurrency":1}),
}))?;
}
on_event(done())?;
Ok(())
}
}
struct SessionBreakingProvider {
session_path: std::path::PathBuf,
}
impl Provider for SessionBreakingProvider {
fn stream_cancellable(
&self,
request: ProviderRequest,
_cancellation: &AgentCancellation,
on_event: &mut dyn FnMut(ProviderEvent) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
let _ = request;
if self.session_path.exists() {
std::fs::remove_file(&self.session_path)?;
}
std::fs::create_dir(&self.session_path)?;
on_event(text("still runs"))?;
on_event(done())?;
Ok(())
}
}
impl Provider for FailingProvider {
fn stream_cancellable(
&self,
_request: ProviderRequest,
_cancellation: &AgentCancellation,
on_event: &mut dyn FnMut(ProviderEvent) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
for event in self.events.clone() {
on_event(event)?;
}
anyhow::bail!("provider broke")
}
}
fn event_types(session: &crate::sessions::Session) -> Vec<String> {
session
.read_events()
.unwrap()
.into_iter()
.map(|event| event.event_type)
.collect()
}
fn assert_event_types(session: &crate::sessions::Session, expected: &[&str]) {
assert_eq!(event_types(session), expected);
}
fn assert_assistant_chunk(session: &crate::sessions::Session, expected: &str) {
assert_eq!(
assistant_chunk_text(&session.read_events().unwrap()),
expected
);
}
fn assistant_chunk_text(events: &[SessionEvent]) -> String {
events
.iter()
.filter(|event| event.event_type == "assistant_chunk")
.filter_map(|event| event.payload["text"].as_str())
.collect::<String>()
}
fn text(s: impl Into<String>) -> ProviderEvent {
ProviderEvent::TextDelta(s.into())
}
fn done() -> ProviderEvent {
ProviderEvent::Done
}
fn text_done(s: &str) -> Vec<ProviderEvent> {
vec![text(s), done()]
}
fn read_done(id: &str) -> Vec<ProviderEvent> {
vec![read_call(id), done()]
}
fn tool_call(id: &str, name: &str, arguments: serde_json::Value) -> ProviderEvent {
ProviderEvent::ToolCall(ToolCall {
id: id.to_string(),
name: name.to_string(),
arguments,
})
}
fn run_request<'a, 'sink>(prompt: &'a str, cwd: &'a std::path::Path) -> AgentRunRequest<'a, 'sink> {
AgentRunRequest {
initial_instructions: &[],
ttsr: crate::config::TtsrSettings::default(),
prompt,
tools: None,
hooks: None,
session: None,
cwd,
output_sink: None,
cancellation: AgentCancellation::default(),
session_title_job: None,
semantic_progress_timeout: None,
invocation_mode: crate::output::InvocationMode::Print,
agent_id: None,
herdr_reporter: None,
}
}
fn hook_settings_with(
phase: fn(&mut crate::config::HookSettings) -> &mut Vec<crate::config::HookDefinition>,
label: &str,
command: &str,
policy: Option<crate::config::HookFailurePolicy>,
) -> crate::config::HookSettings {
let mut settings = crate::config::HookSettings {
enabled: true,
..crate::config::HookSettings::default()
};
phase(&mut settings).push(crate::config::HookDefinition {
label: Some(label.into()),
command: command.into(),
failure_policy: policy,
..crate::config::HookDefinition::default()
});
settings
}
fn before_tool_hook(
label: &str,
command: &str,
policy: Option<crate::config::HookFailurePolicy>,
) -> crate::config::HookSettings {
hook_settings_with(|settings| &mut settings.before_tool, label, command, policy)
}
fn after_tool_hook(
label: &str,
command: &str,
policy: Option<crate::config::HookFailurePolicy>,
) -> crate::config::HookSettings {
hook_settings_with(|settings| &mut settings.after_tool, label, command, policy)
}
fn after_assistant_hook(
label: &str,
command: &str,
policy: Option<crate::config::HookFailurePolicy>,
) -> crate::config::HookSettings {
hook_settings_with(
|settings| &mut settings.after_assistant,
label,
command,
policy,
)
}
fn response_function_call_item(call_id: &str, name: &str) -> serde_json::Value {
json!({
"type": "function_call",
"call_id": call_id,
"name": name,
"arguments": "{}",
"status": "completed"
})
}
fn read_call(id: &str) -> ProviderEvent {
tool_call(id, "read", json!({"path":"file.txt:raw"}))
}
fn herdr_statuses(lines: &Arc<Mutex<Vec<String>>>) -> Vec<(String, String, String)> {
lines
.lock()
.unwrap()
.iter()
.map(|line| {
let value: serde_json::Value = serde_json::from_str(line).unwrap();
(
value["params"]["state"].as_str().unwrap().to_string(),
value["params"]["custom_status"]
.as_str()
.unwrap()
.to_string(),
value["params"]["message"].as_str().unwrap().to_string(),
)
})
.collect()
}
fn expected_prompt_cache_key_for_session_id(session_id: &str) -> String {
use sha2::{Digest, Sha256};
let digest = Sha256::digest(session_id.as_bytes());
format!("magi-code-session-{}", crate::hex::lower_hex(digest))
.chars()
.take("magi-code-session-".len() + 32)
.collect()
}
fn expected_prompt_cache_key_for_subagent(
provider_id: &str,
model: &str,
agent_id: &str,
) -> String {
use sha2::{Digest, Sha256};
let material = format!("provider={provider_id}\nmodel={model}\nidentity={agent_id}");
let digest = Sha256::digest(material.as_bytes());
format!("magi-code-session-{}", crate::hex::lower_hex(digest))
.chars()
.take("magi-code-session-".len() + 32)
.collect()
}
fn write_call(id: &str, path: &str, content: &str) -> ProviderEvent {
tool_call(id, "write", json!({"path": path, "content": content}))
}
fn bash_call(id: &str, command: &str) -> ProviderEvent {
tool_call(id, "bash", json!({"command": command}))
}
fn edit_call(id: &str, path: &str, _old_text: &str, new_text: &str) -> ProviderEvent {
tool_call(
id,
"hash_edit",
json!({"input": format!("[{path}#ABCD]\nSWAP 1.=1:\n+{new_text}")}),
)
}
mod cancellation;
mod hooks_redaction;
mod misc;
mod prompt_context;
mod session_replay;
mod titles_herdr;
mod tools_continuation;
mod ttsr;