diff --git a/README.md b/README.md index 3bbf0c94f..00c196d7c 100644 --- a/README.md +++ b/README.md @@ -101,6 +101,7 @@ cargo run -p decodex --bin decodex -- probe stdio:// cargo run -p decodex --bin decodex -- project list cargo run -p decodex --bin decodex -- status cargo run -p decodex --bin decodex -- diagnose --json +cargo run -p decodex --bin decodex -- maintenance prune --dry-run cargo run -p decodex --bin decodex -- run --dry-run cargo run -p decodex --bin decodex -- serve --interval 60s --listen-address 127.0.0.1:8912 ``` @@ -143,6 +144,13 @@ matching `accounts.jsonl` entry. `~/.codex/decodex/agent-evidence//` and prints the same handoff index for repair agents. +`decodex maintenance prune --dry-run` reports local Decodex storage retention +candidates without applying retention changes. Add `--apply` to rotate +oversized local logs and agent-evidence event streams, prune old backup files, compact +old terminal-run protocol events after preserving their summary, and checkpoint the +SQLite WAL. `decodex serve` also runs the auto-safe subset at startup and periodically +while it is polling. + ## Static Site The public site is an Astro static site under `site/`. It renders checked-in content and diff --git a/apps/decodex/src/cli.rs b/apps/decodex/src/cli.rs index a4fa2c03e..ef410b6b4 100644 --- a/apps/decodex/src/cli.rs +++ b/apps/decodex/src/cli.rs @@ -18,6 +18,7 @@ use crate::{ accounts::{self, AccountImportRequest, AccountLoginRequest, AccountUseRequest}, agent, archive_hygiene::{self, ArchiveHygieneRequest}, + maintenance::{self, MaintenanceMode, MaintenancePruneRequest, MaintenanceScope}, manual::{self, ManualCommitRequest, ManualLandRequest}, orchestrator::{self, DiagnoseRequest, IssueDispatchMode, RunOnceRequest, ServeRequest}, prelude::eyre, @@ -62,6 +63,7 @@ impl Cli { Command::Diagnose(args) => args.run(config_path), Command::Recover(args) => args.run(config_path), Command::ArchiveLinear(args) => args.run(config_path), + Command::Maintenance(args) => args.run(), Command::Account(args) => args.run(), Command::Probe(args) => args.run(), Command::Attempt(args) => args.run(config_path), @@ -137,6 +139,8 @@ enum Command { Recover(RecoverCommand), /// Dry-run or archive old terminal Linear issues by repo label. ArchiveLinear(ArchiveLinearCommand), + /// Maintain local Decodex logs, evidence, backups, and runtime storage. + Maintenance(MaintenanceCommand), /// Manage the global Decodex Codex account pool. Account(AccountCommand), /// Validate the local app-server integration boundary. @@ -620,6 +624,49 @@ impl ArchiveLinearCommand { } } +#[derive(Debug, Args)] +struct MaintenanceCommand { + #[command(subcommand)] + command: MaintenanceSubcommand, +} +impl MaintenanceCommand { + fn run(&self) -> crate::prelude::Result<()> { + match &self.command { + MaintenanceSubcommand::Prune(args) => args.run(), + } + } +} + +#[derive(Debug, Subcommand)] +enum MaintenanceSubcommand { + /// Inspect or apply local Decodex storage retention. + Prune(MaintenancePruneCommand), +} + +#[derive(Debug, Args)] +struct MaintenancePruneCommand { + /// Report candidates without applying retention changes. + #[arg(long, conflicts_with = "apply")] + dry_run: bool, + /// Apply safe file retention, state-aware runtime compaction, and WAL checkpointing. + #[arg(long, conflicts_with = "dry_run")] + apply: bool, + /// Emit the maintenance report as JSON. + #[arg(long)] + json: bool, +} +impl MaintenancePruneCommand { + fn run(&self) -> crate::prelude::Result<()> { + let mode = if self.apply { MaintenanceMode::Apply } else { MaintenanceMode::DryRun }; + + maintenance::run_prune_command(MaintenancePruneRequest { + mode, + scope: MaintenanceScope::Full, + json: self.json, + }) + } +} + #[derive(Debug, Args)] struct ProbeCommand { /// Override the expected app-server transport during probing. diff --git a/apps/decodex/src/lib.rs b/apps/decodex/src/lib.rs index 308b8c177..f27241f4b 100644 --- a/apps/decodex/src/lib.rs +++ b/apps/decodex/src/lib.rs @@ -14,6 +14,7 @@ mod commit_message; mod default_branch_sync; mod git_credentials; mod github; +mod maintenance; mod manual; mod orchestrator; mod pull_request; @@ -57,8 +58,8 @@ fn init_tracing() -> Result { let (non_blocking, guard) = tracing_appender::non_blocking( RollingFileAppender::builder() - .rotation(Rotation::WEEKLY) - .max_log_files(3) + .rotation(Rotation::DAILY) + .max_log_files(30) .filename_suffix("log") .build(log_dir)?, ); diff --git a/apps/decodex/src/maintenance.rs b/apps/decodex/src/maintenance.rs new file mode 100644 index 000000000..f3094e7e6 --- /dev/null +++ b/apps/decodex/src/maintenance.rs @@ -0,0 +1,1110 @@ +use std::{ + cmp::Reverse, + collections::BTreeMap, + fs::{self, Metadata, OpenOptions}, + io::{self, Write as _}, + path::{Path, PathBuf}, + time::{Duration, SystemTime}, +}; + +use rusqlite::{self, Connection}; +use serde::Serialize; +use time::{OffsetDateTime, format_description::well_known::Rfc3339}; + +use crate::{ + prelude::{Result, eyre}, + runtime, +}; + +const DEFAULT_LOG_ROTATE_BYTES: u64 = 10 * 1_024 * 1_024; +const DEFAULT_LOG_RETENTION_DAYS: u64 = 30; +const DEFAULT_EVIDENCE_ROTATE_BYTES: u64 = 10 * 1_024 * 1_024; +const DEFAULT_EVIDENCE_RETENTION_DAYS: u64 = 30; +const DEFAULT_PROTOCOL_EVENT_RETENTION_DAYS: i64 = 14; +const DEFAULT_BACKUP_KEEP_RECENT: usize = 3; +const DEFAULT_BACKUP_RETENTION_DAYS: u64 = 7; + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum MaintenanceMode { + DryRun, + Apply, +} +impl MaintenanceMode { + fn as_str(self) -> &'static str { + match self { + Self::DryRun => "dry-run", + Self::Apply => "apply", + } + } + + fn applies(self) -> bool { + matches!(self, Self::Apply) + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum MaintenanceScope { + Full, + AutoSafe, +} + +#[derive(Clone, Copy, Debug)] +pub(crate) struct MaintenancePruneRequest { + pub(crate) mode: MaintenanceMode, + pub(crate) scope: MaintenanceScope, + pub(crate) json: bool, +} + +#[derive(Debug, Serialize)] +pub(crate) struct MaintenanceReport { + schema: &'static str, + mode: String, + scope: String, + generated_at: String, + pub(crate) logs: FileMaintenanceReport, + pub(crate) agent_evidence: FileMaintenanceReport, + pub(crate) backups: BackupMaintenanceReport, + pub(crate) runtime: RuntimeMaintenanceReport, + pub(crate) wal_checkpoint: Option, +} + +#[derive(Debug, Default, Serialize)] +pub(crate) struct FileMaintenanceReport { + root: String, + rotate_candidates: usize, + pub(crate) rotated_files: usize, + rotate_bytes: u64, + delete_candidates: usize, + pub(crate) deleted_files: usize, + delete_bytes: u64, + actions: Vec, +} + +#[derive(Debug, Default, Serialize)] +pub(crate) struct BackupMaintenanceReport { + root: String, + delete_candidates: usize, + pub(crate) deleted_files: usize, + delete_bytes: u64, + actions: Vec, +} + +#[derive(Debug, Default, Serialize)] +pub(crate) struct RuntimeMaintenanceReport { + database_path: String, + protocol_event_retention_days: i64, + protected_run_count: usize, + protocol_run_candidates: usize, + protocol_event_candidates: u64, + compacted_runs: usize, + compacted_events: u64, + actions: Vec, +} + +#[derive(Debug, Serialize)] +pub(crate) struct WalCheckpointReport { + pub(crate) mode: &'static str, + busy: i64, + log_frames: i64, + checkpointed_frames: i64, +} + +#[derive(Clone, Copy)] +struct MaintenancePolicy { + log_rotate_bytes: u64, + log_retention: Duration, + evidence_rotate_bytes: u64, + evidence_retention: Duration, + protocol_event_retention_days: i64, + backup_keep_recent: usize, + backup_retention: Duration, +} +impl MaintenancePolicy { + fn default() -> Self { + Self { + log_rotate_bytes: DEFAULT_LOG_ROTATE_BYTES, + log_retention: Duration::from_secs(DEFAULT_LOG_RETENTION_DAYS * 24 * 60 * 60), + evidence_rotate_bytes: DEFAULT_EVIDENCE_ROTATE_BYTES, + evidence_retention: Duration::from_secs(DEFAULT_EVIDENCE_RETENTION_DAYS * 24 * 60 * 60), + protocol_event_retention_days: DEFAULT_PROTOCOL_EVENT_RETENTION_DAYS, + backup_keep_recent: DEFAULT_BACKUP_KEEP_RECENT, + backup_retention: Duration::from_secs(DEFAULT_BACKUP_RETENTION_DAYS * 24 * 60 * 60), + } + } +} + +#[derive(Debug, Serialize)] +struct FileMaintenanceAction { + action: &'static str, + path: String, + bytes: u64, + target: Option, + reason: String, +} + +#[derive(Debug, Serialize)] +struct RuntimeMaintenanceAction { + action: &'static str, + run_id: String, + issue_id: String, + status: String, + event_count: u64, + last_event_at: Option, + reason: String, +} + +struct RuntimeProtocolCandidate { + run_id: String, + issue_id: String, + status: String, + event_count: u64, + last_sequence_number: Option, + last_event_type: Option, + last_event_at: Option, + last_event_at_unix: Option, +} + +#[derive(Clone)] +struct BackupCandidate { + path: PathBuf, + bytes: u64, + modified: SystemTime, +} + +pub(crate) fn run_prune_command(request: MaintenancePruneRequest) -> Result<()> { + let report = run_prune_with_policy(request, MaintenancePolicy::default())?; + + if request.json { + println!("{}", serde_json::to_string_pretty(&report)?); + } else { + print_prune_report(&report)?; + } + + Ok(()) +} + +pub(crate) fn run_auto_safe_prune() -> Result { + run_prune_with_policy( + MaintenancePruneRequest { + mode: MaintenanceMode::Apply, + scope: MaintenanceScope::AutoSafe, + json: false, + }, + MaintenancePolicy::default(), + ) +} + +pub(crate) fn ensure_protocol_event_summary_table(connection: &Connection) -> Result<()> { + connection.execute_batch( + "CREATE TABLE IF NOT EXISTS protocol_event_summaries ( + run_id TEXT PRIMARY KEY NOT NULL, + event_count INTEGER NOT NULL, + last_sequence_number INTEGER, + last_event_type TEXT, + last_event_at TEXT, + last_event_at_unix INTEGER, + compacted_at TEXT NOT NULL, + compacted_at_unix INTEGER NOT NULL + );", + )?; + + Ok(()) +} + +fn run_prune_with_policy( + request: MaintenancePruneRequest, + policy: MaintenancePolicy, +) -> Result { + let generated_at = OffsetDateTime::now_utc(); + let system_now = SystemTime::now(); + let logs = maintain_logs(request.mode, policy, system_now, generated_at)?; + let agent_evidence = maintain_agent_evidence(request.mode, policy, system_now, generated_at)?; + let backups = maintain_backups(request.mode, policy, system_now)?; + let runtime = maintain_runtime(request.mode, request.scope, policy, generated_at)?; + let wal_checkpoint = maintain_wal(request.mode, request.scope)?; + + Ok(MaintenanceReport { + schema: "decodex.maintenance_report/1", + mode: request.mode.as_str().to_owned(), + scope: match request.scope { + MaintenanceScope::Full => String::from("full"), + MaintenanceScope::AutoSafe => String::from("auto-safe"), + }, + generated_at: generated_at.format(&Rfc3339)?, + logs, + agent_evidence, + backups, + runtime, + wal_checkpoint, + }) +} + +fn maintain_logs( + mode: MaintenanceMode, + policy: MaintenancePolicy, + system_now: SystemTime, + generated_at: OffsetDateTime, +) -> Result { + let root = runtime::log_dir()?; + let mut report = FileMaintenanceReport { + root: root.display().to_string(), + ..FileMaintenanceReport::default() + }; + + if !root.exists() { + return Ok(report); + } + + for entry in fs::read_dir(&root)? { + let entry = entry?; + let path = entry.path(); + + if !path.is_file() + || path.extension().and_then(|extension| extension.to_str()) != Some("log") + { + continue; + } + + let metadata = entry.metadata()?; + let size = metadata.len(); + + if file_is_older_than(&metadata, system_now, policy.log_retention) { + report.delete_candidates += 1; + report.delete_bytes = report.delete_bytes.saturating_add(size); + + report.actions.push(FileMaintenanceAction { + action: "delete", + path: path.display().to_string(), + bytes: size, + target: None, + reason: format!("log is older than {DEFAULT_LOG_RETENTION_DAYS} days"), + }); + + if mode.applies() { + fs::remove_file(&path)?; + + report.deleted_files += 1; + } + + continue; + } + if size > policy.log_rotate_bytes { + let rotated_path = rotated_path(&path, generated_at)?; + + report.rotate_candidates += 1; + report.rotate_bytes = report.rotate_bytes.saturating_add(size); + + report.actions.push(FileMaintenanceAction { + action: "rotate", + path: path.display().to_string(), + bytes: size, + target: Some(rotated_path.display().to_string()), + reason: format!( + "size {} exceeds {} byte log rotation threshold", + size, policy.log_rotate_bytes + ), + }); + + if mode.applies() { + copy_truncate(&path, &rotated_path)?; + + report.rotated_files += 1; + } + } + } + + Ok(report) +} + +fn maintain_agent_evidence( + mode: MaintenanceMode, + policy: MaintenancePolicy, + system_now: SystemTime, + generated_at: OffsetDateTime, +) -> Result { + let root = runtime::agent_evidence_dir()?; + let mut report = FileMaintenanceReport { + root: root.display().to_string(), + ..FileMaintenanceReport::default() + }; + + if !root.exists() { + return Ok(report); + } + + for entry in fs::read_dir(&root)? { + let entry = entry?; + let service_root = entry.path(); + + if !service_root.is_dir() { + continue; + } + + let events_path = service_root.join("events.jsonl"); + + if events_path.is_file() { + let metadata = fs::metadata(&events_path)?; + let size = metadata.len(); + + if size > policy.evidence_rotate_bytes { + let rotated_path = rotated_path(&events_path, generated_at)?; + + report.rotate_candidates += 1; + report.rotate_bytes = report.rotate_bytes.saturating_add(size); + + report.actions.push(FileMaintenanceAction { + action: "rotate", + path: events_path.display().to_string(), + bytes: size, + target: Some(rotated_path.display().to_string()), + reason: format!( + "size {} exceeds {} byte agent-evidence event threshold", + size, policy.evidence_rotate_bytes + ), + }); + + if mode.applies() { + copy_truncate(&events_path, &rotated_path)?; + + report.rotated_files += 1; + } + } + } + + for event_entry in fs::read_dir(&service_root)? { + let event_entry = event_entry?; + let path = event_entry.path(); + let Some(file_name) = path.file_name().and_then(|name| name.to_str()) else { + continue; + }; + + if !path.is_file() + || !file_name.starts_with("events.") + || path.extension().and_then(|extension| extension.to_str()) != Some("jsonl") + { + continue; + } + + let metadata = event_entry.metadata()?; + + if file_is_older_than(&metadata, system_now, policy.evidence_retention) { + let size = metadata.len(); + + report.delete_candidates += 1; + report.delete_bytes = report.delete_bytes.saturating_add(size); + + report.actions.push(FileMaintenanceAction { + action: "delete", + path: path.display().to_string(), + bytes: size, + target: None, + reason: format!( + "rotated agent-evidence event stream is older than {DEFAULT_EVIDENCE_RETENTION_DAYS} days" + ), + }); + + if mode.applies() { + fs::remove_file(&path)?; + + report.deleted_files += 1; + } + } + } + } + + Ok(report) +} + +fn maintain_backups( + mode: MaintenanceMode, + policy: MaintenancePolicy, + system_now: SystemTime, +) -> Result { + let root = runtime::decodex_home_dir()?; + let mut report = BackupMaintenanceReport { + root: root.display().to_string(), + ..BackupMaintenanceReport::default() + }; + + if !root.exists() { + return Ok(report); + } + + let mut groups = BTreeMap::>::new(); + + collect_backup_candidates(&root, &mut groups)?; + + for candidates in groups.values_mut() { + candidates.sort_by_key(|candidate| Reverse(candidate.modified)); + + for (index, candidate) in candidates.iter().enumerate() { + let young_enough = system_now + .duration_since(candidate.modified) + .map(|age| age <= policy.backup_retention) + .unwrap_or(true); + let recent_enough = index < policy.backup_keep_recent; + + if recent_enough || young_enough { + continue; + } + + report.delete_candidates += 1; + report.delete_bytes = report.delete_bytes.saturating_add(candidate.bytes); + + report.actions.push(FileMaintenanceAction { + action: "delete", + path: candidate.path.display().to_string(), + bytes: candidate.bytes, + target: None, + reason: format!( + "backup is outside the latest {} files and older than {DEFAULT_BACKUP_RETENTION_DAYS} days", + policy.backup_keep_recent + ), + }); + + if mode.applies() { + fs::remove_file(&candidate.path)?; + + report.deleted_files += 1; + } + } + } + + Ok(report) +} + +fn maintain_runtime( + mode: MaintenanceMode, + scope: MaintenanceScope, + policy: MaintenancePolicy, + generated_at: OffsetDateTime, +) -> Result { + let database_path = runtime::runtime_db_path()?; + let mut report = RuntimeMaintenanceReport { + database_path: database_path.display().to_string(), + protocol_event_retention_days: policy.protocol_event_retention_days, + ..RuntimeMaintenanceReport::default() + }; + + if scope == MaintenanceScope::AutoSafe || !database_path.exists() { + return Ok(report); + } + + let mut connection = Connection::open(&database_path)?; + + connection.busy_timeout(Duration::from_secs(5))?; + + if mode.applies() { + ensure_protocol_event_summary_table(&connection)?; + } + + let cutoff_unix = + generated_at.unix_timestamp() - policy.protocol_event_retention_days * 24 * 60 * 60; + let candidates = protocol_event_compaction_candidates(&connection, cutoff_unix)?; + + report.protocol_run_candidates = candidates.len(); + report.protocol_event_candidates = + candidates.iter().map(|candidate| candidate.event_count).sum::(); + report.protected_run_count = protected_protocol_run_count(&connection)?; + + for candidate in &candidates { + report.actions.push(RuntimeMaintenanceAction { + action: "compact-protocol-events", + run_id: candidate.run_id.clone(), + issue_id: candidate.issue_id.clone(), + status: candidate.status.clone(), + event_count: candidate.event_count, + last_event_at: candidate.last_event_at.clone(), + reason: format!( + "terminal run has no active lease, retained worktree, or review marker and its latest protocol event is older than {} days", + policy.protocol_event_retention_days + ), + }); + } + + if mode.applies() && !candidates.is_empty() { + compact_protocol_events(&mut connection, &candidates, generated_at)?; + + report.compacted_runs = candidates.len(); + report.compacted_events = report.protocol_event_candidates; + } + + Ok(report) +} + +fn maintain_wal( + mode: MaintenanceMode, + scope: MaintenanceScope, +) -> Result> { + if mode == MaintenanceMode::DryRun { + return Ok(None); + } + + let database_path = runtime::runtime_db_path()?; + + if !database_path.exists() { + return Ok(None); + } + + let connection = Connection::open(database_path)?; + let checkpoint_mode = match scope { + MaintenanceScope::Full => "TRUNCATE", + MaintenanceScope::AutoSafe => "PASSIVE", + }; + let mut statement = connection.prepare(&format!("PRAGMA wal_checkpoint({checkpoint_mode})"))?; + let report = statement.query_row([], |row| { + Ok(WalCheckpointReport { + mode: checkpoint_mode, + busy: row.get(0)?, + log_frames: row.get(1)?, + checkpointed_frames: row.get(2)?, + }) + })?; + + Ok(Some(report)) +} + +fn protocol_event_compaction_candidates( + connection: &Connection, + cutoff_unix: i64, +) -> Result> { + let mut statement = connection.prepare( + "SELECT + attempts.run_id, + attempts.issue_id, + attempts.status, + totals.event_count, + totals.last_sequence_number, + last.event_type, + last.created_at, + last.created_at_unix + FROM ( + SELECT + run_id, + COUNT(*) AS event_count, + MAX(sequence_number) AS last_sequence_number, + MAX(created_at_unix) AS last_created_at_unix + FROM protocol_events + GROUP BY run_id + ) totals + JOIN run_attempts attempts ON attempts.run_id = totals.run_id + JOIN protocol_events last + ON last.run_id = totals.run_id + AND last.sequence_number = totals.last_sequence_number + LEFT JOIN leases active_lease ON active_lease.run_id = attempts.run_id + LEFT JOIN worktrees retained_worktree ON retained_worktree.issue_id = attempts.issue_id + LEFT JOIN review_handoffs review_handoff ON review_handoff.issue_id = attempts.issue_id + LEFT JOIN review_orchestrations review_orchestration + ON review_orchestration.issue_id = attempts.issue_id + WHERE attempts.status IN ('succeeded', 'failed', 'interrupted', 'terminated') + AND totals.last_created_at_unix < ?1 + AND active_lease.run_id IS NULL + AND retained_worktree.issue_id IS NULL + AND review_handoff.issue_id IS NULL + AND review_orchestration.issue_id IS NULL + ORDER BY totals.last_created_at_unix ASC, attempts.run_id ASC", + )?; + let rows = statement.query_map(rusqlite::params![cutoff_unix], |row| { + Ok(RuntimeProtocolCandidate { + run_id: row.get(0)?, + issue_id: row.get(1)?, + status: row.get(2)?, + event_count: row.get::<_, i64>(3).map(|value| value.max(0) as u64)?, + last_sequence_number: row.get(4)?, + last_event_type: row.get(5)?, + last_event_at: row.get(6)?, + last_event_at_unix: row.get(7)?, + }) + })?; + let mut candidates = Vec::new(); + + for row in rows { + candidates.push(row?); + } + + Ok(candidates) +} + +fn protected_protocol_run_count(connection: &Connection) -> Result { + let count = connection.query_row( + "SELECT COUNT(DISTINCT attempts.run_id) + FROM run_attempts attempts + JOIN protocol_events events ON events.run_id = attempts.run_id + LEFT JOIN leases active_lease ON active_lease.run_id = attempts.run_id + LEFT JOIN worktrees retained_worktree ON retained_worktree.issue_id = attempts.issue_id + LEFT JOIN review_handoffs review_handoff ON review_handoff.issue_id = attempts.issue_id + LEFT JOIN review_orchestrations review_orchestration + ON review_orchestration.issue_id = attempts.issue_id + WHERE active_lease.run_id IS NOT NULL + OR retained_worktree.issue_id IS NOT NULL + OR review_handoff.issue_id IS NOT NULL + OR review_orchestration.issue_id IS NOT NULL + OR attempts.status NOT IN ('succeeded', 'failed', 'interrupted', 'terminated')", + [], + |row| row.get::<_, i64>(0), + )?; + + Ok(count.max(0) as usize) +} + +fn compact_protocol_events( + connection: &mut Connection, + candidates: &[RuntimeProtocolCandidate], + generated_at: OffsetDateTime, +) -> Result<()> { + let generated_at_text = generated_at.format(&Rfc3339)?; + let generated_at_unix = generated_at.unix_timestamp(); + let transaction = connection.transaction()?; + + for candidate in candidates { + transaction.execute( + "INSERT OR REPLACE INTO protocol_event_summaries ( + run_id, event_count, last_sequence_number, last_event_type, last_event_at, + last_event_at_unix, compacted_at, compacted_at_unix + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)", + rusqlite::params![ + &candidate.run_id, + i64::try_from(candidate.event_count).map_err(|_error| { + eyre::eyre!( + "Protocol event count for run `{}` overflowed i64.", + candidate.run_id + ) + })?, + candidate.last_sequence_number, + candidate.last_event_type.as_deref(), + candidate.last_event_at.as_deref(), + candidate.last_event_at_unix, + &generated_at_text, + generated_at_unix, + ], + )?; + transaction.execute( + "DELETE FROM protocol_events WHERE run_id = ?1", + rusqlite::params![&candidate.run_id], + )?; + } + + transaction.commit()?; + + Ok(()) +} + +fn copy_truncate(path: &Path, rotated_path: &Path) -> Result<()> { + if let Some(parent) = rotated_path.parent() { + fs::create_dir_all(parent)?; + } + + fs::copy(path, rotated_path)?; + OpenOptions::new().write(true).truncate(true).open(path)?; + + Ok(()) +} + +fn rotated_path(path: &Path, generated_at: OffsetDateTime) -> Result { + let parent = path.parent().ok_or_else(|| { + eyre::eyre!("Maintenance target `{}` has no parent directory.", path.display()) + })?; + let file_name = path.file_name().and_then(|name| name.to_str()).ok_or_else(|| { + eyre::eyre!("Maintenance target `{}` has no UTF-8 file name.", path.display()) + })?; + let Some((prefix, suffix)) = file_name.rsplit_once('.') else { + return Ok(parent.join(format!("{file_name}.{}", generated_at.unix_timestamp()))); + }; + let candidate = parent.join(format!("{prefix}.{}.{suffix}", generated_at.unix_timestamp())); + + next_available_path(candidate) +} + +fn next_available_path(path: PathBuf) -> Result { + if !path.exists() { + return Ok(path); + } + + let parent = path.parent().ok_or_else(|| { + eyre::eyre!("Maintenance target `{}` has no parent directory.", path.display()) + })?; + let file_name = path.file_name().and_then(|name| name.to_str()).ok_or_else(|| { + eyre::eyre!("Maintenance target `{}` has no UTF-8 file name.", path.display()) + })?; + + for index in 1..=999 { + let candidate = parent.join(format!("{file_name}.{index}")); + + if !candidate.exists() { + return Ok(candidate); + } + } + + eyre::bail!("Could not allocate a unique maintenance rotation path for `{}`.", path.display()); +} + +fn file_is_older_than(metadata: &Metadata, system_now: SystemTime, retention: Duration) -> bool { + metadata + .modified() + .ok() + .and_then(|modified| system_now.duration_since(modified).ok()) + .is_some_and(|age| age > retention) +} + +fn collect_backup_candidates( + root: &Path, + groups: &mut BTreeMap>, +) -> Result<()> { + for entry in fs::read_dir(root)? { + let entry = entry?; + let path = entry.path(); + let file_type = entry.file_type()?; + + if file_type.is_dir() { + collect_backup_candidates(&path, groups)?; + + continue; + } + if !file_type.is_file() { + continue; + } + + let Some(file_name) = path.file_name().and_then(|name| name.to_str()) else { + continue; + }; + let Some((backup_prefix, _suffix)) = file_name.split_once(".bak-") else { + continue; + }; + let metadata = entry.metadata()?; + let group_key = path + .parent() + .map(|parent| parent.join(backup_prefix).display().to_string()) + .unwrap_or_else(|| backup_prefix.to_owned()); + + groups.entry(group_key).or_default().push(BackupCandidate { + path, + bytes: metadata.len(), + modified: metadata.modified().unwrap_or(SystemTime::UNIX_EPOCH), + }); + } + + Ok(()) +} + +fn print_prune_report(report: &MaintenanceReport) -> Result<()> { + let mut stdout = io::stdout().lock(); + + writeln!(stdout, "Decodex maintenance prune ({}, {})", report.mode, report.scope)?; + writeln!( + stdout, + "logs: rotate {}/{} files ({} bytes), delete {}/{} files ({} bytes)", + report.logs.rotated_files, + report.logs.rotate_candidates, + report.logs.rotate_bytes, + report.logs.deleted_files, + report.logs.delete_candidates, + report.logs.delete_bytes + )?; + writeln!( + stdout, + "agent-evidence: rotate {}/{} streams ({} bytes), delete {}/{} files ({} bytes)", + report.agent_evidence.rotated_files, + report.agent_evidence.rotate_candidates, + report.agent_evidence.rotate_bytes, + report.agent_evidence.deleted_files, + report.agent_evidence.delete_candidates, + report.agent_evidence.delete_bytes + )?; + writeln!( + stdout, + "backups: delete {}/{} files ({} bytes)", + report.backups.deleted_files, report.backups.delete_candidates, report.backups.delete_bytes + )?; + writeln!( + stdout, + "runtime: compact {}/{} terminal runs ({} protocol events), protected runs {}", + report.runtime.compacted_runs, + report.runtime.protocol_run_candidates, + report.runtime.protocol_event_candidates, + report.runtime.protected_run_count + )?; + + match &report.wal_checkpoint { + Some(checkpoint) => writeln!( + stdout, + "wal: {} checkpoint busy={} log_frames={} checkpointed_frames={}", + checkpoint.mode, checkpoint.busy, checkpoint.log_frames, checkpoint.checkpointed_frames + )?, + None => writeln!(stdout, "wal: skipped")?, + } + + Ok(()) +} + +#[cfg(test)] +mod tests { + use std::fs; + + use rusqlite::OptionalExtension as _; + use tempfile::TempDir; + + use crate::{ + maintenance::{ + self, Connection, MaintenanceMode, MaintenancePolicy, MaintenancePruneRequest, + MaintenanceScope, OffsetDateTime, + }, + test_support::TestEnvVarGuard, + }; + + const TEST_RUNTIME_SCHEMA: &str = "PRAGMA journal_mode = WAL; + CREATE TABLE projects ( + service_id TEXT PRIMARY KEY NOT NULL, + config_path TEXT NOT NULL, + repo_root TEXT NOT NULL, + worktree_root TEXT NOT NULL, + workflow_path TEXT NOT NULL, + tracker_api_key_env_var TEXT NOT NULL, + github_token_env_var TEXT NOT NULL, + enabled INTEGER NOT NULL, + config_fingerprint TEXT NOT NULL, + updated_at TEXT NOT NULL, + updated_at_unix INTEGER NOT NULL + ); + CREATE TABLE leases ( + issue_id TEXT PRIMARY KEY NOT NULL, + project_id TEXT NOT NULL, + run_id TEXT NOT NULL, + issue_state TEXT NOT NULL + ); + CREATE TABLE run_attempts ( + run_id TEXT PRIMARY KEY NOT NULL, + project_id TEXT, + issue_id TEXT NOT NULL, + attempt_number INTEGER NOT NULL, + status TEXT NOT NULL, + thread_id TEXT, + turn_id TEXT, + updated_at TEXT NOT NULL, + updated_at_unix INTEGER NOT NULL + ); + CREATE TABLE protocol_events ( + run_id TEXT NOT NULL, + sequence_number INTEGER NOT NULL, + event_type TEXT NOT NULL, + created_at TEXT NOT NULL, + created_at_unix INTEGER NOT NULL, + PRIMARY KEY (run_id, sequence_number) + ); + CREATE TABLE worktrees ( + issue_id TEXT PRIMARY KEY NOT NULL, + project_id TEXT NOT NULL, + branch_name TEXT NOT NULL, + worktree_path TEXT NOT NULL + ); + CREATE TABLE linear_execution_events ( + idempotency_key TEXT PRIMARY KEY NOT NULL, + service_id TEXT NOT NULL, + issue_id TEXT NOT NULL, + event_type TEXT NOT NULL, + event_timestamp TEXT NOT NULL, + event_unix INTEGER, + payload_json TEXT NOT NULL, + recorded_at TEXT NOT NULL, + recorded_at_unix INTEGER NOT NULL + ); + CREATE TABLE review_handoffs ( + project_id TEXT NOT NULL, + issue_id TEXT NOT NULL, + branch_name TEXT NOT NULL, + run_id TEXT NOT NULL, + attempt_number INTEGER NOT NULL, + pr_url TEXT NOT NULL, + target_base_ref_name TEXT, + pr_head_ref_name TEXT NOT NULL, + pr_head_oid TEXT NOT NULL, + updated_at TEXT NOT NULL, + updated_at_unix INTEGER NOT NULL, + PRIMARY KEY (project_id, issue_id, branch_name) + ); + CREATE TABLE review_orchestrations ( + project_id TEXT NOT NULL, + issue_id TEXT NOT NULL, + branch_name TEXT NOT NULL, + run_id TEXT NOT NULL, + attempt_number INTEGER NOT NULL, + pr_url TEXT NOT NULL, + head_sha TEXT NOT NULL, + phase TEXT NOT NULL, + request_comment_database_id INTEGER, + request_created_at_unix_epoch INTEGER, + request_description_thumbs_up_count INTEGER, + request_retry_count INTEGER NOT NULL, + external_round_count INTEGER NOT NULL, + auto_merge_enabled_at_unix_epoch INTEGER, + updated_at TEXT NOT NULL, + updated_at_unix INTEGER NOT NULL, + PRIMARY KEY (project_id, issue_id, branch_name, run_id, attempt_number) + );"; + + #[test] + fn prune_compacts_only_terminal_unowned_protocol_events() { + let temp_dir = TempDir::new().expect("temp dir should create"); + let _home_guard = + TestEnvVarGuard::set("HOME", temp_dir.path().to_str().expect("home should be utf-8")); + let connection = bootstrap_test_runtime_db(&temp_dir); + let now = OffsetDateTime::now_utc(); + let old = now.unix_timestamp() - 30 * 24 * 60 * 60; + let fresh = now.unix_timestamp() - 2 * 24 * 60 * 60; + + insert_attempt(&connection, "old-run", "old-issue", "succeeded"); + insert_event(&connection, "old-run", 1, old); + insert_event(&connection, "old-run", 2, old + 60); + insert_attempt(&connection, "active-run", "active-issue", "running"); + insert_event(&connection, "active-run", 1, old); + + connection + .execute( + "INSERT INTO leases (issue_id, project_id, run_id, issue_state) + VALUES ('active-issue', 'decodex', 'active-run', 'In Progress')", + [], + ) + .expect("active lease should insert"); + + insert_attempt(&connection, "retained-run", "retained-issue", "failed"); + insert_event(&connection, "retained-run", 1, old); + + connection + .execute( + "INSERT INTO worktrees (issue_id, project_id, branch_name, worktree_path) + VALUES ('retained-issue', 'decodex', 'xy/retained', '/tmp/retained')", + [], + ) + .expect("retained worktree should insert"); + + insert_attempt(&connection, "fresh-run", "fresh-issue", "succeeded"); + insert_event(&connection, "fresh-run", 1, fresh); + + let report = maintenance::run_prune_with_policy( + MaintenancePruneRequest { + mode: MaintenanceMode::Apply, + scope: MaintenanceScope::Full, + json: false, + }, + MaintenancePolicy { protocol_event_retention_days: 14, ..MaintenancePolicy::default() }, + ) + .expect("maintenance should run"); + + assert_eq!(report.runtime.compacted_runs, 1); + assert_eq!(report.runtime.compacted_events, 2); + assert_eq!(protocol_event_count(&connection, "old-run"), 0); + assert_eq!(protocol_summary_event_count(&connection, "old-run"), Some(2)); + assert_eq!(protocol_event_count(&connection, "active-run"), 1); + assert_eq!(protocol_event_count(&connection, "retained-run"), 1); + assert_eq!(protocol_event_count(&connection, "fresh-run"), 1); + } + + #[test] + fn prune_rotates_oversized_logs_and_agent_evidence_events() { + let temp_dir = TempDir::new().expect("temp dir should create"); + let _home_guard = + TestEnvVarGuard::set("HOME", temp_dir.path().to_str().expect("home should be utf-8")); + let log_dir = temp_dir.path().join(".codex/decodex/logs"); + let evidence_dir = temp_dir.path().join(".codex/decodex/agent-evidence/decodex"); + let log_path = log_dir.join("decodex.log"); + let events_path = evidence_dir.join("events.jsonl"); + + fs::create_dir_all(&log_dir).expect("log dir should create"); + fs::create_dir_all(&evidence_dir).expect("evidence dir should create"); + fs::write(&log_path, b"0123456789abcdef").expect("log should write"); + fs::write(&events_path, b"0123456789abcdef").expect("events should write"); + + let report = maintenance::run_prune_with_policy( + MaintenancePruneRequest { + mode: MaintenanceMode::Apply, + scope: MaintenanceScope::AutoSafe, + json: false, + }, + MaintenancePolicy { + log_rotate_bytes: 8, + evidence_rotate_bytes: 8, + ..MaintenancePolicy::default() + }, + ) + .expect("maintenance should run"); + + assert_eq!(report.logs.rotated_files, 1); + assert_eq!(report.agent_evidence.rotated_files, 1); + assert_eq!(fs::metadata(&log_path).expect("log should remain").len(), 0); + assert_eq!(fs::metadata(&events_path).expect("events should remain").len(), 0); + assert_eq!( + fs::read_dir(&log_dir) + .expect("log dir should list") + .filter_map(std::result::Result::ok) + .filter(|entry| entry.path() != log_path) + .count(), + 1 + ); + assert_eq!( + fs::read_dir(&evidence_dir) + .expect("evidence dir should list") + .filter_map(std::result::Result::ok) + .filter(|entry| entry.path() != events_path) + .count(), + 1 + ); + } + + fn insert_attempt(connection: &Connection, run_id: &str, issue_id: &str, status: &str) { + connection + .execute( + "INSERT INTO run_attempts ( + run_id, project_id, issue_id, attempt_number, status, updated_at, updated_at_unix + ) VALUES (?1, 'decodex', ?2, 1, ?3, '2026-05-01T00:00:00Z', 0)", + rusqlite::params![run_id, issue_id, status], + ) + .expect("attempt should insert"); + } + + fn bootstrap_test_runtime_db(temp_dir: &TempDir) -> Connection { + let decodex_home = temp_dir.path().join(".codex/decodex"); + + fs::create_dir_all(&decodex_home).expect("decodex home should create"); + + let database_path = decodex_home.join("runtime.sqlite3"); + let connection = Connection::open(&database_path).expect("runtime DB should open"); + + connection.execute_batch(TEST_RUNTIME_SCHEMA).expect("schema should bootstrap"); + + maintenance::ensure_protocol_event_summary_table(&connection) + .expect("summary table should create"); + + connection + } + + fn insert_event(connection: &Connection, run_id: &str, sequence_number: i64, created_at: i64) { + connection + .execute( + "INSERT INTO protocol_events ( + run_id, sequence_number, event_type, created_at, created_at_unix + ) VALUES (?1, ?2, 'event', '2026-05-01T00:00:00Z', ?3)", + rusqlite::params![run_id, sequence_number, created_at], + ) + .expect("event should insert"); + } + + fn protocol_event_count(connection: &Connection, run_id: &str) -> i64 { + connection + .query_row( + "SELECT COUNT(*) FROM protocol_events WHERE run_id = ?1", + rusqlite::params![run_id], + |row| row.get(0), + ) + .expect("event count should read") + } + + fn protocol_summary_event_count(connection: &Connection, run_id: &str) -> Option { + connection + .query_row( + "SELECT event_count FROM protocol_event_summaries WHERE run_id = ?1", + rusqlite::params![run_id], + |row| row.get(0), + ) + .optional() + .expect("summary should read") + } +} diff --git a/apps/decodex/src/orchestrator.rs b/apps/decodex/src/orchestrator.rs index 412803725..772062241 100644 --- a/apps/decodex/src/orchestrator.rs +++ b/apps/decodex/src/orchestrator.rs @@ -25,7 +25,7 @@ use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; use time::{OffsetDateTime, format_description::well_known::Rfc3339}; -use crate::{agent, default_branch_sync, git_credentials, state}; +use crate::{agent, default_branch_sync, git_credentials, maintenance, state}; #[rustfmt::skip] use crate::{agent::{ACTIVE_RUN_IDLE_TIMEOUT, AppServerCapabilityPreflightFailure, AppServerDynamicToolFailure, AppServerHomePreflightFailure, AppServerProcessEnv, AppServerRunRequest, AppServerRunResult, AppServerTransportFailure, AppServerTurnFailure, ISSUE_DELIVERY_CLOSEOUT_COMPLETE_TOOL_NAME, ISSUE_LABEL_ADD_TOOL_NAME, ISSUE_PROGRESS_CHECKPOINT_TOOL_NAME, ISSUE_REVIEW_CHECKPOINT_TOOL_NAME, ISSUE_REVIEW_HANDOFF_TOOL_NAME, ISSUE_REVIEW_REPAIR_COMPLETE_TOOL_NAME, ISSUE_TERMINAL_FINALIZE_TOOL_NAME, ISSUE_TRANSITION_TOOL_NAME, DecodexRunContext, DecodexToolBridge, ReviewExecutionMode, ReviewHandoffContext, ReviewHandoffWritebackFailed, ReviewPolicyStopReason, ReviewPolicyStopRequested, RunCompletionDisposition, TrackerToolBridge, TurnContinuationGuard}, config::{InternalReviewMode, ServiceConfig}, git_credentials::GitCredentialSource, github, prelude::{Result, eyre}, state::{ChildAgentActivityBucket, ChildAgentActivitySummary, CodexAccountActivitySummary, ProjectRegistration, ProjectRunStatus, ProtocolActivitySummary, RUN_OPERATION_AGENT_RUN, RUN_OPERATION_APP_SERVER_PREFLIGHT, RUN_OPERATION_GIT_CREDENTIALS, RUN_OPERATION_IDLE, RUN_OPERATION_RECONCILIATION, RUN_OPERATION_REPO_GATE, RUN_OPERATION_REVIEW_WRITEBACK, RUN_OPERATION_WAITING_EXTERNAL, ReviewHandoffMarker, ReviewOrchestrationMarker, RunActivityMarker, RunAttempt, StateStore, WorktreeMapping}, tracker::{IssueTracker, TrackerComment, TrackerIssue, linear::LinearClient, records}, workflow::{WorkflowDocument, WorkflowExecution}, worktree::{WorktreeManager, WorktreeSpec}}; diff --git a/apps/decodex/src/orchestrator/entrypoints.rs b/apps/decodex/src/orchestrator/entrypoints.rs index d0d99ac1a..e3cd3b76a 100644 --- a/apps/decodex/src/orchestrator/entrypoints.rs +++ b/apps/decodex/src/orchestrator/entrypoints.rs @@ -117,6 +117,8 @@ pub(crate) fn run_control_plane(request: ServeRequest<'_>) -> Result<()> { let state_store = Arc::new(runtime::open_runtime_store()?); + run_control_plane_maintenance("startup"); + if request.api_only { let operator_state_endpoint = OperatorStateEndpoint::start(request.listen_address, Arc::clone(&state_store))?; @@ -173,6 +175,7 @@ pub(crate) fn run_control_plane(request: ServeRequest<'_>) -> Result<()> { let global_config_path = runtime::global_config_path()?; let project_config_dir = runtime::project_config_dir()?; let mut project_runtimes: HashMap = HashMap::new(); + let mut next_maintenance_at = Instant::now() + Duration::from_secs(60 * 60); tracing::info!( poll_interval_s = poll_interval.as_secs(), @@ -188,6 +191,13 @@ pub(crate) fn run_control_plane(request: ServeRequest<'_>) -> Result<()> { loop { let tick_started_at = Instant::now(); + + if tick_started_at >= next_maintenance_at { + run_control_plane_maintenance("scheduled"); + + next_maintenance_at = tick_started_at + Duration::from_secs(60 * 60); + } + let snapshot = run_control_plane_tick(&state_store, &mut project_runtimes)?; if let Err(error) = operator_state_endpoint.publish_snapshot(&snapshot) { @@ -333,6 +343,33 @@ pub(crate) fn run_diagnose(request: DiagnoseRequest<'_>) -> Result<()> { Ok(()) } +fn run_control_plane_maintenance(trigger: &'static str) { + match maintenance::run_auto_safe_prune() { + Ok(report) => { + tracing::info!( + trigger = trigger, + log_rotated_files = report.logs.rotated_files, + evidence_rotated_files = report.agent_evidence.rotated_files, + backup_deleted_files = report.backups.deleted_files, + wal_checkpoint_mode = report + .wal_checkpoint + .as_ref() + .map(|checkpoint| checkpoint.mode) + .unwrap_or("skipped"), + "Completed Decodex auto-safe maintenance." + ); + }, + Err(error) => { + let _ = error; + + tracing::warn!( + trigger = trigger, + "Decodex auto-safe maintenance failed; sensitive runtime details were withheld from control-plane logs." + ); + }, + } +} + fn runtime_recovery_warning(prefix: &str, error: &Report) -> String { format!("{prefix}:{}", runtime_recovery_error_class(error)) } diff --git a/apps/decodex/src/state/internal.rs b/apps/decodex/src/state/internal.rs index 4fd99c4c2..d09265f82 100644 --- a/apps/decodex/src/state/internal.rs +++ b/apps/decodex/src/state/internal.rs @@ -249,6 +249,16 @@ CREATE TABLE IF NOT EXISTS protocol_events ( created_at_unix INTEGER NOT NULL, PRIMARY KEY (run_id, sequence_number) ); +CREATE TABLE IF NOT EXISTS protocol_event_summaries ( + run_id TEXT PRIMARY KEY NOT NULL, + event_count INTEGER NOT NULL, + last_sequence_number INTEGER, + last_event_type TEXT, + last_event_at TEXT, + last_event_at_unix INTEGER, + compacted_at TEXT NOT NULL, + compacted_at_unix INTEGER NOT NULL +); CREATE TABLE IF NOT EXISTS worktrees ( issue_id TEXT PRIMARY KEY NOT NULL, project_id TEXT NOT NULL, @@ -306,7 +316,7 @@ CREATE TABLE IF NOT EXISTS schema_meta ( value TEXT NOT NULL ); INSERT INTO schema_meta (key, value) -VALUES ('schema_version', '3') +VALUES ('schema_version', '4') ON CONFLICT(key) DO UPDATE SET value = excluded.value; "#, )?; @@ -595,6 +605,8 @@ ON CONFLICT(key) DO UPDATE SET value = excluded.value; } fn load_protocol_event_summaries(&self, state: &mut StateData) -> Result<()> { + self.load_compacted_protocol_event_summaries(state)?; + let mut statement = self.connection.prepare( "SELECT totals.run_id, totals.event_count, totals.last_sequence_number, \ last.event_type, last.created_at, last.created_at_unix \ @@ -629,6 +641,33 @@ ON CONFLICT(key) DO UPDATE SET value = excluded.value; Ok(()) } + fn load_compacted_protocol_event_summaries(&self, state: &mut StateData) -> Result<()> { + let mut statement = self.connection.prepare( + "SELECT run_id, event_count, last_sequence_number, last_event_type, last_event_at, \ + last_event_at_unix FROM protocol_event_summaries ORDER BY run_id", + )?; + let rows = statement.query_map([], |row| { + Ok(( + row.get::<_, String>(0)?, + ProtocolEventSummaryRecord { + event_count: row.get(1)?, + last_sequence_number: row.get(2)?, + last_event_type: row.get(3)?, + last_event_at: row.get(4)?, + last_event_at_unix: row.get(5)?, + }, + )) + })?; + + for row in rows { + let (run_id, summary) = row?; + + state.event_summaries.insert(run_id, summary); + } + + Ok(()) + } + fn load_worktrees(&self, state: &mut StateData) -> Result<()> { let mut statement = self .connection diff --git a/docs/spec/agent-evidence.md b/docs/spec/agent-evidence.md index bd06b4024..4086f787c 100644 --- a/docs/spec/agent-evidence.md +++ b/docs/spec/agent-evidence.md @@ -132,6 +132,13 @@ blocker count, run capsule count, warning count, and connector backoff count. Th stream exists so a future agent can identify when evidence changed without diffing all JSON files. +The event stream is append-only between maintenance windows, but it is not permanent +runtime authority. `decodex maintenance prune --apply` and the auto-safe maintenance +subset in `decodex serve` may copy-truncate an oversized `events.jsonl` into a rotated +local sibling file and later delete old rotated event files. The current +`handoff-index.json`, blocker snapshots, and run capsules remain the compact diagnostic +surface for repair agents. + ## Privacy Boundary Agent evidence may include local filesystem paths, issue identifiers, PR URLs, diff --git a/docs/spec/runtime.md b/docs/spec/runtime.md index a74fba469..75506f7a1 100644 --- a/docs/spec/runtime.md +++ b/docs/spec/runtime.md @@ -355,7 +355,21 @@ After a process restart, recent-run history, active lease ownership, retained po ## Retention and cleanup - Lease and session mappings: remove when the run closes. -- Attempt and event journals: retain in the runtime database until explicit cleanup policy removes old rows. +- Attempt records, terminal outcome, and locally cached Linear execution ledger links + remain runtime history. Raw protocol event rows for terminal runs may be compacted by + `decodex maintenance prune --apply` once the latest event is at least 14 days old, + but only after Decodex writes the compact run summary and confirms that no active + lease, retained worktree, review handoff, review orchestration, or cleanup blocker + still owns that run or issue. +- `decodex maintenance prune --dry-run` is the read-only retention path for inspecting + local cleanup candidates without applying retention changes. The `--apply` mode owns + state-aware protocol-event + compaction, old backup pruning, local log and agent-evidence event-stream rotation, + and SQLite WAL checkpointing. Operators must not delete `runtime.sqlite3-wal` + directly. +- `decodex serve` runs the auto-safe maintenance subset at startup and periodically + while polling. That subset may rotate oversized local files, prune old backups, and + run a passive WAL checkpoint, but it must not compact runtime protocol events. - Worktrees: retain while the issue is non-terminal, and also retain terminal owned lanes while authoritative post-merge closeout or deterministic cleanup is still incomplete. - Terminal issue cleanup: once the issue reaches a terminal tracker state and no authoritative post-merge tail remains pending, remove the worktree during reconciliation or startup cleanup. - If an issue becomes non-terminal but no longer eligible while `decodex` is still preparing the lane, keep the worktree and skip execution for that pass.