harn_vm/flow/predicates/
executor.rs

1//! Budgeted executor for Flow invariant predicates.
2//!
3//! The executor is intentionally small: callers provide predicate runners, and
4//! this module owns scheduling, per-kind budgets, semantic cheap-judge limits,
5//! and deterministic replay drift detection.
6
7use std::collections::{BTreeMap, BTreeSet};
8use std::sync::atomic::{AtomicBool, Ordering};
9use std::sync::{Arc, Mutex};
10use std::time::{Duration, Instant};
11
12use async_trait::async_trait;
13use futures::StreamExt;
14use serde::{Deserialize, Serialize};
15use sha2::{Digest, Sha256};
16use tokio::sync::Semaphore;
17
18use super::compose::verdict_strictness;
19use crate::flow::{InvariantBlockError, InvariantResult, PredicateHash, Slice};
20
21const DEFAULT_DETERMINISTIC_BUDGET: Duration = Duration::from_millis(50);
22const DEFAULT_SEMANTIC_BUDGET: Duration = Duration::from_secs(2);
23const DEFAULT_SEMANTIC_TOKEN_CAP: u64 = 1024;
24const DEFAULT_MAX_DETERMINISTIC_LANES: usize = 16;
25const DEFAULT_MAX_SEMANTIC_LANES: usize = 2;
26const DEFAULT_MAX_DETERMINISTIC_LANES_PER_SLICE: usize = usize::MAX;
27const DEFAULT_MAX_SEMANTIC_LANES_PER_SLICE: usize = 1;
28const DEFAULT_SLICE_DETERMINISTIC_ENVELOPE: Duration = Duration::from_secs(5);
29const DEFAULT_SLICE_SEMANTIC_ENVELOPE: Duration = Duration::from_secs(20);
30
31/// Predicate execution mode declared by predicate author annotations.
32#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
33#[serde(rename_all = "snake_case")]
34pub enum PredicateKind {
35    /// Pure Harn predicate. No shell, network, LLM, or host side effects.
36    Deterministic,
37    /// Semantic predicate. May make one `cheap_judge` call over pre-baked
38    /// evidence within the semantic wall-clock and token budgets.
39    Semantic,
40}
41
42/// How a semantic predicate uses its deterministic fallback.
43#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
44#[serde(rename_all = "snake_case")]
45pub enum SemanticFallbackPolicy {
46    /// The semantic and deterministic results are composed by strictness.
47    #[default]
48    Enforce,
49    /// The deterministic result is evaluated and recorded for replay audit,
50    /// but does not alter the live semantic verdict.
51    Advisory,
52}
53
54/// Request passed to the cheap semantic judge.
55#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
56pub struct CheapJudgeRequest {
57    pub prompt: String,
58    pub evidence_key: String,
59    pub evidence: String,
60}
61
62/// Response returned by the cheap semantic judge.
63#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
64pub struct CheapJudgeResponse {
65    pub passes: bool,
66    #[serde(default, skip_serializing_if = "Option::is_none")]
67    pub reason: Option<String>,
68    #[serde(default)]
69    pub input_tokens: u64,
70    #[serde(default)]
71    pub output_tokens: u64,
72    #[serde(default, skip_serializing_if = "Option::is_none")]
73    pub provider_id: Option<String>,
74    #[serde(default, skip_serializing_if = "Option::is_none")]
75    pub model_id: Option<String>,
76    #[serde(default, skip_serializing_if = "Option::is_none")]
77    pub cheap_judge_version: Option<String>,
78}
79
80/// Replay-audit metadata for a semantic predicate's judge call.
81#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
82pub struct SemanticReplayAuditMetadata {
83    #[serde(default, skip_serializing_if = "Option::is_none")]
84    pub provider_id: Option<String>,
85    #[serde(default, skip_serializing_if = "Option::is_none")]
86    pub model_id: Option<String>,
87    pub prompt_hash: String,
88    pub evidence_hashes: BTreeMap<String, String>,
89    pub token_cap: u64,
90    #[serde(default, skip_serializing_if = "Option::is_none")]
91    pub cheap_judge_version: Option<String>,
92}
93
94/// Host-provided adapter for semantic predicate judging.
95#[async_trait]
96pub trait CheapJudge: Send + Sync {
97    async fn cheap_judge(
98        &self,
99        request: CheapJudgeRequest,
100    ) -> Result<CheapJudgeResponse, InvariantBlockError>;
101}
102
103/// Predicate runner supplied by a collector or Harn adapter.
104#[async_trait]
105pub trait PredicateRunner: Send + Sync {
106    fn hash(&self) -> PredicateHash;
107    fn name(&self) -> String {
108        self.hash().as_str().to_string()
109    }
110    fn kind(&self) -> PredicateKind;
111    fn fallback_hash(&self) -> Option<PredicateHash> {
112        None
113    }
114    fn fallback_policy(&self) -> SemanticFallbackPolicy {
115        SemanticFallbackPolicy::Enforce
116    }
117    fn fallback_diagnostic(&self) -> Option<InvariantBlockError> {
118        None
119    }
120    fn raw_result(&self) -> Option<serde_json::Value> {
121        None
122    }
123    fn enforced(&self) -> bool {
124        true
125    }
126
127    /// Static evidence captured when the predicate was authored. Semantic
128    /// predicates may only judge over this map; the executor never fetches
129    /// evidence during evaluation.
130    fn evidence(&self) -> BTreeMap<String, String> {
131        BTreeMap::new()
132    }
133
134    async fn evaluate(&self, context: PredicateContext) -> InvariantResult;
135}
136
137/// Runtime context made available to a predicate invocation.
138#[derive(Clone)]
139pub struct PredicateContext {
140    inner: Arc<PredicateContextInner>,
141}
142
143struct PredicateContextInner {
144    slice: Arc<Slice>,
145    kind: PredicateKind,
146    evidence: BTreeMap<String, String>,
147    cheap_judge: Option<Arc<dyn CheapJudge>>,
148    semantic_token_cap: u64,
149    cancel_token: Arc<AtomicBool>,
150    judge_state: Mutex<JudgeBudgetState>,
151}
152
153#[derive(Default)]
154struct JudgeBudgetState {
155    calls: u64,
156    tokens: u64,
157    block_error: Option<InvariantBlockError>,
158    semantic_audit: Option<SemanticReplayAuditMetadata>,
159}
160
161impl PredicateContext {
162    fn new(
163        slice: Arc<Slice>,
164        kind: PredicateKind,
165        evidence: BTreeMap<String, String>,
166        cheap_judge: Option<Arc<dyn CheapJudge>>,
167        semantic_token_cap: u64,
168        cancel_token: Arc<AtomicBool>,
169    ) -> Self {
170        Self {
171            inner: Arc::new(PredicateContextInner {
172                slice,
173                kind,
174                evidence,
175                cheap_judge,
176                semantic_token_cap,
177                cancel_token,
178                judge_state: Mutex::new(JudgeBudgetState::default()),
179            }),
180        }
181    }
182
183    pub fn slice(&self) -> &Slice {
184        &self.inner.slice
185    }
186
187    pub fn kind(&self) -> PredicateKind {
188        self.inner.kind
189    }
190
191    pub fn evidence(&self, key: &str) -> Option<&str> {
192        self.inner.evidence.get(key).map(String::as_str)
193    }
194
195    pub fn is_cancelled(&self) -> bool {
196        self.inner.cancel_token.load(Ordering::SeqCst)
197    }
198
199    /// Invoke the semantic cheap judge with executor-enforced limits.
200    ///
201    /// Deterministic predicates cannot call this method. Semantic predicates
202    /// get one call per predicate evaluation attempt, and the request must
203    /// refer to evidence already present in the predicate's evidence map.
204    pub async fn cheap_judge(
205        &self,
206        prompt: impl Into<String>,
207        evidence_key: impl Into<String>,
208    ) -> Result<CheapJudgeResponse, InvariantBlockError> {
209        if self.inner.kind != PredicateKind::Semantic {
210            return Err(self.record_block(InvariantBlockError::new(
211                "side_effect_denied",
212                "deterministic predicates cannot invoke cheap_judge",
213            )));
214        }
215
216        let prompt = prompt.into();
217        let evidence_key = evidence_key.into();
218        let Some(evidence) = self.inner.evidence.get(&evidence_key).cloned() else {
219            return Err(self.record_block(InvariantBlockError::new(
220                "evidence_missing",
221                format!(
222                    "semantic predicate requested evidence key '{evidence_key}' that was not pre-baked"
223                ),
224            )));
225        };
226
227        let estimated_tokens = estimate_tokens(&prompt).saturating_add(estimate_tokens(&evidence));
228        {
229            let mut state = self
230                .inner
231                .judge_state
232                .lock()
233                .expect("predicate judge state lock");
234            if state.calls >= 1 {
235                let error = InvariantBlockError::budget_exceeded(
236                    "semantic predicate exceeded one cheap_judge call",
237                );
238                state.block_error = Some(error.clone());
239                return Err(error);
240            }
241            if state.tokens.saturating_add(estimated_tokens) > self.inner.semantic_token_cap {
242                let error = InvariantBlockError::budget_exceeded(format!(
243                    "semantic predicate cheap_judge request exceeds token cap {}",
244                    self.inner.semantic_token_cap
245                ));
246                state.block_error = Some(error.clone());
247                return Err(error);
248            }
249            state.calls += 1;
250            state.tokens = state.tokens.saturating_add(estimated_tokens);
251        }
252
253        let Some(judge) = self.inner.cheap_judge.clone() else {
254            return Err(self.record_block(InvariantBlockError::new(
255                "llm_unavailable",
256                "semantic predicate cheap_judge was requested but no judge is installed",
257            )));
258        };
259
260        let response = match judge
261            .cheap_judge(CheapJudgeRequest {
262                prompt: prompt.clone(),
263                evidence_key: evidence_key.clone(),
264                evidence: evidence.clone(),
265            })
266            .await
267        {
268            Ok(response) => response,
269            Err(error) => return Err(self.record_block(error)),
270        };
271        {
272            let mut state = self
273                .inner
274                .judge_state
275                .lock()
276                .expect("predicate judge state lock");
277            state.semantic_audit = Some(SemanticReplayAuditMetadata {
278                provider_id: response.provider_id.clone(),
279                model_id: response.model_id.clone(),
280                prompt_hash: stable_hash(prompt.as_bytes()),
281                evidence_hashes: self
282                    .inner
283                    .evidence
284                    .iter()
285                    .map(|(key, value)| (key.clone(), stable_hash(value.as_bytes())))
286                    .collect(),
287                token_cap: self.inner.semantic_token_cap,
288                cheap_judge_version: response.cheap_judge_version.clone(),
289            });
290        }
291        let response_tokens = response.input_tokens.saturating_add(response.output_tokens);
292        {
293            let mut state = self
294                .inner
295                .judge_state
296                .lock()
297                .expect("predicate judge state lock");
298            state.tokens = state.tokens.saturating_add(response_tokens);
299            if state.tokens > self.inner.semantic_token_cap {
300                let error = InvariantBlockError::budget_exceeded(format!(
301                    "semantic predicate cheap_judge response exceeded token cap {}",
302                    self.inner.semantic_token_cap
303                ));
304                state.block_error = Some(error.clone());
305                return Err(error);
306            }
307        }
308        Ok(response)
309    }
310
311    fn cancel(&self) {
312        self.inner.cancel_token.store(true, Ordering::SeqCst);
313    }
314
315    fn block_error(&self) -> Option<InvariantBlockError> {
316        self.inner
317            .judge_state
318            .lock()
319            .expect("predicate judge state lock")
320            .block_error
321            .clone()
322    }
323
324    fn semantic_audit(&self) -> Option<SemanticReplayAuditMetadata> {
325        self.inner
326            .judge_state
327            .lock()
328            .expect("predicate judge state lock")
329            .semantic_audit
330            .clone()
331    }
332
333    fn record_block(&self, error: InvariantBlockError) -> InvariantBlockError {
334        self.inner
335            .judge_state
336            .lock()
337            .expect("predicate judge state lock")
338            .block_error = Some(error.clone());
339        error
340    }
341}
342
343/// Cross-slice scheduling configuration.
344///
345/// The fairness scheduler enforces three orthogonal limits on top of the
346/// existing per-predicate hard timeouts:
347///
348/// 1. **Global lane caps** (`max_deterministic_lanes`, `max_semantic_lanes`)
349///    bound how many predicates of each kind may run concurrently across all
350///    queued slices. Deterministic and semantic lanes are independent so
351///    deterministic work continues to make progress while semantic predicates
352///    wait for cheap-judge slots.
353/// 2. **Per-slice lane caps**
354///    (`max_deterministic_lanes_per_slice`, `max_semantic_lanes_per_slice`)
355///    keep one slice from monopolizing all global lanes. With the default
356///    semantic-per-slice cap of 1, two slices each holding a permit alternate
357///    fairly through the global semantic semaphore (FIFO under tokio).
358/// 3. **Aggregate per-slice envelopes**
359///    (`slice_deterministic_envelope`, `slice_semantic_envelope`) bound the
360///    total wall-clock spent on each kind within one slice. Once exhausted,
361///    every remaining predicate of that kind for that slice short-circuits to
362///    a structured `budget_exceeded` block — never a panic and never an
363///    implicit approval.
364#[derive(Clone, Debug)]
365pub struct PredicateSchedulerConfig {
366    /// Global cap on concurrent deterministic predicate evaluations across
367    /// all slices.
368    pub max_deterministic_lanes: usize,
369    /// Global cap on concurrent semantic predicate evaluations across all
370    /// slices.
371    pub max_semantic_lanes: usize,
372    /// Per-slice cap on concurrent deterministic predicate evaluations.
373    pub max_deterministic_lanes_per_slice: usize,
374    /// Per-slice cap on concurrent semantic predicate evaluations. Keep this
375    /// strictly less than `max_semantic_lanes` to enforce fairness when more
376    /// than one slice is queued.
377    pub max_semantic_lanes_per_slice: usize,
378    /// Aggregate wall-clock envelope summed across all deterministic predicate
379    /// attempts for one slice. Both the first attempt and the replay attempt
380    /// of a deterministic predicate count against this envelope. Setting this
381    /// to `Duration::ZERO` disables aggregate enforcement and falls back to
382    /// the per-predicate hard timeout alone.
383    pub slice_deterministic_envelope: Duration,
384    /// Aggregate wall-clock envelope summed across all semantic predicate
385    /// attempts for one slice. `Duration::ZERO` disables aggregate
386    /// enforcement.
387    pub slice_semantic_envelope: Duration,
388}
389
390impl Default for PredicateSchedulerConfig {
391    fn default() -> Self {
392        Self {
393            max_deterministic_lanes: DEFAULT_MAX_DETERMINISTIC_LANES,
394            max_semantic_lanes: DEFAULT_MAX_SEMANTIC_LANES,
395            max_deterministic_lanes_per_slice: DEFAULT_MAX_DETERMINISTIC_LANES_PER_SLICE,
396            max_semantic_lanes_per_slice: DEFAULT_MAX_SEMANTIC_LANES_PER_SLICE,
397            slice_deterministic_envelope: DEFAULT_SLICE_DETERMINISTIC_ENVELOPE,
398            slice_semantic_envelope: DEFAULT_SLICE_SEMANTIC_ENVELOPE,
399        }
400    }
401}
402
403/// Executor configuration.
404#[derive(Clone, Debug)]
405pub struct PredicateExecutorConfig {
406    pub deterministic_budget: Duration,
407    pub semantic_budget: Duration,
408    pub semantic_token_cap: u64,
409    /// Re-run successful deterministic predicates to detect result drift.
410    pub replay_deterministic: bool,
411    /// Cross-slice fairness and aggregate-envelope scheduling knobs.
412    pub scheduler: PredicateSchedulerConfig,
413}
414
415impl Default for PredicateExecutorConfig {
416    fn default() -> Self {
417        Self {
418            deterministic_budget: DEFAULT_DETERMINISTIC_BUDGET,
419            semantic_budget: DEFAULT_SEMANTIC_BUDGET,
420            semantic_token_cap: DEFAULT_SEMANTIC_TOKEN_CAP,
421            replay_deterministic: true,
422            scheduler: PredicateSchedulerConfig::default(),
423        }
424    }
425}
426
427/// Budgeted predicate executor.
428#[derive(Clone)]
429pub struct PredicateExecutor {
430    config: PredicateExecutorConfig,
431    cheap_judge: Option<Arc<dyn CheapJudge>>,
432}
433
434impl PredicateExecutor {
435    pub fn new(config: PredicateExecutorConfig) -> Self {
436        Self {
437            config,
438            cheap_judge: None,
439        }
440    }
441
442    pub fn with_cheap_judge(
443        config: PredicateExecutorConfig,
444        cheap_judge: Arc<dyn CheapJudge>,
445    ) -> Self {
446        Self {
447            config,
448            cheap_judge: Some(cheap_judge),
449        }
450    }
451
452    /// Evaluate predicates for a single slice using the scheduler defaults.
453    ///
454    /// This is shorthand for [`execute_slices`](Self::execute_slices) with a
455    /// single-element queue. Even one slice still flows through the cross-slice
456    /// scheduler so the aggregate per-slice envelope and lane caps apply.
457    pub async fn execute_slice(
458        &self,
459        slice: &Slice,
460        predicates: &[Arc<dyn PredicateRunner>],
461    ) -> PredicateExecutionReport {
462        let mut reports = self
463            .execute_slices(vec![(slice.clone(), predicates.to_vec())])
464            .await;
465        reports.pop().unwrap_or(PredicateExecutionReport {
466            records: Vec::new(),
467            skipped: Vec::new(),
468        })
469    }
470
471    /// Evaluate one slice serially through the same budget, replay, fallback,
472    /// and record implementation used by the concurrent scheduler.
473    pub async fn execute_slice_serial(
474        &self,
475        slice: &Slice,
476        predicates: &[Arc<dyn PredicateRunner>],
477    ) -> PredicateExecutionReport {
478        let scheduler = &self.config.scheduler;
479        let lanes = SliceLanes::new(
480            Arc::new(Semaphore::new(1)),
481            Arc::new(Semaphore::new(1)),
482            1,
483            1,
484        );
485        let envelope = SliceEnvelope::new(
486            scheduler.slice_deterministic_envelope,
487            scheduler.slice_semantic_envelope,
488        );
489        let slice = Arc::new(slice.clone());
490        let mut records = Vec::with_capacity(predicates.len());
491        for runner in predicates {
492            records.push(
493                self.execute_one(
494                    slice.clone(),
495                    runner.clone(),
496                    lanes.clone(),
497                    envelope.clone(),
498                )
499                .await,
500            );
501        }
502        self.apply_semantic_fallbacks(&mut records);
503        records.sort_by(|left, right| left.predicate_hash.cmp(&right.predicate_hash));
504        PredicateExecutionReport {
505            records,
506            skipped: Vec::new(),
507        }
508    }
509
510    /// Select named predicates and their semantic fallback dependencies, then
511    /// execute them serially. This is the public-adapter seam: selection,
512    /// dependency skips, enforcement membership, budgets, and fallback
513    /// composition all stay in the executor.
514    pub async fn execute_named_slice_serial(
515        &self,
516        slice: &Slice,
517        predicates: &[Arc<dyn PredicateRunner>],
518        requested_names: Option<&BTreeSet<String>>,
519        include_semantic: bool,
520    ) -> PredicateExecutionReport {
521        let directly_selected = predicates
522            .iter()
523            .filter(|runner| {
524                requested_names
525                    .map(|names| names.contains(&runner.name()))
526                    .unwrap_or(true)
527                    && (include_semantic || runner.kind() != PredicateKind::Semantic)
528            })
529            .map(|runner| runner.hash())
530            .collect::<BTreeSet<_>>();
531        let by_hash = predicates
532            .iter()
533            .map(|runner| (runner.hash(), runner.clone()))
534            .collect::<BTreeMap<_, _>>();
535        let mut selected = directly_selected.clone();
536        for runner in predicates {
537            if directly_selected.contains(&runner.hash())
538                && runner.kind() == PredicateKind::Semantic
539            {
540                if let Some(fallback_hash) = runner.fallback_hash() {
541                    if by_hash.contains_key(&fallback_hash) {
542                        selected.insert(fallback_hash);
543                    }
544                }
545            }
546        }
547
548        let selected_runners = predicates
549            .iter()
550            .filter(|runner| selected.contains(&runner.hash()))
551            .cloned()
552            .collect::<Vec<_>>();
553        let skipped = predicates
554            .iter()
555            .filter(|runner| !selected.contains(&runner.hash()))
556            .map(|runner| PredicateExecutionSkip {
557                name: runner.name(),
558                predicate_hash: runner.hash(),
559                kind: runner.kind(),
560                reason: if runner.kind() == PredicateKind::Semantic
561                    && requested_names
562                        .map(|names| names.contains(&runner.name()))
563                        .unwrap_or(true)
564                    && !include_semantic
565                {
566                    "semantic predicates require an explicit include_semantic option".to_string()
567                } else {
568                    "not selected".to_string()
569                },
570            })
571            .collect::<Vec<_>>();
572
573        let mut report = self.execute_slice_serial(slice, &selected_runners).await;
574        for record in &mut report.records {
575            record.enforced &= directly_selected.contains(&record.predicate_hash);
576        }
577        report.skipped = skipped;
578        report
579    }
580
581    /// Evaluate predicates for several candidate slices fairly.
582    ///
583    /// All slices share the global deterministic and semantic lane semaphores
584    /// configured in [`PredicateSchedulerConfig`]. Each slice additionally
585    /// owns its own per-slice lane semaphores (so one slice cannot occupy all
586    /// global semantic lanes), and each slice independently tracks aggregate
587    /// wall-clock against the per-kind envelopes. Once a slice exhausts an
588    /// envelope, all remaining predicates of that kind for that slice
589    /// short-circuit to a `budget_exceeded` `Block`.
590    ///
591    /// Output preserves input slice order. Within each slice, records are
592    /// sorted by predicate hash so reports are deterministic regardless of
593    /// the order predicates finished.
594    pub async fn execute_slices(
595        &self,
596        slices: Vec<(Slice, Vec<Arc<dyn PredicateRunner>>)>,
597    ) -> Vec<PredicateExecutionReport> {
598        let scheduler = &self.config.scheduler;
599        let det_global = Arc::new(Semaphore::new(clamp_permits(
600            scheduler.max_deterministic_lanes,
601        )));
602        let sem_global = Arc::new(Semaphore::new(clamp_permits(scheduler.max_semantic_lanes)));
603
604        let slice_futures = slices
605            .into_iter()
606            .map(|(slice, predicates)| {
607                let executor = self.clone();
608                let det_global = det_global.clone();
609                let sem_global = sem_global.clone();
610                async move {
611                    executor
612                        .execute_slice_inner(slice, predicates, det_global, sem_global)
613                        .await
614                }
615            })
616            .collect::<Vec<_>>();
617
618        futures::future::join_all(slice_futures).await
619    }
620
621    async fn execute_slice_inner(
622        &self,
623        slice: Slice,
624        predicates: Vec<Arc<dyn PredicateRunner>>,
625        det_global: Arc<Semaphore>,
626        sem_global: Arc<Semaphore>,
627    ) -> PredicateExecutionReport {
628        let scheduler = &self.config.scheduler;
629        let slice_rc = Arc::new(slice);
630        let lanes = SliceLanes::new(
631            det_global,
632            sem_global,
633            scheduler.max_deterministic_lanes_per_slice,
634            scheduler.max_semantic_lanes_per_slice,
635        );
636        let envelope = SliceEnvelope::new(
637            scheduler.slice_deterministic_envelope,
638            scheduler.slice_semantic_envelope,
639        );
640
641        // The lane semaphores enforce real concurrency. `buffer_unordered`
642        // just decides how many futures we polled-but-not-yet-completed at
643        // once; cap it at the actual predicate count so we don't allocate
644        // internal slots for predicates that don't exist.
645        let buffer = predicates.len().max(1);
646
647        let mut records = futures::stream::iter(predicates)
648            .map(|runner| {
649                let executor = self.clone();
650                let slice = slice_rc.clone();
651                let lanes = lanes.clone();
652                let envelope = envelope.clone();
653                async move { executor.execute_one(slice, runner, lanes, envelope).await }
654            })
655            .buffer_unordered(buffer)
656            .collect::<Vec<_>>()
657            .await;
658
659        self.apply_semantic_fallbacks(&mut records);
660        records.sort_by(|left, right| left.predicate_hash.cmp(&right.predicate_hash));
661        PredicateExecutionReport {
662            records,
663            skipped: Vec::new(),
664        }
665    }
666
667    async fn execute_one(
668        &self,
669        slice: Arc<Slice>,
670        runner: Arc<dyn PredicateRunner>,
671        lanes: SliceLanes,
672        envelope: SliceEnvelope,
673    ) -> PredicateExecutionRecord {
674        let started = Instant::now();
675        let predicate_hash = runner.hash();
676        let name = runner.name();
677        let kind = runner.kind();
678        let first = self
679            .run_attempt(slice.clone(), runner.as_ref(), &lanes, &envelope)
680            .await;
681        let first_hash = hash_result(&first.result);
682        let mut result = first.result;
683        let mut attempts = 1;
684        let mut second_hash = None;
685        let semantic_replay_audit = first.semantic_audit;
686
687        if self.config.replay_deterministic
688            && kind == PredicateKind::Deterministic
689            && !result.is_blocking()
690        {
691            let second = self
692                .run_attempt(slice, runner.as_ref(), &lanes, &envelope)
693                .await;
694            attempts = 2;
695            let replay_hash = hash_result(&second.result);
696            second_hash = replay_hash.clone();
697            if second.result.is_blocking() {
698                result = second.result;
699            } else {
700                match (first_hash.as_ref(), replay_hash.as_ref()) {
701                    (Some(left), Some(right)) if left == right => {}
702                    (Some(left), Some(right)) => {
703                        result = InvariantResult::block(InvariantBlockError::nondeterministic_drift(
704                            format!(
705                                "deterministic predicate result drifted across replay: {left} != {right}"
706                            ),
707                        ));
708                    }
709                    _ => {
710                        result = InvariantResult::block(InvariantBlockError::new(
711                            "result_hash_failed",
712                            "failed to hash deterministic predicate replay result",
713                        ));
714                    }
715                }
716            }
717        }
718
719        PredicateExecutionRecord {
720            name,
721            predicate_hash,
722            kind,
723            fallback_hash: runner.fallback_hash(),
724            fallback_policy: runner.fallback_policy(),
725            fallback_diagnostic: runner.fallback_diagnostic(),
726            result,
727            raw_result: runner.raw_result(),
728            enforced: runner.enforced(),
729            elapsed_ms: started.elapsed().as_millis() as u64,
730            attempts,
731            replayable: kind == PredicateKind::Deterministic,
732            first_result_hash: first_hash,
733            second_result_hash: second_hash,
734            semantic_replay_audit,
735        }
736    }
737
738    async fn run_attempt(
739        &self,
740        slice: Arc<Slice>,
741        runner: &dyn PredicateRunner,
742        lanes: &SliceLanes,
743        envelope: &SliceEnvelope,
744    ) -> PredicateAttempt {
745        let kind = runner.kind();
746        let timeout = match kind {
747            PredicateKind::Deterministic => self.config.deterministic_budget,
748            PredicateKind::Semantic => self.config.semantic_budget,
749        };
750
751        if let Some(attempt) =
752            envelope_exhausted_attempt(envelope, kind, "before this predicate started")
753        {
754            return attempt;
755        }
756
757        let _permits = lanes.acquire(kind).await;
758
759        // Re-check after acquiring the lane: the queue may have been long
760        // enough that the slice's envelope drained while we waited.
761        if let Some(attempt) =
762            envelope_exhausted_attempt(envelope, kind, "while waiting for a lane")
763        {
764            return attempt;
765        }
766
767        let context = PredicateContext::new(
768            slice,
769            kind,
770            runner.evidence(),
771            self.cheap_judge.clone(),
772            self.config.semantic_token_cap,
773            Arc::new(AtomicBool::new(false)),
774        );
775        let started = Instant::now();
776        let attempt = match tokio::time::timeout(timeout, runner.evaluate(context.clone())).await {
777            Ok(result) => PredicateAttempt {
778                result: context
779                    .block_error()
780                    .map(InvariantResult::block)
781                    .unwrap_or(result),
782                semantic_audit: context.semantic_audit(),
783            },
784            Err(_) => {
785                context.cancel();
786                PredicateAttempt {
787                    result: InvariantResult::block(InvariantBlockError::budget_exceeded(format!(
788                        "{kind:?} predicate exceeded {}ms budget",
789                        timeout.as_millis()
790                    ))),
791                    semantic_audit: context.semantic_audit(),
792                }
793            }
794        };
795        envelope.charge(kind, started.elapsed());
796        attempt
797    }
798
799    fn apply_semantic_fallbacks(&self, records: &mut [PredicateExecutionRecord]) {
800        let by_hash = records
801            .iter()
802            .map(|record| {
803                (
804                    record.predicate_hash.clone(),
805                    (record.kind, record.result.clone()),
806                )
807            })
808            .collect::<BTreeMap<_, _>>();
809
810        for record in records {
811            if record.kind != PredicateKind::Semantic {
812                continue;
813            }
814            if let Some(diagnostic) = record.fallback_diagnostic.take() {
815                record.result = InvariantResult::block(diagnostic);
816                continue;
817            }
818            let Some(fallback_hash) = record.fallback_hash.as_ref() else {
819                record.result = InvariantResult::block(InvariantBlockError::new(
820                    "fallback_missing",
821                    "semantic predicate did not declare a deterministic fallback",
822                ));
823                continue;
824            };
825            let Some((fallback_kind, fallback_result)) = by_hash.get(fallback_hash) else {
826                record.result = InvariantResult::block(InvariantBlockError::new(
827                    "fallback_unselected",
828                    format!(
829                        "semantic predicate fallback {} was not selected for evaluation",
830                        fallback_hash.as_str()
831                    ),
832                ));
833                continue;
834            };
835            if *fallback_kind != PredicateKind::Deterministic {
836                record.result = InvariantResult::block(InvariantBlockError::new(
837                    "fallback_not_deterministic",
838                    format!(
839                        "semantic predicate fallback {} is not deterministic",
840                        fallback_hash.as_str()
841                    ),
842                ));
843                continue;
844            }
845            if record.fallback_policy == SemanticFallbackPolicy::Enforce {
846                record.result = stricter_result(&record.result, fallback_result);
847            }
848        }
849    }
850}
851
852impl Default for PredicateExecutor {
853    fn default() -> Self {
854        Self::new(PredicateExecutorConfig::default())
855    }
856}
857
858/// Per-predicate execution metadata.
859#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
860pub struct PredicateExecutionRecord {
861    pub name: String,
862    #[serde(rename = "hash")]
863    pub predicate_hash: PredicateHash,
864    pub kind: PredicateKind,
865    #[serde(default, skip_serializing_if = "Option::is_none")]
866    pub fallback_hash: Option<PredicateHash>,
867    #[serde(default)]
868    pub fallback_policy: SemanticFallbackPolicy,
869    #[serde(default, skip_serializing_if = "Option::is_none")]
870    pub fallback_diagnostic: Option<InvariantBlockError>,
871    pub result: InvariantResult,
872    #[serde(default, skip_serializing_if = "Option::is_none")]
873    pub raw_result: Option<serde_json::Value>,
874    /// Whether this record independently contributes to the slice verdict.
875    /// Dependency-only advisory fallbacks remain visible but are not enforced.
876    #[serde(default = "default_true")]
877    pub enforced: bool,
878    pub elapsed_ms: u64,
879    pub attempts: u8,
880    pub replayable: bool,
881    #[serde(default, skip_serializing_if = "Option::is_none")]
882    pub first_result_hash: Option<String>,
883    #[serde(default, skip_serializing_if = "Option::is_none")]
884    pub second_result_hash: Option<String>,
885    #[serde(default, skip_serializing_if = "Option::is_none")]
886    pub semantic_replay_audit: Option<SemanticReplayAuditMetadata>,
887}
888
889/// Complete result of executing all predicates for one slice.
890#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
891pub struct PredicateExecutionReport {
892    pub records: Vec<PredicateExecutionRecord>,
893    #[serde(default)]
894    pub skipped: Vec<PredicateExecutionSkip>,
895}
896
897#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
898pub struct PredicateExecutionSkip {
899    pub name: String,
900    #[serde(rename = "hash")]
901    pub predicate_hash: PredicateHash,
902    pub kind: PredicateKind,
903    pub reason: String,
904}
905
906impl PredicateExecutionReport {
907    pub fn invariants_applied(&self) -> Vec<(PredicateHash, InvariantResult)> {
908        self.records
909            .iter()
910            .map(|record| (record.predicate_hash.clone(), record.result.clone()))
911            .collect()
912    }
913
914    pub fn is_allowed(&self) -> bool {
915        self.records
916            .iter()
917            .filter(|record| record.enforced)
918            .all(|record| !record.result.is_blocking())
919    }
920}
921
922fn hash_result(result: &InvariantResult) -> Option<String> {
923    let bytes = serde_json::to_vec(result).ok()?;
924    Some(hex::encode(Sha256::digest(bytes)))
925}
926
927fn stable_hash(bytes: &[u8]) -> String {
928    format!("sha256:{}", hex::encode(Sha256::digest(bytes)))
929}
930
931fn stricter_result(left: &InvariantResult, right: &InvariantResult) -> InvariantResult {
932    if verdict_strictness(&left.verdict) >= verdict_strictness(&right.verdict) {
933        left.clone()
934    } else {
935        right.clone()
936    }
937}
938
939fn estimate_tokens(value: &str) -> u64 {
940    value.split_whitespace().count().max(1) as u64
941}
942
943fn default_true() -> bool {
944    true
945}
946
947fn clamp_permits(value: usize) -> usize {
948    value.clamp(1, Semaphore::MAX_PERMITS)
949}
950
951/// Short-circuit return for predicates whose slice envelope is already
952/// exhausted. Returns `None` when the envelope still has headroom (or is
953/// disabled by a zero budget).
954fn envelope_exhausted_attempt(
955    envelope: &SliceEnvelope,
956    kind: PredicateKind,
957    when: &str,
958) -> Option<PredicateAttempt> {
959    let remaining = envelope.remaining(kind)?;
960    if !remaining.is_zero() {
961        return None;
962    }
963    Some(PredicateAttempt {
964        result: InvariantResult::block(InvariantBlockError::budget_exceeded(format!(
965            "slice {kind:?} envelope exhausted {when}"
966        ))),
967        semantic_audit: None,
968    })
969}
970
971struct PredicateAttempt {
972    result: InvariantResult,
973    semantic_audit: Option<SemanticReplayAuditMetadata>,
974}
975
976/// Per-slice lane semaphores. Pairs a global (cross-slice) semaphore with a
977/// per-slice semaphore so a single slice cannot occupy every global lane.
978#[derive(Clone)]
979struct SliceLanes {
980    deterministic_global: Arc<Semaphore>,
981    deterministic_local: Arc<Semaphore>,
982    semantic_global: Arc<Semaphore>,
983    semantic_local: Arc<Semaphore>,
984}
985
986impl SliceLanes {
987    fn new(
988        deterministic_global: Arc<Semaphore>,
989        semantic_global: Arc<Semaphore>,
990        deterministic_per_slice: usize,
991        semantic_per_slice: usize,
992    ) -> Self {
993        Self {
994            deterministic_global,
995            deterministic_local: Arc::new(Semaphore::new(clamp_permits(deterministic_per_slice))),
996            semantic_global,
997            semantic_local: Arc::new(Semaphore::new(clamp_permits(semantic_per_slice))),
998        }
999    }
1000
1001    async fn acquire(&self, kind: PredicateKind) -> LaneTickets {
1002        let (global, local) = match kind {
1003            PredicateKind::Deterministic => (&self.deterministic_global, &self.deterministic_local),
1004            PredicateKind::Semantic => (&self.semantic_global, &self.semantic_local),
1005        };
1006        // Take per-slice first to preserve global FIFO admission. Acquiring
1007        // the global permit first would let one slice park its waiters at the
1008        // head of the global queue and starve other slices' lanes.
1009        let local_ticket = local
1010            .clone()
1011            .acquire_owned()
1012            .await
1013            .expect("predicate lane semaphore closed");
1014        let global_ticket = global
1015            .clone()
1016            .acquire_owned()
1017            .await
1018            .expect("predicate lane semaphore closed");
1019        LaneTickets {
1020            _local: local_ticket,
1021            _global: global_ticket,
1022        }
1023    }
1024}
1025
1026/// RAII permit holder. Both permits release when the value drops.
1027struct LaneTickets {
1028    _local: tokio::sync::OwnedSemaphorePermit,
1029    _global: tokio::sync::OwnedSemaphorePermit,
1030}
1031
1032/// Aggregate per-kind wall-clock counters for one slice's predicate envelope.
1033#[derive(Clone)]
1034struct SliceEnvelope {
1035    deterministic_used: Arc<Mutex<Duration>>,
1036    semantic_used: Arc<Mutex<Duration>>,
1037    deterministic_budget: Duration,
1038    semantic_budget: Duration,
1039}
1040
1041impl SliceEnvelope {
1042    fn new(deterministic_budget: Duration, semantic_budget: Duration) -> Self {
1043        Self {
1044            deterministic_used: Arc::new(Mutex::new(Duration::ZERO)),
1045            semantic_used: Arc::new(Mutex::new(Duration::ZERO)),
1046            deterministic_budget,
1047            semantic_budget,
1048        }
1049    }
1050
1051    fn counter(&self, kind: PredicateKind) -> &Mutex<Duration> {
1052        match kind {
1053            PredicateKind::Deterministic => &self.deterministic_used,
1054            PredicateKind::Semantic => &self.semantic_used,
1055        }
1056    }
1057
1058    fn budget(&self, kind: PredicateKind) -> Duration {
1059        match kind {
1060            PredicateKind::Deterministic => self.deterministic_budget,
1061            PredicateKind::Semantic => self.semantic_budget,
1062        }
1063    }
1064
1065    fn remaining(&self, kind: PredicateKind) -> Option<Duration> {
1066        let budget = self.budget(kind);
1067        if budget.is_zero() {
1068            return None;
1069        }
1070        let used = *self.counter(kind).lock().expect("slice envelope lock");
1071        Some(budget.saturating_sub(used))
1072    }
1073
1074    fn charge(&self, kind: PredicateKind, elapsed: Duration) {
1075        let mut used = self.counter(kind).lock().expect("slice envelope lock");
1076        *used = used.saturating_add(elapsed);
1077    }
1078}
1079
1080#[cfg(test)]
1081#[path = "executor_tests.rs"]
1082mod tests;
harn_vm/flow/predicates/executor.rs

harn_vm/flow/predicates/
executor.rs
