69
approvals: Arc<dyn ApprovalHandler>,
70
config: AgentConfig,
71
) -> Self {
72
- Self {
73
- engine: ChatEngine::new(),
74
- sink,
75
- tools: tokio::sync::RwLock::new(HashMap::new()),
76
- executor,
77
- approvals,
78
- session_grants: tokio::sync::Mutex::new(HashSet::new()),
79
- turn: tokio::sync::Mutex::new(()),
80
- cancelled: AtomicBool::new(false),
81
- agent_config: config.validated(),
82
- }
72
+ Self::with_agent_and_app_id(sink, executor, approvals, config, None)
73
}
74
75
/// Associate telemetry with an Onde app id. See [`ChatEngine::with_app_id`].
76
pub fn with_app_id(sink: S, onde_app_id: Option<String>) -> Self {
77
+ Self::with_agent_and_app_id(
78
+ sink,
79
+ Arc::new(RejectingExecutor),
80
+ Arc::new(DenyApprovals),
81
+ AgentConfig::default(),
82
+ onde_app_id,
83
+ )
84
+ }
85
+
86
+ /// The full constructor: tools, approvals, and an Onde app id together.
87
+ ///
88
+ /// The narrower constructors each fill in defaults for what they don't
89
+ /// take. Use this one when you need both host tools and telemetry, which
90
+ /// the others cannot express between them.
91
+ pub fn with_agent_and_app_id(
92
+ sink: S,
93
+ executor: Arc<dyn ToolExecutor>,
94
+ approvals: Arc<dyn ApprovalHandler>,
95
+ config: AgentConfig,
96
+ onde_app_id: Option<String>,
97
+ ) -> Self {
98
Self {
99
engine: ChatEngine::with_app_id(onde_app_id),
100
sink,
101
tools: tokio::sync::RwLock::new(HashMap::new()),
91
- executor: Arc::new(RejectingExecutor),
92
- approvals: Arc::new(DenyApprovals),
102
+ executor,
103
+ approvals,
104
session_grants: tokio::sync::Mutex::new(HashSet::new()),
105
turn: tokio::sync::Mutex::new(()),
106
cancelled: AtomicBool::new(false),
96
- agent_config: AgentConfig::default(),
107
+ agent_config: config.validated(),
108
}
109
}
110
213
/// Never returns an error: a failure is reported in
214
/// [`ChatReply::error`] so a caller has one shape to render either way.
215
/// The reply also goes to the sink, so a host can render it from an event
205
- /// instead of awaiting this call — see [`StatusSink::replied`].
216
+ /// instead of awaiting this call — see [`EventSink::replied`].
217
pub async fn send(&self, message: impl Into<String>) -> ChatReply {
218
let reply = match self.engine.send_message(message).await {
219
Ok(result) => ChatReply::ok(result.text, result.duration_display),
324
}
325
326
/// Run a complete local agent turn, including any requested host tools.
327
+ ///
328
+ /// With no tools registered this is one inference call. Otherwise Ed
329
+ /// alternates between the model and the host's executor until the model
330
+ /// answers in text, or until
331
+ /// [`max_tool_rounds`](AgentConfig::max_tool_rounds) is spent, whichever
332
+ /// comes first. Reaching the budget is reported to the sink as a warning,
333
+ /// and yields [`AgentError::ToolRoundsExhausted`] if there is no usable
334
+ /// text by then.
335
+ ///
336
+ /// Reports [`EventSink::agent_replied`] on success. It does *not* report
337
+ /// [`EventSink::replied`]; that belongs to [`send`](Ed::send).
338
pub async fn run(&self, message: impl Into<String>) -> Result<AgentReply, AgentError> {
339
let _turn = self.turn.lock().await;
340
self.cancelled.store(false, Ordering::Release);
341
342
let registered = self.tools.read().await.clone();
343
if registered.is_empty() {
322
- let reply = self.send(message).await;
323
- return match (reply.reply, reply.duration, reply.error) {
324
- (Some(text), Some(duration), None) => {
325
- let agent_reply = AgentReply {
326
- text,
327
- duration_seconds: 0.0,
328
- duration,
329
- tool_rounds: 0,
330
- };
331
- self.sink.agent_replied(&agent_reply);
332
- Ok(agent_reply)
333
- }
334
- (_, _, Some(reason)) => Err(AgentError::Inference { reason }),
335
- _ => Err(AgentError::EmptyReply),
344
+ // Deliberately not routed through `send`: that reports a
345
+ // `ChatReply` to the sink, and a host listening for both events
346
+ // would render this turn's answer twice. An agent turn reports
347
+ // `agent_replied` and nothing else.
348
+ let result =
349
+ self.engine
350
+ .send_message(message)
351
+ .await
352
+ .map_err(|error| AgentError::Inference {
353
+ reason: error.to_string(),
354
+ })?;
355
+ if result.text.trim().is_empty() {
356
+ return Err(AgentError::EmptyReply);
357
+ }
358
+ let agent_reply = AgentReply {
359
+ text: result.text,
360
+ duration_seconds: result.duration_secs,
361
+ duration: result.duration_display,
362
+ tool_rounds: 0,
363
};
364
+ self.sink.agent_replied(&agent_reply);
365
+ return Ok(agent_reply);
366
}
367
368
match self.engine.tool_calling_support().await {
392
})?;
393
let mut duration_seconds = result.duration_secs;
394
let mut rounds = 0_u8;
395
+ let max_rounds = self.agent_config.max_tool_rounds;
396
+ let mut exhausted = false;
397
398
while !result.tool_calls.is_empty() {
399
+ if !self.agent_config.allows_round(rounds) {
400
+ // The budget only stops Ed *offering* tools; it cannot stop the
401
+ // model asking. `onde` parses tool calls out of every response
402
+ // whether or not tools were sent, so a model that keeps
403
+ // emitting tool-call syntax would loop here until `rounds`
404
+ // overflowed and wrapped back under the limit. This is the
405
+ // loop's only hard exit.
406
+ //
407
+ // The model's last turn is already in history carrying these
408
+ // calls. Leaving them unanswered would make the next turn's
409
+ // message sequence invalid, so close them out the way
410
+ // cancelling does.
411
+ let unanswered = result
412
+ .tool_calls
413
+ .iter()
414
+ .map(|pending| ToolResult {
415
+ tool_call_id: pending.id.clone(),
416
+ content: "tool was not executed: the agent ran out of tool rounds"
417
+ .to_string(),
418
+ })
419
+ .collect();
420
+ self.engine.record_tool_results(unanswered).await;
421
+ self.sink.warning(&format!(
422
+ "the agent stopped after {max_rounds} tool rounds; the model kept requesting tools"
423
+ ));
424
+ exhausted = true;
425
+ break;
426
+ }
427
rounds += 1;
428
let mut tool_results = Vec::with_capacity(result.tool_calls.len());
429
460
return Err(AgentError::Cancelled);
461
}
462
404
- let next_tools = if rounds < self.agent_config.max_tool_rounds {
463
+ let next_tools = if self.agent_config.allows_round(rounds) {
464
Some(definitions.as_slice())
465
} else {
466
None
476
}
477
478
if result.text.trim().is_empty() {
420
- return Err(AgentError::EmptyReply);
479
+ return Err(if exhausted {
480
+ AgentError::ToolRoundsExhausted { rounds: max_rounds }
481
+ } else {
482
+ AgentError::EmptyReply
483
+ });
484
}
485
let reply = AgentReply {
486
text: result.text,
511
///
512
/// Leaves the model loaded — this starts a new conversation, it doesn't
513
/// tear down the engine.
514
+ ///
515
+ /// Also drops every `AllowForSession` grant. The user approved those
516
+ /// mutating tools for a conversation, and this ends the conversation;
517
+ /// carrying the grants into the next one would keep consent alive past the
518
+ /// thing it was given for.
519
pub async fn clear_history(&self) -> usize {
520
let cleared = self.engine.clear_history().await;
521
+ self.session_grants.lock().await.clear();
522
info!("ed: cleared {cleared} messages");
523
cleared
524
}
761
assert!(ed.session_grants.lock().await.contains("save_note"));
762
}
763
764
+ #[tokio::test]
765
+ async fn clearing_the_conversation_revokes_session_grants() {
766
+ // The grant was given for a conversation. Clearing the history ends
767
+ // that conversation, so the next one has to ask again.
768
+ let executor = Arc::new(CountingExecutor(AtomicUsize::new(0)));
769
+ let ed = Ed::with_agent(
770
+ NoopSink,
771
+ executor.clone(),
772
+ Arc::new(FixedApproval(ApprovalDecision::AllowForSession)),
773
+ AgentConfig::default(),
774
+ );
775
+ ed.register_tool(AgentToolDefinition::mutating(
776
+ "save_note",
777
+ "Save a note",
778
+ r#"{"type":"object"}"#,
779
+ ))
780
+ .await
781
+ .unwrap();
782
+ let tools = ed.tools.read().await.clone();
783
+ ed.execute_tool_call(&call("save_note"), &tools).await;
784
+ assert!(ed.session_grants.lock().await.contains("save_note"));
785
+
786
+ ed.clear_history().await;
787
+
788
+ assert!(ed.session_grants.lock().await.is_empty());
789
+ }
790
+
791
#[tokio::test]
792
async fn unknown_tools_return_an_error_result_instead_of_executing() {
793
let executor = Arc::new(CountingExecutor(AtomicUsize::new(0)));