Program engine: attempts act as their run (credential in effects, journal as the run)

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

@@ -7478,6 +7478,9 @@ async fn serve_reserved_unix_with_lookup_ready(
                     if let Some(agent) = ancestry.agent() {
                         request.extensions_mut().insert(BoundAgent(agent.to_owned()));
                     }
+                    if let HarnessAncestry::Run(credential) = &ancestry {
+                        request.extensions_mut().insert(credential.clone());
+                    }
                     if matches!(ancestry, HarnessAncestry::Unavailable) {
                         request.extensions_mut().insert(HarnessAncestry::Unavailable);
                     }
@@ -7521,11 +7524,41 @@ mod gateway_listener_tests;
 #[derive(Clone, Debug, Default, PartialEq, Eq)]
 pub enum HarnessAncestry {
     Bound(String),
+    /// A program effect: an ancestor carries the run credential its attempt was started with
+    /// (docs/st3/program-engine.md §3). The guard resolves it to `mission-run/RUN`.
+    Run(RunCredential),
     Person,
     #[default]
     Unavailable,
 }
 
+/// A run credential read from a caller's exec-time environment. Its `Debug` never shows the
+/// secret, so ancestry diagnostics cannot leak it.
+#[derive(Clone, PartialEq, Eq)]
+pub struct RunCredential(String);
+
+impl RunCredential {
+    pub fn new(secret: impl Into<String>) -> Self {
+        Self(secret.into())
+    }
+
+    pub fn secret(&self) -> &str {
+        &self.0
+    }
+}
+
+impl std::fmt::Debug for RunCredential {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        formatter.write_str("RunCredential(..)")
+    }
+}
+
+/// The run a request's credential binds, set by the serving guard and never by a client.
+#[derive(Clone, Debug)]
+pub struct BoundRun(pub crate::store::run_credentials::RunBinding);
+
+pub use crate::store::run_credentials::RUN_CREDENTIAL_ENV;
+
 impl HarnessAncestry {
     fn agent(&self) -> Option<&str> {
         match self { Self::Bound(agent) => Some(agent), _ => None }
@@ -7548,6 +7581,15 @@ impl HarnessAncestry {
     fn of_environment<'a>(lookup: &impl Fn(&str) -> Option<&'a str>) -> Self {
         let primary = lookup("ST_AGENT");
         let legacy = lookup("FRACTAL_AGENT");
+        // A program effect acts as its run, never as an agent too: one environment naming both
+        // cannot be bound to either.
+        if let Some(credential) = lookup(RUN_CREDENTIAL_ENV) {
+            return if credential.is_empty() || primary.is_some() || legacy.is_some() {
+                Self::Unavailable
+            } else {
+                Self::Run(RunCredential::new(credential))
+            };
+        }
         if primary.is_none() && legacy.is_none() {
             return Self::Person;
         }
@@ -7997,6 +8039,38 @@ pub fn authorize_worker(peer: Option<&LocalPeer>, observe: &dyn Fn() -> anyhow::
 #[derive(Clone, Debug)]
 pub struct BoundAgent(pub String);
 
+/// The identity a local request is bound to by its caller's ancestry: an agent harness, or the
+/// mission run a program effect acts as. Handlers that hold a bound caller to its own identity
+/// read this one extractor, so a bound run is never mistaken for an unbound person.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub enum BoundActor {
+    Agent(String),
+    Run(String),
+}
+
+impl BoundActor {
+    pub fn of(extensions: &axum::http::Extensions) -> Option<Self> {
+        extensions.get::<BoundAgent>().map(|agent| Self::Agent(agent.0.clone()))
+            .or_else(|| extensions.get::<BoundRun>().map(|run| Self::Run(run.0.run.clone())))
+    }
+
+    /// The bound identity: the agent as the harness names it, or `mission-run/RUN`.
+    pub fn actor(&self) -> &str {
+        match self { Self::Agent(actor) | Self::Run(actor) => actor }
+    }
+}
+
+impl<S: Send + Sync> axum::extract::OptionalFromRequestParts<S> for BoundActor {
+    type Rejection = std::convert::Infallible;
+
+    async fn from_request_parts(
+        parts: &mut axum::http::request::Parts,
+        _state: &S,
+    ) -> Result<Option<Self>, Self::Rejection> {
+        Ok(Self::of(&parts.extensions))
+    }
+}
+
 // The serving boundary supplies the immutable ancestry binding and its identity store before
 // any route or admission effect runs, including for directly constructed routers.
 async fn guard_bound_actor(
@@ -8007,6 +8081,16 @@ async fn guard_bound_actor(
     request.extensions_mut().remove::<ancestry_diagnostics::VerifiedWorkerFallback>();
     let ancestry_diagnostic = ancestry_diagnostics::RequestDiagnostic::capture(&request);
     let bound_agent = request.extensions().get::<BoundAgent>().map(|agent| agent.0.clone());
+    // A run credential binds only while its attempt runs; one that no longer binds leaves the
+    // caller unverifiable, so it writes nothing.
+    request.extensions_mut().remove::<BoundRun>();
+    if let Some(credential) = request.extensions().get::<RunCredential>().cloned() {
+        let lookup = store.clone();
+        match tokio::task::spawn_blocking(move || lookup.run_binding(credential.secret())).await {
+            Ok(Ok(binding)) => { request.extensions_mut().insert(BoundRun(binding)); }
+            _ => { request.extensions_mut().insert(HarnessAncestry::Unavailable); }
+        }
+    }
     let client_request = request.uri().path().starts_with("/v1/client/");
     // This guard is outside all router layers. Authenticate here, before ancestry admission,
     // independently of the ingress observe/enforce rollout.
@@ -8064,7 +8148,9 @@ pub async fn guard_bound_request(
     bound_agent: Option<&str>,
     store: &Store,
 ) -> Result<Request<Body>, ApiError> {
-    let Some(bound_agent) = bound_agent else {
+    // A bound run follows the rules of a bound agent, as the actor `mission-run/RUN`.
+    let bound_run = request.extensions().get::<BoundRun>().map(|run| run.0.run.clone());
+    let Some(bound_agent) = bound_agent.or(bound_run.as_deref()) else {
         return if matches!(request.extensions().get::<HarnessAncestry>(), Some(HarnessAncestry::Unavailable)) {
             guard_unverifiable_request(request).await
         } else {
@@ -8089,9 +8175,11 @@ pub async fn guard_bound_request(
         && !client_v0::administrative_get(&request) {
         return Ok(request);
     }
-    let bound_identity = public_identity(
-        store, crate::store::identities::IdentityKind::Agent, bound_agent,
-    )?.identity;
+    let bound_identity = if bound_run.as_deref() == Some(bound_agent) {
+        bound_agent.to_owned()
+    } else {
+        public_identity(store, crate::store::identities::IdentityKind::Agent, bound_agent)?.identity
+    };
     // Free mode lets an agent do what its person may do, but always as itself: a harness never
     // names a person, or another agent, as the local client-v0 actor.
     for person in [client_v0::LOCAL_PERSON_HEADER, wire_names::LOCAL_PERSON_HEADER_ALIAS]

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

@@ -556,24 +556,40 @@ impl ClientSession {
         };
         // Free mode: an agent's local session holds every scope a person's does. Its actions
         // still record the agent as their actor.
-        Ok(Self {
-            actor: person.into(),
-            authority_actor: person.into(),
+        Ok(Self::acting(person))
+    }
+
+    /// A program effect's local session, acting as `run` (docs/st3/program-engine.md §3). Only a
+    /// request its run credential binds to that same run may name it; run authority then decides
+    /// each write.
+    pub fn local_run(run: &str, bound_run: Option<&str>) -> Result<Self, ApiError> {
+        if bound_run != Some(run) {
+            return Err(forbidden(
+                "a trusted Unix client acts as a mission run only when its run credential binds that run",
+            ));
+        }
+        Ok(Self::acting(run))
+    }
+
+    fn acting(actor: &str) -> Self {
+        Self {
+            actor: actor.into(),
+            authority_actor: actor.into(),
             pairing_grant: None,
             pairing_issued_here: false,
             grant_expires_at: None,
             assistant_write: None,
             transport: "unix",
             local_peer: false,
             peer_process: None,
-            custom_forms,
+            custom_forms: false,
             conversation_blocks: false,
             conversation_delivery_links: false,
             conversation_message_participants: false,
             assistant_content_handles: false,
             messages_paging: false,
             scopes: ALL_SCOPES.iter().map(|scope| (*scope).to_owned()).collect(),
-        })
+        }
     }
 
     pub fn pairing() -> Self {
@@ -775,10 +791,15 @@ pub fn authenticate(
                 .headers()
                 .get(LOCAL_PERSON_HEADER)
                 .and_then(|value| value.to_str().ok());
-            let mut session = ClientSession::local(person)?;
+            let bound_run = request.extensions().get::<super::BoundRun>().map(|bound| bound.0.run.as_str());
+            let mut session = match person.filter(|person| person.starts_with("mission-run/")) {
+                Some(run) => ClientSession::local_run(run, bound_run)?,
+                None => ClientSession::local(person)?,
+            };
             session.peer_process = request.extensions().get::<super::LocalPeer>()
                 .and_then(|peer| peer.peer_process.clone());
             session.local_peer = request.extensions().get::<super::BoundAgent>().is_some()
+                || bound_run.is_some()
                 || request.extensions().get::<super::LocalPeer>()
                     .is_some_and(|peer| peer.ancestor.is_some())
                 || request.extensions().get::<super::NativeDeliveryPeer>().is_some();

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

@@ -1439,10 +1439,10 @@ fn ignored_authority_warnings(
 
 pub async fn apply_with_bound(
     State(state): State<AppState>,
-    bound: Option<Extension<BoundAgent>>,
+    bound: Option<BoundActor>,
     Json(request): Json<ApplyRequest>,
 ) -> Result<Json<ApplyResponse>, ApiError> {
-    apply_with_authority(state, request, bound.as_ref().map(|bound| bound.0.0.as_str())).await
+    apply_with_authority(state, request, bound.as_ref().map(BoundActor::actor)).await
 }
 
 pub async fn get_mission(

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

@@ -233,7 +233,7 @@ pub async fn operational_repair_plan(
 
 pub async fn apply_operational_repair(
     State(state): State<AppState>,
-    bound: Option<Extension<BoundAgent>>,
+    bound: Option<BoundActor>,
     Json(request): Json<OperationalRepairApplyRequest>,
 ) -> Result<Json<OperationalRepairResult>, ApiError> {
     if bound.is_some() || request.actor.is_some() {
@@ -249,14 +249,14 @@ pub async fn apply_operational_repair(
 /// before mutation, even if it would silently ignore an unknown actor field.
 pub async fn apply_attributed_operational_repair(
     State(state): State<AppState>,
-    bound: Option<Extension<BoundAgent>>,
+    bound: Option<BoundActor>,
     Json(request): Json<OperationalRepairApplyRequest>,
 ) -> Result<Json<OperationalRepairResult>, ApiError> {
     let bound = bound.ok_or_else(|| ApiError::bad(FractalError::new(
         "foreign-agent-actor",
         "attributed repair apply requires an authenticated harness binding",
     )))?;
-    apply_operational_repair_with_actor(state, request, Some(bound.0.0)).await
+    apply_operational_repair_with_actor(state, request, Some(bound.actor().to_owned())).await
 }
 
 async fn apply_operational_repair_with_actor(

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

@@ -45,17 +45,19 @@ pub(super) fn generic_claims_page(
 
 pub async fn put_document(
     State(state): State<AppState>,
-    bound: Option<axum::Extension<BoundAgent>>,
+    bound: Option<BoundActor>,
     Json(request): Json<DocumentPutRequest>,
 ) -> Result<Json<DocumentVersion>, ApiError> {
+    // A seat writes only its own seat documents; a run, which has none, writes none.
     if let Some(bound) = bound.as_ref()
-        && !crate::handoff::may_write(&bound.0.0, &request.name)
+        && !crate::handoff::may_write(bound.actor(), &request.name)
     {
         return Err(ApiError::bad(FractalError::new(
             "document-not-yours",
             format!(
-                "`{}` belongs to another seat; a seat writes only doc/agents/… for itself",
-                request.name
+                "`{}` belongs to a seat; a seat writes only doc/agents/… for itself, and `{}` is not that seat",
+                request.name,
+                bound.actor(),
             ),
         )));
     }
@@ -66,7 +68,7 @@ pub async fn put_document(
             &request.bytes,
             &request.expected_document,
             &request.idempotency_key,
-            bound.as_ref().map(|bound| bound.0.0.as_str()),
+            bound.as_ref().map(BoundActor::actor),
         )
         .map_err(ApiError::bad)?;
     signal_changed(&state);