fix(rig): wire WorkloadSpec.lifecycle and report the lifecycle_snapshots drift class (#10553)

* feat(rig): give WorkloadSpec.lifecycle a reader and repair a snapshot class

Closes the two residual gaps in #10317 and #10319 after the pipeline
lifecycle step, snapshot state, and shared-path/service repair landed.

#10317 — `WorkloadSpec.lifecycle` had zero readers anywhere in the
workspace: no `.lifecycle()` call site, no `.lifecycle` field read
outside its own accessor, no cross-crate use. It was parsed, defaulted
in four test constructors, serialized, documented — and never executed.

Wire it rather than delete it: `PipelineStep::Lifecycle` can now take
its contract from a `workload` reference instead of an inline payload.
Exactly one source is legal, and resolution is fail-closed — unknown
extension, unmatched `path`, ambiguous selection, and a workload with no
contract all fail before a phase runs. One declaration on the workload
now governs every op the rig runs against it, instead of an `up` copy
and a `down` copy drifting apart.

Inline `lifecycle` becomes optional, which serde no longer rejects at
load time, so `rig lint` gains the "exactly one contract source" check
plus shape validation of the reference.

#10319 — `repair` now reports the `lifecycle_snapshots` class it had no
row for. Deliberately report-only: the handle is opaque, so deleting the
record does not reap the environment behind it, it only strands one. A
handle a declared step still owns is `skipped` with the op that reaps
it; a handle no declared step owns is `blocked`.

* test(rig): update lifecycle step spec tests for the optional contract body

CI caught the one consumer my grep missed: I searched `crates/` for
`PipelineStep::Lifecycle` but the `#[path]`-included test modules live
under `tests/`, so `tests/core/rig/spec_test.rs` was never in scope.

Also extends the round-trip coverage to the `workload` reference form:
the reference carries no inline body, and does not grow an empty one on
serialize.

---------

Co-authored-by: chubes-bot <266378653+homeboy-ci[bot]@users.noreply.github.com>

Author: chubes4

crates/homeboy-rig/src/lib.rs

@@ -98,14 +98,14 @@ pub use spec::{
     DependencyMaterializationStepSpec, DiscoverSpec, ExecutableRequirementSpec,
     FilesystemAssertionKind, FilesystemAssertionSpec, LifecycleContract, LifecyclePhaseContract,
     LifecyclePhaseKind, LifecyclePhaseResult, LifecyclePhaseStatus, LifecycleResultMetadata,
-    LifecycleSnapshotRef, NewerThanSpec, NormalizedDependencyMaterializationStep, PatchOp,
-    PipelineStep, RigRequirementsSpec, RigResourceRetentionSpec, RigResourcesSpec, RigSpec,
-    RunnerToolRequirementSpec, ServiceKind, ServiceSpec, SharedPathOp, SharedPathSpec, StackOp,
-    SymlinkSpec, TimeSource, TraceConfig, TraceDependencySpec, TraceExperimentArtifactSpec,
-    TraceExperimentCommandSpec, TraceExperimentSpec, TraceGuardrailSpec,
-    TraceNativePublicPreviewSpec, TracePhaseTemplateSpec, TracePreviewAssetFanoutSpec,
-    TraceProfileSpec, TracePublicPreviewMode, TracePublicPreviewSpec, TraceVariantSpec,
-    WorkloadSpec, RIG_RESOURCE_CLASSES, RIG_RESOURCE_CLASS_EXCLUSIVE,
+    LifecycleSnapshotRef, LifecycleWorkloadKind, LifecycleWorkloadRef, NewerThanSpec,
+    NormalizedDependencyMaterializationStep, PatchOp, PipelineStep, RigRequirementsSpec,
+    RigResourceRetentionSpec, RigResourcesSpec, RigSpec, RunnerToolRequirementSpec, ServiceKind,
+    ServiceSpec, SharedPathOp, SharedPathSpec, StackOp, SymlinkSpec, TimeSource, TraceConfig,
+    TraceDependencySpec, TraceExperimentArtifactSpec, TraceExperimentCommandSpec,
+    TraceExperimentSpec, TraceGuardrailSpec, TraceNativePublicPreviewSpec, TracePhaseTemplateSpec,
+    TracePreviewAssetFanoutSpec, TraceProfileSpec, TracePublicPreviewMode, TracePublicPreviewSpec,
+    TraceVariantSpec, WorkloadSpec, RIG_RESOURCE_CLASSES, RIG_RESOURCE_CLASS_EXCLUSIVE,
     RIG_RESOURCE_CLASS_LIFECYCLE_SNAPSHOTS, RIG_RESOURCE_CLASS_PATHS, RIG_RESOURCE_CLASS_PORTS,
     RIG_RESOURCE_CLASS_PROCESS_PATTERNS,
 };
@@ -124,8 +124,8 @@ pub use workloads::{
     extension_workload_inputs, invocation_requirements_for_extension_workloads,
     required_component_id_for_workload, required_extension_ids_for_workload,
     runner_capabilities_for_extension, trace_dependencies_for_extension,
-    workload_path_expansions_for_extension, workloads_for_extension, RigExtensionWorkloadInputs,
-    RigWorkloadKind, RigWorkloadPathExpansion,
+    workload_lifecycle_contract, workload_path_expansions_for_extension, workloads_for_extension,
+    RigExtensionWorkloadInputs, RigWorkloadKind, RigWorkloadPathExpansion,
 };
 
 use discovery::discover_rigs_for_install;

crates/homeboy-rig/src/lint.rs

@@ -522,6 +522,7 @@ fn portability_failures(root: &Path, files: &[PathBuf]) -> Result<Vec<String>> {
         };
         failures.extend(shared_path_failures(root, file, &value));
         failures.extend(resource_retention_failures(root, file, &value));
+        failures.extend(lifecycle_step_source_failures(root, file, &value));
     }
 
     Ok(failures)
@@ -584,6 +585,87 @@ fn shared_path_failures(root: &Path, file: &Path, rig: &serde_json::Value) -> Ve
     failures
 }
 
+/// A `kind: "lifecycle"` step takes its contract from exactly one source:
+/// an inline `lifecycle` payload, or a `workload` reference.
+///
+/// Both fields are optional in the schema so either shape parses, which means
+/// serde no longer rejects a step that declares neither. The executor still
+/// fails closed at run time, but a rig author should not have to run `rig up`
+/// to find out — that is what this lint is for.
+fn lifecycle_step_source_failures(
+    root: &Path,
+    file: &Path,
+    rig: &serde_json::Value,
+) -> Vec<String> {
+    let rel = display_relative(root, file);
+    let Some(pipelines) = rig.get("pipeline").and_then(|value| value.as_object()) else {
+        return Vec::new();
+    };
+
+    let mut failures = Vec::new();
+    for (name, steps) in pipelines {
+        let Some(steps) = steps.as_array() else {
+            continue;
+        };
+        for (index, step) in steps.iter().enumerate() {
+            let Some(step) = step.as_object() else {
+                continue;
+            };
+            if step.get("kind").and_then(|value| value.as_str()) != Some("lifecycle") {
+                continue;
+            }
+            let inline = step.contains_key("lifecycle");
+            let reference = step.contains_key("workload");
+            let location = format!("{rel}: pipeline.{name}[{index}] lifecycle step");
+            match (inline, reference) {
+                (true, true) => failures.push(format!(
+                    "{location} declares both `lifecycle` and `workload`; use exactly one contract source"
+                )),
+                (false, false) => failures.push(format!(
+                    "{location} declares neither `lifecycle` nor `workload`; use exactly one contract source"
+                )),
+                _ => {}
+            }
+
+            if let Some(workload) = step.get("workload") {
+                failures.extend(lifecycle_workload_ref_failures(&location, workload));
+            }
+        }
+    }
+    failures
+}
+
+/// Shape checks for a `workload` reference. Resolution against the declared
+/// workload maps stays at run time — lint only rejects a reference that could
+/// never resolve for anyone.
+fn lifecycle_workload_ref_failures(location: &str, workload: &serde_json::Value) -> Vec<String> {
+    let Some(workload) = workload.as_object() else {
+        return vec![format!("{location} `workload` must be an object")];
+    };
+
+    let mut failures = Vec::new();
+    match workload.get("kind").and_then(|value| value.as_str()) {
+        Some("bench" | "fuzz" | "trace") => {}
+        Some(other) => failures.push(format!(
+            "{location} `workload.kind` must be one of bench, fuzz, trace (found '{other}')"
+        )),
+        None => failures.push(format!(
+            "{location} `workload.kind` is required and must be one of bench, fuzz, trace"
+        )),
+    }
+    let extension = workload
+        .get("extension")
+        .and_then(|value| value.as_str())
+        .unwrap_or("")
+        .trim();
+    if extension.is_empty() {
+        failures.push(format!(
+            "{location} `workload.extension` is required and must be a non-empty extension id"
+        ));
+    }
+    failures
+}
+
 /// `delete_after_ttl` without a `ttl` is contract-invalid: the lifecycle record
 /// would fail validation and the run's whole lifecycle index would be dropped.
 /// Catch it at lint time instead of losing the artifact at runtime.
@@ -1439,4 +1521,134 @@ mod tests {
         assert!(error.contains("fuzz_profile profile missing references write"));
         assert!(error.contains("bench_profile profile missing references slow"));
     }
+
+    #[test]
+    fn package_lint_reports_lifecycle_step_without_a_contract_source() {
+        let temp = tempfile::TempDir::new().expect("temp package");
+        let rig_dir = temp.path().join("rigs").join("lifecycle");
+        fs::create_dir_all(&rig_dir).expect("rig dir");
+        fs::write(
+            rig_dir.join("rig.json"),
+            r#"{
+                "id": "lifecycle",
+                "pipeline": {
+                    "up": [{ "kind": "lifecycle", "op": "prepare" }]
+                }
+            }"#,
+        )
+        .expect("write rig");
+
+        let outcome = run_package_lint_at(temp.path()).expect("lint package");
+        let step = portability_step(&outcome);
+
+        assert_eq!(step.status, "fail");
+        let error = step.error.as_ref().expect("error");
+        assert!(
+            error.contains("declares neither `lifecycle` nor `workload`"),
+            "{error}"
+        );
+        assert!(error.contains("pipeline.up[0]"), "{error}");
+    }
+
+    #[test]
+    fn package_lint_reports_lifecycle_step_with_two_contract_sources() {
+        let temp = tempfile::TempDir::new().expect("temp package");
+        let rig_dir = temp.path().join("rigs").join("lifecycle");
+        fs::create_dir_all(&rig_dir).expect("rig dir");
+        fs::write(
+            rig_dir.join("rig.json"),
+            r#"{
+                "id": "lifecycle",
+                "pipeline": {
+                    "up": [{
+                        "kind": "lifecycle",
+                        "lifecycle": { "phases": [] },
+                        "workload": { "kind": "fuzz", "extension": "generic" }
+                    }]
+                }
+            }"#,
+        )
+        .expect("write rig");
+
+        let outcome = run_package_lint_at(temp.path()).expect("lint package");
+        let step = portability_step(&outcome);
+
+        assert_eq!(step.status, "fail");
+        let error = step.error.as_ref().expect("error");
+        assert!(
+            error.contains("declares both `lifecycle` and `workload`"),
+            "{error}"
+        );
+    }
+
+    #[test]
+    fn package_lint_reports_malformed_lifecycle_workload_reference() {
+        let temp = tempfile::TempDir::new().expect("temp package");
+        let rig_dir = temp.path().join("rigs").join("lifecycle");
+        fs::create_dir_all(&rig_dir).expect("rig dir");
+        fs::write(
+            rig_dir.join("rig.json"),
+            r#"{
+                "id": "lifecycle",
+                "pipeline": {
+                    "up": [{
+                        "kind": "lifecycle",
+                        "workload": { "kind": "sandbox", "extension": "  " }
+                    }]
+                }
+            }"#,
+        )
+        .expect("write rig");
+
+        let outcome = run_package_lint_at(temp.path()).expect("lint package");
+        let step = portability_step(&outcome);
+
+        assert_eq!(step.status, "fail");
+        let error = step.error.as_ref().expect("error");
+        assert!(error.contains("`workload.kind` must be one of"), "{error}");
+        assert!(
+            error.contains("`workload.extension` is required"),
+            "{error}"
+        );
+    }
+
+    #[test]
+    fn package_lint_accepts_both_lifecycle_contract_sources_used_alone() {
+        let temp = tempfile::TempDir::new().expect("temp package");
+        let rig_dir = temp.path().join("rigs").join("lifecycle");
+        fs::create_dir_all(&rig_dir).expect("rig dir");
+        fs::write(
+            rig_dir.join("rig.json"),
+            r#"{
+                "id": "lifecycle",
+                "fuzz_workloads": {
+                    "generic": [{
+                        "path": "fuzz/a.workload.json",
+                        "lifecycle": {
+                            "phases": [{ "id": "make", "phase": "prepare", "command": "true" }]
+                        }
+                    }]
+                },
+                "pipeline": {
+                    "up": [{
+                        "kind": "lifecycle",
+                        "workload": { "kind": "fuzz", "extension": "generic" }
+                    }],
+                    "down": [{
+                        "kind": "lifecycle",
+                        "op": "teardown",
+                        "lifecycle": {
+                            "phases": [{ "id": "reap", "phase": "teardown", "command": "true" }]
+                        }
+                    }]
+                }
+            }"#,
+        )
+        .expect("write rig");
+
+        let outcome = run_package_lint_at(temp.path()).expect("lint package");
+        let step = portability_step(&outcome);
+
+        assert_eq!(step.status, "pass", "{:?}", step.error);
+    }
 }

crates/homeboy-rig/src/pipeline/lifecycle_step.rs

@@ -16,7 +16,8 @@ use std::time::Duration;
 use super::super::expand::{expand_vars, settings_env};
 use super::super::spec::{
     LifecycleContract, LifecyclePhaseContract, LifecyclePhaseKind, LifecyclePhaseResult,
-    LifecyclePhaseStatus, LifecycleResultMetadata, LifecycleSnapshotRef, RigSpec,
+    LifecyclePhaseStatus, LifecycleResultMetadata, LifecycleSnapshotRef, LifecycleWorkloadRef,
+    RigSpec,
 };
 use super::super::state::{now_rfc3339, LifecycleSnapshotState, RigState};
 use super::super::toolchain;
@@ -35,14 +36,17 @@ use homeboy_core::server::{
 /// names the thing it is holding a handle to.
 const DEFAULT_SNAPSHOT_KIND: &str = "lifecycle_snapshot";
 
+#[allow(clippy::too_many_arguments)]
 pub(super) fn run_lifecycle_step(
     rig: &RigSpec,
     step_id: Option<&str>,
     component: Option<&str>,
-    contract: &LifecycleContract,
+    contract: Option<&LifecycleContract>,
+    workload: Option<&LifecycleWorkloadRef>,
     op: LifecyclePhaseKind,
     settings: &[(String, String)],
 ) -> Result<()> {
+    let contract = resolve_contract(rig, contract, workload)?;
     let (result, failure) = execute_lifecycle_phases(rig, component, contract, op, settings)?;
 
     // Persist before propagating a phase failure: a handle captured before the
@@ -56,8 +60,36 @@ pub(super) fn run_lifecycle_step(
     }
 }
 
+/// Pick the contract this step executes: the inline payload, or the one a
+/// rig-owned workload declares.
+///
+/// Exactly one source is legal. Declaring both is a spec bug that would leave
+/// the reader guessing which contract governs, and declaring neither leaves
+/// nothing to run — both fail here, before any phase executes and before any
+/// state is touched.
+fn resolve_contract<'a>(
+    rig: &'a RigSpec,
+    contract: Option<&'a LifecycleContract>,
+    workload: Option<&LifecycleWorkloadRef>,
+) -> Result<&'a LifecycleContract> {
+    match (contract, workload) {
+        (Some(contract), None) => Ok(contract),
+        (None, Some(reference)) => {
+            super::super::workloads::workload_lifecycle_contract(rig, reference)
+        }
+        (Some(_), Some(_)) => Err(step_error(
+            rig,
+            "lifecycle step declares both `lifecycle` and `workload`; declare the contract inline or reference a workload, not both",
+        )),
+        (None, None) => Err(step_error(
+            rig,
+            "lifecycle step declares neither `lifecycle` nor `workload`; provide an inline `homeboy/lifecycle-contract/v1` payload or reference a workload that declares one",
+        )),
+    }
+}
+
 /// Stable ownership key for the handles a step captures.
-fn step_key(step_id: Option<&str>, component: Option<&str>) -> String {
+pub(crate) fn step_key(step_id: Option<&str>, component: Option<&str>) -> String {
     step_id
         .map(str::trim)
         .filter(|value| !value.is_empty())

crates/homeboy-rig/src/pipeline/mod.rs

@@ -28,6 +28,10 @@ use homeboy_core::error::Result;
 pub use outcome::{PipelineOutcome, PipelineStepOutcome};
 
 pub(super) use command_step::run_command_step;
+/// Ownership key a `lifecycle` step's captured handles are recorded under.
+/// Exported so `repair` classifies live handles with the same key the executor
+/// writes, instead of a second copy that can drift.
+pub(super) use lifecycle_step::step_key as lifecycle_step_key;
 
 pub fn run_pipeline(rig: &RigSpec, name: &str, fail_fast: bool) -> Result<PipelineOutcome> {
     run_pipeline_with_settings(rig, name, fail_fast, &[])
@@ -339,13 +343,15 @@ fn run_step(
             step_id,
             component,
             lifecycle,
+            workload,
             op,
             ..
         } => lifecycle_step::run_lifecycle_step(
             rig,
             step_id.as_deref(),
             component.as_deref(),
-            lifecycle,
+            lifecycle.as_ref(),
+            workload.as_ref(),
             *op,
             settings,
         ),