Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

Merge branch 'orchestration' into develop

Browse Source 
# Conflicts:
#	web/workspace/deno.json
This commit is contained in:
Hare
2026-08-14 14:49:01 +09:00
parent ecbd003579 ace134e2f0
commit 1ed835fb79
81 changed files with 3997 additions and 6039 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker-runtime/src/config_bundle.rs
+45
View File
@@ -24,6 +24,8 @@ pub struct ConfigBundle {
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub declarations: Vec<ConfigDeclaration>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub prompt_catalog: Option<worker::EffectivePromptCatalog>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub profile_source_archive: Option<ProfileSourceArchive>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub profile_source_archive_handle: Option<BackendResourceHandle>,
@@ -69,6 +71,16 @@ impl ConfigBundle {
));
}
if let Some(prompt_catalog) = &self.prompt_catalog {
lines.push(format!(
"prompt_catalog\0{}\0{}\0{}\0{}",
prompt_catalog.config_revision,
prompt_catalog.schema_fingerprint,
prompt_catalog.toolchain_fingerprint,
prompt_catalog.catalog_digest
));
}
if let Some(archive) = &self.profile_source_archive {
lines.push(format!(
"profile_archive\0{}\0{}\0{}",
@@ -274,6 +286,12 @@ pub(crate) fn validate_config_bundle(bundle: &ConfigBundle) -> Result<(), Runtim
validate_declaration_reference(&bundle.metadata.id, declaration)?;
}
if let Some(prompt_catalog) = &bundle.prompt_catalog {
prompt_catalog.verify_digest().map_err(|error| {
RuntimeError::InvalidRequest(format!("invalid Prompt catalog projection: {error}"))
})?;
}
if let Some(archive) = &bundle.profile_source_archive {
validate_profile_source_archive_ref(&archive.reference).map_err(|err| {
RuntimeError::InvalidRequest(format!("invalid profile source archive: {err}"))
@@ -582,6 +600,7 @@ mod tests {
name: "credential".to_string(),
reference: reference.to_string(),
}],
prompt_catalog: None,
profile_source_archive: None,
profile_source_archive_handle: None,
}
@@ -615,6 +634,32 @@ mod tests {
validate_config_bundle(&bundle_with_declaration("vault:team.api-key")).unwrap();
}
#[test]
fn validates_immutable_prompt_catalog_projection() {
let mut bundle = bundle_with_declaration("secret:github-token");
bundle.prompt_catalog = Some(
worker::EffectivePromptCatalog::new(
std::collections::BTreeMap::from([("default".to_string(), "hello".to_string())]),
7,
"schema",
"toolchain",
)
.unwrap(),
);
bundle = bundle.with_computed_digest();
validate_config_bundle(&bundle).unwrap();
bundle
.prompt_catalog
.as_mut()
.unwrap()
.templates
.insert("default".into(), "tampered".into());
bundle = bundle.with_computed_digest();
let error = validate_config_bundle(&bundle).unwrap_err();
assert!(error.to_string().contains("catalog digest mismatch"));
}
#[test]
fn bundle_summary_redacts_runtime_internal_resource_handle() {
let mut bundle = bundle_with_declaration("secret:github-token");
crates/worker-runtime/src/fs_store.rs
+8 -1
View File
@@ -1,5 +1,5 @@
use crate::catalog::{CreateWorkerRequest, WorkingDirectoryStatus};
use crate::config_bundle::ConfigBundle;
use crate::config_bundle::{ConfigBundle, validate_config_bundle};
use crate::diagnostics::{DiagnosticSeverity, RuntimeDiagnostic};
use crate::error::RuntimeError;
use crate::identity::{WorkerId, WorkerRef};
@@ -324,6 +324,13 @@ impl RuntimeSnapshot {
message: format!("runtime snapshot backend is {:?}", self.backend),
});
}
for bundle in self.config_bundles.values() {
validate_config_bundle(bundle).map_err(|error| RuntimeError::StoreCorrupt {
operation: "read runtime snapshot",
path: path.to_path_buf(),
message: format!("invalid config bundle {}: {error}", bundle.metadata.id),
})?;
}
Ok(())
}
crates/worker-runtime/src/http_server.rs
+12
View File
@@ -1811,12 +1811,21 @@ mod tests {
label: Some("test".to_string()),
}],
declarations: Vec::new(),
prompt_catalog: None,
profile_source_archive: None,
profile_source_archive_handle: None,
}
.with_computed_digest()
}
fn store_coder_test_bundle(runtime: &Runtime) {
runtime
.store_config_bundle(test_bundle(ProfileSelector::Builtin(
"builtin:coder".to_string(),
)))
.unwrap();
}
fn scoped_task_request(objective: &str, workspace_id: &str) -> CreateWorkerRequest {
let mut request = task_request(objective);
request.workspace_api = Some(WorkspaceApiRef {
@@ -1922,6 +1931,7 @@ mod tests {
let runtime =
Runtime::with_execution_backend(RuntimeOptions::default(), Arc::new(AcceptingBackend))
.unwrap();
store_coder_test_bundle(&runtime);
let (auth, signer) = auth_config_and_signer();
let token_a = token_for_workspace(&signer, "workspace-a");
let token_b = token_for_workspace(&signer, "workspace-b");
@@ -2002,6 +2012,7 @@ mod tests {
let runtime =
Runtime::with_execution_backend(RuntimeOptions::default(), Arc::new(AcceptingBackend))
.unwrap();
store_coder_test_bundle(&runtime);
let (auth, signer_a, signer_b) = auth_config_and_two_signers();
let token_a = token_for_workspace(&signer_a, "workspace-a");
let token_b = token_for_workspace(&signer_b, "workspace-a");
@@ -2648,6 +2659,7 @@ mod ws_tests {
label: Some("ws".to_string()),
}],
declarations: Vec::new(),
prompt_catalog: None,
profile_source_archive: None,
profile_source_archive_handle: None,
}
crates/worker-runtime/src/profile_archive.rs
+3 -3
View File
@@ -743,10 +743,10 @@ mod tests {
.load(Some("profiles/main.dcdl"), "./shared.dcdl")
.unwrap();
match loaded {
LoadedImport::Source(source) => {
assert_eq!(source.key, "profiles/shared.dcdl");
LoadedImport::Source { key, .. } => {
assert_eq!(key, "profiles/shared.dcdl");
}
LoadedImport::Value(_) => panic!("expected source import"),
LoadedImport::Value { .. } => panic!("expected source import"),
}
}
crates/worker-runtime/src/runtime.rs
+110 -25
View File
@@ -261,6 +261,19 @@ impl Runtime {
digest: bundle.metadata.digest.clone(),
};
let summary = bundle.summary();
if let Some(existing) = state.config_bundles.get(&bundle.metadata.id) {
if existing.metadata.digest != bundle.metadata.digest {
return Err(RuntimeError::ConfigBundleDigestMismatch {
bundle_id: bundle.metadata.id.clone(),
expected_digest: existing.metadata.digest.clone(),
actual_digest: bundle.metadata.digest.clone(),
});
}
return Ok(ConfigBundleAvailability {
reference,
summary: existing.summary(),
});
}
state
.config_bundles
.insert(bundle.metadata.id.clone(), bundle);
@@ -526,6 +539,7 @@ impl Runtime {
message: "worker creation requires an execution backend".to_string(),
}
})?;
let config_bundle = state.resolve_config_bundle_ref(request.config_bundle.as_ref())?;
let worker_id = WorkerId::generated(state.next_worker_sequence);
state.next_worker_sequence += 1;
@@ -551,7 +565,7 @@ impl Runtime {
workspace_scope: scope.cloned(),
context: self.execution_context(worker_ref.clone()),
working_directory: None,
config_bundle: None,
config_bundle,
};
(backend, worker_ref, spawn_request)
};
@@ -868,7 +882,7 @@ impl Runtime {
let (backend, request) = {
let mut state = self.lock()?;
state.ensure_running()?;
let (worker_request, previous_working_directory, config_bundle, run_generation) = {
let (worker_request, previous_working_directory, run_generation) = {
let worker = state.worker(worker_ref)?;
if worker.execution_handle.is_some() {
return Ok(worker.detail());
@@ -879,19 +893,14 @@ impl Runtime {
worker_ref.worker_id
)));
}
let config_bundle = worker
.request
.config_bundle
.as_ref()
.and_then(|bundle_ref| state.config_bundles.get(&bundle_ref.id))
.cloned();
(
worker.request.clone(),
worker.working_directory.clone(),
config_bundle,
worker.run_generation.saturating_add(1).max(1),
)
};
let config_bundle =
state.resolve_config_bundle_ref(worker_request.config_bundle.as_ref())?;
let backend = state.execution_backend.clone().ok_or_else(|| {
RuntimeError::WorkerExecutionUnavailable {
worker_id: worker_ref.worker_id.clone(),
@@ -1558,31 +1567,20 @@ impl Runtime {
.collect::<Vec<_>>();
let mut candidates = Vec::with_capacity(worker_ids.len());
for worker_id in worker_ids {
let (
worker_ref,
request,
previous_working_directory,
config_bundle,
run_generation,
) = {
let (worker_ref, request, previous_working_directory, run_generation) = {
let worker = state
.workers
.get(&worker_id)
.expect("collected Worker exists");
let config_bundle = worker
.request
.config_bundle
.as_ref()
.and_then(|bundle_ref| state.config_bundles.get(&bundle_ref.id))
.cloned();
(
worker.worker_ref.clone(),
worker.request.clone(),
worker.working_directory.clone(),
config_bundle,
worker.run_generation.saturating_add(1).max(1),
)
};
let config_bundle =
state.resolve_config_bundle_ref(request.config_bundle.as_ref())?;
state
.workers
.get_mut(&worker_id)
@@ -2075,6 +2073,17 @@ impl RuntimeState {
})
}
fn resolve_config_bundle_ref(
&self,
reference: Option<&ConfigBundleRef>,
) -> Result<Option<ConfigBundle>, RuntimeError> {
let Some(reference) = reference else {
return Ok(None);
};
self.check_config_bundle_ref(reference)?;
Ok(self.config_bundles.get(&reference.id).cloned())
}
fn validate_worker_config_boundary(
&self,
_request: &CreateWorkerRequest,
@@ -2831,6 +2840,7 @@ mod tests {
name: "read".to_string(),
reference: "capability:read".to_string(),
}],
prompt_catalog: None,
profile_source_archive: None,
profile_source_archive_handle: None,
}
@@ -2843,6 +2853,7 @@ mod tests {
restore_result: Mutex<Option<WorkerExecutionSpawnResult>>,
restore_count: Mutex<u64>,
run_generations: Mutex<Vec<u64>>,
config_bundles: Mutex<Vec<Option<ConfigBundle>>>,
contexts: Mutex<BTreeMap<WorkerId, WorkerExecutionContext>>,
dispatched_inputs: Mutex<Vec<WorkerInput>>,
preserve_commit_ack_submission_id: AtomicBool,
@@ -2890,6 +2901,10 @@ mod tests {
.lock()
.unwrap()
.push(request.run_generation);
self.config_bundles
.lock()
.unwrap()
.push(request.config_bundle.clone());
self.contexts
.lock()
.unwrap()
@@ -2913,6 +2928,10 @@ mod tests {
.lock()
.unwrap()
.push(request.run_generation);
self.config_bundles
.lock()
.unwrap()
.push(request.config_bundle.clone());
if let Some(result) = self.restore_result.lock().unwrap().clone() {
return result;
}
@@ -3409,11 +3428,30 @@ mod tests {
#[test]
fn synced_config_bundle_is_stored_checked_and_used_for_worker_creation() {
let runtime = runtime_with_backend();
let bundle = test_bundle();
let backend = Arc::new(TestExecutionBackend::default());
let runtime =
Runtime::with_execution_backend(RuntimeOptions::default(), backend.clone()).unwrap();
let mut bundle = test_bundle();
bundle.prompt_catalog = Some(
worker::EffectivePromptCatalog::new(
BTreeMap::from([("default".to_string(), "workspace prompt".to_string())]),
7,
"schema",
"toolchain",
)
.unwrap(),
);
bundle = bundle.with_computed_digest();
let availability = runtime.store_config_bundle(bundle.clone()).unwrap();
assert_eq!(availability.reference.id, "bundle-1");
assert_eq!(availability.reference.digest, bundle.metadata.digest);
let mut conflicting_bundle = bundle.clone();
conflicting_bundle.profiles[0].label = Some("conflicting".to_string());
conflicting_bundle = conflicting_bundle.with_computed_digest();
assert!(matches!(
runtime.store_config_bundle(conflicting_bundle),
Err(RuntimeError::ConfigBundleDigestMismatch { .. })
));
let listed = runtime.list_config_bundles().unwrap();
assert_eq!(listed.len(), 1);
@@ -3428,6 +3466,53 @@ mod tests {
.create_worker(bundled_task_request("synced", &bundle))
.unwrap();
assert_eq!(detail.config_bundle, Some(availability.reference));
assert_eq!(
backend.config_bundles.lock().unwrap().as_slice(),
&[Some(bundle.clone())]
);
runtime.stop_worker(&detail.worker_ref, None).unwrap();
runtime.restore_worker(&detail.worker_ref).unwrap();
assert_eq!(
backend.config_bundles.lock().unwrap().as_slice(),
&[Some(bundle.clone()), Some(bundle)]
);
}
#[test]
fn restore_fails_closed_when_recorded_config_bundle_is_missing_or_mismatched() {
let (runtime, backend) = runtime_and_backend();
let bundle = test_bundle();
let detail = runtime
.create_worker(bundled_task_request("missing-on-restore", &bundle))
.unwrap();
runtime.stop_worker(&detail.worker_ref, None).unwrap();
runtime.lock().unwrap().config_bundles.clear();
assert!(matches!(
runtime.restore_worker(&detail.worker_ref),
Err(RuntimeError::ConfigBundleMissing { .. })
));
assert_eq!(backend.config_bundles.lock().unwrap().len(), 1);
let (runtime, backend) = runtime_and_backend();
let bundle = test_bundle();
let detail = runtime
.create_worker(bundled_task_request("mismatch-on-restore", &bundle))
.unwrap();
runtime.stop_worker(&detail.worker_ref, None).unwrap();
let mut replacement = bundle.clone();
replacement.profiles[0].label = Some("replacement".to_string());
replacement = replacement.with_computed_digest();
runtime
.lock()
.unwrap()
.config_bundles
.insert(replacement.metadata.id.clone(), replacement);
assert!(matches!(
runtime.restore_worker(&detail.worker_ref),
Err(RuntimeError::ConfigBundleDigestMismatch { .. })
));
assert_eq!(backend.config_bundles.lock().unwrap().len(), 1);
}
#[test]
crates/worker-runtime/src/worker_backend.rs
+20 -5
View File
@@ -53,7 +53,7 @@ use worker::feature::builtin::{
#[cfg(feature = "ws-server")]
use worker::ipc::protocol_session::{live_log_entry_event, subscribe_worker_protocol_session};
use worker::{
PromptLoader, SegmentLogSink, WORKER_INPUT_SUBMISSION_EXTENSION_DOMAIN, Worker,
PromptCatalogSource, SegmentLogSink, WORKER_INPUT_SUBMISSION_EXTENSION_DOMAIN, Worker,
WorkerController, WorkerControllerTransport, WorkerError, WorkerFilesystemAuthority,
WorkerHandle, WorkerSharedState, WorkerWorkspaceContext, WorkspaceClient, WorkspaceId,
};
@@ -329,12 +329,12 @@ impl ProfileRuntimeWorkerFactory {
fn restore_fallback_manifest(
worker_name: &str,
) -> Result<(manifest::WorkerManifest, PromptLoader), String> {
) -> Result<(manifest::WorkerManifest, PromptCatalogSource), String> {
let mut config = manifest::WorkerManifestConfig::builtin_defaults();
config.worker.name = Some(worker_name.to_string());
let manifest = manifest::WorkerManifest::try_from(config)
.map_err(|err| format!("failed to build restore fallback manifest: {err}"))?;
Ok((manifest, PromptLoader::builtins_only()))
Ok((manifest, PromptCatalogSource::builtins_only()))
}
async fn resolve_profile_source_archive(
&self,
@@ -566,7 +566,7 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
let archive = self
.resolve_profile_source_archive(&request.request.profile_source)
.await?;
let (manifest, loader) = {
let (manifest, mut loader) = {
let manifest = archive
.resolve_profile(selector, &worker_root, &worker_name)
.map_err(|err| format!("failed to resolve profile source archive: {err}"))?;
@@ -584,6 +584,13 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
)?
}
};
if let Some(prompt_catalog) = request
.config_bundle
.as_ref()
.and_then(|bundle| bundle.prompt_catalog.clone())
{
loader = loader.with_effective_catalog(prompt_catalog);
}
let flow_transition_enabled = manifest.feature.flow.enabled;
let worker_aggregate_dir = self.worker_aggregate_dir(&request.worker_ref)?;
@@ -719,7 +726,14 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
self.worker_mutation_identity.as_ref(),
self.embedded_worker_mutation_dispatcher.as_ref(),
);
let (manifest, loader) = Self::restore_fallback_manifest(&worker_name)?;
let (manifest, mut loader) = Self::restore_fallback_manifest(&worker_name)?;
if let Some(prompt_catalog) = request
.config_bundle
.as_ref()
.and_then(|bundle| bundle.prompt_catalog.clone())
{
loader = loader.with_effective_catalog(prompt_catalog);
}
let worker_aggregate_dir = self.worker_aggregate_dir(&request.worker_ref)?;
let session_dir = worker_aggregate_dir.join("session");
@@ -2090,6 +2104,7 @@ mod tests {
label: Some("adapter-test".to_string()),
}],
declarations: Vec::new(),
prompt_catalog: None,
profile_source_archive: Some(sample_profile_archive()),
profile_source_archive_handle: None,
}