1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
mod output;
mod receive;
mod tools;
use crate::{
AgentEvent, AgentEventSink, AgentRequest, ApprovalDecider, CancellationToken, ChatProvider,
TokenUsage, ToolDefinition, ToolExecutor,
};
pub use output::{AgentError, AgentLimits, AgentOutput};
#[allow(clippy::too_many_arguments)]
pub async fn run_agent_with_sink(
provider: &dyn ChatProvider,
tools: &dyn ToolExecutor,
request: AgentRequest,
definitions: Vec<ToolDefinition>,
limits: AgentLimits,
approval: &dyn ApprovalDecider,
sink: &dyn AgentEventSink,
cancellation: CancellationToken,
) -> Result<AgentOutput, AgentError> {
let mut messages = crate::history::build_messages(
request.system_prompt.as_deref(),
&request.context_blocks,
&request.prompt,
&request.history,
)?;
let mut events = Vec::new();
let mut tool_count = 0;
let mut used_bounded_sql_query = false;
let mut tool_metadata = Vec::new();
let mut usage = TokenUsage::default();
for _ in 0..limits.max_turns {
check_cancelled(&cancellation)?;
let (assistant, turn_usage) = receive::receive(
provider,
&request.model,
&messages,
&definitions,
sink,
&cancellation,
&mut events,
)
.await?;
// Providers report cumulative counts per response; sum across turns.
usage.input_tokens += turn_usage.input_tokens;
usage.output_tokens += turn_usage.output_tokens;
messages.push(assistant.clone());
if assistant.tool_calls.is_empty() {
check_cancelled(&cancellation)?;
emit(&mut events, sink, AgentEvent::Complete).await;
return Ok(AgentOutput {
answer: assistant.content,
events,
used_bounded_sql_query,
tool_metadata,
usage,
});
}
// When every call in the message is valid and auto-runnable, the
// calls are independent: run them concurrently instead of paying
// their latency sequentially. `auto_runnable` is the single policy
// for "may this run with no questions asked"; the sequential path
// below applies the same gates, so the two cannot drift.
let batch_parallel = assistant.tool_calls.len() > 1
&& assistant.tool_calls.iter().all(|call| {
tools::invalid_reason(call, &definitions).is_none()
&& definitions
.iter()
.find(|definition| definition.name == call.name)
.is_some_and(|definition| tools::auto_runnable(definition, &limits))
});
if batch_parallel {
tool_count += assistant.tool_calls.len();
if tool_count > limits.max_tool_calls {
return Err(AgentError::Limit("tool calls"));
}
check_cancelled(&cancellation)?;
for call in &assistant.tool_calls {
emit(
&mut events,
sink,
AgentEvent::ToolRequested {
name: call.name.clone(),
arguments: call.arguments.clone(),
},
)
.await;
if definitions
.iter()
.find(|definition| definition.name == call.name)
.is_some_and(|definition| definition.effect.database_data)
{
used_bounded_sql_query = true;
}
}
let results =
tools::execute_batch(tools, &assistant.tool_calls.clone(), &definitions).await;
for (call, (result, summary)) in assistant.tool_calls.iter().zip(results) {
let (message, truncated) =
tools::tool_message(call.id.clone(), result, limits.context_byte_budget);
tool_metadata.push(crate::ToolMetadata {
name: call.name.clone(),
status: if summary.contains("failed") {
"failed"
} else {
"completed"
}
.into(),
});
messages.push(message);
check_cancelled(&cancellation)?;
emit(
&mut events,
sink,
AgentEvent::ToolCompleted {
name: call.name.clone(),
summary: output::completion_summary(summary, truncated),
},
)
.await;
}
continue;
}
for call in assistant.tool_calls {
if let Some(reason) = tools::invalid_reason(&call, &definitions) {
if call.id.trim().is_empty() {
return Err(AgentError::InvalidToolCall);
}
tool_count += 1;
if tool_count > limits.max_tool_calls {
return Err(AgentError::Limit("tool calls"));
}
check_cancelled(&cancellation)?;
emit(
&mut events,
sink,
AgentEvent::ToolRequested {
name: call.name.clone(),
arguments: call.arguments.clone(),
},
)
.await;
// Feed the problem back as the tool result so the model can
// retry with a valid call on its next turn.
tool_metadata.push(crate::ToolMetadata {
name: call.name.clone(),
status: "failed".into(),
});
let (message, _) = tools::tool_message(
call.id,
serde_json::json!({"error": reason}),
limits.context_byte_budget,
);
messages.push(message);
emit(
&mut events,
sink,
AgentEvent::ToolCompleted {
name: call.name,
summary: "tool call failed validation".into(),
},
)
.await;
continue;
}
tool_count += 1;
if tool_count > limits.max_tool_calls {
return Err(AgentError::Limit("tool calls"));
}
check_cancelled(&cancellation)?;
emit(
&mut events,
sink,
AgentEvent::ToolRequested {
name: call.name.clone(),
arguments: call.arguments.clone(),
},
)
.await;
let definition = definitions
.iter()
.find(|tool| tool.name == call.name)
.expect("validated");
let approved = !definition.effect.requires_approval
|| approval.approve(definition, &call.arguments).await;
// Apply the same policy the batch path consults (`auto_runnable`),
// split into its gates so the denial can name which one refused.
// `requires_approval` was already resolved into `approved`, so a
// tool that needed approval and got it still runs; the remaining
// gates bind whether or not approval was granted. This is the one
// place the sequential path decides auto-run — keeping it here in
// terms of the shared gates means a gate added to `tools.rs`
// cannot apply to the batch path and not this one.
let candidate_denied = tools::candidate_denied(definition, &limits);
let side_effect_denied = tools::external_side_effect_gated(definition);
let executed = approved && !candidate_denied && !side_effect_denied;
let (result, summary) = if executed {
check_cancelled(&cancellation)?;
// Indicates a database-row-producing query tool ran.
if definition.effect.database_data {
used_bounded_sql_query = true;
}
tools::execute(tools, &call.name, call.arguments, definition.read_only).await
} else {
emit(
&mut events,
sink,
AgentEvent::ToolDenied {
name: call.name.clone(),
reason: if side_effect_denied {
"external side effect requires approval".into()
} else if candidate_denied {
"candidate writes are not permitted".into()
} else {
"approval was not granted".into()
},
},
)
.await;
(
serde_json::json!({"error":"tool call denied by approval policy"}),
"read-only database tool denied",
)
};
tool_metadata.push(crate::ToolMetadata {
name: call.name.clone(),
status: if executed {
if summary.contains("failed") {
"failed"
} else {
"completed"
}
} else {
"denied"
}
.into(),
});
let (message, truncated) =
tools::tool_message(call.id, result, limits.context_byte_budget);
messages.push(message);
if executed {
check_cancelled(&cancellation)?;
emit(
&mut events,
sink,
AgentEvent::ToolCompleted {
name: call.name,
summary: output::completion_summary(summary, truncated),
},
)
.await;
}
}
// Intra-loop context budget: the pre-loop trim bounds history, but
// assistant turns and tool results accumulate here. Trim the oldest
// tool-result pairs (the assistant turn that issued each call plus its
// `tool` message) until the conversation fits, keeping the newest
// context — the same recency policy the pre-loop path uses. A single
// result is already capped at construction, so this resolves
// accumulation; if trimming everything still leaves the newest result
// over budget, truncate it rather than aborting the whole run (S4
// invariant 1: one result must never kill the run by itself).
output::trim_to_budget(&mut messages, limits.context_byte_budget);
}
Err(AgentError::Limit("turns"))
}
pub(super) async fn emit(
events: &mut Vec<AgentEvent>,
sink: &dyn AgentEventSink,
event: AgentEvent,
) {
sink.emit(event.clone()).await;
events.push(event);
}
fn check_cancelled(token: &CancellationToken) -> Result<(), AgentError> {
if token.is_cancelled() {
Err(AgentError::Cancelled)
} else {
Ok(())
}
}