feat: propagate workspace prompt revisions

crates/worker-runtime/src/execution.rs

@@ -357,6 +357,16 @@ pub trait WorkerExecutionBackend: Send + Sync + 'static {
         ))
     }
 
+    /// Observe a newer immutable Workspace Prompt projection. Profile-backed
+    /// execution uses this as a revision notification; other backends may
+    /// safely ignore it.
+    fn observe_workspace_prompt_projection(
+        &self,
+        _projection: worker::WorkspacePromptProjection,
+    ) -> Result<(), String> {
+        Ok(())
+    }
+
     fn dispatch_input(
         &self,
         handle: &WorkerExecutionHandle,
@@ -469,6 +479,13 @@ impl WorkerExecutionBackendRef {
         self.backend.cleanup_working_directory(working_directory_id)
     }
 
+    pub(crate) fn observe_workspace_prompt_projection(
+        &self,
+        projection: worker::WorkspacePromptProjection,
+    ) -> Result<(), String> {
+        self.backend.observe_workspace_prompt_projection(projection)
+    }
+
     pub(crate) fn dispatch_input(
         &self,
         handle: &WorkerExecutionHandle,

crates/worker-runtime/src/http_server.rs

@@ -195,6 +195,10 @@ fn runtime_http_router_with_optional_auth(
             "/v1/config-bundles/{bundle_id}/availability",
             get(check_config_bundle),
         )
+        .route(
+            "/v1/workspace-prompt-projections",
+            post(observe_workspace_prompt_projection),
+        )
         .route(
             "/v1/working-directories",
             get(list_working_directories).post(create_working_directory),
@@ -285,6 +289,18 @@ pub struct RuntimeHttpConfigBundleSyncRequest {
     pub bundle: ConfigBundle,
 }
 
+/// Server-owned notification carrying the Workspace's current immutable Prompt projection.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct RuntimeHttpWorkspacePromptProjectionRequest {
+    pub projection: worker::WorkspacePromptProjection,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct RuntimeHttpWorkspacePromptProjectionResponse {
+    pub workspace_id: String,
+    pub config_revision: u64,
+}
+
 /// Config bundle availability response used by sync/check endpoints.
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub struct RuntimeHttpConfigBundleAvailabilityResponse {
@@ -431,6 +447,23 @@ async fn store_config_bundle(
     }))
 }
 
+async fn observe_workspace_prompt_projection(
+    State(state): State<RuntimeHttpState>,
+    body: Result<Json<RuntimeHttpWorkspacePromptProjectionRequest>, JsonRejection>,
+) -> RestResult<RuntimeHttpWorkspacePromptProjectionResponse> {
+    let Json(request) = body.map_err(RuntimeHttpRestError::json_rejection)?;
+    let workspace_id = request.projection.workspace_id.clone();
+    let config_revision = request.projection.config_revision;
+    state
+        .runtime
+        .observe_workspace_prompt_projection(request.projection)
+        .map_err(RuntimeHttpRestError::runtime)?;
+    Ok(Json(RuntimeHttpWorkspacePromptProjectionResponse {
+        workspace_id,
+        config_revision,
+    }))
+}
+
 async fn check_config_bundle(
     State(state): State<RuntimeHttpState>,
     Path(bundle_id): Path<String>,
@@ -1518,7 +1551,10 @@ fn required_runtime_permission(method: &Method, path: &str) -> Option<&'static s
     {
         return Some("workdirs:operate");
     }
-    if path.starts_with("/v1/config-bundles") || path.starts_with("/v1/working-directories") {
+    if path.starts_with("/v1/config-bundles")
+        || path.starts_with("/v1/workspace-prompt-projections")
+        || path.starts_with("/v1/working-directories")
+    {
         return Some("workers:create");
     }
     if path.ends_with("/workspace-api") {

crates/worker-runtime/src/runtime.rs

@@ -300,6 +300,29 @@ impl Runtime {
         state.check_config_bundle_ref(reference)
     }
 
+    /// Notify the execution backend of the Workspace's current immutable
+    /// Prompt projection. The Runtime keeps this cache outside persisted Worker
+    /// restore authority.
+    pub fn observe_workspace_prompt_projection(
+        &self,
+        projection: worker::WorkspacePromptProjection,
+    ) -> Result<(), RuntimeError> {
+        let backend = {
+            let state = self.lock()?;
+            state.ensure_running()?;
+            state.execution_backend.clone().ok_or_else(|| {
+                RuntimeError::ExecutionBackendUnavailable {
+                    message:
+                        "Workspace Prompt projection notification requires an execution backend"
+                            .to_string(),
+                }
+            })?
+        };
+        backend
+            .observe_workspace_prompt_projection(projection)
+            .map_err(|message| RuntimeError::ExecutionBackendUnavailable { message })
+    }
+
     /// Stop the Runtime.  v0 keeps data readable after stop, but rejects new
     /// create/send/worker lifecycle mutations.
     pub fn stop_runtime(&self) -> Result<(), RuntimeError> {

crates/worker-runtime/src/worker_backend.rs

@@ -84,6 +84,13 @@ pub struct RuntimeWorkerController {
 /// controller-backed Worker for a Runtime catalog entry.
 #[async_trait]
 pub trait RuntimeWorkerFactory: Send + Sync + 'static {
+    fn observe_workspace_prompt_projection(
+        &self,
+        _projection: worker::WorkspacePromptProjection,
+    ) -> Result<(), String> {
+        Ok(())
+    }
+
     async fn spawn_controller(
         &self,
         request: WorkerExecutionSpawnRequest,
@@ -211,12 +218,22 @@ impl WorkerObservationProvider for RuntimeGrantedWorkerObservationProvider {
 }
 
 #[derive(Debug, Default)]
-struct WorkspacePromptProjectionCache {
+pub(crate) struct WorkspacePromptProjectionCache {
     active: Mutex<HashMap<String, Arc<worker::WorkspacePromptProjection>>>,
 }
 
 impl WorkspacePromptProjectionCache {
-    fn observe(
+    pub(crate) fn active(
+        &self,
+        workspace_id: &str,
+    ) -> Result<Option<Arc<worker::WorkspacePromptProjection>>, String> {
+        self.active
+            .lock()
+            .map(|active| active.get(workspace_id).cloned())
+            .map_err(|_| "Workspace Prompt projection cache lock was poisoned".to_string())
+    }
+
+    pub(crate) fn observe(
         &self,
         projection: worker::WorkspacePromptProjection,
     ) -> Result<Arc<worker::WorkspacePromptProjection>, String> {
@@ -448,6 +465,7 @@ impl RuntimeWorkspaceBackendRef {
         workspace_scope: Option<&crate::runtime::RuntimeWorkspaceScope>,
         mutation_identity: Option<&RuntimeIdentityMaterial>,
         embedded_dispatcher: Option<&Arc<dyn EmbeddedWorkerMutationDispatcher>>,
+        prompt_projection_cache: Option<Arc<WorkspacePromptProjectionCache>>,
     ) -> WorkerWorkspaceContext {
         match self {
             Self::None => WorkerWorkspaceContext::no_workspace(),
@@ -462,6 +480,9 @@ impl RuntimeWorkspaceBackendRef {
                     runtime_id.clone(),
                     worker_ref.worker_id.to_string(),
                 );
+                if let Some(cache) = prompt_projection_cache {
+                    client = client.with_prompt_projection_cache(cache);
+                }
                 if let (Some(scope), Some(identity)) = (workspace_scope, mutation_identity) {
                     client = client.with_worker_remove(RuntimeWorkerMutationForwarder::remote(
                         identity,
@@ -560,6 +581,13 @@ fn runtime_local_workdir_session(
 
 #[async_trait]
 impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
+    fn observe_workspace_prompt_projection(
+        &self,
+        projection: worker::WorkspacePromptProjection,
+    ) -> Result<(), String> {
+        self.prompt_projection_cache.observe(projection).map(|_| ())
+    }
+
     async fn spawn_controller(
         &self,
         request: WorkerExecutionSpawnRequest,
@@ -599,6 +627,7 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
             request.workspace_scope.as_ref(),
             self.worker_mutation_identity.as_ref(),
             self.embedded_worker_mutation_dispatcher.as_ref(),
+            Some(self.prompt_projection_cache.clone()),
         );
         let selector = profile.as_ref();
         let archive = self
@@ -785,6 +814,7 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
             request.workspace_scope.as_ref(),
             self.worker_mutation_identity.as_ref(),
             self.embedded_worker_mutation_dispatcher.as_ref(),
+            Some(self.prompt_projection_cache.clone()),
         );
         let (manifest, loader) = Self::restore_fallback_manifest(&worker_name)?;
 
@@ -1341,6 +1371,13 @@ where
         &self.backend_id
     }
 
+    fn observe_workspace_prompt_projection(
+        &self,
+        projection: worker::WorkspacePromptProjection,
+    ) -> Result<(), String> {
+        self.factory.observe_workspace_prompt_projection(projection)
+    }
+
     fn create_working_directory(
         &self,
         request: &WorkingDirectoryRequest,
@@ -1922,6 +1959,46 @@ mod tests {
     use manifest::{Scope, WorkerManifest};
     use session_store::{LogEntry, WorkerMetadataStore};
 
+    #[test]
+    fn workspace_prompt_projection_notification_advances_shared_cache() {
+        let cache = WorkspacePromptProjectionCache::default();
+        let catalog_v1 = worker::EffectivePromptCatalog::new(
+            BTreeMap::from([("default".to_string(), "prompt-v1".to_string())]),
+            8,
+            "schema",
+            "toolchain",
+        )
+        .unwrap();
+        let projection_v1 = worker::WorkspacePromptProjection::new(
+            "workspace-a",
+            "source-v1",
+            catalog_v1.catalog_digest.clone(),
+            catalog_v1,
+        )
+        .unwrap();
+        let catalog_v2 = worker::EffectivePromptCatalog::new(
+            BTreeMap::from([("default".to_string(), "prompt-v2".to_string())]),
+            9,
+            "schema",
+            "toolchain",
+        )
+        .unwrap();
+        let projection_v2 = worker::WorkspacePromptProjection::new(
+            "workspace-a",
+            "source-v2",
+            catalog_v2.catalog_digest.clone(),
+            catalog_v2.clone(),
+        )
+        .unwrap();
+
+        cache.observe(projection_v1).unwrap();
+        cache.observe(projection_v2).unwrap();
+
+        let active = cache.active("workspace-a").unwrap().unwrap();
+        assert_eq!(active.config_revision, 9);
+        assert_eq!(active.catalog.catalog_digest, catalog_v2.catalog_digest);
+    }
+
     #[test]
     fn workspace_prompt_projection_cache_rejects_same_revision_source_drift() {
         let catalog = worker::EffectivePromptCatalog::new(
@@ -1968,12 +2045,12 @@ mod tests {
         let scope = crate::runtime::RuntimeWorkspaceScope::new("workspace-a", "server-main");
 
         let before_restart =
-            backend.worker_context(&worker_ref, Some(&scope), Some(&identity), None);
+            backend.worker_context(&worker_ref, Some(&scope), Some(&identity), None, None);
         let adapter = WorkerRuntimeExecutionBackend::new(FailingFactory).unwrap();
         let (after_restore_kind, after_restore_workspace_id) = adapter
             .run_on_adapter_runtime(async move {
                 let after_restore =
-                    backend.worker_context(&worker_ref, Some(&scope), Some(&identity), None);
+                    backend.worker_context(&worker_ref, Some(&scope), Some(&identity), None, None);
                 let client = after_restore.client_handle();
                 Ok((
                     client.kind().to_string(),
@@ -2163,6 +2240,7 @@ mod tests {
                 request.workspace_scope.as_ref(),
                 None,
                 None,
+                None,
             );
             let workspace_client = workspace_context.client_handle();
             self.observed_workspace_clients.lock().unwrap().push((

crates/worker-runtime/src/worker_source.rs

@@ -2,8 +2,8 @@ use std::sync::Arc;
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
 use worker::{
-    WorkspaceClient, WorkspaceClientError, WorkspaceRequest, WorkspaceRequestMethod,
+    WorkspaceClient, WorkspaceClientError, WorkspacePromptProjection, WorkspaceRequest,
-    WorkspaceResponse,
+    WorkspaceRequestMethod, WorkspaceResponse,
 };
 
 use crate::auth::{
@@ -12,6 +12,7 @@ use crate::auth::{
     WorkerMutationSourceClaims, new_token_id,
 };
 use crate::runtime::RuntimeWorkspaceScope;
+use crate::worker_backend::WorkspacePromptProjectionCache;
 
 pub const DEFAULT_WORKER_MUTATION_SOURCE_TTL_SECONDS: u64 = 60;
 
@@ -293,6 +294,7 @@ pub struct RuntimeOwnedWorkspaceClient {
     worker_id: String,
     request_timeout: Option<Duration>,
     worker_remove: Option<RuntimeWorkerMutationForwarder>,
+    prompt_projection_cache: Option<Arc<WorkspacePromptProjectionCache>>,
 }
 
 impl RuntimeOwnedWorkspaceClient {
@@ -309,6 +311,7 @@ impl RuntimeOwnedWorkspaceClient {
             worker_id: worker_id.into(),
             request_timeout: None,
             worker_remove: None,
+            prompt_projection_cache: None,
         }
     }
 
@@ -317,6 +320,14 @@ impl RuntimeOwnedWorkspaceClient {
         self
     }
 
+    pub(crate) fn with_prompt_projection_cache(
+        mut self,
+        cache: Arc<WorkspacePromptProjectionCache>,
+    ) -> Self {
+        self.prompt_projection_cache = Some(cache);
+        self
+    }
+
     #[cfg(test)]
     fn with_request_timeout(mut self, request_timeout: Option<Duration>) -> Self {
         self.request_timeout = request_timeout;
@@ -385,6 +396,40 @@ impl WorkspaceClient for RuntimeOwnedWorkspaceClient {
         }
     }
 
+    fn current_prompt_projection(
+        &self,
+    ) -> Result<Option<WorkspacePromptProjection>, WorkspaceClientError> {
+        let Some(cache) = self.prompt_projection_cache.as_ref() else {
+            return Ok(None);
+        };
+        if let Some(projection) = cache
+            .active(&self.workspace_id)
+            .map_err(WorkspaceClientError::Request)?
+        {
+            return Ok(Some((*projection).clone()));
+        }
+        let response = self.execute(WorkspaceRequest::get(format!(
+            "/api/w/{}/config-sources/active/prompt-projection",
+            self.workspace_id
+        )))?;
+        if !(200..300).contains(&response.status) {
+            return Err(WorkspaceClientError::Request(format!(
+                "active Workspace Prompt projection request failed with HTTP {}: {}",
+                response.status, response.body
+            )));
+        }
+        let projection: WorkspacePromptProjection =
+            serde_json::from_str(&response.body).map_err(|error| {
+                WorkspaceClientError::Request(format!(
+                    "invalid active Workspace Prompt projection response: {error}"
+                ))
+            })?;
+        let projection = cache
+            .observe(projection)
+            .map_err(WorkspaceClientError::Request)?;
+        Ok(Some((*projection).clone()))
+    }
+
     fn execute_worker_remove(
         &self,
         target_runtime_id: &str,
@@ -521,6 +566,41 @@ mod tests {
         verify_worker_mutation_source_proof,
     };
 
+    #[test]
+    fn current_prompt_projection_uses_the_shared_runtime_cache_without_http() {
+        let cache = Arc::new(WorkspacePromptProjectionCache::default());
+        let catalog = worker::EffectivePromptCatalog::new(
+            std::collections::BTreeMap::from([(
+                "default".to_string(),
+                "workspace prompt".to_string(),
+            )]),
+            3,
+            "schema",
+            "toolchain",
+        )
+        .unwrap();
+        let projection = WorkspacePromptProjection::new(
+            "workspace-a",
+            "source-3",
+            catalog.catalog_digest.clone(),
+            catalog,
+        )
+        .unwrap();
+        cache.observe(projection).unwrap();
+        let client = RuntimeOwnedWorkspaceClient::new(
+            "workspace-a",
+            "http://127.0.0.1:1",
+            "runtime-a",
+            "worker-a",
+        )
+        .with_prompt_projection_cache(cache);
+
+        let projection = client.current_prompt_projection().unwrap().unwrap();
+
+        assert_eq!(projection.config_revision, 3);
+        assert_eq!(projection.source_digest, "source-3");
+    }
+
     #[test]
     fn ordinary_workspace_forwarding_stamps_legacy_source_only_inside_runtime() {
         use std::io::{Read, Write};

crates/workspace-server/src/hosts.rs

@@ -45,6 +45,7 @@ use worker_runtime::http_server::{
     RuntimeHttpWorkerLifecycleResponse, RuntimeHttpWorkerResponse,
     RuntimeHttpWorkerWorkspaceApiRequest, RuntimeHttpWorkersResponse,
     RuntimeHttpWorkingDirectoriesResponse, RuntimeHttpWorkingDirectoryResponse,
+    RuntimeHttpWorkspacePromptProjectionRequest, RuntimeHttpWorkspacePromptProjectionResponse,
 };
 use worker_runtime::identity::{
     RuntimeWorkerRef, WorkerId as EmbeddedWorkerId, WorkerRef as EmbeddedWorkerRef,
@@ -778,6 +779,13 @@ pub trait WorkspaceWorkerRuntime: Send + Sync {
         }
     }
 
+    fn observe_workspace_prompt_projection(
+        &self,
+        _projection: worker::WorkspacePromptProjection,
+    ) -> Result<(), String> {
+        Ok(())
+    }
+
     fn sync_config_bundle(&self, _bundle: ConfigBundle) -> ConfigBundleSyncResult {
         ConfigBundleSyncResult {
             state: WorkerOperationState::Unsupported,
@@ -1185,6 +1193,36 @@ impl RuntimeRegistry {
         Ok(runtime.replace_worker_workspace_api(worker_id, workspace_api))
     }
 
+    pub fn observe_workspace_prompt_projection(
+        &self,
+        projection: worker::WorkspacePromptProjection,
+    ) -> Vec<RuntimeDiagnostic> {
+        let runtimes = self
+            .runtimes
+            .read()
+            .map(|runtimes| runtimes.clone())
+            .unwrap_or_default();
+        runtimes
+            .into_iter()
+            .filter_map(|runtime| {
+                runtime
+                    .observe_workspace_prompt_projection(projection.clone())
+                    .err()
+                    .map(|message| {
+                        diagnostic(
+                            "workspace_prompt_projection_notification_failed",
+                            DiagnosticSeverity::Warning,
+                            format!(
+                                "runtime '{}' rejected Workspace Prompt projection revision {}: {message}",
+                                runtime.runtime_id(), projection.config_revision
+                            ),
+                        )
+                    })
+            })
+            .take(MAX_DIAGNOSTICS)
+            .collect()
+    }
+
     pub fn spawn_worker(
         &self,
         runtime_id: &str,
@@ -2040,6 +2078,15 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime {
         }
     }
 
+    fn observe_workspace_prompt_projection(
+        &self,
+        projection: worker::WorkspacePromptProjection,
+    ) -> Result<(), String> {
+        self.runtime
+            .observe_workspace_prompt_projection(projection)
+            .map_err(|error| error.to_string())
+    }
+
     fn sync_config_bundle(&self, bundle: ConfigBundle) -> ConfigBundleSyncResult {
         match self.runtime.store_config_bundle(bundle) {
             Ok(availability) => ConfigBundleSyncResult {
@@ -3155,6 +3202,18 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
         }
     }
 
+    fn observe_workspace_prompt_projection(
+        &self,
+        projection: worker::WorkspacePromptProjection,
+    ) -> Result<(), String> {
+        self.post_json::<_, RuntimeHttpWorkspacePromptProjectionResponse>(
+            "/v1/workspace-prompt-projections",
+            &RuntimeHttpWorkspacePromptProjectionRequest { projection },
+        )
+        .map(|_| ())
+        .map_err(|error| error.message)
+    }
+
     fn sync_config_bundle(&self, bundle: ConfigBundle) -> ConfigBundleSyncResult {
         let request = RuntimeHttpConfigBundleSyncRequest { bundle };
         match self.post_json::<_, RuntimeHttpConfigBundleAvailabilityResponse>(
@@ -4507,6 +4566,7 @@ mod tests {
         runtime_id: String,
         host_id: String,
         workers: Vec<WorkerSummary>,
+        observed_prompt_revisions: Arc<Mutex<Vec<u64>>>,
     }
 
     impl FixtureRuntime {
@@ -4542,6 +4602,7 @@ mod tests {
                     working_directory: None,
                     diagnostics: Vec::new(),
                 }],
+                observed_prompt_revisions: Arc::new(Mutex::new(Vec::new())),
             }
         }
     }
@@ -4551,6 +4612,17 @@ mod tests {
             &self.runtime_id
         }
 
+        fn observe_workspace_prompt_projection(
+            &self,
+            projection: worker::WorkspacePromptProjection,
+        ) -> Result<(), String> {
+            self.observed_prompt_revisions
+                .lock()
+                .map_err(|_| "prompt projection observations poisoned".to_string())?
+                .push(projection.config_revision);
+            Ok(())
+        }
+
         fn runtime_summary(&self, _limit: usize) -> RuntimeSummary {
             RuntimeSummary {
                 runtime_id: self.runtime_id.clone(),
@@ -4647,6 +4719,36 @@ mod tests {
         assert_eq!(from_runtime_a.label, "worker from runtime a");
     }
 
+    #[test]
+    fn registry_broadcasts_workspace_prompt_projection_revisions() {
+        let runtime =
+            FixtureRuntime::with_worker("runtime-a", "host-a", "worker-a", "worker from runtime a");
+        let observed = runtime.observed_prompt_revisions.clone();
+        let registry = RuntimeRegistry::new(vec![Arc::new(runtime)]);
+        let catalog = worker::EffectivePromptCatalog::new(
+            std::collections::BTreeMap::from([(
+                "default".to_string(),
+                "workspace prompt".to_string(),
+            )]),
+            12,
+            "schema",
+            "toolchain",
+        )
+        .unwrap();
+        let projection = worker::WorkspacePromptProjection::new(
+            "workspace-a",
+            "source-12",
+            catalog.catalog_digest.clone(),
+            catalog,
+        )
+        .unwrap();
+
+        let diagnostics = registry.observe_workspace_prompt_projection(projection);
+
+        assert!(diagnostics.is_empty());
+        assert_eq!(*observed.lock().unwrap(), vec![12]);
+    }
+
     #[test]
     fn registry_worker_list_can_be_scoped_by_runtime_id() {
         let registry = RuntimeRegistry::new(vec![

crates/workspace-server/src/server.rs

@@ -2673,6 +2673,14 @@ async fn scoped_commit_workspace_config_tree(
     let state = api
         .config_store
         .commit_evaluated_workspace_config(&path.workspace_id, &candidate)?;
+    if let Ok(projection) = api
+        .prompt_projection_cache
+        .resolve(&path.workspace_id, &state)
+    {
+        let _diagnostics = api
+            .runtime
+            .observe_workspace_prompt_projection((*projection).clone());
+    }
     Ok((
         StatusCode::CREATED,
         Json(WorkspaceConfigTreeResponse {