termivar_scanner/adaptive/
pipeline.rs

1//! Outcome-driven adaptive pipeline decisions.
2//!
3//! This module converts immutable verification outcomes and knowledge
4//! snapshots into declarative runner directives. It does not execute plugins,
5//! send probes, sleep, or mutate the knowledge base.
6
7use std::collections::{BTreeMap, BTreeSet};
8
9use serde::{Deserialize, Deserializer, Serialize};
10use termivar_core::{
11    EntityId, EvidenceId, EvidenceValue, HttpEvidencePredicate, Outcome, OutcomeStatus,
12    ReasoningModelError, VerificationStage,
13};
14use thiserror::Error;
15
16use crate::{Expression, ExpressionEvaluation, KnowledgeLayer, KnowledgeSnapshot, RuleEngineError};
17
18/// Validation and consistency errors raised by adaptive decisions.
19#[derive(Debug, Error)]
20#[non_exhaustive]
21pub enum AdaptivePipelineError {
22    /// A required rule, action, or explanation value was empty.
23    #[error("{field} must not be empty")]
24    EmptyValue { field: &'static str },
25
26    /// An outcome selector contained no statuses.
27    #[error("outcome selector must contain at least one status")]
28    EmptyStatuses,
29
30    /// An outcome selector contained no verification stages.
31    #[error("outcome selector must contain at least one verification stage")]
32    EmptyStages,
33
34    /// A rule application limit was zero.
35    #[error("adaptation rule {rule_id} must allow at least one application")]
36    ZeroRuleApplications { rule_id: String },
37
38    /// An invalid schedule directive contained no action identity.
39    #[error("scheduled action id must not be empty")]
40    EmptyScheduledAction,
41
42    /// A throttle directive had no delay.
43    #[error("throttle delay must be greater than zero")]
44    ZeroThrottleDelay,
45
46    /// A rule identity was reused with different semantics.
47    #[error("adaptation rule identity {id} already has a different definition")]
48    RuleIdentityConflict { id: String },
49
50    /// The outcome and knowledge snapshot refer to different subjects.
51    #[error("adaptive outcome subject {expected} does not match snapshot subject {actual}")]
52    SnapshotSubjectMismatch {
53        /// Subject declared by the outcome.
54        expected: EntityId,
55        /// Subject captured by the snapshot.
56        actual: EntityId,
57    },
58
59    /// Outcome provenance was absent from the supplied snapshot.
60    #[error("adaptive snapshot is missing outcome evidence {evidence_id}")]
61    MissingOutcomeEvidence { evidence_id: EvidenceId },
62
63    /// A matched conditional rule cited no immutable evidence.
64    #[error("matched adaptation rule {rule_id} did not cite any evidence")]
65    MissingContributingEvidence { rule_id: String },
66
67    /// Global transition limits must be positive.
68    #[error("maximum adaptive transitions must be greater than zero")]
69    ZeroTransitions,
70
71    /// Per-action scheduling limits must be positive.
72    #[error("maximum action schedules must be greater than zero")]
73    ZeroActionSchedules,
74
75    /// A decision was recorded twice or out of order.
76    #[error("adaptive decision sequence {actual} does not match ledger sequence {expected}")]
77    DecisionSequenceMismatch { expected: u32, actual: u32 },
78
79    /// A ledger counter could not be incremented safely.
80    #[error("adaptive ledger counter overflowed")]
81    CounterOverflow,
82
83    /// Persisted ledger state contained an invalid identity or counter.
84    #[error("invalid adaptive ledger entry for {field}")]
85    InvalidLedgerEntry { field: &'static str },
86
87    /// Expression evaluation failed.
88    #[error(transparent)]
89    Rule(#[from] RuleEngineError),
90
91    /// A core reasoning-domain value was invalid.
92    #[error(transparent)]
93    Reasoning(#[from] ReasoningModelError),
94}
95
96fn non_empty(
97    value: impl Into<String>,
98    field: &'static str,
99) -> Result<String, AdaptivePipelineError> {
100    let value = value.into();
101    if value.trim().is_empty() {
102        return Err(AdaptivePipelineError::EmptyValue { field });
103    }
104    Ok(value)
105}
106
107fn is_false(value: &bool) -> bool {
108    !*value
109}
110
111/// Outcome statuses and verification stages accepted by one policy rule.
112#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
113pub struct OutcomeSelector {
114    statuses: BTreeSet<OutcomeStatus>,
115    stages: BTreeSet<VerificationStage>,
116}
117
118impl OutcomeSelector {
119    /// Creates a non-empty status and stage selector.
120    pub fn new(
121        statuses: BTreeSet<OutcomeStatus>,
122        stages: BTreeSet<VerificationStage>,
123    ) -> Result<Self, AdaptivePipelineError> {
124        if statuses.is_empty() {
125            return Err(AdaptivePipelineError::EmptyStatuses);
126        }
127        if stages.is_empty() {
128            return Err(AdaptivePipelineError::EmptyStages);
129        }
130        Ok(Self { statuses, stages })
131    }
132
133    /// Selects statuses produced by either verification stage.
134    pub fn any_stage(statuses: BTreeSet<OutcomeStatus>) -> Result<Self, AdaptivePipelineError> {
135        Self::new(
136            statuses,
137            BTreeSet::from([VerificationStage::Passive, VerificationStage::Active]),
138        )
139    }
140
141    /// Returns accepted outcome statuses.
142    pub fn statuses(&self) -> &BTreeSet<OutcomeStatus> {
143        &self.statuses
144    }
145
146    /// Returns accepted verification stages.
147    pub fn stages(&self) -> &BTreeSet<VerificationStage> {
148        &self.stages
149    }
150
151    /// Returns whether this selector accepts an outcome.
152    pub fn matches(&self, outcome: &Outcome) -> bool {
153        self.statuses.contains(&outcome.status()) && self.stages.contains(&outcome.stage())
154    }
155}
156
157impl<'de> Deserialize<'de> for OutcomeSelector {
158    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
159    where
160        D: Deserializer<'de>,
161    {
162        #[derive(Deserialize)]
163        #[serde(deny_unknown_fields)]
164        struct WireSelector {
165            statuses: BTreeSet<OutcomeStatus>,
166            stages: BTreeSet<VerificationStage>,
167        }
168
169        let wire = WireSelector::deserialize(deserializer)?;
170        Self::new(wire.statuses, wire.stages).map_err(serde::de::Error::custom)
171    }
172}
173
174/// Side-effect-free command emitted for the runner or scheduler.
175#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
176#[serde(tag = "directive", rename_all = "snake_case")]
177#[non_exhaustive]
178pub enum PipelineDirective {
179    /// The verified action completed the current objective.
180    Complete,
181    /// Add a named planner or plugin action to the execution queue.
182    ScheduleAction {
183        /// Stable action identity understood by the runner.
184        action_id: String,
185    },
186    /// Re-run planning, optionally excluding the action that produced the outcome.
187    Replan {
188        /// Whether the source action should be suppressed in the next plan.
189        suppress_current_action: bool,
190    },
191    /// Apply backpressure before optionally retrying the source action.
192    Throttle {
193        /// Required delay in milliseconds.
194        delay_ms: u64,
195        /// Whether the runner should retry the source action after the delay.
196        retry_current_action: bool,
197    },
198    /// Collect active verification evidence before another decision.
199    AwaitActiveVerification,
200    /// Preserve the case for a human decision.
201    AwaitHumanReview,
202    /// Stop adaptation because the global transition budget was exhausted.
203    Halt,
204}
205
206impl<'de> Deserialize<'de> for PipelineDirective {
207    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
208    where
209        D: Deserializer<'de>,
210    {
211        #[derive(Deserialize)]
212        #[serde(tag = "directive", rename_all = "snake_case", deny_unknown_fields)]
213        enum WireDirective {
214            Complete {},
215            ScheduleAction {
216                action_id: String,
217            },
218            Replan {
219                suppress_current_action: bool,
220            },
221            Throttle {
222                delay_ms: u64,
223                retry_current_action: bool,
224            },
225            AwaitActiveVerification {},
226            AwaitHumanReview {},
227            Halt {},
228        }
229
230        let directive = match WireDirective::deserialize(deserializer)? {
231            WireDirective::Complete {} => Self::Complete,
232            WireDirective::ScheduleAction { action_id } => Self::ScheduleAction { action_id },
233            WireDirective::Replan {
234                suppress_current_action,
235            } => Self::Replan {
236                suppress_current_action,
237            },
238            WireDirective::Throttle {
239                delay_ms,
240                retry_current_action,
241            } => Self::Throttle {
242                delay_ms,
243                retry_current_action,
244            },
245            WireDirective::AwaitActiveVerification {} => Self::AwaitActiveVerification,
246            WireDirective::AwaitHumanReview {} => Self::AwaitHumanReview,
247            WireDirective::Halt {} => Self::Halt,
248        };
249        directive.validate().map_err(serde::de::Error::custom)?;
250        Ok(directive)
251    }
252}
253
254impl PipelineDirective {
255    fn validate(&self) -> Result<(), AdaptivePipelineError> {
256        match self {
257            Self::ScheduleAction { action_id } if action_id.trim().is_empty() => {
258                Err(AdaptivePipelineError::EmptyScheduledAction)
259            },
260            Self::Throttle { delay_ms: 0, .. } => Err(AdaptivePipelineError::ZeroThrottleDelay),
261            Self::Complete
262            | Self::ScheduleAction { .. }
263            | Self::Replan { .. }
264            | Self::Throttle { .. }
265            | Self::AwaitActiveVerification
266            | Self::AwaitHumanReview
267            | Self::Halt => Ok(()),
268        }
269    }
270
271    /// Returns the explicitly scheduled action, if any.
272    pub fn scheduled_action_id(&self) -> Option<&str> {
273        match self {
274            Self::ScheduleAction { action_id } => Some(action_id),
275            _ => None,
276        }
277    }
278
279    fn repeated_action_id<'a>(&'a self, source_action_id: &'a str) -> Option<&'a str> {
280        match self {
281            Self::ScheduleAction { action_id } => Some(action_id),
282            Self::Throttle {
283                retry_current_action: true,
284                ..
285            } => Some(source_action_id),
286            _ => None,
287        }
288    }
289}
290
291/// Declarative mapping from an outcome and optional evidence expression to a directive.
292#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
293pub struct AdaptationRule {
294    id: String,
295    selector: OutcomeSelector,
296    priority: u16,
297    condition: Option<Expression>,
298    #[serde(skip_serializing_if = "is_false")]
299    condition_policy_guard: bool,
300    directive: PipelineDirective,
301    rationale: String,
302    max_applications: u16,
303}
304
305impl AdaptationRule {
306    /// Creates a validated adaptation rule.
307    #[allow(clippy::too_many_arguments)]
308    pub fn new(
309        id: impl Into<String>,
310        selector: OutcomeSelector,
311        priority: u16,
312        condition: Option<Expression>,
313        directive: PipelineDirective,
314        rationale: impl Into<String>,
315        max_applications: u16,
316    ) -> Result<Self, AdaptivePipelineError> {
317        let id = non_empty(id, "adaptation rule id")?;
318        if max_applications == 0 {
319            return Err(AdaptivePipelineError::ZeroRuleApplications { rule_id: id });
320        }
321        directive.validate()?;
322        Ok(Self {
323            id,
324            selector,
325            priority,
326            condition_policy_guard: condition.is_some(),
327            condition,
328            directive,
329            rationale: non_empty(rationale, "adaptation rationale")?,
330            max_applications,
331        })
332    }
333
334    /// Returns the stable rule identity.
335    pub fn id(&self) -> &str {
336        &self.id
337    }
338
339    /// Returns the accepted outcome selector.
340    pub fn selector(&self) -> &OutcomeSelector {
341        &self.selector
342    }
343
344    /// Returns the deterministic conflict-resolution priority.
345    pub fn priority(&self) -> u16 {
346        self.priority
347    }
348
349    /// Returns the optional evidence condition.
350    pub fn condition(&self) -> Option<&Expression> {
351        self.condition.as_ref()
352    }
353
354    /// Returns the emitted directive template.
355    pub fn directive(&self) -> &PipelineDirective {
356        &self.directive
357    }
358
359    /// Returns the human-readable policy explanation.
360    pub fn rationale(&self) -> &str {
361        &self.rationale
362    }
363
364    /// Returns the maximum number of times this rule may win in one ledger.
365    pub fn max_applications(&self) -> u16 {
366        self.max_applications
367    }
368}
369
370impl<'de> Deserialize<'de> for AdaptationRule {
371    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
372    where
373        D: Deserializer<'de>,
374    {
375        #[derive(Deserialize)]
376        #[serde(transparent)]
377        struct RequiredCondition(Option<Expression>);
378
379        #[derive(Deserialize)]
380        #[serde(deny_unknown_fields)]
381        struct WireRule {
382            id: String,
383            selector: OutcomeSelector,
384            priority: u16,
385            condition: RequiredCondition,
386            #[serde(default)]
387            condition_policy_guard: Option<bool>,
388            directive: PipelineDirective,
389            rationale: String,
390            max_applications: u16,
391        }
392
393        let wire = WireRule::deserialize(deserializer)?;
394        if let Some(condition_policy_guard) = wire.condition_policy_guard {
395            if !condition_policy_guard || wire.condition.0.is_none() {
396                return Err(serde::de::Error::custom(
397                    "adaptation rule condition guard is inconsistent",
398                ));
399            }
400        }
401        Self::new(
402            wire.id,
403            wire.selector,
404            wire.priority,
405            wire.condition.0,
406            wire.directive,
407            wire.rationale,
408            wire.max_applications,
409        )
410        .map_err(serde::de::Error::custom)
411    }
412}
413
414/// Result of registering an adaptation rule identity.
415#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
416#[serde(rename_all = "snake_case")]
417#[non_exhaustive]
418pub enum AdaptiveRuleWrite {
419    /// A new policy rule was registered.
420    Inserted,
421    /// The identical policy rule was already registered.
422    Unchanged,
423}
424
425/// Limits applied across an adaptive decision ledger.
426#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
427pub struct AdaptationLimits {
428    max_transitions: u32,
429    max_action_schedules: u16,
430}
431
432impl AdaptationLimits {
433    /// Creates positive global transition and action scheduling limits.
434    pub fn new(
435        max_transitions: u32,
436        max_action_schedules: u16,
437    ) -> Result<Self, AdaptivePipelineError> {
438        if max_transitions == 0 {
439            return Err(AdaptivePipelineError::ZeroTransitions);
440        }
441        if max_action_schedules == 0 {
442            return Err(AdaptivePipelineError::ZeroActionSchedules);
443        }
444        Ok(Self {
445            max_transitions,
446            max_action_schedules,
447        })
448    }
449
450    /// Returns the maximum recorded directives.
451    pub fn max_transitions(self) -> u32 {
452        self.max_transitions
453    }
454
455    /// Returns the maximum times one action may be scheduled.
456    pub fn max_action_schedules(self) -> u16 {
457        self.max_action_schedules
458    }
459}
460
461impl Default for AdaptationLimits {
462    fn default() -> Self {
463        Self {
464            max_transitions: 64,
465            max_action_schedules: 3,
466        }
467    }
468}
469
470impl<'de> Deserialize<'de> for AdaptationLimits {
471    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
472    where
473        D: Deserializer<'de>,
474    {
475        #[derive(Deserialize)]
476        struct WireLimits {
477            max_transitions: u32,
478            max_action_schedules: u16,
479        }
480
481        let wire = WireLimits::deserialize(deserializer)?;
482        Self::new(wire.max_transitions, wire.max_action_schedules).map_err(serde::de::Error::custom)
483    }
484}
485
486/// Replayable counters and suppressions for one adaptive scan session.
487#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
488pub struct AdaptationLedger {
489    transitions: u32,
490    rule_applications: BTreeMap<String, u16>,
491    action_schedules: BTreeMap<String, u16>,
492    suppressed_actions: BTreeSet<String>,
493}
494
495impl AdaptationLedger {
496    /// Creates an empty ledger.
497    pub fn new() -> Self {
498        Self::default()
499    }
500
501    /// Returns the number of recorded directives.
502    pub fn transitions(&self) -> u32 {
503        self.transitions
504    }
505
506    /// Returns how often a rule produced a directive.
507    pub fn rule_applications(&self, rule_id: &str) -> u16 {
508        self.rule_applications.get(rule_id).copied().unwrap_or(0)
509    }
510
511    /// Returns how often an explicit action was scheduled.
512    pub fn action_schedules(&self, action_id: &str) -> u16 {
513        self.action_schedules.get(action_id).copied().unwrap_or(0)
514    }
515
516    /// Returns action identities suppressed from subsequent replanning.
517    pub fn suppressed_actions(&self) -> &BTreeSet<String> {
518        &self.suppressed_actions
519    }
520
521    /// Records one decision exactly once and in sequence order.
522    pub fn record(&mut self, decision: &AdaptiveDecision) -> Result<(), AdaptivePipelineError> {
523        if decision.sequence != self.transitions {
524            return Err(AdaptivePipelineError::DecisionSequenceMismatch {
525                expected: self.transitions,
526                actual: decision.sequence,
527            });
528        }
529        if decision.transition_limit_reached {
530            return Ok(());
531        }
532        let mut candidate = self.clone();
533        candidate.transitions = candidate
534            .transitions
535            .checked_add(1)
536            .ok_or(AdaptivePipelineError::CounterOverflow)?;
537        if let Some(rule_id) = decision.selected_rule_id() {
538            increment(&mut candidate.rule_applications, rule_id)?;
539        }
540        if let Some(action_id) = decision.directive().scheduled_action_id() {
541            increment(&mut candidate.action_schedules, action_id)?;
542        }
543        if matches!(
544            decision.directive(),
545            PipelineDirective::Replan {
546                suppress_current_action: true
547            }
548        ) {
549            candidate
550                .suppressed_actions
551                .insert(decision.source_action_id.clone());
552        }
553        *self = candidate;
554        Ok(())
555    }
556}
557
558impl<'de> Deserialize<'de> for AdaptationLedger {
559    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
560    where
561        D: Deserializer<'de>,
562    {
563        #[derive(Deserialize)]
564        struct WireLedger {
565            transitions: u32,
566            rule_applications: BTreeMap<String, u16>,
567            action_schedules: BTreeMap<String, u16>,
568            suppressed_actions: BTreeSet<String>,
569        }
570
571        let wire = WireLedger::deserialize(deserializer)?;
572        validate_counter_map(&wire.rule_applications, "rule applications")
573            .map_err(serde::de::Error::custom)?;
574        validate_counter_map(&wire.action_schedules, "action schedules")
575            .map_err(serde::de::Error::custom)?;
576        let transitions = u64::from(wire.transitions);
577        let rule_applications: u64 = wire
578            .rule_applications
579            .values()
580            .map(|count| u64::from(*count))
581            .sum();
582        let action_schedules: u64 = wire
583            .action_schedules
584            .values()
585            .map(|count| u64::from(*count))
586            .sum();
587        if rule_applications > transitions
588            || action_schedules > transitions
589            || wire.suppressed_actions.len() as u64 > transitions
590        {
591            return Err(serde::de::Error::custom(
592                AdaptivePipelineError::InvalidLedgerEntry {
593                    field: "counters exceed transitions",
594                },
595            ));
596        }
597        if wire
598            .suppressed_actions
599            .iter()
600            .any(|action_id| action_id.trim().is_empty())
601        {
602            return Err(serde::de::Error::custom(
603                AdaptivePipelineError::InvalidLedgerEntry {
604                    field: "suppressed actions",
605                },
606            ));
607        }
608        Ok(Self {
609            transitions: wire.transitions,
610            rule_applications: wire.rule_applications,
611            action_schedules: wire.action_schedules,
612            suppressed_actions: wire.suppressed_actions,
613        })
614    }
615}
616
617fn validate_counter_map(
618    counters: &BTreeMap<String, u16>,
619    field: &'static str,
620) -> Result<(), AdaptivePipelineError> {
621    if counters
622        .iter()
623        .any(|(identity, count)| identity.trim().is_empty() || *count == 0)
624    {
625        return Err(AdaptivePipelineError::InvalidLedgerEntry { field });
626    }
627    Ok(())
628}
629
630fn increment(
631    counters: &mut BTreeMap<String, u16>,
632    identity: &str,
633) -> Result<(), AdaptivePipelineError> {
634    let next = counters
635        .get(identity)
636        .copied()
637        .unwrap_or(0)
638        .checked_add(1)
639        .ok_or(AdaptivePipelineError::CounterOverflow)?;
640    counters.insert(identity.to_owned(), next);
641    Ok(())
642}
643
644/// Explainable evaluation of one adaptation policy.
645#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
646pub struct AdaptationRuleEvaluation {
647    rule_id: String,
648    selector_matched: bool,
649    condition: Option<ExpressionEvaluation>,
650    outcome_evidence_matched: bool,
651    rule_limit_exhausted: bool,
652    action_limit_exhausted: bool,
653    policy_suppressed: bool,
654    eligible: bool,
655    selected: bool,
656}
657
658impl AdaptationRuleEvaluation {
659    /// Returns the evaluated rule identity.
660    pub fn rule_id(&self) -> &str {
661        &self.rule_id
662    }
663
664    /// Returns whether outcome status and stage matched.
665    pub fn selector_matched(&self) -> bool {
666        self.selector_matched
667    }
668
669    /// Returns the optional evidence expression trace.
670    pub fn condition(&self) -> Option<&ExpressionEvaluation> {
671        self.condition.as_ref()
672    }
673
674    /// Returns whether conditional evidence belongs to the source outcome.
675    pub fn outcome_evidence_matched(&self) -> bool {
676        self.outcome_evidence_matched
677    }
678
679    /// Returns whether the rule's retry budget was exhausted.
680    pub fn rule_limit_exhausted(&self) -> bool {
681        self.rule_limit_exhausted
682    }
683
684    /// Returns whether the scheduled action reached its global limit.
685    pub fn action_limit_exhausted(&self) -> bool {
686        self.action_limit_exhausted
687    }
688
689    /// Returns whether learned, adaptive, or operator policy suppressed the repeated action.
690    pub fn policy_suppressed(&self) -> bool {
691        self.policy_suppressed
692    }
693
694    /// Returns whether the rule could participate in winner selection.
695    pub fn eligible(&self) -> bool {
696        self.eligible
697    }
698
699    /// Returns whether this rule produced the directive.
700    pub fn selected(&self) -> bool {
701        self.selected
702    }
703}
704
705/// Immutable adaptive decision for one verification outcome.
706#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
707pub struct AdaptiveDecision {
708    sequence: u32,
709    subject: EntityId,
710    case_id: String,
711    source_action_id: String,
712    outcome_status: OutcomeStatus,
713    outcome_stage: VerificationStage,
714    selected_rule_id: Option<String>,
715    directive: PipelineDirective,
716    rationale: String,
717    evaluations: Vec<AdaptationRuleEvaluation>,
718    transition_limit_reached: bool,
719}
720
721impl AdaptiveDecision {
722    /// Returns the zero-based ledger sequence.
723    pub fn sequence(&self) -> u32 {
724        self.sequence
725    }
726
727    /// Returns the decision subject.
728    pub fn subject(&self) -> &EntityId {
729        &self.subject
730    }
731
732    /// Returns the verification case that triggered adaptation.
733    pub fn case_id(&self) -> &str {
734        &self.case_id
735    }
736
737    /// Returns the action that produced the source outcome.
738    pub fn source_action_id(&self) -> &str {
739        &self.source_action_id
740    }
741
742    /// Returns the source outcome classification.
743    pub fn outcome_status(&self) -> OutcomeStatus {
744        self.outcome_status
745    }
746
747    /// Returns the source verification stage.
748    pub fn outcome_stage(&self) -> VerificationStage {
749        self.outcome_stage
750    }
751
752    /// Returns the winning policy identity, if a policy matched.
753    pub fn selected_rule_id(&self) -> Option<&str> {
754        self.selected_rule_id.as_deref()
755    }
756
757    /// Returns the side-effect-free runner directive.
758    pub fn directive(&self) -> &PipelineDirective {
759        &self.directive
760    }
761
762    /// Returns the decision explanation.
763    pub fn rationale(&self) -> &str {
764        &self.rationale
765    }
766
767    /// Returns policy evaluations in stable rule-ID order.
768    pub fn evaluations(&self) -> &[AdaptationRuleEvaluation] {
769        &self.evaluations
770    }
771
772    /// Returns whether the global transition budget forced a halt.
773    pub fn transition_limit_reached(&self) -> bool {
774        self.transition_limit_reached
775    }
776}
777
778/// Deterministic outcome-to-directive policy engine.
779///
780/// # Example
781///
782/// ```rust
783/// use termivar_scanner::AdaptivePipeline;
784///
785/// let pipeline = AdaptivePipeline::with_standard_policies()?;
786/// assert_eq!(pipeline.len(), 3);
787/// # Ok::<(), Box<dyn std::error::Error>>(())
788/// ```
789#[derive(Debug, Clone, Default)]
790pub struct AdaptivePipeline {
791    rules: BTreeMap<String, AdaptationRule>,
792}
793
794impl AdaptivePipeline {
795    /// Creates an empty policy engine with deterministic fallback behavior.
796    pub fn new() -> Self {
797        Self::default()
798    }
799
800    /// Creates policies for common HTTP control-flow signals.
801    ///
802    /// - `403` schedules `http.403-bypass`;
803    /// - `404` schedules `http.enumeration`;
804    /// - `429` throttles for two seconds and retries the current action.
805    ///
806    /// The pipeline only proposes directives. A [`crate::DecisionLoop`] grants
807    /// execution authority separately: scheduled actions must be registered,
808    /// satisfy current planner policy, and be submitted with an explicit host
809    /// suppression context. These example action IDs are not implicit executor
810    /// registrations.
811    pub fn with_standard_policies() -> Result<Self, AdaptivePipelineError> {
812        let mut pipeline = Self::new();
813        let unresolved = OutcomeSelector::any_stage(BTreeSet::from([
814            OutcomeStatus::Blocked,
815            OutcomeStatus::Unknown,
816            OutcomeStatus::NeedsReview,
817        ]))?;
818        let status = HttpEvidencePredicate::RESPONSE_STATUS.into_knowledge();
819        pipeline.register(AdaptationRule::new(
820            "http.429.throttle",
821            unresolved.clone(),
822            300,
823            Some(Expression::equals(
824                KnowledgeLayer::Evidence,
825                status.clone(),
826                EvidenceValue::Unsigned(429),
827            )),
828            PipelineDirective::Throttle {
829                delay_ms: 2_000,
830                retry_current_action: true,
831            },
832            "rate limiting requires backpressure before retry",
833            3,
834        )?)?;
835        pipeline.register(AdaptationRule::new(
836            "http.403.bypass",
837            unresolved.clone(),
838            200,
839            Some(Expression::equals(
840                KnowledgeLayer::Evidence,
841                status.clone(),
842                EvidenceValue::Unsigned(403),
843            )),
844            PipelineDirective::ScheduleAction {
845                action_id: "http.403-bypass".into(),
846            },
847            "access control response requires a dedicated bypass action",
848            2,
849        )?)?;
850        pipeline.register(AdaptationRule::new(
851            "http.404.enumerate",
852            unresolved,
853            100,
854            Some(Expression::equals(
855                KnowledgeLayer::Evidence,
856                status,
857                EvidenceValue::Unsigned(404),
858            )),
859            PipelineDirective::ScheduleAction {
860                action_id: "http.enumeration".into(),
861            },
862            "missing resource response redirects discovery to enumeration",
863            1,
864        )?)?;
865        Ok(pipeline)
866    }
867
868    /// Registers an idempotent policy rule.
869    pub fn register(
870        &mut self,
871        rule: AdaptationRule,
872    ) -> Result<AdaptiveRuleWrite, AdaptivePipelineError> {
873        if let Some(existing) = self.rules.get(rule.id()) {
874            return if existing == &rule {
875                Ok(AdaptiveRuleWrite::Unchanged)
876            } else {
877                Err(AdaptivePipelineError::RuleIdentityConflict {
878                    id: rule.id.clone(),
879                })
880            };
881        }
882        self.rules.insert(rule.id.clone(), rule);
883        Ok(AdaptiveRuleWrite::Inserted)
884    }
885
886    /// Returns the number of registered policies.
887    pub fn len(&self) -> usize {
888        self.rules.len()
889    }
890
891    /// Returns whether no policies are registered.
892    pub fn is_empty(&self) -> bool {
893        self.rules.is_empty()
894    }
895
896    /// Produces a pure decision without changing the supplied ledger.
897    pub fn decide(
898        &self,
899        outcome: &Outcome,
900        snapshot: &KnowledgeSnapshot,
901        ledger: &AdaptationLedger,
902        limits: AdaptationLimits,
903    ) -> Result<AdaptiveDecision, AdaptivePipelineError> {
904        self.decide_with_suppressed_actions(outcome, snapshot, ledger, limits, &BTreeSet::new())
905    }
906
907    /// Produces a pure decision while excluding actions suppressed by external policy.
908    ///
909    /// The set may be derived from an [`crate::ExperienceStore`], operator
910    /// policy, or another scheduler concern. Ledger suppressions are always
911    /// merged with the supplied set.
912    pub fn decide_with_suppressed_actions(
913        &self,
914        outcome: &Outcome,
915        snapshot: &KnowledgeSnapshot,
916        ledger: &AdaptationLedger,
917        limits: AdaptationLimits,
918        suppressed_actions: &BTreeSet<String>,
919    ) -> Result<AdaptiveDecision, AdaptivePipelineError> {
920        validate_snapshot(outcome, snapshot)?;
921        if ledger.transitions >= limits.max_transitions {
922            return Ok(AdaptiveDecision {
923                sequence: ledger.transitions,
924                subject: outcome.subject().clone(),
925                case_id: outcome.case_id().to_owned(),
926                source_action_id: outcome.action_id().to_owned(),
927                outcome_status: outcome.status(),
928                outcome_stage: outcome.stage(),
929                selected_rule_id: None,
930                directive: PipelineDirective::Halt,
931                rationale: "adaptive transition budget exhausted".into(),
932                evaluations: Vec::new(),
933                transition_limit_reached: true,
934            });
935        }
936
937        let mut evaluations = Vec::with_capacity(self.rules.len());
938        for rule in self.rules.values() {
939            let selector_matched = rule.selector.matches(outcome);
940            let condition = if selector_matched {
941                rule.condition
942                    .as_ref()
943                    .map(|condition| condition.evaluate(snapshot))
944                    .transpose()?
945            } else {
946                None
947            };
948            if condition
949                .as_ref()
950                .is_some_and(|condition| condition.matched() && condition.evidence_ids().is_empty())
951            {
952                return Err(AdaptivePipelineError::MissingContributingEvidence {
953                    rule_id: rule.id.clone(),
954                });
955            }
956            let condition_matched = condition
957                .as_ref()
958                .map_or(rule.condition.is_none(), ExpressionEvaluation::matched);
959            let outcome_evidence_matched = condition.as_ref().is_none_or(|condition| {
960                !condition.evidence_ids().is_disjoint(outcome.evidence_ids())
961            });
962            let rule_limit_exhausted = ledger.rule_applications(rule.id()) >= rule.max_applications;
963            let action_limit_exhausted =
964                rule.directive
965                    .scheduled_action_id()
966                    .is_some_and(|action_id| {
967                        ledger.action_schedules(action_id) >= limits.max_action_schedules
968                    });
969            let policy_suppressed = rule
970                .directive
971                .repeated_action_id(outcome.action_id())
972                .is_some_and(|action_id| {
973                    suppressed_actions.contains(action_id)
974                        || ledger.suppressed_actions().contains(action_id)
975                });
976            let eligible = selector_matched
977                && condition_matched
978                && outcome_evidence_matched
979                && !rule_limit_exhausted
980                && !action_limit_exhausted
981                && !policy_suppressed;
982            evaluations.push(AdaptationRuleEvaluation {
983                rule_id: rule.id.clone(),
984                selector_matched,
985                condition,
986                outcome_evidence_matched,
987                rule_limit_exhausted,
988                action_limit_exhausted,
989                policy_suppressed,
990                eligible,
991                selected: false,
992            });
993        }
994
995        let mut candidates: Vec<_> = evaluations
996            .iter()
997            .filter(|evaluation| evaluation.eligible)
998            .map(|evaluation| evaluation.rule_id.clone())
999            .collect();
1000        candidates.sort_by(|left, right| {
1001            self.rules[right]
1002                .priority
1003                .cmp(&self.rules[left].priority)
1004                .then_with(|| left.cmp(right))
1005        });
1006        let selected_rule_id = candidates.first().cloned();
1007        if let Some(selected) = &selected_rule_id {
1008            if let Some(evaluation) = evaluations
1009                .iter_mut()
1010                .find(|evaluation| &evaluation.rule_id == selected)
1011            {
1012                evaluation.selected = true;
1013            }
1014        }
1015
1016        let (directive, rationale) = selected_rule_id
1017            .as_ref()
1018            .map(|rule_id| {
1019                let rule = &self.rules[rule_id];
1020                (rule.directive.clone(), rule.rationale.clone())
1021            })
1022            .unwrap_or_else(|| fallback(outcome));
1023
1024        Ok(AdaptiveDecision {
1025            sequence: ledger.transitions,
1026            subject: outcome.subject().clone(),
1027            case_id: outcome.case_id().to_owned(),
1028            source_action_id: outcome.action_id().to_owned(),
1029            outcome_status: outcome.status(),
1030            outcome_stage: outcome.stage(),
1031            selected_rule_id,
1032            directive,
1033            rationale,
1034            evaluations,
1035            transition_limit_reached: false,
1036        })
1037    }
1038
1039    /// Produces and atomically records one in-order adaptive decision.
1040    pub fn decide_and_record(
1041        &self,
1042        outcome: &Outcome,
1043        snapshot: &KnowledgeSnapshot,
1044        ledger: &mut AdaptationLedger,
1045        limits: AdaptationLimits,
1046    ) -> Result<AdaptiveDecision, AdaptivePipelineError> {
1047        self.decide_and_record_with_suppressed_actions(
1048            outcome,
1049            snapshot,
1050            ledger,
1051            limits,
1052            &BTreeSet::new(),
1053        )
1054    }
1055
1056    /// Produces and records a decision with explicit policy suppressions.
1057    pub fn decide_and_record_with_suppressed_actions(
1058        &self,
1059        outcome: &Outcome,
1060        snapshot: &KnowledgeSnapshot,
1061        ledger: &mut AdaptationLedger,
1062        limits: AdaptationLimits,
1063        suppressed_actions: &BTreeSet<String>,
1064    ) -> Result<AdaptiveDecision, AdaptivePipelineError> {
1065        let decision = self.decide_with_suppressed_actions(
1066            outcome,
1067            snapshot,
1068            ledger,
1069            limits,
1070            suppressed_actions,
1071        )?;
1072        ledger.record(&decision)?;
1073        Ok(decision)
1074    }
1075}
1076
1077fn fallback(outcome: &Outcome) -> (PipelineDirective, String) {
1078    match outcome.status() {
1079        OutcomeStatus::Success => (
1080            PipelineDirective::Complete,
1081            "verified objective completed".into(),
1082        ),
1083        OutcomeStatus::FalsePositive | OutcomeStatus::ConfirmedNegative => (
1084            PipelineDirective::Replan {
1085                suppress_current_action: true,
1086            },
1087            "negative conclusion suppresses the source action before replanning".into(),
1088        ),
1089        OutcomeStatus::Unknown | OutcomeStatus::NeedsReview
1090            if outcome.stage() == VerificationStage::Passive =>
1091        {
1092            (
1093                PipelineDirective::AwaitActiveVerification,
1094                "passive evidence is inconclusive".into(),
1095            )
1096        },
1097        OutcomeStatus::Blocked | OutcomeStatus::Unknown | OutcomeStatus::NeedsReview => (
1098            PipelineDirective::AwaitHumanReview,
1099            "no eligible automated adaptation policy remains".into(),
1100        ),
1101        _ => (
1102            PipelineDirective::AwaitHumanReview,
1103            "outcome status is not covered by the current adaptation policy".into(),
1104        ),
1105    }
1106}
1107
1108fn validate_snapshot(
1109    outcome: &Outcome,
1110    snapshot: &KnowledgeSnapshot,
1111) -> Result<(), AdaptivePipelineError> {
1112    if outcome.subject() != snapshot.subject() {
1113        return Err(AdaptivePipelineError::SnapshotSubjectMismatch {
1114            expected: outcome.subject().clone(),
1115            actual: snapshot.subject().clone(),
1116        });
1117    }
1118    let evidence_ids: BTreeSet<_> = snapshot
1119        .evidence()
1120        .iter()
1121        .map(|evidence| evidence.id())
1122        .collect();
1123    for evidence_id in outcome.evidence_ids() {
1124        if !evidence_ids.contains(evidence_id) {
1125            return Err(AdaptivePipelineError::MissingOutcomeEvidence {
1126                evidence_id: evidence_id.clone(),
1127            });
1128        }
1129    }
1130    Ok(())
1131}
1132
1133#[cfg(test)]
1134mod tests {
1135    use super::*;
1136    use crate::{
1137        ExperiencePolicy, ExperienceStore, PassiveVerifier, VerificationCase, VerificationRule,
1138        VerifierWrite,
1139    };
1140    use termivar_core::{
1141        BayesianEvidence, ConfidenceScore, Evidence, EvidenceKind, EvidenceSource, Hypothesis,
1142        HypothesisState, HypothesisStrength, KnowledgePredicate, Probability,
1143    };
1144
1145    struct Fixture {
1146        knowledge: crate::KnowledgeBase,
1147        evidence_id: EvidenceId,
1148    }
1149
1150    fn subject() -> EntityId {
1151        EntityId::new("endpoint:https://example.test").unwrap()
1152    }
1153
1154    fn status_predicate() -> KnowledgePredicate {
1155        HttpEvidencePredicate::RESPONSE_STATUS.into_knowledge()
1156    }
1157
1158    fn fixture(status: u64) -> Fixture {
1159        let knowledge = crate::KnowledgeBase::new();
1160        let evidence = Evidence::new(
1161            subject(),
1162            EvidenceKind::Http,
1163            status_predicate(),
1164            EvidenceValue::Unsigned(status),
1165            EvidenceSource::new("http.executor", "response-status").unwrap(),
1166            ConfidenceScore::MAX,
1167        );
1168        let evidence_id = evidence.id().clone();
1169        knowledge.insert_evidence(evidence).unwrap();
1170        let mut hypothesis = Hypothesis::with_id(
1171            "hypothesis:http",
1172            subject(),
1173            KnowledgePredicate::new("vulnerability", "candidate").unwrap(),
1174            EvidenceValue::Boolean(true),
1175            Probability::from_percent(50).unwrap(),
1176        )
1177        .unwrap();
1178        hypothesis
1179            .observe(
1180                BayesianEvidence::new(
1181                    evidence_id.clone(),
1182                    Probability::from_percent(80).unwrap(),
1183                    Probability::from_percent(20).unwrap(),
1184                    "HTTP response contributes to the candidate",
1185                )
1186                .unwrap(),
1187            )
1188            .unwrap();
1189        hypothesis.set_strength(HypothesisStrength::Strong);
1190        hypothesis.set_state(HypothesisState::Supported);
1191        knowledge.upsert_hypothesis(hypothesis).unwrap();
1192        Fixture {
1193            knowledge,
1194            evidence_id,
1195        }
1196    }
1197
1198    fn verified_outcome(
1199        status: OutcomeStatus,
1200        stage: VerificationStage,
1201        evidence_id: EvidenceId,
1202    ) -> Outcome {
1203        Outcome::verified(
1204            "case:http:1",
1205            subject(),
1206            "http.probe",
1207            "hypothesis:http",
1208            "verify.http-status",
1209            stage,
1210            status,
1211            Probability::from_percent(90).unwrap(),
1212            "HTTP status verification",
1213            BTreeSet::from([evidence_id]),
1214        )
1215        .unwrap()
1216    }
1217
1218    fn unknown_outcome(stage: VerificationStage) -> Outcome {
1219        Outcome::unknown(
1220            "case:http:1",
1221            subject(),
1222            "http.probe",
1223            "hypothesis:http",
1224            stage,
1225            "No verifier rule matched",
1226        )
1227        .unwrap()
1228    }
1229
1230    #[test]
1231    fn verifier_blocked_403_schedules_bypass_action() {
1232        let fixture = fixture(403);
1233        let mut verifier = PassiveVerifier::new();
1234        assert_eq!(
1235            verifier
1236                .register(
1237                    VerificationRule::new(
1238                        "verify.403",
1239                        VerificationStage::Passive,
1240                        100,
1241                        Expression::equals(
1242                            KnowledgeLayer::Evidence,
1243                            status_predicate(),
1244                            EvidenceValue::Unsigned(403),
1245                        ),
1246                        OutcomeStatus::Blocked,
1247                        Probability::from_percent(95).unwrap(),
1248                        "Access was blocked",
1249                    )
1250                    .unwrap(),
1251                )
1252                .unwrap(),
1253            VerifierWrite::Inserted
1254        );
1255        let outcome = verifier
1256            .verify(
1257                &fixture.knowledge,
1258                &VerificationCase::new("case:http:1", subject(), "http.probe", "hypothesis:http")
1259                    .unwrap(),
1260            )
1261            .unwrap();
1262        let pipeline = AdaptivePipeline::with_standard_policies().unwrap();
1263        let snapshot = fixture.knowledge.snapshot_for_subject(&subject());
1264        let decision = pipeline
1265            .decide(
1266                outcome.outcome(),
1267                &snapshot,
1268                &AdaptationLedger::new(),
1269                AdaptationLimits::default(),
1270            )
1271            .unwrap();
1272
1273        assert_eq!(decision.selected_rule_id(), Some("http.403.bypass"));
1274        assert_eq!(
1275            decision.directive(),
1276            &PipelineDirective::ScheduleAction {
1277                action_id: "http.403-bypass".into()
1278            }
1279        );
1280    }
1281
1282    #[test]
1283    fn review_404_redirects_to_enumeration() {
1284        let fixture = fixture(404);
1285        let pipeline = AdaptivePipeline::with_standard_policies().unwrap();
1286        let snapshot = fixture.knowledge.snapshot_for_subject(&subject());
1287        let outcome = verified_outcome(
1288            OutcomeStatus::NeedsReview,
1289            VerificationStage::Passive,
1290            fixture.evidence_id,
1291        );
1292        let decision = pipeline
1293            .decide(
1294                &outcome,
1295                &snapshot,
1296                &AdaptationLedger::new(),
1297                AdaptationLimits::default(),
1298            )
1299            .unwrap();
1300
1301        assert_eq!(decision.selected_rule_id(), Some("http.404.enumerate"));
1302        assert_eq!(
1303            decision.directive().scheduled_action_id(),
1304            Some("http.enumeration")
1305        );
1306        let selected: Vec<_> = decision
1307            .evaluations()
1308            .iter()
1309            .filter(|evaluation| evaluation.selected())
1310            .map(AdaptationRuleEvaluation::rule_id)
1311            .collect();
1312        assert_eq!(selected, vec!["http.404.enumerate"]);
1313    }
1314
1315    #[test]
1316    fn stale_status_evidence_cannot_redirect_a_new_outcome() {
1317        let fixture = fixture(403);
1318        let current = Evidence::new(
1319            subject(),
1320            EvidenceKind::Http,
1321            status_predicate(),
1322            EvidenceValue::Unsigned(200),
1323            EvidenceSource::new("http.executor", "current-response-status").unwrap(),
1324            ConfidenceScore::MAX,
1325        );
1326        let current_id = current.id().clone();
1327        fixture.knowledge.insert_evidence(current).unwrap();
1328        let snapshot = fixture.knowledge.snapshot_for_subject(&subject());
1329        let outcome = verified_outcome(
1330            OutcomeStatus::NeedsReview,
1331            VerificationStage::Passive,
1332            current_id,
1333        );
1334        let pipeline = AdaptivePipeline::with_standard_policies().unwrap();
1335        let decision = pipeline
1336            .decide(
1337                &outcome,
1338                &snapshot,
1339                &AdaptationLedger::new(),
1340                AdaptationLimits::default(),
1341            )
1342            .unwrap();
1343
1344        assert!(decision.selected_rule_id().is_none());
1345        assert_eq!(
1346            decision.directive(),
1347            &PipelineDirective::AwaitActiveVerification
1348        );
1349        let stale = decision
1350            .evaluations()
1351            .iter()
1352            .find(|evaluation| evaluation.rule_id() == "http.403.bypass")
1353            .unwrap();
1354        assert!(stale.condition().unwrap().matched());
1355        assert!(!stale.outcome_evidence_matched());
1356        assert!(!stale.eligible());
1357    }
1358
1359    #[test]
1360    fn blocked_429_throttles_and_retries() {
1361        let fixture = fixture(429);
1362        let pipeline = AdaptivePipeline::with_standard_policies().unwrap();
1363        let snapshot = fixture.knowledge.snapshot_for_subject(&subject());
1364        let outcome = verified_outcome(
1365            OutcomeStatus::Blocked,
1366            VerificationStage::Active,
1367            fixture.evidence_id,
1368        );
1369        let decision = pipeline
1370            .decide(
1371                &outcome,
1372                &snapshot,
1373                &AdaptationLedger::new(),
1374                AdaptationLimits::default(),
1375            )
1376            .unwrap();
1377
1378        assert_eq!(decision.selected_rule_id(), Some("http.429.throttle"));
1379        assert_eq!(
1380            decision.directive(),
1381            &PipelineDirective::Throttle {
1382                delay_ms: 2_000,
1383                retry_current_action: true
1384            }
1385        );
1386    }
1387
1388    #[test]
1389    fn learned_suppression_prevents_repeating_a_scheduled_action() {
1390        let fixture = fixture(403);
1391        let mut experience = ExperienceStore::new();
1392        for attempt in 0..10 {
1393            experience
1394                .observe(
1395                    Outcome::verified(
1396                        format!("case:bypass:{attempt}"),
1397                        subject(),
1398                        "http.403-bypass",
1399                        "hypothesis:http",
1400                        "verify.403-bypass",
1401                        VerificationStage::Active,
1402                        OutcomeStatus::ConfirmedNegative,
1403                        Probability::from_percent(90).unwrap(),
1404                        "active negative control rejected the bypass",
1405                        BTreeSet::from([fixture.evidence_id.clone()]),
1406                    )
1407                    .unwrap(),
1408                )
1409                .unwrap();
1410        }
1411        let suppressed = experience.suppressed_actions(&subject(), ExperiencePolicy::default());
1412        let outcome = verified_outcome(
1413            OutcomeStatus::Blocked,
1414            VerificationStage::Passive,
1415            fixture.evidence_id,
1416        );
1417        let decision = AdaptivePipeline::with_standard_policies()
1418            .unwrap()
1419            .decide_with_suppressed_actions(
1420                &outcome,
1421                &fixture.knowledge.snapshot_for_subject(&subject()),
1422                &AdaptationLedger::new(),
1423                AdaptationLimits::default(),
1424                &suppressed,
1425            )
1426            .unwrap();
1427
1428        assert_eq!(decision.directive(), &PipelineDirective::AwaitHumanReview);
1429        let bypass = decision
1430            .evaluations()
1431            .iter()
1432            .find(|evaluation| evaluation.rule_id() == "http.403.bypass")
1433            .unwrap();
1434        assert!(bypass.policy_suppressed());
1435        assert!(!bypass.eligible());
1436    }
1437
1438    #[test]
1439    fn rule_application_limit_prevents_adaptation_loops() {
1440        let fixture = fixture(403);
1441        let pipeline = AdaptivePipeline::with_standard_policies().unwrap();
1442        let snapshot = fixture.knowledge.snapshot_for_subject(&subject());
1443        let outcome = verified_outcome(
1444            OutcomeStatus::Blocked,
1445            VerificationStage::Passive,
1446            fixture.evidence_id,
1447        );
1448        let mut ledger = AdaptationLedger::new();
1449
1450        for _ in 0..2 {
1451            let decision = pipeline
1452                .decide_and_record(
1453                    &outcome,
1454                    &snapshot,
1455                    &mut ledger,
1456                    AdaptationLimits::default(),
1457                )
1458                .unwrap();
1459            assert_eq!(decision.selected_rule_id(), Some("http.403.bypass"));
1460        }
1461        let exhausted = pipeline
1462            .decide(&outcome, &snapshot, &ledger, AdaptationLimits::default())
1463            .unwrap();
1464
1465        assert_eq!(ledger.rule_applications("http.403.bypass"), 2);
1466        assert_eq!(ledger.action_schedules("http.403-bypass"), 2);
1467        assert_eq!(
1468            serde_json::from_value::<AdaptationLedger>(serde_json::to_value(&ledger).unwrap())
1469                .unwrap(),
1470            ledger
1471        );
1472        assert_eq!(exhausted.directive(), &PipelineDirective::AwaitHumanReview);
1473        let evaluation = exhausted
1474            .evaluations()
1475            .iter()
1476            .find(|evaluation| evaluation.rule_id() == "http.403.bypass")
1477            .unwrap();
1478        assert!(evaluation.rule_limit_exhausted());
1479        assert!(!evaluation.eligible());
1480    }
1481
1482    #[test]
1483    fn action_schedule_limit_applies_across_policy_rules() {
1484        let fixture = fixture(403);
1485        let mut pipeline = AdaptivePipeline::with_standard_policies().unwrap();
1486        pipeline
1487            .register(
1488                AdaptationRule::new(
1489                    "http.403.alternate",
1490                    OutcomeSelector::any_stage(BTreeSet::from([OutcomeStatus::Blocked])).unwrap(),
1491                    250,
1492                    Some(Expression::equals(
1493                        KnowledgeLayer::Evidence,
1494                        status_predicate(),
1495                        EvidenceValue::Unsigned(403),
1496                    )),
1497                    PipelineDirective::ScheduleAction {
1498                        action_id: "http.403-bypass".into(),
1499                    },
1500                    "alternate policy schedules the same bounded action",
1501                    2,
1502                )
1503                .unwrap(),
1504            )
1505            .unwrap();
1506        let snapshot = fixture.knowledge.snapshot_for_subject(&subject());
1507        let outcome = verified_outcome(
1508            OutcomeStatus::Blocked,
1509            VerificationStage::Passive,
1510            fixture.evidence_id,
1511        );
1512        let limits = AdaptationLimits::new(64, 1).unwrap();
1513        let mut ledger = AdaptationLedger::new();
1514        pipeline
1515            .decide_and_record(&outcome, &snapshot, &mut ledger, limits)
1516            .unwrap();
1517        let exhausted = pipeline
1518            .decide(&outcome, &snapshot, &ledger, limits)
1519            .unwrap();
1520
1521        assert_eq!(ledger.action_schedules("http.403-bypass"), 1);
1522        assert_eq!(exhausted.directive(), &PipelineDirective::AwaitHumanReview);
1523        assert!(exhausted
1524            .evaluations()
1525            .iter()
1526            .filter(|evaluation| {
1527                evaluation
1528                    .condition()
1529                    .is_some_and(ExpressionEvaluation::matched)
1530            })
1531            .all(|evaluation| evaluation.action_limit_exhausted()));
1532    }
1533
1534    #[test]
1535    fn global_transition_limit_emits_unrecorded_halt() {
1536        let fixture = fixture(404);
1537        let pipeline = AdaptivePipeline::with_standard_policies().unwrap();
1538        let snapshot = fixture.knowledge.snapshot_for_subject(&subject());
1539        let outcome = unknown_outcome(VerificationStage::Passive);
1540        let limits = AdaptationLimits::new(1, 3).unwrap();
1541        let mut ledger = AdaptationLedger::new();
1542        pipeline
1543            .decide_and_record(&outcome, &snapshot, &mut ledger, limits)
1544            .unwrap();
1545        let halted = pipeline
1546            .decide_and_record(&outcome, &snapshot, &mut ledger, limits)
1547            .unwrap();
1548
1549        assert_eq!(halted.directive(), &PipelineDirective::Halt);
1550        assert!(halted.transition_limit_reached());
1551        assert_eq!(ledger.transitions(), 1);
1552    }
1553
1554    #[test]
1555    fn fallback_transitions_cover_outcome_lifecycle() {
1556        let fixture = fixture(200);
1557        let snapshot = fixture.knowledge.snapshot_for_subject(&subject());
1558        let pipeline = AdaptivePipeline::new();
1559        let ledger = AdaptationLedger::new();
1560        let limits = AdaptationLimits::default();
1561
1562        let success = verified_outcome(
1563            OutcomeStatus::Success,
1564            VerificationStage::Passive,
1565            fixture.evidence_id.clone(),
1566        );
1567        assert_eq!(
1568            pipeline
1569                .decide(&success, &snapshot, &ledger, limits)
1570                .unwrap()
1571                .directive(),
1572            &PipelineDirective::Complete
1573        );
1574        let false_positive = verified_outcome(
1575            OutcomeStatus::FalsePositive,
1576            VerificationStage::Active,
1577            fixture.evidence_id.clone(),
1578        );
1579        let mut replanning_ledger = AdaptationLedger::new();
1580        let replan = pipeline
1581            .decide_and_record(&false_positive, &snapshot, &mut replanning_ledger, limits)
1582            .unwrap();
1583        assert_eq!(
1584            replan.directive(),
1585            &PipelineDirective::Replan {
1586                suppress_current_action: true
1587            }
1588        );
1589        assert!(replanning_ledger
1590            .suppressed_actions()
1591            .contains("http.probe"));
1592        let confirmed_negative = verified_outcome(
1593            OutcomeStatus::ConfirmedNegative,
1594            VerificationStage::Active,
1595            fixture.evidence_id,
1596        );
1597        assert_eq!(
1598            pipeline
1599                .decide(&confirmed_negative, &snapshot, &ledger, limits)
1600                .unwrap()
1601                .directive(),
1602            &PipelineDirective::Replan {
1603                suppress_current_action: true
1604            }
1605        );
1606        assert_eq!(
1607            pipeline
1608                .decide(
1609                    &unknown_outcome(VerificationStage::Passive),
1610                    &snapshot,
1611                    &ledger,
1612                    limits,
1613                )
1614                .unwrap()
1615                .directive(),
1616            &PipelineDirective::AwaitActiveVerification
1617        );
1618        assert_eq!(
1619            pipeline
1620                .decide(
1621                    &unknown_outcome(VerificationStage::Active),
1622                    &snapshot,
1623                    &ledger,
1624                    limits,
1625                )
1626                .unwrap()
1627                .directive(),
1628            &PipelineDirective::AwaitHumanReview
1629        );
1630    }
1631
1632    #[test]
1633    fn equal_priority_rules_use_stable_identity_order() {
1634        let fixture = fixture(200);
1635        let snapshot = fixture.knowledge.snapshot_for_subject(&subject());
1636        let mut pipeline = AdaptivePipeline::new();
1637        let selector =
1638            OutcomeSelector::any_stage(BTreeSet::from([OutcomeStatus::Blocked])).unwrap();
1639        for id in ["zeta", "alpha"] {
1640            pipeline
1641                .register(
1642                    AdaptationRule::new(
1643                        id,
1644                        selector.clone(),
1645                        10,
1646                        None,
1647                        PipelineDirective::AwaitHumanReview,
1648                        format!("{id} rationale"),
1649                        1,
1650                    )
1651                    .unwrap(),
1652                )
1653                .unwrap();
1654        }
1655        let outcome = verified_outcome(
1656            OutcomeStatus::Blocked,
1657            VerificationStage::Passive,
1658            fixture.evidence_id,
1659        );
1660
1661        let first = pipeline
1662            .decide(
1663                &outcome,
1664                &snapshot,
1665                &AdaptationLedger::new(),
1666                AdaptationLimits::default(),
1667            )
1668            .unwrap();
1669        let second = pipeline
1670            .decide(
1671                &outcome,
1672                &snapshot,
1673                &AdaptationLedger::new(),
1674                AdaptationLimits::default(),
1675            )
1676            .unwrap();
1677
1678        assert_eq!(first, second);
1679        assert_eq!(first.selected_rule_id(), Some("alpha"));
1680    }
1681
1682    #[test]
1683    fn wire_invariants_reject_invalid_rules_limits_and_ledgers() {
1684        let rule = AdaptationRule::new(
1685            "schedule",
1686            OutcomeSelector::any_stage(BTreeSet::from([OutcomeStatus::Blocked])).unwrap(),
1687            10,
1688            None,
1689            PipelineDirective::ScheduleAction {
1690                action_id: "http.bypass".into(),
1691            },
1692            "schedule bypass",
1693            1,
1694        )
1695        .unwrap();
1696        let mut encoded = serde_json::to_value(&rule).unwrap();
1697        assert_eq!(
1698            serde_json::from_value::<AdaptationRule>(encoded.clone()).unwrap(),
1699            rule
1700        );
1701        encoded["directive"]["action_id"] = serde_json::json!("");
1702        assert!(serde_json::from_value::<AdaptationRule>(encoded).is_err());
1703
1704        assert!(
1705            serde_json::from_value::<PipelineDirective>(serde_json::json!({
1706                "directive": "schedule_action",
1707                "action_id": ""
1708            }))
1709            .is_err()
1710        );
1711        assert!(
1712            serde_json::from_value::<PipelineDirective>(serde_json::json!({
1713                "directive": "throttle",
1714                "delay_ms": 0,
1715                "retry_current_action": true
1716            }))
1717            .is_err()
1718        );
1719        assert!(
1720            serde_json::from_value::<PipelineDirective>(serde_json::json!({
1721                "directive": "complete",
1722                "action_id": "must-not-be-ignored"
1723            }))
1724            .is_err()
1725        );
1726
1727        assert!(
1728            serde_json::from_value::<OutcomeSelector>(serde_json::json!({
1729                "statuses": [],
1730                "stages": ["passive"]
1731            }))
1732            .is_err()
1733        );
1734        assert!(
1735            serde_json::from_value::<OutcomeSelector>(serde_json::json!({
1736                "statuses": ["blocked"],
1737                "stages": ["passive"],
1738                "condition": "must-not-be-ignored"
1739            }))
1740            .is_err()
1741        );
1742        assert!(
1743            serde_json::from_value::<AdaptationLimits>(serde_json::json!({
1744                "max_transitions": 0,
1745                "max_action_schedules": 1
1746            }))
1747            .is_err()
1748        );
1749        assert!(
1750            serde_json::from_value::<AdaptationLedger>(serde_json::json!({
1751                "transitions": 1,
1752                "rule_applications": {"": 1},
1753                "action_schedules": {},
1754                "suppressed_actions": []
1755            }))
1756            .is_err()
1757        );
1758        assert!(
1759            serde_json::from_value::<AdaptationLedger>(serde_json::json!({
1760                "transitions": 1,
1761                "rule_applications": {"rule": 2},
1762                "action_schedules": {},
1763                "suppressed_actions": []
1764            }))
1765            .is_err()
1766        );
1767    }
1768
1769    #[test]
1770    fn adaptation_rule_wire_cannot_broaden_a_condition_to_unconditional() {
1771        let conditional = AdaptationRule::new(
1772            "conditional.schedule",
1773            OutcomeSelector::any_stage(BTreeSet::from([OutcomeStatus::Blocked])).unwrap(),
1774            10,
1775            Some(Expression::equals(
1776                KnowledgeLayer::Evidence,
1777                status_predicate(),
1778                EvidenceValue::Unsigned(403),
1779            )),
1780            PipelineDirective::ScheduleAction {
1781                action_id: "http.bypass".into(),
1782            },
1783            "schedule only for a correlated 403",
1784            1,
1785        )
1786        .unwrap();
1787        let encoded = serde_json::to_value(&conditional).unwrap();
1788        assert_eq!(encoded["condition_policy_guard"], true);
1789
1790        let mut legacy_guardless = encoded.clone();
1791        legacy_guardless
1792            .as_object_mut()
1793            .unwrap()
1794            .remove("condition_policy_guard");
1795        let restored_legacy = serde_json::from_value::<AdaptationRule>(legacy_guardless).unwrap();
1796        assert_eq!(restored_legacy, conditional);
1797        assert_eq!(
1798            serde_json::to_value(&restored_legacy).unwrap()["condition_policy_guard"],
1799            true
1800        );
1801
1802        let mismatch = fixture(200);
1803        let mismatch_outcome = verified_outcome(
1804            OutcomeStatus::Blocked,
1805            VerificationStage::Passive,
1806            mismatch.evidence_id,
1807        );
1808        let mut pipeline = AdaptivePipeline::new();
1809        pipeline.register(restored_legacy).unwrap();
1810        assert_eq!(
1811            pipeline
1812                .decide(
1813                    &mismatch_outcome,
1814                    &mismatch.knowledge.snapshot_for_subject(&subject()),
1815                    &AdaptationLedger::new(),
1816                    AdaptationLimits::default(),
1817                )
1818                .unwrap()
1819                .directive(),
1820            &PipelineDirective::AwaitHumanReview
1821        );
1822
1823        let mut misspelled = encoded.clone();
1824        let condition = misspelled
1825            .as_object_mut()
1826            .unwrap()
1827            .remove("condition")
1828            .unwrap();
1829        misspelled["conditon"] = condition;
1830        assert!(serde_json::from_value::<AdaptationRule>(misspelled).is_err());
1831
1832        let mut missing = encoded.clone();
1833        missing.as_object_mut().unwrap().remove("condition");
1834        assert!(serde_json::from_value::<AdaptationRule>(missing).is_err());
1835
1836        let mut nulled = encoded;
1837        nulled["condition"] = serde_json::Value::Null;
1838        assert!(serde_json::from_value::<AdaptationRule>(nulled).is_err());
1839
1840        let mut inconsistent = serde_json::to_value(&conditional).unwrap();
1841        inconsistent["condition_policy_guard"] = serde_json::json!(false);
1842        assert!(serde_json::from_value::<AdaptationRule>(inconsistent).is_err());
1843
1844        let unconditional = AdaptationRule::new(
1845            "unconditional.review",
1846            OutcomeSelector::any_stage(BTreeSet::from([OutcomeStatus::NeedsReview])).unwrap(),
1847            1,
1848            None,
1849            PipelineDirective::AwaitHumanReview,
1850            "explicit unconditional fallback",
1851            1,
1852        )
1853        .unwrap();
1854        let unconditional_wire = serde_json::to_value(&unconditional).unwrap();
1855        assert!(unconditional_wire["condition"].is_null());
1856        assert!(unconditional_wire.get("condition_policy_guard").is_none());
1857        assert_eq!(
1858            serde_json::from_value::<AdaptationRule>(unconditional_wire).unwrap(),
1859            unconditional
1860        );
1861    }
1862
1863    #[test]
1864    fn outcome_provenance_must_exist_in_snapshot() {
1865        let outcome_fixture = fixture(403);
1866        let snapshot_fixture = fixture(403);
1867        let outcome = verified_outcome(
1868            OutcomeStatus::Blocked,
1869            VerificationStage::Passive,
1870            outcome_fixture.evidence_id,
1871        );
1872        let snapshot = snapshot_fixture.knowledge.snapshot_for_subject(&subject());
1873
1874        assert!(matches!(
1875            AdaptivePipeline::new().decide(
1876                &outcome,
1877                &snapshot,
1878                &AdaptationLedger::new(),
1879                AdaptationLimits::default(),
1880            ),
1881            Err(AdaptivePipelineError::MissingOutcomeEvidence { .. })
1882        ));
1883    }
1884}
termivar_scanner/adaptive/pipeline.rs

termivar_scanner/adaptive/
pipeline.rs
