fix: keep mandatory Subject callbacks independent of waiting Host parents

crates/workspace-server/src/server.rs

@@ -20372,11 +20372,26 @@ fn require_dedicated_subjektiv_consolidation_worker(
     }
 }
 
-fn subjektiv_subject_scope(
+// Store-only authority discovery. A mandatory Session attribution callback must
+// reject consolidation roles without consulting Runtime: its create/restore parent
+// can already hold a Host worker while Runtime waits for that callback to finish.
+enum SubjektivStoredScopeAuthority {
+    Subject,
+    JobConsolidation,
+    LegacyConsolidation { singleton_key: String },
+}
+
+struct SubjektivStoredWorkerScope {
+    subject_id: String,
+    worker: RuntimeWorkerRef,
+    authority: SubjektivStoredScopeAuthority,
+}
+
+fn subjektiv_worker_scope_from_store(
     api: &WorkspaceApi,
     workspace_id: &str,
     context: &server_api::ServerRequestContext,
-) -> ApiResult<(String, RuntimeWorkerRef, SubjektivWorkerAuthority)> {
+) -> ApiResult<SubjektivStoredWorkerScope> {
     let source = context.runtime_source.as_ref().ok_or_else(|| {
         Error::WorkspacePermissionDenied(
             "subjektiv operations require authenticated Runtime-owned source authority".to_string(),
@@ -20389,11 +20404,11 @@ fn subjektiv_subject_scope(
     })?;
     let worker = RuntimeWorkerRef::new(&source.runtime_id, worker_id);
     if let Some(grant) = subjektiv_job::active_grant(api, workspace_id, &worker)? {
-        return Ok((
+        return Ok(SubjektivStoredWorkerScope {
-            grant.subject_id,
+            subject_id: grant.subject_id,
             worker,
-            SubjektivWorkerAuthority::Consolidation,
+            authority: SubjektivStoredScopeAuthority::JobConsolidation,
-        ));
+        });
     }
     let lease = api
         .store
@@ -20404,15 +20419,47 @@ fn subjektiv_subject_scope(
             )
         })?;
     let (subject_id, authority) = subjektiv_scope_from_singleton_key(&lease.key)?;
-    if authority == SubjektivWorkerAuthority::Consolidation {
+    let authority = match authority {
-        let mut summary = api
+        SubjektivWorkerAuthority::Subject => SubjektivStoredScopeAuthority::Subject,
-            .runtime
+        SubjektivWorkerAuthority::Consolidation => {
-            .worker(&worker)
+            SubjektivStoredScopeAuthority::LegacyConsolidation {
-            .map_err(|error| error.into_error())?;
+                singleton_key: lease.key.clone(),
-        summary.singleton_key = Some(lease.key.clone());
+            }
-        require_dedicated_subjektiv_consolidation_worker(&summary, workspace_id, &lease.key)?;
+        }
-    }
+    };
-    Ok((subject_id.to_string(), worker, authority))
+    Ok(SubjektivStoredWorkerScope {
+        subject_id: subject_id.to_owned(),
+        worker,
+        authority,
+    })
+}
+
+// Mixed callers run this complete policy boundary on Host. Only legacy
+// consolidation authority needs synchronous Runtime profile/tag verification.
+fn subjektiv_subject_scope(
+    api: &WorkspaceApi,
+    workspace_id: &str,
+    context: &server_api::ServerRequestContext,
+) -> ApiResult<(String, RuntimeWorkerRef, SubjektivWorkerAuthority)> {
+    let scope = subjektiv_worker_scope_from_store(api, workspace_id, context)?;
+    let authority = match scope.authority {
+        SubjektivStoredScopeAuthority::Subject => SubjektivWorkerAuthority::Subject,
+        SubjektivStoredScopeAuthority::JobConsolidation => SubjektivWorkerAuthority::Consolidation,
+        SubjektivStoredScopeAuthority::LegacyConsolidation { singleton_key } => {
+            let mut summary = api
+                .runtime
+                .worker(&scope.worker)
+                .map_err(|error| error.into_error())?;
+            summary.singleton_key = Some(singleton_key.clone());
+            require_dedicated_subjektiv_consolidation_worker(
+                &summary,
+                workspace_id,
+                &singleton_key,
+            )?;
+            SubjektivWorkerAuthority::Consolidation
+        }
+    };
+    Ok((scope.subject_id, scope.worker, authority))
 }
 
 fn require_subjektiv_worker_authority(
@@ -20436,8 +20483,8 @@ fn subjektiv_session_attribution(
     context: &server_api::ServerRequestContext,
     session_id: &str,
 ) -> ApiResult<crate::subjektiv::SubjectSessionAttribution> {
-    let (subject_id, worker, authority) = subjektiv_subject_scope(api, workspace_id, context)?;
+    let scope = subjektiv_worker_scope_from_store(api, workspace_id, context)?;
-    if authority != SubjektivWorkerAuthority::Subject {
+    if !matches!(scope.authority, SubjektivStoredScopeAuthority::Subject) {
         return Err(Error::WorkspacePermissionDenied(
             "subject Session attribution requires the current subject Worker".into(),
         )
@@ -20448,9 +20495,9 @@ fn subjektiv_session_attribution(
         return Err(Error::InvalidInput("committed session_id is invalid".to_string()).into());
     }
     crate::subjektiv::SubjectSessionAttribution::new(
-        subject_id,
+        scope.subject_id,
-        worker.runtime_id,
+        scope.worker.runtime_id,
-        worker.worker_id,
+        scope.worker.worker_id,
         session_id,
     )
     .map_err(|error| Error::InvalidInput(error.to_string()).into())
@@ -20885,8 +20932,11 @@ async fn scoped_record_subjektiv_session(
     context: server_api::ServerRequestContext,
     Json(request): Json<server_api::SubjektivRecordSessionRequest>,
 ) -> ApiResult<Json<server_api::SubjektivRecordSessionResponse>> {
-    api.host_executors
+    // Runtime create/restore waits for this mandatory callback while holding
-        .host()?
+    // Host capacity. Its Subject-only authority and whole attribution SQL must
+    // use the independent DB lane, never the parent's waiting Host lane.
+    api.store
+        .db_executor()
         .run(move || {
             validate_workspace_scope(&api, &path.workspace_id)?;
             let attribution = if request.create_if_missing {
@@ -22251,63 +22301,86 @@ async fn scoped_subjektiv_memory_backend(
     context: server_api::ServerRequestContext,
     Json(request): Json<server_api::SubjektivMemoryBackendRequest>,
 ) -> ApiResult<Json<server_api::SubjektivMemoryBackendResponse>> {
-    api.host_executors
+    use server_api::SubjektivMemoryBackendOperation as Op;
-        .host()?
+    // Subject-only operations are complete DB boundaries. This includes resident
+    // context callbacks during Runtime create/restore; they must not queue behind
+    // the Host parents waiting for them. Other roles can need legacy Runtime checks.
+    let (expected, label) = match &request.operation {
+        Op::ResidentSummary(_) => (
+            Some(SubjektivWorkerAuthority::Subject),
+            "resident surface read",
+        ),
+        Op::ResidentContext(_) => (
+            Some(SubjektivWorkerAuthority::Subject),
+            "resident subject context read",
+        ),
+        Op::ValidateProposal(_) => (
+            Some(SubjektivWorkerAuthority::Subject),
+            "Memory proposal validation",
+        ),
+        Op::ReceiptStatus(_) => (
+            Some(SubjektivWorkerAuthority::Subject),
+            "explicit candidate receipt lookup",
+        ),
+        Op::StageExplicit(_) => (
+            Some(SubjektivWorkerAuthority::Subject),
+            "explicit candidate staging",
+        ),
+        Op::ListCandidates(_) => (
+            Some(SubjektivWorkerAuthority::Consolidation),
+            "candidate listing",
+        ),
+        Op::ReadCandidate(_) => (
+            Some(SubjektivWorkerAuthority::Consolidation),
+            "candidate read",
+        ),
+        Op::DecideCandidate(_) => (
+            Some(SubjektivWorkerAuthority::Consolidation),
+            "candidate decision",
+        ),
+        Op::PrepareSurface(_) => (
+            Some(SubjektivWorkerAuthority::Consolidation),
+            "surface generation preparation",
+        ),
+        Op::PublishSurface(_) => (
+            Some(SubjektivWorkerAuthority::Consolidation),
+            "surface publication",
+        ),
+        Op::FailSurface(_) => (
+            Some(SubjektivWorkerAuthority::Consolidation),
+            "surface generation failure",
+        ),
+        Op::Query(_) | Op::Read(_) | Op::ListChanges(_) => (None, "recall"),
+    };
+    let executor = if expected == Some(SubjektivWorkerAuthority::Subject) {
+        api.store.db_executor()
+    } else {
+        api.host_executors.host()?
+    };
+    executor
         .run(move || {
             validate_workspace_scope(&api, &path.workspace_id)?;
-            let (subject_id, source_worker, authority) =
+            let (subject_id, source_worker, authority) = if expected
-                subjektiv_subject_scope(&api, &path.workspace_id, &context)?;
+                == Some(SubjektivWorkerAuthority::Subject)
+            {
+                let scope = subjektiv_worker_scope_from_store(&api, &path.workspace_id, &context)?;
+                if !matches!(scope.authority, SubjektivStoredScopeAuthority::Subject) {
+                    return Err(Error::WorkspacePermissionDenied(format!(
+                        "subjektiv {label} is not authorized for this Worker role"
+                    ))
+                    .into());
+                }
+                (
+                    scope.subject_id,
+                    scope.worker,
+                    SubjektivWorkerAuthority::Subject,
+                )
+            } else {
+                subjektiv_subject_scope(&api, &path.workspace_id, &context)?
+            };
             let store = open_subjektiv_store(&api)?;
-            use server_api::SubjektivMemoryBackendOperation as Op;
             // Authentication, live Worker/Job authority and committed Session provenance
             // remain Backend-owned. Only the domain operation/projection is shared.
-            let (expected, label) = match &request.operation {
-                Op::ResidentSummary(_) => (
-                    Some(SubjektivWorkerAuthority::Subject),
-                    "resident surface read",
-                ),
-                Op::ResidentContext(_) => (
-                    Some(SubjektivWorkerAuthority::Subject),
-                    "resident subject context read",
-                ),
-                Op::ValidateProposal(_) => (
-                    Some(SubjektivWorkerAuthority::Subject),
-                    "Memory proposal validation",
-                ),
-                Op::ReceiptStatus(_) => (
-                    Some(SubjektivWorkerAuthority::Subject),
-                    "explicit candidate receipt lookup",
-                ),
-                Op::StageExplicit(_) => (
-                    Some(SubjektivWorkerAuthority::Subject),
-                    "explicit candidate staging",
-                ),
-                Op::ListCandidates(_) => (
-                    Some(SubjektivWorkerAuthority::Consolidation),
-                    "candidate listing",
-                ),
-                Op::ReadCandidate(_) => (
-                    Some(SubjektivWorkerAuthority::Consolidation),
-                    "candidate read",
-                ),
-                Op::DecideCandidate(_) => (
-                    Some(SubjektivWorkerAuthority::Consolidation),
-                    "candidate decision",
-                ),
-                Op::PrepareSurface(_) => (
-                    Some(SubjektivWorkerAuthority::Consolidation),
-                    "surface generation preparation",
-                ),
-                Op::PublishSurface(_) => (
-                    Some(SubjektivWorkerAuthority::Consolidation),
-                    "surface publication",
-                ),
-                Op::FailSurface(_) => (
-                    Some(SubjektivWorkerAuthority::Consolidation),
-                    "surface generation failure",
-                ),
-                Op::Query(_) | Op::Read(_) | Op::ListChanges(_) => (None, "recall"),
-            };
             if let Some(expected) = expected {
                 require_subjektiv_worker_authority(authority, expected, label)?;
             }

crates/workspace-server/src/server/tests/subject_spawn_tests.rs

@@ -1,34 +1,45 @@
 use super::*;
 
-#[tokio::test]
+#[tokio::test(flavor = "current_thread")]
-async fn subject_session_callback_completes_while_create_and_deletion_are_waiting() {
+async fn subject_callbacks_finish_before_waiting_host_parents_release_capacity() {
     let workspace = tempfile::tempdir().unwrap();
     let mut api = test_api(workspace.path()).await;
     let identity = RuntimeIdentityMaterial::generate("runtime-test").unwrap();
     configure_runtime_request_auth(&mut api, &identity, "runtime-test");
     let subjektiv = open_subjektiv_store(&api).unwrap();
-    let subject = subjektiv
-        .create_subject(crate::subjektiv::SubjectRole::new("companion").unwrap())
-        .unwrap();
-    let singleton_key = crate::subjektiv::subject_worker_singleton_key(&subject.id).unwrap();
     let memory = api
         .config_store
         .get_workspace_memory_settings(TEST_WORKSPACE_ID)
         .unwrap();
-    let reserved = api
+    let mut callbacks = Vec::new();
-        .config_store
+    for index in 0..2 {
-        .reserve_worker_create(
+        let subject = subjektiv
-            TEST_WORKSPACE_ID,
+            .create_subject(
-            "runtime-test",
+                crate::subjektiv::SubjectRole::new(format!("companion-{index}")).unwrap(),
-            "subject-start",
+            )
-            &"f".repeat(64),
+            .unwrap();
-            Some(&singleton_key),
+        let singleton_key = crate::subjektiv::subject_worker_singleton_key(&subject.id).unwrap();
-            &memory,
+        let reserved = api
-        )
+            .config_store
-        .unwrap();
+            .reserve_worker_create(
-    let worker_id = reserved.worker_id.to_string();
+                TEST_WORKSPACE_ID,
+                "runtime-test",
+                &format!("subject-start-{index}"),
+                &"f".repeat(64),
+                Some(&singleton_key),
+                &memory,
+            )
+            .unwrap();
+        callbacks.push((
+            subject.id,
+            reserved.worker_id.to_string(),
+            session_store::new_session_id().to_string(),
+        ));
+    }
     let server = WorkspaceServerApi::new_for_test(api.config.clone(), api.store.clone());
-    // Reuse the fixture API, but exercise the real outer dispatcher and both auth layers.
+    // Real dispatcher, source proof authentication, live singleton authority and
+    // attribution SQL. Only Runtime's create/restore wait is modeled below; no
+    // product process or complete create endpoint is spawned by this route test.
     server
         .apis
         .lock()
@@ -39,61 +50,210 @@ async fn subject_session_callback_completes_while_create_and_deletion_are_waitin
         build_inner_router(api.clone()),
     );
     let gate = server.admission_for_workspace(TEST_WORKSPACE_ID).await;
-    let create = gate.admit(false).unwrap();
+    let host = api.host_executors.host().unwrap(); // production 2 workers/16 queued
-    let mut deletion = Box::pin(gate.lock_deletion());
+    let mut parents = Vec::new();
-    assert!(futures::poll!(&mut deletion).is_pending());
+    let mut finish_callbacks = Vec::new();
+    for _ in 0..2 {
+        let create = gate.admit(false).unwrap();
+        let (entered, ready) = tokio::sync::oneshot::channel();
+        let (finish_callback, callback_reply) = std::sync::mpsc::channel();
+        finish_callbacks.push(finish_callback);
+        parents.push(tokio::spawn(run_host_workflow(host.clone(), async move {
+            let _create = create;
+            entered.send(()).unwrap();
+            // The actual Host parent cannot finish until its mandatory callbacks
+            // return. Timeout/disconnection is failure cleanup, never success.
+            callback_reply
+                .recv_timeout(std::time::Duration::from_secs(5))
+                .map_err(|error| {
+                    Error::InvalidInput(format!("Runtime callback did not complete: {error}"))
+                })?;
+            Ok::<_, Error>(())
+        })));
+        tokio::time::timeout(std::time::Duration::from_secs(2), ready)
+            .await
+            .unwrap()
+            .unwrap();
+    }
     let app = workspace_server_router(server);
     let path = format!("/api/w/{TEST_WORKSPACE_ID}/subjektiv/sessions");
-    let session_id = session_store::new_session_id().to_string();
+    let memory_path = format!("/api/w/{TEST_WORKSPACE_ID}/subjektiv/memory");
-    let body = serde_json::to_vec(&json!({ "session_id": session_id, "create_if_missing": true }))
+    for (_, worker_id, _) in &callbacks {
-        .unwrap();
+        // Resident context is also a pure Subject callback used by Runtime
-    let response = tokio::time::timeout(
+        // bootstrap/restore. It must not reenter the waiting Host lane either.
-        std::time::Duration::from_secs(2),
+        let response = tokio::time::timeout(
-        app.clone().oneshot(runtime_source_request(
+            std::time::Duration::from_secs(2),
-            &identity,
+            app.clone().oneshot(runtime_source_request(
-            Some(&worker_id),
+                &identity,
-            "POST",
+                Some(worker_id),
-            &path,
+                "POST",
-            body.clone(),
+                &memory_path,
-        )),
+                serde_json::to_vec(&server_api::SubjektivMemoryBackendRequest {
-    )
+                    operation: server_api::SubjektivMemoryBackendOperation::ResidentContext(
-    .await
+                        Default::default(),
-    .expect("Session attribution must not wait for its own Worker creation")
+                    ),
-    .unwrap();
+                })
-    let status = response.status();
-    let response_body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
-    assert_eq!(status, StatusCode::OK, "{response_body:?}");
-    let response: Value = serde_json::from_slice(&response_body).unwrap();
-    assert_eq!(response["subject_id"], subject.id);
-    assert_eq!(response["session_id"], session_id);
-    let stored = subjektiv.session_attribution(&session_id).unwrap().unwrap();
-    assert_eq!(stored.worker_id, worker_id);
-    assert_eq!(stored.subject_id, subject.id);
-
-    // The callback lane does not bypass authentication.
-    let response = tokio::time::timeout(
-        std::time::Duration::from_secs(2),
-        app.clone().oneshot(
-            Request::builder()
-                .method("POST")
-                .uri(&path)
-                .header(CONTENT_TYPE, "application/json")
-                .body(Body::from(body))
                 .unwrap(),
-        ),
+            )),
-    )
+        )
-    .await
+        .await
-    .unwrap()
+        .expect("resident context must not wait for its Host parents")
-    .unwrap();
+        .unwrap();
-    assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
+        let status = response.status();
+        let body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
+        assert_eq!(status, StatusCode::OK, "{body:?}");
+        let resident: server_api::SubjektivMemoryBackendResponse =
+            serde_json::from_slice(&body).unwrap();
+        assert!(matches!(
+            resident,
+            server_api::SubjektivMemoryBackendResponse::ResidentContext(_)
+        ));
+    }
+    // Resident memory is not an allowed deletion-drain callback. Preserve that
+    // admission fence; only the exact Session attribution route reenters below.
+    let mut deletion = Box::pin(gate.lock_deletion());
     assert!(futures::poll!(&mut deletion).is_pending());
-    drop(create);
+    for (subject_id, worker_id, session_id) in &callbacks {
+        let body =
+            serde_json::to_vec(&json!({ "session_id": session_id, "create_if_missing": true }))
+                .unwrap();
+        let response = tokio::time::timeout(
+            std::time::Duration::from_secs(2),
+            app.clone().oneshot(runtime_source_request(
+                &identity,
+                Some(worker_id),
+                "POST",
+                &path,
+                body,
+            )),
+        )
+        .await
+        .expect("Session attribution must not wait for its Host parents")
+        .unwrap();
+        let status = response.status();
+        let response_body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
+        assert_eq!(status, StatusCode::OK, "{response_body:?}");
+        let response: Value = serde_json::from_slice(&response_body).unwrap();
+        assert_eq!(response["subject_id"], *subject_id);
+        assert_eq!(response["session_id"], *session_id);
+        let stored = subjektiv.session_attribution(session_id).unwrap().unwrap();
+        assert_eq!(stored.worker_id, *worker_id);
+        assert_eq!(stored.subject_id, *subject_id);
+    }
+    // Neither actual parent nor its admission lease was released to make these
+    // callbacks succeed. Only durable attribution completion releases them.
+    assert!(parents.iter().all(|parent| !parent.is_finished()));
+    assert!(futures::poll!(&mut deletion).is_pending());
+    for finish_callback in finish_callbacks {
+        finish_callback.send(()).unwrap();
+    }
+    for parent in parents {
+        parent.await.unwrap().unwrap();
+    }
+    host.drain().await.unwrap();
+    api.store.db_executor().drain().await.unwrap();
     let _deletion = tokio::time::timeout(std::time::Duration::from_secs(2), deletion)
         .await
         .unwrap();
     assert!(gate.admit(true).is_none());
 }
 
+#[tokio::test]
+async fn consolidation_subject_callbacks_are_denied_without_runtime_lookup() {
+    let workspace = tempfile::tempdir().unwrap();
+    let mut api = test_api(workspace.path()).await;
+    let identity = RuntimeIdentityMaterial::generate("runtime-test").unwrap();
+    configure_runtime_request_auth(&mut api, &identity, "runtime-test");
+    let subjektiv = open_subjektiv_store(&api).unwrap();
+    let subject = subjektiv
+        .create_subject(crate::subjektiv::SubjectRole::new("companion").unwrap())
+        .unwrap();
+    let key = format!("{SUBJEKTIV_CONSOLIDATION_SINGLETON_PREFIX}{}", subject.id);
+    let memory = api
+        .config_store
+        .get_workspace_memory_settings(TEST_WORKSPACE_ID)
+        .unwrap();
+    let reservation = api
+        .config_store
+        .reserve_worker_create(
+            TEST_WORKSPACE_ID,
+            "runtime-test",
+            "consolidation-start",
+            &"f".repeat(64),
+            Some(&key),
+            &memory,
+        )
+        .unwrap();
+    // runtime-test has no Worker adapter: authority denial must not try to look
+    // up its profile/tags through Runtime before rejecting this disallowed role.
+    let server = WorkspaceServerApi::new_for_test(api.config.clone(), api.store.clone());
+    server
+        .apis
+        .lock()
+        .await
+        .insert(TEST_WORKSPACE_ID.into(), api.clone());
+    server
+        .routers
+        .lock()
+        .await
+        .insert(TEST_WORKSPACE_ID.into(), build_inner_router(api.clone()));
+    let app = workspace_server_router(server);
+    let path = format!("/api/w/{TEST_WORKSPACE_ID}/subjektiv/sessions");
+    let worker_id = reservation.worker_id.to_string();
+    let session_id = session_store::new_session_id().to_string();
+    for (callback_path, callback_body) in [
+        (
+            path.clone(),
+            serde_json::to_vec(&json!({"session_id": session_id, "create_if_missing": true}))
+                .unwrap(),
+        ),
+        (
+            format!("/api/w/{TEST_WORKSPACE_ID}/subjektiv/memory"),
+            serde_json::to_vec(&server_api::SubjektivMemoryBackendRequest {
+                operation: server_api::SubjektivMemoryBackendOperation::ResidentContext(
+                    Default::default(),
+                ),
+            })
+            .unwrap(),
+        ),
+    ] {
+        let response = app
+            .clone()
+            .oneshot(runtime_source_request(
+                &identity,
+                Some(&worker_id),
+                "POST",
+                &callback_path,
+                callback_body,
+            ))
+            .await
+            .unwrap();
+        let status = response.status();
+        let body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
+        assert_eq!(status, StatusCode::FORBIDDEN, "{callback_path}: {body:?}");
+    }
+    assert!(
+        subjektiv
+            .session_attribution(&session_id)
+            .unwrap()
+            .is_none()
+    );
+    let response = app
+        .oneshot(
+            Request::builder()
+                .method("POST")
+                .uri(path)
+                .header(CONTENT_TYPE, "application/json")
+                .body(Body::from(
+                    json!({"session_id": session_id, "create_if_missing": true}).to_string(),
+                ))
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+    assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
+}
+
 #[tokio::test]
 async fn workspace_deletion_waits_for_session_attribution_and_fences_both_lanes() {
     let gate = Arc::new(WorkspaceAdmission::default());