lash_core/runtime/
turn_queue.rs

1use super::process::ProcessWakeDelivery;
2use crate::{PluginMessage, TurnCause, TurnInput};
3
4#[derive(Clone, Debug, PartialEq, serde::Serialize, serde::Deserialize)]
5#[serde(tag = "kind", rename_all = "snake_case")]
6pub enum SessionCommand {
7    // No generation guard: the command drains asynchronously, so any
8    // generation observed at enqueue time may legitimately have advanced by
9    // drain time, and the refresh recomputes the surface from live sources
10    // regardless — a guard could only fail spuriously.
11    RefreshToolCatalog { reason: String },
12    ResetSession { reason: String },
13}
14
15impl SessionCommand {
16    pub fn kind(&self) -> &'static str {
17        match self {
18            Self::RefreshToolCatalog { .. } => "refresh_tool_catalog",
19            Self::ResetSession { .. } => "reset_session",
20        }
21    }
22
23    pub fn source_key(&self, idempotency_key: impl AsRef<str>) -> String {
24        format!("command:{}:{}", self.kind(), idempotency_key.as_ref())
25    }
26}
27
28#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
29pub struct SessionCommandReceipt {
30    pub session_id: String,
31    pub batch_id: String,
32    pub source_key: String,
33}
34
35#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
36#[serde(rename_all = "snake_case")]
37pub enum DeliveryPolicy {
38    EarliestSafeBoundary,
39    AfterCurrentTurnCommit,
40}
41
42impl DeliveryPolicy {
43    pub fn as_str(self) -> &'static str {
44        match self {
45            Self::EarliestSafeBoundary => "earliest_safe_boundary",
46            Self::AfterCurrentTurnCommit => "after_current_turn_commit",
47        }
48    }
49
50    pub fn from_wire_str(value: &str) -> Option<Self> {
51        match value {
52            "earliest_safe_boundary" => Some(Self::EarliestSafeBoundary),
53            "after_current_turn_commit" => Some(Self::AfterCurrentTurnCommit),
54            _ => None,
55        }
56    }
57}
58
59#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
60#[serde(rename_all = "snake_case")]
61pub enum SlotPolicy {
62    Join,
63    Exclusive,
64}
65
66impl SlotPolicy {
67    pub fn as_str(self) -> &'static str {
68        match self {
69            Self::Join => "join",
70            Self::Exclusive => "exclusive",
71        }
72    }
73
74    pub fn from_wire_str(value: &str) -> Option<Self> {
75        match value {
76            "join" => Some(Self::Join),
77            "exclusive" => Some(Self::Exclusive),
78            _ => None,
79        }
80    }
81}
82
83#[derive(Clone, Debug, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
84#[serde(rename_all = "snake_case")]
85pub enum MergeKey {
86    Never,
87    PayloadDefault,
88    Group(String),
89}
90
91#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
92#[serde(tag = "type", rename_all = "snake_case")]
93pub enum QueuedWorkPayload {
94    ProcessWake { wake: Box<ProcessWakeDelivery> },
95    SessionCommand { command: Box<SessionCommand> },
96}
97
98#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
99#[serde(rename_all = "snake_case")]
100pub enum QueuedWorkClass {
101    SessionCommand,
102    TurnWork,
103}
104
105impl QueuedWorkPayload {
106    pub fn process_wake(wake: ProcessWakeDelivery) -> Self {
107        Self::ProcessWake {
108            wake: Box::new(wake),
109        }
110    }
111
112    pub fn session_command(command: SessionCommand) -> Self {
113        Self::SessionCommand {
114            command: Box::new(command),
115        }
116    }
117
118    pub fn work_class(&self) -> QueuedWorkClass {
119        match self {
120            Self::SessionCommand { .. } => QueuedWorkClass::SessionCommand,
121            Self::ProcessWake { .. } => QueuedWorkClass::TurnWork,
122        }
123    }
124}
125
126#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
127pub struct QueuedWorkItem {
128    pub item_id: String,
129    pub payload: QueuedWorkPayload,
130}
131
132#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
133pub struct QueuedWorkBatch {
134    pub batch_id: String,
135    pub session_id: String,
136    pub enqueue_seq: u64,
137    #[serde(default, skip_serializing_if = "Option::is_none")]
138    pub source_key: Option<String>,
139    pub delivery_policy: DeliveryPolicy,
140    pub slot_policy: SlotPolicy,
141    pub merge_key: MergeKey,
142    pub available_at_ms: u64,
143    pub enqueued_at_ms: u64,
144    pub items: Vec<QueuedWorkItem>,
145}
146
147impl QueuedWorkBatch {
148    pub fn work_class(&self) -> Option<QueuedWorkClass> {
149        work_class_for_payloads(self.items.iter().map(|item| &item.payload))
150    }
151
152    pub fn is_session_command_work(&self) -> bool {
153        self.work_class() == Some(QueuedWorkClass::SessionCommand)
154    }
155
156    pub fn is_turn_work(&self) -> bool {
157        self.work_class() == Some(QueuedWorkClass::TurnWork)
158    }
159}
160
161#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
162pub struct QueuedWorkBatchDraft {
163    pub session_id: String,
164    #[serde(default, skip_serializing_if = "Option::is_none")]
165    pub source_key: Option<String>,
166    pub delivery_policy: DeliveryPolicy,
167    pub slot_policy: SlotPolicy,
168    pub merge_key: MergeKey,
169    pub available_at_ms: u64,
170    pub payloads: Vec<QueuedWorkPayload>,
171}
172
173impl QueuedWorkBatchDraft {
174    pub fn new(
175        session_id: impl Into<String>,
176        delivery_policy: DeliveryPolicy,
177        slot_policy: SlotPolicy,
178        payloads: impl Into<Vec<QueuedWorkPayload>>,
179    ) -> Self {
180        Self {
181            session_id: session_id.into(),
182            source_key: None,
183            delivery_policy,
184            slot_policy,
185            merge_key: MergeKey::Never,
186            available_at_ms: 0,
187            payloads: payloads.into(),
188        }
189    }
190
191    pub fn with_source_key(mut self, source_key: impl Into<String>) -> Self {
192        self.source_key = Some(source_key.into());
193        self
194    }
195
196    pub fn with_available_at_ms(mut self, available_at_ms: u64) -> Self {
197        self.available_at_ms = available_at_ms;
198        self
199    }
200
201    pub fn with_merge_key(mut self, merge_key: MergeKey) -> Self {
202        self.merge_key = merge_key;
203        self
204    }
205
206    pub fn work_class(&self) -> Option<QueuedWorkClass> {
207        work_class_for_payloads(self.payloads.iter())
208    }
209}
210
211fn work_class_for_payloads<'a>(
212    payloads: impl IntoIterator<Item = &'a QueuedWorkPayload>,
213) -> Option<QueuedWorkClass> {
214    let mut payloads = payloads.into_iter();
215    let first = payloads.next()?.work_class();
216    payloads
217        .all(|payload| payload.work_class() == first)
218        .then_some(first)
219}
220
221#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
222#[serde(rename_all = "snake_case")]
223pub enum QueuedWorkClaimBoundary {
224    ActiveTurnCheckpoint,
225    Idle,
226}
227
228#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
229pub struct QueuedWorkCompletion {
230    pub session_id: String,
231    pub claim_id: String,
232    pub lease_token: String,
233    pub batch_ids: Vec<String>,
234}
235
236#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
237pub struct QueuedWorkClaim {
238    pub session_id: String,
239    pub claim_id: String,
240    pub owner: crate::LeaseOwnerIdentity,
241    pub lease_token: String,
242    pub fencing_token: u64,
243    /// The session-execution-lease generation this claim pins. The claim is
244    /// live exactly while this generation still holds the session lease
245    /// (ADR 0029).
246    pub session_lease_generation: u64,
247    pub batches: Vec<QueuedWorkBatch>,
248}
249
250impl QueuedWorkClaim {
251    pub fn completion(&self) -> QueuedWorkCompletion {
252        QueuedWorkCompletion {
253            session_id: self.session_id.clone(),
254            claim_id: self.claim_id.clone(),
255            lease_token: self.lease_token.clone(),
256            batch_ids: self
257                .batches
258                .iter()
259                .map(|batch| batch.batch_id.clone())
260                .collect(),
261        }
262    }
263
264    pub fn is_empty(&self) -> bool {
265        self.batches.iter().all(|batch| batch.items.is_empty())
266    }
267
268    pub fn materialize_for_checkpoint(&self) -> QueuedCheckpointWork {
269        let messages = Vec::new();
270        let transient_messages = Vec::new();
271        let mut turn_causes = Vec::new();
272        for batch in &self.batches {
273            for item in &batch.items {
274                match &item.payload {
275                    QueuedWorkPayload::ProcessWake { wake } => {
276                        turn_causes.push(crate::process_wake_turn_cause(wake));
277                    }
278                    QueuedWorkPayload::SessionCommand { .. } => {}
279                }
280            }
281        }
282        QueuedCheckpointWork {
283            messages,
284            transient_messages,
285            turn_causes,
286        }
287    }
288
289    pub async fn materialize_for_checkpoint_with_attachments(
290        &self,
291        _attachment_store: &crate::SessionAttachmentStore,
292    ) -> Result<QueuedCheckpointWork, String> {
293        let messages = Vec::new();
294        let transient_messages = Vec::new();
295        let mut turn_causes = Vec::new();
296        for batch in &self.batches {
297            for item in &batch.items {
298                match &item.payload {
299                    QueuedWorkPayload::ProcessWake { wake } => {
300                        turn_causes.push(crate::process_wake_turn_cause(wake));
301                    }
302                    QueuedWorkPayload::SessionCommand { .. } => {}
303                }
304            }
305        }
306        Ok(QueuedCheckpointWork {
307            messages,
308            transient_messages,
309            turn_causes,
310        })
311    }
312
313    pub fn exclusive_session_command(&self) -> Option<(&QueuedWorkBatch, &SessionCommand)> {
314        if self.batches.len() != 1 {
315            return None;
316        }
317        let batch = self.batches.first()?;
318        if batch.slot_policy != SlotPolicy::Exclusive || batch.items.len() != 1 {
319            return None;
320        }
321        let item = batch.items.first()?;
322        match &item.payload {
323            QueuedWorkPayload::SessionCommand { command } => Some((batch, command.as_ref())),
324            _ => None,
325        }
326    }
327
328    pub fn materialize_for_turn(&self) -> QueuedTurnWork {
329        let checkpoint = self.materialize_for_checkpoint();
330        QueuedTurnWork {
331            input: TurnInput::empty(),
332            messages: checkpoint.messages,
333            turn_causes: checkpoint.turn_causes,
334        }
335    }
336}
337
338#[derive(Clone, Debug, Default)]
339pub struct QueuedCheckpointWork {
340    pub messages: Vec<PluginMessage>,
341    pub transient_messages: Vec<PluginMessage>,
342    pub turn_causes: Vec<TurnCause>,
343}
344
345#[derive(Clone, Debug)]
346pub struct QueuedTurnWork {
347    pub input: TurnInput,
348    pub messages: Vec<PluginMessage>,
349    pub turn_causes: Vec<TurnCause>,
350}
351
352pub fn process_wake_batch_draft(wake: ProcessWakeDelivery) -> QueuedWorkBatchDraft {
353    let source_key = format!("process:{}:event:{}:wake", wake.process_id, wake.sequence);
354    QueuedWorkBatchDraft::new(
355        wake.target_session_id.clone(),
356        DeliveryPolicy::EarliestSafeBoundary,
357        SlotPolicy::Exclusive,
358        vec![QueuedWorkPayload::process_wake(wake)],
359    )
360    .with_source_key(source_key)
361}
lash_core/runtime/turn_queue.rs

lash_core/runtime/
turn_queue.rs
