blut_graph_core/
execute.rs

1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3use alloc::boxed::Box;
4use alloc::collections::BTreeMap;
5use alloc::string::{String, ToString};
6use alloc::vec::Vec;
7use core::fmt;
8
9use crate::model::{
10    AuthorizedPlan, BufferId, CompiledNode, CompiledPlan, Effect, ExecutionRealm, GraphId,
11    ImplementationId, InputBinding, KernelId, NodeId, OutputBinding, PlanId, StepId,
12};
13
14#[derive(Clone, Debug, PartialEq, Eq)]
15pub enum ExecutionError {
16    UnknownKernel(KernelId),
17    MissingBuffer(BufferId),
18    MissingInvocation(crate::PortRef),
19    UnexpectedInvocation(crate::PortRef),
20    OutputArity {
21        kernel: KernelId,
22        expected: usize,
23        actual: usize,
24    },
25    KernelFailed {
26        kernel: KernelId,
27        failure: StructuredFailure,
28    },
29    UndeclaredFailure(KernelId, String),
30    UnsafeRetry(KernelId),
31    TransactionPrepare(String),
32    TransactionCommit(String),
33    InvalidGap(KernelId),
34    StatefulPlanUnsupported,
35}
36
37#[derive(Clone, Debug, PartialEq, Eq)]
38pub struct FailureEvidence {
39    pub semantic_type: String,
40    pub payload: Vec<u8>,
41}
42
43#[derive(Clone, Debug, PartialEq, Eq)]
44pub struct StructuredFailure {
45    pub domain: String,
46    pub code: String,
47    pub message: String,
48    pub retryable: bool,
49    pub evidence: Vec<FailureEvidence>,
50}
51
52impl ExecutionError {
53    const fn retryable(&self) -> bool {
54        matches!(
55            self,
56            Self::KernelFailed {
57                failure: StructuredFailure {
58                    retryable: true,
59                    ..
60                },
61                ..
62            }
63        )
64    }
65}
66
67impl fmt::Display for ExecutionError {
68    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
69        write!(f, "{self:?}")
70    }
71}
72
73#[cfg(feature = "std")]
74impl std::error::Error for ExecutionError {}
75
76#[derive(Clone, Debug, PartialEq, Eq)]
77pub struct ExecutionAttempt {
78    pub step: StepId,
79    pub semantic_nodes: Vec<NodeId>,
80    pub kernel: KernelId,
81    pub implementation_id: ImplementationId,
82    pub attempts: u32,
83    pub kernel_succeeded: bool,
84    pub completed: bool,
85}
86
87#[derive(Clone, Debug, PartialEq, Eq)]
88pub struct ExecutionReceipt {
89    pub invocation_id: [u8; 32],
90    pub graph_id: GraphId,
91    pub plan_id: PlanId,
92    pub realm: ExecutionRealm,
93    pub completed_nodes: Vec<NodeId>,
94    pub attempts: Vec<ExecutionAttempt>,
95    pub committed_transactions: Vec<String>,
96    pub gaps: Vec<GapReceipt>,
97}
98
99#[derive(Clone, Debug, PartialEq, Eq)]
100pub struct KernelGap {
101    pub output_index: u32,
102    pub offset: u64,
103    /// Known missing extent. `None` preserves unknown cardinality.
104    pub length: Option<u64>,
105    pub domain: String,
106    pub code: String,
107}
108
109#[derive(Clone, Debug, PartialEq, Eq)]
110pub struct GapReceipt {
111    pub step: StepId,
112    pub semantic_nodes: Vec<NodeId>,
113    pub gap: KernelGap,
114}
115
116#[derive(Clone, Debug, PartialEq, Eq)]
117pub struct KernelExecution<V> {
118    pub outputs: Vec<V>,
119    pub gaps: Vec<KernelGap>,
120}
121
122#[derive(Clone, Debug, PartialEq, Eq)]
123pub struct ExecutionFailure {
124    pub error: ExecutionError,
125    pub receipt: ExecutionReceipt,
126}
127
128#[derive(Clone, Debug, PartialEq, Eq)]
129pub struct ExecutionResult<V> {
130    pub terminal_values: BTreeMap<NodeId, Vec<V>>,
131    pub receipt: ExecutionReceipt,
132}
133
134pub trait KernelExecutor {
135    type Value;
136
137    /// Execute one attempt from immutable inputs and return one value per
138    /// physical output buffer. When a node has no connected output buffer,
139    /// returned values are terminal invocation outputs.
140    fn execute(
141        &mut self,
142        node: &CompiledNode,
143        inputs: &[Option<&Self::Value>],
144    ) -> Result<Vec<Self::Value>, ExecutionError>;
145
146    /// Gap-aware execution hook. Existing atomic kernels implement only
147    /// `execute`; explicitly partial kernels override this method and return
148    /// structured gaps alongside their ordinary output records.
149    fn execute_with_gaps(
150        &mut self,
151        node: &CompiledNode,
152        inputs: &[Option<&Self::Value>],
153    ) -> Result<KernelExecution<Self::Value>, ExecutionError> {
154        self.execute(node, inputs).map(|outputs| KernelExecution {
155            outputs,
156            gaps: Vec::new(),
157        })
158    }
159}
160
161pub trait TransactionalSink {
162    fn prepare(&mut self, idempotency_key: &str) -> Result<(), ExecutionError>;
163    fn commit(&mut self, idempotency_key: &str) -> Result<String, ExecutionError>;
164    fn abort(&mut self, idempotency_key: &str);
165}
166
167pub struct PlanExecutor<'a, K, S> {
168    kernels: &'a mut K,
169    sink: &'a mut S,
170}
171
172impl<'a, K, S> PlanExecutor<'a, K, S>
173where
174    K: KernelExecutor,
175    S: TransactionalSink,
176{
177    pub fn new(kernels: &'a mut K, sink: &'a mut S) -> Self {
178        Self { kernels, sink }
179    }
180
181    pub fn execute(
182        &mut self,
183        authorized: &AuthorizedPlan,
184        invocation_id: [u8; 32],
185        invocation_inputs: BTreeMap<crate::PortRef, K::Value>,
186    ) -> Result<ExecutionResult<K::Value>, Box<ExecutionFailure>> {
187        let plan = authorized.as_plan();
188        let mut receipt = ExecutionReceipt {
189            invocation_id,
190            graph_id: plan.graph_id,
191            plan_id: plan.plan_id,
192            realm: plan.realm,
193            completed_nodes: Vec::new(),
194            attempts: Vec::new(),
195            committed_transactions: Vec::new(),
196            gaps: Vec::new(),
197        };
198        let mut buffers: BTreeMap<BufferId, K::Value> = BTreeMap::new();
199        let mut terminal_values = BTreeMap::new();
200
201        if !plan.feedback.is_empty()
202            || plan
203                .nodes
204                .iter()
205                .any(|node| node.state.scope != crate::StateScope::Stateless)
206        {
207            return Err(Box::new(ExecutionFailure {
208                error: ExecutionError::StatefulPlanUnsupported,
209                receipt,
210            }));
211        }
212
213        if let Some(unexpected) = invocation_inputs
214            .keys()
215            .find(|port| plan.invocation_ports.binary_search(port).is_err())
216        {
217            return Err(Box::new(ExecutionFailure {
218                error: ExecutionError::UnexpectedInvocation(unexpected.clone()),
219                receipt,
220            }));
221        }
222        if let Some(missing) = plan
223            .invocation_ports
224            .iter()
225            .find(|port| !invocation_inputs.contains_key(*port))
226        {
227            return Err(Box::new(ExecutionFailure {
228                error: ExecutionError::MissingInvocation(missing.clone()),
229                receipt,
230            }));
231        }
232
233        for node in &plan.nodes {
234            let key = idempotency_key(plan, &invocation_id, node.id, node.implementation_id);
235            let attempt_index = receipt.attempts.len();
236            receipt.attempts.push(ExecutionAttempt {
237                step: node.id,
238                semantic_nodes: node.semantic_nodes.clone(),
239                kernel: node.kernel,
240                implementation_id: node.implementation_id,
241                attempts: 0,
242                kernel_succeeded: false,
243                completed: false,
244            });
245            if node.effect == Effect::Transactional
246                && let Err(error) = self.sink.prepare(&key)
247            {
248                return Err(Box::new(ExecutionFailure { error, receipt }));
249            }
250
251            let mut input_values = Vec::with_capacity(node.input_bindings.len());
252            for binding in &node.input_bindings {
253                match binding {
254                    InputBinding::Absent => input_values.push(None),
255                    InputBinding::Invocation(invocation) => {
256                        let port = &plan.invocation_ports[*invocation as usize];
257                        match invocation_inputs.get(port) {
258                            Some(value) => input_values.push(Some(value)),
259                            None => {
260                                if node.effect == Effect::Transactional {
261                                    self.sink.abort(&key);
262                                }
263                                return Err(Box::new(ExecutionFailure {
264                                    error: ExecutionError::MissingInvocation(port.clone()),
265                                    receipt,
266                                }));
267                            }
268                        }
269                    }
270                    InputBinding::Buffer(buffer) => match buffers.get(buffer) {
271                        Some(value) => input_values.push(Some(value)),
272                        None => {
273                            if node.effect == Effect::Transactional {
274                                self.sink.abort(&key);
275                            }
276                            return Err(Box::new(ExecutionFailure {
277                                error: ExecutionError::MissingBuffer(*buffer),
278                                receipt,
279                            }));
280                        }
281                    },
282                    InputBinding::Feedback(_) => {
283                        unreachable!("stateful plans fail before execution")
284                    }
285                }
286            }
287
288            let outputs = loop {
289                receipt.attempts[attempt_index].attempts += 1;
290                match self.kernels.execute_with_gaps(node, &input_values) {
291                    Ok(outputs) => break outputs,
292                    Err(error) => {
293                        if let ExecutionError::KernelFailed { failure, .. } = &error
294                            && !node.failure.domains.contains(&failure.domain)
295                        {
296                            if node.effect == Effect::Transactional {
297                                self.sink.abort(&key);
298                            }
299                            return Err(Box::new(ExecutionFailure {
300                                error: ExecutionError::UndeclaredFailure(
301                                    node.kernel,
302                                    failure.domain.clone(),
303                                ),
304                                receipt,
305                            }));
306                        }
307                        let failures = receipt.attempts[attempt_index].attempts - 1;
308                        if failures >= u32::from(node.retry_limit) || !error.retryable() {
309                            if node.effect == Effect::Transactional {
310                                self.sink.abort(&key);
311                            }
312                            return Err(Box::new(ExecutionFailure { error, receipt }));
313                        }
314                        if !matches!(
315                            node.effect,
316                            Effect::Pure
317                                | Effect::Idempotent
318                                | Effect::Transactional
319                                | Effect::AtLeastOnce
320                        ) {
321                            return Err(Box::new(ExecutionFailure {
322                                error: ExecutionError::UnsafeRetry(node.kernel),
323                                receipt,
324                            }));
325                        }
326                    }
327                }
328            };
329
330            if outputs.outputs.len() != node.output_bindings.len() {
331                if node.effect == Effect::Transactional {
332                    self.sink.abort(&key);
333                }
334                return Err(Box::new(ExecutionFailure {
335                    error: ExecutionError::OutputArity {
336                        kernel: node.kernel,
337                        expected: node.output_bindings.len(),
338                        actual: outputs.outputs.len(),
339                    },
340                    receipt,
341                }));
342            }
343            if outputs.gaps.iter().any(|gap| {
344                node.partiality != crate::Partiality::ExplicitGaps
345                    || gap.output_index as usize >= outputs.outputs.len()
346                    || gap.length == Some(0)
347                    || !node.failure.domains.contains(&gap.domain)
348            }) {
349                if node.effect == Effect::Transactional {
350                    self.sink.abort(&key);
351                }
352                return Err(Box::new(ExecutionFailure {
353                    error: ExecutionError::InvalidGap(node.kernel),
354                    receipt,
355                }));
356            }
357            receipt
358                .gaps
359                .extend(outputs.gaps.into_iter().map(|gap| GapReceipt {
360                    step: node.id,
361                    semantic_nodes: node.semantic_nodes.clone(),
362                    gap,
363                }));
364            let mut terminals = Vec::new();
365            for (binding, output) in node.output_bindings.iter().copied().zip(outputs.outputs) {
366                match binding {
367                    OutputBinding::Buffer(buffer) => {
368                        buffers.insert(buffer, output);
369                    }
370                    OutputBinding::Terminal => terminals.push(output),
371                }
372            }
373            if !terminals.is_empty()
374                && let Some(terminal) = node.semantic_nodes.last()
375            {
376                terminal_values.insert(*terminal, terminals);
377            }
378
379            receipt.attempts[attempt_index].kernel_succeeded = true;
380            if node.effect == Effect::Transactional {
381                match self.sink.commit(&key) {
382                    Ok(transaction) => receipt.committed_transactions.push(transaction),
383                    Err(error) => return Err(Box::new(ExecutionFailure { error, receipt })),
384                }
385            }
386            receipt.attempts[attempt_index].completed = true;
387            receipt
388                .completed_nodes
389                .extend(node.semantic_nodes.iter().copied());
390
391            for buffer in node
392                .input_bindings
393                .iter()
394                .filter_map(|binding| match binding {
395                    InputBinding::Buffer(buffer) => Some(buffer),
396                    InputBinding::Invocation(_)
397                    | InputBinding::Feedback(_)
398                    | InputBinding::Absent => None,
399                })
400            {
401                if plan
402                    .buffers
403                    .get(buffer.0 as usize)
404                    .is_some_and(|buffer_plan| node.id == buffer_plan.last_consumer)
405                {
406                    buffers.remove(buffer);
407                }
408            }
409        }
410
411        Ok(ExecutionResult {
412            terminal_values,
413            receipt,
414        })
415    }
416}
417
418fn idempotency_key(
419    plan: &CompiledPlan,
420    invocation_id: &[u8; 32],
421    step: StepId,
422    implementation_id: ImplementationId,
423) -> String {
424    let mut hasher = blake3::Hasher::new_derive_key("blut.transaction.v2");
425    hasher.update(&plan.plan_id.0);
426    hasher.update(invocation_id);
427    hasher.update(&step.0.to_le_bytes());
428    hasher.update(&implementation_id.0);
429    hasher.finalize().to_hex().as_str().to_string()
430}
431
432#[cfg(test)]
433mod tests {
434    use alloc::collections::BTreeMap;
435    use alloc::vec;
436
437    use super::*;
438    use crate::model::{
439        CompiledNode, CompiledPlan, Determinism, ExecutionRealm, GraphId, ImplementationId,
440        KernelId, NodeId, PlanId, ResourceEnvelope,
441    };
442
443    struct Kernels {
444        failures_remaining: u32,
445    }
446
447    impl KernelExecutor for Kernels {
448        type Value = u32;
449
450        fn execute(
451            &mut self,
452            node: &CompiledNode,
453            inputs: &[Option<&Self::Value>],
454        ) -> Result<Vec<Self::Value>, ExecutionError> {
455            if self.failures_remaining > 0 {
456                self.failures_remaining -= 1;
457                return Err(ExecutionError::KernelFailed {
458                    kernel: node.kernel,
459                    failure: StructuredFailure {
460                        domain: "test.kernel".into(),
461                        code: "retry".into(),
462                        message: "retry".into(),
463                        retryable: true,
464                        evidence: Vec::new(),
465                    },
466                });
467            }
468            let value = inputs.iter().flatten().map(|value| **value).sum::<u32>() + 1;
469            Ok(vec![value; node.output_bindings.len()])
470        }
471    }
472
473    #[derive(Default)]
474    struct Sink {
475        prepared: Vec<String>,
476        committed: Vec<String>,
477        fail_commit: bool,
478    }
479
480    impl TransactionalSink for Sink {
481        fn prepare(&mut self, key: &str) -> Result<(), ExecutionError> {
482            self.prepared.push(key.into());
483            Ok(())
484        }
485
486        fn commit(&mut self, key: &str) -> Result<String, ExecutionError> {
487            if self.fail_commit {
488                return Err(ExecutionError::TransactionCommit("injected".into()));
489            }
490            self.committed.push(key.into());
491            Ok(key.into())
492        }
493
494        fn abort(&mut self, _key: &str) {}
495    }
496
497    fn node(
498        id: u32,
499        inputs: Vec<BufferId>,
500        outputs: Vec<BufferId>,
501        effect: Effect,
502    ) -> CompiledNode {
503        let input_count = inputs.len();
504        let output_count = outputs.len().max(1);
505        CompiledNode {
506            id: StepId(id),
507            semantic_nodes: vec![NodeId(id)],
508            semantic_types: vec![crate::NodeTypeRef {
509                type_name: "test".into(),
510                version: 1,
511            }],
512            semantic_configs: vec![BTreeMap::new()],
513            kernel: KernelId(id),
514            implementation_id: ImplementationId([id as u8 + 1; 32]),
515            resources: ResourceEnvelope::bounded(0, 0, 1),
516            determinism: Determinism::BitExact,
517            lowering: "test".into(),
518            conversion: None,
519            input_ports: (0..inputs.len())
520                .map(|index| format!("in-{index}"))
521                .collect(),
522            output_ports: if outputs.is_empty() {
523                vec!["out".into()]
524            } else {
525                (0..outputs.len())
526                    .map(|index| format!("out-{index}"))
527                    .collect()
528            },
529            input_contracts: (0..input_count)
530                .map(|index| {
531                    crate::CompiledPortContract::opaque(
532                        format!("in-{index}"),
533                        "test",
534                        crate::Layout::Canonical,
535                        4,
536                    )
537                })
538                .collect(),
539            output_contracts: (0..output_count)
540                .map(|index| {
541                    crate::CompiledPortContract::opaque(
542                        if output_count == 1 {
543                            "out".into()
544                        } else {
545                            format!("out-{index}")
546                        },
547                        "test",
548                        crate::Layout::Canonical,
549                        4,
550                    )
551                })
552                .collect(),
553            input_bindings: inputs.into_iter().map(InputBinding::Buffer).collect(),
554            output_bindings: if outputs.is_empty() {
555                vec![OutputBinding::Terminal]
556            } else {
557                outputs.into_iter().map(OutputBinding::Buffer).collect()
558            },
559            partiality: crate::Partiality::Atomic,
560            failure: crate::FailureContract {
561                domains: vec!["test.kernel".into()],
562            },
563            effect,
564            retry_limit: 0,
565            state: crate::StateContract::stateless(),
566            subgraph_path: vec![],
567        }
568    }
569
570    #[test]
571    fn fanout_and_join_follow_buffers_not_node_iteration() {
572        let mut plan = CompiledPlan {
573            schema_version: 3,
574            graph_id: GraphId([1; 32]),
575            plan_id: PlanId([2; 32]),
576            realm: ExecutionRealm::HostStream,
577            order: vec![NodeId(0), NodeId(1), NodeId(2), NodeId(3)],
578            nodes: vec![
579                node(0, vec![], vec![BufferId(0)], Effect::Pure),
580                node(1, vec![BufferId(0)], vec![BufferId(1)], Effect::Pure),
581                node(2, vec![BufferId(0)], vec![BufferId(2)], Effect::Pure),
582                node(3, vec![BufferId(1), BufferId(2)], vec![], Effect::Pure),
583            ],
584            buffers: vec![
585                crate::BufferPlan {
586                    id: BufferId(0),
587                    layout: crate::Layout::Canonical,
588                    capacity_bytes: 4,
589                    producer: StepId(0),
590                    consumers: vec![StepId(1), StepId(2)],
591                    last_consumer: StepId(2),
592                    aliases: None,
593                },
594                crate::BufferPlan {
595                    id: BufferId(1),
596                    layout: crate::Layout::Canonical,
597                    capacity_bytes: 4,
598                    producer: StepId(1),
599                    consumers: vec![StepId(3)],
600                    last_consumer: StepId(3),
601                    aliases: None,
602                },
603                crate::BufferPlan {
604                    id: BufferId(2),
605                    layout: crate::Layout::Canonical,
606                    capacity_bytes: 4,
607                    producer: StepId(2),
608                    consumers: vec![StepId(3)],
609                    last_consumer: StepId(3),
610                    aliases: None,
611                },
612            ],
613            feedback: vec![],
614            invocation_ports: vec![],
615            propagated_proofs: vec![],
616            propagated_policy: vec![],
617            resulting_fidelity: u16::MAX,
618            peak_bytes: 12,
619            persistent_state_bytes: 0,
620            session: None,
621        };
622        plan.plan_id = PlanId(crate::compile::hash_plan(&plan));
623        let plan = AuthorizedPlan::new(plan);
624        let roots = BTreeMap::new();
625        let mut kernels = Kernels {
626            failures_remaining: 0,
627        };
628        let mut sink = Sink::default();
629        let result = PlanExecutor::new(&mut kernels, &mut sink)
630            .execute(&plan, [9; 32], roots)
631            .unwrap();
632        assert_eq!(result.terminal_values[&NodeId(3)], vec![5]);
633        assert_eq!(result.receipt.completed_nodes, plan.order);
634    }
635
636    #[test]
637    fn failure_returns_attempt_receipt_and_invocations_have_distinct_keys() {
638        let mut plan = CompiledPlan {
639            schema_version: 3,
640            graph_id: GraphId([1; 32]),
641            plan_id: PlanId([2; 32]),
642            realm: ExecutionRealm::BlutDurable,
643            order: vec![NodeId(0)],
644            nodes: vec![node(0, vec![], vec![], Effect::Transactional)],
645            buffers: vec![],
646            feedback: vec![],
647            invocation_ports: vec![],
648            propagated_proofs: vec![],
649            propagated_policy: vec![],
650            resulting_fidelity: u16::MAX,
651            peak_bytes: 0,
652            persistent_state_bytes: 0,
653            session: None,
654        };
655        plan.nodes[0].retry_limit = 1;
656        plan.plan_id = PlanId(crate::compile::hash_plan(&plan));
657        let plan = AuthorizedPlan::new(plan);
658        let roots = BTreeMap::new();
659        let mut kernels = Kernels {
660            failures_remaining: 2,
661        };
662        let mut sink = Sink::default();
663        let failure = PlanExecutor::new(&mut kernels, &mut sink)
664            .execute(&plan, [1; 32], roots)
665            .unwrap_err();
666        assert_eq!(failure.receipt.attempts[0].attempts, 2);
667        assert!(!failure.receipt.attempts[0].completed);
668
669        let key_a = idempotency_key(&plan, &[1; 32], StepId(0), plan.nodes[0].implementation_id);
670        let key_b = idempotency_key(&plan, &[2; 32], StepId(0), plan.nodes[0].implementation_id);
671        assert_ne!(key_a, key_b);
672    }
673
674    #[test]
675    fn commit_failure_returns_the_completed_attempt_without_a_commit_receipt() {
676        let mut plan = CompiledPlan {
677            schema_version: 3,
678            graph_id: GraphId([1; 32]),
679            plan_id: PlanId([2; 32]),
680            realm: ExecutionRealm::BlutDurable,
681            order: vec![NodeId(0)],
682            nodes: vec![node(0, vec![], vec![], Effect::Transactional)],
683            buffers: vec![],
684            feedback: vec![],
685            invocation_ports: vec![],
686            propagated_proofs: vec![],
687            propagated_policy: vec![],
688            resulting_fidelity: u16::MAX,
689            peak_bytes: 0,
690            persistent_state_bytes: 0,
691            session: None,
692        };
693        plan.plan_id = PlanId(crate::compile::hash_plan(&plan));
694        let plan = AuthorizedPlan::new(plan);
695        let roots = BTreeMap::new();
696        let mut kernels = Kernels {
697            failures_remaining: 0,
698        };
699        let mut sink = Sink {
700            fail_commit: true,
701            ..Sink::default()
702        };
703        let failure = PlanExecutor::new(&mut kernels, &mut sink)
704            .execute(&plan, [3; 32], roots)
705            .unwrap_err();
706        assert!(matches!(
707            failure.error,
708            ExecutionError::TransactionCommit(_)
709        ));
710        assert!(failure.receipt.completed_nodes.is_empty());
711        assert!(failure.receipt.attempts[0].kernel_succeeded);
712        assert!(!failure.receipt.attempts[0].completed);
713        assert!(failure.receipt.committed_transactions.is_empty());
714    }
715
716    #[test]
717    fn explicit_partial_node_emits_a_structured_gap_receipt() {
718        struct GapKernels;
719        impl KernelExecutor for GapKernels {
720            type Value = u32;
721
722            fn execute(
723                &mut self,
724                _node: &CompiledNode,
725                _inputs: &[Option<&Self::Value>],
726            ) -> Result<Vec<Self::Value>, ExecutionError> {
727                unreachable!("gap-aware hook is used")
728            }
729
730            fn execute_with_gaps(
731                &mut self,
732                _node: &CompiledNode,
733                _inputs: &[Option<&Self::Value>],
734            ) -> Result<KernelExecution<Self::Value>, ExecutionError> {
735                Ok(KernelExecution {
736                    outputs: vec![7],
737                    gaps: vec![KernelGap {
738                        output_index: 0,
739                        offset: 12,
740                        length: Some(4),
741                        domain: "biosignal.missing".into(),
742                        code: "packet-loss".into(),
743                    }],
744                })
745            }
746        }
747
748        let mut partial = node(0, vec![], vec![], Effect::Pure);
749        partial.partiality = crate::Partiality::ExplicitGaps;
750        partial.failure.domains = vec!["biosignal.missing".into()];
751        let mut plan = CompiledPlan {
752            schema_version: 3,
753            graph_id: GraphId([1; 32]),
754            plan_id: PlanId([0; 32]),
755            realm: ExecutionRealm::HostStream,
756            order: vec![NodeId(0)],
757            nodes: vec![partial],
758            buffers: vec![],
759            feedback: vec![],
760            invocation_ports: vec![],
761            propagated_proofs: vec![],
762            propagated_policy: vec![],
763            resulting_fidelity: u16::MAX,
764            peak_bytes: 0,
765            persistent_state_bytes: 0,
766            session: None,
767        };
768        plan.plan_id = PlanId(crate::compile::hash_plan(&plan));
769        let plan = AuthorizedPlan::new(plan);
770        let mut kernels = GapKernels;
771        let mut sink = Sink::default();
772        let result = PlanExecutor::new(&mut kernels, &mut sink)
773            .execute(&plan, [8; 32], BTreeMap::new())
774            .unwrap();
775        assert_eq!(result.receipt.gaps.len(), 1);
776        assert_eq!(result.receipt.gaps[0].gap.code, "packet-loss");
777        assert_eq!(result.terminal_values[&NodeId(0)], vec![7]);
778    }
779}
blut_graph_core/execute.rs

blut_graph_core/
execute.rs
