1#![forbid(unsafe_code)]
2
3use std::error::Error;
4use std::fmt;
5use std::sync::atomic::{AtomicU64, Ordering};
6use std::sync::Arc;
7use std::thread;
8use std::time::{Duration, SystemTime, UNIX_EPOCH};
9
10use phaseo::Phaseo;
11use serde::{Deserialize, Serialize};
12use serde_json::{json, Value};
13
14static RUN_SEQUENCE: AtomicU64 = AtomicU64::new(1);
15
16#[derive(Clone, Debug, PartialEq, Eq)]
17pub struct AgentError {
18 message: String,
19}
20
21impl AgentError {
22 pub fn new(message: impl Into<String>) -> Self {
23 Self {
24 message: message.into(),
25 }
26 }
27
28 pub fn message(&self) -> &str {
29 &self.message
30 }
31}
32
33impl fmt::Display for AgentError {
34 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
35 formatter.write_str(&self.message)
36 }
37}
38
39impl Error for AgentError {}
40
41impl From<phaseo::PhaseoError> for AgentError {
42 fn from(error: phaseo::PhaseoError) -> Self {
43 Self::new(error.to_string())
44 }
45}
46
47#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq)]
48pub struct ToolCall {
49 pub id: String,
50 pub name: String,
51 #[serde(default)]
52 pub input: Value,
53}
54
55#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq)]
56pub struct Message {
57 pub role: String,
58 #[serde(default)]
59 pub content: String,
60 #[serde(default, skip_serializing_if = "Vec::is_empty")]
61 pub tool_calls: Vec<ToolCall>,
62 #[serde(default, skip_serializing_if = "Option::is_none")]
63 pub tool_call_id: Option<String>,
64 #[serde(default, skip_serializing_if = "Option::is_none")]
65 pub name: Option<String>,
66 #[serde(default)]
67 pub is_error: bool,
68}
69
70impl Message {
71 pub fn user(content: impl Into<String>) -> Self {
72 Self {
73 role: "user".to_string(),
74 content: content.into(),
75 ..Self::default()
76 }
77 }
78
79 pub fn assistant(content: impl Into<String>) -> Self {
80 Self {
81 role: "assistant".to_string(),
82 content: content.into(),
83 ..Self::default()
84 }
85 }
86}
87
88#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq)]
89pub struct UsageSummary {
90 pub input_tokens: u64,
91 pub output_tokens: u64,
92 pub cached_tokens: u64,
93 pub total_tokens: u64,
94 pub cost: f64,
95}
96
97impl UsageSummary {
98 fn add(&mut self, other: &Self) {
99 self.input_tokens += other.input_tokens;
100 self.output_tokens += other.output_tokens;
101 self.cached_tokens += other.cached_tokens;
102 self.total_tokens += other.total_tokens;
103 self.cost += other.cost;
104 }
105}
106
107#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq)]
108pub struct ModelRequest {
109 pub agent_id: String,
110 pub model: String,
111 pub instructions: String,
112 pub messages: Vec<Message>,
113 pub tools: Vec<ToolSpec>,
114 #[serde(default)]
115 pub context: Value,
116}
117
118#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq)]
119pub struct ModelResponse {
120 pub message: Message,
121 #[serde(default)]
122 pub usage: UsageSummary,
123 pub request_id: Option<String>,
124 pub provider: Option<String>,
125 pub model: Option<String>,
126 pub finish_reason: Option<String>,
127}
128
129pub trait ModelClient {
130 fn generate(&mut self, request: &ModelRequest) -> Result<ModelResponse, AgentError>;
131}
132
133#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
134pub struct ToolSpec {
135 pub id: String,
136 pub description: String,
137 pub parameters: Value,
138}
139
140pub type ToolExecutor =
141 Arc<dyn Fn(Value, &RuntimeContext) -> Result<Value, AgentError> + Send + Sync>;
142
143#[derive(Clone)]
144pub struct Tool {
145 pub id: String,
146 pub description: String,
147 pub parameters: Value,
148 pub execute: Option<ToolExecutor>,
149 pub require_approval: bool,
150}
151
152impl fmt::Debug for Tool {
153 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
154 formatter
155 .debug_struct("Tool")
156 .field("id", &self.id)
157 .field("description", &self.description)
158 .field("parameters", &self.parameters)
159 .field("has_executor", &self.execute.is_some())
160 .field("require_approval", &self.require_approval)
161 .finish()
162 }
163}
164
165impl Tool {
166 pub fn new<F>(
167 id: impl Into<String>,
168 description: impl Into<String>,
169 parameters: Value,
170 execute: F,
171 ) -> Self
172 where
173 F: Fn(Value, &RuntimeContext) -> Result<Value, AgentError> + Send + Sync + 'static,
174 {
175 Self {
176 id: id.into(),
177 description: description.into(),
178 parameters,
179 execute: Some(Arc::new(execute)),
180 require_approval: false,
181 }
182 }
183
184 pub fn external(
185 id: impl Into<String>,
186 description: impl Into<String>,
187 parameters: Value,
188 ) -> Self {
189 Self {
190 id: id.into(),
191 description: description.into(),
192 parameters,
193 execute: None,
194 require_approval: false,
195 }
196 }
197
198 pub fn require_approval(mut self) -> Self {
199 self.require_approval = true;
200 self
201 }
202
203 fn spec(&self) -> ToolSpec {
204 ToolSpec {
205 id: self.id.clone(),
206 description: self.description.clone(),
207 parameters: self.parameters.clone(),
208 }
209 }
210}
211
212pub fn define_tool(tool: Tool) -> Tool {
213 tool
214}
215
216#[derive(Clone, Debug)]
217pub struct RuntimeContext {
218 pub run_id: String,
219 pub agent_id: String,
220 pub step_index: usize,
221 pub context: Value,
222 pub tool_call: ToolCall,
223}
224
225#[derive(Clone, Debug)]
226pub struct HumanReviewContext {
227 pub run_id: String,
228 pub agent_id: String,
229 pub step_index: usize,
230 pub messages: Vec<Message>,
231 pub response: ModelResponse,
232 pub context: Value,
233}
234
235#[derive(Clone, Debug)]
236pub struct HumanReviewRequest {
237 pub reason: String,
238 pub payload: Value,
239}
240
241pub type HumanReviewer =
242 Arc<dyn Fn(&HumanReviewContext) -> Option<HumanReviewRequest> + Send + Sync>;
243
244#[derive(Clone)]
245pub struct AgentDefinition {
246 pub id: String,
247 pub model: String,
248 pub instructions: String,
249 pub tools: Vec<Tool>,
250 pub max_steps: usize,
251 pub max_retries: usize,
252 pub retry_backoff: Duration,
253 pub human_review: Option<HumanReviewer>,
254}
255
256impl fmt::Debug for AgentDefinition {
257 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
258 formatter
259 .debug_struct("AgentDefinition")
260 .field("id", &self.id)
261 .field("model", &self.model)
262 .field("instructions", &self.instructions)
263 .field("tools", &self.tools)
264 .field("max_steps", &self.max_steps)
265 .field("max_retries", &self.max_retries)
266 .field("retry_backoff", &self.retry_backoff)
267 .field("has_human_review", &self.human_review.is_some())
268 .finish()
269 }
270}
271
272impl AgentDefinition {
273 pub fn new(id: impl Into<String>, model: impl Into<String>) -> Self {
274 Self {
275 id: id.into(),
276 model: model.into(),
277 instructions: String::new(),
278 tools: Vec::new(),
279 max_steps: 8,
280 max_retries: 0,
281 retry_backoff: Duration::from_millis(250),
282 human_review: None,
283 }
284 }
285
286 pub fn instructions(mut self, instructions: impl Into<String>) -> Self {
287 self.instructions = instructions.into();
288 self
289 }
290
291 pub fn tool(mut self, tool: Tool) -> Self {
292 self.tools.push(tool);
293 self
294 }
295
296 pub fn max_steps(mut self, max_steps: usize) -> Self {
297 self.max_steps = max_steps.max(1);
298 self
299 }
300
301 pub fn model_retries(mut self, max_retries: usize, backoff: Duration) -> Self {
302 self.max_retries = max_retries;
303 self.retry_backoff = backoff;
304 self
305 }
306
307 pub fn human_review<F>(mut self, reviewer: F) -> Self
308 where
309 F: Fn(&HumanReviewContext) -> Option<HumanReviewRequest> + Send + Sync + 'static,
310 {
311 self.human_review = Some(Arc::new(reviewer));
312 self
313 }
314}
315
316#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
317pub struct PendingToolCall {
318 pub call: ToolCall,
319 pub kind: String,
320 pub reason: String,
321}
322
323#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
324pub struct HumanPause {
325 pub reason: String,
326 pub payload: Value,
327 pub kind: String,
328 pub pending_tool_calls: Vec<PendingToolCall>,
329}
330
331#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
332pub struct RunStep {
333 pub index: usize,
334 pub status: String,
335 pub model_attempts: usize,
336 pub tool_calls: Vec<ToolCall>,
337 pub request_id: Option<String>,
338 pub provider: Option<String>,
339 pub model: Option<String>,
340 pub finish_reason: Option<String>,
341 pub error: Option<String>,
342 pub usage: UsageSummary,
343}
344
345#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
346pub struct RunRecord {
347 pub id: String,
348 pub agent_id: String,
349 pub model: String,
350 pub max_steps: usize,
351 pub status: String,
352 pub input: Value,
353 pub context: Value,
354 pub step_count: usize,
355 pub pause: Option<HumanPause>,
356 pub stop_reason: Option<String>,
357 pub created_at_ms: u64,
358 pub updated_at_ms: u64,
359}
360
361#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
362pub struct RunResult {
363 pub run: RunRecord,
364 pub steps: Vec<RunStep>,
365 pub output: Value,
366 pub messages: Vec<Message>,
367 pub usage: UsageSummary,
368}
369
370#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
371pub struct AgentEvent {
372 pub event_type: String,
373 pub run_id: String,
374 pub agent_id: String,
375 pub timestamp_ms: u64,
376 pub details: Value,
377}
378
379#[derive(Clone, Debug)]
380pub struct RunOptions {
381 pub input: Value,
382 pub context: Value,
383 pub model: Option<String>,
384 pub max_steps: Option<usize>,
385}
386
387impl RunOptions {
388 pub fn new(input: impl Into<Value>) -> Self {
389 Self {
390 input: input.into(),
391 context: Value::Null,
392 model: None,
393 max_steps: None,
394 }
395 }
396}
397
398#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
399pub struct ToolDecision {
400 pub tool_call_id: String,
401 pub reason: Option<String>,
402}
403
404#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
405pub struct ToolOutput {
406 pub tool_call_id: String,
407 pub output: Value,
408}
409
410#[derive(Clone, Debug)]
411pub struct ContinueOptions {
412 pub result: RunResult,
413 pub human_input: Option<String>,
414 pub approvals: Vec<ToolDecision>,
415 pub tool_outputs: Vec<ToolOutput>,
416}
417
418impl ContinueOptions {
419 pub fn new(result: RunResult) -> Self {
420 Self {
421 result,
422 human_input: None,
423 approvals: Vec::new(),
424 tool_outputs: Vec::new(),
425 }
426 }
427}
428
429pub struct Agent {
430 definition: AgentDefinition,
431}
432
433pub fn create_agent(definition: AgentDefinition) -> Agent {
434 Agent { definition }
435}
436
437impl Agent {
438 pub fn run(
439 &self,
440 client: &mut dyn ModelClient,
441 options: RunOptions,
442 ) -> Result<RunResult, AgentError> {
443 self.run_with_events(client, options, None)
444 }
445
446 pub fn run_with_events(
447 &self,
448 client: &mut dyn ModelClient,
449 options: RunOptions,
450 mut on_event: Option<&mut dyn FnMut(&AgentEvent)>,
451 ) -> Result<RunResult, AgentError> {
452 let timestamp = now_ms();
453 let run_id = new_run_id();
454 let model = options
455 .model
456 .unwrap_or_else(|| self.definition.model.clone());
457 let max_steps = options
458 .max_steps
459 .unwrap_or(self.definition.max_steps)
460 .max(1);
461 let input = options.input;
462 let mut result = RunResult {
463 run: RunRecord {
464 id: run_id.clone(),
465 agent_id: self.definition.id.clone(),
466 model,
467 max_steps,
468 status: "running".to_string(),
469 input: input.clone(),
470 context: options.context,
471 step_count: 0,
472 pause: None,
473 stop_reason: None,
474 created_at_ms: timestamp,
475 updated_at_ms: timestamp,
476 },
477 steps: Vec::new(),
478 output: Value::Null,
479 messages: vec![Message::user(value_to_text(&input))],
480 usage: UsageSummary::default(),
481 };
482 emit(&mut on_event, &result, "run.started", Value::Null);
483 self.drive(client, &mut result, &mut on_event)?;
484 Ok(result)
485 }
486
487 pub fn continue_run(
488 &self,
489 client: &mut dyn ModelClient,
490 options: ContinueOptions,
491 ) -> Result<RunResult, AgentError> {
492 self.continue_with_events(client, options, None)
493 }
494
495 pub fn continue_with_events(
496 &self,
497 client: &mut dyn ModelClient,
498 options: ContinueOptions,
499 mut on_event: Option<&mut dyn FnMut(&AgentEvent)>,
500 ) -> Result<RunResult, AgentError> {
501 let ContinueOptions {
502 mut result,
503 human_input,
504 approvals,
505 tool_outputs,
506 } = options;
507 let pause = result
508 .run
509 .pause
510 .clone()
511 .ok_or_else(|| AgentError::new("Agent run is not waiting for human input"))?;
512
513 if pause.kind == "human_review" {
514 let input = human_input
515 .filter(|value| !value.trim().is_empty())
516 .ok_or_else(|| AgentError::new("Human input is required to resume this run"))?;
517 result.messages.push(Message::user(input));
518 } else {
519 self.resolve_pending_tools(&mut result, &pause, &approvals, &tool_outputs)?;
520 }
521
522 result.run.pause = None;
523 result.run.status = "running".to_string();
524 result.run.updated_at_ms = now_ms();
525 emit(&mut on_event, &result, "run.resumed", Value::Null);
526 self.drive(client, &mut result, &mut on_event)?;
527 Ok(result)
528 }
529
530 fn drive(
531 &self,
532 client: &mut dyn ModelClient,
533 result: &mut RunResult,
534 on_event: &mut Option<&mut dyn FnMut(&AgentEvent)>,
535 ) -> Result<(), AgentError> {
536 let max_steps = result.run.max_steps.max(1);
537
538 while result.run.step_count < max_steps {
539 let step_index = result.run.step_count;
540 emit(
541 on_event,
542 result,
543 "model.request.started",
544 json!({"step_index": step_index}),
545 );
546
547 let request = ModelRequest {
548 agent_id: self.definition.id.clone(),
549 model: result.run.model.clone(),
550 instructions: self.definition.instructions.clone(),
551 messages: result.messages.clone(),
552 tools: self.definition.tools.iter().map(Tool::spec).collect(),
553 context: result.run.context.clone(),
554 };
555
556 let (response, attempts) = self.generate_with_retry(client, &request)?;
557 result.run.step_count += 1;
558 result.run.updated_at_ms = now_ms();
559 result.usage.add(&response.usage);
560 result.messages.push(response.message.clone());
561 result.steps.push(RunStep {
562 index: step_index,
563 status: "completed".to_string(),
564 model_attempts: attempts,
565 tool_calls: response.message.tool_calls.clone(),
566 request_id: response.request_id.clone(),
567 provider: response.provider.clone(),
568 model: response.model.clone(),
569 finish_reason: response.finish_reason.clone(),
570 error: None,
571 usage: response.usage.clone(),
572 });
573 emit(
574 on_event,
575 result,
576 "model.response.completed",
577 json!({"step_index": step_index, "attempts": attempts}),
578 );
579
580 if !response.message.tool_calls.is_empty() {
581 let pending = self.pending_tools(&response.message.tool_calls)?;
582 let automatic: Vec<ToolCall> = response
583 .message
584 .tool_calls
585 .iter()
586 .filter(|call| !pending.iter().any(|item| item.call.id == call.id))
587 .cloned()
588 .collect();
589 self.execute_tools(result, &automatic, on_event)?;
590 if !pending.is_empty() {
591 let pause_kind = if pending.iter().all(|call| call.kind == "external_output") {
592 "external_output"
593 } else if pending.iter().all(|call| call.kind == "approval") {
594 "tool_approval"
595 } else {
596 "tool_input"
597 };
598 result.run.status = "waiting_for_human".to_string();
599 result.run.pause = Some(HumanPause {
600 reason: "Pending tool calls require input".to_string(),
601 payload: json!({"tool_calls": pending}),
602 kind: pause_kind.to_string(),
603 pending_tool_calls: pending,
604 });
605 emit(on_event, result, "run.paused", Value::Null);
606 return Ok(());
607 }
608 continue;
609 }
610
611 if let Some(reviewer) = &self.definition.human_review {
612 let review_context = HumanReviewContext {
613 run_id: result.run.id.clone(),
614 agent_id: result.run.agent_id.clone(),
615 step_index,
616 messages: result.messages.clone(),
617 response: response.clone(),
618 context: result.run.context.clone(),
619 };
620 if let Some(review) = reviewer(&review_context) {
621 result.run.status = "waiting_for_human".to_string();
622 result.run.pause = Some(HumanPause {
623 reason: review.reason,
624 payload: review.payload,
625 kind: "human_review".to_string(),
626 pending_tool_calls: Vec::new(),
627 });
628 emit(on_event, result, "run.paused", Value::Null);
629 return Ok(());
630 }
631 }
632
633 result.output = Value::String(response.message.content);
634 result.run.status = "completed".to_string();
635 emit(on_event, result, "run.completed", Value::Null);
636 return Ok(());
637 }
638
639 result.run.status = "stopped".to_string();
640 result.run.stop_reason = Some(format!("max_steps:{max_steps}"));
641 emit(
642 on_event,
643 result,
644 "run.stopped",
645 json!({"reason": result.run.stop_reason}),
646 );
647 Ok(())
648 }
649
650 fn generate_with_retry(
651 &self,
652 client: &mut dyn ModelClient,
653 request: &ModelRequest,
654 ) -> Result<(ModelResponse, usize), AgentError> {
655 let mut attempts = 0;
656 loop {
657 attempts += 1;
658 match client.generate(request) {
659 Ok(response) => return Ok((response, attempts)),
660 Err(error) if attempts <= self.definition.max_retries => {
661 if !self.definition.retry_backoff.is_zero() {
662 thread::sleep(self.definition.retry_backoff);
663 }
664 let _ = error;
665 }
666 Err(error) => return Err(error),
667 }
668 }
669 }
670
671 fn pending_tools(&self, calls: &[ToolCall]) -> Result<Vec<PendingToolCall>, AgentError> {
672 calls
673 .iter()
674 .filter_map(|call| {
675 let tool = self
676 .definition
677 .tools
678 .iter()
679 .find(|tool| tool.id == call.name);
680 match tool {
681 None => Some(Err(AgentError::new(format!(
682 "Model requested unknown tool: {}",
683 call.name
684 )))),
685 Some(tool) if tool.require_approval => Some(Ok(PendingToolCall {
686 call: call.clone(),
687 kind: "approval".to_string(),
688 reason: "Tool requires approval".to_string(),
689 })),
690 Some(tool) if tool.execute.is_none() => Some(Ok(PendingToolCall {
691 call: call.clone(),
692 kind: "external_output".to_string(),
693 reason: "Tool output must be supplied externally".to_string(),
694 })),
695 Some(_) => None,
696 }
697 })
698 .collect()
699 }
700
701 fn execute_tools(
702 &self,
703 result: &mut RunResult,
704 calls: &[ToolCall],
705 on_event: &mut Option<&mut dyn FnMut(&AgentEvent)>,
706 ) -> Result<(), AgentError> {
707 for call in calls {
708 let tool = self
709 .definition
710 .tools
711 .iter()
712 .find(|tool| tool.id == call.name)
713 .ok_or_else(|| AgentError::new(format!("Unknown tool: {}", call.name)))?;
714 let executor = tool.execute.as_ref().ok_or_else(|| {
715 AgentError::new(format!("Tool requires external output: {}", call.name))
716 })?;
717 let runtime = RuntimeContext {
718 run_id: result.run.id.clone(),
719 agent_id: result.run.agent_id.clone(),
720 step_index: result.run.step_count.saturating_sub(1),
721 context: result.run.context.clone(),
722 tool_call: call.clone(),
723 };
724 emit(
725 on_event,
726 result,
727 "tool.started",
728 json!({"tool_call_id": call.id, "name": call.name}),
729 );
730 let (output, is_error) = match executor(call.input.clone(), &runtime) {
731 Ok(output) => (output, false),
732 Err(error) => (json!({"error": error.to_string()}), true),
733 };
734 result.messages.push(Message {
735 role: "tool".to_string(),
736 content: value_to_text(&output),
737 tool_call_id: Some(call.id.clone()),
738 name: Some(call.name.clone()),
739 is_error,
740 ..Message::default()
741 });
742 emit(
743 on_event,
744 result,
745 "tool.completed",
746 json!({"tool_call_id": call.id, "name": call.name, "is_error": is_error}),
747 );
748 }
749 Ok(())
750 }
751
752 fn resolve_pending_tools(
753 &self,
754 result: &mut RunResult,
755 pause: &HumanPause,
756 approvals: &[ToolDecision],
757 tool_outputs: &[ToolOutput],
758 ) -> Result<(), AgentError> {
759 for pending in &pause.pending_tool_calls {
760 if let Some(output) = tool_outputs
761 .iter()
762 .find(|output| output.tool_call_id == pending.call.id)
763 {
764 result.messages.push(Message {
765 role: "tool".to_string(),
766 content: value_to_text(&output.output),
767 tool_call_id: Some(pending.call.id.clone()),
768 name: Some(pending.call.name.clone()),
769 ..Message::default()
770 });
771 continue;
772 }
773
774 let approved = approvals
775 .iter()
776 .any(|decision| decision.tool_call_id == pending.call.id);
777 if approved {
778 self.execute_tools(result, std::slice::from_ref(&pending.call), &mut None)?;
779 continue;
780 }
781
782 return Err(AgentError::new(format!(
783 "No approval or external output supplied for tool call {}",
784 pending.call.id
785 )));
786 }
787 Ok(())
788 }
789}
790
791pub struct GatewayAgentClient {
792 client: Phaseo,
793 model: String,
794}
795
796impl GatewayAgentClient {
797 pub fn new(client: Phaseo, model: impl Into<String>) -> Self {
798 Self {
799 client,
800 model: model.into(),
801 }
802 }
803
804 pub fn from_env(model: impl Into<String>) -> Result<Self, AgentError> {
805 Ok(Self::new(Phaseo::from_env()?, model))
806 }
807}
808
809pub fn create_gateway_agent_client(
810 model: impl Into<String>,
811) -> Result<GatewayAgentClient, AgentError> {
812 GatewayAgentClient::from_env(model)
813}
814
815impl ModelClient for GatewayAgentClient {
816 fn generate(&mut self, request: &ModelRequest) -> Result<ModelResponse, AgentError> {
817 let model = if request.model.trim().is_empty() {
818 &self.model
819 } else {
820 &request.model
821 };
822 let input = response_input(&request.messages);
823 let tools: Vec<Value> = request
824 .tools
825 .iter()
826 .map(|tool| {
827 json!({
828 "type": "function",
829 "name": tool.id,
830 "description": tool.description,
831 "parameters": tool.parameters,
832 })
833 })
834 .collect();
835 let mut body = json!({
836 "model": model,
837 "input": input,
838 "metadata": {"phaseo_agent_id": request.agent_id},
839 });
840 if !request.instructions.is_empty() {
841 body["instructions"] = Value::String(request.instructions.clone());
842 }
843 if !tools.is_empty() {
844 body["tools"] = Value::Array(tools);
845 }
846
847 let response = self.client.responses(&body)?;
848 model_response_from_body(response.body, response.request_id)
849 }
850}
851
852fn response_input(messages: &[Message]) -> Vec<Value> {
853 let mut input = Vec::new();
854 for message in messages {
855 if message.role == "tool" {
856 input.push(json!({
857 "type": "function_call_output",
858 "call_id": message.tool_call_id,
859 "output": message.content,
860 }));
861 continue;
862 }
863 if !message.content.is_empty() {
864 input.push(json!({
865 "type": "message",
866 "role": message.role,
867 "content": message.content,
868 }));
869 }
870 for call in &message.tool_calls {
871 input.push(json!({
872 "type": "function_call",
873 "call_id": call.id,
874 "name": call.name,
875 "arguments": call.input.to_string(),
876 }));
877 }
878 }
879 input
880}
881
882fn model_response_from_body(
883 body: Value,
884 header_request_id: Option<String>,
885) -> Result<ModelResponse, AgentError> {
886 let mut text = body
887 .get("output_text")
888 .and_then(Value::as_str)
889 .unwrap_or_default()
890 .to_string();
891 let mut tool_calls = Vec::new();
892 if let Some(output) = body.get("output").and_then(Value::as_array) {
893 for item in output {
894 match item.get("type").and_then(Value::as_str) {
895 Some("function_call") => {
896 let raw = item
897 .get("arguments")
898 .and_then(Value::as_str)
899 .unwrap_or("{}");
900 tool_calls.push(ToolCall {
901 id: item
902 .get("call_id")
903 .or_else(|| item.get("id"))
904 .and_then(Value::as_str)
905 .unwrap_or_default()
906 .to_string(),
907 name: item
908 .get("name")
909 .and_then(Value::as_str)
910 .unwrap_or_default()
911 .to_string(),
912 input: serde_json::from_str(raw)
913 .unwrap_or_else(|_| Value::String(raw.into())),
914 });
915 }
916 Some("message") if text.is_empty() => {
917 text = item
918 .get("content")
919 .and_then(Value::as_array)
920 .into_iter()
921 .flatten()
922 .filter_map(|part| part.get("text").and_then(Value::as_str))
923 .collect::<Vec<_>>()
924 .join("");
925 }
926 _ => {}
927 }
928 }
929 }
930
931 if text.is_empty() && tool_calls.is_empty() {
932 return Err(AgentError::new(
933 "Phaseo response contained neither output text nor tool calls",
934 ));
935 }
936
937 let usage = body.get("usage").cloned().unwrap_or(Value::Null);
938 let input_tokens = integer(&usage, &["input_tokens", "prompt_tokens"]);
939 let output_tokens = integer(&usage, &["output_tokens", "completion_tokens"]);
940 let total_tokens = integer(&usage, &["total_tokens"]).max(input_tokens + output_tokens);
941 Ok(ModelResponse {
942 message: Message {
943 role: "assistant".to_string(),
944 content: text,
945 tool_calls,
946 ..Message::default()
947 },
948 usage: UsageSummary {
949 input_tokens,
950 output_tokens,
951 cached_tokens: usage
952 .pointer("/input_tokens_details/cached_tokens")
953 .and_then(Value::as_u64)
954 .unwrap_or_else(|| integer(&usage, &["cached_tokens", "cache_read_input_tokens"])),
955 total_tokens,
956 cost: body
957 .pointer("/meta/cost")
958 .or_else(|| body.get("cost"))
959 .and_then(Value::as_f64)
960 .unwrap_or_default(),
961 },
962 request_id: header_request_id
963 .or_else(|| body.get("id").and_then(Value::as_str).map(str::to_string)),
964 provider: body
965 .get("provider")
966 .and_then(Value::as_str)
967 .map(str::to_string),
968 model: body
969 .get("model")
970 .and_then(Value::as_str)
971 .map(str::to_string),
972 finish_reason: body
973 .get("status")
974 .and_then(Value::as_str)
975 .map(str::to_string),
976 })
977}
978
979fn integer(value: &Value, keys: &[&str]) -> u64 {
980 keys.iter()
981 .find_map(|key| value.get(key).and_then(Value::as_u64))
982 .unwrap_or_default()
983}
984
985fn value_to_text(value: &Value) -> String {
986 value
987 .as_str()
988 .map(str::to_string)
989 .unwrap_or_else(|| value.to_string())
990}
991
992fn now_ms() -> u64 {
993 SystemTime::now()
994 .duration_since(UNIX_EPOCH)
995 .unwrap_or_default()
996 .as_millis()
997 .try_into()
998 .unwrap_or(u64::MAX)
999}
1000
1001fn new_run_id() -> String {
1002 let sequence = RUN_SEQUENCE.fetch_add(1, Ordering::Relaxed);
1003 format!("run_{}_{sequence}", now_ms())
1004}
1005
1006fn emit(
1007 handler: &mut Option<&mut dyn FnMut(&AgentEvent)>,
1008 result: &RunResult,
1009 event_type: &str,
1010 details: Value,
1011) {
1012 if let Some(handler) = handler.as_deref_mut() {
1013 handler(&AgentEvent {
1014 event_type: event_type.to_string(),
1015 run_id: result.run.id.clone(),
1016 agent_id: result.run.agent_id.clone(),
1017 timestamp_ms: now_ms(),
1018 details,
1019 });
1020 }
1021}
1022
1023#[cfg(test)]
1024mod tests {
1025 use super::*;
1026
1027 struct ToolLoopClient {
1028 calls: usize,
1029 }
1030
1031 impl ModelClient for ToolLoopClient {
1032 fn generate(&mut self, _request: &ModelRequest) -> Result<ModelResponse, AgentError> {
1033 self.calls += 1;
1034 if self.calls == 1 {
1035 return Ok(ModelResponse {
1036 message: Message {
1037 role: "assistant".to_string(),
1038 tool_calls: vec![ToolCall {
1039 id: "call_1".to_string(),
1040 name: "lookup".to_string(),
1041 input: json!({"slug": "presets"}),
1042 }],
1043 ..Message::default()
1044 },
1045 ..ModelResponse::default()
1046 });
1047 }
1048 Ok(ModelResponse {
1049 message: Message::assistant("Presets define stable routing defaults."),
1050 ..ModelResponse::default()
1051 })
1052 }
1053 }
1054
1055 #[test]
1056 fn executes_a_tool_loop_and_emits_events() {
1057 let tool = Tool::new(
1058 "lookup",
1059 "Lookup documentation",
1060 json!({"type": "object"}),
1061 |input, _context| Ok(json!({"slug": input["slug"], "ok": true})),
1062 );
1063 let agent = create_agent(
1064 AgentDefinition::new("support", "openai/gpt-5.4-nano")
1065 .instructions("Use tools when helpful")
1066 .tool(tool),
1067 );
1068 let mut client = ToolLoopClient { calls: 0 };
1069 let mut events = Vec::new();
1070 let mut capture = |event: &AgentEvent| events.push(event.event_type.clone());
1071 let result = agent
1072 .run_with_events(
1073 &mut client,
1074 RunOptions::new("Explain presets"),
1075 Some(&mut capture),
1076 )
1077 .unwrap();
1078
1079 assert_eq!(result.run.status, "completed");
1080 assert_eq!(
1081 result.output,
1082 json!("Presets define stable routing defaults.")
1083 );
1084 assert_eq!(result.steps.len(), 2);
1085 assert!(result.messages.iter().any(|message| message.role == "tool"));
1086 assert!(events.iter().any(|event| event == "tool.completed"));
1087 }
1088
1089 struct RetryReviewClient {
1090 calls: usize,
1091 models: Vec<String>,
1092 }
1093
1094 impl ModelClient for RetryReviewClient {
1095 fn generate(&mut self, request: &ModelRequest) -> Result<ModelResponse, AgentError> {
1096 self.calls += 1;
1097 self.models.push(request.model.clone());
1098 if self.calls == 1 {
1099 return Err(AgentError::new("temporary failure"));
1100 }
1101 Ok(ModelResponse {
1102 message: Message::assistant("Deploy the change"),
1103 ..ModelResponse::default()
1104 })
1105 }
1106 }
1107
1108 #[test]
1109 fn retries_pauses_and_resumes_human_review() {
1110 let agent = create_agent(
1111 AgentDefinition::new("review", "openai/gpt-5.4-nano")
1112 .model_retries(1, Duration::ZERO)
1113 .human_review(|context| {
1114 if context
1115 .messages
1116 .iter()
1117 .any(|message| message.role == "user" && message.content == "approved")
1118 {
1119 None
1120 } else {
1121 Some(HumanReviewRequest {
1122 reason: "Approve deployment".to_string(),
1123 payload: json!({"output": context.response.message.content}),
1124 })
1125 }
1126 }),
1127 );
1128 let mut client = RetryReviewClient {
1129 calls: 0,
1130 models: Vec::new(),
1131 };
1132 let mut run_options = RunOptions::new("Prepare deployment");
1133 run_options.model = Some("openai/override-model".to_string());
1134 run_options.max_steps = Some(3);
1135 let paused = agent.run(&mut client, run_options).unwrap();
1136 assert_eq!(paused.run.status, "waiting_for_human");
1137 assert_eq!(paused.steps[0].model_attempts, 2);
1138 assert_eq!(paused.run.model, "openai/override-model");
1139 assert_eq!(paused.run.max_steps, 3);
1140
1141 let mut options = ContinueOptions::new(paused);
1142 options.human_input = Some("approved".to_string());
1143 let completed = agent.continue_run(&mut client, options).unwrap();
1144 assert_eq!(completed.run.status, "completed");
1145 assert_eq!(completed.output, json!("Deploy the change"));
1146 assert!(client
1147 .models
1148 .iter()
1149 .all(|model| model == "openai/override-model"));
1150 serde_json::to_string(&completed).unwrap();
1151 }
1152
1153 struct MixedToolClient {
1154 calls: usize,
1155 }
1156
1157 impl ModelClient for MixedToolClient {
1158 fn generate(&mut self, _request: &ModelRequest) -> Result<ModelResponse, AgentError> {
1159 self.calls += 1;
1160 if self.calls == 1 {
1161 return Ok(ModelResponse {
1162 message: Message {
1163 role: "assistant".to_string(),
1164 tool_calls: vec![
1165 ToolCall {
1166 id: "call_local".to_string(),
1167 name: "local".to_string(),
1168 input: json!({"value": 1}),
1169 },
1170 ToolCall {
1171 id: "call_external".to_string(),
1172 name: "external".to_string(),
1173 input: json!({"value": 2}),
1174 },
1175 ],
1176 ..Message::default()
1177 },
1178 ..ModelResponse::default()
1179 });
1180 }
1181 Ok(ModelResponse {
1182 message: Message::assistant("Both tool results received."),
1183 ..ModelResponse::default()
1184 })
1185 }
1186 }
1187
1188 #[test]
1189 fn executes_automatic_tools_before_pausing_for_external_outputs() {
1190 let agent = create_agent(
1191 AgentDefinition::new("mixed-tools", "openai/gpt-5.4-nano")
1192 .tool(Tool::new(
1193 "local",
1194 "Run locally",
1195 json!({"type": "object"}),
1196 |input, _context| Ok(json!({"local": input["value"]})),
1197 ))
1198 .tool(Tool::external(
1199 "external",
1200 "Run externally",
1201 json!({"type": "object"}),
1202 )),
1203 );
1204 let mut client = MixedToolClient { calls: 0 };
1205 let paused = agent
1206 .run(&mut client, RunOptions::new("Use both tools"))
1207 .unwrap();
1208
1209 let pause = paused.run.pause.as_ref().unwrap();
1210 assert_eq!(pause.kind, "external_output");
1211 assert_eq!(pause.pending_tool_calls.len(), 1);
1212 assert_eq!(pause.pending_tool_calls[0].call.id, "call_external");
1213 assert!(paused.messages.iter().any(|message| {
1214 message.tool_call_id.as_deref() == Some("call_local")
1215 && message.content.contains("local")
1216 }));
1217
1218 let mut options = ContinueOptions::new(paused);
1219 options.tool_outputs.push(ToolOutput {
1220 tool_call_id: "call_external".to_string(),
1221 output: json!({"external": 2}),
1222 });
1223 let completed = agent.continue_run(&mut client, options).unwrap();
1224 assert_eq!(completed.run.status, "completed");
1225 assert_eq!(completed.output, json!("Both tool results received."));
1226 }
1227
1228 #[test]
1229 fn parses_gateway_tool_calls() {
1230 let response = model_response_from_body(
1231 json!({
1232 "id": "resp_1",
1233 "model": "test/model",
1234 "status": "completed",
1235 "output": [{
1236 "type": "function_call",
1237 "call_id": "call_1",
1238 "name": "lookup",
1239 "arguments": "{\"slug\":\"rust\"}"
1240 }],
1241 "usage": {"input_tokens": 3, "output_tokens": 2}
1242 }),
1243 None,
1244 )
1245 .unwrap();
1246 assert_eq!(response.message.tool_calls[0].input["slug"], "rust");
1247 assert_eq!(response.usage.total_tokens, 5);
1248 }
1249
1250 #[test]
1251 fn reads_nested_responses_cached_token_usage() {
1252 let response = model_response_from_body(
1253 json!({
1254 "id": "resp_cached",
1255 "output_text": "cached",
1256 "usage": {
1257 "input_tokens": 10,
1258 "output_tokens": 2,
1259 "input_tokens_details": {"cached_tokens": 7}
1260 }
1261 }),
1262 None,
1263 )
1264 .unwrap();
1265 assert_eq!(response.usage.cached_tokens, 7);
1266 }
1267}