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
154/// Durable acceptance evidence returned to an ingress caller.
155///
156/// The receipt deliberately carries only stable routing and idempotency
157/// identity. Queue dispatch and the pending row's mutable lifecycle state are
158/// observed separately through the pending-input reconciliation surface.
159#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
160pub struct TurnInputAcceptanceReceipt {
161    pub input_id: String,
162    pub session_id: String,
163    #[serde(default, skip_serializing_if = "Option::is_none")]
164    pub source_key: Option<String>,
165    pub ingress: TurnInputIngress,
166}
167
168/// Durable evidence that an admitted input became canonical conversation input.
169///
170/// This is the application stage between [`TurnInputAcceptanceReceipt`]
171/// (admission) and the terminal turn commit (settlement). It deliberately
172/// carries identity only: hosts correlate an input to its canonical turn and
173/// committed message without parsing or retaining display text.
174#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
175pub struct TurnInputApplication {
176    pub input_id: String,
177    #[serde(default, skip_serializing_if = "Option::is_none")]
178    pub source_key: Option<String>,
179    pub turn_id: String,
180    pub committed_message_id: String,
181    /// Present for active-turn checkpoint application and absent when the
182    /// input formed the initial canonical input of an idle queued turn.
183    #[serde(default, skip_serializing_if = "Option::is_none")]
184    pub checkpoint: Option<CheckpointKind>,
185}
186
187impl From<&PendingTurnInput> for TurnInputAcceptanceReceipt {
188    fn from(input: &PendingTurnInput) -> Self {
189        Self {
190            input_id: input.input_id.clone(),
191            session_id: input.session_id.clone(),
192            source_key: input.source_key.clone(),
193            ingress: input.ingress.clone(),
194        }
195    }
196}
197
198impl PendingTurnInput {
199    pub fn source_or_id(&self) -> &str {
200        self.source_key.as_deref().unwrap_or(&self.input_id)
201    }
202
203    pub fn accepted_input(&self) -> Option<crate::AcceptedInjectedTurnInput> {
204        plugin_message_from_turn_input(&self.input).map(|message| {
205            crate::AcceptedInjectedTurnInput {
206                id: self
207                    .source_key
208                    .as_deref()
209                    .map(source_key_display_id)
210                    .or_else(|| Some(self.input_id.clone())),
211                message,
212            }
213        })
214    }
215}
216
217#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
218#[serde(tag = "kind", content = "value", rename_all = "snake_case")]
219pub enum PendingTurnInputCancelTarget {
220    InputId(String),
221    SourceKey(String),
222}
223
224impl PendingTurnInputCancelTarget {
225    pub fn input_id(input_id: impl Into<String>) -> Self {
226        Self::InputId(input_id.into())
227    }
228
229    pub fn source_key(source_key: impl Into<String>) -> Self {
230        Self::SourceKey(source_key.into())
231    }
232}
233
234#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
235pub struct PendingTurnInputClaimDiagnostics {
236    pub state: TurnInputState,
237    #[serde(default, skip_serializing_if = "Option::is_none")]
238    pub claim_id: Option<String>,
239    #[serde(default, skip_serializing_if = "Option::is_none")]
240    pub claim_owner: Option<crate::LeaseOwnerIdentity>,
241    /// The session-execution-lease generation the live claim pins, when a claim
242    /// holds the row. `None` when the row carries no claim (ADR 0029).
243    #[serde(default, skip_serializing_if = "Option::is_none")]
244    pub claim_session_lease_generation: Option<u64>,
245    pub claim_fencing_token: u64,
246}
247
248#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
249#[serde(tag = "outcome", content = "data", rename_all = "snake_case")]
250pub enum PendingTurnInputCancelOutcome {
251    Cancelled(PendingTurnInput),
252    AlreadyClaimed {
253        input: PendingTurnInput,
254        #[serde(default, skip_serializing_if = "Option::is_none")]
255        claim: Option<PendingTurnInputClaimDiagnostics>,
256    },
257    AlreadyCompleted(PendingTurnInput),
258    AlreadyCancelled(PendingTurnInput),
259    NotFound,
260}
261
262impl PendingTurnInputCancelOutcome {
263    pub fn is_cancelled(&self) -> bool {
264        matches!(self, Self::Cancelled(_))
265    }
266
267    pub fn input(&self) -> Option<&PendingTurnInput> {
268        match self {
269            Self::Cancelled(input)
270            | Self::AlreadyClaimed { input, .. }
271            | Self::AlreadyCompleted(input)
272            | Self::AlreadyCancelled(input) => Some(input),
273            Self::NotFound => None,
274        }
275    }
276}
277
278#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
279pub struct PendingTurnInputCancelResult {
280    pub target: PendingTurnInputCancelTarget,
281    pub outcome: PendingTurnInputCancelOutcome,
282}
283
284#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
285#[serde(tag = "outcome", content = "data", rename_all = "snake_case")]
286pub enum PendingTurnInputSuffixCancelOutcome {
287    AnchorNotFound {
288        anchor: PendingTurnInputCancelTarget,
289    },
290    Outcomes {
291        anchor: PendingTurnInputCancelTarget,
292        outcomes: Vec<PendingTurnInputCancelOutcome>,
293    },
294}
295
296#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
297#[serde(tag = "kind", rename_all = "snake_case")]
298pub enum TurnInputClaimMode {
299    ActiveTurn {
300        turn_id: String,
301        checkpoint: CheckpointKind,
302    },
303    NextTurn,
304}
305
306#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
307pub struct TurnInputCompletion {
308    pub session_id: String,
309    pub claim_id: String,
310    pub lease_token: String,
311    pub input_ids: Vec<String>,
312    #[serde(default, skip_serializing_if = "Vec::is_empty")]
313    pub applications: Vec<TurnInputApplication>,
314}
315
316#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
317pub struct TurnInputClaim {
318    pub session_id: String,
319    pub claim_id: String,
320    pub owner: crate::LeaseOwnerIdentity,
321    pub lease_token: String,
322    pub fencing_token: u64,
323    /// The session-execution-lease generation this claim pins. The claim is
324    /// live exactly while this generation still holds the session lease
325    /// (ADR 0029).
326    pub session_lease_generation: u64,
327    pub mode: TurnInputClaimMode,
328    pub inputs: Vec<PendingTurnInput>,
329    #[serde(default, skip_serializing_if = "Vec::is_empty")]
330    pub applications: Vec<TurnInputApplication>,
331}
332
333impl TurnInputClaim {
334    pub fn completion(&self) -> TurnInputCompletion {
335        TurnInputCompletion {
336            session_id: self.session_id.clone(),
337            claim_id: self.claim_id.clone(),
338            lease_token: self.lease_token.clone(),
339            input_ids: self
340                .inputs
341                .iter()
342                .map(|input| input.input_id.clone())
343                .collect(),
344            applications: self.applications.clone(),
345        }
346    }
347
348    pub fn record_initial_turn_application(&mut self, turn_id: &str, committed_message_id: &str) {
349        self.applications = self
350            .inputs
351            .iter()
352            .filter(|input| {
353                input.input.items.iter().any(|item| match item {
354                    crate::InputItem::Text { text } => !text.is_empty(),
355                    crate::InputItem::Attachment { .. } => true,
356                })
357            })
358            .map(|input| TurnInputApplication {
359                input_id: input.input_id.clone(),
360                source_key: input.source_key.clone(),
361                turn_id: turn_id.to_string(),
362                committed_message_id: committed_message_id.to_string(),
363                checkpoint: None,
364            })
365            .collect();
366    }
367
368    pub fn record_checkpoint_applications(
369        &mut self,
370        turn_id: &str,
371        checkpoint: CheckpointKind,
372        committed_messages: &[crate::Message],
373    ) {
374        let committed_message_ids = committed_messages
375            .iter()
376            .map(|message| message.id.as_str())
377            .collect::<std::collections::HashSet<_>>();
378        self.applications = self
379            .inputs
380            .iter()
381            .filter_map(|input| {
382                let committed_message_id = ingress_message_id(&input.input_id);
383                committed_message_ids
384                    .contains(committed_message_id.as_str())
385                    .then(|| TurnInputApplication {
386                        input_id: input.input_id.clone(),
387                        source_key: input.source_key.clone(),
388                        turn_id: turn_id.to_string(),
389                        committed_message_id,
390                        checkpoint: Some(checkpoint),
391                    })
392            })
393            .collect();
394    }
395
396    pub fn accepted_turn_inputs(&self) -> Vec<crate::AcceptedInjectedTurnInput> {
397        self.inputs
398            .iter()
399            .filter_map(PendingTurnInput::accepted_input)
400            .collect()
401    }
402
403    pub async fn materialize_for_checkpoint(
404        &self,
405        attachment_store: &crate::SessionAttachmentStore,
406        attachment_source_policy: &dyn crate::AttachmentSourcePolicy,
407    ) -> Result<QueuedCheckpointTurnInput, String> {
408        let mut messages = Vec::new();
409        for input in &self.inputs {
410            if let Some(message) = committed_message_from_pending_input(
411                input,
412                attachment_store,
413                attachment_source_policy,
414            )
415            .await?
416            {
417                messages.push(message);
418            }
419        }
420        Ok(QueuedCheckpointTurnInput {
421            messages,
422            turn_causes: Vec::new(),
423        })
424    }
425
426    pub fn materialize_for_turn(&self) -> TurnInput {
427        let mut input_items = Vec::new();
428        let mut protocol_turn_options = None;
429        let mut trace_turn_id = None;
430        for pending in &self.inputs {
431            input_items.extend(pending.input.items.clone());
432            if protocol_turn_options.is_none() {
433                protocol_turn_options = pending.input.protocol_turn_options.clone();
434            }
435            if trace_turn_id.is_none() {
436                trace_turn_id = pending.input.trace_turn_id.clone();
437            }
438        }
439        TurnInput {
440            items: input_items,
441            protocol_turn_options,
442            trace_turn_id,
443            protocol_extension: None,
444            turn_context: crate::TurnContext::default(),
445        }
446    }
447}
448
449#[derive(Clone, Debug, Default)]
450pub struct QueuedCheckpointTurnInput {
451    pub messages: Vec<crate::Message>,
452    pub turn_causes: Vec<TurnCause>,
453}
454
455pub(crate) fn source_key_display_id(source: &str) -> String {
456    source
457        .strip_prefix("host:")
458        .or_else(|| source.strip_prefix("injection:"))
459        .unwrap_or(source)
460        .to_string()
461}
462
463pub(crate) fn plugin_message_from_turn_input(input: &TurnInput) -> Option<PluginMessage> {
464    let mut text = Vec::new();
465    let mut attachments = Vec::new();
466    for item in &input.items {
467        match item {
468            crate::InputItem::Text { text: item_text } if !item_text.is_empty() => {
469                text.push(item_text.clone());
470            }
471            crate::InputItem::Text { .. } => {}
472            crate::InputItem::Attachment { source } => attachments.push(source.clone()),
473        }
474    }
475    if text.is_empty() && attachments.is_empty() {
476        return None;
477    }
478    Some(PluginMessage {
479        id: None,
480        role: crate::MessageRole::User,
481        content: text.join("\n"),
482        origin: None,
483        parts: Vec::new(),
484        attachments,
485    })
486}
487
488async fn committed_message_from_pending_input(
489    pending: &PendingTurnInput,
490    attachment_store: &crate::SessionAttachmentStore,
491    attachment_source_policy: &dyn crate::AttachmentSourcePolicy,
492) -> Result<Option<crate::Message>, String> {
493    let normalized = super::io::normalize_input_items(
494        &pending.input.items,
495        attachment_store,
496        attachment_source_policy,
497    )
498    .await?;
499    let message_id = ingress_message_id(&pending.input_id);
500    let mut parts = Vec::new();
501    for item in normalized {
502        match item {
503            super::NormalizedItem::Text(text) if !text.is_empty() => {
504                let part_id = format!("{message_id}.p{}", parts.len());
505                parts.push(crate::Part {
506                    id: part_id,
507                    kind: crate::PartKind::Text,
508                    content: text,
509                    attachment: None,
510                    tool_call_id: None,
511                    tool_name: None,
512                    tool_replay: None,
513                    prune_state: crate::PruneState::Intact,
514                    reasoning_meta: None,
515                    response_meta: None,
516                });
517            }
518            super::NormalizedItem::Text(_) => {}
519            super::NormalizedItem::Attachment(source) => {
520                let part_id = format!("{message_id}.p{}", parts.len());
521                parts.push(crate::Part {
522                    id: part_id,
523                    kind: crate::PartKind::Attachment,
524                    content: String::new(),
525                    attachment: Some(crate::session_model::message::PartAttachment { source }),
526                    tool_call_id: None,
527                    tool_name: None,
528                    tool_replay: None,
529                    prune_state: crate::PruneState::Intact,
530                    reasoning_meta: None,
531                    response_meta: None,
532                });
533            }
534        }
535    }
536    if parts.is_empty() {
537        return Ok(None);
538    }
539    Ok(Some(crate::Message {
540        id: message_id,
541        role: crate::MessageRole::User,
542        origin: None,
543        parts: crate::shared_parts(parts),
544    }))
545}
546
547pub(crate) fn ingress_message_id(input_id: &str) -> String {
548    format!("m_ingress_{input_id}")
549}
lash_core/runtime/turn_input_ingress.rs

lash_core/runtime/
turn_input_ingress.rs
