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