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