rch_common/e2e/
process_triage.rs

1//! Stable process triage integration contract.
2//!
3//! This module defines a shared schema for invoking external process-triage
4//! helpers and safely interpreting their actions/results.
5
6use schemars::generate::SchemaSettings;
7use schemars::{JsonSchema, Schema};
8use serde::{Deserialize, Serialize};
9use std::collections::HashSet;
10
11/// Stable schema version for the process-triage integration contract.
12///
13/// Sourced from the central [`crate::schema_versions`] registry so cross-component
14/// drift is caught by the registry's pinned-snapshot test.
15pub const PROCESS_TRIAGE_CONTRACT_SCHEMA_VERSION: &str = crate::schema_versions::current_version(
16    crate::schema_versions::SchemaComponent::ProcessTriageContract,
17);
18
19/// Stable command surface for invoking the process-triage adapter.
20#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize, JsonSchema)]
21#[serde(rename_all = "snake_case")]
22pub enum ProcessTriageAdapterCommand {
23    Analyze,
24    Execute,
25    Health,
26    Version,
27}
28
29impl ProcessTriageAdapterCommand {
30    /// Return CLI arguments for the command surface.
31    pub fn args(self) -> &'static [&'static str] {
32        match self {
33            Self::Analyze => &["process-triage", "analyze", "--json"],
34            Self::Execute => &["process-triage", "execute", "--json"],
35            Self::Health => &["process-triage", "health", "--json"],
36            Self::Version => &["process-triage", "version", "--json"],
37        }
38    }
39}
40
41/// Supported process-triage action classes ordered by risk.
42#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize, JsonSchema)]
43#[serde(rename_all = "snake_case")]
44pub enum ProcessTriageActionClass {
45    ObserveOnly,
46    SoftTerminate,
47    HardTerminate,
48    ReclaimDisk,
49}
50
51impl ProcessTriageActionClass {
52    fn risk_rank(self) -> u8 {
53        match self {
54            Self::ObserveOnly => 0,
55            Self::ReclaimDisk => 1,
56            Self::SoftTerminate => 2,
57            Self::HardTerminate => 3,
58        }
59    }
60}
61
62/// Trigger class that initiated process triage.
63#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
64#[serde(rename_all = "snake_case")]
65pub enum ProcessTriageTrigger {
66    DiskPressure,
67    WorkerHealth,
68    BuildTimeout,
69    Manual,
70}
71
72/// Process classification label from detector.
73#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
74#[serde(rename_all = "snake_case")]
75pub enum ProcessClassification {
76    BuildRelated,
77    Suspicious,
78    Interactive,
79    SystemCritical,
80    Unknown,
81}
82
83/// Escalation levels used by safe-action policy.
84#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
85#[serde(rename_all = "snake_case")]
86pub enum ProcessTriageEscalationLevel {
87    Automatic,
88    Supervised,
89    ManualReview,
90    Blocked,
91}
92
93/// Failure taxonomy for process triage adapter interactions.
94#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
95#[serde(rename_all = "snake_case")]
96pub enum ProcessTriageFailureKind {
97    DetectorUncertain,
98    PolicyViolation,
99    TransportError,
100    ExecutorRuntimeError,
101    Timeout,
102    PartialResult,
103    InvalidRequest,
104}
105
106/// Adapter command budget used in timeout/retry policy.
107#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
108pub struct ProcessTriageCommandBudget {
109    pub command: ProcessTriageAdapterCommand,
110    pub timeout_secs: u64,
111    pub retries: u32,
112}
113
114/// Timeout policy contract.
115#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
116pub struct ProcessTriageTimeoutPolicy {
117    pub request_timeout_secs: u64,
118    pub action_timeout_secs: u64,
119    pub total_timeout_secs: u64,
120}
121
122impl Default for ProcessTriageTimeoutPolicy {
123    fn default() -> Self {
124        Self {
125            request_timeout_secs: 8,
126            action_timeout_secs: 15,
127            total_timeout_secs: 30,
128        }
129    }
130}
131
132/// Retry policy contract.
133#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
134pub struct ProcessTriageRetryPolicy {
135    pub max_attempts: u32,
136    pub initial_backoff_ms: u64,
137    pub max_backoff_ms: u64,
138    pub backoff_multiplier_percent: u16,
139}
140
141impl Default for ProcessTriageRetryPolicy {
142    fn default() -> Self {
143        Self {
144            max_attempts: 3,
145            initial_backoff_ms: 250,
146            max_backoff_ms: 2_000,
147            backoff_multiplier_percent: 200,
148        }
149    }
150}
151
152/// Escalation thresholds used by policy evaluation.
153#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
154pub struct ProcessTriageEscalationThresholds {
155    pub min_confidence_for_automatic: u8,
156    pub max_actions_before_manual_review: u32,
157    pub max_hard_terminations_before_manual_review: u32,
158}
159
160impl Default for ProcessTriageEscalationThresholds {
161    fn default() -> Self {
162        Self {
163            min_confidence_for_automatic: 85,
164            max_actions_before_manual_review: 5,
165            max_hard_terminations_before_manual_review: 1,
166        }
167    }
168}
169
170/// Safe-action policy with explicit allowlist/denylist and escalation rules.
171#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
172pub struct ProcessTriageSafeActionPolicy {
173    pub policy_version: String,
174    pub allow_action_classes: Vec<ProcessTriageActionClass>,
175    pub deny_action_classes: Vec<ProcessTriageActionClass>,
176    pub managed_process_patterns: Vec<String>,
177    pub protected_process_patterns: Vec<String>,
178    pub escalation: ProcessTriageEscalationThresholds,
179    pub require_audit_record: bool,
180}
181
182impl Default for ProcessTriageSafeActionPolicy {
183    fn default() -> Self {
184        Self {
185            policy_version: "v1".to_string(),
186            allow_action_classes: vec![
187                ProcessTriageActionClass::ObserveOnly,
188                ProcessTriageActionClass::SoftTerminate,
189                ProcessTriageActionClass::ReclaimDisk,
190            ],
191            deny_action_classes: vec![ProcessTriageActionClass::HardTerminate],
192            managed_process_patterns: vec![
193                "cargo".to_string(),
194                "rustc".to_string(),
195                "clang".to_string(),
196            ],
197            protected_process_patterns: vec![
198                "sshd".to_string(),
199                "systemd".to_string(),
200                "init".to_string(),
201            ],
202            escalation: ProcessTriageEscalationThresholds::default(),
203            require_audit_record: true,
204        }
205    }
206}
207
208/// Full contract bundle for process triage integration.
209#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
210pub struct ProcessTriageContract {
211    pub schema_version: String,
212    pub timeout_policy: ProcessTriageTimeoutPolicy,
213    pub retry_policy: ProcessTriageRetryPolicy,
214    pub command_budgets: Vec<ProcessTriageCommandBudget>,
215    pub safe_action_policy: ProcessTriageSafeActionPolicy,
216}
217
218impl Default for ProcessTriageContract {
219    fn default() -> Self {
220        Self {
221            schema_version: PROCESS_TRIAGE_CONTRACT_SCHEMA_VERSION.to_string(),
222            timeout_policy: ProcessTriageTimeoutPolicy::default(),
223            retry_policy: ProcessTriageRetryPolicy::default(),
224            command_budgets: vec![
225                ProcessTriageCommandBudget {
226                    command: ProcessTriageAdapterCommand::Analyze,
227                    timeout_secs: 8,
228                    retries: 1,
229                },
230                ProcessTriageCommandBudget {
231                    command: ProcessTriageAdapterCommand::Execute,
232                    timeout_secs: 15,
233                    retries: 1,
234                },
235                ProcessTriageCommandBudget {
236                    command: ProcessTriageAdapterCommand::Health,
237                    timeout_secs: 3,
238                    retries: 2,
239                },
240                ProcessTriageCommandBudget {
241                    command: ProcessTriageAdapterCommand::Version,
242                    timeout_secs: 2,
243                    retries: 0,
244                },
245            ],
246            safe_action_policy: ProcessTriageSafeActionPolicy::default(),
247        }
248    }
249}
250
251/// Observed process sample from detector.
252#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
253pub struct ProcessDescriptor {
254    pub pid: u32,
255    pub ppid: Option<u32>,
256    pub owner: String,
257    pub command: String,
258    pub classification: ProcessClassification,
259    pub cpu_percent_milli: u32,
260    pub rss_mb: u32,
261    pub runtime_secs: u64,
262}
263
264/// Action request item proposed by detector/planner.
265#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
266pub struct ProcessTriageActionRequest {
267    pub action_class: ProcessTriageActionClass,
268    pub pid: u32,
269    pub reason_code: String,
270    pub signal: Option<String>,
271}
272
273/// Adapter request schema.
274#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
275pub struct ProcessTriageRequest {
276    pub schema_version: String,
277    pub correlation_id: String,
278    pub worker_id: String,
279    pub observed_at_unix_ms: i64,
280    pub trigger: ProcessTriageTrigger,
281    pub detector_confidence_percent: u8,
282    pub retry_attempt: u32,
283    pub candidate_processes: Vec<ProcessDescriptor>,
284    pub requested_actions: Vec<ProcessTriageActionRequest>,
285}
286
287/// Action execution outcome.
288#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
289#[serde(rename_all = "snake_case")]
290pub enum ProcessTriageActionOutcome {
291    Skipped,
292    Executed,
293    Failed,
294    Escalated,
295}
296
297/// Action result in response.
298#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
299pub struct ProcessTriageActionResult {
300    pub pid: u32,
301    pub action_class: ProcessTriageActionClass,
302    pub outcome: ProcessTriageActionOutcome,
303    pub note: Option<String>,
304}
305
306/// High-level response status.
307#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
308#[serde(rename_all = "snake_case")]
309pub enum ProcessTriageResponseStatus {
310    Applied,
311    PartiallyApplied,
312    EscalatedNoAction,
313    RejectedByPolicy,
314    Failed,
315}
316
317/// Failure payload.
318#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
319pub struct ProcessTriageFailure {
320    pub kind: ProcessTriageFailureKind,
321    pub code: String,
322    pub message: String,
323    pub remediation: Vec<String>,
324}
325
326/// Audit record required for every process triage response.
327#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
328pub struct ProcessTriageAuditRecord {
329    pub policy_version: String,
330    pub evaluated_by: String,
331    pub evaluated_at_unix_ms: i64,
332    pub decision_code: String,
333    pub requires_operator_ack: bool,
334    pub audit_required: bool,
335}
336
337/// Adapter response schema.
338#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
339pub struct ProcessTriageResponse {
340    pub schema_version: String,
341    pub correlation_id: String,
342    pub status: ProcessTriageResponseStatus,
343    pub escalation_level: ProcessTriageEscalationLevel,
344    pub executed_actions: Vec<ProcessTriageActionResult>,
345    pub failure: Option<ProcessTriageFailure>,
346    pub audit: ProcessTriageAuditRecord,
347}
348
349/// Deterministic policy decision for a single action request.
350#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
351pub struct ProcessTriagePolicyDecision {
352    pub permitted: bool,
353    pub escalation_level: ProcessTriageEscalationLevel,
354    pub effective_action: Option<ProcessTriageActionClass>,
355    pub decision_code: String,
356    pub reason: String,
357    pub requires_operator_ack: bool,
358    pub audit_required: bool,
359}
360
361/// Contract validation failures.
362#[derive(Debug, thiserror::Error, PartialEq, Eq)]
363pub enum ProcessTriageContractError {
364    #[error("schema version mismatch: expected {expected}, got {actual}")]
365    SchemaVersionMismatch { expected: String, actual: String },
366    #[error("detector confidence percent must be <= 100, got {0}")]
367    InvalidConfidence(u8),
368    #[error("requested_actions must not be empty")]
369    EmptyRequestedActions,
370    #[error("requested action references unknown pid {0}")]
371    UnknownActionPid(u32),
372    #[error("timeout policy has invalid value for {field}: {value}")]
373    InvalidTimeout { field: &'static str, value: u64 },
374    #[error("retry policy has invalid value for {field}: {value}")]
375    InvalidRetryPolicy { field: &'static str, value: u64 },
376    #[error("allowlist/denylist conflict for action class {0:?}")]
377    AllowDenyConflict(ProcessTriageActionClass),
378    #[error("escalation threshold min_confidence_for_automatic must be <= 100, got {0}")]
379    InvalidEscalationConfidence(u8),
380}
381
382impl ProcessTriageRequest {
383    /// Validate request shape and semantic constraints.
384    pub fn validate(&self) -> Result<(), ProcessTriageContractError> {
385        if self.schema_version != PROCESS_TRIAGE_CONTRACT_SCHEMA_VERSION {
386            return Err(ProcessTriageContractError::SchemaVersionMismatch {
387                expected: PROCESS_TRIAGE_CONTRACT_SCHEMA_VERSION.to_string(),
388                actual: self.schema_version.clone(),
389            });
390        }
391        if self.detector_confidence_percent > 100 {
392            return Err(ProcessTriageContractError::InvalidConfidence(
393                self.detector_confidence_percent,
394            ));
395        }
396        if self.requested_actions.is_empty() {
397            return Err(ProcessTriageContractError::EmptyRequestedActions);
398        }
399
400        let candidate_pids: HashSet<u32> = self.candidate_processes.iter().map(|p| p.pid).collect();
401        for action in &self.requested_actions {
402            if !candidate_pids.contains(&action.pid) {
403                return Err(ProcessTriageContractError::UnknownActionPid(action.pid));
404            }
405        }
406
407        Ok(())
408    }
409}
410
411impl ProcessTriageContract {
412    /// Validate contract configuration.
413    pub fn validate(&self) -> Result<(), ProcessTriageContractError> {
414        if self.timeout_policy.request_timeout_secs == 0 {
415            return Err(ProcessTriageContractError::InvalidTimeout {
416                field: "request_timeout_secs",
417                value: self.timeout_policy.request_timeout_secs,
418            });
419        }
420        if self.timeout_policy.action_timeout_secs == 0 {
421            return Err(ProcessTriageContractError::InvalidTimeout {
422                field: "action_timeout_secs",
423                value: self.timeout_policy.action_timeout_secs,
424            });
425        }
426        if self.timeout_policy.total_timeout_secs == 0 {
427            return Err(ProcessTriageContractError::InvalidTimeout {
428                field: "total_timeout_secs",
429                value: self.timeout_policy.total_timeout_secs,
430            });
431        }
432        if self.retry_policy.max_attempts == 0 {
433            return Err(ProcessTriageContractError::InvalidRetryPolicy {
434                field: "max_attempts",
435                value: self.retry_policy.max_attempts as u64,
436            });
437        }
438        if self.retry_policy.initial_backoff_ms == 0 {
439            return Err(ProcessTriageContractError::InvalidRetryPolicy {
440                field: "initial_backoff_ms",
441                value: self.retry_policy.initial_backoff_ms,
442            });
443        }
444        if self.retry_policy.max_backoff_ms < self.retry_policy.initial_backoff_ms {
445            return Err(ProcessTriageContractError::InvalidRetryPolicy {
446                field: "max_backoff_ms",
447                value: self.retry_policy.max_backoff_ms,
448            });
449        }
450        if self
451            .safe_action_policy
452            .escalation
453            .min_confidence_for_automatic
454            > 100
455        {
456            return Err(ProcessTriageContractError::InvalidEscalationConfidence(
457                self.safe_action_policy
458                    .escalation
459                    .min_confidence_for_automatic,
460            ));
461        }
462
463        let allow: HashSet<ProcessTriageActionClass> = self
464            .safe_action_policy
465            .allow_action_classes
466            .iter()
467            .copied()
468            .collect();
469        let deny: HashSet<ProcessTriageActionClass> = self
470            .safe_action_policy
471            .deny_action_classes
472            .iter()
473            .copied()
474            .collect();
475
476        if let Some(action) = allow.intersection(&deny).next() {
477            return Err(ProcessTriageContractError::AllowDenyConflict(*action));
478        }
479
480        Ok(())
481    }
482}
483
484/// Evaluate a requested action against the safe-action policy.
485pub fn evaluate_triage_action(
486    request: &ProcessTriageRequest,
487    contract: &ProcessTriageContract,
488    action: &ProcessTriageActionRequest,
489) -> ProcessTriagePolicyDecision {
490    let policy = &contract.safe_action_policy;
491    let allow: HashSet<ProcessTriageActionClass> =
492        policy.allow_action_classes.iter().copied().collect();
493    let deny: HashSet<ProcessTriageActionClass> =
494        policy.deny_action_classes.iter().copied().collect();
495
496    if deny.contains(&action.action_class) {
497        return ProcessTriagePolicyDecision {
498            permitted: false,
499            escalation_level: ProcessTriageEscalationLevel::Blocked,
500            effective_action: None,
501            decision_code: "PT_BLOCK_DENYLIST".to_string(),
502            reason: format!("action class {:?} is denylisted", action.action_class),
503            requires_operator_ack: true,
504            audit_required: policy.require_audit_record,
505        };
506    }
507
508    if !allow.contains(&action.action_class) {
509        return ProcessTriagePolicyDecision {
510            permitted: false,
511            escalation_level: ProcessTriageEscalationLevel::Blocked,
512            effective_action: None,
513            decision_code: "PT_BLOCK_NOT_ALLOWLISTED".to_string(),
514            reason: format!("action class {:?} is not allowlisted", action.action_class),
515            requires_operator_ack: true,
516            audit_required: policy.require_audit_record,
517        };
518    }
519
520    let target = request
521        .candidate_processes
522        .iter()
523        .find(|proc_desc| proc_desc.pid == action.pid);
524
525    if let Some(proc_desc) = target {
526        let cmd_lower = proc_desc.command.to_ascii_lowercase();
527        if pattern_matches(&cmd_lower, &policy.protected_process_patterns) {
528            return ProcessTriagePolicyDecision {
529                permitted: false,
530                escalation_level: ProcessTriageEscalationLevel::Blocked,
531                effective_action: None,
532                decision_code: "PT_BLOCK_PROTECTED_PROCESS".to_string(),
533                reason: format!(
534                    "target process '{}' matches protected patterns",
535                    proc_desc.command
536                ),
537                requires_operator_ack: true,
538                audit_required: policy.require_audit_record,
539            };
540        }
541        if !policy.managed_process_patterns.is_empty()
542            && !pattern_matches(&cmd_lower, &policy.managed_process_patterns)
543        {
544            return ProcessTriagePolicyDecision {
545                permitted: false,
546                escalation_level: ProcessTriageEscalationLevel::Blocked,
547                effective_action: None,
548                decision_code: "PT_BLOCK_OUT_OF_SCOPE_PROCESS".to_string(),
549                reason: format!(
550                    "target process '{}' does not match managed patterns",
551                    proc_desc.command
552                ),
553                requires_operator_ack: true,
554                audit_required: policy.require_audit_record,
555            };
556        }
557    }
558
559    if request.detector_confidence_percent < policy.escalation.min_confidence_for_automatic {
560        return ProcessTriagePolicyDecision {
561            permitted: false,
562            escalation_level: ProcessTriageEscalationLevel::ManualReview,
563            effective_action: None,
564            decision_code: "PT_MANUAL_LOW_CONFIDENCE".to_string(),
565            reason: format!(
566                "detector confidence {} is below automatic threshold {}",
567                request.detector_confidence_percent, policy.escalation.min_confidence_for_automatic
568            ),
569            requires_operator_ack: true,
570            audit_required: policy.require_audit_record,
571        };
572    }
573
574    if request.retry_attempt + 1 >= contract.retry_policy.max_attempts {
575        return ProcessTriagePolicyDecision {
576            permitted: false,
577            escalation_level: ProcessTriageEscalationLevel::ManualReview,
578            effective_action: None,
579            decision_code: "PT_MANUAL_RETRY_EXHAUSTED".to_string(),
580            reason: format!(
581                "retry attempt {} reached max attempts {}",
582                request.retry_attempt + 1,
583                contract.retry_policy.max_attempts
584            ),
585            requires_operator_ack: true,
586            audit_required: policy.require_audit_record,
587        };
588    }
589
590    let hard_kill_requests = request
591        .requested_actions
592        .iter()
593        .filter(|req| req.action_class == ProcessTriageActionClass::HardTerminate)
594        .count() as u32;
595
596    if hard_kill_requests > policy.escalation.max_hard_terminations_before_manual_review {
597        return ProcessTriagePolicyDecision {
598            permitted: false,
599            escalation_level: ProcessTriageEscalationLevel::ManualReview,
600            effective_action: None,
601            decision_code: "PT_MANUAL_HARD_KILL_THRESHOLD".to_string(),
602            reason: format!(
603                "requested hard terminations {} exceeds threshold {}",
604                hard_kill_requests, policy.escalation.max_hard_terminations_before_manual_review
605            ),
606            requires_operator_ack: true,
607            audit_required: policy.require_audit_record,
608        };
609    }
610
611    if request.requested_actions.len() as u32 > policy.escalation.max_actions_before_manual_review {
612        let downgraded_action = if action.action_class.risk_rank()
613            > ProcessTriageActionClass::ObserveOnly.risk_rank()
614        {
615            ProcessTriageActionClass::ObserveOnly
616        } else {
617            action.action_class
618        };
619
620        return ProcessTriagePolicyDecision {
621            permitted: true,
622            escalation_level: ProcessTriageEscalationLevel::Supervised,
623            effective_action: Some(downgraded_action),
624            decision_code: "PT_SUPERVISED_ACTION_VOLUME".to_string(),
625            reason: format!(
626                "requested action count {} exceeds threshold {}, action downgraded for supervised mode",
627                request.requested_actions.len(),
628                policy.escalation.max_actions_before_manual_review
629            ),
630            requires_operator_ack: true,
631            audit_required: policy.require_audit_record,
632        };
633    }
634
635    ProcessTriagePolicyDecision {
636        permitted: true,
637        escalation_level: ProcessTriageEscalationLevel::Automatic,
638        effective_action: Some(action.action_class),
639        decision_code: "PT_ALLOW_AUTOMATIC".to_string(),
640        reason: "action satisfies allowlist and escalation thresholds".to_string(),
641        requires_operator_ack: false,
642        audit_required: policy.require_audit_record,
643    }
644}
645
646fn pattern_matches(command_lower: &str, patterns: &[String]) -> bool {
647    patterns
648        .iter()
649        .map(|p| p.to_ascii_lowercase())
650        .any(|p| !p.is_empty() && command_lower.contains(&p))
651}
652
653/// JSON schema for request payload.
654pub fn process_triage_request_schema() -> Schema {
655    SchemaSettings::draft07()
656        .into_generator()
657        .into_root_schema_for::<ProcessTriageRequest>()
658}
659
660/// JSON schema for response payload.
661pub fn process_triage_response_schema() -> Schema {
662    SchemaSettings::draft07()
663        .into_generator()
664        .into_root_schema_for::<ProcessTriageResponse>()
665}
666
667#[cfg(test)]
668mod tests {
669    use super::*;
670    use serde_json::Value;
671
672    fn sample_request() -> ProcessTriageRequest {
673        ProcessTriageRequest {
674            schema_version: PROCESS_TRIAGE_CONTRACT_SCHEMA_VERSION.to_string(),
675            correlation_id: "corr-123".to_string(),
676            worker_id: "worker-a".to_string(),
677            observed_at_unix_ms: 1_768_768_123_000,
678            trigger: ProcessTriageTrigger::WorkerHealth,
679            detector_confidence_percent: 96,
680            retry_attempt: 0,
681            candidate_processes: vec![
682                ProcessDescriptor {
683                    pid: 1001,
684                    ppid: Some(1000),
685                    owner: "ubuntu".to_string(),
686                    command: "cargo test --workspace".to_string(),
687                    classification: ProcessClassification::BuildRelated,
688                    cpu_percent_milli: 92500,
689                    rss_mb: 2100,
690                    runtime_secs: 240,
691                },
692                ProcessDescriptor {
693                    pid: 1002,
694                    ppid: Some(1),
695                    owner: "root".to_string(),
696                    command: "sshd: ubuntu@pts/4".to_string(),
697                    classification: ProcessClassification::SystemCritical,
698                    cpu_percent_milli: 100,
699                    rss_mb: 32,
700                    runtime_secs: 8600,
701                },
702            ],
703            requested_actions: vec![ProcessTriageActionRequest {
704                action_class: ProcessTriageActionClass::SoftTerminate,
705                pid: 1001,
706                reason_code: "stuck_compile".to_string(),
707                signal: Some("TERM".to_string()),
708            }],
709        }
710    }
711
712    fn extract_ref_name(schema: &Value) -> Option<String> {
713        if let Some(reference) = schema.get("$ref").and_then(Value::as_str) {
714            return reference.rsplit('/').next().map(str::to_string);
715        }
716
717        for key in ["anyOf", "oneOf", "allOf"] {
718            if let Some(reference) =
719                schema
720                    .get(key)
721                    .and_then(Value::as_array)
722                    .and_then(|variants| {
723                        variants
724                            .iter()
725                            .find_map(|variant| variant.get("$ref").and_then(Value::as_str))
726                    })
727            {
728                return reference.rsplit('/').next().map(str::to_string);
729            }
730        }
731
732        None
733    }
734
735    fn find_schema_properties(
736        schema_json: &Value,
737        required: &[&str],
738    ) -> serde_json::Map<String, Value> {
739        let root_properties = schema_json
740            .get("properties")
741            .and_then(Value::as_object)
742            .filter(|properties| required.iter().all(|key| properties.contains_key(*key)))
743            .cloned();
744        if let Some(properties) = root_properties {
745            return properties;
746        }
747
748        let definition_properties = schema_json
749            .get("definitions")
750            .and_then(Value::as_object)
751            .and_then(|definitions| {
752                definitions.values().find_map(|node| {
753                    let properties = node.get("properties")?.as_object()?;
754                    if required.iter().all(|key| properties.contains_key(*key)) {
755                        Some(properties.clone())
756                    } else {
757                        None
758                    }
759                })
760            });
761        if let Some(properties) = definition_properties {
762            return properties;
763        }
764
765        panic!("schema properties not found for required keys: {required:?}");
766    }
767
768    fn definition_properties(
769        schema_json: &Value,
770        definition_name: &str,
771    ) -> serde_json::Map<String, Value> {
772        schema_json
773            .get("definitions")
774            .and_then(Value::as_object)
775            .and_then(|definitions| definitions.get(definition_name))
776            .and_then(|definition| definition.get("properties"))
777            .and_then(Value::as_object)
778            .cloned()
779            .unwrap_or_else(|| panic!("definition '{definition_name}' missing properties"))
780    }
781
782    fn definition_enum(schema_json: &Value, definition_name: &str) -> Vec<String> {
783        schema_json
784            .get("definitions")
785            .and_then(Value::as_object)
786            .and_then(|definitions| definitions.get(definition_name))
787            .and_then(|definition| definition.get("enum"))
788            .and_then(Value::as_array)
789            .map(|values| {
790                values
791                    .iter()
792                    .filter_map(Value::as_str)
793                    .map(str::to_string)
794                    .collect()
795            })
796            .unwrap_or_else(|| panic!("definition '{definition_name}' missing enum values"))
797    }
798
799    #[test]
800    fn process_triage_contract_request_roundtrip() {
801        let request = sample_request();
802        request.validate().expect("sample request should validate");
803
804        let json = serde_json::to_string(&request).expect("serialize request");
805        let restored: ProcessTriageRequest =
806            serde_json::from_str(&json).expect("deserialize request");
807        assert_eq!(
808            restored.schema_version,
809            PROCESS_TRIAGE_CONTRACT_SCHEMA_VERSION
810        );
811        assert_eq!(restored.worker_id, "worker-a");
812        assert_eq!(restored.requested_actions.len(), 1);
813    }
814
815    #[test]
816    fn process_triage_contract_policy_validation_rejects_allow_deny_overlap() {
817        let mut contract = ProcessTriageContract::default();
818        contract
819            .safe_action_policy
820            .deny_action_classes
821            .push(ProcessTriageActionClass::SoftTerminate);
822
823        let result = contract.validate();
824        assert!(matches!(
825            result,
826            Err(ProcessTriageContractError::AllowDenyConflict(
827                ProcessTriageActionClass::SoftTerminate
828            ))
829        ));
830    }
831
832    #[test]
833    fn process_triage_contract_blocks_protected_process() {
834        let contract = ProcessTriageContract::default();
835        let request = ProcessTriageRequest {
836            requested_actions: vec![ProcessTriageActionRequest {
837                action_class: ProcessTriageActionClass::SoftTerminate,
838                pid: 1002,
839                reason_code: "force_cleanup".to_string(),
840                signal: Some("TERM".to_string()),
841            }],
842            ..sample_request()
843        };
844
845        let decision = evaluate_triage_action(&request, &contract, &request.requested_actions[0]);
846        assert!(!decision.permitted);
847        assert_eq!(
848            decision.escalation_level,
849            ProcessTriageEscalationLevel::Blocked
850        );
851        assert_eq!(decision.decision_code, "PT_BLOCK_PROTECTED_PROCESS");
852    }
853
854    #[test]
855    fn process_triage_contract_requires_manual_review_on_low_confidence() {
856        let contract = ProcessTriageContract::default();
857        let request = ProcessTriageRequest {
858            detector_confidence_percent: 40,
859            ..sample_request()
860        };
861
862        let decision = evaluate_triage_action(&request, &contract, &request.requested_actions[0]);
863        assert!(!decision.permitted);
864        assert_eq!(
865            decision.escalation_level,
866            ProcessTriageEscalationLevel::ManualReview
867        );
868        assert_eq!(decision.decision_code, "PT_MANUAL_LOW_CONFIDENCE");
869    }
870
871    #[test]
872    fn process_triage_contract_respects_denylist() {
873        let mut contract = ProcessTriageContract::default();
874        contract.safe_action_policy.allow_action_classes = vec![
875            ProcessTriageActionClass::ObserveOnly,
876            ProcessTriageActionClass::HardTerminate,
877        ];
878        contract.safe_action_policy.deny_action_classes =
879            vec![ProcessTriageActionClass::HardTerminate];
880
881        let request = ProcessTriageRequest {
882            requested_actions: vec![ProcessTriageActionRequest {
883                action_class: ProcessTriageActionClass::HardTerminate,
884                pid: 1001,
885                reason_code: "stuck_compile".to_string(),
886                signal: Some("KILL".to_string()),
887            }],
888            ..sample_request()
889        };
890
891        let decision = evaluate_triage_action(&request, &contract, &request.requested_actions[0]);
892        assert!(!decision.permitted);
893        assert_eq!(
894            decision.escalation_level,
895            ProcessTriageEscalationLevel::Blocked
896        );
897        assert_eq!(decision.decision_code, "PT_BLOCK_DENYLIST");
898    }
899
900    #[test]
901    fn process_triage_contract_schema_contains_core_fields() {
902        let schema = process_triage_request_schema();
903        let schema_json = serde_json::to_value(&schema).expect("schema to json");
904        let root_properties = schema_json
905            .get("properties")
906            .and_then(|props| props.as_object())
907            .cloned();
908        let definition_properties = schema_json
909            .get("definitions")
910            .and_then(|defs| defs.as_object())
911            .and_then(|defs| {
912                defs.values().find_map(|node| {
913                    let properties = node.get("properties")?.as_object()?;
914                    if properties.contains_key("worker_id")
915                        && properties.contains_key("requested_actions")
916                    {
917                        Some(properties.clone())
918                    } else {
919                        None
920                    }
921                })
922            });
923        let properties = root_properties
924            .or(definition_properties)
925            .expect("request properties in schema");
926
927        assert!(properties.contains_key("schema_version"));
928        assert!(properties.contains_key("worker_id"));
929        assert!(properties.contains_key("requested_actions"));
930    }
931
932    #[test]
933    fn process_triage_contract_response_schema_requires_audit_record() {
934        let schema = process_triage_response_schema();
935        let schema_json = serde_json::to_value(&schema).expect("schema to json");
936        let response_properties =
937            find_schema_properties(&schema_json, &["status", "executed_actions", "audit"]);
938        let audit_schema = response_properties
939            .get("audit")
940            .expect("response schema should contain audit field");
941        let audit_properties = if let Some(definition_name) = extract_ref_name(audit_schema) {
942            definition_properties(&schema_json, &definition_name)
943        } else {
944            audit_schema
945                .get("properties")
946                .and_then(Value::as_object)
947                .cloned()
948                .expect("audit schema should provide properties")
949        };
950
951        assert!(audit_properties.contains_key("policy_version"));
952        assert!(audit_properties.contains_key("decision_code"));
953        assert!(audit_properties.contains_key("audit_required"));
954    }
955
956    #[test]
957    fn process_triage_contract_response_schema_exposes_failure_kind_taxonomy() {
958        let schema = process_triage_response_schema();
959        let schema_json = serde_json::to_value(&schema).expect("schema to json");
960        let response_properties = find_schema_properties(&schema_json, &["status", "failure"]);
961        let failure_schema = response_properties
962            .get("failure")
963            .expect("response schema should contain failure field");
964        let failure_definition = extract_ref_name(failure_schema)
965            .expect("failure field should reference ProcessTriageFailure schema");
966        let failure_properties = definition_properties(&schema_json, &failure_definition);
967        let kind_schema = failure_properties
968            .get("kind")
969            .expect("failure schema should contain kind field");
970        let kind_values = if let Some(definition_name) = extract_ref_name(kind_schema) {
971            definition_enum(&schema_json, &definition_name)
972        } else {
973            kind_schema
974                .get("enum")
975                .and_then(Value::as_array)
976                .map(|values| {
977                    values
978                        .iter()
979                        .filter_map(Value::as_str)
980                        .map(str::to_string)
981                        .collect::<Vec<_>>()
982                })
983                .expect("kind schema should expose enum values")
984        };
985
986        for expected in [
987            "detector_uncertain",
988            "policy_violation",
989            "transport_error",
990            "executor_runtime_error",
991            "timeout",
992            "partial_result",
993            "invalid_request",
994        ] {
995            assert!(
996                kind_values.iter().any(|value| value == expected),
997                "missing failure kind: {}",
998                expected
999            );
1000        }
1001    }
1002
1003    #[test]
1004    fn process_triage_contract_parser_compatibility() {
1005        let json = r#"{
1006            "schema_version":"1.0.0",
1007            "correlation_id":"corr-xyz",
1008            "worker_id":"worker-z",
1009            "observed_at_unix_ms":1768768123000,
1010            "trigger":"disk_pressure",
1011            "detector_confidence_percent":88,
1012            "retry_attempt":1,
1013            "candidate_processes":[{
1014                "pid":4242,
1015                "ppid":1,
1016                "owner":"ubuntu",
1017                "command":"cargo clippy --workspace",
1018                "classification":"build_related",
1019                "cpu_percent_milli":50000,
1020                "rss_mb":700,
1021                "runtime_secs":128
1022            }],
1023            "requested_actions":[{
1024                "action_class":"soft_terminate",
1025                "pid":4242,
1026                "reason_code":"timeout",
1027                "signal":"TERM"
1028            }]
1029        }"#;
1030
1031        let request: ProcessTriageRequest = serde_json::from_str(json).expect("compat parse");
1032        assert_eq!(
1033            request.schema_version,
1034            PROCESS_TRIAGE_CONTRACT_SCHEMA_VERSION
1035        );
1036        assert_eq!(request.trigger, ProcessTriageTrigger::DiskPressure);
1037        assert_eq!(request.requested_actions.len(), 1);
1038    }
1039}
rch_common/e2e/process_triage.rs

rch_common/e2e/
process_triage.rs
