lash_core/runtime/process/
events.rs

1use std::collections::BTreeMap;
2use std::time::SystemTime;
3
4use serde::{Deserialize, Serialize};
5
6use super::model::{ProcessId, RecoveryDisposition, SessionScope, SessionScopeId};
7use super::validation::process_event_payload_hash;
8
9#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
10pub struct ProcessEventType {
11    pub name: String,
12    pub payload_schema: crate::LashSchema,
13    pub semantics: ProcessEventSemanticsSpec,
14}
15
16#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
17pub struct ProcessEventSemanticsSpec {
18    #[serde(default, skip_serializing_if = "Option::is_none")]
19    pub terminal: Option<ProcessTerminalSpec>,
20    #[serde(default, skip_serializing_if = "Option::is_none")]
21    pub wake: Option<ProcessWakeSpec>,
22}
23
24#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
25pub struct ProcessTerminalSpec {
26    pub state: ProcessTerminalState,
27    #[serde(default, skip_serializing_if = "Option::is_none")]
28    pub await_output: Option<ProcessValueSelector>,
29}
30
31#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
32pub struct ProcessWakeSpec {
33    #[serde(default, skip_serializing_if = "Option::is_none")]
34    pub when: Option<ProcessValueSelector>,
35    pub input: ProcessValueSelector,
36    #[serde(default)]
37    pub dedupe_key: ProcessWakeDedupeKey,
38}
39
40#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
41#[serde(rename_all = "snake_case")]
42pub enum ProcessWakeDedupeKey {
43    #[default]
44    EventIdentity,
45    Selector(ProcessValueSelector),
46    Const(String),
47}
48
49#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
50#[serde(rename_all = "snake_case")]
51pub enum ProcessValueSelector {
52    Payload,
53    Pointer(String),
54    Const(serde_json::Value),
55    Template {
56        template: String,
57        #[serde(default)]
58        fields: BTreeMap<String, ProcessValueSelector>,
59    },
60    Present(String),
61}
62
63#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
64pub struct ProcessEventSemantics {
65    #[serde(default, skip_serializing_if = "Option::is_none")]
66    pub terminal: Option<ProcessTerminalSemantics>,
67    #[serde(default, skip_serializing_if = "Option::is_none")]
68    pub wake: Option<ProcessWake>,
69}
70
71#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
72#[serde(rename_all = "snake_case")]
73pub enum ProcessTerminalState {
74    Completed,
75    Failed,
76    Cancelled,
77    /// The owner stopped executing the work without recording an outcome. The
78    /// true result is unknowable and no cleanup is assumed to have run. Peer of
79    /// the other three terminals; see ADR 0019.
80    Abandoned,
81}
82
83/// Who wrote an [`ProcessTerminalState::Abandoned`] terminal — the exactly-one
84/// legitimate writer per path (ADR 0019).
85#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
86#[serde(rename_all = "snake_case")]
87pub enum AbandonWriter {
88    /// The owner abandoned its own OwnerBound work inline at graceful drain,
89    /// under its own live lease.
90    OwnerDrain,
91    /// The recovery sweep abandoned an OwnerBound, started row whose holder is
92    /// provably dead.
93    Sweep,
94    /// The sweep reconciled a durable Abandon Request into Abandoned once the
95    /// row's lease had lapsed.
96    ReconciledRequest,
97}
98
99/// Evidence attached to an [`ProcessTerminalState::Abandoned`] terminal: which
100/// path wrote it, the dead-or-lapsed owner identity it was established against
101/// (absent for an externally-owned row lash never executed), and when.
102#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
103pub struct AbandonEvidence {
104    pub writer: AbandonWriter,
105    #[serde(default, skip_serializing_if = "Option::is_none")]
106    pub owner: Option<crate::LeaseOwnerIdentity>,
107    pub epoch_ms: u64,
108}
109
110/// Authority under which an *unleased* terminal completion
111/// ([`ProcessRegistry::complete_process`](super::registry::ProcessRegistry::complete_process))
112/// is written.
113///
114/// Lash-owned workers fence terminal writes with a process lease
115/// (`complete_process_with_lease`), which the store validates against the
116/// persisted `(owner, lease_token, fencing_token)`. The unleased path is
117/// reserved for writers whose single-writer discipline lives *outside* the Lash
118/// lease. In-process Rust cannot make such a token unforgeable; the value of
119/// this type is instead **explicitness + a single validation choke point per
120/// backend + audit evidence** on the terminal write. Every backend calls
121/// [`validate`](Self::validate) against the row's declared
122/// [`RecoveryDisposition`] inside its completion operation, and records the
123/// authority on the durable terminal event (see [`terminal_append_request`]).
124///
125/// There is deliberately no `Default`: a caller must name its authority, the
126/// same footgun-prevention stance the runtime takes elsewhere.
127#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
128#[serde(tag = "authority", rename_all = "snake_case")]
129pub enum ProcessCompletionAuthority {
130    /// An external actor closes an [`RecoveryDisposition::ExternallyOwned`] row
131    /// it holds a handle grant for (the `shell.start` detach path, ADR 0019).
132    /// `granted_to` is the session-scope identity the caller verified holds the
133    /// grant — the audit trail for who closed the row out of band. Rejected on
134    /// any lash-executed disposition: those have a lease-fenced single writer.
135    ExternalOwner { granted_to: String },
136    /// A workflow-key-coalesced substrate (e.g. Restate keyed by `process_id`)
137    /// completes a row it ran itself. Its single-writer discipline is the
138    /// engine's per-key coalescing, not a Lash lease; `workflow_key` records the
139    /// key that served as that discipline. Valid for the lash-executed
140    /// dispositions ([`RecoveryDisposition::Rerunnable`] and
141    /// [`RecoveryDisposition::OwnerBound`], which Restate runs), and rejected on
142    /// [`RecoveryDisposition::ExternallyOwned`] rows — a substrate never runs
143    /// one, so it may not close one.
144    WorkflowKey { workflow_key: String },
145    /// The sweep reconciled a durable Abandon Request on an
146    /// [`RecoveryDisposition::ExternallyOwned`] row (whose lease had lapsed, or
147    /// which Lash never leased) into an
148    /// [`ProcessTerminalState::Abandoned`] terminal. Carries no owner: the
149    /// closure is authorized by the recorded request, not a live writer. Only
150    /// ever writes an `Abandoned` terminal.
151    ReconciledAbandon,
152}
153
154impl ProcessCompletionAuthority {
155    /// Construct [`ExternalOwner`](Self::ExternalOwner) authority naming the
156    /// session-scope identity that holds the handle grant.
157    pub fn external_owner(granted_to: impl Into<String>) -> Self {
158        Self::ExternalOwner {
159            granted_to: granted_to.into(),
160        }
161    }
162
163    /// Construct [`WorkflowKey`](Self::WorkflowKey) authority naming the
164    /// coalescing key that serves as the substrate's single-writer discipline.
165    pub fn workflow_key(workflow_key: impl Into<String>) -> Self {
166        Self::WorkflowKey {
167            workflow_key: workflow_key.into(),
168        }
169    }
170
171    /// Short, stable label for diagnostics.
172    pub fn label(&self) -> &'static str {
173        match self {
174            Self::ExternalOwner { .. } => "external-owner",
175            Self::WorkflowKey { .. } => "workflow-key",
176            Self::ReconciledAbandon => "reconciled-abandon",
177        }
178    }
179
180    /// Validate this authority against the row's declared recovery disposition
181    /// and the terminal outcome being written. This is the single per-backend
182    /// choke point that keeps unleased completion honest: each `complete_process`
183    /// implementation calls it before appending the terminal event, so the
184    /// disposition×authority contract is enforced uniformly across memory,
185    /// SQLite, and Postgres rather than at each scattered caller.
186    pub fn validate(
187        &self,
188        process_id: &str,
189        disposition: RecoveryDisposition,
190        await_output: &ProcessAwaitOutput,
191    ) -> Result<(), crate::PluginError> {
192        let reject = |reason: &str| {
193            Err(crate::PluginError::Session(format!(
194                "process `{process_id}` cannot be completed with {} authority: {reason}",
195                self.label()
196            )))
197        };
198        match self {
199            Self::ExternalOwner { .. } => {
200                if disposition != RecoveryDisposition::ExternallyOwned {
201                    return reject(
202                        "only externally-owned rows may be completed by an external owner; a \
203                         lash-executed row has a lease-fenced single writer",
204                    );
205                }
206            }
207            Self::WorkflowKey { .. } => {
208                if disposition == RecoveryDisposition::ExternallyOwned {
209                    return reject(
210                        "externally-owned rows are never executed by a workflow substrate; they \
211                         close through their external owner or a reconciled abandon request",
212                    );
213                }
214            }
215            Self::ReconciledAbandon => {
216                if disposition != RecoveryDisposition::ExternallyOwned {
217                    return reject(
218                        "reconciled-abandon closes only externally-owned rows; a lash-executed \
219                         row is abandoned under its lease",
220                    );
221                }
222                if await_output.terminal_state() != ProcessTerminalState::Abandoned {
223                    return reject("reconciled-abandon writes only an Abandoned terminal");
224                }
225            }
226        }
227        Ok(())
228    }
229}
230
231/// Terminal event type name for a terminal state.
232pub fn terminal_event_type_name(state: ProcessTerminalState) -> &'static str {
233    match state {
234        ProcessTerminalState::Completed => "process.completed",
235        ProcessTerminalState::Failed => "process.failed",
236        ProcessTerminalState::Cancelled => "process.cancelled",
237        ProcessTerminalState::Abandoned => "process.abandoned",
238    }
239}
240
241/// Build the replay-keyed terminal event append for a completion.
242///
243/// The single source of truth for the terminal event's type, replay key, and
244/// payload shape, shared by every completion path (leased and unleased) across
245/// all backends. When `authority` is supplied — the unleased
246/// [`ProcessRegistry::complete_process`](super::registry::ProcessRegistry::complete_process)
247/// path — it is recorded alongside `await_output` as durable audit evidence
248/// (the leased path's evidence is the lease it releases, so it passes `None`
249/// and the payload is byte-identical to the historical shape). The
250/// `await_output` selector (`/await_output`) is untouched by the sibling key.
251pub fn terminal_append_request(
252    process_id: &str,
253    await_output: &ProcessAwaitOutput,
254    authority: Option<&ProcessCompletionAuthority>,
255) -> ProcessEventAppendRequest {
256    let event_type = terminal_event_type_name(await_output.terminal_state());
257    let mut payload = serde_json::json!({ "await_output": await_output });
258    if let Some(authority) = authority {
259        payload["completion_authority"] =
260            serde_json::to_value(authority).expect("completion authority serializes");
261    }
262    ProcessEventAppendRequest::new(event_type, payload)
263        .with_replay_key(format!("process:{process_id}:terminal:{event_type}"))
264}
265
266#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
267pub struct ProcessTerminalSemantics {
268    pub state: ProcessTerminalState,
269    pub await_output: ProcessAwaitOutput,
270}
271
272#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
273#[serde(tag = "type", rename_all = "snake_case")]
274pub enum ProcessAwaitOutput {
275    Success {
276        value: serde_json::Value,
277        #[serde(default, skip_serializing_if = "Option::is_none")]
278        control: Option<crate::ToolControl>,
279    },
280    Failure {
281        class: crate::ToolFailureClass,
282        code: String,
283        message: String,
284        #[serde(default, skip_serializing_if = "Option::is_none")]
285        raw: Option<serde_json::Value>,
286        #[serde(default, skip_serializing_if = "Option::is_none")]
287        control: Option<crate::ToolControl>,
288    },
289    Cancelled {
290        message: String,
291        #[serde(default, skip_serializing_if = "Option::is_none")]
292        raw: Option<serde_json::Value>,
293        #[serde(default, skip_serializing_if = "Option::is_none")]
294        control: Option<crate::ToolControl>,
295    },
296    /// The owner stopped executing without recording an outcome. Written only by
297    /// the sweep or an owner's graceful drain, never round-tripped from a tool
298    /// (a tool cannot self-report abandonment); see [`AbandonEvidence`]. The
299    /// evidence is boxed so this rare terminal does not enlarge the pervasive
300    /// `ProcessAwaitOutput` that flows through every tool result.
301    Abandoned {
302        evidence: Box<AbandonEvidence>,
303        #[serde(default, skip_serializing_if = "Option::is_none")]
304        control: Option<crate::ToolControl>,
305    },
306}
307
308impl ProcessAwaitOutput {
309    pub fn terminal_state(&self) -> ProcessTerminalState {
310        match self {
311            Self::Success { .. } => ProcessTerminalState::Completed,
312            Self::Failure { .. } => ProcessTerminalState::Failed,
313            Self::Cancelled { .. } => ProcessTerminalState::Cancelled,
314            Self::Abandoned { .. } => ProcessTerminalState::Abandoned,
315        }
316    }
317
318    pub fn from_tool_output(output: crate::ToolCallOutput) -> Self {
319        let control = output.control;
320        match output.outcome {
321            crate::ToolCallOutcome::Success(value) => Self::Success {
322                value: value.to_json_value(),
323                control,
324            },
325            crate::ToolCallOutcome::Failure(failure) => Self::Failure {
326                class: failure.class,
327                code: failure.code,
328                message: failure.message,
329                raw: failure.raw.map(|value| value.to_json_value()),
330                control,
331            },
332            crate::ToolCallOutcome::Cancelled(cancellation) => Self::Cancelled {
333                message: cancellation.message,
334                raw: cancellation.raw.map(|value| value.to_json_value()),
335                control,
336            },
337        }
338    }
339
340    pub fn into_tool_output(self) -> crate::ToolCallOutput {
341        match self {
342            Self::Success { value, control } => {
343                let mut output = crate::ToolCallOutput::success(value);
344                output.control = control;
345                output
346            }
347            Self::Failure {
348                class,
349                code,
350                message,
351                raw,
352                control,
353            } => {
354                let mut failure = crate::ToolFailure::tool(class, code, message);
355                failure.raw = raw.map(crate::ToolValue::from);
356                let mut output = crate::ToolCallOutput::failure(failure);
357                output.control = control;
358                output
359            }
360            Self::Cancelled {
361                message,
362                raw,
363                control,
364            } => {
365                let mut cancellation = crate::ToolCancellation::runtime(message);
366                cancellation.raw = raw.map(crate::ToolValue::from);
367                let mut output = crate::ToolCallOutput::cancelled(cancellation);
368                output.control = control;
369                output
370            }
371            // Abandonment has no `ToolCallOutcome` peer: a tool never self-reports
372            // it. To a caller awaiting the result it surfaces one-directionally as
373            // an external failure whose raw payload names it abandoned and carries
374            // the evidence, while the process layer keeps `Abandoned` a distinct
375            // terminal (ADR 0019). `from_tool_output` therefore never reverses this.
376            Self::Abandoned { evidence, control } => {
377                let raw = serde_json::to_value(&evidence)
378                    .ok()
379                    .map(crate::ToolValue::from);
380                let message = match evidence.writer {
381                    AbandonWriter::OwnerDrain => {
382                        "process abandoned: owner drained without recording an outcome".to_string()
383                    }
384                    AbandonWriter::Sweep => {
385                        "process abandoned: recovery observed the owner provably dead".to_string()
386                    }
387                    AbandonWriter::ReconciledRequest => {
388                        "process abandoned: reconciled abandon request after the lease lapsed"
389                            .to_string()
390                    }
391                };
392                let mut failure = crate::ToolFailure::tool(
393                    crate::ToolFailureClass::External,
394                    "process_abandoned",
395                    message,
396                );
397                failure.raw = raw;
398                let mut output = crate::ToolCallOutput::failure(failure);
399                output.control = control;
400                output
401            }
402        }
403    }
404}
405
406#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
407pub struct ProcessWake {
408    pub input: String,
409    pub dedupe_key: String,
410}
411
412pub fn process_signal_event_type(signal_name: &str) -> Result<String, crate::PluginError> {
413    validate_process_signal_name(signal_name)?;
414    Ok(format!("signal.{signal_name}"))
415}
416
417pub fn process_signal_name_from_event_type(event_type: &str) -> Option<&str> {
418    event_type.strip_prefix("signal.")
419}
420
421pub fn process_signal_wait_key(process_id: &str, signal_name: &str, ordinal: u64) -> String {
422    format!("process:{process_id}:signal.{signal_name}:{ordinal}")
423}
424
425pub fn validate_process_signal_name(signal_name: &str) -> Result<(), crate::PluginError> {
426    let valid = !signal_name.is_empty()
427        && signal_name
428            .chars()
429            .all(|ch| ch.is_ascii_alphanumeric() || ch == '_' || ch == '-');
430    if valid {
431        Ok(())
432    } else {
433        Err(crate::PluginError::Session(format!(
434            "process signal name must be non-empty and contain only ASCII letters, digits, `_`, or `-`, got `{signal_name}`"
435        )))
436    }
437}
438
439#[derive(Clone, Debug, Serialize, Deserialize)]
440pub struct ProcessEvent {
441    pub process_id: ProcessId,
442    pub sequence: u64,
443    pub event_type: String,
444    pub payload: serde_json::Value,
445    pub invocation: crate::RuntimeInvocation,
446    pub semantics: ProcessEventSemantics,
447    pub occurred_at: SystemTime,
448}
449
450#[derive(Clone, Debug, Serialize, Deserialize)]
451pub struct ProcessEventAppendResult {
452    pub event: ProcessEvent,
453    #[serde(default, skip_serializing_if = "Option::is_none")]
454    pub wake_delivery: Option<ProcessWakeDelivery>,
455}
456
457#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
458pub struct ProcessEventAppendRequest {
459    pub event_type: String,
460    pub payload: serde_json::Value,
461    #[serde(default, skip_serializing_if = "Option::is_none")]
462    pub replay: Option<crate::RuntimeReplay>,
463    #[serde(default, skip_serializing_if = "Option::is_none")]
464    pub wake_target_scope: Option<SessionScope>,
465}
466
467impl ProcessEventAppendRequest {
468    pub fn new(event_type: impl Into<String>, payload: serde_json::Value) -> Self {
469        Self {
470            event_type: event_type.into(),
471            payload,
472            replay: None,
473            wake_target_scope: None,
474        }
475    }
476
477    pub fn with_replay_key(mut self, replay_key: impl Into<String>) -> Self {
478        self.replay = Some(crate::RuntimeReplay {
479            key: replay_key.into(),
480        });
481        self
482    }
483
484    pub fn with_optional_replay(mut self, replay: Option<crate::RuntimeReplay>) -> Self {
485        self.replay = replay;
486        self
487    }
488
489    pub fn with_wake_target_scope(mut self, scope: SessionScope) -> Self {
490        self.wake_target_scope = Some(scope);
491        self
492    }
493
494    pub fn with_optional_wake_target_scope(mut self, scope: Option<SessionScope>) -> Self {
495        self.wake_target_scope = scope;
496        self
497    }
498
499    pub fn cancel_requested(process_id: &str, reason: Option<String>) -> Self {
500        let payload = serde_json::json!({
501            "reason": reason,
502        });
503        let replay_key = process_event_payload_hash("process.cancel_requested", &payload)
504            .unwrap_or_else(|_| format!("process:{process_id}:cancel_requested"));
505        Self::new("process.cancel_requested", payload).with_replay_key(format!(
506            "process:{process_id}:cancel_requested:{replay_key}"
507        ))
508    }
509}
510
511#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
512pub struct ProcessWakeDelivery {
513    pub wake_id: String,
514    pub target_session_id: String,
515    pub target_scope_id: SessionScopeId,
516    pub process_id: ProcessId,
517    pub sequence: u64,
518    #[serde(default = "default_process_wake_event_type")]
519    pub event_type: String,
520    #[serde(default = "default_process_wake_event_invocation")]
521    pub event_invocation: crate::RuntimeInvocation,
522    #[serde(default, skip_serializing_if = "Option::is_none")]
523    pub process_caused_by: Option<crate::CausalRef>,
524    pub dedupe_key: String,
525    pub input: String,
526    pub created_at_ms: u64,
527}
528
529fn default_process_wake_event_type() -> String {
530    "process.wake".to_string()
531}
532
533fn default_process_wake_event_invocation() -> crate::RuntimeInvocation {
534    crate::RuntimeInvocation {
535        scope: crate::RuntimeScope::new(""),
536        subject: crate::RuntimeSubject::ProcessEvent {
537            process_id: String::new(),
538            sequence: 0,
539            event_type: default_process_wake_event_type(),
540        },
541        caused_by: None,
542        replay: None,
543    }
544}
545
546pub(super) fn default_process_event_types() -> Vec<ProcessEventType> {
547    vec![
548        ProcessEventType {
549            name: "process.cancel_requested".to_string(),
550            payload_schema: crate::LashSchema::any(),
551            semantics: ProcessEventSemanticsSpec::default(),
552        },
553        ProcessEventType {
554            name: "process.waiting".to_string(),
555            payload_schema: crate::LashSchema::any(),
556            semantics: ProcessEventSemanticsSpec::default(),
557        },
558        ProcessEventType {
559            name: "process.resumed".to_string(),
560            payload_schema: crate::LashSchema::any(),
561            semantics: ProcessEventSemanticsSpec::default(),
562        },
563        terminal_event_type("process.completed", ProcessTerminalState::Completed),
564        terminal_event_type("process.failed", ProcessTerminalState::Failed),
565        terminal_event_type("process.cancelled", ProcessTerminalState::Cancelled),
566        terminal_event_type("process.abandoned", ProcessTerminalState::Abandoned),
567    ]
568}
569
570fn terminal_event_type(name: &str, state: ProcessTerminalState) -> ProcessEventType {
571    ProcessEventType {
572        name: name.to_string(),
573        payload_schema: crate::LashSchema::any(),
574        semantics: ProcessEventSemanticsSpec {
575            terminal: Some(ProcessTerminalSpec {
576                state,
577                await_output: Some(ProcessValueSelector::Pointer("/await_output".to_string())),
578            }),
579            ..ProcessEventSemanticsSpec::default()
580        },
581    }
582}
lash_core/runtime/process/events.rs

lash_core/runtime/process/
events.rs
