1use crate::{CheckpointKind, PluginMessage, TurnCause, TurnInput};
2
3pub const TURN_INPUT_CLAIM_TTL_MS: u64 = 30 * 1000;
4
5#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
6#[serde(tag = "scope", rename_all = "snake_case")]
7pub enum TurnInputIngress {
8 ActiveTurn {
9 turn_id: String,
10 #[serde(default)]
11 min_boundary: TurnInputCheckpointBoundary,
12 },
13 NextTurn,
14}
15
16impl TurnInputIngress {
17 pub fn active_turn(
18 turn_id: impl Into<String>,
19 min_boundary: TurnInputCheckpointBoundary,
20 ) -> Self {
21 Self::ActiveTurn {
22 turn_id: turn_id.into(),
23 min_boundary,
24 }
25 }
26
27 pub fn next_turn() -> Self {
28 Self::NextTurn
29 }
30
31 pub fn active_turn_id(&self) -> Option<&str> {
32 match self {
33 Self::ActiveTurn { turn_id, .. } => Some(turn_id),
34 Self::NextTurn => None,
35 }
36 }
37
38 pub fn admits_checkpoint(&self, checkpoint: CheckpointKind) -> bool {
39 match self {
40 Self::ActiveTurn { min_boundary, .. } => min_boundary.admits(checkpoint),
41 Self::NextTurn => false,
42 }
43 }
44}
45
46#[derive(
47 Clone, Copy, Debug, Default, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize,
48)]
49#[serde(rename_all = "snake_case")]
50pub enum TurnInputCheckpointBoundary {
51 #[default]
52 AfterWork,
53 BeforeCompletion,
54}
55
56impl TurnInputCheckpointBoundary {
57 pub fn admits(self, checkpoint: CheckpointKind) -> bool {
58 match self {
59 Self::AfterWork => true,
60 Self::BeforeCompletion => checkpoint == CheckpointKind::BeforeCompletion,
61 }
62 }
63}
64
65#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
66#[serde(rename_all = "snake_case")]
67pub enum TurnInputState {
68 PendingActive,
69 DeferredNextTurn,
70 Accepted,
71 Cancelled,
72 Completed,
73}
74
75impl TurnInputState {
76 pub fn as_str(self) -> &'static str {
77 match self {
78 Self::PendingActive => "pending_active",
79 Self::DeferredNextTurn => "deferred_next_turn",
80 Self::Accepted => "accepted",
81 Self::Cancelled => "cancelled",
82 Self::Completed => "completed",
83 }
84 }
85
86 pub fn from_wire_str(value: &str) -> Option<Self> {
87 match value {
88 "pending_active" => Some(Self::PendingActive),
89 "deferred_next_turn" => Some(Self::DeferredNextTurn),
90 "accepted" => Some(Self::Accepted),
91 "cancelled" => Some(Self::Cancelled),
92 "completed" => Some(Self::Completed),
93 _ => None,
94 }
95 }
96
97 pub fn is_next_turn_pending(self) -> bool {
98 matches!(self, Self::DeferredNextTurn)
99 }
100}
101
102#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
103pub struct PendingTurnInputDraft {
104 pub session_id: String,
105 #[serde(default, skip_serializing_if = "Option::is_none")]
106 pub input_id: Option<String>,
107 #[serde(default, skip_serializing_if = "Option::is_none")]
108 pub source_key: Option<String>,
109 pub ingress: TurnInputIngress,
110 pub input: TurnInput,
111}
112
113impl PendingTurnInputDraft {
114 pub fn new(session_id: impl Into<String>, ingress: TurnInputIngress, input: TurnInput) -> Self {
115 Self {
116 session_id: session_id.into(),
117 input_id: None,
118 source_key: None,
119 ingress,
120 input,
121 }
122 }
123
124 pub fn with_input_id(mut self, input_id: impl Into<String>) -> Self {
125 self.input_id = Some(input_id.into());
126 self
127 }
128
129 pub fn with_source_key(mut self, source_key: impl Into<String>) -> Self {
130 self.source_key = Some(source_key.into());
131 self
132 }
133}
134
135#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
136pub struct PendingTurnInput {
137 pub input_id: String,
138 pub session_id: String,
139 pub enqueue_seq: u64,
140 #[serde(default, skip_serializing_if = "Option::is_none")]
141 pub source_key: Option<String>,
142 pub ingress: TurnInputIngress,
143 pub state: TurnInputState,
144 pub enqueued_at_ms: u64,
145 pub input: TurnInput,
146}
147
148impl PendingTurnInput {
149 pub fn source_or_id(&self) -> &str {
150 self.source_key.as_deref().unwrap_or(&self.input_id)
151 }
152
153 pub fn accepted_input(&self) -> Option<crate::AcceptedInjectedTurnInput> {
154 plugin_message_from_turn_input(&self.input).map(|message| {
155 crate::AcceptedInjectedTurnInput {
156 id: self
157 .source_key
158 .as_deref()
159 .map(source_key_display_id)
160 .or_else(|| Some(self.input_id.clone())),
161 message,
162 }
163 })
164 }
165}
166
167#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
168#[serde(tag = "kind", rename_all = "snake_case")]
169pub enum TurnInputClaimMode {
170 ActiveTurn {
171 turn_id: String,
172 checkpoint: CheckpointKind,
173 },
174 NextTurn,
175}
176
177#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
178pub struct TurnInputCompletion {
179 pub session_id: String,
180 pub claim_id: String,
181 pub lease_token: String,
182 pub input_ids: Vec<String>,
183}
184
185#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
186pub struct TurnInputClaim {
187 pub session_id: String,
188 pub claim_id: String,
189 pub owner: crate::LeaseOwnerIdentity,
190 pub lease_token: String,
191 pub fencing_token: u64,
192 pub claimed_at_epoch_ms: u64,
193 pub expires_at_epoch_ms: u64,
194 pub mode: TurnInputClaimMode,
195 pub inputs: Vec<PendingTurnInput>,
196}
197
198impl TurnInputClaim {
199 pub fn completion(&self) -> TurnInputCompletion {
200 TurnInputCompletion {
201 session_id: self.session_id.clone(),
202 claim_id: self.claim_id.clone(),
203 lease_token: self.lease_token.clone(),
204 input_ids: self
205 .inputs
206 .iter()
207 .map(|input| input.input_id.clone())
208 .collect(),
209 }
210 }
211
212 pub fn accepted_turn_inputs(&self) -> Vec<crate::AcceptedInjectedTurnInput> {
213 self.inputs
214 .iter()
215 .filter_map(PendingTurnInput::accepted_input)
216 .collect()
217 }
218
219 pub async fn materialize_for_checkpoint(
220 &self,
221 attachment_store: &dyn crate::AttachmentStore,
222 ) -> Result<QueuedCheckpointTurnInput, String> {
223 let mut transient_messages = Vec::new();
224 for input in &self.inputs {
225 if let Some(message) =
226 plugin_message_from_turn_input_with_attachments(&input.input, attachment_store)
227 .await?
228 {
229 transient_messages.push(message);
230 }
231 }
232 Ok(QueuedCheckpointTurnInput {
233 transient_messages,
234 turn_causes: Vec::new(),
235 })
236 }
237
238 pub fn materialize_for_turn(&self) -> TurnInput {
239 let mut input_items = Vec::new();
240 let mut image_blobs = std::collections::HashMap::new();
241 let mut protocol_turn_options = None;
242 let mut trace_turn_id = None;
243 for pending in &self.inputs {
244 input_items.extend(pending.input.items.clone());
245 image_blobs.extend(pending.input.image_blobs.clone());
246 if protocol_turn_options.is_none() {
247 protocol_turn_options = pending.input.protocol_turn_options.clone();
248 }
249 if trace_turn_id.is_none() {
250 trace_turn_id = pending.input.trace_turn_id.clone();
251 }
252 }
253 TurnInput {
254 items: input_items,
255 image_blobs,
256 protocol_turn_options,
257 trace_turn_id,
258 protocol_extension: None,
259 turn_context: crate::TurnContext::default(),
260 }
261 }
262}
263
264#[derive(Clone, Debug, Default)]
265pub struct QueuedCheckpointTurnInput {
266 pub transient_messages: Vec<PluginMessage>,
267 pub turn_causes: Vec<TurnCause>,
268}
269
270pub(crate) fn source_key_display_id(source: &str) -> String {
271 source
272 .strip_prefix("host:")
273 .or_else(|| source.strip_prefix("injection:"))
274 .unwrap_or(source)
275 .to_string()
276}
277
278pub(crate) fn plugin_message_from_turn_input(input: &TurnInput) -> Option<PluginMessage> {
279 let mut text = Vec::new();
280 let mut images = Vec::new();
281 for item in &input.items {
282 match item {
283 crate::InputItem::Text { text: item_text } if !item_text.is_empty() => {
284 text.push(item_text.clone());
285 }
286 crate::InputItem::Text { .. } => {}
287 crate::InputItem::ImageRef { id } => {
288 if let Some(bytes) = input.image_blobs.get(id).cloned() {
289 images.push(bytes);
290 }
291 }
292 }
293 }
294 if text.is_empty() && images.is_empty() {
295 return None;
296 }
297 Some(PluginMessage {
298 role: crate::MessageRole::User,
299 content: text.join("\n"),
300 origin: None,
301 parts: Vec::new(),
302 images,
303 })
304}
305
306pub(crate) async fn plugin_message_from_turn_input_with_attachments(
307 input: &TurnInput,
308 attachment_store: &dyn crate::AttachmentStore,
309) -> Result<Option<PluginMessage>, String> {
310 let normalized =
311 super::io::normalize_input_items(&input.items, &input.image_blobs, attachment_store)
312 .await?;
313 let has_image = normalized
314 .iter()
315 .any(|item| matches!(item, super::NormalizedItem::Image(_)));
316 if !has_image {
317 return Ok(plugin_message_from_turn_input(input));
318 }
319
320 let mut content = Vec::new();
321 let mut parts = Vec::new();
322 for item in normalized {
323 match item {
324 super::NormalizedItem::Text(text) if !text.is_empty() => {
325 let part_id = format!("pending.p{}", parts.len());
326 content.push(text.clone());
327 parts.push(crate::Part {
328 id: part_id,
329 kind: crate::PartKind::Text,
330 content: text,
331 attachment: None,
332 tool_call_id: None,
333 tool_name: None,
334 tool_replay: None,
335 prune_state: crate::PruneState::Intact,
336 reasoning_meta: None,
337 response_meta: None,
338 });
339 }
340 super::NormalizedItem::Text(_) => {}
341 super::NormalizedItem::Image(reference) => {
342 let part_id = format!("pending.p{}", parts.len());
343 parts.push(crate::Part {
344 id: part_id,
345 kind: crate::PartKind::Image,
346 content: String::new(),
347 attachment: Some(crate::session_model::message::PartAttachment { reference }),
348 tool_call_id: None,
349 tool_name: None,
350 tool_replay: None,
351 prune_state: crate::PruneState::Intact,
352 reasoning_meta: None,
353 response_meta: None,
354 });
355 }
356 }
357 }
358 if parts.is_empty() {
359 return Ok(None);
360 }
361 Ok(Some(PluginMessage {
362 role: crate::MessageRole::User,
363 content: content.join("\n"),
364 origin: None,
365 parts,
366 images: Vec::new(),
367 }))
368}