lash_core/runtime/
turn_input_ingress.rs

1use crate::{CheckpointKind, PluginMessage, TurnCause, TurnInput};
2
3#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
4#[serde(tag = "scope", rename_all = "snake_case")]
5pub enum TurnInputIngress {
6    ActiveTurn {
7        turn_id: String,
8        #[serde(default)]
9        min_boundary: TurnInputCheckpointBoundary,
10    },
11    NextTurn,
12}
13
14impl TurnInputIngress {
15    pub fn active_turn(
16        turn_id: impl Into<String>,
17        min_boundary: TurnInputCheckpointBoundary,
18    ) -> Self {
19        Self::ActiveTurn {
20            turn_id: turn_id.into(),
21            min_boundary,
22        }
23    }
24
25    pub fn next_turn() -> Self {
26        Self::NextTurn
27    }
28
29    pub fn active_turn_id(&self) -> Option<&str> {
30        match self {
31            Self::ActiveTurn { turn_id, .. } => Some(turn_id),
32            Self::NextTurn => None,
33        }
34    }
35
36    pub fn admits_checkpoint(&self, checkpoint: CheckpointKind) -> bool {
37        match self {
38            Self::ActiveTurn { min_boundary, .. } => min_boundary.admits(checkpoint),
39            Self::NextTurn => false,
40        }
41    }
42}
43
44#[derive(
45    Clone, Copy, Debug, Default, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize,
46)]
47#[serde(rename_all = "snake_case")]
48pub enum TurnInputCheckpointBoundary {
49    #[default]
50    AfterWork,
51    BeforeCompletion,
52}
53
54impl TurnInputCheckpointBoundary {
55    pub fn admits(self, checkpoint: CheckpointKind) -> bool {
56        match self {
57            Self::AfterWork => true,
58            Self::BeforeCompletion => checkpoint == CheckpointKind::BeforeCompletion,
59        }
60    }
61}
62
63#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
64#[serde(rename_all = "snake_case")]
65pub enum TurnInputState {
66    PendingActive,
67    DeferredNextTurn,
68    Accepted,
69    Cancelled,
70    Completed,
71}
72
73impl TurnInputState {
74    pub fn as_str(self) -> &'static str {
75        match self {
76            Self::PendingActive => "pending_active",
77            Self::DeferredNextTurn => "deferred_next_turn",
78            Self::Accepted => "accepted",
79            Self::Cancelled => "cancelled",
80            Self::Completed => "completed",
81        }
82    }
83
84    pub fn from_wire_str(value: &str) -> Option<Self> {
85        match value {
86            "pending_active" => Some(Self::PendingActive),
87            "deferred_next_turn" => Some(Self::DeferredNextTurn),
88            "accepted" => Some(Self::Accepted),
89            "cancelled" => Some(Self::Cancelled),
90            "completed" => Some(Self::Completed),
91            _ => None,
92        }
93    }
94
95    pub fn is_next_turn_pending(self) -> bool {
96        matches!(self, Self::DeferredNextTurn)
97    }
98}
99
100#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
101pub struct PendingTurnInputDraft {
102    pub session_id: String,
103    #[serde(default, skip_serializing_if = "Option::is_none")]
104    pub input_id: Option<String>,
105    #[serde(default, skip_serializing_if = "Option::is_none")]
106    pub source_key: Option<String>,
107    pub ingress: TurnInputIngress,
108    pub input: TurnInput,
109}
110
111impl PendingTurnInputDraft {
112    pub fn new(session_id: impl Into<String>, ingress: TurnInputIngress, input: TurnInput) -> Self {
113        Self {
114            session_id: session_id.into(),
115            input_id: None,
116            source_key: None,
117            ingress,
118            input,
119        }
120    }
121
122    pub fn with_input_id(mut self, input_id: impl Into<String>) -> Self {
123        self.input_id = Some(input_id.into());
124        self
125    }
126
127    pub fn with_source_key(mut self, source_key: impl Into<String>) -> Self {
128        self.source_key = Some(source_key.into());
129        self
130    }
131
132    pub fn submitted_content_matches(
133        &self,
134        existing: &PendingTurnInput,
135    ) -> Result<bool, serde_json::Error> {
136        Ok(self.ingress == existing.ingress
137            && serde_json::to_value(&self.input)? == serde_json::to_value(&existing.input)?)
138    }
139}
140
141#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
142pub struct PendingTurnInput {
143    pub input_id: String,
144    pub session_id: String,
145    pub enqueue_seq: u64,
146    #[serde(default, skip_serializing_if = "Option::is_none")]
147    pub source_key: Option<String>,
148    pub ingress: TurnInputIngress,
149    pub state: TurnInputState,
150    pub enqueued_at_ms: u64,
151    pub input: TurnInput,
152}
153
154impl PendingTurnInput {
155    pub fn source_or_id(&self) -> &str {
156        self.source_key.as_deref().unwrap_or(&self.input_id)
157    }
158
159    pub fn accepted_input(&self) -> Option<crate::AcceptedInjectedTurnInput> {
160        plugin_message_from_turn_input(&self.input).map(|message| {
161            crate::AcceptedInjectedTurnInput {
162                id: self
163                    .source_key
164                    .as_deref()
165                    .map(source_key_display_id)
166                    .or_else(|| Some(self.input_id.clone())),
167                message,
168            }
169        })
170    }
171}
172
173#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
174#[serde(tag = "kind", content = "value", rename_all = "snake_case")]
175pub enum PendingTurnInputCancelTarget {
176    InputId(String),
177    SourceKey(String),
178}
179
180impl PendingTurnInputCancelTarget {
181    pub fn input_id(input_id: impl Into<String>) -> Self {
182        Self::InputId(input_id.into())
183    }
184
185    pub fn source_key(source_key: impl Into<String>) -> Self {
186        Self::SourceKey(source_key.into())
187    }
188}
189
190#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
191pub struct PendingTurnInputClaimDiagnostics {
192    pub state: TurnInputState,
193    #[serde(default, skip_serializing_if = "Option::is_none")]
194    pub claim_id: Option<String>,
195    #[serde(default, skip_serializing_if = "Option::is_none")]
196    pub claim_owner: Option<crate::LeaseOwnerIdentity>,
197    /// The session-execution-lease generation the live claim pins, when a claim
198    /// holds the row. `None` when the row carries no claim (ADR 0029).
199    #[serde(default, skip_serializing_if = "Option::is_none")]
200    pub claim_session_lease_generation: Option<u64>,
201    pub claim_fencing_token: u64,
202}
203
204#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
205#[serde(tag = "outcome", content = "data", rename_all = "snake_case")]
206pub enum PendingTurnInputCancelOutcome {
207    Cancelled(PendingTurnInput),
208    AlreadyClaimed {
209        input: PendingTurnInput,
210        #[serde(default, skip_serializing_if = "Option::is_none")]
211        claim: Option<PendingTurnInputClaimDiagnostics>,
212    },
213    AlreadyCompleted(PendingTurnInput),
214    AlreadyCancelled(PendingTurnInput),
215    NotFound,
216}
217
218impl PendingTurnInputCancelOutcome {
219    pub fn is_cancelled(&self) -> bool {
220        matches!(self, Self::Cancelled(_))
221    }
222
223    pub fn input(&self) -> Option<&PendingTurnInput> {
224        match self {
225            Self::Cancelled(input)
226            | Self::AlreadyClaimed { input, .. }
227            | Self::AlreadyCompleted(input)
228            | Self::AlreadyCancelled(input) => Some(input),
229            Self::NotFound => None,
230        }
231    }
232}
233
234#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
235pub struct PendingTurnInputCancelResult {
236    pub target: PendingTurnInputCancelTarget,
237    pub outcome: PendingTurnInputCancelOutcome,
238}
239
240#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
241#[serde(tag = "outcome", content = "data", rename_all = "snake_case")]
242pub enum PendingTurnInputSuffixCancelOutcome {
243    AnchorNotFound {
244        anchor: PendingTurnInputCancelTarget,
245    },
246    Outcomes {
247        anchor: PendingTurnInputCancelTarget,
248        outcomes: Vec<PendingTurnInputCancelOutcome>,
249    },
250}
251
252#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
253#[serde(tag = "kind", rename_all = "snake_case")]
254pub enum TurnInputClaimMode {
255    ActiveTurn {
256        turn_id: String,
257        checkpoint: CheckpointKind,
258    },
259    NextTurn,
260}
261
262#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
263pub struct TurnInputCompletion {
264    pub session_id: String,
265    pub claim_id: String,
266    pub lease_token: String,
267    pub input_ids: Vec<String>,
268}
269
270#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
271pub struct TurnInputClaim {
272    pub session_id: String,
273    pub claim_id: String,
274    pub owner: crate::LeaseOwnerIdentity,
275    pub lease_token: String,
276    pub fencing_token: u64,
277    /// The session-execution-lease generation this claim pins. The claim is
278    /// live exactly while this generation still holds the session lease
279    /// (ADR 0029).
280    pub session_lease_generation: u64,
281    pub mode: TurnInputClaimMode,
282    pub inputs: Vec<PendingTurnInput>,
283}
284
285impl TurnInputClaim {
286    pub fn completion(&self) -> TurnInputCompletion {
287        TurnInputCompletion {
288            session_id: self.session_id.clone(),
289            claim_id: self.claim_id.clone(),
290            lease_token: self.lease_token.clone(),
291            input_ids: self
292                .inputs
293                .iter()
294                .map(|input| input.input_id.clone())
295                .collect(),
296        }
297    }
298
299    pub fn accepted_turn_inputs(&self) -> Vec<crate::AcceptedInjectedTurnInput> {
300        self.inputs
301            .iter()
302            .filter_map(PendingTurnInput::accepted_input)
303            .collect()
304    }
305
306    pub async fn materialize_for_checkpoint(
307        &self,
308        attachment_store: &crate::SessionAttachmentStore,
309    ) -> Result<QueuedCheckpointTurnInput, String> {
310        let mut transient_messages = Vec::new();
311        for input in &self.inputs {
312            if let Some(message) =
313                plugin_message_from_turn_input_with_attachments(&input.input, attachment_store)
314                    .await?
315            {
316                transient_messages.push(message);
317            }
318        }
319        Ok(QueuedCheckpointTurnInput {
320            transient_messages,
321            turn_causes: Vec::new(),
322        })
323    }
324
325    pub fn materialize_for_turn(&self) -> TurnInput {
326        let mut input_items = Vec::new();
327        let mut image_blobs = std::collections::HashMap::new();
328        let mut protocol_turn_options = None;
329        let mut trace_turn_id = None;
330        for pending in &self.inputs {
331            input_items.extend(pending.input.items.clone());
332            image_blobs.extend(pending.input.image_blobs.clone());
333            if protocol_turn_options.is_none() {
334                protocol_turn_options = pending.input.protocol_turn_options.clone();
335            }
336            if trace_turn_id.is_none() {
337                trace_turn_id = pending.input.trace_turn_id.clone();
338            }
339        }
340        TurnInput {
341            items: input_items,
342            image_blobs,
343            protocol_turn_options,
344            trace_turn_id,
345            protocol_extension: None,
346            turn_context: crate::TurnContext::default(),
347        }
348    }
349}
350
351#[derive(Clone, Debug, Default)]
352pub struct QueuedCheckpointTurnInput {
353    pub transient_messages: Vec<PluginMessage>,
354    pub turn_causes: Vec<TurnCause>,
355}
356
357pub(crate) fn source_key_display_id(source: &str) -> String {
358    source
359        .strip_prefix("host:")
360        .or_else(|| source.strip_prefix("injection:"))
361        .unwrap_or(source)
362        .to_string()
363}
364
365pub(crate) fn plugin_message_from_turn_input(input: &TurnInput) -> Option<PluginMessage> {
366    let mut text = Vec::new();
367    let mut images = Vec::new();
368    for item in &input.items {
369        match item {
370            crate::InputItem::Text { text: item_text } if !item_text.is_empty() => {
371                text.push(item_text.clone());
372            }
373            crate::InputItem::Text { .. } => {}
374            crate::InputItem::ImageRef { id } => {
375                if let Some(bytes) = input.image_blobs.get(id).cloned() {
376                    images.push(bytes);
377                }
378            }
379        }
380    }
381    if text.is_empty() && images.is_empty() {
382        return None;
383    }
384    Some(PluginMessage {
385        role: crate::MessageRole::User,
386        content: text.join("\n"),
387        origin: None,
388        parts: Vec::new(),
389        images,
390    })
391}
392
393pub(crate) async fn plugin_message_from_turn_input_with_attachments(
394    input: &TurnInput,
395    attachment_store: &crate::SessionAttachmentStore,
396) -> Result<Option<PluginMessage>, String> {
397    let normalized =
398        super::io::normalize_input_items(&input.items, &input.image_blobs, attachment_store)
399            .await?;
400    let has_image = normalized
401        .iter()
402        .any(|item| matches!(item, super::NormalizedItem::Image(_)));
403    if !has_image {
404        return Ok(plugin_message_from_turn_input(input));
405    }
406
407    let mut content = Vec::new();
408    let mut parts = Vec::new();
409    for item in normalized {
410        match item {
411            super::NormalizedItem::Text(text) if !text.is_empty() => {
412                let part_id = format!("pending.p{}", parts.len());
413                content.push(text.clone());
414                parts.push(crate::Part {
415                    id: part_id,
416                    kind: crate::PartKind::Text,
417                    content: text,
418                    attachment: None,
419                    tool_call_id: None,
420                    tool_name: None,
421                    tool_replay: None,
422                    prune_state: crate::PruneState::Intact,
423                    reasoning_meta: None,
424                    response_meta: None,
425                });
426            }
427            super::NormalizedItem::Text(_) => {}
428            super::NormalizedItem::Image(reference) => {
429                let part_id = format!("pending.p{}", parts.len());
430                parts.push(crate::Part {
431                    id: part_id,
432                    kind: crate::PartKind::Image,
433                    content: String::new(),
434                    attachment: Some(crate::session_model::message::PartAttachment { reference }),
435                    tool_call_id: None,
436                    tool_name: None,
437                    tool_replay: None,
438                    prune_state: crate::PruneState::Intact,
439                    reasoning_meta: None,
440                    response_meta: None,
441                });
442            }
443        }
444    }
445    if parts.is_empty() {
446        return Ok(None);
447    }
448    Ok(Some(PluginMessage {
449        role: crate::MessageRole::User,
450        content: content.join("\n"),
451        origin: None,
452        parts,
453        images: Vec::new(),
454    }))
455}
lash_core/runtime/turn_input_ingress.rs

lash_core/runtime/
turn_input_ingress.rs
