From 17d6789b41cbd3e99d265e74ed9bc4e805f07345 Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 19 Aug 2026 06:01:03 +0900 Subject: [PATCH] feat: propagate workspace prompt revisions --- crates/worker-runtime/src/execution.rs | 17 ++++ crates/worker-runtime/src/http_server.rs | 38 +++++++- crates/worker-runtime/src/runtime.rs | 23 +++++ crates/worker-runtime/src/worker_backend.rs | 86 ++++++++++++++++- crates/worker-runtime/src/worker_source.rs | 84 +++++++++++++++- crates/workspace-server/src/hosts.rs | 102 ++++++++++++++++++++ crates/workspace-server/src/server.rs | 8 ++ 7 files changed, 351 insertions(+), 7 deletions(-) diff --git a/crates/worker-runtime/src/execution.rs b/crates/worker-runtime/src/execution.rs index 381e0c84..642e23e5 100644 --- a/crates/worker-runtime/src/execution.rs +++ b/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, diff --git a/crates/worker-runtime/src/http_server.rs b/crates/worker-runtime/src/http_server.rs index 3bdfbeff..f345ae02 100644 --- a/crates/worker-runtime/src/http_server.rs +++ b/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, + body: Result, JsonRejection>, +) -> RestResult { + 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, Path(bundle_id): Path, @@ -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") { diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index ec308e7f..dd8645f7 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/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> { diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index 2ee34bb3..963fe5c3 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/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>>, } impl WorkspacePromptProjectionCache { - fn observe( + pub(crate) fn active( + &self, + workspace_id: &str, + ) -> Result>, 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, String> { @@ -448,6 +465,7 @@ impl RuntimeWorkspaceBackendRef { workspace_scope: Option<&crate::runtime::RuntimeWorkspaceScope>, mutation_identity: Option<&RuntimeIdentityMaterial>, embedded_dispatcher: Option<&Arc>, + prompt_projection_cache: Option>, ) -> 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(( diff --git a/crates/worker-runtime/src/worker_source.rs b/crates/worker-runtime/src/worker_source.rs index fd8daa44..43a838b0 100644 --- a/crates/worker-runtime/src/worker_source.rs +++ b/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, - WorkspaceResponse, + WorkspaceClient, WorkspaceClientError, WorkspacePromptProjection, WorkspaceRequest, + 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, worker_remove: Option, + prompt_projection_cache: Option>, } 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, + ) -> Self { + self.prompt_projection_cache = Some(cache); + self + } + #[cfg(test)] fn with_request_timeout(mut self, request_timeout: Option) -> Self { self.request_timeout = request_timeout; @@ -385,6 +396,40 @@ impl WorkspaceClient for RuntimeOwnedWorkspaceClient { } } + fn current_prompt_projection( + &self, + ) -> Result, 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}; diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index ab94b0b7..6087d38d 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/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 { + 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, + observed_prompt_revisions: Arc>>, } 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![ diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 718312d4..e282e916 100644 --- a/crates/workspace-server/src/server.rs +++ b/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 {