From de6d686db9d384acb7c94a68880bb8dede2ec521 Mon Sep 17 00:00:00 2001 From: Johannes Schickling Date: Tue, 10 Nov 2026 16:02:51 +0100 Subject: [PATCH 1/4] Run linked child runs and person asks from program steps --- crates/fractal-kernel/src/model.rs | 4 + .../fractal-missions/src/store/person_work.rs | 25 +- crates/fractal-reconcile/src/reconcile.rs | 201 +++++++- .../src/reconcile/program.rs | 379 +++++++++++++-- .../src/reconcile/program/tests.rs | 8 +- .../src/reconcile/tests/program_children.rs | 432 ++++++++++++++++++ docs/st3/program-engine.md | 97 ++-- release-notes/program-engine-children.json | 11 + 8 files changed, 1073 insertions(+), 84 deletions(-) create mode 100644 crates/fractal-reconcile/src/reconcile/tests/program_children.rs create mode 100644 release-notes/program-engine-children.json diff --git a/crates/fractal-kernel/src/model.rs b/crates/fractal-kernel/src/model.rs index 5cc768a12b0..352722958fe 100644 --- a/crates/fractal-kernel/src/model.rs +++ b/crates/fractal-kernel/src/model.rs @@ -3455,6 +3455,10 @@ pub struct PersonRenameReport { pub namespace_notice: String, } +/// The requester of a program step's person asks (docs/st3/program-engine.md §6). The daemon asks +/// on the program's behalf: a program step has no claimant, so its attempt fences the ask instead. +pub const PROGRAM_ASKER: &str = "daemon/program"; + #[derive(Clone, Debug, Deserialize, Serialize)] pub struct PersonAskRequest { #[serde(skip)] diff --git a/crates/fractal-missions/src/store/person_work.rs b/crates/fractal-missions/src/store/person_work.rs index 014697e8d00..2af70aa1807 100644 --- a/crates/fractal-missions/src/store/person_work.rs +++ b/crates/fractal-missions/src/store/person_work.rs @@ -418,11 +418,14 @@ pub trait PersonWorkStore { impl PersonWorkStore for Store { fn ask_person(&self, input: &PersonAskRequest) -> Result { + let program = input.actor == crate::model::PROGRAM_ASKER; if !crate::model::person_subject(&input.person) - || !(input.actor.starts_with("agent/") || input.actor == "daemon/browser-pairing") + || !(input.actor.starts_with("agent/") || input.actor == "daemon/browser-pairing" || program) || input.title.trim().is_empty() || input.reason.trim().is_empty() || input.idempotency_key.is_empty() + // A program asks only on its own step. + || (program && (input.step.is_none() || input.new_run.is_some())) { return Err(FractalError::new( "invalid-person-ask", @@ -487,12 +490,20 @@ impl PersonWorkStore for Store { Ok(proven && legacy["fields"]["step"] == origin_subject && legacy["fields"]["step_attempt"].as_u64() == Some(origin.attempt as u64) && matches!(origin.status.as_str(), "ready" | "blocked" | "claimed" | "working")) }).transpose().map_err(internal)?.unwrap_or(false); - if step_owner_is_terminal_tx(tx, &origin_subject)? - || !declaration_live(tx, &input.actor).map_err(internal)? - || (!legacy_origin && (!matches!(origin.status.as_str(), "claimed" | "working") - || origin.claimant.as_deref() != Some(input.actor.as_str()) - || origin.claim_incarnation != input.incarnation - || origin.claim_expires_at_unix_ms.is_none_or(|expiry| expiry <= now_ms()))) { + // A program step has no claimant: the daemon asks for the step's current attempt, + // which the request names as its incarnation (docs/st3/program-engine.md §6). + let stale = if program { + !origin.agentless + || !matches!(origin.status.as_str(), "ready" | "working" | "blocked" | "waiting-person") + || input.incarnation.as_deref() != Some(origin.attempt.to_string().as_str()) + } else { + !declaration_live(tx, &input.actor).map_err(internal)? + || (!legacy_origin && (!matches!(origin.status.as_str(), "claimed" | "working") + || origin.claimant.as_deref() != Some(input.actor.as_str()) + || origin.claim_incarnation != input.incarnation + || origin.claim_expires_at_unix_ms.is_none_or(|expiry| expiry <= now_ms()))) + }; + if step_owner_is_terminal_tx(tx, &origin_subject)? || stale { return Err(FractalError::new("stale-work-ask", format!( "cannot ask from step `{origin_subject}`: status `{}`, claimant `{}`, expected incarnation `{}`, given `{}`; only its live claimant can block this step. Refresh `work ls` and use its current claim and incarnation, or omit --step for an independent ask", origin.status, origin.claimant.as_deref().unwrap_or(""), origin.claim_incarnation.as_deref().unwrap_or(""), input.incarnation.as_deref().unwrap_or("")))); diff --git a/crates/fractal-reconcile/src/reconcile.rs b/crates/fractal-reconcile/src/reconcile.rs index 8fb43d58276..af5d8306e78 100644 --- a/crates/fractal-reconcile/src/reconcile.rs +++ b/crates/fractal-reconcile/src/reconcile.rs @@ -10591,9 +10591,13 @@ impl Reconciler { self.isolate("mission-run", &subject, || { let (run, mission) = self.store.mission_run_for_evaluation(id)?; // Eager cancellation (§6): a program step that left its attempt in this run's - // latest transition stops now, before an inactive run returns below. + // latest transition stops now, with its child runs and asks, before an + // inactive run returns below. Then what the run's programs wait on is started + // or resolved, with its reads noted so their ends evaluate the run again. + let mut settled = false; if let Some(engine) = self.program_engine.get() { - engine.release_left(&run); + engine.release_left(&run)?; + settled = self.settle_program_waits(engine, &run)?; } if active_only && !matches!(run.status.as_str(), "running" | "standing" | "blocked") @@ -10634,7 +10638,9 @@ impl Reconciler { } } } - let evaluated = self.evaluate_active_mission_run(&run, mission); + let evaluated = self + .evaluate_active_mission_run(&run, mission) + .map(|evaluated| evaluated || settled); // Recovery is diagnostic too: admission and execution writes come first. if (!first_readiness_pending(&run) || now_ms().saturating_sub(first_readiness_since(&run)) @@ -11221,6 +11227,14 @@ impl Reconciler { } return Ok(changed); } + // A program waiting on its person asks stays loaded (§6), so a restarted daemon + // replays it to the same wait. Its step's timeout pauses meanwhile (DQ6). + if view.status == "waiting-person" + && let Some(program) = &step.spec.program + { + changed |= self.dispatch_program(run, &step, view, program)?; + return Ok(changed); + } if matches!( view.status.as_str(), "completed" | "cancelled" | "waiting-person" @@ -14405,6 +14419,186 @@ impl Reconciler { } } + /// Start or resolve what the programs of `run` wait on (§6). It runs inside the run's + /// evaluation, so each read is noted: a child run's end, an ask's answer, or the end of a run + /// that holds a full mission's capacity evaluates this run again. Returns whether it wrote. + fn settle_program_waits(&self, engine: &program::ProgramEngine, run: &MissionRunView) -> Result { + let mut changed = false; + for request in engine.requests(&run.subject) { + changed |= match &request.wait { + program::Wait::Run { mission, input } => self.settle_program_child(engine, run, &request, mission, input)?, + program::Wait::Ask(ask) => self.settle_program_ask(engine, &request, ask)?, + }; + } + Ok(changed) + } + + /// A linked child run: started once per key and attempt, adopted when the program resumes, + /// and resolved with its outcome once it ends. A start that finds the mission at capacity + /// waits until a run holding the capacity ends. + fn settle_program_child( + &self, + engine: &program::ProgramEngine, + run: &MissionRunView, + request: &program::Request, + mission: &str, + input: &BTreeMap, + ) -> Result { + let idempotency_key = request.idempotency_key(); + let child = self.store.mission_run_subject_for_idempotency_key(&idempotency_key); + let Some(status) = self.store.mission_run_status(&child)? else { + return match self.start_program_child(run, &request.step, mission, input, idempotency_key) { + Ok(started) => { + engine.waiting(request, &started.subject)?; + Ok(true) + } + Err(error) if error.code == "mission-run-capacity" => { + for holder in self.store.active_mission_runs_for_mission(mission)? { + smallclaims::touched::note_read(|| holder.subject.clone()); + } + Ok(false) + } + Err(error) => { + let reason = format!( + "the program could not start its child run `{}` of mission `{mission}`: {}", + request.key, error.message + ); + engine.fail(&request.step, request.attempt, &reason)?; + Ok(true) + } + }; + }; + if !matches!(status.as_str(), "completed" | "failed" | "cancelled") { + return Ok(false); + } + let exit = self.program_child_exit(&child, &status)?; + engine.resolve(request, &child, exit)?; + Ok(true) + } + + /// The one place a program starts a child run: linked to the program step, for the parent + /// run's requester, in its workspace. + fn start_program_child( + &self, + run: &MissionRunView, + step: &str, + mission: &str, + input: &BTreeMap, + idempotency_key: String, + ) -> std::result::Result { + self.store.create_child_mission_run( + &MissionRunRequest { + mission: mission.to_owned(), + revision: None, + workspace: run.workspace.clone(), + requester: Some(run.requester.clone()), + mode: Some(run.mode.clone()), + inputs: input.clone(), + idempotency_key, + }, + run, + step, + None, + ) + } + + /// A child run's end as its program receives it: `{run, status: 'completed', products?}` + /// with each product its steps declare, by full subject, or `ChildFailed {run, status, + /// reason}`. + fn program_child_exit(&self, child: &str, status: &str) -> Result { + let view = self + .store + .mission_run_steps(child, false)? + .with_context(|| format!("the child run {child} disappeared"))?; + if status == "completed" { + let mut value = serde_json::json!({"run": child, "status": status}); + let products = self.program_child_products(&view)?; + if !products.is_empty() { + value["products"] = Value::Object(products); + } + return Ok(serde_json::json!({"_tag": "Success", "value": value})); + } + let reason = self + .store + .latest_claim(child, Some("mission-run.state"))? + .and_then(|claim| claim.body["fields"]["reason"].as_str().map(str::to_owned)) + .or_else(|| { + view.steps.iter().find(|step| step.status == "failed").map(|step| { + format!("step `{}` failed: {}", step.step, step.blocked_reason.as_deref().unwrap_or("no reason recorded")) + }) + }) + .unwrap_or_else(|| format!("the child run is {status}")); + Ok(serde_json::json!({"_tag": "Failure", "error": {"_tag": "ChildFailed", "run": child, "status": status, "reason": reason}})) + } + + fn program_child_products(&self, child: &MissionRunView) -> Result> { + let mut products = serde_json::Map::new(); + let Some(spec) = self.store.mission_spec(&child.mission, Some(&child.revision))? else { + return Ok(products); + }; + for step in flatten_mission_steps(&spec) { + let Some(view) = child.steps.iter().find(|view| view.step == step.spec.path) else { + continue; + }; + let variables = run_variables(child, &step, view); + for product in &step.spec.products { + let subject = crate::mission::interpolate(&product.subject, &variables)?; + if let Some(value) = self.subject_value(&subject)? { + products.insert(subject, value); + } + } + } + Ok(products) + } + + /// A person ask on the program step: asked once per key and attempt, and resolved with the + /// person's answer, the default at its deadline, or `AskCancelled` when it is withdrawn. + fn settle_program_ask( + &self, + engine: &program::ProgramEngine, + request: &program::Request, + ask: &crate::model::PersonAskRequest, + ) -> Result { + let (subject, asked) = match &request.child { + Some(subject) => (subject.clone(), false), + None => match self.store.ask_person(ask) { + Ok(view) => { + engine.waiting(request, &view.subject)?; + (view.subject, true) + } + Err(error) => { + let reason = format!("the program's ask `{}` was refused: {}", request.key, error.message); + engine.fail(&request.step, request.attempt, &reason)?; + return Ok(true); + } + }, + }; + smallclaims::touched::note_read(|| subject.clone()); + let exit = if let Some(done) = self.store.latest_claim(&subject, Some("work.person-done"))? { + let fields = &done.body["fields"]; + let resolved_by = fields["resolved_by"].as_str().or(done.actor.as_deref()).unwrap_or_default(); + let mut value = serde_json::json!({"resolved_by": resolved_by}); + if let Some(answer) = fields["answer"].as_object() { + value["answer"] = Value::Object( + ["type", "outcome", "id", "label", "text"] + .into_iter() + .filter_map(|name| answer.get(name).filter(|part| part.is_string()).map(|part| (name.to_owned(), part.clone()))) + .collect(), + ); + } + if let Some(summary) = fields["summary"].as_str() { + value["summary"] = Value::from(summary); + } + serde_json::json!({"_tag": "Success", "value": value}) + } else if self.store.latest_claim(&subject, Some("work.person-cancelled"))?.is_some() { + serde_json::json!({"_tag": "Failure", "error": {"_tag": "AskCancelled", "ask": subject}}) + } else { + return Ok(asked); + }; + engine.resolve(request, &subject, exit)?; + Ok(true) + } + fn materialize_step_declarations( &self, run: &MissionRunView, @@ -21190,6 +21384,7 @@ pub mod tests { mod seat_admission_queue; mod terminal_step_guard; mod program_engine; + mod program_children; #[test] fn native_exec_and_gate_shell_resolve_the_declared_path() { use super::{NativeRuntime, RuntimeControl}; diff --git a/crates/fractal-reconcile/src/reconcile/program.rs b/crates/fractal-reconcile/src/reconcile/program.rs index 74665a83de8..5740478a6b0 100644 --- a/crates/fractal-reconcile/src/reconcile/program.rs +++ b/crates/fractal-reconcile/src/reconcile/program.rs @@ -28,9 +28,11 @@ use anyhow::{Context as _, Result, anyhow}; use serde_json::{Value, json}; use sha2::Digest as _; -use crate::MissionsStore; -use crate::model::{ClaimInput, ClaimRecord, FractalError, MissionRunView, ProgramSpec}; +use crate::model::{ + ClaimInput, ClaimRecord, FractalError, MissionRunView, PROGRAM_ASKER, PersonAskRequest, PersonStepResponse, ProgramSpec, +}; use crate::store::Store; +use crate::{MissionsStore, PersonWorkStore}; /// The claim kind of a program journal entry, on the program's `step-run` subject. pub const JOURNAL_KIND: &str = "program.step"; @@ -107,6 +109,42 @@ struct Running { dispatch: Dispatch, /// The host generation running it; a host that exits takes exactly its own programs with it. generation: u64, + /// By journal key: the child runs and asks the program waits on. + requests: BTreeMap, +} + +/// A child run or person ask a program waits on (§6). The run's evaluation starts it and, once it +/// ends, journals its result under `key` and resolves it ([`ProgramEngine::resolve`]). +#[derive(Clone, Debug)] +pub struct Request { + pub step: String, + pub attempt: u32, + pub key: String, + pub wait: Wait, + /// The child run or ask subject once it exists. + pub child: Option, + seq: Value, + since: Instant, +} + +#[derive(Clone, Debug)] +pub enum Wait { + /// A linked child run of `mission` with these inputs. + Run { mission: String, input: BTreeMap }, + /// A native person ask on the program step, asked by [`PROGRAM_ASKER`]. + Ask(Box), +} + +impl Request { + /// The idempotency key of the child run or ask: one per key and attempt, so a resumed attempt + /// adopts it and a retried attempt starts a new one. + pub fn idempotency_key(&self) -> String { + idempotency_key(&self.step, self.attempt, &self.key) + } +} + +fn idempotency_key(step: &str, attempt: u32, key: &str) -> String { + format!("program:{step}:{attempt}:{key}") } #[derive(Default)] @@ -181,7 +219,7 @@ impl ProgramEngine { "cwd": dispatch.cwd, "journal": journal, }); - inner.running.insert(dispatch.step.clone(), Running { dispatch, generation }); + inner.running.insert(dispatch.step.clone(), Running { dispatch, generation, requests: BTreeMap::new() }); send(&inner, &message); Ok(Dispatched::Started) } @@ -203,31 +241,192 @@ impl ProgramEngine { /// Eager cancellation (§6), called from every evaluation of a run: each program attempt of /// `run` this engine runs whose step left it (ended, failed, timed out, retried, cancelled with - /// its run or replaced by a new generation) is stopped now, with its effects. - pub fn release_left(&self, run: &MissionRunView) { + /// its run or replaced by a new generation) is stopped now, with its effects, and the child + /// runs and asks it still waits on are cancelled. + pub fn release_left(&self, run: &MissionRunView) -> Result<()> { let terminal = matches!(run.status.as_str(), "completed" | "failed" | "cancelled"); + let left = { + let mut inner = self.lock(); + let left = inner + .running + .iter() + .filter(|(_, running)| running.dispatch.run == run.subject) + .filter_map(|(step, running)| { + let attempt = running.dispatch.attempt; + let reason = match run.steps.iter().find(|view| view.subject == *step) { + None => Some("the run moved to another generation".to_owned()), + Some(_) if terminal => Some(format!("the run is {}", run.status)), + Some(view) if view.attempt != attempt => Some(format!("the step moved to attempt {}", view.attempt)), + Some(view) if !RUNNING_STATES.contains(&view.status.as_str()) => Some(format!("the step is {}", view.status)), + Some(_) => None, + }; + reason.map(|reason| (step.clone(), attempt, running.generation, reason)) + }) + .collect::>(); + for (step, attempt, generation, reason) in &left { + inner.running.remove(step); + eprintln!("fractal: program {step} attempt {attempt} cancelled: {reason}"); + send_to(&inner, *generation, &json!({"t": "cancel", "step": step, "attempt": attempt, "reason": reason})); + } + left + }; + for (step, attempt, _, reason) in left { + self.cancel_waits(&step, attempt, &reason)?; + } + Ok(()) + } + + /// What the programs of `run` wait on, for its evaluation to start and resolve. + pub fn requests(&self, run: &str) -> Vec { + let inner = self.lock(); + inner + .running + .values() + .filter(|running| running.dispatch.run == run) + .flat_map(|running| running.requests.values().cloned()) + .collect() + } + + /// The child run or ask `request` waits on now exists: journal it as the key's `child`. + pub fn waiting(&self, request: &Request, child: &str) -> Result<()> { + let Some(program) = self.program_of(request, |running| { + if let Some(pending) = running.requests.get_mut(&request.key) { + pending.child = Some(child.to_owned()); + } + }) else { + return Ok(()); + }; + self.write_journal(&request.step, request.attempt, &program, &request.key, "started", &Value::Null, Some(child))?; + Ok(()) + } + + /// The child run or ask `child` that `request` waited on ended with `exit` (`{_tag, value}` or + /// `{_tag, error}`): journal it under the key, then hand it to the program (`resolved`). A key + /// that already holds a result keeps it, and the program gets that one. + pub fn resolve(&self, request: &Request, child: &str, exit: Value) -> Result<()> { + let Some(program) = self.program_of(request, |_| {}) else { return Ok(()) }; + let (status, mut result) = match exit["_tag"].as_str() { + Some("Success") => ("completed", json!({"value": exit["value"]})), + _ => ("failed", json!({"error": exit["error"]})), + }; + result["ms"] = Value::from(u64::try_from(request.since.elapsed().as_millis()).unwrap_or(u64::MAX)); + let exit = match self.write_journal(&request.step, request.attempt, &program, &request.key, status, &result, Some(child)) { + Ok(_) => exit, + Err(error) => { + let mut journaled = self + .journal(&request.step, request.attempt)? + .into_iter() + .find(|entry| entry["key"] == request.key.as_str()) + .ok_or_else(|| anyhow!("journaling `{}` of {}: {error}", request.key, request.step))?; + if let Some(entry) = journaled.as_object_mut() { + entry.remove("key"); + } + journaled + } + }; let mut inner = self.lock(); - let left = inner + let Some((generation, pending)) = inner .running - .iter() - .filter(|(_, running)| running.dispatch.run == run.subject) - .filter_map(|(step, running)| { - let attempt = running.dispatch.attempt; - let reason = match run.steps.iter().find(|view| view.subject == *step) { - None => Some("the run moved to another generation".to_owned()), - Some(_) if terminal => Some(format!("the run is {}", run.status)), - Some(view) if view.attempt != attempt => Some(format!("the step moved to attempt {}", view.attempt)), - Some(view) if !RUNNING_STATES.contains(&view.status.as_str()) => Some(format!("the step is {}", view.status)), - Some(_) => None, - }; - reason.map(|reason| (step.clone(), attempt, reason)) - }) - .collect::>(); - for (step, attempt, reason) in left { - inner.running.remove(&step); - eprintln!("fractal: program {step} attempt {attempt} cancelled: {reason}"); - send(&inner, &json!({"t": "cancel", "step": step, "attempt": attempt, "reason": reason})); + .get_mut(&request.step) + .filter(|running| running.dispatch.attempt == request.attempt) + .and_then(|running| Some((running.generation, running.requests.remove(&request.key)?))) + else { + return Ok(()); + }; + send_to( + &inner, + generation, + &json!({"t": "resolved", "step": request.step, "attempt": request.attempt, "seq": pending.seq, "key": request.key, "exit": exit}), + ); + Ok(()) + } + + /// The program asked for something the daemon refuses (a defect of the bundle): its step fails + /// and the program stops, with what it already waits on. + pub fn fail(&self, step: &str, attempt: u32, reason: &str) -> Result<()> { + { + let mut inner = self.lock(); + let running = inner.running.get(step).filter(|running| running.dispatch.attempt == attempt); + if let Some(generation) = running.map(|running| running.generation) + && forget(&mut inner, step, attempt, generation) + { + send_to(&inner, generation, &json!({"t": "cancel", "step": step, "attempt": attempt, "reason": reason})); + } + } + eprintln!("fractal: program {step} attempt {attempt} failed: {reason}"); + self.cancel_waits(step, attempt, reason)?; + if self.left(step, attempt)?.is_none() { + self.store.set_step_state_unless_terminal(step, "failed", Some(reason))?; + } + (self.wake)(step); + Ok(()) + } + + /// The bundle hash of the program that made `request`, while this engine still runs that + /// attempt, after `update` saw its entry. + fn program_of(&self, request: &Request, update: impl FnOnce(&mut Running)) -> Option { + let mut inner = self.lock(); + let running = inner.running.get_mut(&request.step).filter(|running| running.dispatch.attempt == request.attempt)?; + update(running); + Some(running.dispatch.program.bundle_hash.clone()) + } + + /// Cancel the child runs and asks this attempt journaled as waits and has no result for, once + /// nothing waits on them: the program ended or its step left the attempt. A run that ends + /// cancels its children itself, and its asks end with it. + fn cancel_waits(&self, step: &str, attempt: u32, reason: &str) -> Result<()> { + let reason = format!("its program step {step} (attempt {attempt}) no longer waits on it: {reason}"); + for child in self.open_waits(step, attempt)? { + if child.starts_with("mission-run/") { + if !matches!(self.store.mission_run_status(&child)?.as_deref(), None | Some("completed" | "failed" | "cancelled")) { + self.store.request_mission_run_cancellation(&child, &reason)?; + } + continue; + } + if !self.store.step_run(&child)?.is_some_and(|ask| matches!(ask.status.as_str(), "ready" | "pending")) { + continue; + } + let cancelled = self.store.finish_person_step( + &PersonStepResponse { + delegation: None, + subject: child.clone(), + actor: PROGRAM_ASKER.to_owned(), + summary: reason.clone(), + evidence: Vec::new(), + episode: None, + answer: None, + idempotency_key: format!("program-cancel:{child}"), + }, + true, + ); + match cancelled { + // `stale-fence`: the ask no longer waits (answered, or its run or generation ended). + Err(error) if error.code != "stale-fence" => return Err(error.into()), + _ => {} + } } + Ok(()) + } + + /// The child runs and asks this attempt journaled under a key that has no result yet. + fn open_waits(&self, step: &str, attempt: u32) -> Result> { + let mut waits = BTreeMap::new(); + for claim in self.journal_claims(step)? { + let fields = &claim.body["fields"]; + let Some(key) = fields["key"].as_str().filter(|_| fields["attempt"].as_u64() == Some(u64::from(attempt))) else { + continue; + }; + match (fields["status"].as_str(), fields["child"].as_str()) { + (Some("started"), Some(child)) => { + waits.insert(key.to_owned(), child.to_owned()); + } + (Some("completed" | "failed"), _) => { + waits.remove(key); + } + _ => {} + } + } + Ok(waits.into_values().collect()) } /// The bundle as this revision pinned it, read from the store's blob of its hash (never from @@ -255,20 +454,8 @@ impl ProgramEngine { /// store order. A retried attempt starts from the top, because an earlier attempt's /// compensation may have undone what its completed steps did. fn journal(&self, step: &str, attempt: u32) -> Result> { - let mut claims = Vec::new(); - let mut before = None; - loop { - // Newest first; each page continues below the oldest index of the previous one. - let page = self.store.claims_for_subject_kind_at(step, JOURNAL_KIND, before, true, JOURNAL_PAGE)?; - claims.extend(page.claims); - match page.next_cursor { - Some(cursor) => before = Some(cursor), - None => break, - } - } - claims.sort_by_key(|claim| claim.store_index); let mut entries = Vec::new(); - for claim in claims { + for claim in self.journal_claims(step)? { let fields = &claim.body["fields"]; if fields["attempt"].as_u64() != Some(u64::from(attempt)) || fields["key"] == RESULT_KEY { continue; @@ -282,6 +469,23 @@ impl ProgramEngine { Ok(entries) } + /// Every journal entry of `step`, in store order. + fn journal_claims(&self, step: &str) -> Result> { + let mut claims = Vec::new(); + let mut before = None; + loop { + // Newest first; each page continues below the oldest index of the previous one. + let page = self.store.claims_for_subject_kind_at(step, JOURNAL_KIND, before, true, JOURNAL_PAGE)?; + claims.extend(page.claims); + match page.next_cursor { + Some(cursor) => before = Some(cursor), + None => break, + } + } + claims.sort_by_key(|claim| claim.store_index); + Ok(claims) + } + /// The live host's generation, starting one when there is none or the last one exited. fn ensure_host(self: &Arc, inner: &mut Inner) -> Result { if let Some(host) = inner.host.as_mut() { @@ -430,12 +634,53 @@ impl ProgramEngine { let attempt = u32::try_from(attempt)?; match message["t"].as_str() { Some("journal") => self.journal_entry(step, attempt, generation, &program, message), + Some("run" | "ask") => self.request(step, attempt, generation, &program, message), Some("done") => self.done(step, attempt, generation, &program, message), Some("unresponsive") => self.unresponsive(step, attempt, generation, message), other => Err(anyhow!("unknown message type {other:?}")), } } + /// A program waits on a child run or a person ask (§6). Registered here, it is started and + /// resolved by its run's evaluation. The `started` entry, written before the request is + /// registered, is the change that evaluates the run; a request the daemon cannot accept fails + /// the step. + fn request(&self, step: &str, attempt: u32, generation: u64, program: &str, message: &Value) -> Result<()> { + if self.stop_if_left(step, attempt, generation)? { + return Ok(()); + } + let Some(key) = message["key"].as_str().filter(|key| !key.is_empty() && *key != RESULT_KEY) else { + return self.fail(step, attempt, &format!("the program made a request without a key or with the reserved key {RESULT_KEY}")); + }; + let wait = match wait(step, attempt, key, message) { + Ok(wait) => wait, + Err(reason) => return self.fail(step, attempt, &format!("the program made an invalid request for `{key}`: {reason}")), + }; + self.write_journal(step, attempt, program, key, "started", &Value::Null, None)?; + { + let mut inner = self.lock(); + let Some(running) = inner + .running + .get_mut(step) + .filter(|running| running.dispatch.attempt == attempt && running.generation == generation) + else { + return Ok(()); + }; + let request = Request { + step: step.to_owned(), + attempt, + key: key.to_owned(), + wait, + child: None, + seq: message["seq"].clone(), + since: Instant::now(), + }; + running.requests.insert(key.to_owned(), request); + } + (self.wake)(step); + Ok(()) + } + /// Why the step no longer runs this attempt, if it does not. fn left(&self, step: &str, attempt: u32) -> Result> { Ok(match self.store.step_run(step)? { @@ -449,10 +694,13 @@ impl ProgramEngine { /// True when it was stopped. fn stop_if_left(&self, step: &str, attempt: u32, generation: u64) -> Result { let Some(reason) = self.left(step, attempt)? else { return Ok(false) }; - let mut inner = self.lock(); - if forget(&mut inner, step, attempt, generation) { - send_to(&inner, generation, &json!({"t": "cancel", "step": step, "attempt": attempt, "reason": reason})); + { + let mut inner = self.lock(); + if forget(&mut inner, step, attempt, generation) { + send_to(&inner, generation, &json!({"t": "cancel", "step": step, "attempt": attempt, "reason": reason})); + } } + self.cancel_waits(step, attempt, &reason)?; Ok(true) } @@ -532,7 +780,11 @@ impl ProgramEngine { if !forget(&mut self.lock(), step, attempt, generation) { return Ok(()); } - if self.left(step, attempt)?.is_some() { + // A program can end while a child run or ask it started is still open (a lost race): + // nothing waits on it any more. + let left = self.left(step, attempt)?; + self.cancel_waits(step, attempt, left.as_deref().unwrap_or("the program ended"))?; + if left.is_some() { return Ok(()); } let exit = &message["exit"]; @@ -562,6 +814,7 @@ impl ProgramEngine { inner.faults.insert(step.to_owned(), (attempt, reason.clone())); } eprintln!("fractal: program {step} attempt {attempt}: {reason}"); + self.cancel_waits(step, attempt, &reason)?; if self.left(step, attempt)?.is_none() { self.store.set_step_state_unless_terminal(step, "failed", Some(&reason))?; } @@ -694,6 +947,50 @@ fn write_host(mut stdin: ChildStdin, queue: mpsc::Receiver>) { } } +/// What a `run` or `ask` message asks the daemon to wait on, or why it cannot. +fn wait(step: &str, attempt: u32, key: &str, message: &Value) -> std::result::Result { + let text = |name: &str| { + message[name] + .as_str() + .filter(|value| !value.trim().is_empty()) + .map(str::to_owned) + .ok_or_else(|| format!("`{name}` must be a non-empty string")) + }; + if message["t"] == "run" { + let mission = text("mission")?; + let input = serde_json::from_value::>(message["input"].clone()) + .map_err(|_| "`input` must map each mission input to a string".to_owned())?; + let mission = mission.strip_prefix("mission/").unwrap_or(&mission).to_owned(); + return Ok(Wait::Run { mission, input }); + } + let deadline_ms = match &message["deadline"] { + Value::Null => None, + Value::String(deadline) => Some(crate::declaration_syntax::parse_duration(deadline, true).map_err(|error| error.message)?), + _ => return Err("`deadline` must be a duration such as `10m`".into()), + }; + let default_answer = match &message["default"] { + Value::Null => None, + Value::String(answer) => Some(answer.clone()), + _ => return Err("`default` must name an answer of the request".into()), + }; + let title = text("title")?; + Ok(Wait::Ask(Box::new(PersonAskRequest { + legacy_request: None, + person: text("for")?, + reason: title.clone(), + title, + actor: PROGRAM_ASKER.to_owned(), + step: Some(step.to_owned()), + new_run: None, + // A program step has no claim: its attempt fences the ask. + incarnation: Some(attempt.to_string()), + request: Some(message["request"].clone()).filter(|request| !request.is_null()), + deadline_ms, + default_answer, + idempotency_key: idempotency_key(step, attempt, key), + }))) +} + /// A typed failure keeps its tag and fields, so `fractal missions show` says which error ended /// the program. fn failure_reason(error: &Value) -> String { diff --git a/crates/fractal-reconcile/src/reconcile/program/tests.rs b/crates/fractal-reconcile/src/reconcile/program/tests.rs index bda84a45fa1..49cce72609b 100644 --- a/crates/fractal-reconcile/src/reconcile/program/tests.rs +++ b/crates/fractal-reconcile/src/reconcile/program/tests.rs @@ -105,18 +105,18 @@ fn messages_for_another_attempt_or_host_change_nothing() { env: BTreeMap::new(), cwd: "/".into(), }; - engine.lock().running.insert(step.into(), Running { dispatch, generation: 1 }); + engine.lock().running.insert(step.into(), Running { dispatch, generation: 1, requests: BTreeMap::new() }); let message = |attempt: u32, t: &str| { json!({"t": t, "step": step, "attempt": attempt, "seq": 1, "key": "merge", "status": "completed", "value": 1, - "exit": {"_tag": "Success", "value": 1}, "ms": 5}) + "exit": {"_tag": "Success", "value": 1}, "ms": 5, "mission": "publish", "input": {}, "for": "person/owner", "title": "Go?"}) }; - for t in ["journal", "done", "unresponsive"] { + for t in ["journal", "run", "ask", "done", "unresponsive"] { // An earlier attempt of the step, then the right attempt from a replaced host. engine.handle(&message(1, t), 1).unwrap(); engine.handle(&message(2, t), 2).unwrap(); } assert_eq!(journal_claims(&engine, step), 0); - assert!(engine.lock().running.contains_key(step)); + assert!(engine.lock().running.get(step).is_some_and(|running| running.requests.is_empty())); assert_eq!(engine.host_fault(step, 2), None); // A late message for the earlier attempt never removes the attempt that replaced it, and a // refused entry is answered rather than left waiting (no host here, so nothing is sent). diff --git a/crates/fractal-reconcile/src/reconcile/tests/program_children.rs b/crates/fractal-reconcile/src/reconcile/tests/program_children.rs new file mode 100644 index 00000000000..8ee99d11131 --- /dev/null +++ b/crates/fractal-reconcile/src/reconcile/tests/program_children.rs @@ -0,0 +1,432 @@ +//! Linked child runs and person asks of program steps, with a real Bun host +//! (docs/st3/program-engine.md §6, §7): adoption on resume, the capacity wait and its reactive +//! wake, cancellation, retries, asks answered, defaulted and withdrawn, and the exit shapes the +//! `fractal/effect` program API decodes. These tests fail, never skip, when no `bun` is on `PATH`. +use super::*; +use crate::model::{AnswerInput, PROGRAM_ASKER, PersonStepResponse}; +use crate::reconcile::program::{JOURNAL_KIND, Liveness, ProgramEngine, RESULT_KEY}; +use serde_json::json; +use std::time::Instant; + +/// A hand-written bundle against HostContext v1. Mode `child` starts one linked child run, mode +/// `ask` opens three concurrent asks; then both wait for a gate file in a journaled step, so a +/// test can replace the host after the waits resolved. +const BUNDLE: &str = r#" +const step = async (ctx, key, effect) => { + const replayed = ctx.journal.get(key) + if (replayed !== undefined) return replayed.value + ctx.started(key) + const at = Date.now() + const value = await effect() + await ctx.record({ key, status: 'completed', value, ms: Date.now() - at }) + return value +} +const wait = (ctx, key, request) => ctx.journal.get(key) ?? request() +const gate = (ctx) => + step(ctx, 'gate', async () => { + const out = await ctx.spawn(['sh', '-c', 'while [ ! -e "$GATE" ]; do sleep 0.05; done; echo opened'], { env: { GATE: ctx.payload.gate } }) + return out.stdout.trim() + }) +const DECISION = { + version: 1, type: 'decision', question: 'Ship it?', why_person: 'The owner decides releases.', + answers: [ + { id: 'ship', label: 'Ship', outcome: 'accept', consequence: 'It ships.' }, + { id: 'hold', label: 'Hold', outcome: 'decline', consequence: 'Nothing ships.' }, + ], +} +export default { + [Symbol.for('fractal.program.v1')]: async (ctx) => { + if (ctx.payload.mode === 'child') { + const published = await wait(ctx, 'publish', () => ctx.child({ key: 'publish', mission: 'program-child', input: { head: 'h1' } })) + if (published._tag === 'Failure') return published + return { _tag: 'Success', value: { published: published.value, gate: await gate(ctx) } } + } + const asks = await Promise.all([ + wait(ctx, 'approve', () => ctx.ask({ key: 'approve', for: 'person/owner', title: 'Ship it?', request: DECISION, deadline: '2m', default: 'ship' })), + wait(ctx, 'note', () => ctx.ask({ key: 'note', for: 'person/owner', title: 'Anything to add?', request: null })), + wait(ctx, 'withdrawn', () => ctx.ask({ key: 'withdrawn', for: 'person/owner', title: 'Still needed?', request: null })), + ]) + return { _tag: 'Success', value: { asks, gate: await gate(ctx) } } + }, +} +"#; + +struct Fixture { + store: Arc, + reconciler: Reconciler, + engine: Arc, + dir: tempfile::TempDir, +} + +struct Started { + run: String, + step: String, + gate: std::path::PathBuf, +} + +/// One journal entry: key, status, attempt, child. +type Entry = (String, String, u64, Option); + +fn fixture() -> Fixture { + let dir = tempfile::tempdir().unwrap(); + let store = Arc::new(Store::open_memory("node").unwrap()); + store.set_ask_deadlines(true); + let bundle = dir.path().join("program.mjs"); + std::fs::write(&bundle, BUNDLE).unwrap(); + let hash = store.put_blob(BUNDLE.as_bytes()).unwrap(); + apply_source( + &store, + &format!( + r#"version 2 +agent "publisher" {{ workspace "/tmp"; command "true"; restart "never" }} +mission "program-child" state="ready" {{ + goal "Publish one head at a time." + concurrent-runs max=1 + input "head" kind="text" + step "publish" {{ + assigned-to "agent/publisher" + produces {{ resource "mission-run/${{ST_MISSION_RUN}}/published" {{ kind "custom.st3.program-test" }} }} + }} +}} +mission "program-parent" state="ready" {{ + goal "Wait on child runs and asks." + concurrent-runs max=4 + input "mode" kind="text" + input "gate" kind="text" + step "program" {{ + retry {{ attempts 2 }} + program "{}" sha256="{hash}" + }} +}}"#, + bundle.display() + ), + "program-children-definition", + ); + let reconciler = Reconciler::new(store.clone(), Arc::new(FakeRuntime::default()), "node".into(), Arc::new(Notify::new())); + // As the daemon's engine does: a change off the reconciler thread marks the step for the next + // pass. Nothing signals a follow-reads evaluation: tests call that themselves. + let incremental = reconciler.incremental.clone(); + let wake = Arc::new(move |step: &str| incremental.touch(step)); + let engine = ProgramEngine::new(store.clone(), wake, dir.path().join("programs"), Liveness::default()); + assert!(reconciler.program_engine.set(engine.clone()).is_ok()); + Fixture { store, reconciler, engine, dir } +} + +impl Fixture { + fn start(&self, name: &str, mode: &str) -> Started { + let gate = self.dir.path().join(format!("{name}.gate")); + let inputs = BTreeMap::from([("mode".to_owned(), mode.to_owned()), ("gate".to_owned(), gate.display().to_string())]); + let run = self + .store + .create_mission_run(&MissionRunRequest { + mission: "program-parent".into(), + revision: None, + workspace: self.dir.path().display().to_string(), + requester: Some("person/owner".into()), + mode: None, + inputs, + idempotency_key: format!("program-parent-{name}"), + }) + .unwrap(); + let step = run.steps.iter().find(|step| step.step == "program").unwrap().subject.clone(); + Started { run: run.subject, step, gate } + } + + /// A run of the child mission that nothing links to a program. + fn holder(&self) -> String { + self.store + .create_mission_run(&MissionRunRequest { + mission: "program-child".into(), + revision: None, + workspace: self.dir.path().display().to_string(), + requester: Some("person/owner".into()), + mode: None, + inputs: BTreeMap::from([("head".to_owned(), "h0".to_owned())]), + idempotency_key: "program-child-holder".into(), + }) + .unwrap() + .subject + } + + /// Run passes until `done` holds; the engine itself works off the reconciler thread. + fn until(&self, what: &str, done: impl Fn() -> bool) { + let deadline = Instant::now() + Duration::from_secs(60); + while !done() { + assert!(Instant::now() < deadline, "timed out waiting until {what}"); + self.reconciler.reconcile_once().unwrap(); + std::thread::sleep(Duration::from_millis(50)); + } + } + + fn status(&self, step: &str) -> String { + self.store.step_run(step).unwrap().unwrap().status + } + + fn run_status(&self, run: &str) -> String { + self.store.mission_run_status(run).unwrap().unwrap() + } + + fn journal(&self, step: &str) -> Vec { + let mut claims = self.store.claims_for(step, Some(JOURNAL_KIND)).unwrap(); + claims.sort_by_key(|claim| claim.store_index); + claims + .iter() + .map(|claim| { + let fields = &claim.body["fields"]; + ( + fields["key"].as_str().unwrap().to_owned(), + fields["status"].as_str().unwrap().to_owned(), + fields["attempt"].as_u64().unwrap(), + fields["child"].as_str().map(str::to_owned), + ) + }) + .collect() + } + + /// The result journaled for `key` at `attempt`: `(status, value or error)`. + fn result(&self, step: &str, key: &str, attempt: u64) -> Option<(String, Value)> { + self.store.claims_for(step, Some(JOURNAL_KIND)).unwrap().into_iter().find_map(|claim| { + let fields = &claim.body["fields"]; + let status = fields["status"].as_str()?; + (fields["key"] == key && fields["attempt"].as_u64() == Some(attempt) && status != "started").then(|| { + let result = if status == "failed" { fields["error"].clone() } else { fields["value"].clone() }; + (status.to_owned(), result) + }) + }) + } + + /// The child or ask subject the program journaled for `key` at `attempt`. + fn child(&self, step: &str, key: &str, attempt: u64) -> Option { + self.journal(step) + .into_iter() + .find_map(|(entry, status, at, child)| (entry == key && status == "started" && at == attempt).then_some(child).flatten()) + } + + /// The program's child runs, by the step the program runs as. + fn children(&self, step: &str) -> Vec { + self.store.mission_run_children_for_step(step).unwrap() + } + + /// Complete a child run's only step, after writing its declared product. + fn complete_child(&self, child: &str) { + let id = child.strip_prefix("mission-run/").unwrap(); + self.store + .append_claim(&ClaimInput { + subject: format!("resource/mission-run/{id}/published"), + kind: "resource.observed".into(), + actor: Some("agent/publisher".into()), + fields: BTreeMap::from([ + ("kind".into(), json!("custom.st3.program-test")), + ("head".into(), json!("h1")), + ]), + evidence: Vec::new(), + expected_subject: None, + idempotency_key: None, + }) + .unwrap(); + let step = self.store.mission_run(child).unwrap().unwrap().steps.remove(0).subject; + self.store.set_step_state(&step, "completed", None).unwrap(); + } + + fn kill_host(&self) -> u32 { + let host = self.engine.host_pid().unwrap(); + // SAFETY: the host is this test's own child process. + assert_eq!(unsafe { libc::kill(host as i32, libc::SIGKILL) }, 0); + host + } +} + +fn count(journal: &[Entry], key: &str, status: &str) -> usize { + journal.iter().filter(|(entry, at, _, _)| entry == key && at == status).count() +} + +#[test] +fn a_child_run_is_adopted_on_resume_and_its_result_is_journaled_and_replayed() { + let fixture = fixture(); + let parent = fixture.start("adopt", "child"); + fixture.until("the child run starts", || fixture.child(&parent.step, "publish", 1).is_some()); + let child = fixture.child(&parent.step, "publish", 1).unwrap(); + // The child is linked to the program step under the key's idempotency key. + assert_eq!(fixture.store.mission_run_subject_for_idempotency_key(&format!("program:{}:1:publish", parent.step)), child); + assert_eq!(fixture.children(&parent.step).iter().map(|run| run.subject.clone()).collect::>(), [child.clone()]); + // A new host resumes the program, which asks for the same key again: the daemon adopts the + // child it started instead of starting another. + let host = fixture.kill_host(); + fixture.until("the resumed program waits on the child again", || { + fixture.engine.host_pid().is_some_and(|pid| pid != host) && count(&fixture.journal(&parent.step), "publish", "started") == 3 + }); + fixture.reconciler.reconcile_once().unwrap(); + assert_eq!(fixture.children(&parent.step).len(), 1); + fixture.complete_child(&child); + fixture.until("the child's end is journaled", || fixture.result(&parent.step, "publish", 1).is_some()); + let completed = json!({ + "run": child, + "status": "completed", + "products": { + format!("resource/{child}/published"): {"kind": "custom.st3.program-test", "head": "h1"}, + }, + }); + assert_eq!(fixture.result(&parent.step, "publish", 1), Some(("completed".to_owned(), completed.clone()))); + fixture.until("the program waits on its gate", || count(&fixture.journal(&parent.step), "gate", "started") == 1); + // A host replaced now replays the child's result from the journal: no new request. + let before = count(&fixture.journal(&parent.step), "publish", "started"); + let host = fixture.kill_host(); + fixture.until("the program waits on its gate again", || { + fixture.engine.host_pid().is_some_and(|pid| pid != host) && count(&fixture.journal(&parent.step), "gate", "started") == 2 + }); + std::fs::write(&parent.gate, "").unwrap(); + fixture.until("the program completes", || fixture.status(&parent.step) == "completed"); + let journal = fixture.journal(&parent.step); + assert_eq!(count(&journal, "publish", "started"), before); + assert_eq!(count(&journal, "publish", "completed"), 1); + assert_eq!(fixture.children(&parent.step).len(), 1); + assert_eq!( + fixture.result(&parent.step, RESULT_KEY, 1), + Some(("completed".to_owned(), json!({"published": completed, "gate": "opened"}))) + ); +} + +#[test] +fn a_child_of_a_full_mission_waits_and_starts_when_the_holder_ends() { + let fixture = fixture(); + let holder = fixture.holder(); + let parent = fixture.start("capacity", "child"); + fixture.until("the program asks for its child", || !fixture.engine.requests(&parent.run).is_empty()); + // An evaluation that finds the mission full notes the runs that hold it. + fixture.reconciler.reconcile_once().unwrap(); + assert!(fixture.children(&parent.step).is_empty()); + assert_eq!(fixture.child(&parent.step, "publish", 1), None); + assert_eq!(fixture.status(&parent.step), "working"); + // The holder's end alone, with no pass, starts the child: `concurrent-runs max=1` serializes + // the section, and the waiting start runs next. + fixture.store.request_mission_run_cancellation(&holder, "make room").unwrap(); + fixture.reconciler.evaluate_changed_mission_runs().unwrap(); + let child = fixture.child(&parent.step, "publish", 1).expect("the child starts once the holder ended"); + assert_eq!(fixture.children(&parent.step).iter().map(|run| run.subject.clone()).collect::>(), [child.clone()]); + fixture.complete_child(&child); + std::fs::write(&parent.gate, "").unwrap(); + fixture.until("the program completes", || fixture.status(&parent.step) == "completed"); +} + +#[test] +fn children_are_cancelled_with_their_run_or_their_step_and_a_retry_starts_a_new_one() { + let fixture = fixture(); + // Capacity 1: the second parent's child waits until the first one's is cancelled. + let cancelled = fixture.start("cancelled", "child"); + fixture.until("the first child starts", || fixture.child(&cancelled.step, "publish", 1).is_some()); + let first = fixture.child(&cancelled.step, "publish", 1).unwrap(); + // Cancelling the parent run cascades to its child. + fixture.store.request_mission_run_cancellation(&cancelled.run, "not wanted").unwrap(); + fixture.reconciler.evaluate_changed_mission_runs().unwrap(); + assert_eq!(fixture.run_status(&first), "cancelled"); + + let retried = fixture.start("retried", "child"); + fixture.until("the second parent's child starts", || fixture.child(&retried.step, "publish", 1).is_some()); + let attempt_one = fixture.child(&retried.step, "publish", 1).unwrap(); + // The program step fails while its run goes on (it has another attempt): its child is + // cancelled in that evaluation. + fixture.store.set_step_state(&retried.step, "failed", Some("the active execution timeout expired")).unwrap(); + fixture.reconciler.evaluate_changed_mission_runs().unwrap(); + assert_eq!(fixture.run_status(&attempt_one), "cancelled"); + assert_eq!(fixture.run_status(&retried.run), "running"); + // The retried attempt starts from the top with a new child. + fixture.until("attempt 2 starts its child", || fixture.child(&retried.step, "publish", 2).is_some()); + let attempt_two = fixture.child(&retried.step, "publish", 2).unwrap(); + assert_ne!(attempt_two, attempt_one); + assert_eq!(fixture.result(&retried.step, "publish", 1), None); + // A child that fails is a typed failure in its parent. + let child_step = fixture.store.mission_run(&attempt_two).unwrap().unwrap().steps.remove(0).subject; + fixture.store.set_step_state(&child_step, "failed", Some("the publisher refused the head")).unwrap(); + fixture.until("the parent step fails", || fixture.status(&retried.step) == "failed"); + let (status, error) = fixture.result(&retried.step, "publish", 2).unwrap(); + assert_eq!(status, "failed"); + assert_eq!(error["_tag"], "ChildFailed"); + assert_eq!(error["run"], attempt_two.as_str()); + assert_eq!(error["status"], "failed"); + assert!(error["reason"].as_str().is_some_and(|reason| !reason.is_empty()), "{error}"); + assert_eq!(error.as_object().unwrap().len(), 4, "{error}"); +} + +#[test] +fn asks_are_answered_defaulted_or_withdrawn_and_their_answers_replay() { + let fixture = fixture(); + let parent = fixture.start("asks", "ask"); + let asks = || ["approve", "note", "withdrawn"].map(|key| fixture.child(&parent.step, key, 1)); + fixture.until("the three asks are open", || asks().iter().all(Option::is_some)); + let [approve, note, withdrawn] = asks().map(Option::unwrap); + // The step waits on a person, and its program stays loaded. + fixture.until("the step waits on a person", || fixture.status(&parent.step) == "waiting-person"); + fixture.reconciler.reconcile_once().unwrap(); + assert_eq!(fixture.engine.requests(&parent.run).len(), 3); + for ask in [&approve, ¬e, &withdrawn] { + let claim = fixture.store.latest_claim(ask, Some("work.person-asked")).unwrap().unwrap(); + assert_eq!(claim.actor.as_deref(), Some(PROGRAM_ASKER)); + assert_eq!(claim.body["fields"]["origin_step"], parent.step.as_str()); + } + // One answer resolves its key while the other asks stay open. + fixture + .store + .finish_person_step( + &PersonStepResponse { + delegation: None, + subject: note.clone(), + actor: "person/owner".into(), + summary: "Add the changelog.".into(), + evidence: Vec::new(), + episode: None, + answer: Some(AnswerInput { id: None, text: Some("Add the changelog.".into()), option_notes: Vec::new() }), + idempotency_key: "note-answer".into(), + }, + false, + ) + .unwrap(); + fixture.reconciler.evaluate_changed_mission_runs().unwrap(); + assert_eq!( + fixture.result(&parent.step, "note", 1), + Some(("completed".to_owned(), json!({"summary": "Add the changelog.", "resolved_by": "person/owner"}))) + ); + assert_eq!(fixture.status(&parent.step), "waiting-person"); + // A host replaced now replays the answer and asks the open keys again: no new asks. + let host = fixture.kill_host(); + fixture.until("the resumed program waits again", || { + fixture.engine.host_pid().is_some_and(|pid| pid != host) && fixture.engine.requests(&parent.run).len() == 2 + }); + fixture.reconciler.reconcile_once().unwrap(); + assert_eq!(asks(), [Some(approve.clone()), Some(note.clone()), Some(withdrawn.clone())]); + assert_eq!(count(&fixture.journal(&parent.step), "note", "started"), 2); + // An ask withdrawn while the program waits fails its key with `AskCancelled`. + fixture + .store + .finish_person_step( + &PersonStepResponse { + delegation: None, + subject: withdrawn.clone(), + actor: PROGRAM_ASKER.into(), + summary: "No longer needed.".into(), + evidence: Vec::new(), + episode: None, + answer: None, + idempotency_key: "withdraw".into(), + }, + true, + ) + .unwrap(); + fixture.reconciler.evaluate_changed_mission_runs().unwrap(); + assert_eq!( + fixture.result(&parent.step, "withdrawn", 1), + Some(("failed".to_owned(), json!({"_tag": "AskCancelled", "ask": withdrawn}))) + ); + // The deadline resolves the last ask with its default answer. + fixture.store.reconcile_person_asks_at(now_ms() + 3 * 60_000).unwrap(); + std::fs::write(&parent.gate, "").unwrap(); + fixture.until("the program completes", || fixture.status(&parent.step) == "completed"); + let defaulted = json!({ + "answer": {"type": "decision", "outcome": "accept", "id": "ship", "label": "Ship"}, + "summary": fixture.store.latest_claim(&approve, Some("work.person-done")).unwrap().unwrap().body["fields"]["summary"], + "resolved_by": "deadline", + }); + assert_eq!(fixture.result(&parent.step, "approve", 1), Some(("completed".to_owned(), defaulted.clone()))); + let (_, value) = fixture.result(&parent.step, RESULT_KEY, 1).unwrap(); + assert_eq!(value["asks"][0], json!({"_tag": "Success", "value": defaulted})); + assert_eq!(value["asks"][2], json!({"_tag": "Failure", "error": {"_tag": "AskCancelled", "ask": withdrawn}})); +} diff --git a/docs/st3/program-engine.md b/docs/st3/program-engine.md index 1e510c506b8..a74ada6a81b 100644 --- a/docs/st3/program-engine.md +++ b/docs/st3/program-engine.md @@ -141,7 +141,7 @@ can run an effect to its end but can never journal it. | daemon → host | `nack` | `step`, `attempt`, `seq`, `error` | the entry could not be written; the program fails with a defect | | daemon → host | `cancel` | `step`, `attempt`, `reason` | the step ended or moved to another attempt; stop the program and its effects | | host → daemon | `run` | `step`, `attempt`, `seq`, `key`, `mission`, `input` | start or adopt the linked child run for `key` (§6) | -| host → daemon | `ask` | `step`, `attempt`, `seq`, `key`, `for`, `request`, `deadline`, `default` | open or adopt the person ask for `key` (§6) | +| host → daemon | `ask` | `step`, `attempt`, `seq`, `key`, `for`, `title`, `request`, `deadline`, `default` | open or adopt the person ask for `key` (§6) | | daemon → host | `resolved` | `step`, `attempt`, `seq`, `key`, `exit` | the child run or ask for `key` ended; its result is journaled | | host → daemon | `done` | `step`, `attempt`, `exit` (`{_tag: Success, value}` or `{_tag: Failure, error}`) | the program ended | | host → daemon | `unresponsive` | `step`, `attempt`, `ms` | the program's worker missed the liveness deadline and was terminated (DQ8) | @@ -165,7 +165,8 @@ ended by the next daemon (DQ2). The daemon starts the host with `FRACTAL_PROGRAM_RUNTIME`, or else `bun` on its `PATH`; a member without either cannot run program steps. Cancellation is eager: every evaluation of a run tells the engine which of the run's program attempts left their step (ended, failed, timed out, retried, -cancelled with the run, or replaced by a new generation), and the engine sends `cancel` at once. +cancelled with the run, or replaced by a new generation), and the engine sends `cancel` at once and +cancels the child runs and asks the attempt still waits on (§6). ### 3. Declaration and authority @@ -260,29 +261,60 @@ fail the step (a bug), interrupts (host or daemon gone) are never journaled and Long-lived loops do not grow one journal forever: a recurring program is a schedule of finite runs, and an iterative program that needs more than a few thousand steps runs rounds as child runs (§6). -### 6. Linked child runs and worker turns - -`Fractal.run(mission, inputs)` asks the daemon to create a child run with -`create_child_mission_run`, the path `uses-mission` already takes: `parent_step_run` is the program -step, the idempotency key is `program:STEP:ATTEMPT:KEY`, so a resumed attempt adopts its child -instead of starting another, while a retried attempt starts a new one (§5). The child is natively -linked: cancelling the parent run cancels the child through `cancel_descendant_mission_runs_tx` -(R08); a capacity-full child start waits (mission graph runtime §Run ownership). The engine learns -the child's terminal state from its own store transition, not by polling, and journals it under the -child step's key. - -A worker turn (`Fractal.step(...).pipe(Effect.provide(Fractal.Worker.layer(agent)))`) is a child run -of a one-step mission whose step is assigned to that agent, with the step's `goal` and `success` -schema; the result is the child's declared product. A run-scoped reviewer -(`agent/${ST_MISSION_RUN}/reviewer`, one-shot) is declared by that child mission, so independence -holds as in LandingMission. +### 6. Linked child runs, worker turns and asks + +A program waits on two kinds of things the daemon owns: a linked child run (`run`, from +`Fractal.run` and worker turns) and a native person ask (`ask`, from `Fractal.ask`). The engine +journals a `started` entry for the key when the request arrives, and the program's run evaluation +does the rest: it starts or adopts what the key waits on, journals a second `started` entry whose +`child` names it, and, once it ends, journals the result under the key (with `child` set) and sends +`resolved`. Everything the evaluation looks at is a noted read, so the store transition that ends a +child run or answers an ask evaluates the parent run again; nothing polls. A replayed key never +asks again; a resumed attempt that asks for an open key adopts what that key already waits on. A +request the daemon refuses (an unknown mission, inputs the mission does not accept, an ask without +a person) fails the program step with the reason, as a defect would. + +**Child runs.** The evaluation calls `create_child_mission_run`, the path `uses-mission` takes: +`parent_step_run` is the program step, the inputs are the request's, and the idempotency key is +`program:STEP:ATTEMPT:KEY`, so a resumed attempt adopts its child (the run subject follows from the +key) while a retried attempt starts a new one (§5). The child runs for the parent run's requester in +its workspace. A start that finds the mission at its `concurrent-runs` limit waits: the evaluation +notes reads of the runs that hold the capacity, so the end of any of them evaluates the parent +again and the start is retried then. When the child ends, the program receives: + +| Child | Exit | +|---|---| +| `completed` | Success `{ run, status: 'completed', products? }`; `products` holds the value of each product the child's steps declare, by full subject (`resource/mission-run/CHILD/…`) | +| `failed` / `cancelled` | Failure `{ _tag: 'ChildFailed', run, status, reason }`; `reason` is the run's end reason, else its failed step's | + +A worker turn, `Fractal.step({ id, success }).pipe(Effect.provide(Fractal.Worker.layer({ mission, +input, product })))`, is a child run of `mission` with `input`. Its value is the child's product +`resource/RUN/PRODUCT`, decoded with `success`; the worker mission declares the agent, goal and +product (for example a run-scoped one-shot reviewer `agent/${ST_MISSION_RUN}/reviewer`, so +independence holds as in LandingMission). + +**Asks.** `Fractal.ask({ id, for, title, request, deadline, default })` opens a native person ask on +the program step, with `work ask --step` semantics: the same structured requests, deadlines and +default answers (the fleet's ask-deadline policy applies), and the same inbox. Its requester is +`daemon/program`: a program step has no claimant, so the step's attempt fences the ask instead of a +claim. The step waits on a person (`waiting-person`) while any ask is open; its program stays loaded +and its execution clock pauses (DQ6), and a program may hold several asks at once. The program +receives: + +| Ask | Exit | +|---|---| +| answered by a person | Success `{ answer?: { type, outcome?, id?, label?, text? }, summary?, resolved_by }`, `resolved_by` the person | +| resolved at its deadline | the same, with the default answer (none for a free-text ask) and `resolved_by: 'deadline'` | +| withdrawn while the run goes on | Failure `{ _tag: 'AskCancelled', ask }` | -Cancellation is eager: when the reconciler terminates a program step (run cancelled, timed out), it -tells the engine, which sends `cancel` to the host; the host interrupts the program and kills its -effects' process groups. +An ask that ends with its run is never resolved: the program ended with it. -`Fractal.ask` is a native person ask on the program step (`work ask --step` semantics, deadlines and -defaults), resolved through the same journal key. +**Cancellation.** Cancelling the parent run cancels its children through +`cancel_descendant_mission_runs_tx` (R08), and its asks end with the run. When a program step ends, +fails, times out or moves to a retried attempt while its run goes on, the engine cancels the child +runs and asks that attempt journaled as `child` and holds no result for, in the same evaluation that +sends the host `cancel`. A program that ends while a child run or ask it started is still open (a +lost race) cancels it the same way. ### 7. Serialized sections @@ -296,7 +328,9 @@ Two concurrent landings must not publish one host at once (R09). | Coalescing | the owner can skip superseded heads | a queued publish for an older head finds the host at or past it and is a no-op | | Cost | lane bookkeeping (0.03 s) plus an owner | child run start (no agent: one reconciler pass) | -Choice: a child run. It needs no standing owner and reuses §6. +Choice: a child run. It needs no standing owner and reuses §6: a publication whose start finds the +publish mission's one run active waits, journaled as `started` without a `child`, and starts in the +evaluation the active run's end triggers. ### 8. Visibility and stall detection @@ -339,15 +373,19 @@ command (`Fractal.command(argv, { priority: 'background' })`); this is the progr ### 11. Program API -The author-facing API is the `Fractal` namespace of the `fractal` package (D7), on `effect/workflow`: +The author-facing API is the `Fractal` namespace of the `fractal` package's `fractal/effect` entry +point (`import { Fractal } from 'fractal/effect'`, D7), on `effect/workflow`. The entry point's other +named exports (`ProgramEntry`, `HostContextV1`, `ProgramExit`, `ChildRun`, `ChildFailed`, +`AskAnswer`, `AskCancelled`) are the host contract a bundle implements and the exits it decodes. | `Fractal` | `effect/workflow` | Engine | |---|---|---| | `Fractal.mission(intent, program)` | `Workflow.make` with `intent.input` as payload and `intent.output` as success schema | one `program` step; the bundle's entry runs the workflow body for the step run's current attempt | | `Fractal.step({ id, success, error }, effect)` | `Activity.make` named by the step's key (`id`, `id#2`, …) | a `program.step` entry under that key, replayed by the key | -| `Fractal.ask({ id, for, request, deadline, default })` | `DurableDeferred.make` named by the ask's key | a native person ask on the program step; the answer is journaled under the key | -| `Fractal.run(mission, input)` | a child workflow execution | a linked child run (§6), its terminal result journaled under the key | -| — | `WorkflowEngine` | the host's engine: `activityExecute` returns a journaled result or runs the activity and records it through the host protocol; `deferredResult` reads the journal | +| `Fractal.step({ id, success }).pipe(Effect.provide(Fractal.Worker.layer({ mission, input, product })))` | `DurableDeferred.make` named by the key | a worker turn: `ctx.child` (protocol `run`) of `mission` with `input`; the value is the child's product `resource/RUN/PRODUCT` decoded with `success` (§6) | +| `Fractal.ask({ id, for, title, request, deadline, default })` | `DurableDeferred.make` named by the ask's key | `ctx.ask` (protocol `ask`): a native person ask on the program step; the answer is journaled under the key, `AskCancelled` when it is withdrawn (§6) | +| `Fractal.run(mission, input, { id })` | `DurableDeferred.make` named by the key (default: the mission's id) | `ctx.child` (protocol `run`): a linked child run; its end is journaled under the key, `ChildFailed` when it does not complete (§6) | +| — | `WorkflowEngine` | the host's engine: `activityExecute` returns a journaled result or runs the activity and records it through the host protocol; `deferredResult` reads the journal, else waits for the daemon's `resolved` | Inputs and outputs are Effect Schemas. `intent.input` is a Struct: each field renders as a native mission `input`, `missions start` checks the inputs against it, and the program receives the decoded @@ -375,7 +413,8 @@ program to the same wait. journals, and whether replay checks the body's step sequence against the journal and refuses a mismatch. - **DQ6 Execution clock:** whether time a program waits on a child run or an ask stops the step's - timeout clock, as `work wait` does for seats. + timeout clock, as `work wait` does for seats. Today an open ask pauses it (the step is + `waiting-person`) and a child run does not. - **DQ7 Native graph retirement:** whether native step-graph execution is retired once every data-DSL mission lowers to a program. - **DQ8 Program liveness:** decided: the host pings each program's worker; a worker that does not diff --git a/release-notes/program-engine-children.json b/release-notes/program-engine-children.json new file mode 100644 index 00000000000..a1dca2c3e5e --- /dev/null +++ b/release-notes/program-engine-children.json @@ -0,0 +1,11 @@ +{ + "version": 1, + "summary": "Program steps wait on linked child runs and native person asks: the run's evaluation starts or adopts a child (`create_child_mission_run`, key program:STEP:ATTEMPT:KEY) or asks on the program step as `daemon/program`, journals the result under the program's key and resolves the program; a full mission's start waits until a run holding it ends; a step that ends, fails or retries cancels the children and asks it still waits on", + "replay": {"status": "changed", "detail": "`program.step` claims gain `started` entries for child-run and ask keys, one when the request arrives and one naming the child run or ask in `child`; their results are completed/failed entries with `child` set, fenced like other results (program:STEP:ATTEMPT:KEY:result). Child runs carry idempotency key program:STEP:ATTEMPT:KEY and the program step as parent_step_run; asks are `work.person-asked` claims whose actor is `daemon/program`."}, + "schema": {"status": "none", "detail": "No table or migration; the `child` field of `program.step` was registered with the claim kind."}, + "rules": {"status": "changed", "detail": "`ask_person` admits the requester `daemon/program` for an agentless step's current attempt, given as the ask's incarnation, without a claim; only the daemon acts as `daemon/*`. Agent asks keep the claimant fence."}, + "compatibility": {"status": "changed", "detail": "A program step that waits on an ask is `waiting-person`; its program stays loaded and its timeout clock pauses, as for a worker's ask. Ask deadlines and default answers follow the fleet's ask-deadline policy: a program that names a deadline or default while deadlines are off fails its step with `ask-deadlines-off`."}, + "downtime": {"status": "none", "detail": "No restart or migration beyond a normal deploy; a program waiting on a child run or ask resumes and adopts it."}, + "manual": {"status": "none", "detail": "No operator action."}, + "recovery": {"status": "none", "detail": "A child run or ask is adopted by key after a host or daemon restart; a key already journaled replays without a new request."} +} From 2a62088b1fb7f114eca19fce31c2cc0699cbc20a Mon Sep 17 00:00:00 2001 From: Johannes Schickling Date: Tue, 10 Nov 2026 16:04:57 +0100 Subject: [PATCH 2/4] Borrow test values the program-children assertions reuse --- crates/fractal-reconcile/src/reconcile.rs | 2 +- .../src/reconcile/tests/program_children.rs | 8 ++++---- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/crates/fractal-reconcile/src/reconcile.rs b/crates/fractal-reconcile/src/reconcile.rs index af5d8306e78..2fa7707b1c1 100644 --- a/crates/fractal-reconcile/src/reconcile.rs +++ b/crates/fractal-reconcile/src/reconcile.rs @@ -14533,7 +14533,7 @@ impl Reconciler { fn program_child_products(&self, child: &MissionRunView) -> Result> { let mut products = serde_json::Map::new(); - let Some(spec) = self.store.mission_spec(&child.mission, Some(&child.revision))? else { + let Some(spec) = self.store.mission_spec(&child.mission, Some(child.revision.as_str()))? else { return Ok(products); }; for step in flatten_mission_steps(&spec) { diff --git a/crates/fractal-reconcile/src/reconcile/tests/program_children.rs b/crates/fractal-reconcile/src/reconcile/tests/program_children.rs index 8ee99d11131..827d5a78e76 100644 --- a/crates/fractal-reconcile/src/reconcile/tests/program_children.rs +++ b/crates/fractal-reconcile/src/reconcile/tests/program_children.rs @@ -260,10 +260,10 @@ fn a_child_run_is_adopted_on_resume_and_its_result_is_journaled_and_replayed() { fixture.complete_child(&child); fixture.until("the child's end is journaled", || fixture.result(&parent.step, "publish", 1).is_some()); let completed = json!({ - "run": child, + "run": child.as_str(), "status": "completed", "products": { - format!("resource/{child}/published"): {"kind": "custom.st3.program-test", "head": "h1"}, + (format!("resource/{child}/published")): {"kind": "custom.st3.program-test", "head": "h1"}, }, }); assert_eq!(fixture.result(&parent.step, "publish", 1), Some(("completed".to_owned(), completed.clone()))); @@ -414,7 +414,7 @@ fn asks_are_answered_defaulted_or_withdrawn_and_their_answers_replay() { fixture.reconciler.evaluate_changed_mission_runs().unwrap(); assert_eq!( fixture.result(&parent.step, "withdrawn", 1), - Some(("failed".to_owned(), json!({"_tag": "AskCancelled", "ask": withdrawn}))) + Some(("failed".to_owned(), json!({"_tag": "AskCancelled", "ask": withdrawn.as_str()}))) ); // The deadline resolves the last ask with its default answer. fixture.store.reconcile_person_asks_at(now_ms() + 3 * 60_000).unwrap(); @@ -428,5 +428,5 @@ fn asks_are_answered_defaulted_or_withdrawn_and_their_answers_replay() { assert_eq!(fixture.result(&parent.step, "approve", 1), Some(("completed".to_owned(), defaulted.clone()))); let (_, value) = fixture.result(&parent.step, RESULT_KEY, 1).unwrap(); assert_eq!(value["asks"][0], json!({"_tag": "Success", "value": defaulted})); - assert_eq!(value["asks"][2], json!({"_tag": "Failure", "error": {"_tag": "AskCancelled", "ask": withdrawn}})); + assert_eq!(value["asks"][2], json!({"_tag": "Failure", "error": {"_tag": "AskCancelled", "ask": withdrawn.as_str()}})); } From 450b819d821ef90e8c45b09ef36cdebb62091df3 Mon Sep 17 00:00:00 2001 From: Johannes Schickling Date: Tue, 10 Nov 2026 16:22:54 +0100 Subject: [PATCH 3/4] Check a left step's child for its cancellation, then its end --- .../src/reconcile/tests/program_children.rs | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/crates/fractal-reconcile/src/reconcile/tests/program_children.rs b/crates/fractal-reconcile/src/reconcile/tests/program_children.rs index 827d5a78e76..8ee4e1e4232 100644 --- a/crates/fractal-reconcile/src/reconcile/tests/program_children.rs +++ b/crates/fractal-reconcile/src/reconcile/tests/program_children.rs @@ -323,16 +323,18 @@ fn children_are_cancelled_with_their_run_or_their_step_and_a_retry_starts_a_new_ let retried = fixture.start("retried", "child"); fixture.until("the second parent's child starts", || fixture.child(&retried.step, "publish", 1).is_some()); let attempt_one = fixture.child(&retried.step, "publish", 1).unwrap(); - // The program step fails while its run goes on (it has another attempt): its child is - // cancelled in that evaluation. + // The program step fails while its run goes on (it has another attempt): its child's + // cancellation is requested in that evaluation. fixture.store.set_step_state(&retried.step, "failed", Some("the active execution timeout expired")).unwrap(); fixture.reconciler.evaluate_changed_mission_runs().unwrap(); - assert_eq!(fixture.run_status(&attempt_one), "cancelled"); + let cancelling = fixture.store.mission_run(&attempt_one).unwrap().unwrap(); + assert_ne!((cancelling.status.as_str(), cancelling.phase.as_str()), ("running", "normal")); assert_eq!(fixture.run_status(&retried.run), "running"); // The retried attempt starts from the top with a new child. fixture.until("attempt 2 starts its child", || fixture.child(&retried.step, "publish", 2).is_some()); let attempt_two = fixture.child(&retried.step, "publish", 2).unwrap(); assert_ne!(attempt_two, attempt_one); + assert_eq!(fixture.run_status(&attempt_one), "cancelled"); assert_eq!(fixture.result(&retried.step, "publish", 1), None); // A child that fails is a typed failure in its parent. let child_step = fixture.store.mission_run(&attempt_two).unwrap().unwrap().steps.remove(0).subject; From a7385565ac8d82c48e3cf15f9795c1fdd83fa611 Mon Sep 17 00:00:00 2001 From: Johannes Schickling Date: Tue, 10 Nov 2026 16:35:14 +0100 Subject: [PATCH 4/4] Record and cancel late program children, retry store failures, bound the program asker --- crates/fractal-api-core/src/api/work.rs | 13 ++++++ .../fractal-missions/src/store/person_work.rs | 23 +++++++++- crates/fractal-reconcile/src/reconcile.rs | 18 ++++++-- .../src/reconcile/program.rs | 45 ++++++++++++------- .../src/reconcile/tests/program_children.rs | 23 +++++++++- docs/st3/program-engine.md | 7 ++- 6 files changed, 106 insertions(+), 23 deletions(-) diff --git a/crates/fractal-api-core/src/api/work.rs b/crates/fractal-api-core/src/api/work.rs index 75a2e357a28..46b700e7ab7 100644 --- a/crates/fractal-api-core/src/api/work.rs +++ b/crates/fractal-api-core/src/api/work.rs @@ -40,11 +40,23 @@ fn bind_work_incarnation( Ok(current) } +/// Only the daemon's program engine asks, or cancels an ask, as the program asker. +fn refuse_program_asker(actor: &str) -> Result<(), ApiError> { + if actor == crate::model::PROGRAM_ASKER { + return Err(ApiError::bad(FractalError::new( + "reserved-actor", + format!("`{actor}` is the daemon's program engine; a request cannot act as it"), + ))); + } + Ok(()) +} + pub async fn ask_person( State(state): State, bound: Option>, Json(mut request): Json, ) -> Result, ApiError> { + refuse_program_asker(&request.actor)?; let expected = bind_work_incarnation( &state, bound.as_ref(), @@ -113,6 +125,7 @@ pub async fn cancel_person_ask( State(state): State, Json(request): Json, ) -> Result, ApiError> { + refuse_program_asker(&request.actor)?; let result = state .store .finish_person_step(&request, true) diff --git a/crates/fractal-missions/src/store/person_work.rs b/crates/fractal-missions/src/store/person_work.rs index 2af70aa1807..e91d0dc534c 100644 --- a/crates/fractal-missions/src/store/person_work.rs +++ b/crates/fractal-missions/src/store/person_work.rs @@ -354,6 +354,25 @@ pub struct PersonAskPass { pub next_deadline: Option, } +/// Whether `step`'s definition in its generation's revision runs a program. +fn runs_program(tx: &Transaction<'_>, step: &StepRunView) -> Result { + let body: Option = tx + .query_row( + "SELECT r.body FROM run_generations g + JOIN mission_runs m ON m.id=?2 + JOIN mission_revisions r ON r.mission_id=m.mission_id AND r.revision=g.revision + WHERE g.id=?1", + params![generation_id_from_subject(&step.generation), step.run.strip_prefix("mission-run/").unwrap_or(&step.run)], + |row| row.get(0), + ) + .optional()?; + let Some(body) = body else { return Ok(false) }; + let mission: crate::model::MissionSpec = serde_json::from_str(&body)?; + Ok(fractal_store::store::flatten_mission_step_specs(&mission) + .into_iter() + .any(|spec| spec.path == step.step && spec.program.is_some())) +} + pub trait PersonWorkStore { /// A daemon-owned approval ask, projected in the same transaction as its durable claim. fn ask_person(&self, input: &PersonAskRequest) -> Result; @@ -491,11 +510,13 @@ impl PersonWorkStore for Store { && matches!(origin.status.as_str(), "ready" | "blocked" | "claimed" | "working")) }).transpose().map_err(internal)?.unwrap_or(false); // A program step has no claimant: the daemon asks for the step's current attempt, - // which the request names as its incarnation (docs/st3/program-engine.md §6). + // which the request names as its incarnation (docs/st3/program-engine.md §6). Only a + // step whose definition runs a program may be asked from this way. let stale = if program { !origin.agentless || !matches!(origin.status.as_str(), "ready" | "working" | "blocked" | "waiting-person") || input.incarnation.as_deref() != Some(origin.attempt.to_string().as_str()) + || !runs_program(tx, &origin).map_err(internal)? } else { !declaration_live(tx, &input.actor).map_err(internal)? || (!legacy_origin && (!matches!(origin.status.as_str(), "claimed" | "working") diff --git a/crates/fractal-reconcile/src/reconcile.rs b/crates/fractal-reconcile/src/reconcile.rs index 2fa7707b1c1..460b2546bbc 100644 --- a/crates/fractal-reconcile/src/reconcile.rs +++ b/crates/fractal-reconcile/src/reconcile.rs @@ -14447,17 +14447,27 @@ impl Reconciler { let idempotency_key = request.idempotency_key(); let child = self.store.mission_run_subject_for_idempotency_key(&idempotency_key); let Some(status) = self.store.mission_run_status(&child)? else { + // The runs holding the mission's capacity are read before the start and again after a + // refusal, so the end of any holder, even one that ends in between, evaluates this run. + let note_holders = || -> Result<()> { + for holder in self.store.active_mission_runs_for_mission(mission)? { + smallclaims::touched::note_read(|| holder.subject.clone()); + } + Ok(()) + }; + note_holders()?; return match self.start_program_child(run, &request.step, mission, input, idempotency_key) { Ok(started) => { engine.waiting(request, &started.subject)?; Ok(true) } Err(error) if error.code == "mission-run-capacity" => { - for holder in self.store.active_mission_runs_for_mission(mission)? { - smallclaims::touched::note_read(|| holder.subject.clone()); - } + note_holders()?; Ok(false) } + // A store failure is retried by the next evaluation; only a refused request is + // the program's defect. + Err(error) if error.code == "internal" => Err(error.into()), Err(error) => { let reason = format!( "the program could not start its child run `{}` of mission `{mission}`: {}", @@ -14566,6 +14576,8 @@ impl Reconciler { engine.waiting(request, &view.subject)?; (view.subject, true) } + // A store failure is retried by the next evaluation. + Err(error) if error.code == "internal" => return Err(error.into()), Err(error) => { let reason = format!("the program's ask `{}` was refused: {}", request.key, error.message); engine.fail(&request.step, request.attempt, &reason)?; diff --git a/crates/fractal-reconcile/src/reconcile/program.rs b/crates/fractal-reconcile/src/reconcile/program.rs index 5740478a6b0..3516174af2b 100644 --- a/crates/fractal-reconcile/src/reconcile/program.rs +++ b/crates/fractal-reconcile/src/reconcile/program.rs @@ -123,6 +123,8 @@ pub struct Request { pub wait: Wait, /// The child run or ask subject once it exists. pub child: Option, + /// The bundle hash of the program that asked, which journals the key's entries. + program: String, seq: Value, since: Instant, } @@ -287,16 +289,24 @@ impl ProgramEngine { .collect() } - /// The child run or ask `request` waits on now exists: journal it as the key's `child`. + /// The child run or ask `request` waits on now exists: journal it as the key's `child`, even + /// when the program stopped waiting meanwhile, so nothing it started goes unrecorded. When its + /// step has left the attempt by then, the step's cancellation (which may have read the journal + /// before this entry) did not see it: cancel it here. pub fn waiting(&self, request: &Request, child: &str) -> Result<()> { - let Some(program) = self.program_of(request, |running| { - if let Some(pending) = running.requests.get_mut(&request.key) { - pending.child = Some(child.to_owned()); - } - }) else { - return Ok(()); - }; - self.write_journal(&request.step, request.attempt, &program, &request.key, "started", &Value::Null, Some(child))?; + self.write_journal(&request.step, request.attempt, &request.program, &request.key, "started", &Value::Null, Some(child))?; + let running = self + .program_of(request, |running| { + if let Some(pending) = running.requests.get_mut(&request.key) { + pending.child = Some(child.to_owned()); + } + }) + .is_some(); + // A program removed by its host's exit is resumed and adopts the child; one that ended or + // left its attempt (each sets the step's state before it cancels) does not. + if !running && let Some(reason) = self.left(&request.step, request.attempt)? { + self.cancel_waits(&request.step, request.attempt, &reason)?; + } Ok(()) } @@ -354,10 +364,11 @@ impl ProgramEngine { } } eprintln!("fractal: program {step} attempt {attempt} failed: {reason}"); - self.cancel_waits(step, attempt, reason)?; + // The step's state first: a child started concurrently then sees it left (`waiting`). if self.left(step, attempt)?.is_none() { self.store.set_step_state_unless_terminal(step, "failed", Some(reason))?; } + self.cancel_waits(step, attempt, reason)?; (self.wake)(step); Ok(()) } @@ -672,6 +683,7 @@ impl ProgramEngine { key: key.to_owned(), wait, child: None, + program: program.to_owned(), seq: message["seq"].clone(), since: Instant::now(), }; @@ -780,11 +792,8 @@ impl ProgramEngine { if !forget(&mut self.lock(), step, attempt, generation) { return Ok(()); } - // A program can end while a child run or ask it started is still open (a lost race): - // nothing waits on it any more. - let left = self.left(step, attempt)?; - self.cancel_waits(step, attempt, left.as_deref().unwrap_or("the program ended"))?; - if left.is_some() { + if let Some(reason) = self.left(step, attempt)? { + self.cancel_waits(step, attempt, &reason)?; return Ok(()); } let exit = &message["exit"]; @@ -796,6 +805,10 @@ impl ProgramEngine { _ => ("failed", Some(failure_reason(&exit["error"]))), }; self.store.set_step_state_unless_terminal(step, status, reason.as_deref())?; + // A program can end while a child run or ask it started is still open (a lost race): + // nothing waits on it any more. The step's state comes first, so a child started + // concurrently sees it left (`waiting`). + self.cancel_waits(step, attempt, "the program ended")?; eprintln!("fractal: program {step} {status}"); (self.wake)(step); Ok(()) @@ -814,10 +827,10 @@ impl ProgramEngine { inner.faults.insert(step.to_owned(), (attempt, reason.clone())); } eprintln!("fractal: program {step} attempt {attempt}: {reason}"); - self.cancel_waits(step, attempt, &reason)?; if self.left(step, attempt)?.is_none() { self.store.set_step_state_unless_terminal(step, "failed", Some(&reason))?; } + self.cancel_waits(step, attempt, &reason)?; (self.wake)(step); Ok(()) } diff --git a/crates/fractal-reconcile/src/reconcile/tests/program_children.rs b/crates/fractal-reconcile/src/reconcile/tests/program_children.rs index 8ee4e1e4232..b80d6e86517 100644 --- a/crates/fractal-reconcile/src/reconcile/tests/program_children.rs +++ b/crates/fractal-reconcile/src/reconcile/tests/program_children.rs @@ -4,7 +4,7 @@ //! `fractal/effect` program API decodes. These tests fail, never skip, when no `bun` is on `PATH`. use super::*; use crate::model::{AnswerInput, PROGRAM_ASKER, PersonStepResponse}; -use crate::reconcile::program::{JOURNAL_KIND, Liveness, ProgramEngine, RESULT_KEY}; +use crate::reconcile::program::{JOURNAL_KIND, Liveness, ProgramEngine, RESULT_KEY, Wait}; use serde_json::json; use std::time::Instant; @@ -308,6 +308,27 @@ fn a_child_of_a_full_mission_waits_and_starts_when_the_holder_ends() { fixture.until("the program completes", || fixture.status(&parent.step) == "completed"); } +#[test] +fn a_child_started_after_its_step_left_is_journaled_and_cancelled() { + let fixture = fixture(); + let holder = fixture.holder(); + let parent = fixture.start("late", "child"); + fixture.until("the program asks for its child", || !fixture.engine.requests(&parent.run).is_empty()); + // An evaluation's view of the request, taken before the step leaves its attempt. + let request = fixture.engine.requests(&parent.run).remove(0); + fixture.store.set_step_state(&parent.step, "failed", Some("the active execution timeout expired")).unwrap(); + fixture.reconciler.evaluate_changed_mission_runs().unwrap(); + assert!(fixture.engine.requests(&parent.run).is_empty()); + // That evaluation now starts the child, after the step's own cancellation found nothing. + fixture.store.request_mission_run_cancellation(&holder, "make room").unwrap(); + let run = fixture.store.mission_run(&parent.run).unwrap().unwrap(); + let Wait::Run { mission, input } = &request.wait else { panic!("a child run request") }; + assert!(fixture.reconciler.settle_program_child(&fixture.engine, &run, &request, mission, input).unwrap()); + let late = fixture.child(&parent.step, "publish", 1).expect("the late child is journaled"); + let late = fixture.store.mission_run(&late).unwrap().unwrap(); + assert_ne!((late.status.as_str(), late.phase.as_str()), ("running", "normal"), "the late child is cancelled"); +} + #[test] fn children_are_cancelled_with_their_run_or_their_step_and_a_retry_starts_a_new_one() { let fixture = fixture(); diff --git a/docs/st3/program-engine.md b/docs/st3/program-engine.md index a74ada6a81b..05b63c4adaf 100644 --- a/docs/st3/program-engine.md +++ b/docs/st3/program-engine.md @@ -272,7 +272,9 @@ does the rest: it starts or adopts what the key waits on, journals a second `sta child run or answers an ask evaluates the parent run again; nothing polls. A replayed key never asks again; a resumed attempt that asks for an open key adopts what that key already waits on. A request the daemon refuses (an unknown mission, inputs the mission does not accept, an ask without -a person) fails the program step with the reason, as a defect would. +a person) fails the program step with the reason, as a defect would; a store failure leaves the +step as it is, and the next evaluation tries again. A child run or ask is journaled as `child` even +when it starts just after its program stopped waiting, and is then cancelled with the attempt. **Child runs.** The evaluation calls `create_child_mission_run`, the path `uses-mission` takes: `parent_step_run` is the program step, the inputs are the request's, and the idempotency key is @@ -297,7 +299,8 @@ independence holds as in LandingMission). the program step, with `work ask --step` semantics: the same structured requests, deadlines and default answers (the fleet's ask-deadline policy applies), and the same inbox. Its requester is `daemon/program`: a program step has no claimant, so the step's attempt fences the ask instead of a -claim. The step waits on a person (`waiting-person`) while any ask is open; its program stays loaded +claim. Only a step whose definition runs a program accepts that requester, and the work API refuses +requests that name it. The step waits on a person (`waiting-person`) while any ask is open; its program stays loaded and its execution clock pauses (DQ6), and a program may hold several asks at once. The program receives: