lash_core/runtime/process/
model.rs

1use std::collections::BTreeMap;
2use std::fmt;
3use std::sync::Arc;
4use std::sync::Mutex;
5
6use serde::{Deserialize, Serialize};
7
8use super::events::{
9    ProcessAwaitOutput, ProcessEventType, ProcessTerminalSemantics, ProcessTerminalState,
10    default_process_event_types,
11};
12use super::validation::{
13    ensure_core_event_types, process_registration_hash, validate_process_registration,
14};
15
16pub type ProcessId = String;
17
18/// Opaque position in a store's Process Change Feed.
19///
20/// The wrapped sequence is meaningful only to the registry backend that issued
21/// it. Backends expose constructors/accessors so external store implementations
22/// can persist and bind the position, but consumers should treat values as
23/// cursors, not comparable timestamps.
24#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
25#[serde(transparent)]
26pub struct ProcessChangeCursor(u64);
27
28impl ProcessChangeCursor {
29    pub fn initial() -> Self {
30        Self(0)
31    }
32
33    pub fn from_store_sequence(sequence: u64) -> Self {
34        Self(sequence)
35    }
36
37    pub fn store_sequence(self) -> u64 {
38        self.0
39    }
40}
41
42#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
43#[serde(transparent)]
44pub struct SessionScopeId(String);
45
46impl SessionScopeId {
47    pub fn new(value: impl Into<String>) -> Self {
48        Self(value.into())
49    }
50
51    pub fn as_str(&self) -> &str {
52        &self.0
53    }
54}
55
56impl fmt::Display for SessionScopeId {
57    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
58        formatter.write_str(&self.0)
59    }
60}
61
62impl From<String> for SessionScopeId {
63    fn from(value: String) -> Self {
64        Self::new(value)
65    }
66}
67
68impl From<&str> for SessionScopeId {
69    fn from(value: &str) -> Self {
70        Self::new(value)
71    }
72}
73
74/// Durable executable input for a process.
75///
76/// `ToolCall`, `SessionTurn`, and `External` are kernel process primitives:
77/// core owns their durable representation and execution semantics because they
78/// are how the runtime coordinates tools, child sessions, and externally
79/// completed work. `Engine` is the extension point for deployment-specific
80/// process runtimes; those rows require a matching [`crate::ProcessEngine`] in
81/// the host's process engine registry.
82#[derive(Debug, Serialize, Deserialize)]
83#[serde(tag = "type", rename_all = "snake_case")]
84pub enum ProcessInput {
85    ToolCall {
86        call: crate::PreparedToolCall,
87    },
88    Engine {
89        kind: String,
90        #[serde(default)]
91        payload: serde_json::Value,
92    },
93    SessionTurn {
94        create_request: Box<crate::SessionCreateRequest>,
95        turn_input: Box<crate::TurnInput>,
96        output_contract: crate::ToolOutputContract,
97    },
98    External {
99        #[serde(default)]
100        metadata: serde_json::Value,
101    },
102}
103
104impl Clone for ProcessInput {
105    fn clone(&self) -> Self {
106        match self {
107            Self::ToolCall { call } => Self::ToolCall { call: call.clone() },
108            Self::Engine { kind, payload } => Self::Engine {
109                kind: kind.clone(),
110                payload: payload.clone(),
111            },
112            Self::SessionTurn {
113                create_request,
114                turn_input,
115                output_contract,
116            } => Self::SessionTurn {
117                create_request: create_request.clone(),
118                turn_input: turn_input.clone(),
119                output_contract: output_contract.clone(),
120            },
121            Self::External { metadata } => Self::External {
122                metadata: metadata.clone(),
123            },
124        }
125    }
126}
127
128impl PartialEq for ProcessInput {
129    fn eq(&self, other: &Self) -> bool {
130        serde_json::to_value(self).ok() == serde_json::to_value(other).ok()
131    }
132}
133
134impl ProcessInput {
135    pub fn engine_kind(&self) -> &'static str {
136        match self {
137            Self::ToolCall { .. } => "tool",
138            Self::Engine { .. } => "engine",
139            Self::SessionTurn { .. } => "session_turn",
140            Self::External { .. } => "external",
141        }
142    }
143
144    pub fn engine_specific_kind(&self) -> Option<&str> {
145        match self {
146            Self::Engine { kind, .. } => Some(kind.as_str()),
147            _ => None,
148        }
149    }
150}
151
152/// Producer-declared contract stating what recovery may do with a process row
153/// after owner loss. Required at registration and applied mechanically by the
154/// sweep; never inferred at runtime. See ADR 0019.
155///
156/// There is deliberately no `Default` and no serde default: a producer that
157/// forgets to declare a disposition must fail to compile rather than silently
158/// inherit re-execution.
159#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
160#[serde(rename_all = "snake_case")]
161pub enum RecoveryDisposition {
162    /// Another owner may re-execute the work — the contract for journaled,
163    /// idempotent inputs (engine rows, session-turn rows).
164    Rerunnable,
165    /// The contract binds at first start: before any owner has begun execution
166    /// any worker may claim the row; once execution has started, no other owner
167    /// may ever re-execute it — abandonment is the only recovery.
168    OwnerBound,
169    /// Lash never executes the row at all. Closure comes from an external actor
170    /// calling `complete_process`, or from a reconciled Abandon Request.
171    ExternallyOwned,
172}
173
174/// Durable "execution started" fact: which owner began executing the row and
175/// when. The runner writes it under its live lease immediately before executing
176/// so the sweep can distinguish an OwnerBound row that has started (never
177/// re-run) from one that has not (runnable by anyone). First-writer-wins.
178#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
179pub struct ProcessStarted {
180    pub owner: crate::LeaseOwnerIdentity,
181    pub started_at_ms: u64,
182}
183
184/// Durable, non-terminal marker recording that a non-owner authorized
185/// abandonment without proof the owner is gone. The sweep reconciles it into
186/// [`ProcessTerminalState::Abandoned`](super::events::ProcessTerminalState::Abandoned)
187/// only once the row's lease has lapsed; the marker never terminates anything
188/// by itself and is visible to observers while pending. See ADR 0019.
189#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
190pub struct AbandonRequest {
191    pub requested_by: String,
192    pub requested_at_ms: u64,
193    #[serde(default, skip_serializing_if = "Option::is_none")]
194    pub reason: Option<String>,
195}
196
197#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
198#[serde(transparent)]
199pub struct ProcessExecutionEnvRef(String);
200
201impl ProcessExecutionEnvRef {
202    pub fn new(value: impl Into<String>) -> Self {
203        Self(value.into())
204    }
205
206    pub fn as_str(&self) -> &str {
207        &self.0
208    }
209}
210
211impl fmt::Display for ProcessExecutionEnvRef {
212    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
213        formatter.write_str(&self.0)
214    }
215}
216
217#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
218#[serde(deny_unknown_fields)]
219pub struct ProcessExecutionEnvSpec {
220    #[serde(default)]
221    pub plugin_options: crate::PluginOptions,
222    #[serde(default)]
223    pub policy: crate::SessionPolicy,
224}
225
226impl ProcessExecutionEnvSpec {
227    pub fn new(plugin_options: crate::PluginOptions, policy: crate::SessionPolicy) -> Self {
228        Self {
229            plugin_options,
230            policy,
231        }
232    }
233
234    pub fn stable_ref(&self) -> Result<ProcessExecutionEnvRef, serde_json::Error> {
235        crate::stable_hash::stable_json_sha256_hex(self)
236            .map(|hash| ProcessExecutionEnvRef::new(format!("process-env:sha256:{hash}")))
237    }
238
239    pub fn to_store_bytes(&self) -> Result<Vec<u8>, serde_json::Error> {
240        serde_json::to_vec(self)
241    }
242
243    pub fn from_store_bytes(bytes: &[u8]) -> Result<Self, serde_json::Error> {
244        serde_json::from_slice(bytes)
245    }
246}
247
248#[async_trait::async_trait]
249pub trait ProcessExecutionEnvStore: Send + Sync {
250    fn durability_tier(&self) -> crate::DurabilityTier {
251        crate::DurabilityTier::Inline
252    }
253
254    async fn put_process_execution_env(
255        &self,
256        env_ref: &ProcessExecutionEnvRef,
257        bytes: &[u8],
258    ) -> Result<(), crate::PluginError>;
259
260    async fn get_process_execution_env(
261        &self,
262        env_ref: &ProcessExecutionEnvRef,
263    ) -> Result<Option<Vec<u8>>, crate::PluginError>;
264}
265
266#[derive(Default)]
267pub struct InMemoryProcessExecutionEnvStore {
268    envs: Mutex<BTreeMap<String, Vec<u8>>>,
269}
270
271impl InMemoryProcessExecutionEnvStore {
272    pub fn new() -> Self {
273        Self::default()
274    }
275}
276
277#[async_trait::async_trait]
278impl ProcessExecutionEnvStore for InMemoryProcessExecutionEnvStore {
279    async fn put_process_execution_env(
280        &self,
281        env_ref: &ProcessExecutionEnvRef,
282        bytes: &[u8],
283    ) -> Result<(), crate::PluginError> {
284        self.envs
285            .lock()
286            .map_err(|_| {
287                crate::PluginError::Session("process execution env store lock poisoned".to_string())
288            })?
289            .insert(env_ref.as_str().to_string(), bytes.to_vec());
290        Ok(())
291    }
292
293    async fn get_process_execution_env(
294        &self,
295        env_ref: &ProcessExecutionEnvRef,
296    ) -> Result<Option<Vec<u8>>, crate::PluginError> {
297        Ok(self
298            .envs
299            .lock()
300            .map_err(|_| {
301                crate::PluginError::Session("process execution env store lock poisoned".to_string())
302            })?
303            .get(env_ref.as_str())
304            .cloned())
305    }
306}
307
308pub async fn persist_process_execution_env(
309    env_store: &dyn ProcessExecutionEnvStore,
310    spec: &ProcessExecutionEnvSpec,
311) -> Result<ProcessExecutionEnvRef, crate::PluginError> {
312    let env_ref = spec.stable_ref().map_err(|err| {
313        crate::PluginError::Session(format!("failed to hash process execution env: {err}"))
314    })?;
315    let bytes = spec.to_store_bytes().map_err(|err| {
316        crate::PluginError::Session(format!("failed to encode process execution env: {err}"))
317    })?;
318    env_store
319        .put_process_execution_env(&env_ref, &bytes)
320        .await?;
321    Ok(env_ref)
322}
323
324pub async fn load_process_execution_env(
325    env_store: &dyn ProcessExecutionEnvStore,
326    env_ref: &ProcessExecutionEnvRef,
327) -> Result<ProcessExecutionEnvSpec, crate::PluginError> {
328    let bytes = env_store
329        .get_process_execution_env(env_ref)
330        .await?
331        .ok_or_else(|| {
332            crate::PluginError::Session(format!("missing process execution env `{env_ref}`"))
333        })?;
334    ProcessExecutionEnvSpec::from_store_bytes(&bytes).map_err(|err| {
335        crate::PluginError::Session(format!(
336            "failed to decode process execution env `{env_ref}`: {err}"
337        ))
338    })
339}
340
341/// Execution-local context for process runners. Durable edges live on the
342/// process record.
343#[derive(Clone, Debug, Default, Serialize, Deserialize)]
344pub struct ProcessExecutionContext {
345    #[serde(default, skip_serializing_if = "Option::is_none")]
346    pub causal_invocation: Option<crate::RuntimeInvocation>,
347}
348
349impl ProcessExecutionContext {
350    pub fn with_causal_invocation(mut self, invocation: Option<crate::RuntimeInvocation>) -> Self {
351        self.causal_invocation = invocation;
352        self
353    }
354
355    pub fn is_empty(&self) -> bool {
356        self.causal_invocation.is_none()
357    }
358}
359
360#[derive(Clone)]
361pub struct ProcessOpScope<'scope> {
362    pub(crate) parent_invocation: Option<crate::RuntimeInvocation>,
363    pub(crate) effect_controller: crate::runtime::RuntimeEffectControllerHandle<'scope>,
364    pub(crate) agent_frame_id: Option<crate::AgentFrameId>,
365    pub(crate) target_agent_frame_id: Option<crate::AgentFrameId>,
366}
367
368impl<'scope> ProcessOpScope<'scope> {
369    pub fn new(scoped_effect_controller: crate::ScopedEffectController<'scope>) -> Self {
370        Self {
371            parent_invocation: None,
372            effect_controller: crate::runtime::RuntimeEffectControllerHandle::borrowed(
373                scoped_effect_controller,
374            ),
375            agent_frame_id: None,
376            target_agent_frame_id: None,
377        }
378    }
379
380    pub fn with_parent_invocation(
381        mut self,
382        parent_invocation: Option<crate::RuntimeInvocation>,
383    ) -> Self {
384        self.parent_invocation = parent_invocation;
385        self
386    }
387
388    pub fn with_agent_frame_id(mut self, agent_frame_id: Option<crate::AgentFrameId>) -> Self {
389        self.agent_frame_id = agent_frame_id;
390        self
391    }
392
393    pub fn with_target_agent_frame_id(
394        mut self,
395        agent_frame_id: Option<crate::AgentFrameId>,
396    ) -> Self {
397        self.target_agent_frame_id = agent_frame_id;
398        self
399    }
400
401    pub fn agent_frame_id(&self) -> Option<&str> {
402        self.agent_frame_id.as_deref()
403    }
404
405    pub fn target_agent_frame_id(&self) -> Option<&str> {
406        self.target_agent_frame_id.as_deref()
407    }
408
409    pub(crate) fn controller(&self) -> &dyn crate::RuntimeEffectController {
410        self.effect_controller.controller()
411    }
412}
413
414#[derive(Clone, Debug, Default)]
415pub struct ProcessStartOptions {
416    pub descriptor: Option<ProcessHandleDescriptor>,
417    /// Runtime-internal spawn provenance override. Set by process execution
418    /// contexts so children started *by a process* inherit the parent's
419    /// originator and wake target instead of being stamped with the ephemeral
420    /// execution scope. `None` means the session start path stamps the
421    /// creating session (the in-session meaning of "start"). This rides
422    /// options — not the request — so in-session callers cannot forge
423    /// provenance through the session surface.
424    pub spawn_provenance: Option<ProcessSpawnProvenance>,
425}
426
427/// Provenance a process-run context hands to its children: the chain's
428/// originator and the chain's wake target. Mirrors the trigger fire path,
429/// where the spawned process inherits the registrant and the grant is derived
430/// from the wake target.
431#[derive(Clone, Debug, PartialEq, Eq)]
432pub struct ProcessSpawnProvenance {
433    pub originator: ProcessOriginator,
434    pub wake_target: Option<SessionScope>,
435}
436
437impl ProcessStartOptions {
438    pub fn new() -> Self {
439        Self::default()
440    }
441
442    pub fn with_descriptor(mut self, descriptor: ProcessHandleDescriptor) -> Self {
443        self.descriptor = Some(descriptor);
444        self
445    }
446
447    pub fn with_optional_descriptor(mut self, descriptor: Option<ProcessHandleDescriptor>) -> Self {
448        self.descriptor = descriptor;
449        self
450    }
451
452    pub fn with_spawn_provenance(mut self, spawn_provenance: ProcessSpawnProvenance) -> Self {
453        self.spawn_provenance = Some(spawn_provenance);
454        self
455    }
456
457    pub fn execution_context(&self, scope: &ProcessOpScope<'_>) -> ProcessExecutionContext {
458        ProcessExecutionContext {
459            causal_invocation: scope.parent_invocation.clone(),
460        }
461    }
462}
463
464/// Public host-facing request for starting a visible process handle.
465#[derive(Clone, Debug, Serialize, Deserialize)]
466pub struct ProcessStartRequest {
467    pub id: ProcessId,
468    pub input: ProcessInput,
469    pub disposition: RecoveryDisposition,
470    #[serde(default, skip_serializing_if = "Option::is_none")]
471    pub env_spec: Option<ProcessExecutionEnvSpec>,
472    pub originator: ProcessOriginator,
473    #[serde(default, skip_serializing_if = "Option::is_none")]
474    pub wake_target: Option<SessionScope>,
475    #[serde(default, skip_serializing_if = "Option::is_none")]
476    pub grant: Option<ProcessStartGrant>,
477    #[serde(default)]
478    pub event_types: Vec<ProcessEventType>,
479}
480
481impl ProcessStartRequest {
482    pub fn new(
483        id: impl Into<ProcessId>,
484        input: ProcessInput,
485        disposition: RecoveryDisposition,
486        originator: ProcessOriginator,
487    ) -> Self {
488        Self {
489            id: id.into(),
490            input,
491            disposition,
492            env_spec: None,
493            originator,
494            wake_target: None,
495            grant: None,
496            event_types: default_process_event_types(),
497        }
498    }
499
500    /// External placeholder start: `ProcessInput::External` is always
501    /// [`RecoveryDisposition::ExternallyOwned`] — lash never executes it.
502    pub fn external(
503        id: impl Into<ProcessId>,
504        originator: ProcessOriginator,
505        metadata: serde_json::Value,
506    ) -> Self {
507        Self::new(
508            id,
509            ProcessInput::External { metadata },
510            RecoveryDisposition::ExternallyOwned,
511            originator,
512        )
513    }
514
515    pub fn with_env_spec(mut self, env_spec: ProcessExecutionEnvSpec) -> Self {
516        self.env_spec = Some(env_spec);
517        self
518    }
519
520    pub fn with_wake_target(mut self, wake_target: Option<SessionScope>) -> Self {
521        self.wake_target = wake_target;
522        self
523    }
524
525    pub fn with_grant(mut self, grant: Option<ProcessStartGrant>) -> Self {
526        self.grant = grant;
527        self
528    }
529
530    pub fn with_event_types(
531        mut self,
532        event_types: impl IntoIterator<Item = ProcessEventType>,
533    ) -> Self {
534        self.event_types = event_types.into_iter().collect();
535        self
536    }
537
538    pub fn with_extra_event_types(
539        mut self,
540        event_types: impl IntoIterator<Item = ProcessEventType>,
541    ) -> Self {
542        self.event_types.extend(event_types);
543        self
544    }
545
546    pub fn into_registration(self, env_ref: Option<ProcessExecutionEnvRef>) -> ProcessRegistration {
547        ProcessRegistration::new(
548            self.id,
549            self.input,
550            self.disposition,
551            ProcessProvenance::new(self.originator),
552        )
553        .with_event_types(self.event_types)
554        .with_execution_env_ref(env_ref)
555        .with_wake_target(self.wake_target)
556    }
557}
558
559#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
560pub struct SessionScope {
561    pub session_id: String,
562    #[serde(default, skip_serializing_if = "Option::is_none")]
563    pub agent_frame_id: Option<crate::AgentFrameId>,
564}
565
566#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
567pub struct ProcessProvenance {
568    pub originator: ProcessOriginator,
569    #[serde(default, skip_serializing_if = "Option::is_none")]
570    pub caused_by: Option<crate::CausalRef>,
571}
572
573impl ProcessProvenance {
574    pub fn new(originator: ProcessOriginator) -> Self {
575        Self {
576            originator,
577            caused_by: None,
578        }
579    }
580
581    pub fn host() -> Self {
582        Self::new(ProcessOriginator::host())
583    }
584
585    pub fn session(scope: SessionScope) -> Self {
586        Self::new(ProcessOriginator::session(scope))
587    }
588
589    pub fn with_caused_by(mut self, caused_by: Option<crate::CausalRef>) -> Self {
590        self.caused_by = caused_by;
591        self
592    }
593}
594
595#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
596#[serde(tag = "type", rename_all = "snake_case")]
597pub enum ProcessOriginator {
598    Host {
599        #[serde(default, skip_serializing_if = "Option::is_none")]
600        scope: Option<String>,
601    },
602    Session {
603        scope: SessionScope,
604    },
605}
606
607impl ProcessOriginator {
608    pub fn host() -> Self {
609        Self::Host { scope: None }
610    }
611
612    pub fn host_scoped(scope: impl Into<String>) -> Self {
613        Self::Host {
614            scope: Some(scope.into()),
615        }
616    }
617
618    pub fn session(scope: SessionScope) -> Self {
619        Self::Session { scope }
620    }
621
622    pub fn scope_id(&self) -> String {
623        match self {
624            Self::Host { scope } => scope
625                .as_ref()
626                .map(|scope| format!("host:{scope}"))
627                .unwrap_or_else(|| "host".to_string()),
628            Self::Session { scope } => scope.id().to_string(),
629        }
630    }
631}
632
633impl SessionScope {
634    pub fn new(session_id: impl Into<String>) -> Self {
635        Self {
636            session_id: session_id.into(),
637            agent_frame_id: None,
638        }
639    }
640
641    pub fn for_agent_frame(
642        session_id: impl Into<String>,
643        agent_frame_id: impl Into<crate::AgentFrameId>,
644    ) -> Self {
645        Self {
646            session_id: session_id.into(),
647            agent_frame_id: Some(agent_frame_id.into()),
648        }
649    }
650
651    pub fn id(&self) -> SessionScopeId {
652        match self.agent_frame_id.as_deref() {
653            Some(frame_id) if !frame_id.is_empty() => {
654                SessionScopeId::new(format!("session:{}/frame:{frame_id}", self.session_id))
655            }
656            _ => SessionScopeId::new(format!("session:{}", self.session_id)),
657        }
658    }
659
660    pub fn is_empty(&self) -> bool {
661        self.session_id.is_empty()
662    }
663}
664
665/// Serializable process spec used to start or recover a runtime process.
666#[derive(Debug, Serialize, Deserialize)]
667pub struct ProcessRegistration {
668    pub id: ProcessId,
669    pub input: Arc<ProcessInput>,
670    pub disposition: RecoveryDisposition,
671    pub identity: ProcessIdentity,
672    #[serde(default)]
673    pub event_types: Vec<ProcessEventType>,
674    pub provenance: ProcessProvenance,
675    #[serde(default, skip_serializing_if = "Option::is_none")]
676    pub env_ref: Option<ProcessExecutionEnvRef>,
677    #[serde(default, skip_serializing_if = "Option::is_none")]
678    pub wake_target: Option<SessionScope>,
679}
680
681impl Clone for ProcessRegistration {
682    fn clone(&self) -> Self {
683        Self {
684            id: self.id.clone(),
685            input: Arc::clone(&self.input),
686            disposition: self.disposition,
687            identity: self.identity.clone(),
688            event_types: self.event_types.clone(),
689            provenance: self.provenance.clone(),
690            env_ref: self.env_ref.clone(),
691            wake_target: self.wake_target.clone(),
692        }
693    }
694}
695
696impl ProcessRegistration {
697    pub fn new(
698        id: impl Into<ProcessId>,
699        input: ProcessInput,
700        disposition: RecoveryDisposition,
701        provenance: ProcessProvenance,
702    ) -> Self {
703        let identity = ProcessIdentity::from_process_input(&input);
704        Self {
705            id: id.into(),
706            input: Arc::new(input),
707            disposition,
708            identity,
709            event_types: default_process_event_types(),
710            provenance,
711            env_ref: None,
712            wake_target: None,
713        }
714    }
715
716    pub(crate) fn session_start_draft(
717        id: impl Into<ProcessId>,
718        input: ProcessInput,
719        disposition: RecoveryDisposition,
720    ) -> Self {
721        Self::new(id, input, disposition, ProcessProvenance::host())
722    }
723
724    pub fn with_process_provenance(mut self, provenance: ProcessProvenance) -> Self {
725        self.provenance = provenance;
726        self
727    }
728
729    pub fn with_execution_env_ref(mut self, env_ref: Option<ProcessExecutionEnvRef>) -> Self {
730        self.env_ref = env_ref;
731        self
732    }
733
734    pub fn with_wake_target(mut self, wake_target: Option<SessionScope>) -> Self {
735        self.wake_target = wake_target;
736        self
737    }
738
739    pub fn with_identity(mut self, identity: ProcessIdentity) -> Self {
740        self.identity = identity;
741        self
742    }
743
744    pub fn with_event_types(
745        mut self,
746        event_types: impl IntoIterator<Item = ProcessEventType>,
747    ) -> Self {
748        self.event_types = event_types.into_iter().collect();
749        self
750    }
751
752    pub fn with_extra_event_types(
753        mut self,
754        event_types: impl IntoIterator<Item = ProcessEventType>,
755    ) -> Self {
756        self.event_types.extend(event_types);
757        self
758    }
759}
760
761#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
762#[serde(tag = "state", rename_all = "snake_case")]
763pub enum ProcessStatus {
764    #[default]
765    Running,
766    Completed {
767        await_output: ProcessAwaitOutput,
768    },
769    Failed {
770        await_output: ProcessAwaitOutput,
771    },
772    Cancelled {
773        await_output: ProcessAwaitOutput,
774    },
775    Abandoned {
776        await_output: ProcessAwaitOutput,
777    },
778}
779
780impl ProcessStatus {
781    pub fn from_terminal(terminal: ProcessTerminalSemantics) -> Self {
782        match terminal.state {
783            ProcessTerminalState::Completed => Self::Completed {
784                await_output: terminal.await_output,
785            },
786            ProcessTerminalState::Failed => Self::Failed {
787                await_output: terminal.await_output,
788            },
789            ProcessTerminalState::Cancelled => Self::Cancelled {
790                await_output: terminal.await_output,
791            },
792            ProcessTerminalState::Abandoned => Self::Abandoned {
793                await_output: terminal.await_output,
794            },
795        }
796    }
797
798    pub fn is_terminal(&self) -> bool {
799        !matches!(self, Self::Running)
800    }
801
802    pub fn label(&self) -> &'static str {
803        match self {
804            Self::Running => "running",
805            Self::Completed { .. } => "completed",
806            Self::Failed { .. } => "failed",
807            Self::Cancelled { .. } => "cancelled",
808            Self::Abandoned { .. } => "abandoned",
809        }
810    }
811
812    pub fn terminal_state(&self) -> Option<ProcessTerminalState> {
813        match self {
814            Self::Running => None,
815            Self::Completed { .. } => Some(ProcessTerminalState::Completed),
816            Self::Failed { .. } => Some(ProcessTerminalState::Failed),
817            Self::Cancelled { .. } => Some(ProcessTerminalState::Cancelled),
818            Self::Abandoned { .. } => Some(ProcessTerminalState::Abandoned),
819        }
820    }
821
822    pub fn await_output(&self) -> Option<&ProcessAwaitOutput> {
823        match self {
824            Self::Running => None,
825            Self::Completed { await_output }
826            | Self::Failed { await_output }
827            | Self::Cancelled { await_output }
828            | Self::Abandoned { await_output } => Some(await_output),
829        }
830    }
831
832    pub fn terminal_semantics(&self) -> Option<ProcessTerminalSemantics> {
833        Some(ProcessTerminalSemantics {
834            state: self.terminal_state()?,
835            await_output: self.await_output()?.clone(),
836        })
837    }
838}
839
840/// Durable process row. Session-visible addressability lives in
841/// [`ProcessHandleGrant`], not in the process record.
842#[derive(Clone, Debug, Serialize, Deserialize)]
843pub struct ProcessRecord {
844    pub id: ProcessId,
845    pub registration_hash: String,
846    pub input: Arc<ProcessInput>,
847    /// Declared recovery contract. Required with no serde default: pre-column
848    /// durable rows cannot deserialize and are handled by each store's schema
849    /// version bump (reject-and-recreate), never by an API/serde default.
850    pub disposition: RecoveryDisposition,
851    pub identity: ProcessIdentity,
852    #[serde(default)]
853    pub event_types: Vec<ProcessEventType>,
854    pub provenance: ProcessProvenance,
855    #[serde(default, skip_serializing_if = "Option::is_none")]
856    pub env_ref: Option<ProcessExecutionEnvRef>,
857    #[serde(default, skip_serializing_if = "Option::is_none")]
858    pub wake_target: Option<SessionScope>,
859    #[serde(default)]
860    pub created_at_ms: u64,
861    #[serde(default)]
862    pub updated_at_ms: u64,
863    #[serde(default, skip_serializing_if = "Option::is_none")]
864    pub external_ref: Option<ProcessExternalRef>,
865    /// Durable, lease-fenced execution-started fact (ADR 0019). `None` until a
866    /// runner records it immediately before executing. Boxed so these
867    /// usually-absent facts do not enlarge the pervasive `ProcessRecord` that
868    /// flows through the runtime; serde treats `Option<Box<T>>` identically to
869    /// `Option<T>`, so the persisted JSON is unchanged.
870    #[serde(default, skip_serializing_if = "Option::is_none")]
871    pub first_started: Option<Box<ProcessStarted>>,
872    /// Pending Abandon Request the sweep reconciles once the lease lapses.
873    #[serde(default, skip_serializing_if = "Option::is_none")]
874    pub abandon_request: Option<Box<AbandonRequest>>,
875    #[serde(default, skip_serializing_if = "Option::is_none")]
876    pub wait: Option<WaitState>,
877    #[serde(default)]
878    pub status: ProcessStatus,
879}
880
881#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
882pub struct WaitState {
883    pub kind: WaitKind,
884    pub since_ms: u64,
885}
886
887#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
888#[serde(tag = "kind", rename_all = "snake_case")]
889pub enum WaitKind {
890    Signal {
891        name: String,
892        event_type: String,
893        key: String,
894        ordinal: u64,
895    },
896}
897
898impl ProcessRecord {
899    pub fn from_registration(registration: ProcessRegistration) -> Self {
900        Self::from_registration_with_clock(registration, &crate::SystemClock)
901    }
902
903    pub fn from_registration_with_clock(
904        mut registration: ProcessRegistration,
905        clock: &dyn crate::Clock,
906    ) -> Self {
907        ensure_core_event_types(&mut registration);
908        validate_process_registration(&registration)
909            .expect("process registration should be valid before record construction");
910        let registration_hash = process_registration_hash(&registration)
911            .expect("process registration should hash before record construction");
912        Self::from_prepared_registration(registration, registration_hash, clock.timestamp_ms())
913    }
914
915    pub fn from_prepared_registration(
916        registration: ProcessRegistration,
917        registration_hash: String,
918        now_ms: u64,
919    ) -> Self {
920        Self {
921            id: registration.id,
922            registration_hash,
923            input: registration.input,
924            disposition: registration.disposition,
925            identity: registration.identity,
926            event_types: registration.event_types,
927            provenance: registration.provenance,
928            env_ref: registration.env_ref,
929            wake_target: registration.wake_target,
930            created_at_ms: now_ms,
931            updated_at_ms: now_ms,
932            external_ref: None,
933            first_started: None,
934            abandon_request: None,
935            wait: None,
936            status: ProcessStatus::Running,
937        }
938    }
939
940    pub fn is_terminal(&self) -> bool {
941        self.status.is_terminal()
942    }
943
944    pub fn clear_wake_target_for_session(&mut self, session_id: &str) -> bool {
945        self.clear_wake_target_for_session_with_clock(session_id, &crate::SystemClock)
946    }
947
948    pub fn clear_wake_target_for_session_with_clock(
949        &mut self,
950        session_id: &str,
951        clock: &dyn crate::Clock,
952    ) -> bool {
953        let should_clear = self
954            .wake_target
955            .as_ref()
956            .is_some_and(|scope| scope.session_id == session_id);
957        if should_clear {
958            self.wake_target = None;
959            self.updated_at_ms = clock.timestamp_ms();
960        }
961        should_clear
962    }
963
964    pub fn originator_scope_id(&self) -> String {
965        self.provenance.originator.scope_id()
966    }
967}
968
969/// Canonical process identity stored alongside every durable process row.
970///
971/// `ProcessInput::Engine` keeps its payload opaque to core. Engines therefore
972/// publish their visible kind, display label, and definition identity at the
973/// registration boundary; list, summary, trigger, and observation paths read
974/// this durable field instead of decoding engine payload conventions.
975#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
976pub struct ProcessIdentity {
977    pub kind: String,
978    #[serde(default, skip_serializing_if = "Option::is_none")]
979    pub label: Option<String>,
980    #[serde(default, skip_serializing_if = "Option::is_none")]
981    pub definition: Option<serde_json::Value>,
982}
983
984impl ProcessIdentity {
985    pub fn new(kind: impl Into<String>) -> Self {
986        Self {
987            kind: kind.into(),
988            label: None,
989            definition: None,
990        }
991    }
992
993    pub fn with_label(mut self, label: Option<impl Into<String>>) -> Self {
994        self.label = label.map(Into::into);
995        self
996    }
997
998    pub fn with_definition(mut self, definition: Option<serde_json::Value>) -> Self {
999        self.definition = definition;
1000        self
1001    }
1002
1003    pub fn from_process_input(input: &ProcessInput) -> Self {
1004        match input {
1005            ProcessInput::ToolCall { call } => {
1006                Self::new("tool").with_label(Some(call.tool_name.clone()))
1007            }
1008            ProcessInput::Engine { kind, .. } => Self::new(kind.clone()),
1009            ProcessInput::SessionTurn { create_request, .. } => {
1010                let label = create_request
1011                    .subagent
1012                    .as_ref()
1013                    .map(|subagent| subagent.capability.clone())
1014                    .or_else(|| create_request.usage_source.clone())
1015                    .or_else(|| create_request.session_id.clone());
1016                Self::new("session_turn").with_label(label)
1017            }
1018            ProcessInput::External { metadata } => {
1019                let label = metadata
1020                    .get("label")
1021                    .or_else(|| metadata.get("name"))
1022                    .or_else(|| metadata.get("title"))
1023                    .and_then(serde_json::Value::as_str)
1024                    .map(str::to_string);
1025                Self::new("external").with_label(label)
1026            }
1027        }
1028    }
1029}
1030
1031/// Wire-format version stamped on every persisted [`ProcessLease`].
1032///
1033/// Bump when the on-wire shape of `ProcessLease` changes in a way that older
1034/// code cannot safely deserialize. Version 2 replaced the bare `owner_id`
1035/// string with a full [`LeaseOwnerIdentity`](crate::LeaseOwnerIdentity)
1036/// carrying incarnation and liveness metadata for fenced reclaim.
1037pub const PROCESS_LEASE_SCHEMA_VERSION: u32 = 2;
1038
1039/// Durable session stores owned exclusively by one process execution.
1040pub fn process_runtime_session_ids(process_id: &str) -> [String; 2] {
1041    [
1042        format!("process-env:{process_id}"),
1043        format!("process-session-turn:{process_id}"),
1044    ]
1045}
1046
1047/// Durable lease over a non-terminal background process.
1048///
1049/// The lease pair `(owner, lease_token)` plus `fencing_token` are how lash guarantees that
1050/// one non-terminal process is re-executed by exactly one worker at a time —
1051/// even after a crash, even across two workers that both sweep the same
1052/// registry for recoverable work. The durable backend
1053/// (`lash-sqlite-store`) uses these to serialize concurrent claims on the same
1054/// `process_id`; future distributed durable backends use the *same* fields to
1055/// coordinate workers that don't share a file system.
1056///
1057/// The owner is a full [`LeaseOwnerIdentity`](crate::LeaseOwnerIdentity):
1058/// its persisted liveness metadata is what lets a sweeping worker prove a
1059/// busy holder is *definitely dead* and reclaim the lease before the TTL
1060/// through [`ProcessRegistry::reclaim_process_lease`](super::ProcessRegistry::reclaim_process_lease),
1061/// mirroring the session execution lane.
1062///
1063/// **This is not single-process theatre.** The owner / fencing-token /
1064/// lease-token triple is the public contract that lets any backend detect and
1065/// reject stale writers. Treat it as load-bearing, not defensive.
1066#[derive(Clone, Debug, Serialize, Deserialize)]
1067pub struct ProcessLease {
1068    pub schema_version: u32,
1069    pub process_id: ProcessId,
1070    pub owner: crate::LeaseOwnerIdentity,
1071    pub lease_token: String,
1072    pub fencing_token: u64,
1073    pub claimed_at_epoch_ms: u64,
1074    pub expires_at_epoch_ms: u64,
1075}
1076
1077/// Outcome of claiming (or reclaiming) a [`ProcessLease`].
1078///
1079/// Mirrors [`SessionExecutionLeaseClaimOutcome`](crate::SessionExecutionLeaseClaimOutcome):
1080/// a busy outcome carries the observed holder so the claimant can assess its
1081/// liveness and perform a fenced reclaim on exactly the lease it observed.
1082#[derive(Clone, Debug, Serialize, Deserialize)]
1083pub enum ProcessLeaseClaimOutcome {
1084    Acquired(ProcessLease),
1085    Busy { holder: ProcessLease },
1086}
1087
1088impl ProcessLeaseClaimOutcome {
1089    pub fn acquired(self) -> Option<ProcessLease> {
1090        match self {
1091            Self::Acquired(lease) => Some(lease),
1092            Self::Busy { .. } => None,
1093        }
1094    }
1095}
1096
1097#[derive(Clone, Debug, Serialize, Deserialize)]
1098pub struct ProcessLeaseCompletion {
1099    pub process_id: ProcessId,
1100    pub lease_token: String,
1101}
1102
1103impl ProcessLeaseCompletion {
1104    pub fn from_lease(lease: &ProcessLease) -> Self {
1105        Self {
1106            process_id: lease.process_id.clone(),
1107            lease_token: lease.lease_token.clone(),
1108        }
1109    }
1110}
1111
1112/// Durable backend reference for background work accepted outside the local process.
1113#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq)]
1114pub struct ProcessExternalRef {
1115    pub backend: String,
1116    pub id: String,
1117    #[serde(default, skip_serializing_if = "Option::is_none")]
1118    pub metadata: Option<serde_json::Value>,
1119}
1120
1121#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
1122pub struct ProcessHandleDescriptor {
1123    #[serde(default, skip_serializing_if = "Option::is_none")]
1124    pub kind: Option<String>,
1125    #[serde(default, skip_serializing_if = "Option::is_none")]
1126    pub label: Option<String>,
1127}
1128
1129impl ProcessHandleDescriptor {
1130    pub fn new(kind: Option<impl Into<String>>, label: Option<impl Into<String>>) -> Self {
1131        Self {
1132            kind: kind.map(Into::into),
1133            label: label.map(Into::into),
1134        }
1135    }
1136}
1137
1138#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
1139pub struct ProcessHandleGrant {
1140    pub session_id: String,
1141    pub process_id: ProcessId,
1142    pub descriptor: ProcessHandleDescriptor,
1143}
1144
1145pub type ProcessHandleGrantEntry = (ProcessHandleGrant, ProcessRecord);
1146
1147#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
1148#[serde(rename_all = "snake_case")]
1149pub enum ProcessLifecycleStatus {
1150    #[default]
1151    Running,
1152    Completed,
1153    Failed,
1154    Cancelled,
1155    Abandoned,
1156}
1157
1158impl ProcessLifecycleStatus {
1159    pub fn label(self) -> &'static str {
1160        match self {
1161            Self::Running => "running",
1162            Self::Completed => "completed",
1163            Self::Failed => "failed",
1164            Self::Cancelled => "cancelled",
1165            Self::Abandoned => "abandoned",
1166        }
1167    }
1168
1169    pub fn is_terminal(self) -> bool {
1170        !matches!(self, Self::Running)
1171    }
1172
1173    pub fn terminal_state(self) -> Option<ProcessTerminalState> {
1174        match self {
1175            Self::Running => None,
1176            Self::Completed => Some(ProcessTerminalState::Completed),
1177            Self::Failed => Some(ProcessTerminalState::Failed),
1178            Self::Cancelled => Some(ProcessTerminalState::Cancelled),
1179            Self::Abandoned => Some(ProcessTerminalState::Abandoned),
1180        }
1181    }
1182}
1183
1184impl From<&ProcessStatus> for ProcessLifecycleStatus {
1185    fn from(status: &ProcessStatus) -> Self {
1186        match status {
1187            ProcessStatus::Running => Self::Running,
1188            ProcessStatus::Completed { .. } => Self::Completed,
1189            ProcessStatus::Failed { .. } => Self::Failed,
1190            ProcessStatus::Cancelled { .. } => Self::Cancelled,
1191            ProcessStatus::Abandoned { .. } => Self::Abandoned,
1192        }
1193    }
1194}
1195
1196impl From<ProcessStatus> for ProcessLifecycleStatus {
1197    fn from(status: ProcessStatus) -> Self {
1198        Self::from(&status)
1199    }
1200}
1201
1202#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
1203pub struct ProcessHandleSummary {
1204    #[serde(rename = "__handle__")]
1205    pub handle_type: String,
1206    pub id: ProcessId,
1207    pub process_id: ProcessId,
1208    pub descriptor: ProcessHandleDescriptor,
1209    #[serde(default, skip_serializing_if = "Option::is_none")]
1210    pub definition: Option<serde_json::Value>,
1211    pub status: ProcessLifecycleStatus,
1212}
1213
1214impl ProcessHandleSummary {
1215    pub fn new(
1216        process_id: impl Into<ProcessId>,
1217        descriptor: ProcessHandleDescriptor,
1218        status: ProcessLifecycleStatus,
1219    ) -> Self {
1220        let process_id = process_id.into();
1221        Self {
1222            handle_type: "process".to_string(),
1223            id: process_id.clone(),
1224            process_id,
1225            descriptor,
1226            definition: None,
1227            status,
1228        }
1229    }
1230
1231    pub fn with_definition(mut self, definition: Option<serde_json::Value>) -> Self {
1232        self.definition = definition;
1233        self
1234    }
1235
1236    pub fn from_grant_record(grant: ProcessHandleGrant, record: ProcessRecord) -> Self {
1237        let definition = record.identity.definition.clone();
1238        Self::new(
1239            record.id,
1240            grant.descriptor,
1241            ProcessLifecycleStatus::from(record.status),
1242        )
1243        .with_definition(definition)
1244    }
1245}
1246
1247impl From<ProcessHandleGrantEntry> for ProcessHandleSummary {
1248    fn from((grant, record): ProcessHandleGrantEntry) -> Self {
1249        Self::from_grant_record(grant, record)
1250    }
1251}
1252
1253#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
1254pub struct ProcessCancelSummary {
1255    pub process_id: ProcessId,
1256    pub status: ProcessLifecycleStatus,
1257}
1258
1259impl ProcessCancelSummary {
1260    pub fn from_record(record: ProcessRecord) -> Self {
1261        Self {
1262            process_id: record.id,
1263            status: ProcessLifecycleStatus::from(record.status),
1264        }
1265    }
1266}
1267
1268#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
1269pub enum ProcessStatusFilter {
1270    #[default]
1271    Running,
1272    Completed,
1273    Failed,
1274    Cancelled,
1275    Abandoned,
1276    Any,
1277}
1278
1279impl ProcessStatusFilter {
1280    pub fn decode(value: Option<&str>) -> Result<Self, String> {
1281        match value.unwrap_or("running") {
1282            "running" => Ok(Self::Running),
1283            "completed" => Ok(Self::Completed),
1284            "failed" => Ok(Self::Failed),
1285            "cancelled" => Ok(Self::Cancelled),
1286            "abandoned" => Ok(Self::Abandoned),
1287            "any" => Ok(Self::Any),
1288            other => Err(format!(
1289                "processes.list status must be `running`, `completed`, `failed`, `cancelled`, `abandoned`, or `any`, got `{other}`"
1290            )),
1291        }
1292    }
1293
1294    pub fn list_mode(self) -> ProcessListMode {
1295        match self {
1296            Self::Running => ProcessListMode::Live,
1297            Self::Completed | Self::Failed | Self::Cancelled | Self::Abandoned | Self::Any => {
1298                ProcessListMode::All
1299            }
1300        }
1301    }
1302
1303    pub fn matches(self, status: ProcessLifecycleStatus) -> bool {
1304        match self {
1305            Self::Running => status == ProcessLifecycleStatus::Running,
1306            Self::Completed => status == ProcessLifecycleStatus::Completed,
1307            Self::Failed => status == ProcessLifecycleStatus::Failed,
1308            Self::Cancelled => status == ProcessLifecycleStatus::Cancelled,
1309            Self::Abandoned => status == ProcessLifecycleStatus::Abandoned,
1310            Self::Any => true,
1311        }
1312    }
1313}
1314
1315#[derive(Clone, Debug, Default, PartialEq)]
1316pub struct ProcessListFilter {
1317    pub definition: Option<serde_json::Value>,
1318    pub status: ProcessStatusFilter,
1319    pub waiting: Option<bool>,
1320    pub originator_scope_id: Option<String>,
1321    pub identity_kind: Option<String>,
1322    pub identity_label: Option<String>,
1323    pub caused_by_occurrence_id: Option<String>,
1324    pub caused_by_subscription_id: Option<String>,
1325    /// Inclusive lower bound for `created_at_ms`; paired with
1326    /// `created_at_end_ms` this is a half-open `[start, end)` range.
1327    pub created_at_start_ms: Option<u64>,
1328    /// Exclusive upper bound for `created_at_ms`; paired with
1329    /// `created_at_start_ms` this is a half-open `[start, end)` range.
1330    pub created_at_end_ms: Option<u64>,
1331}
1332
1333impl ProcessListFilter {
1334    pub fn decode(args: &serde_json::Value) -> Result<Self, String> {
1335        let map = args
1336            .as_object()
1337            .ok_or_else(|| "processes.list expects a record of process filters".to_string())?;
1338        for key in map.keys() {
1339            match key.as_str() {
1340                "definition"
1341                | "status"
1342                | "waiting"
1343                | "originator_scope_id"
1344                | "identity_kind"
1345                | "identity_label"
1346                | "caused_by_occurrence_id"
1347                | "caused_by_subscription_id"
1348                | "created_at_start_ms"
1349                | "created_at_end_ms" => {}
1350                _ => return Err(format!("processes.list unknown filter `{key}`")),
1351            }
1352        }
1353        let definition = args.get("definition").cloned();
1354        let status =
1355            ProcessStatusFilter::decode(args.get("status").and_then(serde_json::Value::as_str))?;
1356        let waiting = args
1357            .get("waiting")
1358            .map(|value| {
1359                value
1360                    .as_bool()
1361                    .ok_or_else(|| "processes.list `waiting` filter must be a boolean".to_string())
1362            })
1363            .transpose()?;
1364        let originator_scope_id = optional_string_filter(args, "originator_scope_id")?;
1365        let identity_kind = optional_string_filter(args, "identity_kind")?;
1366        let identity_label = optional_string_filter(args, "identity_label")?;
1367        let caused_by_occurrence_id = optional_string_filter(args, "caused_by_occurrence_id")?;
1368        let caused_by_subscription_id = optional_string_filter(args, "caused_by_subscription_id")?;
1369        let created_at_start_ms = optional_u64_filter(args, "created_at_start_ms")?;
1370        let created_at_end_ms = optional_u64_filter(args, "created_at_end_ms")?;
1371        Ok(Self {
1372            definition,
1373            status,
1374            waiting,
1375            originator_scope_id,
1376            identity_kind,
1377            identity_label,
1378            caused_by_occurrence_id,
1379            caused_by_subscription_id,
1380            created_at_start_ms,
1381            created_at_end_ms,
1382        })
1383    }
1384
1385    pub fn list_mode(&self) -> ProcessListMode {
1386        self.status.list_mode()
1387    }
1388
1389    pub fn matches_entry(&self, entry: &ProcessHandleGrantEntry) -> bool {
1390        let (_grant, record) = entry;
1391        self.matches_record(record)
1392    }
1393
1394    pub fn matches_record(&self, record: &ProcessRecord) -> bool {
1395        let status = ProcessLifecycleStatus::from(&record.status);
1396        self.status.matches(status)
1397            && self
1398                .definition
1399                .as_ref()
1400                .is_none_or(|definition| record.identity.definition.as_ref() == Some(definition))
1401            && self
1402                .waiting
1403                .is_none_or(|waiting| record.wait.is_some() == waiting)
1404            && self
1405                .originator_scope_id
1406                .as_ref()
1407                .is_none_or(|scope_id| record.originator_scope_id() == scope_id.as_str())
1408            && self
1409                .identity_kind
1410                .as_ref()
1411                .is_none_or(|kind| record.identity.kind.as_str() == kind.as_str())
1412            && self
1413                .identity_label
1414                .as_ref()
1415                .is_none_or(|label| record.identity.label.as_deref() == Some(label.as_str()))
1416            && self
1417                .caused_by_occurrence_id
1418                .as_ref()
1419                .is_none_or(|occurrence_id| caused_by_occurrence_matches(record, occurrence_id))
1420            && self
1421                .caused_by_subscription_id
1422                .as_ref()
1423                .is_none_or(|subscription_id| {
1424                    caused_by_subscription_matches(record, subscription_id)
1425                })
1426            && self
1427                .created_at_start_ms
1428                .is_none_or(|start_ms| record.created_at_ms >= start_ms)
1429            && self
1430                .created_at_end_ms
1431                .is_none_or(|end_ms| record.created_at_ms < end_ms)
1432    }
1433}
1434
1435fn optional_string_filter(args: &serde_json::Value, key: &str) -> Result<Option<String>, String> {
1436    args.get(key)
1437        .map(|value| {
1438            value
1439                .as_str()
1440                .map(str::to_string)
1441                .ok_or_else(|| format!("processes.list `{key}` filter must be a string"))
1442        })
1443        .transpose()
1444}
1445
1446fn optional_u64_filter(args: &serde_json::Value, key: &str) -> Result<Option<u64>, String> {
1447    args.get(key)
1448        .map(|value| {
1449            value
1450                .as_u64()
1451                .ok_or_else(|| format!("processes.list `{key}` filter must be an integer"))
1452        })
1453        .transpose()
1454}
1455
1456fn caused_by_occurrence_matches(record: &ProcessRecord, occurrence_id: &str) -> bool {
1457    matches!(
1458        record.provenance.caused_by.as_ref(),
1459        Some(crate::CausalRef::TriggerOccurrence { occurrence_id: actual, .. }) if actual == occurrence_id
1460    )
1461}
1462
1463fn caused_by_subscription_matches(record: &ProcessRecord, subscription_id: &str) -> bool {
1464    matches!(
1465        record.provenance.caused_by.as_ref(),
1466        Some(crate::CausalRef::TriggerOccurrence {
1467            subscription_id: Some(actual),
1468            ..
1469        }) if actual == subscription_id
1470    )
1471}
1472
1473#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
1474#[serde(rename_all = "snake_case")]
1475pub enum ProcessListMode {
1476    #[default]
1477    Live,
1478    All,
1479}
1480
1481impl ProcessListMode {
1482    pub fn as_str(self) -> &'static str {
1483        match self {
1484            Self::Live => "live",
1485            Self::All => "all",
1486        }
1487    }
1488}
1489
1490#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
1491pub struct ProcessStartGrant {
1492    pub session_scope: SessionScope,
1493    pub descriptor: ProcessHandleDescriptor,
1494}
1495
1496#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
1497pub struct ProcessSessionDeleteReport {
1498    pub session_id: String,
1499    pub revoked_handle_count: usize,
1500    pub deleted_wake_count: usize,
1501    pub orphaned_process_ids: Vec<String>,
1502    pub preserved_process_ids: Vec<String>,
1503}
1504
1505#[cfg(test)]
1506mod tests {
1507    use serde_json::json;
1508
1509    use super::*;
1510
1511    fn process_value(component: &str, pos: usize, name: &str) -> serde_json::Value {
1512        json!({
1513            "component": component,
1514            "pos": pos,
1515            "name": name,
1516        })
1517    }
1518
1519    fn engine_entry(
1520        process_id: &str,
1521        definition: serde_json::Value,
1522        process_name: &str,
1523        status: ProcessStatus,
1524    ) -> ProcessHandleGrantEntry {
1525        let mut record = ProcessRecord::from_registration(
1526            ProcessRegistration::new(
1527                process_id,
1528                ProcessInput::Engine {
1529                    kind: "test-engine".to_string(),
1530                    payload: json!({
1531                        "definition": definition.clone(),
1532                        "label": process_name,
1533                    }),
1534                },
1535                RecoveryDisposition::Rerunnable,
1536                ProcessProvenance::host(),
1537            )
1538            .with_identity(
1539                ProcessIdentity::new("test-engine")
1540                    .with_label(Some(process_name))
1541                    .with_definition(Some(definition)),
1542            )
1543            .with_execution_env_ref(Some(ProcessExecutionEnvRef::new(format!(
1544                "process-env:test:{process_id}"
1545            )))),
1546        );
1547        record.status = status;
1548        (
1549            ProcessHandleGrant {
1550                session_id: "session".to_string(),
1551                process_id: process_id.to_string(),
1552                descriptor: ProcessHandleDescriptor::new(Some("test-engine"), Some(process_name)),
1553            },
1554            record,
1555        )
1556    }
1557
1558    #[test]
1559    fn process_list_filter_matches_waiting_facet() {
1560        let process_ref = process_value("target", 0, "target");
1561        let mut waiting_entry = engine_entry(
1562            "waiting",
1563            process_ref.clone(),
1564            "target",
1565            ProcessStatus::Running,
1566        );
1567        waiting_entry.1.wait = Some(WaitState {
1568            since_ms: 42,
1569            kind: WaitKind::Signal {
1570                name: "ready".to_string(),
1571                event_type: "signal.ready".to_string(),
1572                key: "process:waiting:signal.ready:1".to_string(),
1573                ordinal: 1,
1574            },
1575        });
1576        let idle_entry = engine_entry("idle", process_ref, "target", ProcessStatus::Running);
1577        let waiting_filter =
1578            ProcessListFilter::decode(&json!({ "waiting": true })).expect("decode waiting filter");
1579        let idle_filter =
1580            ProcessListFilter::decode(&json!({ "waiting": false })).expect("decode idle filter");
1581
1582        assert_eq!(waiting_filter.list_mode(), ProcessListMode::Live);
1583        assert!(waiting_filter.matches_entry(&waiting_entry));
1584        assert!(!waiting_filter.matches_entry(&idle_entry));
1585        assert!(!idle_filter.matches_entry(&waiting_entry));
1586        assert!(idle_filter.matches_entry(&idle_entry));
1587        assert!(
1588            ProcessListFilter::decode(&json!({ "waiting": "yes" }))
1589                .expect_err("invalid waiting filter")
1590                .contains("must be a boolean")
1591        );
1592    }
1593}
lash_core/runtime/process/model.rs

lash_core/runtime/process/
model.rs
