lash_core/runtime/process/
materialization.rs

1use crate::plugin::PluginError;
2
3use super::events::{
4    AbandonEvidence, AbandonWriter, ProcessAwaitOutput, ProcessEventSemantics,
5    ProcessEventSemanticsSpec, ProcessTerminalSemantics, ProcessTerminalSpec, ProcessTerminalState,
6    ProcessValueSelector, ProcessWake, ProcessWakeDedupeKey, ProcessWakeSpec,
7};
8
9pub fn materialize_process_event_semantics(
10    process_id: &str,
11    sequence: u64,
12    payload: &serde_json::Value,
13    spec: &ProcessEventSemanticsSpec,
14) -> Result<ProcessEventSemantics, PluginError> {
15    materialize_event_semantics(process_id, sequence, payload, spec)
16}
17
18pub(super) fn materialize_event_semantics(
19    process_id: &str,
20    sequence: u64,
21    payload: &serde_json::Value,
22    spec: &ProcessEventSemanticsSpec,
23) -> Result<ProcessEventSemantics, PluginError> {
24    let terminal = spec
25        .terminal
26        .as_ref()
27        .map(|terminal| materialize_terminal_semantics(payload, terminal))
28        .transpose()?;
29    let wake = spec
30        .wake
31        .as_ref()
32        .map(|wake| materialize_wake(process_id, sequence, payload, wake))
33        .transpose()?
34        .flatten();
35    Ok(ProcessEventSemantics { terminal, wake })
36}
37
38fn materialize_terminal_semantics(
39    payload: &serde_json::Value,
40    terminal: &ProcessTerminalSpec,
41) -> Result<ProcessTerminalSemantics, PluginError> {
42    let await_output = match &terminal.await_output {
43        Some(selector) => {
44            let selected = select_value(payload, selector)?;
45            serde_json::from_value::<ProcessAwaitOutput>(selected.clone())
46                .unwrap_or_else(|_| selected_value_to_await_output(terminal.state, selected))
47        }
48        None if terminal.state == ProcessTerminalState::Completed => ProcessAwaitOutput::Success {
49            value: payload.clone(),
50            control: None,
51        },
52        None => {
53            return Err(PluginError::Session(
54                "failed or cancelled terminal events must declare await output".to_string(),
55            ));
56        }
57    };
58    Ok(ProcessTerminalSemantics {
59        state: terminal.state,
60        await_output,
61    })
62}
63
64fn selected_value_to_await_output(
65    state: ProcessTerminalState,
66    value: serde_json::Value,
67) -> ProcessAwaitOutput {
68    match state {
69        ProcessTerminalState::Completed => ProcessAwaitOutput::Success {
70            value,
71            control: None,
72        },
73        ProcessTerminalState::Failed => ProcessAwaitOutput::Failure {
74            class: crate::ToolFailureClass::Execution,
75            code: "process_failed".to_string(),
76            message: selector_value_to_string(&value),
77            raw: Some(value),
78            control: None,
79        },
80        ProcessTerminalState::Cancelled => ProcessAwaitOutput::Cancelled {
81            message: selector_value_to_string(&value),
82            raw: Some(value),
83            control: None,
84        },
85        // Reached only if a producer declares its own `Abandoned` terminal event
86        // and emits a raw value (not a serialized `ProcessAwaitOutput`); the
87        // sweep/drain path writes structured evidence through `complete_process`,
88        // which deserializes directly above. With no structured evidence to carry,
89        // synthesize a minimal owner-drain marker.
90        ProcessTerminalState::Abandoned => ProcessAwaitOutput::Abandoned {
91            evidence: Box::new(AbandonEvidence {
92                writer: AbandonWriter::OwnerDrain,
93                owner: None,
94                epoch_ms: 0,
95            }),
96            control: None,
97        },
98    }
99}
100
101fn materialize_wake(
102    process_id: &str,
103    sequence: u64,
104    payload: &serde_json::Value,
105    wake: &ProcessWakeSpec,
106) -> Result<Option<ProcessWake>, PluginError> {
107    if let Some(when) = &wake.when {
108        let selected = select_value(payload, when)?;
109        if !selector_value_is_truthy(&selected) {
110            return Ok(None);
111        }
112    }
113    let input = selector_value_to_string(&select_value(payload, &wake.input)?);
114    let dedupe_key = match &wake.dedupe_key {
115        ProcessWakeDedupeKey::EventIdentity => format!("{process_id}:{sequence}"),
116        ProcessWakeDedupeKey::Selector(selector) => {
117            selector_value_to_string(&select_value(payload, selector)?)
118        }
119        ProcessWakeDedupeKey::Const(value) => value.clone(),
120    };
121    Ok(Some(ProcessWake { input, dedupe_key }))
122}
123
124pub(super) fn select_value(
125    payload: &serde_json::Value,
126    selector: &ProcessValueSelector,
127) -> Result<serde_json::Value, PluginError> {
128    match selector {
129        ProcessValueSelector::Payload => Ok(payload.clone()),
130        ProcessValueSelector::Pointer(pointer) => {
131            payload.pointer(pointer).cloned().ok_or_else(|| {
132                PluginError::Session(format!("payload pointer `{pointer}` did not match"))
133            })
134        }
135        ProcessValueSelector::Const(value) => Ok(value.clone()),
136        ProcessValueSelector::Template { template, fields } => {
137            let mut rendered = template.clone();
138            for (name, selector) in fields {
139                let value = select_value(payload, selector)?;
140                rendered =
141                    rendered.replace(&format!("{{{name}}}"), &selector_value_to_string(&value));
142            }
143            Ok(serde_json::Value::String(rendered))
144        }
145        ProcessValueSelector::Present(pointer) => {
146            Ok(serde_json::Value::Bool(payload.pointer(pointer).is_some()))
147        }
148    }
149}
150
151fn selector_value_to_string(value: &serde_json::Value) -> String {
152    value
153        .as_str()
154        .map(ToOwned::to_owned)
155        .unwrap_or_else(|| value.to_string())
156}
157
158fn selector_value_is_truthy(value: &serde_json::Value) -> bool {
159    match value {
160        serde_json::Value::Null => false,
161        serde_json::Value::Bool(value) => *value,
162        serde_json::Value::String(value) => !value.is_empty(),
163        serde_json::Value::Array(value) => !value.is_empty(),
164        serde_json::Value::Object(value) => !value.is_empty(),
165        serde_json::Value::Number(_) => true,
166    }
167}
lash_core/runtime/process/materialization.rs

lash_core/runtime/process/
materialization.rs
