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

lash_core/runtime/
turn_input_ingress.rs
