diff --git a/README.md b/README.md index 5e9f226..9eb66e1 100644 --- a/README.md +++ b/README.md @@ -73,10 +73,12 @@ cargo run -- attachment fetch m-1:1.2 --json cargo run -- attachment export m-1:1.2 --json cargo run -- attachment export m-1:1.2 --to ./exports/statement.pdf --json cargo run -- automation rules validate --json +cargo run -- automation rollout --limit 10 --json cargo run -- automation run --json cargo run -- automation run --rule archive-newsletters --limit 25 --json cargo run -- automation show 42 --json cargo run -- automation apply 42 --execute --json +cargo run -- automation prune --older-than-days 30 --json cargo run -- workflow list --json cargo run -- workflow show thread-123 --json cargo run -- triage set thread-123 --bucket urgent --note "reply today" --json diff --git a/docs/operations/automation-rules-and-bulk-actions.md b/docs/operations/automation-rules-and-bulk-actions.md index 7ccf3a0..1dd4218 100644 --- a/docs/operations/automation-rules-and-bulk-actions.md +++ b/docs/operations/automation-rules-and-bulk-actions.md @@ -8,7 +8,9 @@ The current automation slice owns: - a local typed rules file at `.mailroom/automation.toml` - rule validation and preview snapshots +- read-only rollout checks for first-wave micro-batch readiness - persisted automation runs and append-only run events +- stale local snapshot pruning after dry-run review - reviewed thread-first bulk actions for archive, label, and trash - unsubscribe assistance through visible list headers in the review output @@ -105,6 +107,13 @@ Validate the active file: cargo run -- automation rules validate --json ``` +Check first-wave rollout readiness without saving a run: + +```bash +cargo run -- automation rollout --limit 10 --json +cargo run -- automation rollout --rule archive-newsletters --limit 10 --json +``` + Create a review snapshot across all enabled rules: ```bash @@ -138,6 +147,17 @@ cargo run -- automation apply 42 --execute --json Without `--execute`, `automation apply` returns a validation-style error and does not mutate Gmail. +Prune stale local review snapshots after inspecting the dry-run counts: + +```bash +cargo run -- automation prune --older-than-days 30 --json +cargo run -- automation prune --older-than-days 30 --status previewed --execute --json +cargo run -- automation prune --older-than-days 90 --status applied --status apply-failed --json +``` + +`automation prune` deletes only local SQLite automation snapshot rows. It never +mutates Gmail and it never targets in-progress `applying` runs. + ## Output model `--json` uses the standard Mailroom envelope. @@ -180,9 +200,20 @@ The JSON payload also includes header-derived unsubscribe hints: - `precedence_header` - `auto_submitted_header` +`automation rollout` returns the same standard JSON envelope with: + +- `verification` readiness from `audit verification` +- optional `rules` validation detail +- preview-only candidate summaries +- blockers, warnings, next steps, and exact follow-up commands + +Missing or invalid rules are reported as rollout blockers rather than as a +persisted run. + ## Safety model - `automation run` is preview-only and only writes a local snapshot +- `automation rollout` is read-only and writes no automation snapshot - `automation apply` mutates Gmail only when `--execute` is present - `automation apply --execute` requires working Gmail auth up front and aborts before persisting apply results if credentials are missing, expired, or point @@ -191,6 +222,8 @@ The JSON payload also includes header-derived unsubscribe hints: - thread mutations reuse the same Gmail thread cleanup path as the manual cleanup commands - successful apply runs trigger a best-effort mailbox resync afterward +- `automation prune` is dry-run by default and deletes only local snapshot + history when `--execute` is present If a run applies zero candidates, Mailroom records the run transition locally but does not issue Gmail mutations. @@ -215,10 +248,13 @@ This slice adds: 1. Sync local mailbox metadata. 2. Validate the active rules file. -3. Run a preview snapshot. -4. Inspect the saved run by ID. -5. Apply only the reviewed run with `--execute`. -6. Re-run sync or `doctor` if you want to inspect the reconciled local state. +3. Run `automation rollout --limit 10` to check readiness and preview matching + candidates without saving a run. +4. Run a preview snapshot. +5. Inspect the saved run by ID. +6. Apply only the reviewed run with `--execute`. +7. Re-run sync or `doctor` if you want to inspect the reconciled local state. +8. Periodically prune stale preview snapshots after a dry-run count review. For the real-mailbox hardening sequence, do not jump straight from rule editing to `automation apply --execute`. Follow diff --git a/docs/operations/verification-and-hardening.md b/docs/operations/verification-and-hardening.md index 8fc4e0f..fbc2a65 100644 --- a/docs/operations/verification-and-hardening.md +++ b/docs/operations/verification-and-hardening.md @@ -162,12 +162,16 @@ cp config/automation.example.toml .mailroom/automation.toml $EDITOR .mailroom/automation.toml cargo run -- automation rules validate --json +cargo run -- automation rollout --limit 10 --json cargo run -- automation run --limit 10 --json cargo run -- automation show --json ``` Inspect: +- `blockers` +- `warnings` +- preview-only `candidates` - `selected_rule_ids` - `candidate_count` - `candidate_details` @@ -181,6 +185,7 @@ If the preview is surprising, fix the rules before any live apply. Start with micro-batches only: ```bash +cargo run -- automation rollout --rule --limit 10 --json cargo run -- automation run --rule --limit 10 --json cargo run -- automation show --json cargo run -- automation apply --execute --json @@ -230,7 +235,9 @@ cargo run -- audit labels --json cargo run -- audit verification --json cargo run -- sync run --profile deep-audit --json cargo run -- automation rules validate --json +cargo run -- automation rollout --limit 10 --json cargo run -- automation run --limit 10 --json cargo run -- automation show --json cargo run -- automation apply --execute --json +cargo run -- automation prune --older-than-days 30 --json ``` diff --git a/src/automation/mod.rs b/src/automation/mod.rs index 3ff328e..9d04114 100644 --- a/src/automation/mod.rs +++ b/src/automation/mod.rs @@ -3,6 +3,9 @@ mod output; mod rules; mod service; -pub use model::{AutomationRunRequest, DEFAULT_AUTOMATION_RUN_LIMIT}; +pub use model::{ + AutomationPruneRequest, AutomationPruneStatus, AutomationRolloutRequest, AutomationRunRequest, + DEFAULT_AUTOMATION_ROLLOUT_LIMIT, DEFAULT_AUTOMATION_RUN_LIMIT, +}; pub(crate) use service::AutomationServiceError; -pub use service::{apply_run, run_preview, show_run, validate_rules}; +pub use service::{apply_run, prune_runs, rollout, run_preview, show_run, validate_rules}; diff --git a/src/automation/model.rs b/src/automation/model.rs index f4e824c..ab81c34 100644 --- a/src/automation/model.rs +++ b/src/automation/model.rs @@ -1,9 +1,11 @@ +use crate::audit::VerificationAuditReport; use crate::mailbox::SyncRunReport; use crate::store::automation::{AutomationActionKind, AutomationRunDetail}; use serde::{Deserialize, Serialize}; use std::path::PathBuf; pub const DEFAULT_AUTOMATION_RUN_LIMIT: usize = 250; +pub const DEFAULT_AUTOMATION_ROLLOUT_LIMIT: usize = 10; #[derive(Debug, Clone)] pub struct AutomationRunRequest { @@ -11,6 +13,37 @@ pub struct AutomationRunRequest { pub limit: usize, } +#[derive(Debug, Clone)] +pub struct AutomationRolloutRequest { + pub rule_ids: Vec, + pub limit: usize, +} + +#[derive(Debug, Clone)] +pub struct AutomationPruneRequest { + pub older_than_days: u32, + pub statuses: Vec, + pub execute: bool, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum AutomationPruneStatus { + Previewed, + Applied, + ApplyFailed, +} + +impl AutomationPruneStatus { + pub fn as_str(self) -> &'static str { + match self { + Self::Previewed => "previewed", + Self::Applied => "applied", + Self::ApplyFailed => "apply_failed", + } + } +} + #[derive(Debug, Clone, Serialize)] pub struct AutomationRulesValidateReport { pub path: PathBuf, @@ -34,6 +67,34 @@ pub struct AutomationRunPreviewReport { pub detail: AutomationRunDetail, } +#[derive(Debug, Clone, Serialize)] +pub struct AutomationRolloutReport { + pub verification: VerificationAuditReport, + pub rules: Option, + pub selected_rule_ids: Vec, + pub selected_rule_count: usize, + pub candidate_count: usize, + pub candidates: Vec, + pub blocked_rule_ids: Vec, + pub blockers: Vec, + pub warnings: Vec, + pub next_steps: Vec, + pub command_plan: Vec, +} + +#[derive(Debug, Clone, Serialize)] +pub struct AutomationRolloutCandidateSummary { + pub rule_id: String, + pub thread_id: String, + pub message_id: String, + pub action_kind: String, + pub subject: String, + pub from_address: Option, + pub label_names: Vec, + pub has_list_unsubscribe: bool, + pub matched_predicates: Vec, +} + #[derive(Debug, Clone, Serialize)] pub struct AutomationShowReport { pub detail: AutomationRunDetail, @@ -48,6 +109,21 @@ pub struct AutomationApplyReport { pub sync_report: Option, } +#[derive(Debug, Clone, Serialize)] +pub struct AutomationPruneReport { + pub account_id: String, + pub execute: bool, + pub older_than_days: u32, + pub cutoff_epoch_s: i64, + pub statuses: Vec, + pub matched_run_count: i64, + pub matched_candidate_count: i64, + pub matched_event_count: i64, + pub deleted_run_count: i64, + pub warnings: Vec, + pub next_steps: Vec, +} + #[derive(Debug, Clone, Serialize, Deserialize)] pub struct AutomationRuleSet { #[serde(default)] diff --git a/src/automation/output.rs b/src/automation/output.rs index 44539eb..d31d549 100644 --- a/src/automation/output.rs +++ b/src/automation/output.rs @@ -1,6 +1,6 @@ use super::model::{ - AutomationApplyReport, AutomationRulesValidateReport, AutomationRunPreviewReport, - AutomationShowReport, + AutomationApplyReport, AutomationPruneReport, AutomationRolloutReport, + AutomationRulesValidateReport, AutomationRunPreviewReport, AutomationShowReport, }; use anyhow::Result; use std::io::{self, Write}; @@ -70,6 +70,79 @@ impl AutomationRunPreviewReport { } } +impl AutomationRolloutReport { + pub fn print(&self, json: bool) -> Result<()> { + route_output_to_stdout(json, |json, stdout| self.write(json, stdout)) + } + + fn render_plain(&self) -> String { + let mut lines = vec![ + String::from("operation=rollout"), + format!( + "account_id={}", + sanitize(self.verification.account_id.as_deref().unwrap_or("")) + ), + format!("authenticated={}", self.verification.authenticated), + format!("rules_file_exists={}", self.verification.rules_file_exists), + format!("selected_rule_count={}", self.selected_rule_count), + format!( + "selected_rule_ids={}", + sanitize(&self.selected_rule_ids.join(",")) + ), + format!("candidate_count={}", self.candidate_count), + format!("blocked_rule_count={}", self.blocked_rule_ids.len()), + ]; + if !self.blocked_rule_ids.is_empty() { + lines.push(format!( + "blocked_rule_ids={}", + sanitize(&self.blocked_rule_ids.join(",")) + )); + } + if !self.candidates.is_empty() { + lines.push(String::from("results_format=tsv")); + lines.push(String::from( + "rule_id\tthread_id\tmessage_id\taction\thas_unsubscribe\tfrom_address\tsubject\tlabels\tmatched_predicates", + )); + lines.extend(self.candidates.iter().map(|candidate| { + format!( + "{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}", + sanitize(&candidate.rule_id), + sanitize(&candidate.thread_id), + sanitize(&candidate.message_id), + sanitize(&candidate.action_kind), + candidate.has_list_unsubscribe, + sanitize(candidate.from_address.as_deref().unwrap_or_default()), + sanitize(&candidate.subject), + sanitize(&candidate.label_names.join(" | ")), + sanitize(&candidate.matched_predicates.join(", ")), + ) + })); + } + for blocker in &self.blockers { + lines.push(format!("blocker={}", sanitize(blocker))); + } + for warning in &self.warnings { + lines.push(format!("warning={}", sanitize(warning))); + } + for next_step in &self.next_steps { + lines.push(format!("next_step={}", sanitize(next_step))); + } + for command in &self.command_plan { + lines.push(format!("command={}", sanitize(command))); + } + lines.join("\n") + "\n" + } + + fn write(&self, json: bool, writer: &mut W) -> Result<()> { + if json { + crate::cli_output::write_json_success(writer, self)?; + } else { + writer.write_all(self.render_plain().as_bytes())?; + } + Ok(()) + } +} + impl AutomationShowReport { pub fn print(&self, json: bool) -> Result<()> { route_output_to_stdout(json, |json, stdout| self.write(json, stdout)) @@ -89,6 +162,43 @@ impl AutomationShowReport { } } +impl AutomationPruneReport { + pub fn print(&self, json: bool) -> Result<()> { + route_output_to_stdout(json, |json, stdout| self.write(json, stdout)) + } + + fn render_plain(&self) -> String { + let mut lines = vec![ + String::from("operation=prune"), + format!("account_id={}", sanitize(&self.account_id)), + format!("execute={}", self.execute), + format!("older_than_days={}", self.older_than_days), + format!("cutoff_epoch_s={}", self.cutoff_epoch_s), + format!("statuses={}", sanitize(&self.statuses.join(","))), + format!("matched_run_count={}", self.matched_run_count), + format!("matched_candidate_count={}", self.matched_candidate_count), + format!("matched_event_count={}", self.matched_event_count), + format!("deleted_run_count={}", self.deleted_run_count), + ]; + for warning in &self.warnings { + lines.push(format!("warning={}", sanitize(warning))); + } + for next_step in &self.next_steps { + lines.push(format!("next_step={}", sanitize(next_step))); + } + lines.join("\n") + "\n" + } + + fn write(&self, json: bool, writer: &mut W) -> Result<()> { + if json { + crate::cli_output::write_json_success(writer, self)?; + } else { + writer.write_all(self.render_plain().as_bytes())?; + } + Ok(()) + } +} + impl AutomationApplyReport { pub fn print(&self, json: bool) -> Result<()> { route_output_to_stdout(json, |json, stdout| self.write(json, stdout)) diff --git a/src/automation/service.rs b/src/automation/service.rs index 9c55b9b..547a1fd 100644 --- a/src/automation/service.rs +++ b/src/automation/service.rs @@ -1,5 +1,7 @@ use super::model::{ - AutomationApplyReport, AutomationRule, AutomationRuleAction, AutomationRulesValidateReport, + AutomationApplyReport, AutomationPruneReport, AutomationPruneRequest, AutomationPruneStatus, + AutomationRolloutCandidateSummary, AutomationRolloutReport, AutomationRolloutRequest, + AutomationRule, AutomationRuleAction, AutomationRulesValidateReport, AutomationRunPreviewReport, AutomationRunRequest, AutomationShowReport, }; use super::rules::{ResolvedAutomationRules, resolve_rule_selection, validate_rule_file}; @@ -11,6 +13,7 @@ use crate::store::automation::{ AutomationApplyStatus, AutomationMatchReason, AutomationRunCandidateRecord, AutomationRunStatus, AutomationThreadCandidate, CandidateApplyResultInput, CreateAutomationRunInput, FinalizeAutomationRunInput, NewAutomationRunCandidate, + PruneAutomationRunsInput, }; use crate::time::current_epoch_seconds; use anyhow::Result; @@ -25,6 +28,7 @@ use tokio::sync::Semaphore; use tokio::task::{JoinSet, spawn_blocking}; const AUTOMATION_APPLY_CONCURRENCY: usize = 4; +const SECONDS_PER_DAY: i64 = 86_400; #[derive(Debug, Error)] pub enum AutomationServiceError { @@ -34,6 +38,10 @@ pub enum AutomationServiceError { NoActiveAccount, #[error("automation run limit must be greater than zero")] InvalidLimit, + #[error("automation rollout limit must be greater than zero")] + InvalidRolloutLimit, + #[error("automation prune --older-than-days must be greater than zero")] + InvalidPruneWindow, #[error("re-run with --execute to apply automation changes")] ExecuteRequired, #[error( @@ -150,12 +158,155 @@ pub async fn run_preview( Ok(AutomationRunPreviewReport { detail }) } +pub async fn rollout( + config_report: &ConfigReport, + request: AutomationRolloutRequest, +) -> Result { + if request.limit == 0 { + return Err(AutomationServiceError::InvalidRolloutLimit.into()); + } + + ensure_runtime_dirs_task(configured_paths(config_report)?).await?; + init_store_task(config_report).await?; + + let verification = verification_audit_task(config_report).await?; + let mut blockers = Vec::new(); + let mut warnings = verification.warnings.clone(); + let mut selected_rule_ids = Vec::new(); + let mut blocked_rule_ids = Vec::new(); + let mut candidates = Vec::new(); + + let rules = match validate_rule_file(config_report).await { + Ok(report) => Some(report), + Err(error) => { + blockers.push(format!("automation rules are not ready: {error}")); + None + } + }; + + if rules.is_some() { + match resolve_rollout_candidates(config_report, &request).await { + Ok(selection) => { + selected_rule_ids = selection.selected_rule_ids; + blocked_rule_ids = selection.blocked_rule_ids; + candidates = selection.candidates; + } + Err(error) if is_rollout_blocker(&error) => { + blockers.push(error.to_string()); + } + Err(error) => return Err(error), + } + } + + if !blocked_rule_ids.is_empty() { + blockers.push(format!( + "first-wave automation rollout excludes trash rules; remove or disable: {}", + blocked_rule_ids.join(", ") + )); + candidates.clear(); + } + if candidates.is_empty() && blockers.is_empty() { + warnings.push(String::from( + "Selected rules did not match any local thread candidates; inspect the synced cache and rule predicates before applying.", + )); + } + + let command_plan = rollout_command_plan(&selected_rule_ids, request.limit); + let mut next_steps = verification.next_steps.clone(); + if blockers.is_empty() { + next_steps.push(String::from( + "Persist a review snapshot with the matching automation run command, inspect it with automation show, then apply only a reviewed micro-batch.", + )); + } else { + next_steps.push(String::from( + "Clear rollout blockers before creating a persistent automation run.", + )); + } + + Ok(AutomationRolloutReport { + verification, + rules, + selected_rule_count: selected_rule_ids.len(), + selected_rule_ids, + candidate_count: candidates.len(), + candidates, + blocked_rule_ids, + blockers, + warnings, + next_steps, + command_plan, + }) +} + pub async fn show_run(config_report: &ConfigReport, run_id: i64) -> Result { init_store_task(config_report).await?; let detail = load_run_detail_task(config_report, run_id).await?; Ok(AutomationShowReport { detail }) } +pub async fn prune_runs( + config_report: &ConfigReport, + request: AutomationPruneRequest, +) -> Result { + if request.older_than_days == 0 { + return Err(AutomationServiceError::InvalidPruneWindow.into()); + } + + ensure_runtime_dirs_task(configured_paths(config_report)?).await?; + init_store_task(config_report).await?; + let account_id = resolve_automation_account_id_task(config_report).await?; + let statuses = normalize_prune_statuses(request.statuses); + let cutoff_epoch_s = current_epoch_seconds()? + .saturating_sub(i64::from(request.older_than_days) * SECONDS_PER_DAY); + let store_report = prune_automation_runs_task( + config_report, + &PruneAutomationRunsInput { + account_id: account_id.clone(), + cutoff_epoch_s, + statuses: statuses + .iter() + .copied() + .map(prune_status_to_run_status) + .collect(), + execute: request.execute, + }, + ) + .await?; + + let mut warnings = Vec::new(); + if !request.execute { + warnings.push(String::from( + "Dry run only; rerun with --execute to delete matched local automation snapshots.", + )); + } + let next_steps = if request.execute { + vec![String::from( + "Run `cargo run -- doctor --json` if you want to inspect updated local automation counts.", + )] + } else { + vec![String::from( + "Rerun the same prune command with --execute after reviewing the matched counts.", + )] + }; + + Ok(AutomationPruneReport { + account_id, + execute: request.execute, + older_than_days: request.older_than_days, + cutoff_epoch_s, + statuses: statuses + .iter() + .map(|status| status.as_str().to_owned()) + .collect(), + matched_run_count: store_report.matched_run_count, + matched_candidate_count: store_report.matched_candidate_count, + matched_event_count: store_report.matched_event_count, + deleted_run_count: store_report.deleted_run_count, + warnings, + next_steps, + }) +} + pub async fn apply_run( config_report: &ConfigReport, run_id: i64, @@ -353,6 +504,162 @@ struct ApplyOutcome { apply_error: Option, } +#[derive(Debug)] +struct RolloutSelection { + selected_rule_ids: Vec, + blocked_rule_ids: Vec, + candidates: Vec, +} + +async fn resolve_rollout_candidates( + config_report: &ConfigReport, + request: &AutomationRolloutRequest, +) -> Result { + let account_id = resolve_automation_account_id_task(config_report).await?; + let resolved_rules = resolve_rule_selection(config_report, &request.rule_ids).await?; + let selected_rule_ids = resolved_rules + .rules + .iter() + .map(|rule| rule.id.clone()) + .collect::>(); + let blocked_rule_ids = resolved_rules + .rules + .iter() + .filter(|rule| rule.action_kind() == AutomationActionKind::Trash) + .map(|rule| rule.id.clone()) + .collect::>(); + if !blocked_rule_ids.is_empty() { + return Ok(RolloutSelection { + selected_rule_ids, + blocked_rule_ids, + candidates: Vec::new(), + }); + } + + let planned_rules = + resolve_rule_actions_task(config_report, &account_id, &resolved_rules).await?; + let thread_candidates = list_latest_thread_candidates_task(config_report, &account_id).await?; + let now_epoch_ms = current_epoch_seconds()?.saturating_mul(1_000); + let candidates = build_run_candidates( + &thread_candidates, + &planned_rules, + now_epoch_ms, + request.limit, + ) + .iter() + .map(rollout_candidate_summary) + .collect(); + + Ok(RolloutSelection { + selected_rule_ids, + blocked_rule_ids, + candidates, + }) +} + +fn is_rollout_blocker(error: &anyhow::Error) -> bool { + matches!( + error.downcast_ref::(), + Some( + AutomationServiceError::NoActiveAccount + | AutomationServiceError::RuleFileMissing { .. } + | AutomationServiceError::RuleFileRead { .. } + | AutomationServiceError::RuleFileParse { .. } + | AutomationServiceError::RuleValidation { .. } + ) + ) +} + +fn rollout_candidate_summary( + candidate: &NewAutomationRunCandidate, +) -> AutomationRolloutCandidateSummary { + AutomationRolloutCandidateSummary { + rule_id: candidate.rule_id.clone(), + thread_id: candidate.thread_id.clone(), + message_id: candidate.message_id.clone(), + action_kind: candidate.action.kind.as_str().to_owned(), + subject: candidate.subject.clone(), + from_address: candidate.from_address.clone(), + label_names: candidate.label_names.clone(), + has_list_unsubscribe: candidate.has_list_unsubscribe, + matched_predicates: match_reason_tokens(&candidate.reason), + } +} + +fn match_reason_tokens(reason: &AutomationMatchReason) -> Vec { + let mut predicates = Vec::new(); + if let Some(from_address) = &reason.from_address { + predicates.push(format!("from={from_address}")); + } + if !reason.subject_terms.is_empty() { + predicates.push(format!("subject~{}", reason.subject_terms.join("|"))); + } + if !reason.label_names.is_empty() { + predicates.push(format!("label_any={}", reason.label_names.join("|"))); + } + if let Some(days) = reason.older_than_days { + predicates.push(format!("older_than_days={days}")); + } + if let Some(has_attachments) = reason.has_attachments { + predicates.push(format!("has_attachments={has_attachments}")); + } + if let Some(has_list_unsubscribe) = reason.has_list_unsubscribe { + predicates.push(format!("has_list_unsubscribe={has_list_unsubscribe}")); + } + if !reason.list_id_terms.is_empty() { + predicates.push(format!("list_id~{}", reason.list_id_terms.join("|"))); + } + if !reason.precedence_values.is_empty() { + predicates.push(format!("precedence={}", reason.precedence_values.join("|"))); + } + predicates +} + +fn rollout_command_plan(selected_rule_ids: &[String], limit: usize) -> Vec { + let selected_rules = selected_rule_ids + .iter() + .map(|rule_id| format!(" --rule {rule_id}")) + .collect::(); + vec![ + String::from("cargo run -- automation rules validate --json"), + format!("cargo run -- automation run{selected_rules} --limit {limit} --json"), + String::from("cargo run -- automation show --json"), + String::from("cargo run -- automation apply --execute --json"), + String::from("cargo run -- audit verification --json"), + ] +} + +fn normalize_prune_statuses(statuses: Vec) -> Vec { + if statuses.is_empty() { + vec![AutomationPruneStatus::Previewed] + } else { + let mut normalized = Vec::new(); + for status in statuses { + if !normalized.contains(&status) { + normalized.push(status); + } + } + normalized + } +} + +fn prune_status_to_run_status(status: AutomationPruneStatus) -> AutomationRunStatus { + match status { + AutomationPruneStatus::Previewed => AutomationRunStatus::Previewed, + AutomationPruneStatus::Applied => AutomationRunStatus::Applied, + AutomationPruneStatus::ApplyFailed => AutomationRunStatus::ApplyFailed, + } +} + +async fn verification_audit_task( + config_report: &ConfigReport, +) -> Result { + let config_report = config_report.clone(); + spawn_blocking(move || crate::audit::verification(&config_report)) + .await + .map_err(|source| AutomationServiceError::TaskPanic { source })? +} + async fn acquire_apply_run_lock_task( config_report: &ConfigReport, run_id: i64, @@ -947,6 +1254,22 @@ async fn finalize_run_task( Ok(()) } +async fn prune_automation_runs_task( + config_report: &ConfigReport, + input: &PruneAutomationRunsInput, +) -> Result { + let database_path = config_report.config.store.database_path.clone(); + let busy_timeout_ms = config_report.config.store.busy_timeout_ms; + let input = input.clone(); + spawn_blocking(move || { + store::automation::prune_automation_runs(&database_path, busy_timeout_ms, &input) + }) + .await + .map_err(|source| AutomationServiceError::TaskPanic { source })? + .map_err(|source| AutomationServiceError::AutomationWrite { source }) + .map_err(Into::into) +} + async fn append_run_event_task( config_report: &ConfigReport, input: &AppendAutomationRunEventInput, @@ -1015,10 +1338,10 @@ fn configured_paths(config_report: &ConfigReport) -> Result, + /// Maximum number of thread candidates to preview + #[arg(long, default_value_t = crate::automation::DEFAULT_AUTOMATION_ROLLOUT_LIMIT)] + limit: usize, + /// Emit JSON instead of plain text + #[arg(long)] + json: bool, + }, /// Inspect a persisted automation review snapshot Show { /// Numeric automation run ID @@ -299,6 +311,38 @@ pub enum AutomationCommand { #[arg(long)] json: bool, }, + /// Prune stale local automation review snapshots after a dry-run review + Prune { + /// Delete snapshots older than this many days + #[arg(long)] + older_than_days: u32, + /// Restrict pruning to one or more terminal snapshot statuses + #[arg(long = "status", value_enum)] + statuses: Vec, + /// Execute the local snapshot deletion + #[arg(long)] + execute: bool, + /// Emit JSON instead of plain text + #[arg(long)] + json: bool, + }, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, ValueEnum)] +pub enum AutomationPruneStatusArg { + Previewed, + Applied, + ApplyFailed, +} + +impl From for crate::automation::AutomationPruneStatus { + fn from(value: AutomationPruneStatusArg) -> Self { + match value { + AutomationPruneStatusArg::Previewed => Self::Previewed, + AutomationPruneStatusArg::Applied => Self::Applied, + AutomationPruneStatusArg::ApplyFailed => Self::ApplyFailed, + } + } } #[derive(Debug, Subcommand)] diff --git a/src/cli_output/errors.rs b/src/cli_output/errors.rs index 7dcd268..0bc603d 100644 --- a/src/cli_output/errors.rs +++ b/src/cli_output/errors.rs @@ -116,6 +116,8 @@ fn classify_error(error: &AnyhowError) -> (ErrorCode, &'static str) { (ErrorCode::Conflict, "automation.apply.in_progress") } AutomationServiceError::InvalidLimit + | AutomationServiceError::InvalidRolloutLimit + | AutomationServiceError::InvalidPruneWindow | AutomationServiceError::ExecuteRequired | AutomationServiceError::RuleFileMissing { .. } | AutomationServiceError::RuleFileRead { .. } diff --git a/src/cli_output/tests.rs b/src/cli_output/tests.rs index a0cda3c..812531a 100644 --- a/src/cli_output/tests.rs +++ b/src/cli_output/tests.rs @@ -166,6 +166,19 @@ fn automation_apply_in_progress_maps_to_conflict_code() { assert_eq!(exit_code(&report), std::process::ExitCode::from(5)); } +#[test] +fn automation_prune_validation_maps_to_validation_failed_code() { + let error = anyhow!(AutomationServiceError::InvalidPruneWindow); + + let report = describe_error(&error, "automation.prune"); + let value = to_value(json_failure_value(&report)).unwrap(); + + assert_eq!(value["error"]["code"], json!("validation_failed")); + assert_eq!(value["error"]["kind"], json!("automation.validation")); + assert_eq!(value["error"]["operation"], json!("automation.prune")); + assert_eq!(exit_code(&report), std::process::ExitCode::from(2)); +} + #[test] fn attachment_file_errors_map_to_validation_failed_code() { let error = anyhow!(WorkflowServiceError::AttachmentRead { diff --git a/src/handlers/automation.rs b/src/handlers/automation.rs index 00205b8..7e0fb2f 100644 --- a/src/handlers/automation.rs +++ b/src/handlers/automation.rs @@ -25,6 +25,16 @@ pub(crate) async fn handle_automation_command( ) .await? .print(json)?, + AutomationCommand::Rollout { + rule_ids, + limit, + json, + } => automation::rollout( + &config_report, + automation::AutomationRolloutRequest { rule_ids, limit }, + ) + .await? + .print(json)?, AutomationCommand::Show { run_id, json } => automation::show_run(&config_report, run_id) .await? .print(json)?, @@ -35,6 +45,21 @@ pub(crate) async fn handle_automation_command( } => automation::apply_run(&config_report, run_id, execute) .await? .print(json)?, + AutomationCommand::Prune { + older_than_days, + statuses, + execute, + json, + } => automation::prune_runs( + &config_report, + automation::AutomationPruneRequest { + older_than_days, + statuses: statuses.into_iter().map(Into::into).collect(), + execute, + }, + ) + .await? + .print(json)?, } Ok(()) diff --git a/src/lib.rs b/src/lib.rs index f1eb60d..834c093 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -223,6 +223,10 @@ fn command_metadata(command: &Commands) -> CommandMetadata { json: *json, operation: "automation.run", }, + AutomationCommand::Rollout { json, .. } => CommandMetadata { + json: *json, + operation: "automation.rollout", + }, AutomationCommand::Show { json, .. } => CommandMetadata { json: *json, operation: "automation.show", @@ -231,6 +235,10 @@ fn command_metadata(command: &Commands) -> CommandMetadata { json: *json, operation: "automation.apply", }, + AutomationCommand::Prune { json, .. } => CommandMetadata { + json: *json, + operation: "automation.prune", + }, }, Commands::Sync { command } => match command { SyncCommand::Run(args) => CommandMetadata { diff --git a/src/store/automation/mod.rs b/src/store/automation/mod.rs index 08ca98e..9abf6a9 100644 --- a/src/store/automation/mod.rs +++ b/src/store/automation/mod.rs @@ -10,14 +10,15 @@ pub(crate) use read::{ pub(crate) use types::{ AppendAutomationRunEventInput, AutomationActionKind, AutomationActionSnapshot, AutomationApplyStatus, AutomationDoctorReport, AutomationMatchReason, - AutomationRunCandidateRecord, AutomationRunDetail, AutomationRunEventRecord, - AutomationRunRecord, AutomationRunStatus, AutomationStoreReadError, AutomationStoreWriteError, - AutomationThreadCandidate, CandidateApplyResultInput, CreateAutomationRunInput, - FinalizeAutomationRunInput, NewAutomationRunCandidate, + AutomationPruneStoreReport, AutomationRunCandidateRecord, AutomationRunDetail, + AutomationRunEventRecord, AutomationRunRecord, AutomationRunStatus, AutomationStoreReadError, + AutomationStoreWriteError, AutomationThreadCandidate, CandidateApplyResultInput, + CreateAutomationRunInput, FinalizeAutomationRunInput, NewAutomationRunCandidate, + PruneAutomationRunsInput, }; pub(crate) use write::{ append_automation_run_event, claim_automation_run_for_apply, create_automation_run, - finalize_automation_run, record_candidate_apply_result, + finalize_automation_run, prune_automation_runs, record_candidate_apply_result, }; fn is_missing_automation_table_error(error: &rusqlite::Error) -> bool { diff --git a/src/store/automation/tests.rs b/src/store/automation/tests.rs index 95a02f5..e28d041 100644 --- a/src/store/automation/tests.rs +++ b/src/store/automation/tests.rs @@ -2,9 +2,10 @@ use super::{ AppendAutomationRunEventInput, AutomationActionKind, AutomationActionSnapshot, AutomationApplyStatus, AutomationMatchReason, AutomationRunStatus, AutomationStoreWriteError, CandidateApplyResultInput, CreateAutomationRunInput, FinalizeAutomationRunInput, - NewAutomationRunCandidate, append_automation_run_event, claim_automation_run_for_apply, - create_automation_run, finalize_automation_run, get_automation_run_detail, inspect_automation, - list_latest_thread_candidates, record_candidate_apply_result, + NewAutomationRunCandidate, PruneAutomationRunsInput, append_automation_run_event, + claim_automation_run_for_apply, create_automation_run, finalize_automation_run, + get_automation_run_detail, inspect_automation, list_latest_thread_candidates, + prune_automation_runs, record_candidate_apply_result, }; use crate::config::resolve; use crate::store::{accounts, init, mailbox}; @@ -100,6 +101,168 @@ fn create_automation_run_persists_detail_and_doctor_counts() { assert_eq!(doctor.candidate_count, 1); } +#[test] +fn prune_automation_runs_dry_run_reports_without_deleting() { + let repo_root = temp_repo_root(); + let paths = WorkspacePaths::from_repo_root(repo_root.path().to_path_buf()); + paths.ensure_runtime_dirs().unwrap(); + let config_report = resolve(&paths).unwrap(); + init(&config_report).unwrap(); + let account = seed_account(&config_report); + let old_run = create_automation_run( + &config_report.config.store.database_path, + config_report.config.store.busy_timeout_ms, + &CreateAutomationRunInput { + account_id: account.account_id.clone(), + rule_file_path: String::from(".mailroom/automation.toml"), + rule_file_hash: String::from("old"), + selected_rule_ids: vec![String::from("archive-old")], + created_at_epoch_s: 100, + candidates: vec![sample_candidate()], + }, + ) + .unwrap(); + create_automation_run( + &config_report.config.store.database_path, + config_report.config.store.busy_timeout_ms, + &CreateAutomationRunInput { + account_id: account.account_id.clone(), + rule_file_path: String::from(".mailroom/automation.toml"), + rule_file_hash: String::from("new"), + selected_rule_ids: vec![String::from("archive-new")], + created_at_epoch_s: 1_000, + candidates: vec![sample_candidate()], + }, + ) + .unwrap(); + + let report = prune_automation_runs( + &config_report.config.store.database_path, + config_report.config.store.busy_timeout_ms, + &PruneAutomationRunsInput { + account_id: account.account_id, + cutoff_epoch_s: 500, + statuses: vec![AutomationRunStatus::Previewed], + execute: false, + }, + ) + .unwrap(); + + assert_eq!(report.matched_run_count, 1); + assert_eq!(report.matched_candidate_count, 1); + assert_eq!(report.matched_event_count, 1); + assert_eq!(report.deleted_run_count, 0); + assert!( + get_automation_run_detail( + &config_report.config.store.database_path, + config_report.config.store.busy_timeout_ms, + old_run.run.run_id, + ) + .unwrap() + .is_some() + ); + let doctor = inspect_automation( + &config_report.config.store.database_path, + config_report.config.store.busy_timeout_ms, + ) + .unwrap() + .unwrap(); + assert_eq!(doctor.run_count, 2); +} + +#[test] +fn prune_automation_runs_execute_deletes_runs_and_cascades_detail() { + let repo_root = temp_repo_root(); + let paths = WorkspacePaths::from_repo_root(repo_root.path().to_path_buf()); + paths.ensure_runtime_dirs().unwrap(); + let config_report = resolve(&paths).unwrap(); + init(&config_report).unwrap(); + let account = seed_account(&config_report); + let old_run = create_automation_run( + &config_report.config.store.database_path, + config_report.config.store.busy_timeout_ms, + &CreateAutomationRunInput { + account_id: account.account_id.clone(), + rule_file_path: String::from(".mailroom/automation.toml"), + rule_file_hash: String::from("old"), + selected_rule_ids: vec![String::from("archive-old")], + created_at_epoch_s: 100, + candidates: vec![sample_candidate()], + }, + ) + .unwrap(); + let applying_run = create_automation_run( + &config_report.config.store.database_path, + config_report.config.store.busy_timeout_ms, + &CreateAutomationRunInput { + account_id: account.account_id.clone(), + rule_file_path: String::from(".mailroom/automation.toml"), + rule_file_hash: String::from("applying"), + selected_rule_ids: vec![String::from("archive-applying")], + created_at_epoch_s: 100, + candidates: vec![sample_candidate()], + }, + ) + .unwrap(); + finalize_automation_run( + &config_report.config.store.database_path, + config_report.config.store.busy_timeout_ms, + &FinalizeAutomationRunInput { + run_id: applying_run.run.run_id, + status: AutomationRunStatus::Applying, + applied_at_epoch_s: 101, + }, + ) + .unwrap(); + + let report = prune_automation_runs( + &config_report.config.store.database_path, + config_report.config.store.busy_timeout_ms, + &PruneAutomationRunsInput { + account_id: account.account_id, + cutoff_epoch_s: 500, + statuses: vec![ + AutomationRunStatus::Previewed, + AutomationRunStatus::Applied, + AutomationRunStatus::ApplyFailed, + ], + execute: true, + }, + ) + .unwrap(); + + assert_eq!(report.matched_run_count, 1); + assert_eq!(report.matched_candidate_count, 1); + assert_eq!(report.matched_event_count, 1); + assert_eq!(report.deleted_run_count, 1); + assert!( + get_automation_run_detail( + &config_report.config.store.database_path, + config_report.config.store.busy_timeout_ms, + old_run.run.run_id, + ) + .unwrap() + .is_none() + ); + assert!( + get_automation_run_detail( + &config_report.config.store.database_path, + config_report.config.store.busy_timeout_ms, + applying_run.run.run_id, + ) + .unwrap() + .is_some() + ); + let doctor = inspect_automation( + &config_report.config.store.database_path, + config_report.config.store.busy_timeout_ms, + ) + .unwrap() + .unwrap(); + assert_eq!(doctor.run_count, 1); + assert_eq!(doctor.candidate_count, 1); +} + #[test] fn append_automation_run_event_rejects_account_mismatch() { let repo_root = temp_repo_root(); diff --git a/src/store/automation/types.rs b/src/store/automation/types.rs index 00c5cd8..c3181df 100644 --- a/src/store/automation/types.rs +++ b/src/store/automation/types.rs @@ -272,6 +272,22 @@ pub(crate) struct FinalizeAutomationRunInput { pub(crate) applied_at_epoch_s: i64, } +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub(crate) struct PruneAutomationRunsInput { + pub(crate) account_id: String, + pub(crate) cutoff_epoch_s: i64, + pub(crate) statuses: Vec, + pub(crate) execute: bool, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub(crate) struct AutomationPruneStoreReport { + pub(crate) matched_run_count: i64, + pub(crate) matched_candidate_count: i64, + pub(crate) matched_event_count: i64, + pub(crate) deleted_run_count: i64, +} + #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub(crate) struct AutomationDoctorReport { pub(crate) run_count: i64, diff --git a/src/store/automation/write.rs b/src/store/automation/write.rs index e699731..205e69e 100644 --- a/src/store/automation/write.rs +++ b/src/store/automation/write.rs @@ -1,11 +1,11 @@ use super::{ - AppendAutomationRunEventInput, AutomationRunDetail, AutomationRunStatus, - AutomationStoreWriteError, CandidateApplyResultInput, CreateAutomationRunInput, - FinalizeAutomationRunInput, + AppendAutomationRunEventInput, AutomationPruneStoreReport, AutomationRunDetail, + AutomationRunStatus, AutomationStoreWriteError, CandidateApplyResultInput, + CreateAutomationRunInput, FinalizeAutomationRunInput, PruneAutomationRunsInput, }; use crate::store::connection; use anyhow::Result; -use rusqlite::{OptionalExtension, params}; +use rusqlite::{OptionalExtension, params, params_from_iter, types::Value}; use std::path::Path; pub(crate) fn create_automation_run( @@ -266,6 +266,76 @@ pub(crate) fn claim_automation_run_for_apply( Ok(()) } +pub(crate) fn prune_automation_runs( + database_path: &Path, + busy_timeout_ms: u64, + input: &PruneAutomationRunsInput, +) -> Result { + let mut connection = connection::open_or_create(database_path, busy_timeout_ms) + .map_err(|source| AutomationStoreWriteError::open_database(database_path, source))?; + let transaction = connection.transaction()?; + + if input.statuses.is_empty() { + transaction.commit()?; + return Ok(AutomationPruneStoreReport { + matched_run_count: 0, + matched_candidate_count: 0, + matched_event_count: 0, + deleted_run_count: 0, + }); + } + + let status_placeholders = vec!["?"; input.statuses.len()].join(", "); + let predicate = + format!("account_id = ? AND created_at_epoch_s < ? AND status IN ({status_placeholders})"); + let params = prune_params(input); + + let count_sql = format!( + "WITH matched_runs AS ( + SELECT run_id + FROM automation_runs + WHERE {predicate} + ) + SELECT + (SELECT COUNT(*) FROM matched_runs), + (SELECT COUNT(*) FROM automation_run_candidates + WHERE run_id IN (SELECT run_id FROM matched_runs)), + (SELECT COUNT(*) FROM automation_run_events + WHERE run_id IN (SELECT run_id FROM matched_runs))" + ); + let mut report = transaction.query_row(&count_sql, params_from_iter(params.iter()), |row| { + Ok(AutomationPruneStoreReport { + matched_run_count: row.get(0)?, + matched_candidate_count: row.get(1)?, + matched_event_count: row.get(2)?, + deleted_run_count: 0, + }) + })?; + + if input.execute { + let delete_sql = format!("DELETE FROM automation_runs WHERE {predicate}"); + let deleted_run_count = + transaction.execute(&delete_sql, params_from_iter(params.iter()))?; + report.deleted_run_count = i64::try_from(deleted_run_count).unwrap_or(i64::MAX); + } + + transaction.commit()?; + Ok(report) +} + +fn prune_params(input: &PruneAutomationRunsInput) -> Vec { + let mut values = Vec::with_capacity(input.statuses.len() + 2); + values.push(Value::Text(input.account_id.clone())); + values.push(Value::Integer(input.cutoff_epoch_s)); + values.extend( + input + .statuses + .iter() + .map(|status| Value::Text(status.as_str().to_owned())), + ); + values +} + pub(crate) fn append_automation_run_event( database_path: &Path, busy_timeout_ms: u64,