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

Merge commit '7abc6aca45a82665d33115612148536f3a5dd275' into work/T-584-agen-typed-interceptor

Browse Source
This commit is contained in:
Hare
2026-09-03 15:47:21 +09:00
parent 0245980ea5 7abc6aca45
commit e62c7cf4f5
30 changed files with 2979 additions and 434 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/client/src/workspace_product.rs
+23 -45
View File
@@ -1,6 +1,6 @@
use reqwest::Method;
use serde::Serialize;
use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};
use ticket::{
MarkdownText, NewOrchestrationPlanRecord, NewTicket, NewTicketEvent, NewTicketRelation,
OrchestrationPlanKind, OrchestrationPlanRecord, Ticket, TicketBackend, TicketDependencyCheck,
@@ -9,39 +9,17 @@ use ticket::{
TicketRelationKind, TicketRelationView, TicketStateChange, TicketStateSelector, TicketSummary,
};
use workspace_api::{
ListResponse, ObjectiveCreateRequest, ObjectiveDetail, ObjectiveEditRequest,
ObjectiveLinkTicketRequest, ObjectiveStateRequest, ObjectiveSummary,
BrowserCreateWorkerResponse, BrowserWorkspaceOrchestratorResponse,
CreateWorkspaceWorkerRequest, ListResponse, ObjectiveCreateRequest, ObjectiveDetail,
ObjectiveEditRequest, ObjectiveLinkTicketRequest, ObjectiveStateRequest, ObjectiveSummary,
TICKET_ORCHESTRATION_PLANS_QUERY_PATH, TICKET_RELATIONS_QUERY_PATH,
WorkerLaunchOptionsResponse,
};
use crate::{BackendApiClient, BackendWorkspaceClientError};
const DEFAULT_PRODUCT_LIST_LIMIT: usize = 1_000;
#[derive(Debug, Deserialize)]
struct BackendWorkerLaunchOptions {
runtimes: Vec<BackendWorkerLaunchRuntime>,
}
#[derive(Debug, Deserialize)]
struct BackendWorkerLaunchRuntime {
runtime_id: String,
worker_creation_available: bool,
working_directory_required: bool,
}
#[derive(Debug, Deserialize)]
struct BackendCreateWorkerResponse {
runtime_id: String,
worker_id: String,
}
#[derive(Debug, Deserialize)]
struct BackendWorkspaceOrchestratorResponse {
disposition: String,
worker: Option<BackendCreateWorkerResponse>,
}
/// Workspace-scoped Backend client for Ticket and Objective product state.
///
/// Construction requires both the selected Backend URL and Workspace identity.
@@ -267,7 +245,7 @@ impl BackendWorkspaceProductClient {
&self,
ticket_id: &str,
) -> Result<String, BackendWorkspaceClientError> {
let options: BackendWorkerLaunchOptions = self.get_json("/workers/launch-options")?;
let options: WorkerLaunchOptionsResponse = self.get_json("/workers/launch-options")?;
let runtime = options
.runtimes
.iter()
@@ -278,19 +256,19 @@ impl BackendWorkspaceProductClient {
.to_string(),
)
})?;
let response: BackendCreateWorkerResponse = self.send_json(
Method::POST,
"/workers",
Some(&serde_json::json!({
"runtime_id": runtime.runtime_id,
"display_name": format!("intake-{ticket_id}"),
"profile": "builtin:intake",
"initial_submit": [{
"kind": "text",
"content": format!("Please handle intake for Ticket {ticket_id}.")
}]
})),
)?;
let request = CreateWorkspaceWorkerRequest {
runtime_id: runtime.runtime_id.clone(),
display_name: format!("intake-{ticket_id}"),
profile: Some("builtin:intake".to_string()),
ticket_assignment: None,
initial_submit: vec![protocol::Segment::Text {
content: format!("Please handle intake for Ticket {ticket_id}."),
}],
working_directory: None,
control_operation_id: None,
};
let response: BrowserCreateWorkerResponse =
self.send_json(Method::POST, "/workers", Some(&request))?;
Ok(format!(
"Started Intake Worker {}/{} for Ticket {ticket_id}",
response.runtime_id, response.worker_id
@@ -298,7 +276,7 @@ impl BackendWorkspaceProductClient {
}
pub fn start_workspace_orchestrator(&self) -> Result<String, BackendWorkspaceClientError> {
let response: BackendWorkspaceOrchestratorResponse =
let response: BrowserWorkspaceOrchestratorResponse =
self.send_json::<(), _>(Method::POST, "/orchestrator", None)?;
let worker = response.worker.ok_or_else(|| {
BackendWorkspaceClientError::InvalidTarget(
@@ -792,11 +770,11 @@ mod tests {
let (base_url, requests, handle) = response_sequence_server(vec![
(
"200 OK",
r#"{"runtimes":[{"runtime_id":"embedded","worker_creation_available":true,"working_directory_required":false}]}"#,
r#"{"workspace_id":"workspace-a","runtimes":[{"runtime_id":"embedded","display_name":"Embedded","built_in":true,"worker_creation_available":true,"working_directory_required":false,"status":"connected","diagnostics":[]}],"default_profile":null,"profiles":[],"repositories":[],"working_directories":[],"diagnostics":[]}"#,
),
(
"200 OK",
r#"{"runtime_id":"embedded","worker_id":"worker-1"}"#,
r#"{"workspace_id":"workspace-a","runtime_id":"embedded","worker_id":"worker-1","console_href":"/w/workspace-a/workers/worker-1","worker":{"runtime_id":"embedded","worker_id":"worker-1","host_id":"embedded","display_name":"Intake","label":"worker-1","profile":"builtin:intake","singleton_key":null,"tags":[],"workspace":{"visibility":"workspace","identity":"workspace-a","workspace_id":"workspace-a"},"state":"idle","last_seen_at":null,"pinned":false,"retention_state":"active","implementation":{"kind":"runtime","display_hint":"Runtime Worker"},"capabilities":{"can_stop":true,"can_spawn_followup":false},"diagnostics":[]},"diagnostics":[]}"#,
),
]);
let client = BackendWorkspaceProductClient::new_with_access_token(
@@ -824,7 +802,7 @@ mod tests {
#[test]
fn workspace_orchestrator_launch_uses_scoped_backend_route() {
let body = r#"{"disposition":"created","worker":{"runtime_id":"embedded","worker_id":"worker-2"}}"#;
let body = r#"{"workspace_id":"workspace-a","online":true,"disposition":"created","worker":{"runtime_id":"embedded","worker_id":"worker-2","host_id":"embedded","display_name":"Orchestrator","label":"worker-2","profile":"builtin:orchestrator","singleton_key":"workspace-orchestrator","tags":[],"workspace":{"visibility":"workspace","identity":"workspace-a","workspace_id":"workspace-a"},"state":"idle","last_seen_at":null,"pinned":true,"retention_state":"active","implementation":{"kind":"runtime","display_hint":"Runtime Worker"},"capabilities":{"can_stop":true,"can_spawn_followup":false},"diagnostics":[]},"diagnostics":[]}"#;
let (base_url, request, handle) = one_response_server("200 OK", body);
let client = BackendWorkspaceProductClient::new_with_access_token(
base_url,
crates/merge-request/src/lib.rs
+34
View File
@@ -274,6 +274,12 @@ pub struct RegisterReviewerChildSession {
pub reviewer_profile: String,
pub now: DateTime<Utc>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ReviewSubmissionAuthorization {
pub workspace_id: String,
pub subject_ref: String,
}
#[derive(Debug, Clone)]
pub struct SubmitMergeRequestReview {
pub ticket_id: String,
@@ -535,6 +541,34 @@ impl MergeRequestStore {
t.commit()?;
Ok(RequestedMergeRequestReview { request_event: e })
}
pub fn authorize_review_submission(
&self,
ticket_id: &str,
capability_token: &str,
) -> Result<ReviewSubmissionAuthorization, MergeRequestError> {
let connection = self.lock()?;
connection
.query_row(
"SELECT g.workspace_id,g.subject_ref
FROM merge_request_review_grants g
JOIN merge_request_ticket_relations rel
ON rel.workspace_id=g.workspace_id AND rel.merge_request_id=g.merge_request_id
JOIN merge_requests mr
ON mr.workspace_id=g.workspace_id AND mr.merge_request_id=g.merge_request_id
WHERE g.capability_token=?1 AND rel.ticket_id=?2
AND g.status='issued' AND mr.state='open'",
params![capability_token, ticket_id],
|row| {
Ok(ReviewSubmissionAuthorization {
workspace_id: row.get(0)?,
subject_ref: row.get(1)?,
})
},
)
.optional()?
.ok_or_else(|| MergeRequestError::Unauthorized("review grant invalid".into()))
}
pub fn submit_review(
&self,
i: SubmitMergeRequestReview,
crates/merge-request/tests/store.rs
+17
View File
@@ -91,6 +91,23 @@ fn approve(s: &MergeRequestStore, subject: &str, token: &str) -> ReviewEvent {
})
.unwrap()
}
#[test]
fn review_submission_authorization_rejects_invalid_grants_before_side_effects() {
let (_d, store) = fixture();
open(&store);
request(&store, "published-source", "valid-token");
let invalid = store
.authorize_review_submission("T", "invalid-token")
.unwrap_err();
assert!(matches!(invalid, MergeRequestError::Unauthorized(_)));
let authorized = store
.authorize_review_submission("T", "valid-token")
.unwrap();
assert_eq!(authorized.workspace_id, "W");
assert_eq!(authorized.subject_ref, "published-source");
}
#[test]
fn selectors_thread_and_completion_have_no_revision_or_commit_api() {
let (d, s) = fixture();
crates/worker-runtime/src/catalog.rs
+24
View File
@@ -179,6 +179,30 @@ pub struct WorkingDirectoryRequest {
pub materialization: Option<RepositoryMaterializationContext>,
}
/// 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<RepositoryMaterializationContext>,
}
/// 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,
crates/worker-runtime/src/execution.rs
+18
View File
@@ -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<RepositoryRefObservation, WorkingDirectoryDiagnostic> {
Err(WorkingDirectoryDiagnostic::rejected(
"repository_ref_provider_unavailable",
"Worker execution backend does not support Repository ref observation",
))
}
fn list_working_directories(&self) -> Vec<WorkingDirectoryStatus> {
Vec::new()
}
@@ -501,6 +512,13 @@ impl WorkerExecutionBackendRef {
.authorize_working_directory_repository_access(request)
}
pub(crate) fn observe_repository_ref(
&self,
request: &RepositoryRefObservationRequest,
) -> Result<RepositoryRefObservation, WorkingDirectoryDiagnostic> {
self.backend.observe_repository_ref(request)
}
pub(crate) fn list_working_directories(&self) -> Vec<WorkingDirectoryStatus> {
self.backend.list_working_directories()
}
crates/worker-runtime/src/http_server.rs
+85 -17
View File
@@ -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<RuntimeHttpState>,
Extension(auth): Extension<RuntimeAuthContext>,
body: Result<Json<RepositoryRefObservationRequest>, JsonRejection>,
) -> RestResult<crate::catalog::RepositoryRefObservation> {
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<RuntimeHttpState>,
) -> RestResult<RuntimeHttpWorkingDirectoriesResponse> {
@@ -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")
@@ -1941,6 +1970,33 @@ fn status_for_runtime_error(error: &RuntimeError) -> StatusCode {
{
StatusCode::NOT_FOUND
}
RuntimeError::WorkingDirectory(diagnostic)
if matches!(
diagnostic.code.as_str(),
"repository_ref_provider_unavailable"
| "repository_ref_provider_timeout"
| "repository_access_provider_unavailable"
) =>
{
StatusCode::SERVICE_UNAVAILABLE
}
RuntimeError::WorkingDirectory(diagnostic)
if matches!(
diagnostic.code.as_str(),
"repository_ref_provider_auth_failed"
| "repository_access_credential_expired"
| "repository_access_credential_unavailable"
| "repository_access_credential_unauthorized"
| "repository_access_credential_invalid"
) =>
{
StatusCode::FORBIDDEN
}
RuntimeError::WorkingDirectory(diagnostic)
if diagnostic.code == "repository_ref_not_found" =>
{
StatusCode::NOT_FOUND
}
RuntimeError::RuntimeStopped
| RuntimeError::WorkerExecutionUnavailable { .. }
| RuntimeError::ExecutionBackendUnavailable { .. }
@@ -1951,8 +2007,8 @@ fn status_for_runtime_error(error: &RuntimeError) -> StatusCode {
| RuntimeError::InvalidInitialInputKind { .. }
| RuntimeError::ConfigBundleDigestMismatch { .. }
| RuntimeError::InvalidProfileSelector { .. }
| RuntimeError::UnsupportedConfigDeclaration { .. }
| RuntimeError::WorkingDirectory(_) => StatusCode::BAD_REQUEST,
| RuntimeError::UnsupportedConfigDeclaration { .. } => StatusCode::BAD_REQUEST,
RuntimeError::WorkingDirectory(_) => StatusCode::BAD_REQUEST,
RuntimeError::StoreIo { .. }
| RuntimeError::StoreMissing { .. }
| RuntimeError::StoreCorrupt { .. }
@@ -2424,6 +2480,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")
@@ -2934,17 +2994,25 @@ mod tests {
#[test]
fn workdir_runtime_errors_preserve_diagnostic_code() {
let error =
RuntimeError::WorkingDirectory(crate::working_directory::WorkingDirectoryDiagnostic {
code: "working_directory_not_found".to_string(),
message: "working directory missing-workdir was not found".to_string(),
});
assert_eq!(status_for_runtime_error(&error), StatusCode::NOT_FOUND);
assert_eq!(
code_for_runtime_error(&error),
"working_directory_not_found"
);
let cases = [
("working_directory_not_found", StatusCode::NOT_FOUND),
(
"repository_ref_provider_timeout",
StatusCode::SERVICE_UNAVAILABLE,
),
("repository_ref_provider_auth_failed", StatusCode::FORBIDDEN),
("repository_ref_not_found", StatusCode::NOT_FOUND),
];
for (code, expected_status) in cases {
let error = RuntimeError::WorkingDirectory(
crate::working_directory::WorkingDirectoryDiagnostic {
code: code.to_string(),
message: "bounded diagnostic".to_string(),
},
);
assert_eq!(status_for_runtime_error(&error), expected_status);
assert_eq!(code_for_runtime_error(&error), code);
}
}
}
crates/worker-runtime/src/runtime.rs
+91 -17
View File
@@ -1,8 +1,8 @@
use crate::catalog::{
ConfigBundleRef, CreateWorkerRequest, ProfileSelector, WorkerDetail, WorkerLifecycleAck,
WorkerRestoreIntent, WorkerStatus, WorkerSummary, WorkingDirectoryRepositoryAccessRequest,
WorkingDirectoryRequest, WorkingDirectoryStatus as CatalogWorkingDirectoryStatus,
WorkspaceApiRef,
ConfigBundleRef, CreateWorkerRequest, ProfileSelector, RepositoryRefObservation,
RepositoryRefObservationRequest, WorkerDetail, WorkerLifecycleAck, WorkerRestoreIntent,
WorkerStatus, WorkerSummary, WorkingDirectoryRepositoryAccessRequest, WorkingDirectoryRequest,
WorkingDirectoryStatus as CatalogWorkingDirectoryStatus, WorkspaceApiRef,
};
use crate::config_bundle::{
ConfigBundle, ConfigBundleAvailability, ConfigBundleSummary, validate_config_bundle,
@@ -380,6 +380,38 @@ impl Runtime {
self.create_working_directory(request)
}
pub async fn observe_repository_ref_from_resource(
&self,
mut request: RepositoryRefObservationRequest,
) -> Result<RepositoryRefObservation, RuntimeError> {
if let Some(ssh) = request
.materialization
.as_mut()
.and_then(|materialization| materialization.ssh.as_mut())
{
self.resolve_repository_access_resource(ssh).await?;
}
self.observe_repository_ref(request)
}
pub fn observe_repository_ref(
&self,
request: RepositoryRefObservationRequest,
) -> Result<RepositoryRefObservation, RuntimeError> {
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,
@@ -3036,20 +3068,35 @@ fn worker_status_from_run_state(run_state: WorkerExecutionRunState) -> WorkerSta
}
fn repository_resource_error(error: BackendResourceError) -> RuntimeError {
let category = match error {
BackendResourceError::Expired => "expired",
BackendResourceError::Unauthorized { .. } => "unauthorized",
BackendResourceError::UnsupportedKind => "unsupported_kind",
BackendResourceError::MissingResource => "missing_resource",
BackendResourceError::Oversized { .. } => "oversized",
BackendResourceError::DigestMismatch { .. } => "digest_mismatch",
BackendResourceError::ContentTypeMismatch { .. } => "content_type_mismatch",
BackendResourceError::InvalidResponse { .. } => "invalid_response",
BackendResourceError::Transport { .. } => "transport",
let (code, message) = match error {
BackendResourceError::Expired => (
"repository_access_credential_expired",
"Repository access credential lease expired",
),
BackendResourceError::Unauthorized { .. } => (
"repository_access_credential_unauthorized",
"Repository access credential lease was rejected",
),
BackendResourceError::MissingResource => (
"repository_access_credential_unavailable",
"Repository access credential lease is unavailable or already consumed",
),
BackendResourceError::Transport { .. } => (
"repository_access_provider_unavailable",
"Repository access credential provider is unavailable",
),
BackendResourceError::UnsupportedKind
| BackendResourceError::Oversized { .. }
| BackendResourceError::DigestMismatch { .. }
| BackendResourceError::ContentTypeMismatch { .. }
| BackendResourceError::InvalidResponse { .. } => (
"repository_access_credential_invalid",
"Repository access credential response is invalid",
),
};
RuntimeError::InvalidRequest(format!(
"Backend Repository SSH access resource fetch failed: {category}"
))
RuntimeError::WorkingDirectory(
crate::working_directory::WorkingDirectoryDiagnostic::rejected(code, message),
)
}
fn durable_create_worker_request(request: &CreateWorkerRequest) -> CreateWorkerRequest {
@@ -3250,6 +3297,33 @@ mod tests {
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
#[test]
fn repository_resource_failures_keep_typed_credential_diagnostics() {
let cases = [
(
BackendResourceError::Expired,
"repository_access_credential_expired",
),
(
BackendResourceError::MissingResource,
"repository_access_credential_unavailable",
),
(
BackendResourceError::Unauthorized {
message: "denied".to_string(),
},
"repository_access_credential_unauthorized",
),
];
for (error, expected_code) in cases {
let RuntimeError::WorkingDirectory(diagnostic) = repository_resource_error(error)
else {
panic!("Repository resource failure lost its typed diagnostic")
};
assert_eq!(diagnostic.code, expected_code);
}
}
fn internal_worker_ref(
session_id: &str,
parent_session_id: Option<&str>,
crates/worker-runtime/src/worker_backend.rs
+14
View File
@@ -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<RepositoryRefObservation, WorkingDirectoryDiagnostic> {
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<WorkingDirectoryStatus> {
self.working_directory_materializer
.as_ref()
crates/worker-runtime/src/working_directory.rs
+364 -1
View File
@@ -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<RepositoryRefObservation, WorkingDirectoryDiagnostic>;
fn bind_working_directory(
&self,
working_directory_id: &str,
@@ -943,6 +949,72 @@ impl WorkingDirectoryMaterializer for RuntimeGitCacheMaterializer {
self.cache_repository_access(&request.working_directory_id, ssh)
}
fn observe_repository_ref(
&self,
request: &RepositoryRefObservationRequest,
) -> Result<RepositoryRefObservation, WorkingDirectoryDiagnostic> {
let selector = request.selector.trim();
validate_exact_branch_selector(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 +2094,146 @@ fn repository_git_command(
command
}
fn validate_exact_branch_selector(selector: &str) -> Result<(), WorkingDirectoryDiagnostic> {
validate_selector(selector).map_err(|_| {
WorkingDirectoryDiagnostic::new(
"repository_ref_selector_invalid",
"Repository ref observation requires a valid exact branch selector",
)
})?;
if !selector.starts_with("refs/heads/") {
return Err(WorkingDirectoryDiagnostic::new(
"repository_ref_selector_invalid",
"Repository ref observation requires an exact branch selector",
));
}
let status = Command::new("git")
.args(["check-ref-format", selector])
.stdin(Stdio::null())
.stdout(Stdio::null())
.stderr(Stdio::null())
.status()
.map_err(|_| {
WorkingDirectoryDiagnostic::new(
"repository_ref_provider_unavailable",
"Git ref validation could not be started",
)
})?;
if !status.success() {
return Err(WorkingDirectoryDiagnostic::new(
"repository_ref_selector_invalid",
"Repository ref observation requires a valid exact branch selector",
));
}
Ok(())
}
fn read_bounded_command_output(mut reader: impl Read) -> Vec<u8> {
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<String, WorkingDirectoryDiagnostic> {
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 +2705,157 @@ 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_ignores_unpublished_and_stale_workdir_or_cache_refs() {
let seed = create_clean_repo();
let layout = tempfile::tempdir().unwrap();
let provider = layout.path().join("provider.git");
git(
layout.path(),
&[
"clone",
"--bare",
seed.path().to_str().unwrap(),
provider.to_str().unwrap(),
],
);
let cache = layout.path().join("cache");
git(
layout.path(),
&["clone", provider.to_str().unwrap(), cache.to_str().unwrap()],
);
let workdir = layout.path().join("workdir");
git(
layout.path(),
&[
"clone",
provider.to_str().unwrap(),
workdir.to_str().unwrap(),
],
);
git(&workdir, &["config", "user.name", "Yoi Test"]);
git(&workdir, &["config", "user.email", "yoi@example.com"]);
git(&workdir, &["switch", "-c", "published-source"]);
fs::write(workdir.join("source.txt"), "first\n").unwrap();
git(&workdir, &["add", "source.txt"]);
git(&workdir, &["commit", "-m", "source first"]);
let runtime_root = tempfile::tempdir().unwrap();
let materializer = RuntimeGitCacheMaterializer::new(runtime_root.path());
let repository = request(&provider).repository;
let observation_request = RepositoryRefObservationRequest {
repository,
selector: "refs/heads/published-source".to_string(),
materialization: None,
};
assert_eq!(
materializer
.observe_repository_ref(&observation_request)
.unwrap_err()
.code,
"repository_ref_not_found"
);
git(
&workdir,
&["push", "origin", "HEAD:refs/heads/published-source"],
);
let first = materializer
.observe_repository_ref(&observation_request)
.unwrap();
fs::write(workdir.join("source.txt"), "second\n").unwrap();
git(&workdir, &["add", "source.txt"]);
git(&workdir, &["commit", "-m", "source second"]);
let unpublished_second = git_stdout(&workdir, ["rev-parse", "HEAD"]).unwrap();
let still_first = materializer
.observe_repository_ref(&observation_request)
.unwrap();
assert_eq!(still_first.revision_ref, first.revision_ref);
assert_ne!(still_first.revision_ref, unpublished_second);
git(
&workdir,
&["push", "origin", "HEAD:refs/heads/published-source"],
);
let second = materializer
.observe_repository_ref(&observation_request)
.unwrap();
assert_eq!(second.revision_ref, unpublished_second);
assert_ne!(second.revision_ref, first.revision_ref);
assert_ne!(
git_stdout(&cache, ["rev-parse", "HEAD"]).unwrap(),
second.revision_ref
);
}
#[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: repository.clone(),
selector: "HEAD".to_string(),
materialization: None,
})
.unwrap_err();
assert_eq!(non_branch.code, "repository_ref_selector_invalid");
let wildcard = materializer
.observe_repository_ref(&RepositoryRefObservationRequest {
repository,
selector: "refs/heads/release/*".to_string(),
materialization: None,
})
.unwrap_err();
assert_eq!(wildcard.code, "repository_ref_selector_invalid");
}
#[test]
fn local_git_repo_materializes_detached_worktree_under_runtime_root() {
let repo = create_clean_repo();
crates/workspace-api/Cargo.toml
+6 -1
View File
@@ -7,9 +7,10 @@ publish = false
[features]
default = []
typescript = ["dep:ts-rs"]
typescript = ["dep:ts-rs", "protocol/typescript"]
[dependencies]
protocol.workspace = true
serde = { workspace = true, features = ["derive"] }
ts-rs = { version = "12.0.1", optional = true }
@@ -24,6 +25,10 @@ serde_json.workspace = true
name = "generate_workdir_api_types"
required-features = ["typescript"]
[[example]]
name = "generate_worker_launch_api_types"
required-features = ["typescript"]
[[example]]
name = "generate_companion_api_types"
required-features = ["typescript"]
crates/workspace-api/examples/generate_worker_launch_api_types.rs
+3
View File
@@ -0,0 +1,3 @@
fn main() {
print!("{}", workspace_api::worker_launch_api_typescript());
}
crates/workspace-api/src/lib.rs
+320
View File
@@ -579,6 +579,8 @@ pub struct WorkingDirectoryOccupancy {
/// retains the Backend-generated Repository id and is never a Workspace public
/// projection.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[cfg_attr(feature = "typescript", ts(optional_fields = nullable))]
#[serde(deny_unknown_fields)]
pub struct RuntimeWorkingDirectoryCleanupTarget {
pub kind: String,
@@ -590,6 +592,8 @@ pub struct RuntimeWorkingDirectoryCleanupTarget {
/// surfaces must project this through [`WorkingDirectorySummary`] so the UUID is
/// replaced with `repository_key`.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[cfg_attr(feature = "typescript", ts(optional_fields = nullable))]
#[serde(deny_unknown_fields)]
pub struct RuntimeWorkingDirectorySummary {
pub working_directory_id: String,
@@ -607,6 +611,7 @@ pub struct RuntimeWorkingDirectorySummary {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub current_tree: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
#[cfg_attr(feature = "typescript", ts(type = "number | null"))]
pub observed_at_epoch_seconds: Option<u64>,
pub materializer_kind: WorkingDirectoryMaterializerKind,
#[serde(default, skip_serializing_if = "Option::is_none")]
@@ -908,6 +913,7 @@ pub struct RuntimeConnectionTestResponse {
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
pub struct WorkerWorkspaceSummary {
pub visibility: String,
pub identity: String,
@@ -916,12 +922,14 @@ pub struct WorkerWorkspaceSummary {
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
pub struct WorkerImplementationSummary {
pub kind: String,
pub display_hint: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
pub struct WorkerCapabilitySummary {
pub can_stop: bool,
pub can_spawn_followup: bool,
@@ -1095,6 +1103,7 @@ pub struct WorkspaceWorkerDiscoveryPage {
/// do not carry one. The Workspace Server must resolve it from Workspace
/// authority before constructing this response.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
pub struct WorkerSummary {
pub runtime_id: String,
pub worker_id: String,
@@ -1122,6 +1131,142 @@ pub struct WorkerSummary {
pub diagnostics: Vec<Diagnostic>,
}
/// Runtime-owned Worker summary embedded in Worker launch responses.
///
/// This preserves the existing launch wire shape. Workspace-owned Worker list
/// and detail responses use [`WorkerSummary`] instead.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(deny_unknown_fields)]
pub struct WorkerLaunchWorkerSummary {
pub runtime_id: String,
pub worker_id: String,
pub host_id: String,
pub display_name: String,
pub label: String,
pub profile: Option<String>,
pub singleton_key: Option<String>,
pub tags: Vec<String>,
pub workspace: WorkerWorkspaceSummary,
pub state: String,
pub last_seen_at: Option<String>,
pub pinned: bool,
pub retention_state: String,
pub implementation: WorkerImplementationSummary,
pub capabilities: WorkerCapabilitySummary,
#[serde(default, skip_serializing_if = "Option::is_none")]
#[cfg_attr(feature = "typescript", ts(optional = nullable))]
pub working_directory: Option<RuntimeWorkingDirectorySummary>,
#[serde(default)]
pub diagnostics: Vec<Diagnostic>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(deny_unknown_fields)]
pub struct WorkerLaunchOptionsResponse {
pub workspace_id: String,
pub runtimes: Vec<WorkerLaunchRuntimeOption>,
pub default_profile: Option<String>,
pub profiles: Vec<WorkerLaunchProfileCandidate>,
pub repositories: Vec<WorkingDirectoryRepositoryOption>,
pub working_directories: Vec<WorkingDirectorySummary>,
pub diagnostics: Vec<Diagnostic>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(deny_unknown_fields)]
pub struct WorkerLaunchRuntimeOption {
pub runtime_id: String,
pub display_name: String,
pub built_in: bool,
pub worker_creation_available: bool,
pub working_directory_required: bool,
pub status: String,
pub diagnostics: Vec<Diagnostic>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(deny_unknown_fields)]
pub struct WorkerLaunchProfileCandidate {
pub id: String,
pub label: String,
pub description: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(deny_unknown_fields)]
pub struct WorkingDirectoryRepositoryOption {
pub repository_key: String,
#[serde(skip_serializing_if = "Option::is_none")]
#[cfg_attr(feature = "typescript", ts(optional = nullable))]
pub default_selector: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(deny_unknown_fields)]
pub struct BrowserWorkerWorkingDirectorySelection {
pub working_directory_id: String,
#[serde(default)]
pub relative_cwd: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(deny_unknown_fields)]
pub struct CreateWorkspaceWorkerTicketAssignmentRequest {
pub ticket_id: String,
pub operation_id: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(deny_unknown_fields)]
pub struct CreateWorkspaceWorkerRequest {
pub runtime_id: String,
pub display_name: String,
#[serde(default)]
pub profile: Option<String>,
#[serde(default)]
pub ticket_assignment: Option<CreateWorkspaceWorkerTicketAssignmentRequest>,
#[serde(default)]
pub initial_submit: Vec<protocol::Segment>,
#[serde(default)]
pub working_directory: Option<BrowserWorkerWorkingDirectorySelection>,
/// Backend idempotency key used only for authenticated Worker-owned spawn/control.
#[serde(default)]
pub control_operation_id: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(deny_unknown_fields)]
pub struct BrowserCreateWorkerResponse {
pub workspace_id: String,
pub runtime_id: String,
pub worker_id: String,
pub console_href: String,
pub worker: WorkerLaunchWorkerSummary,
pub diagnostics: Vec<Diagnostic>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(deny_unknown_fields)]
pub struct BrowserWorkspaceOrchestratorResponse {
pub workspace_id: String,
pub online: bool,
pub disposition: String,
#[serde(skip_serializing_if = "Option::is_none")]
#[cfg_attr(feature = "typescript", ts(optional = nullable))]
pub worker: Option<WorkerLaunchWorkerSummary>,
pub diagnostics: Vec<Diagnostic>,
}
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum WorkerOperationState {
@@ -1390,6 +1535,74 @@ pub fn workdir_api_typescript() -> String {
)
}
#[cfg(feature = "typescript")]
pub fn worker_launch_api_typescript() -> String {
use ts_rs::TS;
let config = ts_rs::Config::default();
let declarations = [
DiagnosticSeverity::decl(&config),
Diagnostic::decl(&config),
WorkingDirectoryMaterializerKind::decl(&config),
WorkingDirectoryStatusKind::decl(&config),
WorkingDirectoryCleanupTarget::decl(&config),
RuntimeWorkingDirectoryCleanupTarget::decl(&config),
RuntimeWorkingDirectorySummary::decl(&config),
WorkingDirectoryOccupancy::decl(&config),
WorkingDirectorySummary::decl(&config),
WorkerWorkspaceSummary::decl(&config),
WorkerImplementationSummary::decl(&config),
WorkerCapabilitySummary::decl(&config),
WorkerLaunchWorkerSummary::decl(&config),
WorkerLaunchRuntimeOption::decl(&config),
WorkerLaunchProfileCandidate::decl(&config),
WorkingDirectoryRepositoryOption::decl(&config),
WorkerLaunchOptionsResponse::decl(&config),
BrowserWorkerWorkingDirectorySelection::decl(&config),
CreateWorkspaceWorkerTicketAssignmentRequest::decl(&config),
CreateWorkspaceWorkerRequest::decl(&config),
BrowserCreateWorkerResponse::decl(&config),
BrowserWorkspaceOrchestratorResponse::decl(&config),
];
format!(
"// Generated from workspace-api. Do not edit by hand.\n// Regenerate: cargo run -q -p workspace-api --features typescript --example generate_worker_launch_api_types > web/workspace/src/lib/generated/worker-launch-api.ts\n\nimport type {{ Segment }} from \"./protocol\";\n\n{}\n",
declarations
.into_iter()
.map(|declaration| format!("export {declaration}"))
.collect::<Vec<_>>()
.join("\n\n")
)
}
#[cfg(all(test, feature = "typescript"))]
mod worker_launch_typescript_tests {
#[test]
fn generated_worker_launch_api_contract_is_current() {
let expected = super::worker_launch_api_typescript();
let path = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
.join("../../web/workspace/src/lib/generated/worker-launch-api.ts");
let actual = std::fs::read_to_string(&path)
.unwrap_or_else(|error| panic!("failed to read {}: {error}", path.display()));
assert_eq!(
normalize(&actual),
normalize(&expected),
"regenerate Worker launch API TypeScript types with `cargo run -q -p workspace-api --features typescript --example generate_worker_launch_api_types > web/workspace/src/lib/generated/worker-launch-api.ts` and format the generated file",
);
}
fn normalize(value: &str) -> String {
value
.chars()
.filter_map(|character| match character {
character if character.is_whitespace() => None,
',' => Some(';'),
character => Some(character),
})
.collect::<String>()
.replace("=|", "=")
}
}
#[cfg(all(test, feature = "typescript"))]
mod workdir_typescript_tests {
#[test]
@@ -1423,6 +1636,113 @@ mod workdir_typescript_tests {
mod tests {
use super::*;
fn worker_launch_summary() -> WorkerLaunchWorkerSummary {
WorkerLaunchWorkerSummary {
runtime_id: "runtime-a".to_string(),
worker_id: "worker-a".to_string(),
host_id: "host-a".to_string(),
display_name: "Worker A".to_string(),
label: "worker-a".to_string(),
profile: None,
singleton_key: None,
tags: Vec::new(),
workspace: WorkerWorkspaceSummary {
visibility: "workspace".to_string(),
identity: "workspace-a".to_string(),
workspace_id: Some("workspace-a".to_string()),
},
state: "idle".to_string(),
last_seen_at: None,
pinned: false,
retention_state: "active".to_string(),
implementation: WorkerImplementationSummary {
kind: "runtime".to_string(),
display_hint: "Runtime Worker".to_string(),
},
capabilities: WorkerCapabilitySummary {
can_stop: true,
can_spawn_followup: false,
},
working_directory: None,
diagnostics: Vec::new(),
}
}
#[test]
fn worker_launch_optional_omission_and_request_shape_are_stable() {
assert_eq!(
serde_json::to_value(WorkingDirectoryRepositoryOption {
repository_key: "main".to_string(),
default_selector: None,
})
.unwrap(),
serde_json::json!({ "repository_key": "main" })
);
let orchestrator = serde_json::to_value(BrowserWorkspaceOrchestratorResponse {
workspace_id: "workspace-a".to_string(),
online: false,
disposition: "unavailable".to_string(),
worker: None,
diagnostics: Vec::new(),
})
.unwrap();
assert_eq!(
orchestrator,
serde_json::json!({
"workspace_id": "workspace-a",
"online": false,
"disposition": "unavailable",
"diagnostics": [],
})
);
let worker = serde_json::to_value(worker_launch_summary()).unwrap();
assert!(
!worker
.as_object()
.unwrap()
.contains_key("working_directory")
);
assert_eq!(worker["profile"], serde_json::Value::Null);
assert_eq!(worker["singleton_key"], serde_json::Value::Null);
assert_eq!(worker["last_seen_at"], serde_json::Value::Null);
let request = serde_json::to_value(CreateWorkspaceWorkerRequest {
runtime_id: "runtime-a".to_string(),
display_name: "Worker A".to_string(),
profile: None,
ticket_assignment: None,
initial_submit: Vec::new(),
working_directory: None,
control_operation_id: None,
})
.unwrap();
assert_eq!(
request,
serde_json::json!({
"runtime_id": "runtime-a",
"display_name": "Worker A",
"profile": null,
"ticket_assignment": null,
"initial_submit": [],
"working_directory": null,
"control_operation_id": null,
})
);
}
#[test]
fn worker_launch_request_rejects_unknown_fields() {
let error = serde_json::from_value::<CreateWorkspaceWorkerRequest>(serde_json::json!({
"runtime_id": "runtime-a",
"display_name": "Worker A",
"unexpected": true,
}))
.unwrap_err();
assert!(error.to_string().contains("unknown field"));
}
#[test]
fn repository_key_validation_is_canonical_and_bounded() {
let max = "a".repeat(64);
crates/workspace-server/src/hosts.rs
+153 -4
View File
@@ -23,10 +23,10 @@ use worker_runtime::RuntimeWorkspaceScope;
use worker_runtime::auth::{CapabilityTokenSigner, capability_claims};
use worker_runtime::catalog::{
ConfigBundleRef, CreateWorkerRequest, ProfileSelector, ProfileSourceArchiveHttpRef,
ProfileSourceArchiveSource, WorkerDetail as EmbeddedWorkerDetail,
WorkerStatus as EmbeddedWorkerStatus, WorkingDirectoryClaim,
WorkingDirectoryRepositoryAccessRequest, WorkingDirectoryRequest, WorkingDirectoryStatus,
WorkingDirectorySummary, WorkspaceApiRef,
ProfileSourceArchiveSource, RepositoryRefObservation, RepositoryRefObservationRequest,
WorkerDetail as EmbeddedWorkerDetail, WorkerStatus as EmbeddedWorkerStatus,
WorkingDirectoryClaim, WorkingDirectoryRepositoryAccessRequest, WorkingDirectoryRequest,
WorkingDirectoryStatus, WorkingDirectorySummary, WorkspaceApiRef,
};
use worker_runtime::config_bundle::{ConfigBundle, ConfigBundleAvailability, ConfigBundleSummary};
#[cfg(test)]
@@ -830,6 +830,17 @@ pub trait WorkspaceWorkerRuntime: Send + Sync {
))
}
fn observe_repository_ref(
&self,
_request: RepositoryRefObservationRequest,
) -> std::result::Result<RepositoryRefObservation, Error> {
Err(Error::RuntimeOperationFailed {
runtime_id: self.runtime_id().to_string(),
code: "repository_ref_provider_unavailable".to_string(),
message: "Runtime does not support Repository ref observation".to_string(),
})
}
fn list_working_directories(&self) -> RuntimeList<WorkingDirectoryStatus> {
RuntimeList::new(Vec::new(), Vec::new())
}
@@ -1449,6 +1460,31 @@ impl RuntimeRegistry {
})
}
pub fn observe_repository_ref(
&self,
runtime_id: &str,
request: RepositoryRefObservationRequest,
) -> Result<RepositoryRefObservation, RuntimeRegistryError> {
validate_backend_identifier("runtime_id", runtime_id)?;
let runtime = self.runtime(runtime_id)?;
runtime
.observe_repository_ref(request)
.map_err(|error| match error {
Error::RuntimeOperationFailed { code, message, .. } => {
RuntimeRegistryError::RuntimeOperationFailed {
runtime_id: runtime_id.to_string(),
code,
message,
}
}
other => RuntimeRegistryError::RuntimeOperationFailed {
runtime_id: runtime_id.to_string(),
code: "repository_ref_provider_unavailable".to_string(),
message: other.to_string(),
},
})
}
pub fn list_working_directories(
&self,
runtime_id: &str,
@@ -2143,6 +2179,28 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime {
}
}
fn observe_repository_ref(
&self,
request: RepositoryRefObservationRequest,
) -> std::result::Result<RepositoryRefObservation, Error> {
self.runtime
.observe_repository_ref(request)
.map_err(|error| match error {
worker_runtime::error::RuntimeError::WorkingDirectory(diagnostic) => {
Error::RuntimeOperationFailed {
runtime_id: self.runtime_id.clone(),
code: diagnostic.code,
message: diagnostic.message,
}
}
error => Error::RuntimeOperationFailed {
runtime_id: self.runtime_id.clone(),
code: "repository_ref_provider_unavailable".to_string(),
message: error.to_string(),
},
})
}
fn list_working_directories(&self) -> RuntimeList<WorkingDirectoryStatus> {
RuntimeList::new(Vec::new(), Vec::new())
}
@@ -3355,6 +3413,18 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
.map_err(|diagnostic| Error::RegistryInconsistency(diagnostic.message))
}
fn observe_repository_ref(
&self,
request: RepositoryRefObservationRequest,
) -> std::result::Result<RepositoryRefObservation, Error> {
self.post_json::<_, RepositoryRefObservation>("/v1/repository-refs/observe", &request)
.map_err(|diagnostic| Error::RuntimeOperationFailed {
runtime_id: self.runtime_id.clone(),
code: diagnostic.code,
message: diagnostic.message,
})
}
fn list_working_directories(&self) -> RuntimeList<WorkingDirectoryStatus> {
match self.get_json::<RuntimeHttpWorkingDirectoriesResponse>("/v1/working-directories") {
Ok(response) => RuntimeList::new(response.working_directories, Vec::new()),
@@ -4789,6 +4859,85 @@ mod tests {
}
}
struct ObservingExecutionBackend {
response: RepositoryRefObservation,
observed: Arc<Mutex<Vec<RepositoryRefObservationRequest>>>,
}
impl worker_runtime::execution::WorkerExecutionBackend for ObservingExecutionBackend {
fn backend_id(&self) -> &str {
"repository-observation-test-backend"
}
fn spawn_worker(
&self,
_request: worker_runtime::execution::WorkerExecutionSpawnRequest,
) -> worker_runtime::execution::WorkerExecutionSpawnResult {
unreachable!("Repository observation test does not spawn Workers")
}
fn dispatch_input(
&self,
_handle: &worker_runtime::execution::WorkerExecutionHandle,
_input: EmbeddedWorkerInput,
) -> worker_runtime::execution::WorkerExecutionResult {
unreachable!("Repository observation test does not dispatch Worker input")
}
fn observe_repository_ref(
&self,
request: &RepositoryRefObservationRequest,
) -> Result<
RepositoryRefObservation,
worker_runtime::working_directory::WorkingDirectoryDiagnostic,
> {
self.observed.lock().unwrap().push(request.clone());
Ok(self.response.clone())
}
}
#[test]
fn embedded_runtime_forwards_repository_ref_observation_to_execution_backend() {
let observed = Arc::new(Mutex::new(Vec::new()));
let expected = RepositoryRefObservation {
repository_id: "repository-1".to_string(),
source_revision: 7,
source_fingerprint: "sha256:source".to_string(),
selector: "refs/heads/published".to_string(),
revision_ref: "0123456789012345678901234567890123456789".to_string(),
observed_at_epoch_seconds: 42,
};
let runtime = EmbeddedWorkerRuntime::new_memory_with_execution_backend(
"workspace-test",
Arc::new(ObservingExecutionBackend {
response: expected.clone(),
observed: observed.clone(),
}),
)
.unwrap();
let request = RepositoryRefObservationRequest {
repository: worker_runtime::catalog::WorkingDirectoryRepository {
id: "repository-1".to_string(),
provider: "git".to_string(),
source: workspace_api::RepositorySource {
kind: workspace_api::RepositorySourceKind::LocalPath,
uri: "/provider/repository.git".to_string(),
},
source_revision: 7,
source_fingerprint: "sha256:source".to_string(),
selector: None,
},
selector: "refs/heads/published".to_string(),
materialization: None,
};
assert_eq!(
runtime.observe_repository_ref(request.clone()).unwrap(),
expected
);
assert_eq!(observed.lock().unwrap().as_slice(), &[request]);
}
#[derive(Default)]
struct AcceptingExecutionBackend {
contexts:
crates/workspace-server/src/records.rs
+10
View File
@@ -314,12 +314,21 @@ pub struct TicketMergeRequestSummary {
pub review_excerpt: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
pub struct MergeRequestRefDiagnostic {
pub code: String,
pub message: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
pub struct MergeRequestListItem {
pub summary: TicketMergeRequestSummary,
pub ticket_ids: Vec<String>,
pub thread_event_count: usize,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub ref_diagnostics: Vec<MergeRequestRefDiagnostic>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
@@ -470,6 +479,7 @@ pub fn ticket_api_typescript() -> String {
TicketAssignmentPrincipalSummary::decl(&config),
TicketActionEligibility::decl(&config),
TicketMergeRequestSummary::decl(&config),
MergeRequestRefDiagnostic::decl(&config),
MergeRequestListItem::decl(&config),
MergeRequestListResponse::decl(&config),
TicketEvidenceSummary::decl(&config),
crates/workspace-server/src/repositories.rs
+1 -1
View File
@@ -392,7 +392,7 @@ impl RepositoryRegistryReader {
}
}
fn normalize_target_branch_selector(
pub(crate) fn normalize_target_branch_selector(
id: &str,
selector: &str,
) -> Result<String, RepositoryLookupError> {
crates/workspace-server/src/server.rs
+717 -250
View File
File diff suppressed because it is too large Load Diff