code_system_graph_core/
execution_policy.rs

1use std::sync::Arc;
2use std::time::{Duration, Instant};
3
4use code_system_graph_model::stable_id;
5use schemars::JsonSchema;
6use serde::{Deserialize, Serialize};
7use thiserror::Error;
8
9/// Default maximum wall time for one supervised scan or sync pass.
10pub const DEFAULT_MAX_SCAN_WALL_TIME_MS: u64 = 21_600_000;
11/// Default maximum time without verified forward progress.
12pub const DEFAULT_MAX_NO_PROGRESS_TIME_MS: u64 = 300_000;
13/// Default maximum wall time for one repository-local `CodeGraph` synchronization.
14pub const DEFAULT_MAX_CODEGRAPH_SYNC_WALL_TIME_MS_PER_REPO: u64 = 3_600_000;
15/// Default maximum resident memory accepted for one worker process.
16pub const DEFAULT_MAX_WORKER_MEMORY_BYTES: u64 = 17_179_869_184;
17/// Default cooperative shutdown grace period before forced termination.
18pub const DEFAULT_GRACEFUL_TERMINATION_MS: u64 = 5_000;
19/// Default inactivity lease for a foreground watch session.
20pub const DEFAULT_WATCH_IDLE_TIMEOUT_MS: u64 = 28_800_000;
21/// Default absolute lifetime for one foreground watch session.
22pub const DEFAULT_MAX_WATCH_SESSION_WALL_TIME_MS: u64 = 86_400_000;
23/// Default minimum delay between watched sync pass starts.
24pub const DEFAULT_MIN_WATCH_RESCAN_INTERVAL_MS: u64 = 10_000;
25/// Default maximum retained historical checkpoint-cache bytes.
26pub const DEFAULT_MAX_CHECKPOINT_CACHE_BYTES: u64 = 10_737_418_240;
27
28/// Optional operator-owned execution-policy overrides from the workspace manifest.
29#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
30#[serde(rename_all = "camelCase", deny_unknown_fields)]
31pub struct ExecutionPolicyOverrides {
32    /// Optional maximum wall time for one scan or sync pass.
33    pub max_scan_wall_time_ms: Option<u64>,
34    /// Optional maximum time without verified forward progress.
35    pub max_no_progress_time_ms: Option<u64>,
36    /// Optional per-repository `CodeGraph` synchronization wall time.
37    #[serde(rename = "maxCodeGraphSyncWallTimeMsPerRepo")]
38    pub max_codegraph_sync_wall_time_ms_per_repo: Option<u64>,
39    /// Optional maximum worker resident memory.
40    pub max_worker_memory_bytes: Option<u64>,
41    /// Optional cooperative shutdown grace period.
42    pub graceful_termination_ms: Option<u64>,
43    /// Optional watcher inactivity lease.
44    pub watch_idle_timeout_ms: Option<u64>,
45    /// Optional absolute watcher-session lifetime.
46    pub max_watch_session_wall_time_ms: Option<u64>,
47    /// Optional minimum delay between watched sync pass starts.
48    pub min_watch_rescan_interval_ms: Option<u64>,
49    /// Optional maximum retained checkpoint-cache bytes.
50    pub max_checkpoint_cache_bytes: Option<u64>,
51}
52
53/// Effective global execution policy for one workspace operation.
54#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
55#[serde(rename_all = "camelCase")]
56pub struct ExecutionPolicy {
57    /// Maximum wall time for one scan or sync pass.
58    pub max_scan_wall_time_ms: u64,
59    /// Maximum time without verified forward progress.
60    pub max_no_progress_time_ms: u64,
61    /// Maximum wall time for one repository-local `CodeGraph` synchronization.
62    #[serde(rename = "maxCodeGraphSyncWallTimeMsPerRepo")]
63    pub max_codegraph_sync_wall_time_ms_per_repo: u64,
64    /// Maximum resident memory accepted for one worker process.
65    pub max_worker_memory_bytes: u64,
66    /// Cooperative shutdown grace period before forced termination.
67    pub graceful_termination_ms: u64,
68    /// Inactivity lease for a foreground watch session.
69    pub watch_idle_timeout_ms: u64,
70    /// Absolute lifetime for one foreground watch session.
71    pub max_watch_session_wall_time_ms: u64,
72    /// Minimum delay between watched sync pass starts.
73    pub min_watch_rescan_interval_ms: u64,
74    /// Maximum retained historical checkpoint-cache bytes.
75    pub max_checkpoint_cache_bytes: u64,
76}
77
78impl Default for ExecutionPolicy {
79    fn default() -> Self {
80        Self {
81            max_scan_wall_time_ms: DEFAULT_MAX_SCAN_WALL_TIME_MS,
82            max_no_progress_time_ms: DEFAULT_MAX_NO_PROGRESS_TIME_MS,
83            max_codegraph_sync_wall_time_ms_per_repo:
84                DEFAULT_MAX_CODEGRAPH_SYNC_WALL_TIME_MS_PER_REPO,
85            max_worker_memory_bytes: DEFAULT_MAX_WORKER_MEMORY_BYTES,
86            graceful_termination_ms: DEFAULT_GRACEFUL_TERMINATION_MS,
87            watch_idle_timeout_ms: DEFAULT_WATCH_IDLE_TIMEOUT_MS,
88            max_watch_session_wall_time_ms: DEFAULT_MAX_WATCH_SESSION_WALL_TIME_MS,
89            min_watch_rescan_interval_ms: DEFAULT_MIN_WATCH_RESCAN_INTERVAL_MS,
90            max_checkpoint_cache_bytes: DEFAULT_MAX_CHECKPOINT_CACHE_BYTES,
91        }
92    }
93}
94
95impl ExecutionPolicy {
96    /// Resolves a partial global override and validates the effective relationships.
97    ///
98    /// # Errors
99    ///
100    /// Returns [`InvalidExecutionPolicy`] for zero, unrepresentable, or inconsistent values.
101    pub fn resolve(
102        overrides: Option<&ExecutionPolicyOverrides>,
103    ) -> Result<Self, InvalidExecutionPolicy> {
104        let mut policy = Self::default();
105        if let Some(values) = overrides {
106            macro_rules! apply {
107                ($field:ident) => {
108                    if let Some(value) = values.$field {
109                        policy.$field = value;
110                    }
111                };
112            }
113            apply!(max_scan_wall_time_ms);
114            apply!(max_no_progress_time_ms);
115            apply!(max_codegraph_sync_wall_time_ms_per_repo);
116            apply!(max_worker_memory_bytes);
117            apply!(graceful_termination_ms);
118            apply!(watch_idle_timeout_ms);
119            apply!(max_watch_session_wall_time_ms);
120            apply!(min_watch_rescan_interval_ms);
121            apply!(max_checkpoint_cache_bytes);
122        }
123        policy.validate()?;
124        Ok(policy)
125    }
126
127    fn validate(&self) -> Result<(), InvalidExecutionPolicy> {
128        for (field, value) in self.canonical_values() {
129            let invalid_bytes = field.ends_with("Bytes") && usize::try_from(value).is_err();
130            let invalid_sqlite_quota =
131                field == "maxCheckpointCacheBytes" && i64::try_from(value).is_err();
132            let invalid_deadline = field.ends_with("Ms")
133                && Instant::now()
134                    .checked_add(Duration::from_millis(value))
135                    .is_none();
136            if value == 0 || invalid_bytes || invalid_sqlite_quota || invalid_deadline {
137                return Err(InvalidExecutionPolicy::InvalidValue { field, value });
138            }
139        }
140        Self::require_not_greater(
141            "maxNoProgressTimeMs",
142            self.max_no_progress_time_ms,
143            "maxScanWallTimeMs",
144            self.max_scan_wall_time_ms,
145        )?;
146        Self::require_not_greater(
147            "maxCodeGraphSyncWallTimeMsPerRepo",
148            self.max_codegraph_sync_wall_time_ms_per_repo,
149            "maxScanWallTimeMs",
150            self.max_scan_wall_time_ms,
151        )?;
152        Self::require_not_greater(
153            "gracefulTerminationMs",
154            self.graceful_termination_ms,
155            "maxNoProgressTimeMs",
156            self.max_no_progress_time_ms,
157        )?;
158        Self::require_not_greater(
159            "watchIdleTimeoutMs",
160            self.watch_idle_timeout_ms,
161            "maxWatchSessionWallTimeMs",
162            self.max_watch_session_wall_time_ms,
163        )?;
164        Self::require_not_greater(
165            "minWatchRescanIntervalMs",
166            self.min_watch_rescan_interval_ms,
167            "watchIdleTimeoutMs",
168            self.watch_idle_timeout_ms,
169        )
170    }
171
172    fn require_not_greater(
173        field: &'static str,
174        value: u64,
175        maximum_field: &'static str,
176        maximum: u64,
177    ) -> Result<(), InvalidExecutionPolicy> {
178        if value > maximum {
179            return Err(InvalidExecutionPolicy::InvalidRelationship {
180                field,
181                value,
182                maximum_field,
183                maximum,
184            });
185        }
186        Ok(())
187    }
188
189    fn canonical_values(&self) -> [(&'static str, u64); 9] {
190        [
191            ("maxScanWallTimeMs", self.max_scan_wall_time_ms),
192            ("maxNoProgressTimeMs", self.max_no_progress_time_ms),
193            (
194                "maxCodeGraphSyncWallTimeMsPerRepo",
195                self.max_codegraph_sync_wall_time_ms_per_repo,
196            ),
197            ("maxWorkerMemoryBytes", self.max_worker_memory_bytes),
198            ("gracefulTerminationMs", self.graceful_termination_ms),
199            ("watchIdleTimeoutMs", self.watch_idle_timeout_ms),
200            (
201                "maxWatchSessionWallTimeMs",
202                self.max_watch_session_wall_time_ms,
203            ),
204            (
205                "minWatchRescanIntervalMs",
206                self.min_watch_rescan_interval_ms,
207            ),
208            ("maxCheckpointCacheBytes", self.max_checkpoint_cache_bytes),
209        ]
210    }
211
212    /// Returns the stable canonical fingerprint of the effective operational policy.
213    #[must_use]
214    pub fn fingerprint(&self) -> String {
215        let canonical = self
216            .canonical_values()
217            .into_iter()
218            .map(|(name, value)| format!("{name}={value}"))
219            .collect::<Vec<_>>()
220            .join(";");
221        stable_id("execution-policy", &canonical)
222    }
223}
224
225/// Invalid operator-owned execution policy.
226#[derive(Debug, Clone, PartialEq, Eq, Error)]
227pub enum InvalidExecutionPolicy {
228    /// One effective value is zero or cannot be represented internally.
229    #[error("execution policy `{field}` must be positive and representable; received {value}")]
230    InvalidValue {
231        /// Manifest field containing the invalid value.
232        field: &'static str,
233        /// Rejected numeric value.
234        value: u64,
235    },
236    /// One subordinate deadline exceeds its containing deadline.
237    #[error("execution policy `{field}` ({value}) must not exceed `{maximum_field}` ({maximum})")]
238    InvalidRelationship {
239        /// Subordinate manifest field.
240        field: &'static str,
241        /// Supplied subordinate value.
242        value: u64,
243        /// Containing manifest field.
244        maximum_field: &'static str,
245        /// Supplied containing value.
246        maximum: u64,
247    },
248}
249
250/// Observable phase of one supervised scan or synchronization pass.
251#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
252#[serde(rename_all = "snake_case")]
253pub enum JobPhase {
254    /// Workspace configuration and checkout validation.
255    Configuration,
256    /// Filesystem artifact discovery.
257    Discovery,
258    /// Bounded content fingerprinting.
259    Fingerprinting,
260    /// Source-owned extraction and batch encoding.
261    Extraction,
262    /// Candidate graph assembly and linking.
263    GraphAssembly,
264    /// Deterministic community analysis.
265    Communities,
266    /// Atomic snapshot persistence.
267    Publication,
268    /// External repository-local `CodeGraph` synchronization.
269    CodeGraphSync,
270}
271
272/// Resource exhausted by one supervised workspace operation.
273#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
274#[serde(rename_all = "snake_case")]
275pub enum ExecutionResource {
276    /// Absolute wall time.
277    WallTimeMs,
278    /// Time without verified forward progress.
279    NoProgressTimeMs,
280    /// Resident worker memory.
281    WorkerMemoryBytes,
282    /// Deterministic work accounting overflow.
283    WorkUnits,
284    /// The worker terminated without a valid final protocol message.
285    WorkerProcess,
286    /// The worker protocol exceeded its byte limit or was malformed.
287    WorkerProtocolBytes,
288    /// The operator or hosting agent requested cancellation.
289    Cancellation,
290}
291
292/// Observable resource accounting attached to a completed scan or sync pass.
293#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
294#[serde(rename_all = "camelCase")]
295pub struct ExecutionSummary {
296    /// Unique identifier assigned by the supervisor to this pass.
297    pub run_id: String,
298    /// Monotonic wall-clock duration observed by the supervisor.
299    pub duration_ms: u64,
300    /// Largest combined resident set observed for the worker process tree.
301    pub peak_worker_memory_bytes: u64,
302    /// Number of completed units reported through the progress protocol.
303    pub completed_work_units: u64,
304    /// Number of deterministic checkpoint records reused by this pass.
305    pub checkpoint_hits: u64,
306    /// Number of complete deterministic checkpoint records written by this pass.
307    pub checkpoints_written: u64,
308    /// Number of artifact-extractor invocations measured by the worker.
309    pub measured_artifacts: u64,
310    /// Median artifact-extractor duration in milliseconds.
311    pub artifact_duration_p50_ms: u64,
312    /// 95th-percentile artifact-extractor duration in milliseconds.
313    pub artifact_duration_p95_ms: u64,
314    /// 99th-percentile artifact-extractor duration in milliseconds.
315    pub artifact_duration_p99_ms: u64,
316}
317
318/// Typed failure produced when one supervised execution resource is exhausted.
319#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema, Error)]
320#[error(
321    "execution `{run_id}` exceeded {resource:?} during {phase:?}: observed {observed}, maximum {maximum}, completed {completed_units} units"
322)]
323pub struct ExecutionLimitExceeded {
324    /// Stable identifier of the supervised operation.
325    pub run_id: String,
326    /// Phase active when the resource was exhausted.
327    pub phase: JobPhase,
328    /// Exhausted resource.
329    pub resource: ExecutionResource,
330    /// Observed resource value.
331    pub observed: u64,
332    /// Effective maximum.
333    pub maximum: u64,
334    /// Verified work units completed before rejection.
335    pub completed_units: u64,
336}
337
338/// Injectable monotonic time source used by execution watchdogs.
339pub trait MonotonicClock: std::fmt::Debug + Send + Sync {
340    /// Duration since an arbitrary stable origin.
341    fn now(&self) -> Duration;
342}
343
344#[derive(Debug)]
345struct SystemMonotonicClock {
346    origin: Instant,
347}
348
349impl SystemMonotonicClock {
350    fn new() -> Self {
351        Self {
352            origin: Instant::now(),
353        }
354    }
355}
356
357impl MonotonicClock for SystemMonotonicClock {
358    fn now(&self) -> Duration {
359        self.origin.elapsed()
360    }
361}
362
363/// Monotonic progress tracker used inside one worker process.
364#[derive(Debug)]
365pub struct ScanJobTracker {
366    run_id: String,
367    policy: ExecutionPolicy,
368    clock: Arc<dyn MonotonicClock>,
369    started: Duration,
370    last_progress: Duration,
371    phase: JobPhase,
372    completed_units: u64,
373}
374
375impl ScanJobTracker {
376    /// Starts a tracker for one run at configuration validation.
377    #[must_use]
378    pub fn new(run_id: impl Into<String>, policy: ExecutionPolicy) -> Self {
379        Self::with_clock(run_id, policy, Arc::new(SystemMonotonicClock::new()))
380    }
381
382    /// Starts a tracker with an injected monotonic clock for deterministic execution tests.
383    #[must_use]
384    pub fn with_clock(
385        run_id: impl Into<String>,
386        policy: ExecutionPolicy,
387        clock: Arc<dyn MonotonicClock>,
388    ) -> Self {
389        let now = clock.now();
390        Self {
391            run_id: run_id.into(),
392            policy,
393            clock,
394            started: now,
395            last_progress: now,
396            phase: JobPhase::Configuration,
397            completed_units: 0,
398        }
399    }
400
401    /// Changes phase after checking the active deadlines.
402    ///
403    /// # Errors
404    ///
405    /// Returns [`ExecutionLimitExceeded`] when wall time or no-progress time is exhausted.
406    pub fn enter_phase(&mut self, phase: JobPhase) -> Result<(), ExecutionLimitExceeded> {
407        self.check_time()?;
408        self.phase = phase;
409        Ok(())
410    }
411
412    /// Charges verified completed work using checked arithmetic.
413    ///
414    /// # Errors
415    ///
416    /// Returns [`ExecutionLimitExceeded`] on arithmetic overflow or an exhausted deadline.
417    pub fn progress(&mut self, amount: u64) -> Result<(), ExecutionLimitExceeded> {
418        let completed_units = self
419            .completed_units
420            .checked_add(amount)
421            .ok_or_else(|| self.exceeded(ExecutionResource::WorkUnits, u64::MAX, u64::MAX - 1))?;
422        self.check_time()?;
423        self.completed_units = completed_units;
424        self.last_progress = self.clock.now();
425        Ok(())
426    }
427
428    /// Checks monotonic wall time and time since the last verified progress.
429    ///
430    /// # Errors
431    ///
432    /// Returns [`ExecutionLimitExceeded`] when either effective duration is exhausted.
433    pub fn check_time(&self) -> Result<(), ExecutionLimitExceeded> {
434        let now = self.clock.now();
435        self.check_duration(
436            ExecutionResource::WallTimeMs,
437            now.saturating_sub(self.started),
438            self.policy.max_scan_wall_time_ms,
439        )?;
440        self.check_duration(
441            ExecutionResource::NoProgressTimeMs,
442            now.saturating_sub(self.last_progress),
443            self.policy.max_no_progress_time_ms,
444        )
445    }
446
447    fn check_duration(
448        &self,
449        resource: ExecutionResource,
450        observed: Duration,
451        maximum: u64,
452    ) -> Result<(), ExecutionLimitExceeded> {
453        let observed = u64::try_from(observed.as_millis()).unwrap_or(u64::MAX);
454        if observed > maximum {
455            return Err(self.exceeded(resource, observed, maximum));
456        }
457        Ok(())
458    }
459
460    fn exceeded(
461        &self,
462        resource: ExecutionResource,
463        observed: u64,
464        maximum: u64,
465    ) -> ExecutionLimitExceeded {
466        ExecutionLimitExceeded {
467            run_id: self.run_id.clone(),
468            phase: self.phase,
469            resource,
470            observed,
471            maximum,
472            completed_units: self.completed_units,
473        }
474    }
475}
476
477#[cfg(test)]
478mod tests {
479    use std::sync::atomic::{AtomicU64, Ordering};
480
481    use super::*;
482
483    #[derive(Debug, Default)]
484    struct FakeClock {
485        milliseconds: AtomicU64,
486    }
487
488    impl FakeClock {
489        fn advance(&self, milliseconds: u64) {
490            self.milliseconds.fetch_add(milliseconds, Ordering::Relaxed);
491        }
492    }
493
494    impl MonotonicClock for FakeClock {
495        fn now(&self) -> Duration {
496            Duration::from_millis(self.milliseconds.load(Ordering::Relaxed))
497        }
498    }
499
500    #[test]
501    fn defaults_should_be_generous_and_finite() {
502        let policy = ExecutionPolicy::default();
503
504        assert_eq!(policy.max_scan_wall_time_ms, 21_600_000);
505        assert_eq!(policy.max_worker_memory_bytes, 17_179_869_184);
506        assert_eq!(policy.max_checkpoint_cache_bytes, 10_737_418_240);
507    }
508
509    #[test]
510    fn partial_override_should_preserve_other_defaults() {
511        let policy = ExecutionPolicy::resolve(Some(&ExecutionPolicyOverrides {
512            max_scan_wall_time_ms: Some(28_800_000),
513            ..ExecutionPolicyOverrides::default()
514        }))
515        .expect("valid override");
516
517        assert_eq!(policy.max_scan_wall_time_ms, 28_800_000);
518        assert_eq!(policy.max_no_progress_time_ms, 300_000);
519    }
520
521    #[test]
522    fn zero_should_be_rejected() {
523        let error = ExecutionPolicy::resolve(Some(&ExecutionPolicyOverrides {
524            max_worker_memory_bytes: Some(0),
525            ..ExecutionPolicyOverrides::default()
526        }))
527        .expect_err("zero must fail");
528
529        assert!(matches!(
530            error,
531            InvalidExecutionPolicy::InvalidValue {
532                field: "maxWorkerMemoryBytes",
533                value: 0
534            }
535        ));
536    }
537
538    #[test]
539    fn technically_unrepresentable_values_should_be_rejected() {
540        let quota = ExecutionPolicy::resolve(Some(&ExecutionPolicyOverrides {
541            max_checkpoint_cache_bytes: Some(u64::MAX),
542            ..ExecutionPolicyOverrides::default()
543        }))
544        .expect_err("SQLite quota overflow must fail");
545
546        assert!(matches!(quota, InvalidExecutionPolicy::InvalidValue { .. }));
547    }
548
549    #[test]
550    fn subordinate_deadline_should_not_exceed_scan_deadline() {
551        let error = ExecutionPolicy::resolve(Some(&ExecutionPolicyOverrides {
552            max_scan_wall_time_ms: Some(1_000),
553            max_no_progress_time_ms: Some(1_001),
554            max_codegraph_sync_wall_time_ms_per_repo: Some(1_000),
555            graceful_termination_ms: Some(500),
556            ..ExecutionPolicyOverrides::default()
557        }))
558        .expect_err("relationship must fail");
559
560        assert!(matches!(
561            error,
562            InvalidExecutionPolicy::InvalidRelationship {
563                field: "maxNoProgressTimeMs",
564                ..
565            }
566        ));
567    }
568
569    #[test]
570    fn fingerprint_should_ignore_yaml_field_order() {
571        let first = ExecutionPolicy::default();
572        let second = ExecutionPolicy::resolve(Some(&ExecutionPolicyOverrides::default()))
573            .expect("defaults valid");
574
575        assert_eq!(first.fingerprint(), second.fingerprint());
576    }
577
578    #[test]
579    fn injected_clock_should_accept_exact_deadline_and_reject_one_unit_over() {
580        let clock = Arc::new(FakeClock::default());
581        let policy = ExecutionPolicy {
582            max_scan_wall_time_ms: 10,
583            max_no_progress_time_ms: 10,
584            ..ExecutionPolicy::default()
585        };
586        let tracker = ScanJobTracker::with_clock("run", policy, clock.clone());
587
588        clock.advance(10);
589        tracker.check_time().expect("exact deadline is inclusive");
590        clock.advance(1);
591        let error = tracker.check_time().expect_err("one over must fail");
592
593        assert_eq!(error.resource, ExecutionResource::WallTimeMs);
594        assert_eq!(error.observed, 11);
595        assert_eq!(error.maximum, 10);
596    }
597
598    #[test]
599    fn phase_changes_should_not_fake_progress() {
600        let clock = Arc::new(FakeClock::default());
601        let policy = ExecutionPolicy {
602            max_scan_wall_time_ms: 100,
603            max_no_progress_time_ms: 5,
604            ..ExecutionPolicy::default()
605        };
606        let mut tracker = ScanJobTracker::with_clock("run", policy, clock.clone());
607
608        clock.advance(5);
609        tracker
610            .enter_phase(JobPhase::Discovery)
611            .expect("exact idle deadline is inclusive");
612        clock.advance(1);
613        let error = tracker
614            .enter_phase(JobPhase::Fingerprinting)
615            .expect_err("phase churn must not renew watchdog");
616
617        assert_eq!(error.resource, ExecutionResource::NoProgressTimeMs);
618    }
619}
code_system_graph_core/execution_policy.rs

code_system_graph_core/
execution_policy.rs
