feat(store): record which process owns a run, and enumerate unsettled runs

src/store/file/document.rs

@@ -242,6 +242,13 @@ pub fn new_run_record(id: &str, workflow_id: &str, started_at: u64) -> RunRecord
         inputs: serde_json::Map::new(),
         trigger: None,
         origin: None,
+        // Stamped by the caller through `RunRecord::with_executor`. Left unset
+        // here so this factory stays usable in tests and tools that are not
+        // actually executing anything; an unstamped record is treated as
+        // unowned, which is the honest reading.
+        executor: None,
+        // Nobody has asked a run to stop before it has begun.
+        cancel_requested: false,
         // Both are evidence about a run that has ended, so a run that has only
         // just started has neither. They are filled in when it settles.
         summary: None,

src/store/file/store_impl.rs

@@ -179,6 +179,31 @@ impl FileWorkflowStore {
             .join(format!("{}.json", safe_component(run_id)?)))
     }
 
+    /// Every parsable run record in the scope's runs directory, unordered.
+    ///
+    /// A missing directory is an empty listing, not an error: a scope that has
+    /// never run anything has nothing to read. A record this build cannot parse
+    /// is skipped rather than failing the whole listing — history is
+    /// diagnostic, and one corrupt file should not hide the rest of it.
+    pub(super) fn read_run_dir(&self) -> Result<Vec<RunRecord>, WorkflowError> {
+        let entries = match std::fs::read_dir(&self.runs_dir) {
+            Ok(entries) => entries,
+            Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
+            Err(source) => {
+                return Err(WorkflowError::Io {
+                    path: self.runs_dir.clone(),
+                    source,
+                });
+            }
+        };
+        Ok(entries
+            .filter_map(|entry| entry.ok().map(|e| e.path()))
+            .filter(|path| is_json(path))
+            .filter_map(|path| std::fs::read(&path).ok())
+            .filter_map(|body| serde_json::from_slice::<RunRecord>(&body).ok())
+            .collect())
+    }
+
     /// Run one workflow definition mutation while holding its filesystem lock.
     pub(super) fn with_definition_lock<T>(
         &self,

src/store/file/workflow_store_impl.rs

@@ -135,31 +135,25 @@ impl WorkflowStore for FileWorkflowStore {
     }
 
     fn list_runs(&self, workflow_id: &str) -> Result<Vec<RunRecord>, WorkflowError> {
-        let entries = match std::fs::read_dir(&self.runs_dir) {
-            Ok(entries) => entries,
-            Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
-            Err(source) => {
-                return Err(WorkflowError::Io {
-                    path: self.runs_dir.clone(),
-                    source,
-                });
-            }
-        };
-
-        let mut runs: Vec<RunRecord> = entries
-            .filter_map(|entry| entry.ok().map(|e| e.path()))
-            .filter(|path| is_json(path))
-            // A run record this host cannot parse is skipped rather than
-            // failing the listing: history is diagnostic, and one corrupt file
-            // should not hide the rest of it.
-            .filter_map(|path| std::fs::read(&path).ok())
-            .filter_map(|body| serde_json::from_slice::<RunRecord>(&body).ok())
+        let mut runs: Vec<RunRecord> = self
+            .read_run_dir()?
+            .into_iter()
             .filter(|run| run.workflow_id == workflow_id)
             .collect();
         runs.sort_by_key(|run| std::cmp::Reverse(run.started_at));
         Ok(runs)
     }
 
+    fn unsettled_runs(&self) -> Result<Vec<RunRecord>, WorkflowError> {
+        let mut runs: Vec<RunRecord> = self
+            .read_run_dir()?
+            .into_iter()
+            .filter(|run| !run.status.is_settled())
+            .collect();
+        runs.sort_by_key(|run| std::cmp::Reverse(run.started_at));
+        Ok(runs)
+    }
+
     fn list_revisions(&self, workflow_id: &str) -> Result<Vec<WorkflowRevision>, WorkflowError> {
         // Releases before the source/state split kept undo snapshots beside
         // definitions. Merge that history with new workspace-scoped snapshots

src/store/mod.rs

@@ -40,9 +40,9 @@ pub use file::{safe_component, workspace_scope, write_atomic};
 
 pub use self::types::{
     Diagnosis, LEGACY_TRUNCATED_KEY, NoteId, NoteKind, NoteSource, ProposalId, ProposalStatus,
-    ProposalVerification, RunId, RunOrigin, RunRecord, RunStatus, RunStep, TRUNCATED_KEY,
-    TranscriptEntry, WorkflowDefaults, WorkflowError, WorkflowId, WorkflowNote, WorkflowProposal,
-    WorkflowRecord, WorkflowRevision, WorkflowSummary, is_truncated,
+    ProposalVerification, RunExecutor, RunId, RunOrigin, RunRecord, RunStatus, RunStep,
+    TRUNCATED_KEY, TranscriptEntry, WorkflowDefaults, WorkflowError, WorkflowId, WorkflowNote,
+    WorkflowProposal, WorkflowRecord, WorkflowRevision, WorkflowSummary, is_truncated,
 };
 
 /// An exclusive claim over proposal decisions for one workflow.
@@ -146,6 +146,20 @@ pub trait WorkflowStore: Send + Sync {
     /// Every recorded run for a workflow, newest first.
     fn list_runs(&self, workflow_id: &str) -> Result<Vec<RunRecord>, WorkflowError>;
 
+    /// Every recorded run that has not settled, across all workflows.
+    ///
+    /// What a reconciliation sweep reads — the caller that asks "which records
+    /// still claim to be live" is asking across every workflow in the scope, and
+    /// answering that through [`list_runs`](Self::list_runs) would re-read the
+    /// whole runs directory once per workflow.
+    ///
+    /// The default returns nothing, which makes a store that cannot enumerate
+    /// its runs a store a sweep leaves alone rather than one that fails to
+    /// compile.
+    fn unsettled_runs(&self) -> Result<Vec<RunRecord>, WorkflowError> {
+        Ok(Vec::new())
+    }
+
     /// Every superseded copy of a workflow, newest first.
     ///
     /// A workflow that has never been written over has no revisions, which is

src/store/tests/mod.rs

@@ -31,7 +31,9 @@ pub(super) use super::file::{
 pub(super) use super::{
     FileWorkflowStore, WorkflowStore, require, require_run, rollback, undo_last,
 };
-pub(super) use crate::store::types::{RunStatus, WorkflowDefaults, WorkflowError, WorkflowRecord};
+pub(super) use crate::store::types::{
+    RunExecutor, RunRecord, RunStatus, WorkflowDefaults, WorkflowError, WorkflowRecord,
+};
 
 /// A store rooted in a temporary directory, with definitions and runs kept
 /// apart the way the discovered layout keeps them.

src/store/tests/runs.rs

@@ -49,3 +49,84 @@ fn asking_for_a_run_that_was_never_recorded_is_an_error_not_a_silent_none() {
     let err = require_run(&store_in(root.path()), "ghost").expect_err("no such run");
     assert!(matches!(err, WorkflowError::RunNotFound(_)), "got {err:?}");
 }
+
+#[test]
+fn unsettled_runs_spans_every_workflow_and_skips_finished_ones() {
+    // What a reconciliation sweep reads. It has to cross workflow boundaries —
+    // "which records still claim to be live" is a question about the whole
+    // scope — and it must not hand back runs that already have an outcome.
+    let root = tempfile::tempdir().unwrap();
+    let store = store_in(root.path());
+
+    store
+        .record_run(&new_run_record("live-alpha", "alpha", 100))
+        .unwrap();
+    store
+        .record_run(&new_run_record("live-beta", "beta", 300))
+        .unwrap();
+    let mut parked = new_run_record("parked", "alpha", 200);
+    parked.status = RunStatus::PendingApproval;
+    store.record_run(&parked).unwrap();
+    for (id, status) in [
+        ("done", RunStatus::Succeeded),
+        ("broke", RunStatus::Failed),
+        ("stopped", RunStatus::Cancelled),
+        ("cut-off", RunStatus::Interrupted),
+    ] {
+        let mut record = new_run_record(id, "alpha", 50);
+        record.status = status;
+        store.record_run(&record).unwrap();
+    }
+
+    let unsettled = store.unsettled_runs().unwrap();
+    let ids: Vec<&str> = unsettled.iter().map(|r| r.id.as_str()).collect();
+
+    assert_eq!(
+        ids,
+        vec!["live-beta", "parked", "live-alpha"],
+        "newest first, both workflows, nothing settled"
+    );
+}
+
+#[test]
+fn a_scope_that_has_never_run_anything_has_no_unsettled_runs() {
+    let root = tempfile::tempdir().unwrap();
+    let store = store_in(root.path());
+
+    assert!(store.unsettled_runs().unwrap().is_empty());
+}
+
+#[test]
+fn an_executor_and_a_cancel_request_survive_a_write_and_read() {
+    let root = tempfile::tempdir().unwrap();
+    let store = store_in(root.path());
+    let mut record = new_run_record("owned", "alpha", 100).with_executor(Some(RunExecutor {
+        host: "somewhere".to_string(),
+        pid: 4711,
+        started_at_secs: Some(1_700_000_000),
+    }));
+    record.cancel_requested = true;
+
+    store.record_run(&record).unwrap();
+
+    let read = store.get_run("owned").unwrap().unwrap();
+    assert_eq!(read.executor, record.executor);
+    assert!(read.cancel_requested);
+}
+
+#[test]
+fn a_record_written_before_executors_existed_still_parses() {
+    // Both fields are additive, so an older run file must keep loading — and
+    // must read as unowned, which is what makes a sweep treat it as an orphan.
+    let older = serde_json::json!({
+        "id": "legacy",
+        "workflowId": "alpha",
+        "status": "running",
+        "startedAt": 100,
+    });
+
+    let record: RunRecord = serde_json::from_value(older).expect("older records still parse");
+
+    assert!(record.executor.is_none());
+    assert!(!record.cancel_requested);
+}

src/store/types/mod.rs

@@ -46,8 +46,8 @@ pub use proposal::{
 };
 
 pub use run::{
-    LEGACY_TRUNCATED_KEY, RunId, RunOrigin, RunRecord, RunStatus, RunStep, TRUNCATED_KEY,
-    bounded_evidence, bounded_within, is_truncated,
+    LEGACY_TRUNCATED_KEY, RunExecutor, RunId, RunOrigin, RunRecord, RunStatus, RunStep,
+    TRUNCATED_KEY, bounded_evidence, bounded_within, is_truncated,
 };
 pub use transcript::TranscriptEntry;
 pub use workflow::{

src/store/types/run.rs

@@ -109,6 +109,39 @@ impl RunOrigin {
     }
 }
 
+/// The OS process that is executing a run.
+///
+/// A run record on its own cannot say whether the process that wrote it is
+/// still alive, which is the difference between "another host is working on
+/// this" and "this row is a tombstone from a process that was killed". Stamping
+/// the executor turns that unanswerable question into a liveness check.
+///
+/// Both halves of the identity matter. A bare pid is not enough: pids are
+/// recycled, so a run left behind by pid 4711 would look alive the moment
+/// something unrelated is assigned 4711. `started_at_secs` — the executing
+/// process's own start time — makes the pair unique for as long as the record
+/// is worth reading.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct RunExecutor {
+    /// The host the process runs on.
+    ///
+    /// The run store is per-machine today, but a record naming a pid with no
+    /// host is one shared filesystem away from a liveness check that compares
+    /// this machine's process table against another machine's pid. Recording it
+    /// costs nothing and makes the check refuse rather than guess.
+    pub host: String,
+    /// The executing process's id.
+    pub pid: u32,
+    /// When that process itself started, in seconds since the epoch.
+    ///
+    /// The pid-reuse guard. Absent when the platform would not report it, in
+    /// which case a liveness check falls back to the pid alone and errs toward
+    /// leaving the record alone.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub started_at_secs: Option<u64>,
+}
+
 /// Where a run got to.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "snake_case")]
@@ -220,6 +253,28 @@ pub struct RunRecord {
     /// caller could not say anything about itself.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub origin: Option<RunOrigin>,
+    /// The process executing this run, while one is.
+    ///
+    /// Written when the run is admitted and left in place when it settles: a
+    /// finished run's executor is history, not a claim. What reads it is a
+    /// reconciliation sweep, which only consults it for a record that still
+    /// says [`RunStatus::Running`].
+    ///
+    /// Absent on records written before this field existed, which a sweep
+    /// treats as unowned — an old record claiming to run is exactly the kind
+    /// this exists to clean up.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub executor: Option<RunExecutor>,
+    /// Whether someone has asked this run to stop.
+    ///
+    /// The durable half of cancellation. An in-memory cancellation registry can
+    /// only reach a run its own process is executing, so a cancel aimed at a run
+    /// owned by another live process is written here instead; that process
+    /// notices the flag on its next poll and cancels itself. Never cleared — a
+    /// run that was asked to stop and then settled should still say it was
+    /// asked.
+    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
+    pub cancel_requested: bool,
     /// Steps in completion order.
     #[serde(default)]
     pub steps: Vec<RunStep>,
@@ -283,6 +338,12 @@ impl RunRecord {
         self
     }
 
+    /// Record the process executing this run.
+    pub fn with_executor(mut self, executor: Option<RunExecutor>) -> Self {
+        self.executor = executor;
+        self
+    }
+
     /// How long the run took, once it has settled.
     pub fn duration_ms(&self) -> Option<u64> {
         self.finished_at

src/store/types/types_tests.rs

@@ -93,6 +93,8 @@ fn run_records_use_camel_case_on_the_wire() {
         inputs: Default::default(),
         trigger: None,
         origin: None,
+        executor: None,
+        cancel_requested: false,
         summary: None,
         diagnosis: None,
     })
@@ -175,6 +177,8 @@ fn run_evidence_is_omitted_from_the_wire_when_absent() {
         inputs: Default::default(),
         trigger: None,
         origin: None,
+        executor: None,
+        cancel_requested: false,
         summary: None,
         diagnosis: None,
     })
@@ -201,6 +205,8 @@ fn what_a_run_was_started_with_survives_the_wire() {
         inputs: Default::default(),
         trigger: None,
         origin: None,
+        executor: None,
+        cancel_requested: false,
         summary: None,
         diagnosis: None,
     }