1use super::*;
2
3impl Runtime {
4 pub fn start_turn(
5 self: &Arc<Self>,
6 req: StartTurnRequest,
7 ) -> BoxFuture<'_, anyhow::Result<TurnId>> {
8 Box::pin(async move {
9 self.start_turn_admitted(req, false, None)
10 .await?
11 .ok_or_else(|| anyhow::anyhow!("explicit turn was not admitted"))
12 })
13 }
14
15 pub(crate) fn start_goal_turn_if_idle(
16 self: &Arc<Self>,
17 req: StartTurnRequest,
18 ) -> BoxFuture<'_, anyhow::Result<Option<TurnId>>> {
19 self.start_turn_admitted(req, true, None)
20 }
21
22 pub(crate) fn start_event_turn_if_idle(
23 self: &Arc<Self>,
24 req: StartTurnRequest,
25 subscription_id: String,
26 ) -> BoxFuture<'_, anyhow::Result<Option<TurnId>>> {
27 self.start_turn_admitted(req, false, Some(subscription_id))
28 }
29
30 fn start_turn_admitted(
31 self: &Arc<Self>,
32 mut req: StartTurnRequest,
33 goal_only: bool,
34 event_subscription: Option<String>,
35 ) -> BoxFuture<'_, anyhow::Result<Option<TurnId>>> {
36 Box::pin(async move {
37 let _thread_admission = self.thread_admission(&req.thread_id).await;
38 let turn_admission = self.turn_admission.lock().await;
39 let event_only = event_subscription.is_some();
40 if let Some(id) = &event_subscription {
41 if self
42 .thread_activity(&req.thread_id)
43 .await
44 .active_turn_id
45 .is_some()
46 {
47 return Ok(None);
48 }
49 let Some(mut current) = self
51 .thread_events
52 .delivery_request(&req.thread_id, id)
53 .await
54 else {
55 return Ok(None);
56 };
57 current.message = req.message;
58 current.developer_context = req.developer_context;
59 req = current;
60 if self
61 .thread_goal_get(&req.thread_id)
62 .await?
63 .is_some_and(|goal| {
64 matches!(
65 goal.status,
66 roder_api::goals::ThreadGoalStatus::BudgetLimited
67 | roder_api::goals::ThreadGoalStatus::Paused
68 )
69 })
70 {
71 return Ok(None);
72 }
73 }
74 let _goal_admission = if goal_only {
75 if self.has_active_turn_for_thread(&req.thread_id).await {
76 return Ok(None);
77 }
78 let Some((guard, goal)) = self.goals.admit_continuation(&req.thread_id).await?
79 else {
80 return Ok(None);
81 };
82 if let Some(previous) = self.goals.continuation_request(&req.thread_id).await {
83 req = previous;
84 }
85 req.message = crate::goals::continuation_prompt(&goal);
86 Some(guard)
87 } else {
88 None
89 };
90 self.ensure_execution_authority()?;
91 anyhow::ensure!(
92 self.accepting_turns.load(Ordering::Acquire),
93 "runtime is quiescing and cannot accept new turns"
94 );
95 req.workspace = validate_thread_workspace(&req.workspace)?;
96 let team_member = self.teams.member_for_thread(&req.thread_id).await;
97 if event_only
98 && team_member
99 .as_ref()
100 .is_some_and(|(_, member)| member.status == TeamMemberStatus::Closed)
101 {
102 return Ok(None);
103 }
104 let cfg = self.config.read().await.clone();
105 let provider = req
106 .provider_override
107 .clone()
108 .unwrap_or_else(|| cfg.default_provider.clone());
109 self.engine_for(&provider)?;
110 if !event_only {
111 self.thread_events.remember(&req).await;
112 }
113 let turn_id = uuid::Uuid::new_v4().to_string();
114 let mut initial_mailbox_ack = None;
115 if let Some((team_id, member)) = &team_member {
116 let pending = self
117 .teams
118 .reserve_pending_mailbox_messages(team_id, &member.id, &turn_id)
119 .await?;
120 if !pending.is_empty()
121 && let Some(team) = self.read_team(team_id).await
122 {
123 let mailbox = format_mailbox_messages(&team, &pending);
124 req.message = if req.message.trim().is_empty() {
125 mailbox
126 } else {
127 format!("{mailbox}\n\n[Direct task input]\n{}", req.message)
128 };
129 initial_mailbox_ack = Some(MailboxDeliveryAck {
130 team_id: team_id.clone(),
131 message_ids: pending.iter().map(|message| message.id.clone()).collect(),
132 });
133 }
134 }
135 let (abort_handle, abort_registration) = AbortHandle::new_pair();
136 let drain = Arc::new(TurnDrainHandle {
137 thread_id: req.thread_id.clone(),
138 interrupt_requested: AtomicBool::new(false),
139 interrupt_reason: Mutex::new(None),
140 completed: AtomicBool::new(false),
141 completed_notify: Notify::new(),
142 });
143 let (steer_changed, steering) = tokio::sync::watch::channel(0);
144 let active = ActiveTurnHandle {
145 thread_id: req.thread_id.clone(),
146 abort: abort_handle,
147 steers: Arc::new(Mutex::new(Vec::new())),
148 steer_changed,
149 drain,
150 };
151 self.active_turns
152 .write()
153 .await
154 .insert(turn_id.clone(), active);
155 self.record_turn_lifecycle(
156 req.thread_id.clone(),
157 turn_id.clone(),
158 TurnLifecycleState::Running,
159 TurnCleanupState::NotRequested,
160 None,
161 )
162 .await;
163 self.active_turn_contexts.write().await.insert(
164 turn_id.clone(),
165 InheritedTurnContext {
166 workspace: req.workspace.clone(),
167 instructions: req.instructions.clone(),
168 developer_context: req.developer_context.clone(),
169 },
170 );
171 if let Some((team_id, member)) = team_member {
172 let updated = match self
173 .teams
174 .update_member(&team_id, &member.id, |member| {
175 member.current_turn_id = Some(turn_id.clone());
176 member.status = TeamMemberStatus::Running;
177 member.final_message = None;
178 member.terminal_error = None;
179 })
180 .await
181 {
182 Ok(updated) => updated,
183 Err(error) => {
184 self.active_turns.write().await.remove(&turn_id);
185 self.active_turn_contexts.write().await.remove(&turn_id);
186 self.teams
187 .release_mailbox_reservations_for_turn(&turn_id)
188 .await;
189 return Err(error);
190 }
191 };
192 if let Some(member) = updated
193 .members
194 .into_iter()
195 .find(|candidate| candidate.id == member.id)
196 {
197 self.emit(RoderEvent::TeamMemberStatusChanged(
198 TeamMemberStatusChanged {
199 team_id,
200 member_id: member.id,
201 member_thread_id: member.thread_id,
202 status: TeamMemberStatus::Running,
203 timestamp: OffsetDateTime::now_utc(),
204 },
205 ))
206 .await;
207 }
208 }
209 self.goals.remember_turn_options(&req).await;
210 let runtime = Arc::clone(self);
211 let turn_req = req;
212 let thread_id_for_task = turn_req.thread_id.clone();
213 let turn_id_for_task = turn_id.clone();
214 runtime.initialize_turn_usage(&turn_id).await;
215 tokio::spawn(async move {
216 let result = Abortable::new(
217 runtime.run_turn(
218 turn_req,
219 turn_id_for_task.clone(),
220 initial_mailbox_ack,
221 steering,
222 goal_only,
223 ),
224 abort_registration,
225 )
226 .await;
227 runtime
235 .cancel_pending_external_tool_calls_for_turn(&turn_id_for_task)
236 .await;
237 let completed = matches!(&result, Ok(Ok(TurnRunOutcome::Completed)));
238 if let Err(error) = runtime
239 .sync_turn_goal_usage(&thread_id_for_task, &turn_id_for_task)
240 .await
241 {
242 eprintln!("failed to finalize turn token budget: {error}");
243 }
244 let stopped_status = match &result {
245 Ok(Err(error)) => Some(crate::goals::status_after_error(error)),
246 Ok(Ok(TurnRunOutcome::Stopped)) => {
247 Some(roder_api::goals::ThreadGoalStatus::Blocked)
248 }
249 Err(_) => Some(roder_api::goals::ThreadGoalStatus::Paused),
250 _ => None,
251 };
252 if let Err(error) = runtime
253 .goals
254 .finish_turn(&thread_id_for_task, &turn_id_for_task, stopped_status)
255 .await
256 {
257 eprintln!("failed to finalize goal accounting: {error}");
258 }
259 match &result {
260 Ok(Err(err)) => {
261 let (cleanup, ownership) =
262 runtime.await_provider_turn_cleanup(&turn_id_for_task).await;
263 runtime
265 .emit(RoderEvent::TurnFailed(TurnFailed {
266 thread_id: thread_id_for_task.clone(),
267 turn_id: turn_id_for_task.clone(),
268 error: err.to_string(),
269 error_kind: err
270 .downcast_ref::<roder_api::provider_error::ProviderFailure>()
271 .map(|failure| failure.kind.retry_cause().to_string()),
272 usage: None,
273 timestamp: OffsetDateTime::now_utc(),
274 }))
275 .await;
276 runtime
277 .record_turn_lifecycle_with_ownership(
278 thread_id_for_task.clone(),
279 turn_id_for_task.clone(),
280 TurnLifecycleState::Failed,
281 cleanup,
282 Some(TurnLifecycleReason::ProviderFailure),
283 ownership,
284 )
285 .await;
286 let _ = runtime
287 .complete_team_member_turn_with_result(
288 &thread_id_for_task,
289 &turn_id_for_task,
290 TeamMemberStatus::Failed,
291 None,
292 Some(err.to_string()),
293 )
294 .await;
295 }
296 Ok(Ok(TurnRunOutcome::Stopped)) => {
297 let (cleanup, ownership) =
298 runtime.await_provider_turn_cleanup(&turn_id_for_task).await;
299 runtime
300 .record_turn_lifecycle_with_ownership(
301 thread_id_for_task.clone(),
302 turn_id_for_task.clone(),
303 TurnLifecycleState::Failed,
304 cleanup,
305 Some(TurnLifecycleReason::ProviderFailure),
306 ownership,
307 )
308 .await;
309 let _ = runtime
310 .complete_team_member_turn_with_result(
311 &thread_id_for_task,
312 &turn_id_for_task,
313 TeamMemberStatus::Failed,
314 None,
315 Some("turn stopped before completion".to_string()),
316 )
317 .await;
318 }
319 Err(_) => {
320 let reason = if let Some(handle) = runtime
321 .turn_drains
322 .read()
323 .await
324 .get(&turn_id_for_task)
325 .cloned()
326 {
327 handle
328 .interrupt_reason
329 .lock()
330 .await
331 .unwrap_or(TurnLifecycleReason::RuntimeFailure)
332 } else {
333 TurnLifecycleReason::RuntimeFailure
334 };
335 let (cleanup, ownership) =
336 runtime.await_provider_turn_cleanup(&turn_id_for_task).await;
337 runtime
338 .record_turn_lifecycle_with_ownership(
339 thread_id_for_task.clone(),
340 turn_id_for_task.clone(),
341 TurnLifecycleState::Interrupted,
342 cleanup,
343 Some(reason),
344 ownership,
345 )
346 .await;
347 runtime
348 .emit(RoderEvent::TurnInterrupted(TurnInterrupted {
349 thread_id: thread_id_for_task.clone(),
350 turn_id: turn_id_for_task.clone(),
351 usage: None,
352 timestamp: OffsetDateTime::now_utc(),
353 }))
354 .await;
355 let _ = runtime
356 .complete_team_member_turn_with_result(
357 &thread_id_for_task,
358 &turn_id_for_task,
359 TeamMemberStatus::Interrupted,
360 None,
361 None,
362 )
363 .await;
364 }
365 Ok(Ok(TurnRunOutcome::Completed)) => {}
366 }
367 runtime
368 .teams
369 .release_mailbox_reservations_for_turn(&turn_id_for_task)
370 .await;
371 runtime.active_turns.write().await.remove(&turn_id_for_task);
372 runtime.turn_usage.lock().await.remove(&turn_id_for_task);
373 if let Some(drain) = runtime.turn_drains.write().await.remove(&turn_id_for_task) {
374 drain.completed.store(true, Ordering::Release);
375 drain.completed_notify.notify_waiters();
376 }
377 runtime
378 .active_turn_selections
379 .write()
380 .await
381 .remove(&turn_id_for_task);
382 runtime
383 .active_turn_contexts
384 .write()
385 .await
386 .remove(&turn_id_for_task);
387 if !completed {
388 let _ = runtime.provider_turn_cleanups.lock().map(|mut cleanups| {
392 cleanups.remove(&turn_id_for_task);
393 });
394 }
395 runtime.active_turns_changed.notify_waiters();
396 if completed {
397 let _ = runtime
398 .continue_active_goal_after_turn(thread_id_for_task)
399 .await;
400 }
401 });
402 drop(turn_admission);
403 Ok(Some(turn_id))
404 })
405 }
406}