Program engine M1.2: linked child runs and person asks for program steps

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<AppState>,
     bound: Option<Extension<BoundAgent>>,
     Json(mut request): Json<PersonAskRequest>,
 ) -> Result<Json<StepRunView>, 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<AppState>,
     Json(request): Json<PersonStepResponse>,
 ) -> Result<Json<StepRunView>, ApiError> {
+    refuse_program_asker(&request.actor)?;
     let result = state
         .store
         .finish_person_step(&request, true)

crates/fractal-kernel/src/model.rs

@@ -3492,6 +3492,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)]

crates/fractal-missions/src/store/person_work.rs

@@ -354,6 +354,25 @@ pub struct PersonAskPass {
     pub next_deadline: Option<u128>,
 }
 
+/// Whether `step`'s definition in its generation's revision runs a program.
+fn runs_program(tx: &Transaction<'_>, step: &StepRunView) -> Result<bool> {
+    let body: Option<String> = 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<StepRunView, FractalError>;
@@ -418,11 +437,14 @@ pub trait PersonWorkStore {
 impl PersonWorkStore for Store {
 
     fn ask_person(&self, input: &PersonAskRequest) -> Result<StepRunView, FractalError> {
+        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 +509,22 @@ 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). 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")
+                    || 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("<none>"), origin.claim_incarnation.as_deref().unwrap_or("<none>"), input.incarnation.as_deref().unwrap_or("<none>"))));

crates/fractal-reconcile/src/reconcile.rs

@@ -10607,9 +10607,13 @@ impl<R: RuntimeControl> Reconciler<R> {
                 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")
@@ -10650,7 +10654,9 @@ impl<R: RuntimeControl> Reconciler<R> {
                             }
                         }
                     }
-                    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))
@@ -11237,6 +11243,14 @@ impl<R: RuntimeControl> Reconciler<R> {
                         }
                         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"
@@ -14421,6 +14435,198 @@ impl<R: RuntimeControl> Reconciler<R> {
         }
     }
 
+    /// 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<bool> {
+        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<String, String>,
+    ) -> Result<bool> {
+        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" => {
+                    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}`: {}",
+                        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<String, String>,
+        idempotency_key: String,
+    ) -> std::result::Result<MissionRunView, crate::model::FractalError> {
+        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<Value> {
+        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<serde_json::Map<String, Value>> {
+        let mut products = serde_json::Map::new();
+        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) {
+            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<bool> {
+        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)
+                }
+                // 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)?;
+                    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,
@@ -21246,6 +21452,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};