feat(session): expose scoped forks for MOI conversations

crates/astra-server-types/src/lib.rs

@@ -2202,12 +2202,14 @@ impl From<AdminUserRoleRecord> for AdminUserRoleResponse {
 #[cfg(feature = "server")]
 impl From<SessionRecord> for SessionResponse {
     fn from(value: SessionRecord) -> Self {
+        let mut metadata = value.metadata;
+        metadata.remove(astra_services::auth::session::PROVIDER_SCOPE_PROOF_KEY);
         Self {
             session_id: value.session_id,
             user_id: value.user_id,
             agent_id: value.agent_id,
             title: value.title,
-            metadata: value.metadata,
+            metadata,
             status: value.status,
             event_count: value.event_count,
             created_at: value.created_at,

crates/astra-turn-types/src/lib.rs

@@ -227,8 +227,9 @@ pub use session_cursor::{
 };
 pub use session_fork::{
     ForkBasisDimensionV1, ForkDimensionDispositionV1, ForkDimensionEvidenceV1,
-    ForkExcludedAuthorityV1, SESSION_FORK_MANIFEST_SCHEMA_VERSION, SessionForkActivationV1,
-    SessionForkManifestV1, SessionForkStateV1, SessionForkValidationError, SharedManifestPrefixV1,
+    ForkExcludedAuthorityV1, ProviderRunOwner, SESSION_FORK_MANIFEST_SCHEMA_VERSION,
+    SessionForkActivationV1, SessionForkManifestV1, SessionForkStateV1, SessionForkValidationError,
+    SharedManifestPrefixV1,
 };
 pub use session_handoff::{
     HandoffOperationWatermarksV1, HandoffRiskEvidenceV1, MANIFEST_DELTA_SCHEMA_VERSION,

crates/astra-turn-types/src/session_fork.rs

@@ -11,6 +11,13 @@ use crate::{
 
 pub const SESSION_FORK_MANIFEST_SCHEMA_VERSION: u32 = 1;
 
+/// Provider and workspace authenticated by the server, never a client claim.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct ProviderRunOwner {
+    pub provider_id: String,
+    pub provider_scope_id: String,
+}
+
 #[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Hash)]
 #[serde(rename_all = "snake_case")]
 pub enum ForkBasisDimensionV1 {
@@ -103,6 +110,10 @@ pub struct SessionForkManifestV1 {
     pub parent_key: SessionKeyV1,
     pub child_key: SessionKeyV1,
     pub parent_head: SessionContextHeadV1,
+    /// Immutable scope proof for this exact retained prefix and child identity.
+    /// Older manifests remain readable but cannot establish provider ownership.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub provider_owner: Option<ProviderRunOwner>,
     pub dimensions: Vec<ForkDimensionEvidenceV1>,
     pub excluded_authority: Vec<ForkExcludedAuthorityV1>,
     pub state: SessionForkStateV1,
@@ -129,6 +140,13 @@ impl SessionForkManifestV1 {
             return Err(SessionForkValidationError::UnsupportedSchema);
         }
         validate_identity(&self.fork_id)?;
+        if let Some(owner) = &self.provider_owner {
+            for identity in [&owner.provider_id, &owner.provider_scope_id] {
+                if identity.is_empty() || identity.trim() != identity {
+                    return Err(SessionForkValidationError::InvalidIdentity);
+                }
+            }
+        }
         self.shared_prefix().validate_for_child(&self.child_key)?;
         if self.parent_head.schema_version != SESSION_COORDINATION_SCHEMA_VERSION
             || self.parent_head.key != self.parent_key
@@ -308,6 +326,7 @@ mod tests {
             fork_id: "fork-a".into(),
             parent_key: parent_key.clone(),
             child_key: child_key.clone(),
+            provider_owner: None,
             parent_head: SessionContextHeadV1 {
                 schema_version: SESSION_COORDINATION_SCHEMA_VERSION,
                 key: parent_key,

crates/runtime/src/server/chat_handlers.rs

@@ -31,6 +31,7 @@ pub(super) async fn resolve_or_create_chat_session_id(
         agent_id,
         session_id_is_trusted,
         false,
+        None,
     )
     .await
     .map(|resolved| resolved.session_id)
@@ -49,6 +50,7 @@ pub(super) async fn resolve_or_create_chat_session(
     agent_id: Option<String>,
     session_id_is_trusted: bool,
     provider_authorized: bool,
+    provider_principal: Option<&AuthPrincipal>,
 ) -> Result<ResolvedChatSession, (StatusCode, Json<ErrorResponse>)> {
     match requested_session_id {
         Some(session_id) => {
@@ -103,6 +105,40 @@ pub(super) async fn resolve_or_create_chat_session(
                 )])
             });
 
+            if provider_authorized && user.user_id.starts_with("provider_authorized:") {
+                let principal = provider_principal.ok_or_else(|| {
+                    error_response_coded(
+                        StatusCode::FORBIDDEN,
+                        "verified provider identity is required",
+                        "provider_session_creation_forbidden",
+                    )
+                })?;
+                let identity = astra_services::ProviderSessionCreationIdentity::from_principal(
+                    principal,
+                    &format!("chat-{}", uuid::Uuid::new_v4()),
+                )?;
+                let result = state
+                    .session_service
+                    .create_provider_session(
+                        identity,
+                        SessionCreateRequestData {
+                            agent_id,
+                            title: None,
+                            metadata,
+                        },
+                        astra_services::resource_governor::LimitCheck::Allowed,
+                    )
+                    .await?;
+                return Ok(ResolvedChatSession {
+                    session_id: Some(result.session.session_id),
+                    full_llm_capture:
+                        crate::turn::llm::exchange_capture::session_full_llm_capture_enabled(Some(
+                            &result.session.metadata,
+                        )),
+                    session_admission_facts: Some(result.admission_facts),
+                });
+            }
+
             match super::session::session_quota::create_session_with_resource_quota(
                 state,
                 user.user_id.clone(),
@@ -206,6 +242,7 @@ pub(super) async fn chat_handler(
             &principal.origin,
             astra_services::AuthPrincipalOrigin::ProviderAuthorizedRequest(_)
         ),
+        Some(&principal),
     )
     .await?;
     chat_data.session_id = resolved.session_id;
@@ -284,6 +321,7 @@ pub(super) async fn chat_stream_handler(
                 &principal.origin,
                 astra_services::AuthPrincipalOrigin::ProviderAuthorizedRequest(_)
             ),
+            Some(&principal),
         ),
         inject_effective_runtime_context(&state, &principal, &mut chat_data),
     );
@@ -594,7 +632,8 @@ mod session_resolution_tests {
             }));
         state.resource_governor = governor.clone();
         let result =
-            resolve_or_create_chat_session(&state, &test_user(), None, None, false, false).await;
+            resolve_or_create_chat_session(&state, &test_user(), None, None, false, false, None)
+                .await;
         assert!(result.is_err());
         assert_eq!(result.err().unwrap().0, StatusCode::SERVICE_UNAVAILABLE);
         assert_eq!(governor.get_usage("u1").await.sessions_created, 0);
@@ -613,6 +652,7 @@ mod session_resolution_tests {
             Some("agent-1".to_string()),
             false,
             false,
+            None,
         )
         .await
         .expect("session resolution should succeed");
@@ -674,6 +714,7 @@ mod session_resolution_tests {
             None,
             false,
             false,
+            None,
         )
         .await
         .expect("custom session service should resolve an existing session");
@@ -698,6 +739,7 @@ mod session_resolution_tests {
             None,
             false,
             true,
+            None,
         )
         .await
         {

crates/runtime/src/server/edge/edge_callback_handlers.rs

@@ -1504,6 +1504,7 @@ pub(crate) async fn post_provider_interaction_respond_handler(
         ));
     }
 
+    super::super::run::handlers::authorize_provider_run_session(&state, &principal, run_id).await?;
     let target = state
         .execution
         .run_lifecycle_service
@@ -2259,6 +2260,27 @@ mod edge_callback_insert_tests {
 
     #[async_trait]
     impl RunLifecycleService for ApprovalTargetRunLifecycle {
+        async fn authorize_provider_run_scope(
+            &self,
+            user_id: String,
+            run_id: String,
+            owner: astra_services::runs::ProviderRunOwner,
+        ) -> Result<(), (StatusCode, Json<crate::ErrorResponse>)> {
+            if user_id == "u-approval"
+                && run_id == self.run_id
+                && owner.provider_id == "moi"
+                && owner.provider_scope_id == "workspace-a"
+            {
+                Ok(())
+            } else {
+                Err(astra_core::error_response_coded(
+                    StatusCode::FORBIDDEN,
+                    "Session is outside the authorized scope",
+                    "session_fork_forbidden",
+                ))
+            }
+        }
+
         async fn create_run(
             &self,
             _user_id: String,

crates/runtime/src/server/provider_test_support.rs

@@ -14,6 +14,53 @@ use axum::{Json, Router};
 use serde_json::{Value, json};
 use tokio::sync::Notify;
 
+/// Exercise provider/binding execution without a persistent Session. This
+/// test-only boundary uses production request conversion, runtime injection,
+/// lifecycle execution and SSE projection; it does not test inbound HTTP
+/// authentication or Session creation/scope admission.
+pub async fn ephemeral_provider_chat_stream(
+    state: &super::AppState,
+    principal: &astra_services::AuthPrincipal,
+    payload: Value,
+) -> Response {
+    let request: super::ChatRequest = serde_json::from_value(payload).expect("test chat request");
+    assert!(
+        request.session_id.is_none(),
+        "ephemeral fixture cannot address a persistent Session"
+    );
+    let mut request = super::chat_request_into_data(request);
+    let result = async {
+        super::provider_runtime_context::inject_effective_runtime_context(
+            state,
+            principal,
+            &mut request,
+        )
+        .await?;
+        state
+            .execution
+            .run_lifecycle_service
+            .stream_chat(principal.user.user_id.clone(), request)
+            .await
+    }
+    .await;
+    match result {
+        Ok(mut stream) => super::sse_streaming_response(
+            stream.session_id,
+            stream.run_id,
+            None,
+            stream
+                .event_rx
+                .take()
+                .expect("a new ephemeral run streams its events"),
+        ),
+        Err((status, error)) => super::sse_error_response_from_error_with_context(
+            status,
+            error.0,
+            super::SseErrorContext::default(),
+        ),
+    }
+}
+
 #[derive(Clone, Debug)]
 pub struct ProviderRequest {
     pub method: Method,

crates/runtime/src/server/router_builder/sessions.rs

@@ -105,6 +105,14 @@ pub(super) fn add_routes(router: Router<AppState>, state: AppState) -> Router<Ap
             "/sessions/{session_id}/handoffs/{handoff_id}/activate",
             post(session::session_handlers::activate_session_handoff_handler),
         )
+        .route(
+            "/sessions/{session_id}/runs/{run_id}/cursor",
+            get(session::session_handlers::get_session_run_cursor_handler),
+        )
+        .route(
+            "/sessions/{session_id}/forks/{fork_id}/activate-for-continuation",
+            post(session::session_handlers::activate_session_fork_for_continuation_handler),
+        )
         .route(
             "/sessions/{session_id}/forks",
             get(session::session_handlers::list_session_forks_handler)

crates/runtime/src/server/run/handlers.rs

@@ -424,6 +424,25 @@ pub(crate) async fn get_run_projection_handler(
     Ok(Json(RunProjectionResponse::new(projection)))
 }
 
+pub(crate) async fn authorize_provider_run_session(
+    state: &AppState,
+    principal: &AuthPrincipal,
+    run_id: &str,
+) -> Result<(), (StatusCode, Json<ErrorResponse>)> {
+    let AuthPrincipalOrigin::ProviderAuthorizedRequest(context) = &principal.origin else {
+        return Ok(());
+    };
+    let owner = astra_services::runs::ProviderRunOwner {
+        provider_id: context.provider_id.clone(),
+        provider_scope_id: context.provider_scope_id.clone(),
+    };
+    state
+        .execution
+        .run_lifecycle_service
+        .authorize_provider_run_scope(principal.user.user_id.clone(), run_id.to_owned(), owner)
+        .await
+}
+
 pub(crate) async fn stream_run_handler(
     State(state): State<AppState>,
     trace: Option<axum::Extension<crate::server::RequestTrace>>,
@@ -464,6 +483,19 @@ pub(crate) async fn stream_run_handler(
         }
     };
 
+    if let Err((status, error)) = authorize_provider_run_session(&state, &principal, &run_id).await
+    {
+        return sse_error_response_from_error_with_context(
+            status,
+            error.0,
+            SseErrorContext {
+                request_id: request_id.as_deref(),
+                run_id: Some(&run_id),
+                ..SseErrorContext::default()
+            },
+        );
+    }
+
     if query.replay_only {
         return match state
             .execution
@@ -548,6 +580,7 @@ pub(crate) async fn delete_run_handler(
             external_request_descriptor(&method, &uri, &headers, "/chat/runs/{run_id}", &body),
         )
         .await?;
+    authorize_provider_run_session(&state, &principal, &run_id).await?;
     cancel_run_for_user(&state, run_id, principal.user.user_id).await
 }