perf(observation): hoist retention store and reap artifact-root scratch

Terminal-run retention resolved the observation store and the artifact
root inside every per-run artifact plan, so a default --limit 1000 sweep
opened up to 2000 SQLite connections and re-ran the migration ladder for
each. Both are now resolved once per sweep and threaded through a new
`cleanup_persisted_artifacts_with_store`. The apply pass still re-plans
each run immediately before deleting its bytes: the planning loop walks
every candidate before any deletion happens, so those plans are already
stale by then.

`removed_run_count` subtracted `skipped_run_ids` from `removable_run_ids`,
under-reporting by every run the planning loop had already blocked, since
those runs were never in `removable_run_ids`. It now counts what was
deleted.

Adds an `orphaned-artifact-bytes` cleanup category for the two
artifact-root path families that are created outside any durable journal
and are only reclaimed by an in-process branch, so SIGKILL leaks them:
`<run_id>/.artifact-<uuid>.staging` and `_scratch/patch-<label>-<uuid>/`.

It is deliberately not a "path with no artifacts row" reaper. The artifact
root is a shared namespace with at least nine subsystem-owned top-level
subtrees carrying no row at those paths, and artifact bytes are written
before their row is inserted by design, so row absence is the normal state
of a publication that is succeeding. Ownership is proven by name shape
plus a fixed 24h age floor instead, and the database is not consulted.

Size is measured best-effort and never gates removal, inverting the
coupling in `workspace/sync/mod.rs:2602` where a failed `du -sk`
downgraded liveness to "unknown" and silently disabled prune.

Refs #10284, #10285

Author: homeboy-ci[bot]

crates/homeboy-cli/src/commands/cleanup.rs

@@ -11,7 +11,8 @@ use homeboy::core::defaults;
 use homeboy::core::engine;
 use homeboy::core::engine::shell::quote_arg;
 use homeboy::core::observation::runs_service::{
-    self, PersistedArtifactCleanupOptions, RunnerDownloadCleanupOptions,
+    self, OrphanedArtifactBytesCleanupOptions, PersistedArtifactCleanupOptions,
+    RunnerDownloadCleanupOptions,
 };
 use homeboy::core::resource_cleanup_intent::ResourceCleanupIntent;
 use homeboy::core::worktree::{self, WorktreeCleanupOptions, WorktreeCleanupOutput};
@@ -69,6 +70,7 @@ pub enum CleanupCategoryArg {
     WorktreeProviders,
     TerminalRuns,
     PersistedRunArtifacts,
+    OrphanedArtifactBytes,
     RunnerDownloads,
     RemoteLabWorkspaces,
     RuntimeTmp,
@@ -639,6 +641,19 @@ const TERMINAL_RUNS_METADATA: CleanupInventoryCategoryMetadata = CleanupInventor
     apply_command: "homeboy runs retention --apply",
 };
 
+/// Crash residue under the artifact root that no database row can describe.
+/// This is the only artifact-root cleanup that is not row-driven, so it is
+/// scoped to the two name families a single private constructor owns rather
+/// than to "anything without a row" — see
+/// `runs_service::orphaned_artifact_bytes` for why a row join is unsafe here.
+const ORPHANED_ARTIFACT_BYTES_METADATA: CleanupInventoryCategoryMetadata =
+    CleanupInventoryCategoryMetadata {
+        category: "orphaned_artifact_bytes",
+        include_arg: "orphaned-artifact-bytes",
+        dry_run_command: "homeboy cleanup --include orphaned-artifact-bytes",
+        apply_command: "homeboy cleanup --include orphaned-artifact-bytes --apply",
+    };
+
 pub(crate) const RUNNER_DOWNLOADS_METADATA: CleanupInventoryCategoryMetadata =
     CleanupInventoryCategoryMetadata {
         category: "runner_downloads",
@@ -795,6 +810,24 @@ fn cleanup_inventory(args: CleanupArgs) -> homeboy::core::Result<Value> {
         categories.push(persisted_artifacts_category(persisted, resources, apply)?);
     }
 
+    if selected.includes(CleanupCategoryArg::OrphanedArtifactBytes) {
+        let output =
+            runs_service::cleanup_orphaned_artifact_bytes(OrphanedArtifactBytesCleanupOptions {
+                apply,
+                limit: usize::try_from(limit).unwrap_or(usize::MAX),
+            })?;
+        categories.push(category_from_output(
+            ORPHANED_ARTIFACT_BYTES_METADATA,
+            apply,
+            output.planned_count,
+            output.removed_count,
+            output.skipped_count,
+            output.planned_size_bytes,
+            output.removed_size_bytes,
+            output,
+        )?);
+    }
+
     if selected.includes(CleanupCategoryArg::RunnerDownloads) {
         let output = runs_service::cleanup_runner_downloads(RunnerDownloadCleanupOptions {
             apply,
@@ -2041,6 +2074,13 @@ mod tests {
                 "homeboy runs artifact cleanup-persisted",
                 "homeboy runs artifact cleanup-persisted --apply",
             ),
+            (
+                ORPHANED_ARTIFACT_BYTES_METADATA,
+                "orphaned_artifact_bytes",
+                "orphaned-artifact-bytes",
+                "homeboy cleanup --include orphaned-artifact-bytes",
+                "homeboy cleanup --include orphaned-artifact-bytes --apply",
+            ),
             (
                 RUNNER_DOWNLOADS_METADATA,
                 "runner_downloads",

crates/homeboy-core/src/observation/runs_service/mod.rs

@@ -173,13 +173,15 @@ pub enum ArtifactStorage {
 
 mod artifact_links;
 mod artifact_resolve;
+mod orphaned_artifact_bytes;
 mod persisted_cleanup;
 mod run_lookup;
 mod runner_downloads;
 pub mod runner_evidence;
 
 pub use artifact_links::*;
 pub use artifact_resolve::*;
+pub use orphaned_artifact_bytes::*;
 pub use persisted_cleanup::*;
 pub use run_lookup::*;
 pub use runner_downloads::*;
@@ -204,24 +206,22 @@ pub fn retain_terminal_runs(
         ));
     }
     let finished_before = (Utc::now() - Duration::days(options.older_than_days)).to_rfc3339();
+    // The store and artifact root are resolved once for the whole sweep. Both
+    // used to be re-derived inside each per-run plan call, which meant one
+    // connection open plus a full migration ladder per candidate run.
     let mut store = ObservationStore::open_initialized()?;
+    let artifact_root = crate::artifacts::root()?;
     let candidate_run_ids = store.terminal_run_ids_before(&finished_before, options.limit)?;
     let mut artifact_cleanup = Vec::new();
     let mut lifecycle_directories = Vec::new();
     let mut removable_run_ids = Vec::new();
     let mut skipped_run_ids = Vec::new();
     for run_id in &candidate_run_ids {
-        let artifacts = cleanup_persisted_artifacts(PersistedArtifactCleanupOptions {
-            apply: false,
-            older_than_days: 0,
-            run_id: Some(run_id.clone()),
-            kind: None,
-            artifact_type: None,
-            run_kind: None,
-            component_id: None,
-            limit: 10_000,
-            terminal_only: true,
-        })?;
+        let artifacts = cleanup_persisted_artifacts_with_store(
+            &store,
+            &artifact_root,
+            terminal_run_artifact_plan_options(run_id),
+        )?;
         let lifecycle_directory = terminal_run_lifecycle_directory(&store, run_id)?;
         let blocked = artifacts
             .rows
@@ -237,21 +237,25 @@ pub fn retain_terminal_runs(
             }
         }
     }
+    let mut removed_run_count = 0;
     if options.apply {
         // Revalidate and remove artifact bytes before deleting any provenance.
         // A blocked resource leaves the terminal run record and lifecycle root intact.
+        //
+        // This second plan pass is *not* redundant with the loop above: that
+        // loop walks every candidate before any deletion happens, so the
+        // earliest plans are already stale by the time apply begins. Re-reading
+        // each run's classification immediately before deleting its bytes is
+        // the only thing that catches a run whose artifacts became unsafe (or
+        // whose owning run left a terminal state) during the planning window.
+        // What was redundant, and is now gone, is reopening the store and
+        // re-resolving the artifact root on every one of these calls.
         for run_id in &removable_run_ids {
-            let artifacts = cleanup_persisted_artifacts(PersistedArtifactCleanupOptions {
-                apply: false,
-                older_than_days: 0,
-                run_id: Some(run_id.clone()),
-                kind: None,
-                artifact_type: None,
-                run_kind: None,
-                component_id: None,
-                limit: 10_000,
-                terminal_only: true,
-            })?;
+            let artifacts = cleanup_persisted_artifacts_with_store(
+                &store,
+                &artifact_root,
+                terminal_run_artifact_plan_options(run_id),
+            )?;
             if artifacts
                 .rows
                 .iter()
@@ -262,7 +266,6 @@ pub fn retain_terminal_runs(
             }
             let records = store.list_artifacts(run_id)?;
             let run = store.get_run(run_id)?;
-            let artifact_root = crate::artifacts::root()?;
             for row in artifacts.rows.iter().filter(|row| row.action == "remove") {
                 let artifact = records
                     .iter()
@@ -310,24 +313,39 @@ pub fn retain_terminal_runs(
             }
         }
         store.delete_terminal_runs(&deleted)?;
+        // Count what was actually deleted. Subtracting `skipped_run_ids` from
+        // `removable_run_ids` under-reported by every run the planning loop had
+        // already blocked, because those runs are in `skipped_run_ids` but were
+        // never in `removable_run_ids` to begin with.
+        removed_run_count = deleted.len();
     }
     Ok(TerminalRunRetentionOutcome {
         dry_run: !options.apply,
         older_than_days: options.older_than_days,
-        removed_run_count: if options.apply {
-            removable_run_ids
-                .len()
-                .saturating_sub(skipped_run_ids.len())
-        } else {
-            0
-        },
+        removed_run_count,
         candidate_run_ids,
         artifact_cleanup,
         lifecycle_directories,
         skipped_run_ids,
     })
 }
 
+/// Per-run artifact plan used by terminal retention. Always a dry-run plan:
+/// retention removes bytes itself and releases the rows with the owning run.
+fn terminal_run_artifact_plan_options(run_id: &str) -> PersistedArtifactCleanupOptions {
+    PersistedArtifactCleanupOptions {
+        apply: false,
+        older_than_days: 0,
+        run_id: Some(run_id.to_string()),
+        kind: None,
+        artifact_type: None,
+        run_kind: None,
+        component_id: None,
+        limit: 10_000,
+        terminal_only: true,
+    }
+}
+
 fn terminal_run_lifecycle_directory(
     store: &ObservationStore,
     run_id: &str,