Mission run authority: decide every write by a run, audited by default

crates/fractal-daemon/src/service.rs

@@ -1739,6 +1739,7 @@ mod tests {
         limits: crate::config::LimitsConfig::default(),
         logging: crate::config::LoggingConfig::default(),
         reconcile: crate::config::ReconcileConfig::default(),
+        programs: crate::config::ProgramsConfig::default(),
         asks: crate::config::AsksConfig::default(),
         codex_work_without_asking: None,
         wait_for_peers_after_start: None,

crates/fractal-delivery/src/store/delivery.rs

@@ -299,7 +299,7 @@ impl DeliveryStore for Store {
         let child = parent.map(|parent| ChildMissionContext {
             root_revision: parent.root_revision.clone(),
             root_run_id: parent.root_mission_run.trim_start_matches("mission-run/").into(),
-            parent_step_run: normalize_step_run(subscription), default_selector: None,
+            parent_step_run: normalize_step_run(subscription), default_selector: None, started_by: None,
         });
         self.create_mission_run_inner(request, child, None, Some((resource, discovery)))
     }

crates/fractal-kernel/src/config.rs

@@ -151,6 +151,10 @@ pub struct Config {
     /// Background reconciliation start cap, reloaded with daemon policies.
     #[serde(skip_serializing_if = "ReconcileConfig::is_default")]
     pub reconcile: ReconcileConfig,
+    /// What a write outside a mission run's authority does. Written only when it differs from
+    /// the default.
+    #[serde(skip_serializing_if = "ProgramsConfig::is_default")]
+    pub programs: ProgramsConfig,
     /// How long a person has to answer an ask that names no deadline. Written only when it
     /// differs from the default.
     #[serde(skip_serializing_if = "AsksConfig::is_default")]
@@ -460,6 +464,27 @@ impl LimitsConfig {
     }
 }
 
+/// `[programs]`: how the daemon holds a program's run to its authority
+/// (docs/st3/program-engine.md §3).
+#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(default, deny_unknown_fields)]
+pub struct ProgramsConfig {
+    /// `audit` lets a write outside a run's authority proceed and records it; `enforce` refuses it.
+    pub authority: RunAuthorityMode,
+}
+
+impl ProgramsConfig {
+    pub fn is_default(&self) -> bool { self == &Self::default() }
+}
+
+#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "kebab-case")]
+pub enum RunAuthorityMode {
+    #[default]
+    Audit,
+    Enforce,
+}
+
 /// The reconcile pass cap when `[reconcile] max_passes_per_minute` is not set.
 pub const DEFAULT_MAX_PASSES_PER_MINUTE: u32 = 30;
 
@@ -789,6 +814,7 @@ impl Default for Config {
             limits: LimitsConfig::default(),
             logging: LoggingConfig::default(),
             reconcile: ReconcileConfig::default(),
+            programs: ProgramsConfig::default(),
             asks: AsksConfig::default(),
             codex_work_without_asking: None,
             wait_for_peers_after_start: None,

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

@@ -236,6 +236,25 @@ mod identifier_query_tests {
     }
 }
 
+fn child_context(
+    parent: &MissionRunView,
+    parent_step_run: &str,
+    default_selector: Option<WorkSelector>,
+    started_by: Option<String>,
+) -> ChildMissionContext {
+    ChildMissionContext {
+        root_revision: parent.root_revision.clone(),
+        root_run_id: parent
+            .root_mission_run
+            .strip_prefix("mission-run/")
+            .unwrap_or(&parent.root_mission_run)
+            .to_owned(),
+        parent_step_run: normalize_step_run(parent_step_run),
+        default_selector,
+        started_by,
+    }
+}
+
 /// Render a copy; the signed revision and execution spec remain physical.
 fn public_mission_spec(connection: &Connection, body: &str) -> Result<MissionSpec> {
     let mut mission: MissionSpec = serde_json::from_str(body)?;
@@ -1610,6 +1629,15 @@ pub trait MissionsStore {
             parent_step_run: &str,
             default_selector: Option<&WorkSelector>,
         ) -> Result<MissionRunView, FractalError>;
+    /// A child run a program step starts: the parent run starts it, so its creation is written
+    /// as `mission-run/PARENT` and it holds at most its parent's authority
+    /// (docs/st3/program-engine.md §3). The requester stays the parent's.
+    fn create_program_child_mission_run(
+            &self,
+            request: &MissionRunRequest,
+            parent: &MissionRunView,
+            parent_step_run: &str,
+        ) -> Result<MissionRunView, FractalError>;
     /// An occurrence belongs to its schedule, regardless of the requesting member or revision.
     fn scheduled_mission_run_subject(schedule: &str, occurrence: u64) -> String;
     fn create_scheduled_mission_run(
@@ -2939,16 +2967,21 @@ impl MissionsStore for Store {
         ) -> Result<MissionRunView, FractalError> {
         self.create_mission_run_inner(
             request,
-            Some(ChildMissionContext {
-                root_revision: parent.root_revision.clone(),
-                root_run_id: parent
-                    .root_mission_run
-                    .strip_prefix("mission-run/")
-                    .unwrap_or(&parent.root_mission_run)
-                    .to_owned(),
-                parent_step_run: normalize_step_run(parent_step_run),
-                default_selector: default_selector.cloned(),
-            }),
+            Some(child_context(parent, parent_step_run, default_selector.cloned(), None)),
+            None,
+            None,
+        )
+    }
+
+    fn create_program_child_mission_run(
+            &self,
+            request: &MissionRunRequest,
+            parent: &MissionRunView,
+            parent_step_run: &str,
+        ) -> Result<MissionRunView, FractalError> {
+        self.create_mission_run_inner(
+            request,
+            Some(child_context(parent, parent_step_run, None, Some(parent.subject.clone()))),
             None,
             None,
         )
@@ -2984,6 +3017,7 @@ impl MissionsStore for Store {
                 .into(),
             parent_step_run: normalize_step_run(schedule),
             default_selector: None,
+            started_by: None,
         });
         self.create_mission_run_inner(request, child, Some(&subject), None)
     }
@@ -4674,7 +4708,7 @@ impl MissionsStore for Store {
                     &[],
                     None,
                 )
-                .map_err(internal)?;
+                .map_err(smallclaims::error::typed)?;
             }
         }
         revision_seats::carry_completed_agents_tx(
@@ -4717,7 +4751,7 @@ impl MissionsStore for Store {
             None,
             idempotency_key,
         )
-        .map_err(internal)?;
+        .map_err(smallclaims::error::typed)?;
         terminalize_generation_steps_tx(
             &transaction,
             &self.origin,
@@ -4744,7 +4778,7 @@ impl MissionsStore for Store {
             &[],
             None,
         )
-        .map_err(internal)?;
+        .map_err(smallclaims::error::typed)?;
         if reopening {
             // The run's own state history says who reopened it, when, and why.
             append_claim_tx(
@@ -4762,7 +4796,7 @@ impl MissionsStore for Store {
                 &[],
                 None,
             )
-            .map_err(internal)?;
+            .map_err(smallclaims::error::typed)?;
         }
         if let Some(proposal) = proposal {
             let changed = transaction
@@ -4793,7 +4827,7 @@ impl MissionsStore for Store {
                 &[],
                 None,
             )
-            .map_err(internal)?;
+            .map_err(smallclaims::error::typed)?;
         }
         let view = mission_run_view_tx(&transaction, run_id).map_err(internal)?;
         transaction

crates/fractal-store/src/store.rs

@@ -73,6 +73,7 @@ pub mod seat_queue_plan_tests;
 pub mod schedule_standby;
 pub mod mission_capacity;
 pub mod run_credentials;
+pub mod run_authority;
 pub mod marks_switch;
 pub mod glasses;
 pub mod mailbox_wakes;
@@ -354,6 +355,8 @@ mod run_eta_tests;
 #[cfg(test)]
 mod run_credentials_tests;
 #[cfg(test)]
+mod run_authority_tests;
+#[cfg(test)]
 mod run_eta_review_tests;
 pub mod client_summary;
 pub mod unread_mail;
@@ -2435,6 +2438,10 @@ pub struct ChildMissionContext {
     pub root_run_id: String,
     pub parent_step_run: String,
     pub default_selector: Option<WorkSelector>,
+    /// Who starts the child when it is not its requester: `mission-run/PARENT` for a child a
+    /// program starts. Its creation claims are written as this actor, and run authority takes it
+    /// as the child's starter (docs/st3/program-engine.md §3).
+    pub started_by: Option<String>,
 }
 
 fn stale_ref_request_tx(connection: &Connection, resource: &str, discovery: &str) -> Result<Option<String>> {
@@ -7952,7 +7959,7 @@ fn adopt_declared_mission_revision_tx(
             &[],
             Some(batch_id),
         )
-        .map_err(internal)?;
+        .map_err(smallclaims::error::typed)?;
         if status == "draining" {
             transaction
                 .execute(
@@ -8047,7 +8054,7 @@ fn adopt_declared_mission_revision_tx(
                 &[],
                 Some(batch_id),
             )
-            .map_err(internal)?;
+            .map_err(smallclaims::error::typed)?;
             claim_ids.push(claim.id);
         }
     }
@@ -8094,7 +8101,7 @@ fn adopt_declared_mission_revision_tx(
         &[],
         Some(batch_id),
     )
-    .map_err(internal)?;
+    .map_err(smallclaims::error::typed)?;
     claim_ids.push(superseded.id);
     claim_ids.extend(terminalize_generation_steps_tx(
         transaction,
@@ -8122,7 +8129,7 @@ fn adopt_declared_mission_revision_tx(
         &[],
         Some(batch_id),
     )
-    .map_err(internal)?;
+    .map_err(smallclaims::error::typed)?;
     claim_ids.push(created.id);
     Ok((claim_ids, false))
 }
@@ -11605,6 +11612,8 @@ fn replicated_work_validation_fields<'a>(
         owned_sets::validate_receipt(subject, body).map_err(anyhow::Error::new)?;
     }
     let fields = schema_fields_for_body(kind, body)?;
+    run_authority::admit_tx(transaction, origin, actor, kind, subject, &fields)
+        .map_err(anyhow::Error::new)?;
     human_review_reassignment::admit_reassignment(transaction, kind, &fields)
         .map_err(anyhow::Error::new)?;
     if kind == "person.delegation-set"
@@ -21760,7 +21769,7 @@ pub fn terminalize_generation_steps_tx(
             &[],
             batch_id,
         )
-        .map_err(internal)?;
+        .map_err(smallclaims::error::typed)?;
         claim_ids.push(claim.id);
     }
     Ok(claim_ids)
@@ -22230,7 +22239,7 @@ pub fn cancel_descendant_mission_runs_tx(
             &[],
             None,
         )
-        .map_err(internal)?;
+        .map_err(smallclaims::error::typed)?;
         claim_ids.push(run_claim.id);
         let generation_claim = append_claim_tx(
             transaction,
@@ -22242,7 +22251,7 @@ pub fn cancel_descendant_mission_runs_tx(
             &[],
             None,
         )
-        .map_err(internal)?;
+        .map_err(smallclaims::error::typed)?;
         claim_ids.push(generation_claim.id);
     }
     Ok(claim_ids)

crates/fractal-store/src/store/message_routing.rs

@@ -22,6 +22,12 @@ pub fn legacy_reason(store: &Store, input: &ClaimInput) -> Result<Option<&'stati
     if input.kind != "message.sent" {
         return Ok(Some("not-a-message"));
     }
+    // A mission run's send stays on the graph append, where its run authority decides it
+    // (docs/st3/program-engine.md §3); the stream append has no such decision.
+    let sender = input.fields.get("from").and_then(Value::as_str);
+    if [input.actor.as_deref(), sender].into_iter().flatten().any(|party| party.starts_with("mission-run/")) {
+        return Ok(Some("run-sender"));
+    }
     if messages_shard::shard(store).is_none() {
         return Ok(Some("no-shard"));
     }

crates/fractal-store/src/store/message_routing_tests.rs

@@ -125,6 +125,11 @@ fn a_message_goes_to_the_stream_only_where_its_recipients_host_reads_it() {
     personal.actor = Some("person/ada".into());
     personal.fields.insert("from".into(), json!("person/ada"));
     assert_eq!(reason(&personal), Some("ineligible"));
+    // A mission run's send stays on the graph append, where its run authority decides it.
+    let mut from_run = message("message/route-run", &here);
+    from_run.actor = Some("mission-run/0123abcd".into());
+    from_run.fields.insert("from".into(), json!("mission-run/0123abcd"));
+    assert_eq!(reason(&from_run), Some("run-sender"));
     let mut attached = message("message/route-attached", &here);
     attached.fields.insert("attachments".into(), json!([{"name": "body.txt"}]));
     assert_eq!(reason(&attached), Some("attached"));

crates/fractal-store/src/store/mission_start_production_copy_tests.rs

@@ -122,6 +122,7 @@ fn scheduled_mission_start_writer_hold_on_production_copy() {
                 root_run_id: root_run_id.clone(),
                 parent_step_run: parent_step_run.clone(),
                 default_selector: None,
+                started_by: None,
             });
             let result = store.create_mission_run_inner(&request, context, Some(&subject), None);
             let mut receipt = lend_receipt(started.elapsed());

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

@@ -396,6 +396,9 @@ impl Store {
             request.requester.as_deref().unwrap_or("person/requester"),
             "person",
         );
+        // The starter writes the run's creation: its requester, or the run that started it.
+        let starter = child.as_ref().and_then(|child| child.started_by.clone())
+            .unwrap_or_else(|| requester.clone());
         let mode = request.mode.as_deref().unwrap_or("run");
         if !matches!(mode, "run" | "eval") {
             return Err(FractalError::new(
@@ -514,19 +517,21 @@ impl Store {
             &self.origin,
             &subject,
             "mission-run.created",
-            Some(&requester),
+            Some(&starter),
             &body,
             &[],
             None,
             &request.idempotency_key,
         )
-        .map_err(internal)?;
+        // A typed refusal, such as a starter without the authority to start this mission, keeps
+        // its code.
+        .map_err(smallclaims::error::typed)?;
         append_claim_tx(
             &transaction,
             &self.origin,
             &generation_subject,
             "run-generation.created",
-            Some(&requester),
+            Some(&starter),
             &json!({"fields": {
                 "run": subject,
                 "revision": mission.revision,
@@ -536,7 +541,7 @@ impl Store {
             &[],
             None,
         )
-        .map_err(internal)?;
+        .map_err(smallclaims::error::typed)?;
         stages.enter("view");
         let view = created_mission_run_view_tx(&transaction, &run_id).map_err(internal)?;
         stages.enter("idempotency-record");