From 10ee6a998a533ee3bd42fabc90cf97de479cdc35 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Mon, 24 Aug 2026 05:59:30 +0300 Subject: [PATCH 01/10] feat(store): add run executor tracking and cancellation support Introduce a `RunExecutor` struct to record the host, pid, and process start time of the process executing a run, enabling liveness checks that distinguish active runs from tombstones left by killed processes. Also add a `cancel_requested` flag to `RunRecord` for durable cancellation that works across process boundaries, along with a builder method for setting the executor. Auto-committed-on: dragonfly Co-authored-by: Medulla --- src/store/types/run.rs | 61 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 61 insertions(+) diff --git a/src/store/types/run.rs b/src/store/types/run.rs index fdf46ae..f12a31b 100644 --- a/src/store/types/run.rs +++ b/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, +} + /// 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, + /// 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, + /// 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, @@ -283,6 +338,12 @@ impl RunRecord { self } + /// Record the process executing this run. + pub fn with_executor(mut self, executor: Option) -> Self { + self.executor = executor; + self + } + /// How long the run took, once it has settled. pub fn duration_ms(&self) -> Option { self.finished_at From 756e67c1effb6028f4d4c6aa45d8a3c3f68da4db Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Mon, 24 Aug 2026 05:59:41 +0300 Subject: [PATCH 02/10] feat(store): export RunExecutor from run module The `RunExecutor` type is now publicly re-exported from the store types module so that it can be used by external consumers of the crate. Auto-committed-on: dragonfly Co-authored-by: Medulla --- src/store/types/mod.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/store/types/mod.rs b/src/store/types/mod.rs index 47c53ea..9be746f 100644 --- a/src/store/types/mod.rs +++ b/src/store/types/mod.rs @@ -46,7 +46,8 @@ pub use proposal::{ }; pub use run::{ - LEGACY_TRUNCATED_KEY, RunId, RunOrigin, RunRecord, RunStatus, RunStep, TRUNCATED_KEY, + LEGACY_TRUNCATED_KEY, RunExecutor, RunId, RunOrigin, RunRecord, RunStatus, RunStep, + TRUNCATED_KEY, bounded_evidence, bounded_within, is_truncated, }; pub use transcript::TranscriptEntry; From 9e1dbcaa41f89d7c8941fc40fdba638b4ed57b8c Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Mon, 24 Aug 2026 05:59:55 +0300 Subject: [PATCH 03/10] feat(store): add unsettled_runs method to WorkflowStore Add a new method `unsettled_runs` to the `WorkflowStore` trait that returns every recorded run that has not settled, across all workflows. This enables a reconciliation sweep to query which records still claim to be live without re-reading the runs directory once per workflow. The default implementation returns an empty vector, allowing stores that cannot enumerate their runs to be left alone rather than failing to compile. Auto-committed-on: dragonfly Co-authored-by: Medulla --- src/store/mod.rs | 17 ++++++++++++++++- 1 file changed, 16 insertions(+), 1 deletion(-) diff --git a/src/store/mod.rs b/src/store/mod.rs index 5bfd9d5..9b8e1d8 100644 --- a/src/store/mod.rs +++ b/src/store/mod.rs @@ -40,7 +40,8 @@ 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, + ProposalVerification, RunExecutor, RunId, RunOrigin, RunRecord, RunStatus, RunStep, + TRUNCATED_KEY, TranscriptEntry, WorkflowDefaults, WorkflowError, WorkflowId, WorkflowNote, WorkflowProposal, WorkflowRecord, WorkflowRevision, WorkflowSummary, is_truncated, }; @@ -146,6 +147,20 @@ pub trait WorkflowStore: Send + Sync { /// Every recorded run for a workflow, newest first. fn list_runs(&self, workflow_id: &str) -> Result, 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, 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 From d9ac87b7a266144f9d55bc39ce0f13a48adbd344 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Mon, 24 Aug 2026 06:00:14 +0300 Subject: [PATCH 04/10] refactor(store): extract run directory reading into shared method Extract the common logic for reading and parsing run records from the run directory into a private helper method, then reuse it in both `list_runs` and the new `unsettled_runs` to eliminate duplication and ensure consistent error handling. Auto-committed-on: dragonfly Co-authored-by: Medulla --- src/store/file/workflow_store_impl.rs | 32 +++++++++++---------------- 1 file changed, 13 insertions(+), 19 deletions(-) diff --git a/src/store/file/workflow_store_impl.rs b/src/store/file/workflow_store_impl.rs index e4ff481..8abf079 100644 --- a/src/store/file/workflow_store_impl.rs +++ b/src/store/file/workflow_store_impl.rs @@ -135,31 +135,25 @@ impl WorkflowStore for FileWorkflowStore { } fn list_runs(&self, workflow_id: &str) -> Result, 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 = 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::(&body).ok()) + let mut runs: Vec = 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, WorkflowError> { + let mut runs: Vec = 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, WorkflowError> { // Releases before the source/state split kept undo snapshots beside // definitions. Merge that history with new workspace-scoped snapshots From 965f98afc4e7988bfe79c70702b833d115f390b7 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Mon, 24 Aug 2026 06:00:33 +0300 Subject: [PATCH 05/10] feat(store): add read_run_dir method to list run records Add a method to read all parsable run records from a scope's runs directory, returning an empty list when the directory does not exist. The implementation skips unparseable files rather than failing, so that a single corrupt record does not hide the rest of the run history. Auto-committed-on: dragonfly Co-authored-by: Medulla --- src/store/file/store_impl.rs | 25 +++++++++++++++++++++++++ 1 file changed, 25 insertions(+) diff --git a/src/store/file/store_impl.rs b/src/store/file/store_impl.rs index a086729..0012411 100644 --- a/src/store/file/store_impl.rs +++ b/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, 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::(&body).ok()) + .collect()) + } + /// Run one workflow definition mutation while holding its filesystem lock. pub(super) fn with_definition_lock( &self, From c1df5ecbad6622a10db10f1e8806b77f6076fa1e Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Mon, 24 Aug 2026 06:01:09 +0300 Subject: [PATCH 06/10] fix(store): initialise executor and cancel fields in new_run_record Set executor to None and cancel_requested to false in the factory function so that newly created run records are properly unowned and not prematurely cancelled, keeping the factory usable in tests and tools that do not execute runs. Auto-committed-on: dragonfly Co-authored-by: Medulla --- src/store/file/document.rs | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/src/store/file/document.rs b/src/store/file/document.rs index 569f670..6d0fa60 100644 --- a/src/store/file/document.rs +++ b/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, From 8204ddeebb2fd3384ad879a1bf8f03e624fe195c Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Mon, 24 Aug 2026 06:01:32 +0300 Subject: [PATCH 07/10] test(types): add executor and cancel_requested fields to test structs Add the missing `executor` and `cancel_requested` fields to the test structs in three test functions to match the updated data model, ensuring the tests remain valid after the struct was extended. Auto-committed-on: dragonfly Co-authored-by: Medulla --- src/store/types/types_tests.rs | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/src/store/types/types_tests.rs b/src/store/types/types_tests.rs index a337ee2..33f0d8c 100644 --- a/src/store/types/types_tests.rs +++ b/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, } From 839d0468ef32b3cfb7af67e632d166af4a00cab3 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Mon, 24 Aug 2026 06:02:06 +0300 Subject: [PATCH 08/10] chore: files changed src/store/tests/runs.rs Auto-committed-on: dragonfly Co-authored-by: Medulla --- src/store/tests/runs.rs | 81 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 81 insertions(+) diff --git a/src/store/tests/runs.rs b/src/store/tests/runs.rs index cc914fe..79e867e 100644 --- a/src/store/tests/runs.rs +++ b/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); +} From 06dba7baded1e17a26f0ecafd58e288cc5a9eb74 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Mon, 24 Aug 2026 06:02:26 +0300 Subject: [PATCH 09/10] chore(store): add missing type re-exports in test module Add `RunExecutor` and `RunRecord` to the public re-exports from `crate::store::types` in the test module, so that test code can reference these types without additional imports. Auto-committed-on: dragonfly Co-authored-by: Medulla --- src/store/tests/mod.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/store/tests/mod.rs b/src/store/tests/mod.rs index 2cfb590..5a0197f 100644 --- a/src/store/tests/mod.rs +++ b/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. From 9367c00029c2b5fd2223dc0e7d268491cec5e42c Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Mon, 24 Aug 2026 06:02:59 +0300 Subject: [PATCH 10/10] chore(store): reflow re-exports to remove unnecessary line breaks Consolidated multi-line re-export statements in the store module and its types submodule into single lines, removing the unnecessary line breaks that were splitting related items across lines. This improves readability without changing any exported symbols or behaviour. Auto-committed-on: dragonfly Co-authored-by: Medulla --- src/store/mod.rs | 5 ++--- src/store/types/mod.rs | 3 +-- 2 files changed, 3 insertions(+), 5 deletions(-) diff --git a/src/store/mod.rs b/src/store/mod.rs index 9b8e1d8..cf9d377 100644 --- a/src/store/mod.rs +++ b/src/store/mod.rs @@ -41,9 +41,8 @@ pub use file::{safe_component, workspace_scope, write_atomic}; pub use self::types::{ Diagnosis, LEGACY_TRUNCATED_KEY, NoteId, NoteKind, NoteSource, ProposalId, ProposalStatus, ProposalVerification, RunExecutor, RunId, RunOrigin, RunRecord, RunStatus, RunStep, - TRUNCATED_KEY, - TranscriptEntry, WorkflowDefaults, WorkflowError, WorkflowId, WorkflowNote, WorkflowProposal, - WorkflowRecord, WorkflowRevision, WorkflowSummary, is_truncated, + TRUNCATED_KEY, TranscriptEntry, WorkflowDefaults, WorkflowError, WorkflowId, WorkflowNote, + WorkflowProposal, WorkflowRecord, WorkflowRevision, WorkflowSummary, is_truncated, }; /// An exclusive claim over proposal decisions for one workflow. diff --git a/src/store/types/mod.rs b/src/store/types/mod.rs index 9be746f..6bfb16f 100644 --- a/src/store/types/mod.rs +++ b/src/store/types/mod.rs @@ -47,8 +47,7 @@ pub use proposal::{ pub use run::{ LEGACY_TRUNCATED_KEY, RunExecutor, RunId, RunOrigin, RunRecord, RunStatus, RunStep, - TRUNCATED_KEY, - bounded_evidence, bounded_within, is_truncated, + TRUNCATED_KEY, bounded_evidence, bounded_within, is_truncated, }; pub use transcript::TranscriptEntry; pub use workflow::{