From ee8ee360ef7c3afe8b05854d3046d830a6d6a3e7 Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 21 Aug 2026 12:35:58 +0900 Subject: [PATCH] fix: share workspace runtime worker contracts --- Cargo.lock | 11 ++ Cargo.toml | 3 + crates/client/Cargo.toml | 1 + crates/client/src/backend_runtime.rs | 108 +---------- crates/client/src/lib.rs | 6 +- crates/tui/src/backend_worker_picker.rs | 3 +- crates/workspace-api/Cargo.toml | 13 ++ crates/workspace-api/src/lib.rs | 195 +++++++++++++++++++ crates/workspace-server/Cargo.toml | 1 + crates/workspace-server/src/hosts.rs | 131 ++++++++++++- crates/workspace-server/src/server.rs | 243 +++++++++++++----------- 11 files changed, 498 insertions(+), 217 deletions(-) create mode 100644 crates/workspace-api/Cargo.toml create mode 100644 crates/workspace-api/src/lib.rs diff --git a/Cargo.lock b/Cargo.lock index ed14f7e5..2dd520bc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -540,6 +540,7 @@ dependencies = [ "tokio-tungstenite 0.29.0", "uuid", "workdir", + "workspace-api", ] [[package]] @@ -6135,6 +6136,15 @@ dependencies = [ "worker", ] +[[package]] +name = "workspace-api" +version = "0.1.0" +dependencies = [ + "serde", + "serde_json", + "workdir", +] + [[package]] name = "writeable" version = "0.6.3" @@ -6265,6 +6275,7 @@ dependencies = [ "workdir", "worker", "worker-runtime", + "workspace-api", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index ea432561..1fec7f77 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -27,6 +27,7 @@ members = [ "crates/ticket", "crates/merge-request", "crates/project-record", + "crates/workspace-api", "crates/workspace-server", "tests/e2e", ] @@ -57,6 +58,7 @@ default-members = [ "crates/ticket", "crates/merge-request", "crates/project-record", + "crates/workspace-api", "crates/workspace-server", ] @@ -78,6 +80,7 @@ ticket = { path = "crates/ticket" } project-record = { path = "crates/project-record" } worker = { path = "crates/worker" } worker-runtime = { path = "crates/worker-runtime" } +workspace-api = { path = "crates/workspace-api" } yoi-plugin-pdk = { path = "crates/plugin-pdk" } yoi = { path = "crates/yoi" } protocol = { path = "crates/protocol" } diff --git a/crates/client/Cargo.toml b/crates/client/Cargo.toml index 08e075ee..b92e79ab 100644 --- a/crates/client/Cargo.toml +++ b/crates/client/Cargo.toml @@ -16,6 +16,7 @@ thiserror = { workspace = true } tokio = { workspace = true, features = ["rt", "macros", "net", "io-util", "sync", "time", "process", "fs"] } tokio-tungstenite = { workspace = true } uuid = { workspace = true } +workspace-api.workspace = true workdir = { workspace = true } [dev-dependencies] diff --git a/crates/client/src/backend_runtime.rs b/crates/client/src/backend_runtime.rs index 3661b20e..9ec12f6c 100644 --- a/crates/client/src/backend_runtime.rs +++ b/crates/client/src/backend_runtime.rs @@ -1,13 +1,21 @@ use futures::{SinkExt, StreamExt}; use protocol::stream::{decode_event, encode_method}; use protocol::{ErrorCode, Event, Method}; -use serde::Deserialize; use std::collections::VecDeque; use std::fmt; use tokio::sync::mpsc; use tokio_tungstenite::connect_async; use tokio_tungstenite::tungstenite::Message as TungsteniteMessage; pub use workdir::workspace::WorkingDirectorySummary as BackendWorkingDirectorySummary; +pub use workspace_api::{ + Diagnostic as BackendDiagnostic, DiagnosticSeverity as BackendDiagnosticSeverity, + ListResponse as BackendRuntimeListResponse, RuntimeSummary as BackendRuntimeSummary, + WorkerCapabilitySummary as BackendWorkerCapabilitySummary, + WorkerImplementationSummary as BackendWorkerImplementationSummary, + WorkerRestoreResponse as BackendWorkerRestoreResponse, + WorkerRestoreResult as BackendWorkerRestoreResult, WorkerSummary as BackendWorkerSummary, + WorkerWorkspaceSummary as BackendWorkerWorkspaceSummary, +}; #[derive(Debug, Clone, PartialEq, Eq)] pub struct BackendRuntimeTarget { @@ -93,94 +101,6 @@ impl BackendRuntimeListTarget { } } -#[derive(Debug, Clone, Deserialize)] -pub struct BackendRuntimeListResponse { - pub workspace_id: String, - pub limit: usize, - pub items: Vec, - pub source: String, - #[serde(default)] - pub diagnostics: Vec, -} - -#[derive(Debug, Clone, Deserialize, PartialEq, Eq)] -pub struct BackendRuntimeSummary { - pub runtime_id: String, - pub label: String, - pub kind: String, - pub status: String, - #[serde(default)] - pub host_ids: Vec, - #[serde(default)] - pub diagnostics: Vec, -} - -#[derive(Debug, Clone, Deserialize, PartialEq, Eq)] -pub struct BackendWorkerWorkspaceSummary { - pub visibility: String, - pub identity: String, -} - -#[derive(Debug, Clone, Deserialize, PartialEq, Eq)] -pub struct BackendWorkerImplementationSummary { - pub kind: String, - pub display_hint: String, -} - -#[derive(Debug, Clone, Deserialize, PartialEq, Eq)] -pub struct BackendWorkerCapabilitySummary { - pub can_stop: bool, - pub can_spawn_followup: bool, -} - -#[derive(Debug, Clone, Deserialize, PartialEq, Eq)] -pub struct BackendWorkerSummary { - pub runtime_id: String, - pub worker_id: String, - pub resource_key: String, - pub host_id: String, - #[serde(default)] - pub display_name: String, - pub label: String, - #[serde(default)] - pub profile: Option, - #[serde(default)] - pub singleton_key: Option, - #[serde(default)] - pub tags: Vec, - pub workspace: BackendWorkerWorkspaceSummary, - pub state: String, - #[serde(default)] - pub last_seen_at: Option, - #[serde(default)] - pub pinned: bool, - #[serde(default)] - pub retention_state: String, - pub implementation: BackendWorkerImplementationSummary, - pub capabilities: BackendWorkerCapabilitySummary, - #[serde(default)] - pub working_directory: Option, - #[serde(default)] - pub diagnostics: Vec, -} - -#[derive(Debug, Clone, Deserialize, PartialEq, Eq)] -pub struct BackendWorkerRestoreResult { - pub state: String, - #[serde(default)] - pub worker: Option, - #[serde(default)] - pub diagnostics: Vec, -} - -#[derive(Debug, Clone, Deserialize, PartialEq, Eq)] -pub struct BackendWorkerRestoreResponse { - pub workspace_id: String, - pub runtime_id: String, - pub worker_id: String, - pub result: BackendWorkerRestoreResult, -} - #[derive(Debug)] pub struct BackendRuntimeClient { target: BackendRuntimeTarget, @@ -277,7 +197,7 @@ pub async fn list_backend_workers( } Err(error) => diagnostics.push(BackendDiagnostic { code: "runtime_worker_list_failed".to_string(), - severity: Some("error".to_string()), + severity: BackendDiagnosticSeverity::Error, message: format!( "failed to list workers for runtime {}: {error}", runtime.runtime_id @@ -619,14 +539,6 @@ fn percent_encode(input: &str, keep: impl Fn(u8) -> bool) -> String { encoded } -#[derive(Debug, Clone, Deserialize, PartialEq, Eq)] -pub struct BackendDiagnostic { - pub code: String, - #[serde(default)] - pub severity: Option, - pub message: String, -} - #[cfg(test)] mod tests { use super::*; diff --git a/crates/client/src/lib.rs b/crates/client/src/lib.rs index 45d51a90..c2efc1bd 100644 --- a/crates/client/src/lib.rs +++ b/crates/client/src/lib.rs @@ -22,9 +22,9 @@ pub use backend_auth::{ poll_device_login, start_device_login, wait_for_device_login, }; pub use backend_runtime::{ - BackendDiagnostic, BackendRuntimeClient, BackendRuntimeClientError, BackendRuntimeListResponse, - BackendRuntimeListTarget, BackendRuntimeSummary, BackendRuntimeTarget, - BackendWorkerCapabilitySummary, BackendWorkerImplementationSummary, + BackendDiagnostic, BackendDiagnosticSeverity, BackendRuntimeClient, BackendRuntimeClientError, + BackendRuntimeListResponse, BackendRuntimeListTarget, BackendRuntimeSummary, + BackendRuntimeTarget, BackendWorkerCapabilitySummary, BackendWorkerImplementationSummary, BackendWorkerRestoreResponse, BackendWorkerRestoreResult, BackendWorkerSummary, BackendWorkerWorkspaceSummary, BackendWorkingDirectorySummary, list_backend_stopped_workers, list_backend_workers, restore_backend_worker, diff --git a/crates/tui/src/backend_worker_picker.rs b/crates/tui/src/backend_worker_picker.rs index 2683336f..46e3f28a 100644 --- a/crates/tui/src/backend_worker_picker.rs +++ b/crates/tui/src/backend_worker_picker.rs @@ -46,7 +46,7 @@ pub(crate) async fn run( } Err(error) => response.diagnostics.push(client::BackendDiagnostic { code: "backend_stopped_workers_list_failed".to_string(), - severity: Some("error".to_string()), + severity: client::BackendDiagnosticSeverity::Error, message: error.to_string(), }), } @@ -396,6 +396,7 @@ mod tests { workspace: BackendWorkerWorkspaceSummary { visibility: "workspace".to_string(), identity: "ws".to_string(), + workspace_id: Some("ws".to_string()), }, state: "running".to_string(), last_seen_at: None, diff --git a/crates/workspace-api/Cargo.toml b/crates/workspace-api/Cargo.toml new file mode 100644 index 00000000..f5c05977 --- /dev/null +++ b/crates/workspace-api/Cargo.toml @@ -0,0 +1,13 @@ +[package] +name = "workspace-api" +version = "0.1.0" +edition.workspace = true +license.workspace = true +publish = false + +[dependencies] +serde = { workspace = true, features = ["derive"] } +workdir.workspace = true + +[dev-dependencies] +serde_json.workspace = true diff --git a/crates/workspace-api/src/lib.rs b/crates/workspace-api/src/lib.rs new file mode 100644 index 00000000..a24cd814 --- /dev/null +++ b/crates/workspace-api/src/lib.rs @@ -0,0 +1,195 @@ +//! Shared Workspace HTTP resource contracts. +//! +//! This crate owns transport DTOs exposed by the Workspace Server and consumed +//! by Rust clients. Runtime-internal projections remain in their owning crates; +//! callers must explicitly construct these Workspace-authoritative resources. + +use serde::{Deserialize, Serialize}; +use workdir::workspace::WorkingDirectorySummary; + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum DiagnosticSeverity { + Info, + Warning, + Error, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct Diagnostic { + pub code: String, + pub severity: DiagnosticSeverity, + pub message: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct ListResponse { + pub workspace_id: String, + pub limit: usize, + pub items: Vec, + pub source: String, + #[serde(default)] + pub diagnostics: Vec, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum RuntimeSourceKind { + EmbeddedWorkerRuntime, + RemoteHttp, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum RuntimeSourceStatus { + Active, + Reserved, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum RuntimeIdentityAuthority { + RuntimeRegistryProjection, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct RuntimeSourceSummary { + pub kind: RuntimeSourceKind, + pub status: RuntimeSourceStatus, + pub identity_authority: RuntimeIdentityAuthority, + pub note: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct RuntimeCapabilitySummary { + pub can_list_hosts: bool, + pub can_list_workers: bool, + pub can_get_worker: bool, + pub can_spawn_worker: bool, + pub can_stop_worker: bool, + pub has_workspace_fs: bool, + pub has_shell: bool, + pub has_git: bool, + pub supports_worktrees: bool, + pub supports_backend_internal_tools: bool, + pub workspace_scope: String, + pub max_workers: usize, + pub os: String, + pub arch: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct RuntimeSummary { + pub runtime_id: String, + pub label: String, + pub kind: String, + pub status: String, + pub source: RuntimeSourceSummary, + #[serde(default)] + pub host_ids: Vec, + pub capabilities: RuntimeCapabilitySummary, + #[serde(default)] + pub diagnostics: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct WorkerWorkspaceSummary { + pub visibility: String, + pub identity: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub workspace_id: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct WorkerImplementationSummary { + pub kind: String, + pub display_hint: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct WorkerCapabilitySummary { + pub can_stop: bool, + pub can_spawn_followup: bool, +} + +/// Workspace-authoritative Worker projection. +/// +/// `resource_key` is required here even though Runtime-internal Worker summaries +/// do not carry one. The Workspace Server must resolve it from Workspace +/// authority before constructing this response. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct WorkerSummary { + pub runtime_id: String, + pub worker_id: String, + pub resource_key: String, + pub host_id: String, + #[serde(default)] + pub display_name: String, + pub label: String, + pub profile: Option, + pub singleton_key: Option, + #[serde(default)] + pub tags: Vec, + pub workspace: WorkerWorkspaceSummary, + pub state: String, + pub last_seen_at: Option, + #[serde(default)] + pub pinned: bool, + #[serde(default)] + pub retention_state: String, + pub implementation: WorkerImplementationSummary, + pub capabilities: WorkerCapabilitySummary, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub working_directory: Option, + #[serde(default)] + pub diagnostics: Vec, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum WorkerOperationState { + Accepted, + Unsupported, + Rejected, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct WorkerRestoreResult { + pub state: WorkerOperationState, + pub worker: Option, + #[serde(default)] + pub diagnostics: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct WorkerRestoreResponse { + pub workspace_id: String, + pub runtime_id: String, + pub worker_id: String, + pub result: WorkerRestoreResult, +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn worker_resource_key_is_required() { + let payload = serde_json::json!({ + "runtime_id": "arcadia", + "worker_id": "worker-1", + "host_id": "host", + "display_name": "Coder", + "label": "Coder", + "workspace": { + "visibility": "workspace", + "identity": "workspace-test" + }, + "state": "idle", + "implementation": {"kind": "worker", "display_hint": "Coder"}, + "capabilities": {"can_stop": true, "can_spawn_followup": false} + }); + + assert!(serde_json::from_value::(payload).is_err()); + } +} diff --git a/crates/workspace-server/Cargo.toml b/crates/workspace-server/Cargo.toml index 73e775fb..dd870795 100644 --- a/crates/workspace-server/Cargo.toml +++ b/crates/workspace-server/Cargo.toml @@ -38,6 +38,7 @@ tokio = { workspace = true, features = ["fs", "macros", "net", "rt-multi-thread" tower.workspace = true tokio-tungstenite.workspace = true worker.workspace = true +workspace-api.workspace = true workdir = { workspace = true, features = ["http-client"] } worker-runtime.workspace = true toml.workspace = true diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index 70a81068..29b610ba 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -246,8 +246,6 @@ pub struct WorkerCapabilitySummary { pub struct WorkerSummary { #[serde(flatten)] pub worker: RuntimeWorkerRef, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub resource_key: Option, pub host_id: String, /// Human-readable display name. This is not identity and may be duplicated. pub display_name: String, @@ -271,6 +269,119 @@ pub struct WorkerSummary { pub diagnostics: Vec, } +impl From for workspace_api::Diagnostic { + fn from(diagnostic: RuntimeDiagnostic) -> Self { + let severity = match diagnostic.severity { + DiagnosticSeverity::Info => workspace_api::DiagnosticSeverity::Info, + DiagnosticSeverity::Warning => workspace_api::DiagnosticSeverity::Warning, + DiagnosticSeverity::Error => workspace_api::DiagnosticSeverity::Error, + }; + Self { + code: diagnostic.code, + severity, + message: diagnostic.message, + } + } +} + +impl From for workspace_api::RuntimeSourceSummary { + fn from(source: RuntimeSourceSummary) -> Self { + let kind = match source.kind { + RuntimeSourceKind::EmbeddedWorkerRuntime => { + workspace_api::RuntimeSourceKind::EmbeddedWorkerRuntime + } + RuntimeSourceKind::RemoteHttp => workspace_api::RuntimeSourceKind::RemoteHttp, + }; + let status = match source.status { + RuntimeSourceStatus::Active => workspace_api::RuntimeSourceStatus::Active, + RuntimeSourceStatus::Reserved => workspace_api::RuntimeSourceStatus::Reserved, + }; + let identity_authority = match source.identity_authority { + RuntimeIdentityAuthority::RuntimeRegistryProjection => { + workspace_api::RuntimeIdentityAuthority::RuntimeRegistryProjection + } + }; + Self { + kind, + status, + identity_authority, + note: source.note, + } + } +} + +impl From for workspace_api::RuntimeCapabilitySummary { + fn from(capabilities: RuntimeCapabilitySummary) -> Self { + Self { + can_list_hosts: capabilities.can_list_hosts, + can_list_workers: capabilities.can_list_workers, + can_get_worker: capabilities.can_get_worker, + can_spawn_worker: capabilities.can_spawn_worker, + can_stop_worker: capabilities.can_stop_worker, + has_workspace_fs: capabilities.has_workspace_fs, + has_shell: capabilities.has_shell, + has_git: capabilities.has_git, + supports_worktrees: capabilities.supports_worktrees, + supports_backend_internal_tools: capabilities.supports_backend_internal_tools, + workspace_scope: capabilities.workspace_scope, + max_workers: capabilities.max_workers, + os: capabilities.os, + arch: capabilities.arch, + } + } +} + +impl From for workspace_api::RuntimeSummary { + fn from(runtime: RuntimeSummary) -> Self { + Self { + runtime_id: runtime.runtime_id, + label: runtime.label, + kind: runtime.kind, + status: runtime.status, + source: runtime.source.into(), + host_ids: runtime.host_ids, + capabilities: runtime.capabilities.into(), + diagnostics: runtime.diagnostics.into_iter().map(Into::into).collect(), + } + } +} + +pub(crate) fn workspace_worker_summary( + summary: WorkerSummary, + resource_key: String, +) -> workspace_api::WorkerSummary { + workspace_api::WorkerSummary { + runtime_id: summary.worker.runtime_id, + worker_id: summary.worker.worker_id, + resource_key, + host_id: summary.host_id, + display_name: summary.display_name, + label: summary.label, + profile: summary.profile, + singleton_key: summary.singleton_key, + tags: summary.tags, + workspace: workspace_api::WorkerWorkspaceSummary { + visibility: summary.workspace.visibility, + identity: summary.workspace.identity, + workspace_id: summary.workspace.workspace_id, + }, + state: summary.state, + last_seen_at: summary.last_seen_at, + pinned: summary.pinned, + retention_state: summary.retention_state, + implementation: workspace_api::WorkerImplementationSummary { + kind: summary.implementation.kind, + display_hint: summary.implementation.display_hint, + }, + capabilities: workspace_api::WorkerCapabilitySummary { + can_stop: summary.capabilities.can_stop, + can_spawn_followup: summary.capabilities.can_spawn_followup, + }, + working_directory: summary.working_directory, + diagnostics: summary.diagnostics.into_iter().map(Into::into).collect(), + } +} + #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub struct WorkerRestoreResult { pub state: WorkerOperationState, @@ -509,6 +620,16 @@ pub enum WorkerOperationState { Rejected, } +impl From for workspace_api::WorkerOperationState { + fn from(state: WorkerOperationState) -> Self { + match state { + WorkerOperationState::Accepted => Self::Accepted, + WorkerOperationState::Unsupported => Self::Unsupported, + WorkerOperationState::Rejected => Self::Rejected, + } + } +} + #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub struct WorkerSpawnAcceptanceEvidence { pub kind: String, @@ -1680,7 +1801,6 @@ impl EmbeddedWorkerRuntime { ); WorkerSummary { worker: RuntimeWorkerRef::new(&self.runtime_id, worker_id.clone()), - resource_key: None, host_id: self.host_id.clone(), display_name: display.display_name.clone(), label: display.display_name, @@ -1720,7 +1840,6 @@ impl EmbeddedWorkerRuntime { ); WorkerSummary { worker: RuntimeWorkerRef::new(&self.runtime_id, worker_id.clone()), - resource_key: None, host_id: self.host_id.clone(), display_name: display.display_name.clone(), label: display.display_name, @@ -2806,7 +2925,6 @@ impl RemoteWorkerRuntime { ); WorkerSummary { worker: RuntimeWorkerRef::new(&self.runtime_id, worker_id.clone()), - resource_key: None, host_id: self.host_id.clone(), display_name: display.display_name.clone(), label: display.display_name, @@ -2850,7 +2968,6 @@ impl RemoteWorkerRuntime { ); WorkerSummary { worker: RuntimeWorkerRef::new(&self.runtime_id, worker_id.clone()), - resource_key: None, host_id: self.host_id.clone(), display_name: display.display_name.clone(), label: display.display_name, @@ -4222,7 +4339,6 @@ pub fn placeholder_worker(host_id: impl Into) -> WorkerSummary { let host_id = host_id.into(); WorkerSummary { worker: RuntimeWorkerRef::new("placeholder", "worker-placeholder"), - resource_key: None, host_id, display_name: "Worker runtime actions are not implemented".to_string(), label: "Worker runtime actions are not implemented".to_string(), @@ -4616,7 +4732,6 @@ mod tests { host_id: host_id.to_string(), workers: vec![WorkerSummary { worker: RuntimeWorkerRef::new(runtime_id, worker_id), - resource_key: None, host_id: host_id.to_string(), display_name: label.to_string(), label: label.to_string(), diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index e3046592..47c0aa6e 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -77,13 +77,13 @@ use crate::hosts::{ ConfigBundleCheckResult, ConfigBundleSyncResult, DiagnosticSeverity, EMBEDDED_RUNTIME_ID, EmbeddedWorkerRuntime, HostSummary, RemoteRuntimeConfig, RemoteWorkerRuntime, RuntimeDiagnostic, RuntimeRegistry, RuntimeRegistryError, RuntimeRegistryUnregisterResult, - RuntimeSummary, TicketWorkerRole, WorkerCapabilitySummary, WorkerCompletionsRequest, - WorkerCompletionsResult, WorkerControlOperation, WorkerCreateBinding, - WorkerImplementationSummary, WorkerInputKind, WorkerInputRequest, WorkerInputResult, - WorkerLifecycleRequest, WorkerLifecycleResult, WorkerOperationState, WorkerRestoreResult, - WorkerSpawnAcceptanceRequirement, WorkerSpawnIntent, WorkerSpawnRequest, WorkerSpawnResult, - WorkerSpawnWorkingDirectoryRequest, WorkerSummary, WorkerTicketAssignmentRequest, - WorkerWorkspaceSummary, worker_spawn_create_fingerprint, + TicketWorkerRole, WorkerCapabilitySummary, WorkerCompletionsRequest, WorkerCompletionsResult, + WorkerControlOperation, WorkerCreateBinding, WorkerImplementationSummary, WorkerInputKind, + WorkerInputRequest, WorkerInputResult, WorkerLifecycleRequest, WorkerLifecycleResult, + WorkerOperationState, WorkerRestoreResult, WorkerSpawnAcceptanceRequirement, WorkerSpawnIntent, + WorkerSpawnRequest, WorkerSpawnResult, WorkerSpawnWorkingDirectoryRequest, WorkerSummary, + WorkerTicketAssignmentRequest, WorkerWorkspaceSummary, worker_spawn_create_fingerprint, + workspace_worker_summary, }; use crate::identity::WorkspaceIdentity; use crate::memory_backend::execute_memory_backend_operation_with_authority; @@ -2264,14 +2264,6 @@ enum RuntimeWorkersStatusFilter { Stopped, } -#[derive(Debug, Serialize, Deserialize)] -pub struct WorkerRestoreResponse { - pub workspace_id: String, - #[serde(flatten)] - pub worker_ref: RuntimeWorkerRef, - pub result: WorkerRestoreResult, -} - #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] #[serde(rename_all = "snake_case")] pub enum CleanupTargetKind { @@ -6534,7 +6526,7 @@ async fn scoped_get_profile_source_archive( async fn scoped_list_runtimes( State(api): State, AxumPath(path): AxumPath, -) -> ApiResult>> { +) -> ApiResult>> { validate_workspace_scope(&api, &path.workspace_id)?; list_runtimes(State(api)).await } @@ -6722,7 +6714,7 @@ async fn scoped_worker_remove_source_boundary( async fn scoped_get_workspace_worker( State(api): State, AxumPath(path): AxumPath, -) -> ApiResult> { +) -> ApiResult> { validate_workspace_scope(&api, &path.workspace_id)?; let worker_id = api .store @@ -6738,7 +6730,7 @@ async fn scoped_get_workspace_worker( workers .items .into_iter() - .find(|worker| worker.worker.worker_id == worker_id) + .find(|worker| worker.worker_id == worker_id) .map(Json) .ok_or_else(|| { Error::UnknownWorker { @@ -6751,7 +6743,7 @@ async fn scoped_get_workspace_worker( async fn scoped_list_workers( State(api): State, AxumPath(path): AxumPath, -) -> ApiResult>> { +) -> ApiResult>> { validate_workspace_scope(&api, &path.workspace_id)?; list_workers(State(api)).await } @@ -7010,7 +7002,7 @@ async fn restore_known_worker( State(api): State, AxumPath(path): AxumPath, headers: HeaderMap, -) -> ApiResult> { +) -> ApiResult> { validate_workspace_scope(&api, &path.workspace_id)?; let source = authenticate_worker_mutation_source(&api, &path.workspace_id, &headers)?; let subject = path.worker.clone(); @@ -7590,7 +7582,7 @@ fn build_runtime_cleanup_plan( .items .iter() .filter(|worker| worker.state == "running") - .map(|worker| worker.worker.clone()) + .map(|worker| RuntimeWorkerRef::new(&worker.runtime_id, &worker.worker_id)) .collect(); let (workdir_summaries, mut diagnostics) = match runtime_working_directory_summaries(api, runtime_id) { @@ -8107,7 +8099,7 @@ async fn scoped_list_runtime_workers( State(api): State, AxumPath(path): AxumPath, Query(query): Query, -) -> ApiResult>> { +) -> ApiResult>> { validate_workspace_scope(&api, &path.workspace_id)?; list_runtime_workers(State(api), AxumPath(path.runtime_id), Query(query)).await } @@ -8166,7 +8158,7 @@ async fn scoped_restore_runtime_worker( State(api): State, AxumPath(path): AxumPath, Query(query): Query, -) -> ApiResult> { +) -> ApiResult> { validate_workspace_scope(&api, &path.workspace_id)?; let workspace_id = path.workspace_id.clone(); let runtime_id = path.worker.runtime_id.clone(); @@ -8219,11 +8211,13 @@ async fn scoped_restore_runtime_worker( .into()); } assign_ticket_worker_from_lifecycle(&api, assignment, &runtime_id, &worker_id)?; - return Ok(Json(WorkerRestoreResponse { + let worker = project_workspace_worker(&api, worker)?; + return Ok(Json(workspace_api::WorkerRestoreResponse { workspace_id, - worker_ref: RuntimeWorkerRef::new(&runtime_id, &worker_id), - result: crate::hosts::WorkerRestoreResult { - state: WorkerOperationState::Accepted, + runtime_id: runtime_id.clone(), + worker_id: worker_id.clone(), + result: workspace_api::WorkerRestoreResult { + state: workspace_api::WorkerOperationState::Accepted, worker: Some(worker), diagnostics: Vec::new(), }, @@ -8352,7 +8346,7 @@ async fn scoped_worker_protocol_ws( async fn scoped_list_host_workers( State(api): State, AxumPath(path): AxumPath, -) -> ApiResult>> { +) -> ApiResult>> { validate_workspace_scope(&api, &path.workspace_id)?; list_host_workers(State(api), AxumPath(path.host_id)).await } @@ -9304,21 +9298,21 @@ async fn list_hosts( async fn list_runtimes( State(api): State, -) -> ApiResult>> { +) -> ApiResult>> { let limit = api.config.max_records.min(200); let runtimes = api.runtime.list_runtimes(limit); - Ok(Json(RuntimeListResponse { + Ok(Json(workspace_api::ListResponse { workspace_id: api.config.workspace_id, limit, - items: runtimes.items, + items: runtimes.items.into_iter().map(Into::into).collect(), source: "worker_runtime_registry".to_string(), - diagnostics: runtimes.diagnostics, + diagnostics: runtimes.diagnostics.into_iter().map(Into::into).collect(), })) } async fn list_workers( State(api): State, -) -> ApiResult>> { +) -> ApiResult>> { workers_response(api).map(Json) } @@ -10025,7 +10019,7 @@ async fn post_companion_cancel( #[derive(Debug, Serialize)] struct WorkerShowProjection { #[serde(flatten)] - worker: WorkerSummary, + worker: workspace_api::WorkerSummary, updated_at: String, } @@ -10071,31 +10065,18 @@ async fn get_runtime_worker( .store .list_workdir_registry(&api.config.workspace_id, 500)?; let updated_at = record.updated_at.clone(); - let mut worker = merge_worker_registry_projection(Some(&worker), &record, links, &workdirs); - worker.resource_key = Some( - api.store - .resource_key( - &api.config.workspace_id, - WorkspaceResourceKind::Worker, - &worker_ref.worker_id, - )? - .ok_or_else(|| { - Error::Store(format!( - "Workspace Worker `{}` has no resource key", - worker_ref.worker_id - )) - })?, - ); + let worker = merge_worker_registry_projection(Some(&worker), &record, links, &workdirs); + let worker = project_workspace_worker(&api, worker)?; Ok(Json(WorkerShowProjection { worker, updated_at })) } async fn restore_runtime_worker( State(api): State, AxumPath((runtime_id, worker_id)): AxumPath<(String, String)>, -) -> ApiResult> { +) -> ApiResult> { let worker = resolve_workspace_worker_reference(&api, &runtime_id, &worker_id)?; - let mut result = api.restore_workspace_worker(&worker)?; - if let Some(worker) = result.worker.as_ref() { + let result = api.restore_workspace_worker(&worker)?; + let projected_worker = if let Some(worker) = result.worker.as_ref() { let record = sync_worker_observation(&api, worker)?; let links = api .store @@ -10103,27 +10084,20 @@ async fn restore_runtime_worker( let workdirs = api .store .list_workdir_registry(&api.config.workspace_id, 500)?; - let mut summary = merge_worker_registry_projection(Some(worker), &record, links, &workdirs); - summary.resource_key = Some( - api.store - .resource_key( - &api.config.workspace_id, - WorkspaceResourceKind::Worker, - &record.worker.worker_id, - )? - .ok_or_else(|| { - Error::Store(format!( - "Workspace Worker `{}` has no resource key", - record.worker.worker_id - )) - })?, - ); - result.worker = Some(summary); - } - Ok(Json(WorkerRestoreResponse { + let summary = merge_worker_registry_projection(Some(worker), &record, links, &workdirs); + Some(project_workspace_worker(&api, summary)?) + } else { + None + }; + Ok(Json(workspace_api::WorkerRestoreResponse { workspace_id: api.workspace_id().to_string(), - worker_ref: RuntimeWorkerRef::new(&runtime_id, &worker_id), - result, + runtime_id: runtime_id.clone(), + worker_id: worker_id.clone(), + result: workspace_api::WorkerRestoreResult { + state: result.state.into(), + worker: projected_worker, + diagnostics: result.diagnostics.into_iter().map(Into::into).collect(), + }, })) } @@ -10179,28 +10153,33 @@ async fn list_runtime_workers( State(api): State, AxumPath(runtime_id): AxumPath, Query(query): Query, -) -> ApiResult>> { +) -> ApiResult>> { let limit = api.config.max_records.min(200); - let (worker_list, source) = match query.status { + let (runtime_workers, source) = match query.status { Some(RuntimeWorkersStatusFilter::Stopped) => ( api.runtime .list_stopped_workers_for_runtime(&runtime_id, limit) - .map_err(|err| err.into_error())?, + .map_err(|error| error.into_error())?, "runtime_registry_stopped", ), None => ( api.runtime .list_workers_for_runtime(&runtime_id, limit) - .map_err(|err| err.into_error())?, + .map_err(|error| error.into_error())?, "runtime_registry", ), }; - Ok(Json(RuntimeListResponse { + let items = project_observed_workspace_workers(&api, runtime_workers.items)?; + Ok(Json(workspace_api::ListResponse { workspace_id: api.workspace_id().to_string(), limit, - items: worker_list.items, + items, source: source.to_string(), - diagnostics: worker_list.diagnostics, + diagnostics: runtime_workers + .diagnostics + .into_iter() + .map(Into::into) + .collect(), })) } @@ -11142,22 +11121,70 @@ fn protocol_error_event(message: impl Into) -> protocol::Event { async fn list_host_workers( State(api): State, AxumPath(host_id): AxumPath, -) -> ApiResult>> { +) -> ApiResult>> { let limit = api.config.max_records.min(200); let runtime_workers = api .runtime .list_workers_for_host(&host_id, limit) .map_err(|err| err.into_error())?; - Ok(Json(RuntimeListResponse { - workspace_id: api.config.workspace_id, + let items = project_observed_workspace_workers(&api, runtime_workers.items)?; + Ok(Json(workspace_api::ListResponse { + workspace_id: api.workspace_id().to_string(), limit, - items: runtime_workers.items, + items, source: "worker_runtime_registry".to_string(), - diagnostics: runtime_workers.diagnostics, + diagnostics: runtime_workers + .diagnostics + .into_iter() + .map(Into::into) + .collect(), })) } -fn workers_response(api: WorkspaceApi) -> ApiResult> { +fn project_workspace_worker( + api: &WorkspaceApi, + summary: WorkerSummary, +) -> ApiResult { + let resource_key = api + .store + .resource_key( + &api.config.workspace_id, + WorkspaceResourceKind::Worker, + &summary.worker.worker_id, + )? + .ok_or_else(|| { + Error::Store(format!( + "Workspace Worker `{}` has no resource key", + summary.worker.worker_id + )) + })?; + Ok(workspace_worker_summary(summary, resource_key)) +} + +fn project_observed_workspace_workers( + api: &WorkspaceApi, + workers: Vec, +) -> ApiResult> { + let workdirs = api + .store + .list_workdir_registry(&api.config.workspace_id, 500)?; + workers + .into_iter() + .map(|worker| { + let record = sync_worker_observation(api, &worker)?; + let links = api + .store + .list_worker_workdir_links(&api.config.workspace_id, &record.worker)?; + let summary = + merge_worker_registry_projection(Some(&worker), &record, links, &workdirs); + project_workspace_worker(api, summary) + }) + .collect() +} + +fn workers_response( + api: WorkspaceApi, +) -> ApiResult> { let limit = api.config.max_records.min(200); let runtime_workers = api.runtime.list_workers(limit); let mut observed = std::collections::BTreeMap::new(); @@ -11196,34 +11223,20 @@ fn workers_response(api: WorkspaceApi) -> ApiResult WorkerSummary { WorkerSummary { worker: record.worker.clone(), - resource_key: None, host_id: "backend-registry".to_string(), display_name: record.display_name.clone(), label: record.display_name.clone(), @@ -15979,11 +15991,11 @@ mod tests { ) .await .unwrap(); + assert_eq!(retried_restore.worker_id, first_worker.worker.worker_id); assert_eq!( - retried_restore.worker_ref.worker_id, - first_worker.worker.worker_id + retried_restore.result.state, + workspace_api::WorkerOperationState::Accepted ); - assert_eq!(retried_restore.result.state, WorkerOperationState::Accepted); let restored_assignment = api .store .get_current_ticket_worker_assignment(TEST_WORKSPACE_ID, &second_ticket.id) @@ -18159,6 +18171,23 @@ mod tests { assert_eq!(worker["profile"], "builtin:companion"); assert!(worker.get("role").is_none()); assert_eq!(worker["worker_id"], created["worker_id"]); + let resource_key = worker["resource_key"] + .as_str() + .expect("Workspace Worker list must project a resource key"); + assert!(resource_key.starts_with("W-")); + + let runtime_workers = + get_json(app.clone(), "/api/runtimes/embedded-worker-runtime/workers").await; + let runtime_workers = serde_json::from_value::< + workspace_api::ListResponse, + >(runtime_workers) + .expect("Runtime-scoped Worker list must use the shared Workspace API contract"); + assert!( + runtime_workers + .items + .iter() + .any(|worker| worker.resource_key == resource_key) + ); let detail_path = format!( "/api/runtimes/{}/workers/{}", created["runtime_id"].as_str().unwrap(),