Program engine M1.1a: declare program steps and carry their bundles through apply

crates/fractal-api-client-v0/src/api/client_v0.rs

@@ -7770,6 +7770,7 @@ async fn apply_runtime_control_intent(
             expected_subjects: preview.subject_tokens,
             idempotency_key: format!("{}:runtime-intent", request.idempotency_key),
             actor: Some(actor.to_owned()),
+            program_bundles: Default::default(),
         }),
     )
     .await?

crates/fractal-api-core/src/api.rs

@@ -733,8 +733,8 @@ pub fn router_for_transport(state: AppState, transport: ClientTransportBoundary,
         .route("/v1/gate-checks", post(runs::start_gate_check))
         .route("/v1/gate-checks/{id}", get(runs::read_gate_check))
         .route("/v1/sets/preview", post(owned_sets::preview))
-        .route("/v1/sets/apply", post(owned_sets::apply))
-        .route("/v1/intent/apply", post(launches::apply_with_bound))
+        .route("/v1/sets/apply", post(owned_sets::apply).layer(DefaultBodyLimit::max(PUBLICATION_BODY_LIMIT)))
+        .route("/v1/intent/apply", post(launches::apply_with_bound).layer(DefaultBodyLimit::max(PUBLICATION_BODY_LIMIT)))
         .route("/v1/agents/rename", post(agents::rename_agent))
         .route("/v1/persons/rename", post(agents::rename_person))
         .route("/v1/agents/rename-id", post(agents::rename_agent_id))
@@ -8253,7 +8253,8 @@ pub async fn guard_bound_request(
         return Ok(request);
     }
     let (parts, body) = request.into_parts();
-    let bytes = axum::body::to_bytes(body, 16 * 1024 * 1024)
+    // A publication carries program bundles, so the buffer matches the publication route limit.
+    let bytes = axum::body::to_bytes(body, PUBLICATION_BODY_LIMIT)
         .await
         .map_err(ApiError::internal)?;
     if let Ok(value) = serde_json::from_slice::<Value>(&bytes) {
@@ -8295,7 +8296,7 @@ async fn guard_unverifiable_request(request: Request<Body>) -> Result<Request<Bo
         return Ok(request);
     }
     let (parts, body) = request.into_parts();
-    let bytes = axum::body::to_bytes(body, 16 * 1024 * 1024).await.map_err(ApiError::internal)?;
+    let bytes = axum::body::to_bytes(body, PUBLICATION_BODY_LIMIT).await.map_err(ApiError::internal)?;
     if let Ok(value) = serde_json::from_slice::<Value>(&bytes)
         && ["actor", "requester"].iter().any(|key| value.get(key).is_some_and(|v| !v.is_null()))
     {
@@ -11213,6 +11214,7 @@ async fn apply_with_authority(
     if let Some(error) = publication_refusals(&state, &intent).await?.error() {
         return Err(ApiError::bad(error));
     }
+    store_program_bundles(&state, &intent, &request.program_bundles)?;
     let mut response = state
         .store
         .apply_as(
@@ -11227,6 +11229,46 @@ async fn apply_with_authority(
     Ok(Json(response))
 }
 
+/// The request body a publication may use: program bundles travel inside it as base64.
+pub const PUBLICATION_BODY_LIMIT: usize = 32 * 1024 * 1024;
+
+/// Verify each program bundle a publication carries against the hash its intent names, then stage
+/// the bytes as local blobs; the `mission.published` claim that names the hash promotes them
+/// (docs/st3/program-engine.md §4). A bundle the intent does not name is refused, so a publication
+/// cannot stage arbitrary bytes, and nothing is staged unless every bundle checks out.
+pub(crate) fn store_program_bundles(
+    state: &AppState,
+    intent: &crate::model::NormalizedIntent,
+    bundles: &BTreeMap<String, String>,
+) -> Result<(), ApiError> {
+    use base64::Engine as _;
+    let named = intent.missions.values().flat_map(|mission| mission.programs()).map(|(_, program)| program.bundle_hash.as_str()).collect::<BTreeSet<_>>();
+    let mut verified = Vec::with_capacity(bundles.len());
+    for (hash, encoded) in bundles {
+        if !named.contains(hash.as_str()) {
+            return Err(ApiError::bad(FractalError::new(
+                "unused-program-bundle",
+                format!("the publication carries program bundle {hash}, which none of its programs names"),
+            )));
+        }
+        let bytes = base64::engine::general_purpose::STANDARD.decode(encoded).map_err(|error| {
+            ApiError::bad(FractalError::new("invalid-program-bundle", format!("program bundle {hash} is not base64: {error}")))
+        })?;
+        let actual = hex::encode(Sha256::digest(&bytes));
+        if actual != *hash {
+            return Err(ApiError::bad(FractalError::new(
+                "program-bundle-hash-mismatch",
+                format!("program bundle {hash} has sha256 {actual}"),
+            )));
+        }
+        verified.push(bytes);
+    }
+    for bytes in verified {
+        state.store.put_blob(&bytes).map_err(ApiError::internal)?;
+    }
+    Ok(())
+}
+
 /// One write a rule in audit mode would have refused.
 #[derive(Clone, Debug, Deserialize, Serialize)]
 pub struct RuleAudit {

crates/fractal-api-core/src/api/owned_sets.rs

@@ -53,6 +53,7 @@ pub async fn apply(
     if let Some(error) = publication_refusals(&state, &intent).await?.error() {
         return Err(ApiError::bad(error));
     }
+    store_program_bundles(&state, &intent, &request.program_bundles)?;
     let response = state
         .store
         .apply_owned_set(

crates/fractal-api-core/src/api/tests.rs

@@ -4254,6 +4254,7 @@ fn apply_request(state: &AppState, source: &str, actor: &str, key: &str) -> Appl
         expected_subjects: preview.subject_tokens,
         idempotency_key: key.into(),
         actor: Some(actor.into()),
+        program_bundles: Default::default(),
     }
 }
 
@@ -6665,6 +6666,14 @@ async fn mission_resolves_a_bare_document_to_immutable_bytes() {
 }
 
 async fn preview_and_apply(app: Router, kdl: &str) -> (Value, StatusCode, Value) {
+    preview_and_apply_with_bundles(app, kdl, BTreeMap::new()).await
+}
+
+async fn preview_and_apply_with_bundles(
+    app: Router,
+    kdl: &str,
+    program_bundles: BTreeMap<String, String>,
+) -> (Value, StatusCode, Value) {
     let (status, preview) = json_request(
         app.clone(),
         "/v1/intent/mission",
@@ -6691,13 +6700,82 @@ async fn preview_and_apply(app: Router, kdl: &str) -> (Value, StatusCode, Value)
                 .unwrap(),
             idempotency_key: format!("apply-{}", hex::encode(Sha256::digest(kdl))),
             actor: Some("person/alex".into()),
+            program_bundles,
         })
         .unwrap(),
     )
     .await;
     (preview, status, applied)
 }
 
+/// A publication carries its program bundles as base64, so both caller guards buffer a body up to
+/// the publication route limit (32 MiB) and hand a body past 16 MiB on whole.
+#[tokio::test]
+async fn caller_guards_pass_a_publication_body_past_sixteen_mebibytes() {
+    let root = tempfile::tempdir().unwrap();
+    let state = state(root.path());
+    let padding = "a".repeat(20 * 1024 * 1024);
+    for (bound, body) in [
+        (Some("agent/example/publisher"), format!(r#"{{"actor":"agent/example/publisher","pad":"{padding}"}}"#)),
+        (None, format!(r#"{{"pad":"{padding}"}}"#)),
+    ] {
+        let mut request = Request::builder()
+            .method(axum::http::Method::POST)
+            .uri("/v1/intent/apply")
+            .header("content-type", "application/json")
+            .body(Body::from(body.clone()))
+            .unwrap();
+        if bound.is_none() {
+            request.extensions_mut().insert(HarnessAncestry::Unavailable);
+        }
+        let passed = guard_bound_request(request, bound, &state.store).await.unwrap();
+        let bytes = to_bytes(passed.into_body(), PUBLICATION_BODY_LIMIT).await.unwrap();
+        assert_eq!(bytes.len(), body.len(), "{bound:?}");
+    }
+}
+
+#[tokio::test]
+async fn publication_stores_the_program_bundle_its_declaration_names() {
+    use base64::Engine as _;
+    let root = tempfile::tempdir().unwrap();
+    let state = state(root.path());
+    let store = state.store.clone();
+    let app = router(state);
+    let bytes = b"export default {}";
+    let hash = hex::encode(Sha256::digest(bytes));
+    let kdl = format!(
+        "version 2\nmission \"land\" state=\"ready\" {{\n  goal \"Land.\"\n  step \"program\" {{ program \"/bundles/land.mjs\" sha256=\"{hash}\" }}\n}}"
+    );
+    let encoded = |bytes: &[u8]| base64::engine::general_purpose::STANDARD.encode(bytes);
+    let before = store.index().unwrap();
+    // A declaration whose bytes this member never received would be refused by every replica.
+    let (_, status, error) = preview_and_apply(app.clone(), &kdl).await;
+    assert!(status.is_client_error(), "{error}");
+    assert_eq!(error["code"], "missing-program-bundle");
+    let (_, status, error) =
+        preview_and_apply_with_bundles(app.clone(), &kdl, BTreeMap::from([(hash.clone(), encoded(b"export default 2"))])).await;
+    assert!(status.is_client_error(), "{error}");
+    assert_eq!(error["code"], "program-bundle-hash-mismatch");
+    let (_, status, error) = preview_and_apply_with_bundles(
+        app.clone(),
+        &kdl,
+        BTreeMap::from([(hash.clone(), encoded(bytes)), ("0".repeat(64), encoded(b""))]),
+    )
+    .await;
+    assert!(status.is_client_error(), "{error}");
+    assert_eq!(error["code"], "unused-program-bundle");
+    assert_eq!(store.index().unwrap(), before);
+    assert_eq!(store.get_blob(&hash).unwrap(), None);
+
+    let (_, status, applied) =
+        preview_and_apply_with_bundles(app, &kdl, BTreeMap::from([(hash.clone(), encoded(bytes))])).await;
+    assert_eq!(status, StatusCode::OK, "{applied}");
+    assert_eq!(store.get_blob(&hash).unwrap().as_deref(), Some(&bytes[..]));
+    // The published definition names the blob by `bundle_hash`, which promotes and replicates it.
+    let published = store.latest_claim("mission/land", Some("mission.published")).unwrap().unwrap();
+    assert!(published.body.to_string().contains(&format!("\"bundle_hash\":\"{hash}\"")), "{}", published.body);
+}
+
 #[tokio::test]
 async fn publication_accepts_a_step_assigned_to_an_undeclared_person() {
     let root = tempfile::tempdir().unwrap();
@@ -6938,6 +7016,7 @@ mission "invalid-message" state="ready" {
             expected_subjects: BTreeMap::new(),
             idempotency_key: "reject-invalid-nested-message".into(),
             actor: Some("person/alex".into()),
+            program_bundles: Default::default(),
         })
         .unwrap(),
     )

crates/fractal-kernel/src/model.rs

@@ -872,13 +872,52 @@ pub struct StepSpec {
     /// completes and fails when it fails or is cancelled.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub after_run: Option<String>,
+    /// The program bundle this agentless step runs (docs/st3/program-engine.md). The daemon's
+    /// program engine owns the step's outcome. Omitted when absent, so other steps hash as before.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub program: Option<ProgramSpec>,
     pub gates: Vec<GateSpec>,
     pub nested_mission: Option<Box<MissionSpec>>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub loop_spec: Option<Box<LoopSpec>>,
     pub definition_hash: String,
 }
 
+/// A program step's code: one self-contained ES module, pinned by the SHA-256 of its bytes. The
+/// hash is part of the step's definition, so a mission revision binds exactly one program. The
+/// field is named `bundle_hash` so publishing the declaration promotes the stored blob and every
+/// replica receives it with the declaration; `bundle` is the author's path, kept as provenance.
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+pub struct ProgramSpec {
+    pub bundle: String,
+    pub bundle_hash: String,
+    /// How long the program may go without a new journal entry before its run reports it stalled.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub stalled_after_ms: Option<u64>,
+}
+
+impl MissionSpec {
+    /// Every program this mission declares, with its step's path, including nested missions and
+    /// loop rounds.
+    pub fn programs(&self) -> Vec<(&str, &ProgramSpec)> {
+        let mut programs = Vec::new();
+        for step in self.steps.values() {
+            if let Some(program) = &step.program {
+                programs.push((step.path.as_str(), program));
+            }
+            if let Some(nested) = &step.nested_mission {
+                programs.extend(nested.programs());
+            }
+            if let Some(loop_spec) = &step.loop_spec {
+                for mission in [Some(&loop_spec.round), loop_spec.on_keep.as_ref(), loop_spec.on_discard.as_ref()].into_iter().flatten() {
+                    programs.extend(mission.programs());
+                }
+            }
+        }
+        programs
+    }
+}
+
 /// How the resource edge was authored. Older stored URI edges predate this discriminator.
 #[derive(Clone, Copy, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
 #[serde(rename_all = "snake_case")]
@@ -1782,6 +1821,10 @@ pub struct ApplyRequest {
     pub idempotency_key: String,
     #[serde(default)]
     pub actor: Option<String>,
+    /// The bytes of each program bundle the intent declares, base64 by sha256 (the
+    /// `ProgramSpec::bundle_hash`). The daemon verifies and stores them before it publishes.
+    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
+    pub program_bundles: BTreeMap<String, String>,
 }
 
 #[derive(Clone, Debug, Deserialize, Serialize)]

crates/fractal-missions-base/src/mission.rs

@@ -9,8 +9,8 @@ use crate::declaration_syntax::*;
 use crate::model::{
     AuthorityGrant, AuthorityVerb, BaselineSpec, CompletionSpec, DependencySpec, GateSpec,
     LoopAttentionSpec, LoopExhaustionSpec, LoopSpec, LoopStopSpec, MetricSource, MetricSpec,
-    MissionInputKind, MissionInputSpec, MissionSpec, MissionState, ProductSpec, RetrySpec,
-    RevisionCutover, FractalError, StepSpec, UsedMissionSpec, WorkSelector,
+    MissionInputKind, MissionInputSpec, MissionSpec, MissionState, ProductSpec, ProgramSpec,
+    RetrySpec, RevisionCutover, FractalError, StepSpec, UsedMissionSpec, WorkSelector,
 };
 
 pub const LOOP_ROUND_INPUT: &str = "__st3_loop_round";
@@ -946,6 +946,7 @@ fn parse_loop_group(
         produces_mission: None,
         uses_mission: None,
         after_run: None,
+        program: None,
         gates: Vec::new(),
         nested_mission: None,
         loop_spec: Some(Box::new(loop_spec)),
@@ -1430,6 +1431,7 @@ fn parse_step(
     let mut produces_mission = None;
     let mut uses_mission = None;
     let mut after_run = None;
+    let mut program = None;
     let mut gates = Vec::new();
     let mut nested_mission = None;
     let mut retry = RetrySpec::default();
@@ -1535,6 +1537,7 @@ fn parse_step(
                 "produces-mission" => produces_mission = Some(parse_produced_mission(child)?),
                 "uses-mission" => uses_mission = Some(parse_used_mission(child)?),
                 "after-run" => after_run = Some(parse_after_run(child)?),
+                "program" => program = Some(parse_program(child)?),
                 "gate" => {
                     let gate = crate::gate_parse::parse_gate(child, default_host)?;
                     if matches!(gate, GateSpec::Deadline { .. }) {
@@ -1589,6 +1592,36 @@ fn parse_step(
         }
         agentless = true;
     }
+    if program.is_some() {
+        // The program engine owns the outcome: nothing else may complete or fail the step.
+        if assigned_to.is_some() || !available_to.is_empty() {
+            return Err(FractalError::new(
+                "conflicting-work-selector",
+                format!("step `{path}` runs a program, so it cannot select an agent"),
+            ));
+        }
+        if !gates.is_empty()
+            || !products.is_empty()
+            || produces_mission.is_some()
+            || uses_mission.is_some()
+            || after_run.is_some()
+            || nested_mission.is_some()
+        {
+            return Err(FractalError::new(
+                "program-step-outcome",
+                format!("step `{path}` runs a program, which alone decides its outcome; remove its gates, products, nested or used missions and after-run"),
+            ));
+        }
+        if finally {
+            // Cancellation cleanup ends agentless final work that has no runtime process; a
+            // program compensates inside its body instead.
+            return Err(FractalError::new(
+                "program-step-outcome",
+                format!("step `{path}` runs a program, so it cannot be a finally step; compensate inside the program"),
+            ));
+        }
+        agentless = true;
+    }
     if agentless
         && gates
             .iter()
@@ -1639,6 +1672,7 @@ fn parse_step(
         produces_mission,
         uses_mission,
         after_run,
+        program,
         gates,
         nested_mission,
         loop_spec: None,
@@ -2435,6 +2469,31 @@ fn parse_after_run(node: &KdlNode) -> Result<String, FractalError> {
     Ok(run)
 }
 
+/// `program "/abs/bundle.mjs" sha256="HEX" stalled-after="10m"?`: the bundle's path, the exact
+/// bytes it must have, and how long it may go without journal progress.
+fn parse_program(node: &KdlNode) -> Result<ProgramSpec, FractalError> {
+    ensure_only_properties(node, &["sha256", "stalled-after"])?;
+    let bundle = first_string(node)?;
+    if !bundle.starts_with('/') {
+        return Err(FractalError::new(
+            "invalid-program",
+            format!("program bundle `{bundle}` must be an absolute path"),
+        ));
+    }
+    let bundle_hash = property_string(node, "sha256")?.unwrap_or_default();
+    if bundle_hash.len() != 64 || !bundle_hash.bytes().all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)) {
+        return Err(FractalError::new(
+            "invalid-program",
+            "a program needs sha256=\"<64 lowercase hex digits>\" of its bundle",
+        ));
+    }
+    let stalled_after_ms = property_string(node, "stalled-after")?.map(|value| parse_duration(&value)).transpose()?;
+    if stalled_after_ms == Some(0) {
+        return Err(FractalError::new("invalid-program", "a program's stalled-after must be greater than zero"));
+    }
+    Ok(ProgramSpec { bundle, bundle_hash, stalled_after_ms })
+}
+
 /// The step that a run started with `--after RUN` waits in before its other work starts.
 pub const AFTER_RUN_STEP: &str = "after-run";
 
@@ -2484,6 +2543,7 @@ pub fn mission_after_run(mut mission: MissionSpec, run: &str) -> Result<MissionS
         produces_mission: None,
         uses_mission: None,
         after_run: Some(after_run_subject(run)),
+        program: None,
         gates: Vec::new(),
         nested_mission: None,
         loop_spec: None,