1use std::collections::{BTreeMap, HashMap};
2use std::sync::{Arc, Mutex};
3use std::time::Duration;
4
5use serde_json::{json, Value};
6
7use crate::app_server::durable_resume::{
8 validate_completion, DurableTurnCompletionFuture, DurableTurnResumeOutcome,
9 DurableTurnResumeProvider, DurableTurnResumeRequest,
10};
11use crate::app_server::host::{
12 AgentResolutionRequest, AppServerHost, DefaultAppServerHost, RunConfigResolutionRequest,
13};
14use crate::app_server::outgoing::OutgoingMessageSender;
15use crate::app_server::protocol::{
16 map_run_event_to_notifications, AgentMessageDeltaParams, AppCacheUsage, AppItem,
17 AppServerError, AppServerErrorCode, AppThread, AppTokenUsage, AppTurn, ApprovalDecision,
18 ApprovalRequestParams, ApprovalResolveParams, CheckpointSummary, CheckpointSummaryStatus,
19 InterruptionIdempotencySupport, InterruptionOperationKind, InterruptionSummary,
20 ItemCompletedParams, ItemStartedParams, JsonRpcError, JsonRpcErrorBody, RequestId,
21 ServerNotification, ServerRequest, ThreadStatus, ThreadStatusChangedParams,
22 TurnCompletedParams, TurnResumeParams, TurnResumeResponse, TurnStartParams, TurnStartedParams,
23 TurnStatus, UserInput,
24};
25use crate::app_server::thread_state::{ActiveTurn, SteeringQueue, ThreadStateManager};
26use crate::app_server::thread_store::{ItemAppendOutcome, SqliteThreadStore, ThreadStoreError};
27use crate::app_server::transport::ConnectionId;
28use crate::checkpoint::{
29 CheckpointStatus, OperationKind, ResumeObservation, ResumePolicy, ToolIdempotency,
30 MAX_CHECKPOINT_KEY_BYTES,
31};
32use crate::events::RunEventPayload;
33use crate::runner::CheckpointStartOutcome;
34use crate::runtime::state_v2::{CheckpointStoreV2, CheckpointV2};
35use crate::tools::ApprovalDecision as ToolApprovalDecision;
36use crate::types::{AgentStatus, CacheUsageStatus, Metadata, TaskTokenUsage, ToolExecutionResult};
37use crate::{
38 Agent, ApprovalBroker, ApprovalFuture, ApprovalProvider, ApprovalRequest, BeforeLlmEvent,
39 BeforeLlmPatch, BeforeToolCallEvent, BeforeToolCallPatch, Message, RunConfig, RunHandle,
40 RunResult, Runner, RuntimeHook,
41};
42
43mod approval;
44mod resume;
45
46use approval::tool_approval_decision_from_response;
47use resume::{checkpoint_projection, turn_completion_result};
48
49#[derive(Clone)]
50pub struct AppServerRunAdapter {
51 runner: Runner,
52 host: Arc<dyn AppServerHost>,
53 store: SqliteThreadStore,
54 state: ThreadStateManager,
55 outgoing: OutgoingMessageSender,
56 approval_request_timeout: Duration,
57 turn_approval_timeouts: Arc<Mutex<HashMap<(String, String), Duration>>>,
58 durable_resume_provider: Option<Arc<dyn DurableTurnResumeProvider>>,
59}
60pub(crate) struct PreparedTurnResume {
61 response: TurnResumeResponse,
62 continuation: Option<PreparedTurnResumeContinuation>,
63}
64
65struct PreparedTurnResumeContinuation {
66 request: DurableTurnResumeRequest,
67 completion: DurableTurnCompletionFuture,
68}
69
70impl PreparedTurnResume {
71 pub(crate) fn response(&self) -> &TurnResumeResponse {
72 &self.response
73 }
74}
75
76impl AppServerRunAdapter {
77 pub fn new(
78 runner: Runner,
79 agent: Agent,
80 store: SqliteThreadStore,
81 state: ThreadStateManager,
82 outgoing: OutgoingMessageSender,
83 ) -> Self {
84 Self::with_host(
85 runner,
86 Arc::new(DefaultAppServerHost::from_agent(agent)),
87 store,
88 state,
89 outgoing,
90 )
91 }
92
93 pub fn with_host(
94 runner: Runner,
95 host: Arc<dyn AppServerHost>,
96 store: SqliteThreadStore,
97 state: ThreadStateManager,
98 outgoing: OutgoingMessageSender,
99 ) -> Self {
100 Self {
101 runner,
102 host,
103 store,
104 state,
105 outgoing,
106 approval_request_timeout: Duration::from_secs(30),
107 turn_approval_timeouts: Arc::new(Mutex::new(HashMap::new())),
108 durable_resume_provider: None,
109 }
110 }
111
112 pub fn with_approval_request_timeout(mut self, timeout: Duration) -> Self {
113 self.approval_request_timeout = timeout;
114 self
115 }
116
117 pub fn with_durable_resume_provider(
118 mut self,
119 provider: Arc<dyn DurableTurnResumeProvider>,
120 ) -> Self {
121 self.durable_resume_provider = Some(provider);
122 self
123 }
124
125 pub fn store(&self) -> &SqliteThreadStore {
126 &self.store
127 }
128
129 pub fn state(&self) -> &ThreadStateManager {
130 &self.state
131 }
132
133 pub async fn start_turn(
134 &self,
135 owner_connection_id: ConnectionId,
136 params: TurnStartParams,
137 ) -> Result<AppTurn, AppServerError> {
138 let thread = self
139 .store
140 .get_thread(¶ms.thread_id)
141 .map_err(store_error)?
142 .ok_or_else(AppServerError::thread_not_found)?;
143 if thread.archived_at.is_some() {
144 return Err(AppServerError::thread_archived());
145 }
146 if thread.status == ThreadStatus::Running
147 || self.state.has_active_turn(¶ms.thread_id).await
148 {
149 return Err(AppServerError::invalid_params(
150 "Thread already has an active turn",
151 ));
152 }
153
154 let mut effective_metadata = thread.metadata.clone();
155 effective_metadata.extend(params.metadata.clone());
156 let agent_request = AgentResolutionRequest {
157 thread_id: thread.thread_id.clone(),
158 agent_key: thread.agent_key.clone(),
159 cwd: thread.cwd.clone(),
160 metadata: effective_metadata.clone(),
161 };
162 let config_request = RunConfigResolutionRequest {
163 thread_id: thread.thread_id.clone(),
164 agent_key: thread.agent_key.clone(),
165 cwd: thread.cwd.clone(),
166 metadata: effective_metadata.clone(),
167 };
168 let agent = self
169 .host
170 .resolve_agent(&agent_request)
171 .map_err(|error| AppServerError::internal(error.to_string()))?;
172 let mut config = self
173 .host
174 .build_run_config(&config_request)
175 .map_err(|error| AppServerError::internal(error.to_string()))?;
176
177 let turn = self
178 .store
179 .create_turn(¶ms.thread_id, params.input.clone())
180 .map_err(store_error)?;
181 let steering = SteeringQueue::default();
182 if config.approval_provider.is_none() {
183 config.approval_provider = Some(Arc::new(AppServerApprovalProvider));
184 }
185 if config.approval_broker.is_none() {
186 config.approval_broker = Some(ApprovalBroker::default());
187 }
188 let approval_request_timeout =
189 effective_approval_request_timeout(&config, self.approval_request_timeout);
190 config.hooks.push(Arc::new(SteeringRuntimeHook {
191 queue: steering.clone(),
192 }));
193 config.metadata.extend(effective_metadata);
194 config
195 .metadata
196 .insert("thread_id".to_string(), json!(turn.thread_id));
197 config
198 .metadata
199 .insert("turn_id".to_string(), json!(turn.turn_id));
200 config
201 .metadata
202 .insert("session_id".to_string(), json!(turn.thread_id));
203 let checkpoint_store = config
204 .checkpoint_config
205 .as_ref()
206 .and_then(|checkpoint| checkpoint.store.clone());
207
208 let handle = self
209 .runner
210 .start(&agent, input_text(&turn.input), config)
211 .await
212 .map_err(AppServerError::internal)?;
213 self.store
214 .set_active_turn(&turn.thread_id, Some(&turn.turn_id), ThreadStatus::Running)
215 .map_err(store_error)?;
216 self.turn_approval_timeouts
217 .lock()
218 .unwrap_or_else(std::sync::PoisonError::into_inner)
219 .insert(
220 (turn.thread_id.clone(), turn.turn_id.clone()),
221 approval_request_timeout,
222 );
223 self.state
224 .set_active_turn(
225 turn.thread_id.clone(),
226 ActiveTurn {
227 turn: turn.clone(),
228 handle,
229 steering,
230 owner_connection_id,
231 checkpoint_store,
232 },
233 )
234 .await;
235 Ok(turn)
236 }
237
238 pub async fn notify_turn_started(&self, turn: &AppTurn) -> Result<(), AppServerError> {
239 self.broadcast_to_thread(
240 &turn.thread_id,
241 ServerNotification::ThreadStatusChanged(ThreadStatusChangedParams {
242 thread_id: turn.thread_id.clone(),
243 status: ThreadStatus::Running,
244 }),
245 )
246 .await?;
247 self.broadcast_to_thread(
248 &turn.thread_id,
249 ServerNotification::TurnStarted(TurnStartedParams {
250 thread_id: turn.thread_id.clone(),
251 turn_id: turn.turn_id.clone(),
252 run_id: None,
253 status: None,
254 }),
255 )
256 .await
257 }
258
259 pub async fn spawn_event_forwarding(&self, thread_id: String, turn_id: String) {
260 let Some(active) = self.state.active_turn(&thread_id).await else {
261 return;
262 };
263 self.spawn_event_forwarding_for(active, thread_id, turn_id);
264 }
265
266 fn spawn_event_forwarding_for(&self, active: ActiveTurn, thread_id: String, turn_id: String) {
267 let adapter = self.clone();
268 tokio::spawn(async move {
269 let mut events = active.handle.events();
270 let mut tool_arguments = HashMap::<String, Value>::new();
271 while let Some(event) = events.next().await {
272 match event {
273 Ok(event) => {
274 if let RunEventPayload::ToolCallStarted {
275 tool_call_id,
276 arguments,
277 ..
278 } = event.payload()
279 {
280 tool_arguments.insert(tool_call_id.clone(), arguments.clone());
281 }
282 let mut notifications =
283 map_run_event_to_notifications(&thread_id, &turn_id, &event);
284 for notification in &mut notifications {
285 if let ServerNotification::ApprovalRequested(approval) = notification {
286 if let Some(arguments) = tool_arguments.get(&approval.tool_call_id)
287 {
288 approval.arguments = arguments.clone();
289 }
290 }
291 }
292 if let Some(item) = notifications.iter().find_map(item_from_notification) {
293 match adapter.store.append_item(&thread_id, &turn_id, item) {
294 Ok(ItemAppendOutcome::Inserted) => {}
295 Ok(ItemAppendOutcome::AlreadyPresent) => continue,
296 Err(error) => {
297 let _ = adapter
298 .broadcast_to_thread(
299 &thread_id,
300 ServerNotification::ErrorWarning(
301 crate::app_server::protocol::WarningParams {
302 message: error.to_string(),
303 code: Some(
304 "item_identity_conflict".to_string(),
305 ),
306 },
307 ),
308 )
309 .await;
310 continue;
311 }
312 }
313 }
314 for notification in notifications {
315 if matches!(notification, ServerNotification::ApprovalResolved(_)) {
319 continue;
320 }
321 let _ = adapter
322 .broadcast_to_thread(&thread_id, notification.clone())
323 .await;
324 if let ServerNotification::ApprovalRequested(approval) = notification {
325 adapter
326 .route_approval_request(
327 &thread_id,
328 &turn_id,
329 active.owner_connection_id,
330 &active.handle,
331 approval,
332 )
333 .await;
334 }
335 }
336 }
337 Err(error) => {
338 let _ = adapter
339 .broadcast_to_thread(
340 &thread_id,
341 ServerNotification::ErrorWarning(
342 crate::app_server::protocol::WarningParams {
343 message: error,
344 code: Some("event_stream".to_string()),
345 },
346 ),
347 )
348 .await;
349 }
350 }
351 }
352 let result = events.into_result().await;
353 adapter
354 .complete_turn(
355 thread_id,
356 turn_id,
357 active.owner_connection_id,
358 active.checkpoint_store,
359 result,
360 )
361 .await;
362 });
363 }
364
365 pub async fn queue_steering(
366 &self,
367 thread_id: &str,
368 expected_turn_id: &str,
369 input: Vec<UserInput>,
370 ) -> Result<String, AppServerError> {
371 let active = self
372 .validated_active_turn(thread_id, expected_turn_id)
373 .await?;
374 active
375 .steering
376 .lock()
377 .map_err(|_| AppServerError::internal("steering queue lock poisoned"))?
378 .push_back(input);
379 Ok(active.turn.turn_id)
380 }
381
382 pub async fn queue_follow_up(
383 &self,
384 thread_id: &str,
385 expected_turn_id: &str,
386 input: Vec<UserInput>,
387 ) -> Result<String, AppServerError> {
388 let active = self
389 .validated_active_turn(thread_id, expected_turn_id)
390 .await?;
391 self.state.queue_follow_up(thread_id, input).await;
392 Ok(active.turn.turn_id)
393 }
394
395 pub async fn interrupt_turn(
396 &self,
397 thread_id: &str,
398 expected_turn_id: &str,
399 ) -> Result<InterruptTurnOutcome, AppServerError> {
400 let active = self
401 .validated_active_turn(thread_id, expected_turn_id)
402 .await?;
403 let pending = self.state.pending_approval(thread_id).await;
404 active.handle.cancel();
405 let mut approval_resolved = None;
406 if let Some(pending) = pending.filter(|pending| pending.turn_id == active.turn.turn_id) {
407 let _ = active
408 .handle
409 .approve(
410 &pending.request_id,
411 ToolApprovalDecision::timeout("turn interrupted"),
412 )
413 .await;
414 let _ = self
415 .outgoing
416 .resolve_server_error(
417 pending.connection_id,
418 JsonRpcError {
419 id: RequestId::String(pending.request_id.clone()),
420 error: JsonRpcErrorBody {
421 code: AppServerErrorCode::InternalError.code(),
422 message: "turn interrupted".to_string(),
423 data: None,
424 },
425 },
426 )
427 .await;
428 self.state
429 .clear_pending_approval(thread_id, &pending.request_id)
430 .await;
431 approval_resolved = Some(ApprovalResolveParams {
432 thread_id: thread_id.to_string(),
433 turn_id: active.turn.turn_id.clone(),
434 request_id: pending.request_id,
435 decision: ApprovalDecision::Timeout,
436 reason: "turn interrupted".to_string(),
437 metadata: Metadata::new(),
438 });
439 }
440 Ok(InterruptTurnOutcome {
441 approval_resolved,
442 turn_id: active.turn.turn_id,
443 cancelled: true,
444 })
445 }
446
447 pub async fn notify_approval_resolved(
448 &self,
449 params: ApprovalResolveParams,
450 ) -> Result<(), AppServerError> {
451 let thread_id = params.thread_id.clone();
452 self.broadcast_to_thread(&thread_id, ServerNotification::ApprovalResolved(params))
453 .await
454 }
455
456 pub async fn active_turn(&self, thread_id: &str) -> Option<AppTurn> {
457 self.state
458 .active_turn(thread_id)
459 .await
460 .map(|active| active.turn)
461 }
462
463 async fn validated_active_turn(
464 &self,
465 thread_id: &str,
466 expected_turn_id: &str,
467 ) -> Result<ActiveTurn, AppServerError> {
468 let active = self
469 .state
470 .active_turn(thread_id)
471 .await
472 .ok_or_else(AppServerError::active_turn_not_found)?;
473 if !expected_turn_id.is_empty() && active.turn.turn_id != expected_turn_id {
474 return Err(AppServerError::turn_id_mismatch());
475 }
476 Ok(active)
477 }
478
479 fn persist_turn_completion(
480 &self,
481 completion: &TurnCompletedParams,
482 ) -> Result<AppTurn, AppServerError> {
483 self.store
484 .update_turn(
485 &completion.turn_id,
486 completion.status,
487 completion.run_id.as_deref(),
488 &turn_completion_result(completion),
489 )
490 .map_err(store_error)
491 }
492
493 async fn complete_turn(
494 &self,
495 thread_id: String,
496 turn_id: String,
497 owner_connection_id: ConnectionId,
498 checkpoint_store: Option<Arc<dyn CheckpointStoreV2>>,
499 result: Result<RunResult, String>,
500 ) {
501 let (
502 status,
503 run_id,
504 final_output,
505 completion_reason,
506 completion_tool_name,
507 partial_output,
508 error,
509 token_usage,
510 budget_usage,
511 budget_exhaustion,
512 checkpoint,
513 interruption,
514 ) = match result {
515 Ok(result) => {
516 let status = turn_status(result.status());
517 let error = if status == TurnStatus::Failed {
518 result
519 .result()
520 .error
521 .clone()
522 .or_else(|| result.result().wait_reason.clone())
523 .or_else(|| Some("Turn failed".to_string()))
524 } else {
525 None
526 };
527 let (checkpoint, interruption) =
528 checkpoint_projection(&result, checkpoint_store.as_ref()).unwrap_or_else(
529 |error| {
530 eprintln!(
531 "warning: App Server checkpoint projection failed for turn {}: {error}",
532 turn_id
533 );
534 (None, None)
535 },
536 );
537 (
538 status,
539 Some(result.run_id().to_string()),
540 result.final_output().map(str::to_string),
541 result
542 .completion_reason()
543 .map(|reason| reason.as_str().to_string()),
544 result.completion_tool_name().map(str::to_string),
545 result.partial_output().map(str::to_string),
546 error,
547 Some(app_token_usage(&result.result().token_usage)),
548 result.budget_usage().map(app_json_object),
549 result.budget_exhaustion().map(app_json_object),
550 checkpoint,
551 interruption,
552 )
553 }
554 Err(error) => (
555 TurnStatus::Failed,
556 None,
557 None,
558 Some("failed".to_string()),
559 None,
560 None,
561 Some(error),
562 None,
563 None,
564 None,
565 None,
566 None,
567 ),
568 };
569 let mut stored_result = std::collections::BTreeMap::new();
570 if let Some(final_output) = &final_output {
571 stored_result.insert("finalOutput".to_string(), json!(final_output));
572 }
573 if let Some(completion_reason) = &completion_reason {
574 stored_result.insert("completionReason".to_string(), json!(completion_reason));
575 }
576 if let Some(completion_tool_name) = &completion_tool_name {
577 stored_result.insert(
578 "completionToolName".to_string(),
579 json!(completion_tool_name),
580 );
581 }
582 if let Some(partial_output) = &partial_output {
583 stored_result.insert("partialOutput".to_string(), json!(partial_output));
584 }
585 if let Some(error) = &error {
586 stored_result.insert("error".to_string(), json!(error));
587 }
588 if let Some(token_usage) = &token_usage {
589 stored_result.insert("tokenUsage".to_string(), json!(token_usage));
590 }
591 if let Some(budget_usage) = &budget_usage {
592 stored_result.insert("budgetUsage".to_string(), json!(budget_usage));
593 }
594 if let Some(budget_exhaustion) = &budget_exhaustion {
595 stored_result.insert("budgetExhaustion".to_string(), json!(budget_exhaustion));
596 }
597 if let Some(checkpoint) = &checkpoint {
598 stored_result.insert("checkpoint".to_string(), json!(checkpoint));
599 }
600 if let Some(interruption) = &interruption {
601 stored_result.insert("interruption".to_string(), json!(interruption));
602 }
603 let _ = self
604 .store
605 .update_turn(&turn_id, status, run_id.as_deref(), &stored_result);
606 self.turn_approval_timeouts
607 .lock()
608 .unwrap_or_else(std::sync::PoisonError::into_inner)
609 .remove(&(thread_id.clone(), turn_id.clone()));
610 self.state.clear_active_turn(&thread_id, &turn_id).await;
611 let _ = self
612 .store
613 .set_active_turn(&thread_id, None, ThreadStatus::Idle);
614 let _ = self
615 .broadcast_to_thread(
616 &thread_id,
617 ServerNotification::ThreadStatusChanged(ThreadStatusChangedParams {
618 thread_id: thread_id.clone(),
619 status: ThreadStatus::Idle,
620 }),
621 )
622 .await;
623 let _ = self
624 .broadcast_to_thread(
625 &thread_id,
626 ServerNotification::TurnCompleted(TurnCompletedParams {
627 thread_id: thread_id.clone(),
628 turn_id: turn_id.clone(),
629 run_id,
630 status,
631 final_output,
632 completion_reason,
633 completion_tool_name,
634 partial_output,
635 error,
636 token_usage,
637 budget_usage,
638 budget_exhaustion,
639 checkpoint,
640 interruption,
641 }),
642 )
643 .await;
644
645 if status == TurnStatus::Completed {
646 if let Some(input) = self.state.pop_follow_up(&thread_id).await {
647 let params = TurnStartParams {
648 thread_id: thread_id.clone(),
649 input,
650 metadata: Default::default(),
651 };
652 match self.start_turn(owner_connection_id, params).await {
653 Ok(turn) => {
654 let _ = self.notify_turn_started(&turn).await;
655 if let Some(active) = self.state.active_turn(&thread_id).await {
656 self.spawn_event_forwarding_for(active, thread_id, turn.turn_id);
657 }
658 }
659 Err(error) => {
660 let _ = self
661 .broadcast_to_thread(
662 &thread_id,
663 ServerNotification::ErrorWarning(
664 crate::app_server::protocol::WarningParams {
665 message: error.message().to_string(),
666 code: Some("follow_up".to_string()),
667 },
668 ),
669 )
670 .await;
671 }
672 }
673 }
674 }
675 }
676
677 async fn broadcast_to_thread(
678 &self,
679 thread_id: &str,
680 notification: ServerNotification,
681 ) -> Result<(), AppServerError> {
682 let subscribers = self.state.subscribers(thread_id).await;
683 for connection_id in subscribers {
684 self.outgoing
685 .send_notification(connection_id, notification.clone())
686 .await?;
687 }
688 Ok(())
689 }
690
691 async fn route_approval_request(
692 &self,
693 thread_id: &str,
694 turn_id: &str,
695 owner_connection_id: ConnectionId,
696 handle: &RunHandle,
697 approval: ApprovalRequestParams,
698 ) {
699 if !self
700 .state
701 .is_subscribed(thread_id, owner_connection_id)
702 .await
703 {
704 let reason = "approval client disconnected";
705 let _ = handle
706 .approve(&approval.request_id, ToolApprovalDecision::timeout(reason))
707 .await;
708 let _ = self
709 .broadcast_to_thread(
710 thread_id,
711 ServerNotification::ApprovalResolved(ApprovalResolveParams {
712 thread_id: thread_id.to_string(),
713 turn_id: turn_id.to_string(),
714 request_id: approval.request_id,
715 decision: ApprovalDecision::Timeout,
716 reason: reason.to_string(),
717 metadata: Metadata::new(),
718 }),
719 )
720 .await;
721 return;
722 }
723 self.state
724 .set_pending_approval(
725 thread_id,
726 turn_id,
727 approval.request_id.clone(),
728 owner_connection_id,
729 )
730 .await;
731 let timeout = self
732 .turn_approval_timeouts
733 .lock()
734 .unwrap_or_else(std::sync::PoisonError::into_inner)
735 .get(&(thread_id.to_string(), turn_id.to_string()))
736 .copied()
737 .unwrap_or(self.approval_request_timeout);
738 let response = self
739 .outgoing
740 .send_server_request_with_id_and_timeout(
741 owner_connection_id,
742 RequestId::String(approval.request_id.clone()),
743 ServerRequest::ApprovalRequest(approval.clone()),
744 timeout,
745 )
746 .await;
747 let (decision, protocol_decision) = match response {
748 Ok(value) => tool_approval_decision_from_response(value),
749 Err(error) => (
750 ToolApprovalDecision::timeout(error.message().to_string()),
751 ApprovalDecision::Timeout,
752 ),
753 };
754 let resolution_reason = decision.reason().to_string();
755 let resolution_metadata = decision.metadata().cloned().unwrap_or_default();
756 if self
757 .state
758 .pending_approval(thread_id)
759 .await
760 .is_none_or(|pending| pending.request_id != approval.request_id)
761 {
762 return;
763 }
764 self.state
765 .clear_pending_approval(thread_id, &approval.request_id)
766 .await;
767 let _ = handle.approve(&approval.request_id, decision).await;
768 let _ = self
769 .broadcast_to_thread(
770 thread_id,
771 ServerNotification::ApprovalResolved(ApprovalResolveParams {
772 thread_id: thread_id.to_string(),
773 turn_id: turn_id.to_string(),
774 request_id: approval.request_id,
775 decision: protocol_decision,
776 reason: resolution_reason,
777 metadata: resolution_metadata,
778 }),
779 )
780 .await;
781 }
782}
783
784pub struct InterruptTurnOutcome {
785 pub approval_resolved: Option<ApprovalResolveParams>,
786 pub turn_id: String,
787 pub cancelled: bool,
788}
789
790struct AppServerApprovalProvider;
791
792impl ApprovalProvider for AppServerApprovalProvider {
793 fn should_request(&self, _request: &ApprovalRequest) -> bool {
794 true
795 }
796
797 fn decide(&self, _request: &ApprovalRequest) -> ApprovalFuture<Option<ToolApprovalDecision>> {
798 Box::pin(async { Ok(None) })
799 }
800}
801
802struct SteeringRuntimeHook {
803 queue: SteeringQueue,
804}
805
806impl RuntimeHook for SteeringRuntimeHook {
807 fn before_llm(&self, event: BeforeLlmEvent<'_>) -> Option<BeforeLlmPatch> {
808 let queued = {
809 let Ok(mut queue) = self.queue.lock() else {
810 return None;
811 };
812 queue.drain(..).collect::<Vec<_>>()
813 };
814 if queued.is_empty() {
815 return None;
816 }
817 let mut messages = event.messages.to_vec();
818 messages.extend(
819 queued
820 .iter()
821 .map(|input| Message::user(input_text(input)))
822 .filter(|message| !message.content.is_empty()),
823 );
824 Some(BeforeLlmPatch {
825 messages: Some(messages),
826 tool_schemas: None,
827 })
828 }
829
830 fn before_tool_call(&self, event: BeforeToolCallEvent<'_>) -> Option<BeforeToolCallPatch> {
831 let has_steering = self
832 .queue
833 .lock()
834 .map(|queue| !queue.is_empty())
835 .unwrap_or(false);
836 has_steering.then(|| {
837 ToolExecutionResult::success(
838 event.call.id.clone(),
839 "Tool skipped due to queued steering message.",
840 )
841 .into()
842 })
843 }
844}
845
846fn effective_approval_request_timeout(config: &RunConfig, fallback: Duration) -> Duration {
847 config.approval_timeout.unwrap_or(fallback)
848}
849
850fn item_from_notification(notification: &ServerNotification) -> Option<AppItem> {
851 match notification {
852 ServerNotification::AgentMessageDelta(AgentMessageDeltaParams { item, .. })
853 | ServerNotification::ItemStarted(ItemStartedParams { item })
854 | ServerNotification::ItemCompleted(ItemCompletedParams { item }) => Some(item.clone()),
855 _ => None,
856 }
857}
858
859fn input_text(input: &[UserInput]) -> String {
860 input
861 .iter()
862 .filter_map(|item| {
863 if item.get("type").and_then(Value::as_str) == Some("text") {
864 item.get("text").and_then(Value::as_str).map(str::to_string)
865 } else if let Some(text) = item.get("text").and_then(Value::as_str) {
866 Some(text.to_string())
867 } else if item.is_null() {
868 None
869 } else {
870 Some(item.to_string())
871 }
872 })
873 .filter(|text| !text.is_empty())
874 .collect::<Vec<_>>()
875 .join("\n")
876}
877
878fn turn_status(status: AgentStatus) -> TurnStatus {
879 match status {
880 AgentStatus::WaitUser | AgentStatus::ReconciliationRequired => TurnStatus::Interrupted,
881 AgentStatus::Completed => TurnStatus::Completed,
882 AgentStatus::Pending | AgentStatus::Running => TurnStatus::Running,
883 AgentStatus::Failed | AgentStatus::MaxCycles => TurnStatus::Failed,
884 }
885}
886
887fn app_token_usage(usage: &TaskTokenUsage) -> AppTokenUsage {
888 AppTokenUsage {
889 prompt_tokens: usage.prompt_tokens,
890 completion_tokens: usage.completion_tokens,
891 total_tokens: usage.total_tokens,
892 cached_tokens: usage.cached_tokens,
893 reasoning_tokens: usage.reasoning_tokens,
894 input_tokens: usage.input_tokens,
895 output_tokens: usage.output_tokens,
896 cache_creation_tokens: usage.cache_creation_tokens,
897 cache_usage: AppCacheUsage {
898 status: match usage.cache_usage.status {
899 CacheUsageStatus::ProviderReported => "provider_reported",
900 CacheUsageStatus::AccountingMissing => "accounting_missing",
901 CacheUsageStatus::Unsupported => "unsupported",
902 }
903 .to_string(),
904 read_tokens: usage.cache_usage.read_tokens,
905 write_tokens: usage.cache_usage.write_tokens,
906 uncached_input_tokens: usage.cache_usage.uncached_input_tokens,
907 source: usage.cache_usage.source.clone(),
908 },
909 }
910}
911
912fn app_json_object(value: &impl serde::Serialize) -> BTreeMap<String, Value> {
913 let Value::Object(fields) =
914 serde_json::to_value(value).expect("typed App Server observation must serialize")
915 else {
916 unreachable!("typed App Server observation must serialize as an object");
917 };
918 fields.into_iter().collect()
919}
920
921fn store_error(error: ThreadStoreError) -> AppServerError {
922 AppServerError::internal(error.to_string())
923}
924
925#[cfg(test)]
926mod tests {
927 use super::*;
928
929 #[test]
930 fn turn_run_config_approval_timeout_overrides_adapter_fallback() {
931 let configured = Duration::from_millis(25);
932 let config = RunConfig {
933 approval_timeout: Some(configured),
934 ..RunConfig::default()
935 };
936
937 assert_eq!(
938 effective_approval_request_timeout(&config, Duration::from_secs(30)),
939 configured
940 );
941 assert_eq!(
942 effective_approval_request_timeout(&RunConfig::default(), Duration::from_secs(30)),
943 Duration::from_secs(30)
944 );
945 }
946}