From 56798f9fb4bfe18f4fb5986f1a376f253f0cbe8f Mon Sep 17 00:00:00 2001 From: Hare Date: Thu, 3 Sep 2026 12:56:06 +0900 Subject: [PATCH] feat: observe repository refs through runtime providers --- crates/worker-runtime/src/catalog.rs | 24 ++ crates/worker-runtime/src/execution.rs | 18 ++ crates/worker-runtime/src/http_server.rs | 41 ++- crates/worker-runtime/src/runtime.rs | 30 ++- crates/worker-runtime/src/worker_backend.rs | 14 + .../worker-runtime/src/working_directory.rs | 246 +++++++++++++++++- 6 files changed, 366 insertions(+), 7 deletions(-) diff --git a/crates/worker-runtime/src/catalog.rs b/crates/worker-runtime/src/catalog.rs index 2e5d9b9f..14331018 100644 --- a/crates/worker-runtime/src/catalog.rs +++ b/crates/worker-runtime/src/catalog.rs @@ -179,6 +179,30 @@ pub struct WorkingDirectoryRequest { pub materialization: Option, } +/// Backend-authorized request to freshly resolve one Repository provider ref. +/// +/// Runtime executes this against the registered source itself rather than a Workdir +/// or Runtime cache. Secret material is fetched through `materialization` and never +/// appears in the result. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct RepositoryRefObservationRequest { + pub repository: WorkingDirectoryRepository, + pub selector: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub materialization: Option, +} + +/// Provider-neutral proof of one freshly observed Repository ref. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct RepositoryRefObservation { + pub repository_id: String, + pub source_revision: u64, + pub source_fingerprint: String, + pub selector: String, + pub revision_ref: String, + pub observed_at_epoch_seconds: u64, +} + #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct WorkingDirectoryClaim { pub working_directory_id: String, diff --git a/crates/worker-runtime/src/execution.rs b/crates/worker-runtime/src/execution.rs index ff24ecbb..47660977 100644 --- a/crates/worker-runtime/src/execution.rs +++ b/crates/worker-runtime/src/execution.rs @@ -1,4 +1,5 @@ use crate::catalog::{ + RepositoryRefObservation, RepositoryRefObservationRequest, WorkingDirectoryRepositoryAccessRequest, WorkingDirectoryRequest, WorkingDirectoryStatus, }; use crate::config_bundle::ConfigBundle; @@ -333,6 +334,16 @@ pub trait WorkerExecutionBackend: Send + Sync + 'static { )) } + fn observe_repository_ref( + &self, + _request: &RepositoryRefObservationRequest, + ) -> Result { + Err(WorkingDirectoryDiagnostic::rejected( + "repository_ref_provider_unavailable", + "Worker execution backend does not support Repository ref observation", + )) + } + fn list_working_directories(&self) -> Vec { Vec::new() } @@ -501,6 +512,13 @@ impl WorkerExecutionBackendRef { .authorize_working_directory_repository_access(request) } + pub(crate) fn observe_repository_ref( + &self, + request: &RepositoryRefObservationRequest, + ) -> Result { + self.backend.observe_repository_ref(request) + } + pub(crate) fn list_working_directories(&self) -> Vec { self.backend.list_working_directories() } diff --git a/crates/worker-runtime/src/http_server.rs b/crates/worker-runtime/src/http_server.rs index 1e26482e..81246e5d 100644 --- a/crates/worker-runtime/src/http_server.rs +++ b/crates/worker-runtime/src/http_server.rs @@ -11,9 +11,9 @@ use crate::auth::{ verify_capability_token, }; use crate::catalog::{ - ConfigBundleRef, CreateWorkerRequest, WorkerDetail, WorkerLifecycleAck, WorkerSummary, - WorkingDirectoryRepositoryAccessRequest, WorkingDirectoryRequest, WorkingDirectoryStatus, - WorkspaceApiRef, + ConfigBundleRef, CreateWorkerRequest, RepositoryRefObservationRequest, WorkerDetail, + WorkerLifecycleAck, WorkerSummary, WorkingDirectoryRepositoryAccessRequest, + WorkingDirectoryRequest, WorkingDirectoryStatus, WorkspaceApiRef, }; use crate::config_bundle::{ConfigBundle, ConfigBundleAvailability, ConfigBundleSummary}; use crate::error::RuntimeError; @@ -208,6 +208,7 @@ fn runtime_http_router_with_optional_auth( "/v1/working-directories/repository-access", post(authorize_working_directory_repository_access), ) + .route("/v1/repository-refs/observe", post(observe_repository_ref)) .route( "/v1/working-directories/{working_directory_id}/sessions", post(open_workdir_session), @@ -583,6 +584,31 @@ async fn authorize_working_directory_repository_access( })) } +async fn observe_repository_ref( + State(state): State, + Extension(auth): Extension, + body: Result, JsonRejection>, +) -> RestResult { + let Json(request) = body.map_err(RuntimeHttpRestError::json_rejection)?; + if request + .materialization + .as_ref() + .is_some_and(|materialization| materialization.workspace_id != auth.workspace_id) + { + return Err(RuntimeHttpRestError::new( + StatusCode::FORBIDDEN, + "repository_ref_observation_workspace_mismatch", + "Repository ref observation authority does not match the authenticated Workspace", + )); + } + let observation = state + .runtime + .observe_repository_ref_from_resource(request) + .await + .map_err(RuntimeHttpRestError::runtime)?; + Ok(Json(observation)) +} + async fn list_working_directories( State(state): State, ) -> RestResult { @@ -1750,7 +1776,10 @@ fn required_runtime_permission(method: &Method, path: &str) -> Option<&'static s if path == "/v1/workers" && *method == Method::POST { return Some("workers:create"); } - if path == "/v1/working-directories/repository-access" && *method == Method::POST { + if (path == "/v1/working-directories/repository-access" + || path == "/v1/repository-refs/observe") + && *method == Method::POST + { return Some("workdirs:operate"); } if path.starts_with("/v1/workdir-sessions") @@ -2424,6 +2453,10 @@ mod tests { required_runtime_permission(&Method::POST, "/v1/working-directories/repository-access",), Some("workdirs:operate") ); + assert_eq!( + required_runtime_permission(&Method::POST, "/v1/repository-refs/observe"), + Some("workdirs:operate") + ); assert_eq!( required_runtime_permission(&Method::POST, "/v1/working-directories/wd-1/sessions"), Some("workdirs:operate") diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index dad770a2..793f3a37 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -1,6 +1,7 @@ use crate::catalog::{ - ConfigBundleRef, CreateWorkerRequest, ProfileSelector, WorkerDetail, WorkerLifecycleAck, - WorkerStatus, WorkerSummary, WorkingDirectoryRepositoryAccessRequest, WorkingDirectoryRequest, + ConfigBundleRef, CreateWorkerRequest, ProfileSelector, RepositoryRefObservation, + RepositoryRefObservationRequest, WorkerDetail, WorkerLifecycleAck, WorkerStatus, WorkerSummary, + WorkingDirectoryRepositoryAccessRequest, WorkingDirectoryRequest, WorkingDirectoryStatus as CatalogWorkingDirectoryStatus, WorkspaceApiRef, }; use crate::config_bundle::{ @@ -392,6 +393,31 @@ impl Runtime { self.create_working_directory(request) } + pub async fn observe_repository_ref_from_resource( + &self, + mut request: RepositoryRefObservationRequest, + ) -> Result { + if let Some(ssh) = request + .materialization + .as_mut() + .and_then(|materialization| materialization.ssh.as_mut()) + { + self.resolve_repository_access_resource(ssh).await?; + } + let backend = { + let state = self.lock()?; + state.ensure_running()?; + state.execution_backend.clone().ok_or_else(|| { + RuntimeError::ExecutionBackendUnavailable { + message: "Repository ref observation requires an execution backend".to_string(), + } + })? + }; + backend + .observe_repository_ref(&request) + .map_err(RuntimeError::from) + } + pub fn authorize_working_directory_repository_access( &self, request: WorkingDirectoryRepositoryAccessRequest, diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index e9141379..192365e6 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -20,6 +20,7 @@ use crate::auth::{ }; use crate::catalog::{ CreateWorkerRequest, ProfileSourceArchiveHttpRef, ProfileSourceArchiveSource, + RepositoryRefObservation, RepositoryRefObservationRequest, WorkingDirectoryRepositoryAccessRequest, WorkingDirectoryRequest, WorkingDirectoryStatus, }; use crate::execution::{ @@ -1645,6 +1646,19 @@ where materializer.authorize_repository_access(request) } + fn observe_repository_ref( + &self, + request: &RepositoryRefObservationRequest, + ) -> Result { + let materializer = self.working_directory_materializer.as_ref().ok_or_else(|| { + WorkingDirectoryDiagnostic::rejected( + "repository_ref_provider_unavailable", + "Repository ref observation requested, but no materializer is configured for this Runtime backend", + ) + })?; + materializer.observe_repository_ref(request) + } + fn list_working_directories(&self) -> Vec { self.working_directory_materializer .as_ref() diff --git a/crates/worker-runtime/src/working_directory.rs b/crates/worker-runtime/src/working_directory.rs index 06702830..8efe6bbf 100644 --- a/crates/worker-runtime/src/working_directory.rs +++ b/crates/worker-runtime/src/working_directory.rs @@ -1,5 +1,6 @@ use crate::catalog::{ - MaterializerKind, RepositorySshMaterializationAccess, WorkingDirectoryCleanupTarget, + MaterializerKind, RepositoryRefObservation, RepositoryRefObservationRequest, + RepositorySshMaterializationAccess, WorkingDirectoryCleanupTarget, WorkingDirectoryRepositoryAccessRequest, WorkingDirectoryRequest, WorkingDirectoryStatus, WorkingDirectoryStatusKind, WorkingDirectorySummary, }; @@ -196,6 +197,11 @@ pub trait WorkingDirectoryMaterializer: Send + Sync + 'static { request: &WorkingDirectoryRepositoryAccessRequest, ) -> Result<(), WorkingDirectoryDiagnostic>; + fn observe_repository_ref( + &self, + request: &RepositoryRefObservationRequest, + ) -> Result; + fn bind_working_directory( &self, working_directory_id: &str, @@ -943,6 +949,78 @@ impl WorkingDirectoryMaterializer for RuntimeGitCacheMaterializer { self.cache_repository_access(&request.working_directory_id, ssh) } + fn observe_repository_ref( + &self, + request: &RepositoryRefObservationRequest, + ) -> Result { + let selector = request.selector.trim(); + validate_selector(selector)?; + if !selector.starts_with("refs/heads/") { + return Err(WorkingDirectoryDiagnostic::new( + "repository_ref_selector_invalid", + "Repository ref observation requires an exact branch selector", + )); + } + + let working_request = WorkingDirectoryRequest { + repository: request.repository.clone(), + materializer: MaterializerKind::RuntimeGitCache, + backend_workdir_id: None, + materialization: request.materialization.clone(), + }; + Self::validate_request(&working_request)?; + let access = RepositoryCommandAccess::prepare(&self.runtime_root, &working_request)?; + let mut command = repository_git_command(&working_request, access.as_ref()); + command.args([ + "ls-remote", + "--exit-code", + "--refs", + request.repository.source.uri.as_str(), + selector, + ]); + let output = run_repository_git_stdout(command, request.repository.source.kind)?; + let mut lines = output.lines(); + let line = lines.next().ok_or_else(|| { + WorkingDirectoryDiagnostic::new( + "repository_ref_not_found", + "Repository provider did not return the requested ref", + ) + })?; + if lines.next().is_some() { + return Err(WorkingDirectoryDiagnostic::new( + "repository_ref_response_invalid", + "Repository provider returned an ambiguous ref observation", + )); + } + let (revision_ref, observed_selector) = line.split_once('\t').ok_or_else(|| { + WorkingDirectoryDiagnostic::new( + "repository_ref_response_invalid", + "Repository provider returned an invalid ref observation", + ) + })?; + if observed_selector != selector + || !matches!(revision_ref.len(), 40 | 64) + || !revision_ref.bytes().all(|byte| byte.is_ascii_hexdigit()) + { + return Err(WorkingDirectoryDiagnostic::new( + "repository_ref_response_invalid", + "Repository provider returned an invalid ref observation", + )); + } + + Ok(RepositoryRefObservation { + repository_id: request.repository.id.clone(), + source_revision: request.repository.source_revision, + source_fingerprint: request.repository.source_fingerprint.clone(), + selector: selector.to_string(), + revision_ref: revision_ref.to_ascii_lowercase(), + observed_at_epoch_seconds: SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_secs(), + }) + } + fn bind_working_directory( &self, working_directory_id: &str, @@ -2022,6 +2100,112 @@ fn repository_git_command( command } +fn read_bounded_command_output(mut reader: impl Read) -> Vec { + const MAX_CAPTURE_BYTES: usize = 8192; + let mut captured = Vec::new(); + let mut chunk = [0_u8; 4096]; + loop { + match reader.read(&mut chunk) { + Ok(0) | Err(_) => break, + Ok(read) => { + let remaining = MAX_CAPTURE_BYTES.saturating_sub(captured.len()); + captured.extend_from_slice(&chunk[..read.min(remaining)]); + } + } + } + captured +} + +fn run_repository_git_stdout( + mut command: Command, + source_kind: workspace_api::RepositorySourceKind, +) -> Result { + command + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()); + let mut child = command.spawn().map_err(|_| { + WorkingDirectoryDiagnostic::new( + "repository_ref_provider_unavailable", + "Repository provider operation could not be started", + ) + })?; + let stdout = child.stdout.take().ok_or_else(|| { + WorkingDirectoryDiagnostic::new( + "repository_ref_provider_unavailable", + "Repository provider response could not be captured", + ) + })?; + let stderr = child.stderr.take().ok_or_else(|| { + WorkingDirectoryDiagnostic::new( + "repository_ref_provider_unavailable", + "Repository provider diagnostic could not be captured", + ) + })?; + let stdout_reader = std::thread::spawn(move || read_bounded_command_output(stdout)); + let stderr_reader = std::thread::spawn(move || read_bounded_command_output(stderr)); + let started = Instant::now(); + let status = loop { + if let Some(status) = child.try_wait().map_err(|_| { + WorkingDirectoryDiagnostic::new( + "repository_ref_provider_unavailable", + "Repository provider operation status could not be observed", + ) + })? { + break status; + } + if started.elapsed() >= REPOSITORY_COMMAND_TIMEOUT { + let _ = child.kill(); + let _ = child.wait(); + let _ = stdout_reader.join(); + let _ = stderr_reader.join(); + return Err(WorkingDirectoryDiagnostic::new( + "repository_ref_provider_timeout", + "Repository provider operation exceeded the Runtime time limit", + )); + } + std::thread::sleep(Duration::from_millis(25)); + }; + let stdout = stdout_reader.join().unwrap_or_default(); + let stderr = stderr_reader.join().unwrap_or_default(); + if status.success() { + return String::from_utf8(stdout).map_err(|_| { + WorkingDirectoryDiagnostic::new( + "repository_ref_response_invalid", + "Repository provider returned a non-UTF-8 ref observation", + ) + }); + } + if status.code() == Some(2) { + return Err(WorkingDirectoryDiagnostic::new( + "repository_ref_not_found", + "Repository provider did not return the requested ref", + )); + } + let diagnostic = String::from_utf8_lossy(&stderr).to_ascii_lowercase(); + let auth_failed = source_kind.is_remote() + && [ + "authentication failed", + "permission denied", + "could not read username", + "publickey", + ] + .iter() + .any(|marker| diagnostic.contains(marker)); + Err(WorkingDirectoryDiagnostic::new( + if auth_failed { + "repository_ref_provider_auth_failed" + } else { + "repository_ref_provider_unavailable" + }, + if auth_failed { + "Repository provider rejected the operation-scoped authentication" + } else { + "Repository provider operation failed" + }, + )) +} + fn run_repository_git( mut command: Command, code: &'static str, @@ -2493,6 +2677,66 @@ mod tests { WorkerRef::new(WorkerId::from_legacy_u64(sequence)) } + #[test] + fn repository_ref_observation_reads_the_provider_fresh() { + let repo = create_clean_repo(); + git(repo.path(), &["branch", "published"]); + let runtime_root = tempfile::tempdir().unwrap(); + let materializer = RuntimeGitCacheMaterializer::new(runtime_root.path()); + let repository = request(repo.path()).repository; + let observation_request = RepositoryRefObservationRequest { + repository, + selector: "refs/heads/published".to_string(), + materialization: None, + }; + + let first = materializer + .observe_repository_ref(&observation_request) + .unwrap(); + assert_eq!( + first.revision_ref, + git_stdout(repo.path(), ["rev-parse", "published"]).unwrap() + ); + fs::write(repo.path().join("second.txt"), "second\n").unwrap(); + git(repo.path(), &["add", "second.txt"]); + git(repo.path(), &["commit", "-m", "second"]); + git(repo.path(), &["branch", "-f", "published"]); + + let second = materializer + .observe_repository_ref(&observation_request) + .unwrap(); + assert_ne!(first.revision_ref, second.revision_ref); + assert_eq!( + second.revision_ref, + git_stdout(repo.path(), ["rev-parse", "published"]).unwrap() + ); + } + + #[test] + fn repository_ref_observation_rejects_missing_and_non_branch_selectors() { + let repo = create_clean_repo(); + let runtime_root = tempfile::tempdir().unwrap(); + let materializer = RuntimeGitCacheMaterializer::new(runtime_root.path()); + let repository = request(repo.path()).repository; + + let missing = materializer + .observe_repository_ref(&RepositoryRefObservationRequest { + repository: repository.clone(), + selector: "refs/heads/not-published".to_string(), + materialization: None, + }) + .unwrap_err(); + assert_eq!(missing.code, "repository_ref_not_found"); + let non_branch = materializer + .observe_repository_ref(&RepositoryRefObservationRequest { + repository, + selector: "HEAD".to_string(), + materialization: None, + }) + .unwrap_err(); + assert_eq!(non_branch.code, "repository_ref_selector_invalid"); + } + #[test] fn local_git_repo_materializes_detached_worktree_under_runtime_root() { let repo = create_clean_repo();