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 messages = Vec::new();
311        for input in &self.inputs {
312            if let Some(message) =
313                committed_message_from_pending_input(input, attachment_store).await?
314            {
315                messages.push(message);
316            }
317        }
318        Ok(QueuedCheckpointTurnInput {
319            messages,
320            turn_causes: Vec::new(),
321        })
322    }
323
324    pub fn materialize_for_turn(&self) -> TurnInput {
325        let mut input_items = Vec::new();
326        let mut image_blobs = std::collections::HashMap::new();
327        let mut protocol_turn_options = None;
328        let mut trace_turn_id = None;
329        for pending in &self.inputs {
330            input_items.extend(pending.input.items.clone());
331            image_blobs.extend(pending.input.image_blobs.clone());
332            if protocol_turn_options.is_none() {
333                protocol_turn_options = pending.input.protocol_turn_options.clone();
334            }
335            if trace_turn_id.is_none() {
336                trace_turn_id = pending.input.trace_turn_id.clone();
337            }
338        }
339        TurnInput {
340            items: input_items,
341            image_blobs,
342            protocol_turn_options,
343            trace_turn_id,
344            protocol_extension: None,
345            turn_context: crate::TurnContext::default(),
346        }
347    }
348}
349
350#[derive(Clone, Debug, Default)]
351pub struct QueuedCheckpointTurnInput {
352    pub messages: Vec<crate::Message>,
353    pub turn_causes: Vec<TurnCause>,
354}
355
356pub(crate) fn source_key_display_id(source: &str) -> String {
357    source
358        .strip_prefix("host:")
359        .or_else(|| source.strip_prefix("injection:"))
360        .unwrap_or(source)
361        .to_string()
362}
363
364pub(crate) fn plugin_message_from_turn_input(input: &TurnInput) -> Option<PluginMessage> {
365    let mut text = Vec::new();
366    let mut images = Vec::new();
367    for item in &input.items {
368        match item {
369            crate::InputItem::Text { text: item_text } if !item_text.is_empty() => {
370                text.push(item_text.clone());
371            }
372            crate::InputItem::Text { .. } => {}
373            crate::InputItem::ImageRef { id } => {
374                if let Some(bytes) = input.image_blobs.get(id).cloned() {
375                    images.push(bytes);
376                }
377            }
378        }
379    }
380    if text.is_empty() && images.is_empty() {
381        return None;
382    }
383    Some(PluginMessage {
384        id: None,
385        role: crate::MessageRole::User,
386        content: text.join("\n"),
387        origin: None,
388        parts: Vec::new(),
389        images,
390    })
391}
392
393async fn committed_message_from_pending_input(
394    pending: &PendingTurnInput,
395    attachment_store: &crate::SessionAttachmentStore,
396) -> Result<Option<crate::Message>, String> {
397    let normalized = super::io::normalize_input_items(
398        &pending.input.items,
399        &pending.input.image_blobs,
400        attachment_store,
401    )
402    .await?;
403    let message_id = format!("m_ingress_{}", pending.input_id);
404    let mut parts = Vec::new();
405    for item in normalized {
406        match item {
407            super::NormalizedItem::Text(text) if !text.is_empty() => {
408                let part_id = format!("{message_id}.p{}", parts.len());
409                parts.push(crate::Part {
410                    id: part_id,
411                    kind: crate::PartKind::Text,
412                    content: text,
413                    attachment: None,
414                    tool_call_id: None,
415                    tool_name: None,
416                    tool_replay: None,
417                    prune_state: crate::PruneState::Intact,
418                    reasoning_meta: None,
419                    response_meta: None,
420                });
421            }
422            super::NormalizedItem::Text(_) => {}
423            super::NormalizedItem::Image(reference) => {
424                let part_id = format!("{message_id}.p{}", parts.len());
425                parts.push(crate::Part {
426                    id: part_id,
427                    kind: crate::PartKind::Image,
428                    content: String::new(),
429                    attachment: Some(crate::session_model::message::PartAttachment { reference }),
430                    tool_call_id: None,
431                    tool_name: None,
432                    tool_replay: None,
433                    prune_state: crate::PruneState::Intact,
434                    reasoning_meta: None,
435                    response_meta: None,
436                });
437            }
438        }
439    }
440    if parts.is_empty() {
441        return Ok(None);
442    }
443    Ok(Some(crate::Message {
444        id: message_id,
445        role: crate::MessageRole::User,
446        origin: None,
447        parts: crate::shared_parts(parts),
448    }))
449}
lash_core/runtime/turn_input_ingress.rs

lash_core/runtime/
turn_input_ingress.rs
