fix(durable): reclaim crash-orphaned running executions via staleness sweep

CHANGELOG.md

@@ -15,6 +15,22 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ### Added
 
+- **Durable execution**: added a crash-orphan sweep to the durable retention loop (`Journal::sweep_orphans`,
+  #6254) that reclaims `status='running'` executions whose owner process died without finalizing —
+  previously invisible to the TTL prune (which only ever considers `finalized_at IS NOT NULL` rows) and
+  stuck `running` forever. A `status='running'` row whose `updated_at` is older than the new
+  `[durable.retention] stale_running_after_secs` (default 3600s, `0` disables the sweep) becomes a
+  candidate; it is hard-aborted only after a non-blocking try-acquire of its INV-15 advisory
+  `ExecutionLock` succeeds — a live owner (`ExecutionLocked`) short-circuits to skip, since staleness of
+  `updated_at` alone never proves the owner is dead. The sweep runs before `prune()` on every retention
+  tick (same supervised loop, no new spawn site) and is a documented no-op (warn-once) on backends
+  without an on-disk lock directory (`:memory:`, Postgres, non-Unix). `zeph durable prune` now runs the
+  sweep before the TTL prune and `--dry-run` reports both counts separately.
+- **Durable execution**: `zeph-orchestration`'s `journal_budget` (P2) and `zeph-scheduler`'s
+  `fire_with_durable` (P3) now open their execution via `open_execution_exclusive` instead of the plain
+  `open_execution`, making their `DagRun`/`ScheduledJob` rows' liveness observable to the crash-orphan
+  sweep; on `DurableError::ExecutionLocked` both adapters log and return `Ok(())` — a graceful skip, never
+  a task failure or retry.
 - **TUI**: added a read-only settings view (`S` key or the `settings` command-palette
   entry) listing configured LLM providers, MCP servers, and sub-agent definitions in
   three tabs, sourced as a live snapshot from `MetricsSnapshot` (never re-parsed from
@@ -290,6 +306,13 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
   bug in replan execution — surfaced by the new end-to-end test added for #6287's review pass.
   Gap-task IDs are now remapped to local 0-based IDs for the partial scheduler run and back to the
   original global IDs on the way out.
+- **Durable execution**: `open_execution`/`open_execution_exclusive`'s reopen path un-finalized only
+  `completed`/`failed` rows, leaving `aborted` rows untouched on reopen (INV-16). This was safe before
+  #6254 because `aborted` was a rare, immediately-redriven outcome, but became a hazard once the new
+  crash-orphan sweep makes `aborted` the common outcome of a resumable crash: a resumed execution whose
+  row kept `finalized_at` set was prunable out from under the active resume — the exact hazard the
+  completed/failed un-finalize was built to prevent. Reopening now resets `status='running'` and clears
+  `finalized_at` for a row in ANY terminal status.
 - **Worktree**: `--bare` silently skipped the entire worktree subsystem bootstrap
   (`WorktreeManager` construction, `probe_capabilities`) with no warning when
   `worktree.enabled = true` in the active config — the 6th confirmed instance of the `--bare`

config/default.toml

@@ -1765,6 +1765,10 @@ max_journal_bytes = 1073741824
 prune_batch_size = 500
 # Background prune poll interval (seconds).
 prune_interval_secs = 3600
+# Crash-orphan threshold (#6254): a running execution whose owner process died without
+# finalizing becomes a sweep candidate after this many seconds of inactivity, subject to an
+# advisory-lock liveness check before it is aborted. 0 disables the sweep.
+stale_running_after_secs = 3600
 
 [caveman]
 # Start every session in ultra-compressed (telegraphic) output mode.

crates/zeph-config/src/durable.rs

@@ -149,6 +149,10 @@ pub struct RetentionPolicy {
     pub prune_batch_size: u64,
     /// Background prune poll interval, in seconds.
     pub prune_interval_secs: u64,
+    /// Crash-orphan threshold, in seconds (#6254): a `status='running'` row whose `updated_at`
+    /// is older than this becomes a sweep candidate, subject to an INV-15 flock liveness check
+    /// before it is aborted. `0` disables the sweep entirely.
+    pub stale_running_after_secs: u64,
 }
 
 impl Default for RetentionPolicy {
@@ -160,6 +164,7 @@ impl Default for RetentionPolicy {
             max_journal_bytes: 1_073_741_824,
             prune_batch_size: 500,
             prune_interval_secs: 3600,
+            stale_running_after_secs: 3600,
         }
     }
 }
@@ -196,6 +201,7 @@ mod tests {
         assert_eq!(cfg.retention.max_journal_bytes, 1_073_741_824);
         assert_eq!(cfg.retention.prune_batch_size, 500);
         assert_eq!(cfg.retention.prune_interval_secs, 3600);
+        assert_eq!(cfg.retention.stale_running_after_secs, 3600);
     }
 
     #[test]

crates/zeph-config/src/migrate/infra.rs

@@ -668,7 +668,8 @@ pub fn migrate_durable_config(toml_src: &str) -> Result<MigrationResult, Migrate
          # max_executions = 10000\n\
          # max_journal_bytes = 1073741824\n\
          # prune_batch_size = 500\n\
-         # prune_interval_secs = 3600\n";
+         # prune_interval_secs = 3600\n\
+         # stale_running_after_secs = 3600  # crash-orphan threshold (#6254); 0 disables the sweep\n";
     let output = format!("{}{}", toml_src.trim_end(), block);
     Ok(MigrationResult {
         output,
@@ -778,6 +779,73 @@ pub fn migrate_durable_shared_db(toml_src: &str) -> Result<MigrationResult, Migr
     })
 }
 
+/// Adds a commented `# stale_running_after_secs = 3600` advisory line to an existing active
+/// `[durable.retention]` table that predates the crash-orphan sweep (#6254) and lacks the field.
+///
+/// Purely additive with the spec default (`3600`, matching the runtime `#[serde(default)]`
+/// value): a config that never runs this migration step still loads with the correct default at
+/// runtime, so this is a self-documentation convenience, not a correctness requirement. No-op
+/// when `[durable.retention]` is absent (a fresh `[durable]` migration via
+/// [`migrate_durable_config`] already includes the field in its commented block) or
+/// `stale_running_after_secs` (active or commented) is already present.
+///
+/// # Errors
+///
+/// Returns [`MigrateError`] if the source is not valid TOML.
+pub fn migrate_durable_stale_running_after_secs(
+    toml_src: &str,
+) -> Result<MigrationResult, MigrateError> {
+    // Anchored multiline pattern: matches `[durable.retention]` with optional inline comment,
+    // followed by LF or CRLF.
+    static DURABLE_RETENTION_HEADER_RE: std::sync::LazyLock<Regex> =
+        std::sync::LazyLock::new(|| {
+            Regex::new(r"(?m)^[ \t]*\[durable\.retention\][ \t]*(?:#[^\r\n]*)?\r?\n")
+                .expect("static pattern")
+        });
+
+    if !section_header_present(toml_src, "durable.retention") {
+        return Ok(MigrationResult {
+            output: toml_src.to_owned(),
+            changed_count: 0,
+            sections_changed: Vec::new(),
+        });
+    }
+
+    let already_present = toml_src.lines().any(|l| {
+        l.trim()
+            .trim_start_matches('#')
+            .trim()
+            .starts_with("stale_running_after_secs")
+    });
+    if already_present || !DURABLE_RETENTION_HEADER_RE.is_match(toml_src) {
+        return Ok(MigrationResult {
+            output: toml_src.to_owned(),
+            changed_count: 0,
+            sections_changed: Vec::new(),
+        });
+    }
+
+    let comment = "# stale_running_after_secs = 3600  # crash-orphan threshold (#6254); 0 disables \
+        the sweep\n";
+    let output = DURABLE_RETENTION_HEADER_RE
+        .replacen(toml_src, 1, |caps: &regex::Captures| {
+            format!("{}{comment}", &caps[0])
+        })
+        .into_owned();
+
+    let changed = output != toml_src;
+    let changed_count = usize::from(changed);
+    Ok(MigrationResult {
+        output,
+        changed_count,
+        sections_changed: if changed {
+            vec!["durable.retention.stale_running_after_secs".to_owned()]
+        } else {
+            Vec::new()
+        },
+    })
+}
+
 /// Adds a commented-out `[security.content_isolation.nli]` section to configs that predate the
 /// SONAR NLI entailment check stage (#5438). Idempotent: no-op when the real or commented
 /// `[security.content_isolation.nli]` header is already present, so running `--migrate-config`

crates/zeph-config/src/migrate/mod.rs

@@ -601,33 +601,34 @@ use steps::{
     MigrateAcpSubagentsConfig, MigrateAgentBudgetHint, MigrateAgentRetryToToolsRetry,
     MigrateAutodreamConfig, MigrateCavemanConfig, MigrateCocoonProviderNotice,
     MigrateCocoonShowBalance, MigrateCompressionPredictorConfig, MigrateDatabaseUrl,
-    MigrateDeepLinkConfig, MigrateDurableConfig, MigrateDurableSharedDb, MigrateEgressConfig,
-    MigrateEmbedProviderRename, MigrateEvalModelToProvider, MigrateFidelityTimeoutDefaults,
-    MigrateFiveSignalConfig, MigrateFocusAutoConsolidateMinWindow, MigrateForgettingConfig,
-    MigrateGoalsConfig, MigrateGonkagateToGonka, MigrateHooksPermissionDeniedConfig,
-    MigrateHooksTurnComplete, MigrateKnowledgeConfig, MigrateLlmStreamLimits,
-    MigrateMagicDocsConfig, MigrateMcpElicitationConfig, MigrateMcpMaxConnectAttempts,
-    MigrateMcpRetryAndToolTimeout, MigrateMcpTrustLevels, MigrateMemoryGraph,
-    MigrateMemoryGraphRecallIncludeImported, MigrateMemoryHebbian,
-    MigrateMemoryHebbianConsolidation, MigrateMemoryHebbianSpread, MigrateMemoryPersonaConfig,
-    MigrateMemoryReasoning, MigrateMemoryReasoningJudge, MigrateMemoryRetrieval,
-    MigrateMemoryRetrievalQueryBias, MigrateMemoryTypeAwareCompose, MigrateMicrocompactConfig,
-    MigrateNliConfig, MigrateOrchestrationAssetSensitivity, MigrateOrchestrationEnsemble,
-    MigrateOrchestrationPersistence, MigrateOrchestratorProvider, MigrateOtelFilter,
-    MigratePiiFilterNames, MigratePlannerModelToProvider, MigratePolicyProviderAndUtilityWindow,
-    MigrateProviderMaxConcurrent, MigrateQdrantApiKey, MigrateQdrantTimeoutSecs,
-    MigrateQualityConfig, MigrateSandboxConfig, MigrateSandboxEgressFilter, MigrateSchedulerDaemon,
-    MigrateSecretMaskingConfig, MigrateServeConfig, MigrateSessionPersistProviderOverrides,
-    MigrateSessionPersistenceConfig, MigrateSessionProviderPersistence, MigrateSessionRecapConfig,
-    MigrateShadowSentinelConfig, MigrateShellCheckpointsConfig, MigrateShellTransactional,
-    MigrateSkillTrustRequireCheck, MigrateSkillsRegistry, MigrateSttToProvider,
-    MigrateSupervisorConfig, MigrateTelemetryConfig, MigrateToolsCompressionConfig,
-    MigrateTraceMetadata, MigrateTuiDelights, MigrateTuiMouse, MigrateTuiThemeConfig,
-    MigrateTuiThemeDefaults, MigrateUtilityHighGainTools, MigrateVigilConfig,
-    MigrateWorktreeConfig, MigrateWorktreeGitTimeout, MigrateWorktreeQuotaFields,
+    MigrateDeepLinkConfig, MigrateDurableConfig, MigrateDurableSharedDb,
+    MigrateDurableStaleRunningAfterSecs, MigrateEgressConfig, MigrateEmbedProviderRename,
+    MigrateEvalModelToProvider, MigrateFidelityTimeoutDefaults, MigrateFiveSignalConfig,
+    MigrateFocusAutoConsolidateMinWindow, MigrateForgettingConfig, MigrateGoalsConfig,
+    MigrateGonkagateToGonka, MigrateHooksPermissionDeniedConfig, MigrateHooksTurnComplete,
+    MigrateKnowledgeConfig, MigrateLlmStreamLimits, MigrateMagicDocsConfig,
+    MigrateMcpElicitationConfig, MigrateMcpMaxConnectAttempts, MigrateMcpRetryAndToolTimeout,
+    MigrateMcpTrustLevels, MigrateMemoryGraph, MigrateMemoryGraphRecallIncludeImported,
+    MigrateMemoryHebbian, MigrateMemoryHebbianConsolidation, MigrateMemoryHebbianSpread,
+    MigrateMemoryPersonaConfig, MigrateMemoryReasoning, MigrateMemoryReasoningJudge,
+    MigrateMemoryRetrieval, MigrateMemoryRetrievalQueryBias, MigrateMemoryTypeAwareCompose,
+    MigrateMicrocompactConfig, MigrateNliConfig, MigrateOrchestrationAssetSensitivity,
+    MigrateOrchestrationEnsemble, MigrateOrchestrationPersistence, MigrateOrchestratorProvider,
+    MigrateOtelFilter, MigratePiiFilterNames, MigratePlannerModelToProvider,
+    MigratePolicyProviderAndUtilityWindow, MigrateProviderMaxConcurrent, MigrateQdrantApiKey,
+    MigrateQdrantTimeoutSecs, MigrateQualityConfig, MigrateSandboxConfig,
+    MigrateSandboxEgressFilter, MigrateSchedulerDaemon, MigrateSecretMaskingConfig,
+    MigrateServeConfig, MigrateSessionPersistProviderOverrides, MigrateSessionPersistenceConfig,
+    MigrateSessionProviderPersistence, MigrateSessionRecapConfig, MigrateShadowSentinelConfig,
+    MigrateShellCheckpointsConfig, MigrateShellTransactional, MigrateSkillTrustRequireCheck,
+    MigrateSkillsRegistry, MigrateSttToProvider, MigrateSupervisorConfig, MigrateTelemetryConfig,
+    MigrateToolsCompressionConfig, MigrateTraceMetadata, MigrateTuiDelights, MigrateTuiMouse,
+    MigrateTuiThemeConfig, MigrateTuiThemeDefaults, MigrateUtilityHighGainTools,
+    MigrateVigilConfig, MigrateWorktreeConfig, MigrateWorktreeGitTimeout,
+    MigrateWorktreeQuotaFields,
 };
 
-/// Ordered registry of all sequential migration steps (steps 1–86).
+/// Ordered registry of all sequential migration steps (steps 1–87).
 ///
 /// Each entry wraps the corresponding free function and is evaluated lazily at first access.
 /// The ordering is chronological; the dispatch loop in `src/commands/migrate.rs` iterates
@@ -789,6 +790,9 @@ pub static MIGRATIONS: std::sync::LazyLock<Vec<Box<dyn Migration + Send + Sync>>
             // Step 86 — add [orchestration.ensemble] advisory block for ORCH-style
             // deterministic verifier ensemble-merge (spec 073, #6232)
             Box::new(MigrateOrchestrationEnsemble),
+            // Step 87 — add stale_running_after_secs advisory to an existing active
+            // [durable.retention] table for the crash-orphan sweep (spec-064, #6254)
+            Box::new(MigrateDurableStaleRunningAfterSecs),
         ]
     });
 

crates/zeph-config/src/migrate/steps.rs

@@ -53,7 +53,9 @@
 //! step 85 adds a commented `[memory.type_aware_compose]` advisory block for `MemGuard`
 //! type-aware retrieval composition (spec 064, #6086);
 //! step 86 adds a commented `[orchestration.ensemble]` advisory block for ORCH-style
-//! deterministic verifier ensemble-merge (spec 073, #6232).
+//! deterministic verifier ensemble-merge (spec 073, #6232);
+//! step 87 adds a commented `stale_running_after_secs = 3600` advisory to an existing active
+//! `[durable.retention]` table for the crash-orphan sweep (spec-064, #6254).
 //!
 //! Each struct is a zero-size type that delegates to the corresponding free function in
 //! `super`. They exist solely to satisfy the object-safe [`super::Migration`] trait so the
@@ -66,8 +68,8 @@ use super::{
     migrate_autodream_config, migrate_caveman_config, migrate_cocoon_provider_notice,
     migrate_cocoon_show_balance, migrate_compression_predictor_config, migrate_database_url,
     migrate_deep_link_config, migrate_durable_config, migrate_durable_shared_db,
-    migrate_egress_config, migrate_embed_provider_rename, migrate_eval_model_to_provider,
-    migrate_fidelity_timeout_defaults, migrate_five_signal_config,
+    migrate_durable_stale_running_after_secs, migrate_egress_config, migrate_embed_provider_rename,
+    migrate_eval_model_to_provider, migrate_fidelity_timeout_defaults, migrate_five_signal_config,
     migrate_focus_auto_consolidate_min_window, migrate_forgetting_config, migrate_goals_config,
     migrate_hooks_permission_denied_config, migrate_hooks_turn_complete_config,
     migrate_knowledge_config, migrate_llm_stream_limits, migrate_magic_docs_config,
@@ -1073,3 +1075,16 @@ impl Migration for MigrateOrchestrationEnsemble {
         migrate_orchestration_ensemble(toml_src)
     }
 }
+
+/// Step 87 — adds a commented `stale_running_after_secs = 3600` advisory to an existing active
+/// `[durable.retention]` table that predates the crash-orphan sweep (spec-064, #6254).
+pub(super) struct MigrateDurableStaleRunningAfterSecs;
+impl Migration for MigrateDurableStaleRunningAfterSecs {
+    fn name(&self) -> &'static str {
+        "migrate_durable_stale_running_after_secs"
+    }
+
+    fn apply(&self, toml_src: &str) -> Result<MigrationResult, MigrateError> {
+        migrate_durable_stale_running_after_secs(toml_src)
+    }
+}

crates/zeph-config/src/migrate/tests.rs

@@ -9,8 +9,8 @@ use super::*;
 fn migrations_registry_has_all_steps() {
     assert_eq!(
         MIGRATIONS.len(),
-        86,
-        "MIGRATIONS registry must contain all 86 sequential steps"
+        87,
+        "MIGRATIONS registry must contain all 87 sequential steps"
     );
     for m in MIGRATIONS.iter() {
         assert!(
@@ -1817,7 +1817,7 @@ fn migrate_focus_auto_consolidate_noop_when_only_commented_section() {
 
 #[test]
 fn registry_has_fifty_entries() {
-    assert_eq!(MIGRATIONS.len(), 86);
+    assert_eq!(MIGRATIONS.len(), 87);
 }
 
 #[test]
@@ -1855,7 +1855,7 @@ fn registry_is_idempotent_on_empty_input() {
 
 #[test]
 fn registry_preserves_order_matches_dispatch() {
-    // Names must follow the documented step order (steps 1–84).
+    // Names must follow the documented step order (steps 1–87).
     let expected = [
         "migrate_stt_to_provider",
         "migrate_planner_model_to_provider",
@@ -1943,6 +1943,7 @@ fn registry_preserves_order_matches_dispatch() {
         "migrate_a2a_server_remove_inert_fields",
         "migrate_memory_type_aware_compose_config",
         "migrate_orchestration_ensemble",
+        "migrate_durable_stale_running_after_secs",
     ];
     let actual: Vec<&str> = MIGRATIONS.iter().map(|m| m.name()).collect();
     assert_eq!(actual, expected);
@@ -4142,6 +4143,67 @@ fn step_79_still_adds_advisory_when_unsafe_topology_detected() {
     );
 }
 
+// ── migrate_durable_stale_running_after_secs tests (step 87, #6254) ──────
+
+#[test]
+fn step_87_adds_commented_advisory_when_retention_active_and_missing_field() {
+    let src = "[durable.retention]\nttl_completed_secs = 604800\n";
+    let result = migrate_durable_stale_running_after_secs(src).expect("migrate");
+    assert_eq!(result.changed_count, 1);
+    assert!(result.output.contains("# stale_running_after_secs = 3600"));
+    assert!(!result.output.contains("\nstale_running_after_secs ="));
+    assert!(result.output.contains("ttl_completed_secs = 604800"));
+    assert_eq!(
+        result.sections_changed,
+        vec!["durable.retention.stale_running_after_secs".to_owned()]
+    );
+}
+
+#[test]
+fn step_87_noop_when_field_already_present() {
+    let src = "[durable.retention]\nstale_running_after_secs = 7200\n";
+    let result = migrate_durable_stale_running_after_secs(src).expect("migrate");
+    assert_eq!(result.changed_count, 0);
+    assert_eq!(result.output, src);
+}
+
+#[test]
+fn step_87_noop_when_field_comment_already_present() {
+    let src = "[durable.retention]\n# stale_running_after_secs = 3600\n";
+    let result = migrate_durable_stale_running_after_secs(src).expect("migrate");
+    assert_eq!(result.changed_count, 0);
+    assert_eq!(result.output, src);
+}
+
+#[test]
+fn step_87_noop_when_durable_retention_section_absent() {
+    let src = "[durable]\nenabled = true\n";
+    let result = migrate_durable_stale_running_after_secs(src).expect("migrate");
+    assert_eq!(result.changed_count, 0);
+    assert_eq!(result.output, src);
+}
+
+#[test]
+fn step_87_noop_when_durable_retention_only_commented_advisory() {
+    let src = "# [durable.retention]\n# ttl_completed_secs = 604800\n";
+    let result = migrate_durable_stale_running_after_secs(src).expect("migrate");
+    assert_eq!(result.changed_count, 0);
+    assert_eq!(result.output, src);
+}
+
+#[test]
+fn step_87_idempotent_on_own_output() {
+    let src = "[durable.retention]\nttl_completed_secs = 604800\n";
+    let first = migrate_durable_stale_running_after_secs(src).expect("first migrate");
+    assert_eq!(first.changed_count, 1);
+    let second = migrate_durable_stale_running_after_secs(&first.output).expect("second migrate");
+    assert_eq!(second.changed_count, 0, "second run must be a no-op");
+    assert_eq!(
+        second.output, first.output,
+        "output unchanged on second run"
+    );
+}
+
 // ── is_unsafe_shared_topology tests (#6042) ───────────────────────────────
 
 #[test]