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

feat: migrate workspace worker and companion domains to ServerApi

Browse Source
This commit is contained in:
Hare
2026-09-23 10:00:56 +09:00
parent ad12d77ad4
commit 89024d3b01
28 changed files with 7471 additions and 819 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/client/src/backend_runtime.rs
+21 -33
View File
@@ -393,15 +393,16 @@ async fn get_backend_worker_launch_options_with_client(
target: &BackendWorkerLaunchTarget,
api: &BackendApiClient,
) -> Result<BackendWorkerLaunchOptions, BackendRuntimeClientError> {
let path = backend_workspace_workers_launch_options_path(
target
.workspace_id
.as_deref()
.expect("validated Backend Workspace scope"),
);
let response = api.request(HttpMethod::GET, &path)?.send().await?;
let response = api.require_success(response).await?;
Ok(response.json::<BackendWorkerLaunchOptions>().await?)
runtime_server_api_client(&target.base_url, api)?
.workspace_worker_launch_options(
target
.workspace_id
.as_deref()
.expect("validated Backend Workspace scope")
.to_string(),
)
.await
.map_err(Into::into)
}
pub async fn create_backend_worker(
@@ -418,19 +419,17 @@ async fn create_backend_worker_with_client(
request: &BackendCreateWorkerRequest,
api: &BackendApiClient,
) -> Result<BackendCreateWorkerResponse, BackendRuntimeClientError> {
let path = backend_workspace_workers_path(
target
.workspace_id
.as_deref()
.expect("validated Backend Workspace scope"),
);
let response = api
.request(HttpMethod::POST, &path)?
.json(request)
.send()
.await?;
let response = api.require_success(response).await?;
Ok(response.json::<BackendCreateWorkerResponse>().await?)
runtime_server_api_client(&target.base_url, api)?
.workspace_worker_create(
target
.workspace_id
.as_deref()
.expect("validated Backend Workspace scope")
.to_string(),
request.clone(),
)
.await
.map_err(Into::into)
}
pub async fn list_backend_workers(
@@ -680,17 +679,6 @@ fn validate_list_target(
Ok(())
}
fn backend_workspace_workers_path(workspace_id: &str) -> String {
format!("/api/w/{}/workers", path_segment_encode(workspace_id))
}
fn backend_workspace_workers_launch_options_path(workspace_id: &str) -> String {
format!(
"{}/launch-options",
backend_workspace_workers_path(workspace_id)
)
}
fn protocol_ws_request(
target: &BackendRuntimeTarget,
api: &BackendApiClient,
crates/client/src/workspace_product.rs
+13 -40
View File
@@ -1,6 +1,3 @@
use reqwest::Method;
use serde::Serialize;
use serde::de::DeserializeOwned;
use server_api::{
BrowserCreateWorkerResponse, BrowserWorkspaceOrchestratorResponse,
CreateWorkspaceWorkerRequest, ListResponse, MemoryDocumentResponse, MemoryStagingListResponse,
@@ -417,7 +414,10 @@ impl BackendWorkspaceProductClient {
&self,
ticket_id: &str,
) -> Result<String, BackendWorkspaceClientError> {
let options: WorkerLaunchOptionsResponse = self.get_json("/workers/launch-options")?;
let workspace_id = self.workspace_id.clone();
let options: WorkerLaunchOptionsResponse = self.generated(move |client| async move {
client.workspace_worker_launch_options(workspace_id).await
})?;
let runtime = options
.runtimes
.iter()
@@ -439,8 +439,10 @@ impl BackendWorkspaceProductClient {
workdir_attachments: Vec::new(),
control_operation_id: None,
};
let response: BrowserCreateWorkerResponse =
self.send_json(Method::POST, "/workers", Some(&request))?;
let workspace_id = self.workspace_id.clone();
let response: BrowserCreateWorkerResponse = self.generated(move |client| async move {
client.workspace_worker_create(workspace_id, request).await
})?;
Ok(format!(
"Started Intake Worker {}/{} for Ticket {ticket_id}",
response.runtime_id, response.worker_id
@@ -448,8 +450,11 @@ impl BackendWorkspaceProductClient {
}
pub fn start_workspace_orchestrator(&self) -> Result<String, BackendWorkspaceClientError> {
let workspace_id = self.workspace_id.clone();
let response: BrowserWorkspaceOrchestratorResponse =
self.send_json::<(), _>(Method::POST, "/orchestrator", None)?;
self.generated(move |client| async move {
client.workspace_orchestrator_start(workspace_id).await
})?;
let worker = response.worker.ok_or_else(|| {
BackendWorkspaceClientError::InvalidTarget(
"Backend accepted the Orchestrator request without returning a Worker".to_string(),
@@ -464,39 +469,6 @@ impl BackendWorkspaceProductClient {
pub fn default_product_list_limit() -> usize {
DEFAULT_PRODUCT_LIST_LIMIT
}
fn get_json<R: DeserializeOwned>(&self, path: &str) -> Result<R, BackendWorkspaceClientError> {
self.send_json::<(), R>(Method::GET, path, None)
}
fn send_json<B: Serialize + ?Sized, R: DeserializeOwned>(
&self,
method: Method,
path: &str,
body: Option<&B>,
) -> Result<R, BackendWorkspaceClientError> {
let response = self.request(method, path, body)?.send()?;
self.api.check_status(response.status())?;
response.json().map_err(BackendWorkspaceClientError::Http)
}
fn request<B: Serialize + ?Sized>(
&self,
method: Method,
path: &str,
body: Option<&B>,
) -> Result<reqwest::blocking::RequestBuilder, BackendWorkspaceClientError> {
let path = format!(
"/api/w/{}/{}",
encode_path_segment(&self.workspace_id),
path.trim_start_matches('/')
);
let request = self.api.blocking_request(method, &path)?;
Ok(match body {
Some(body) => request.json(body),
None => request,
})
}
}
impl TicketBackend for BackendWorkspaceProductClient {
@@ -785,6 +757,7 @@ fn ticket_list_state_query(query: &TicketListQuery) -> String {
}
}
#[cfg(test)]
fn encode_path_segment(value: &str) -> String {
let mut encoded = String::with_capacity(value.len());
for byte in value.bytes() {
crates/protocol/src/lib.rs
+45
View File
@@ -126,6 +126,7 @@ pub struct WorkerCommandAcknowledgement {
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
#[serde(tag = "kind", content = "state", rename_all = "snake_case")]
pub enum WorkerState {
Idle,
@@ -134,6 +135,7 @@ pub enum WorkerState {
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
#[serde(tag = "kind", content = "state", rename_all = "snake_case")]
pub enum WorkerBusyState {
Run(WorkerRunState),
@@ -142,6 +144,7 @@ pub enum WorkerBusyState {
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
#[serde(rename_all = "snake_case")]
pub enum WorkerRunState {
Running,
@@ -152,6 +155,7 @@ pub enum WorkerRunState {
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
#[serde(rename_all = "snake_case")]
pub enum WorkerMaintenanceState {
Compacting,
@@ -159,8 +163,13 @@ pub enum WorkerMaintenanceState {
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
pub struct WorkerStateSnapshot {
/// Highest lifecycle command id observed by this controller instance.
#[cfg_attr(
feature = "json-schema",
schemars(range(min = 0, max = 9_007_199_254_740_991_u64))
)]
pub last_command_id: u64,
pub state: WorkerState,
}
@@ -453,13 +462,29 @@ impl PasteArtifactAvailability {
#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
pub struct PasteArtifactRef {
pub artifact_id: String,
#[cfg_attr(
feature = "json-schema",
schemars(range(min = 0, max = 9_007_199_254_740_991_u64))
)]
pub created_at_ms: u64,
pub media_type: PasteArtifactMediaType,
/// Availability observed when this immutable reference was committed.
/// Reads revalidate storage and integrity rather than trusting this field.
pub availability: PasteArtifactAvailability,
#[cfg_attr(
feature = "json-schema",
schemars(range(min = 0, max = 9_007_199_254_740_991_u64))
)]
pub byte_len: u64,
#[cfg_attr(
feature = "json-schema",
schemars(range(min = 0, max = 9_007_199_254_740_991_u64))
)]
pub char_count: u64,
#[cfg_attr(
feature = "json-schema",
schemars(range(min = 0, max = 9_007_199_254_740_991_u64))
)]
pub line_count: u64,
pub sha256: String,
pub source_entry_id: String,
@@ -499,8 +524,16 @@ pub struct UploadedFileRef {
pub artifact_id: String,
pub file_name: String,
pub media_type: String,
#[cfg_attr(
feature = "json-schema",
schemars(range(min = 0, max = 9_007_199_254_740_991_u64))
)]
pub created_at_ms: u64,
pub availability: UploadedFileAvailability,
#[cfg_attr(
feature = "json-schema",
schemars(range(min = 0, max = 9_007_199_254_740_991_u64))
)]
pub byte_len: u64,
pub sha256: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
@@ -519,8 +552,20 @@ pub enum Segment {
/// `[Clipboard #N | X chars, Y lines]` chip in `Event::UserMessage`
/// re-broadcast.
Paste {
#[cfg_attr(
feature = "json-schema",
schemars(range(min = 0, max = 4_294_967_295_u32))
)]
id: u32,
#[cfg_attr(
feature = "json-schema",
schemars(range(min = 0, max = 4_294_967_295_u32))
)]
chars: u32,
#[cfg_attr(
feature = "json-schema",
schemars(range(min = 0, max = 4_294_967_295_u32))
)]
lines: u32,
content: String,
},
crates/server-api/Cargo.toml
+1 -3
View File
@@ -13,7 +13,7 @@ typescript = ["dep:ts-rs", "protocol/typescript"]
api-macros = { workspace = true, features = ["reqwest", "axum", "openapi"] }
memory.workspace = true
merge-request.workspace = true
protocol.workspace = true
protocol = { workspace = true, features = ["json-schema"] }
runtime-api.workspace = true
schemars.workspace = true
serde = { workspace = true, features = ["derive"] }
@@ -36,11 +36,9 @@ name = "generate_workdir_api_types"
[[example]]
name = "generate_worker_launch_api_types"
required-features = ["typescript"]
[[example]]
name = "generate_companion_api_types"
required-features = ["typescript"]
[[example]]
name = "generate_memory_api_types"
crates/server-api/examples/generate_companion_api_types.rs
+53 -4
View File
@@ -1,6 +1,55 @@
use std::fs;
use std::path::Path;
use server_api::repository_openapi_typescript::{
COMPANION_OUTPUT_PATH, INPUT_PATH, generate_companion_typescript,
};
fn main() {
print!(
"{}",
server_api::normalized_typescript(server_api::companion_api_typescript())
);
if let Err(error) = run() {
eprintln!("{error}");
std::process::exit(1);
}
}
fn run() -> Result<(), String> {
let mut check = false;
for argument in std::env::args().skip(1) {
match argument.as_str() {
"--check" => check = true,
"--help" | "-h" => {
println!(
"Generate the bounded Companion API TypeScript artifact.\n\nUsage: cargo run -q -p server-api --example generate_companion_api_types -- [--check]"
);
return Ok(());
}
other => return Err(format!("unknown argument `{other}`")),
}
}
let root = Path::new(env!("CARGO_MANIFEST_DIR")).join("../..");
let input_path = root.join(INPUT_PATH);
let output_path = root.join(COMPANION_OUTPUT_PATH);
let input = fs::read_to_string(&input_path)
.map_err(|error| format!("failed to read {}: {error}", input_path.display()))?;
let generated = generate_companion_typescript(&input)
.map_err(|error| format!("failed to generate Companion API types: {error}"))?;
if check {
let checked_in = fs::read_to_string(&output_path)
.map_err(|error| format!("failed to read {}: {error}", output_path.display()))?;
if checked_in != generated {
return Err(format!(
"{} is stale; regenerate it without --check",
output_path.display()
));
}
println!("{} is current", output_path.display());
return Ok(());
}
let temporary_path = output_path.with_extension("ts.tmp");
fs::write(&temporary_path, generated)
.map_err(|error| format!("failed to write {}: {error}", temporary_path.display()))?;
fs::rename(&temporary_path, &output_path)
.map_err(|error| format!("failed to replace {}: {error}", output_path.display()))?;
println!("generated {}", output_path.display());
Ok(())
}
crates/server-api/examples/generate_worker_launch_api_types.rs
+58 -4
View File
@@ -1,6 +1,60 @@
use std::fs;
use std::path::Path;
use server_api::repository_openapi_typescript::{
INPUT_PATH, WORKER_OUTPUT_PATH, generate_worker_typescript,
};
fn main() {
print!(
"{}",
server_api::normalized_typescript(server_api::worker_launch_api_typescript())
);
if let Err(error) = run() {
eprintln!("{error}");
std::process::exit(1);
}
}
fn run() -> Result<(), String> {
let check = parse_check_argument()?;
let root = Path::new(env!("CARGO_MANIFEST_DIR")).join("../..");
let input_path = root.join(INPUT_PATH);
let output_path = root.join(WORKER_OUTPUT_PATH);
let input = fs::read_to_string(&input_path)
.map_err(|error| format!("failed to read {}: {error}", input_path.display()))?;
let generated = generate_worker_typescript(&input)
.map_err(|error| format!("failed to generate Worker API types: {error}"))?;
if check {
let checked_in = fs::read_to_string(&output_path)
.map_err(|error| format!("failed to read {}: {error}", output_path.display()))?;
if checked_in != generated {
return Err(format!(
"{} is stale; regenerate it without --check",
output_path.display()
));
}
println!("{} is current", output_path.display());
return Ok(());
}
let temporary_path = output_path.with_extension("ts.tmp");
fs::write(&temporary_path, generated)
.map_err(|error| format!("failed to write {}: {error}", temporary_path.display()))?;
fs::rename(&temporary_path, &output_path)
.map_err(|error| format!("failed to replace {}: {error}", output_path.display()))?;
println!("generated {}", output_path.display());
Ok(())
}
fn parse_check_argument() -> Result<bool, String> {
let mut check = false;
for argument in std::env::args().skip(1) {
match argument.as_str() {
"--check" => check = true,
"--help" | "-h" => {
println!(
"Generate the bounded Worker API TypeScript artifact.\n\nUsage: cargo run -q -p server-api --example generate_worker_launch_api_types -- [--check]"
);
return Ok(true);
}
other => return Err(format!("unknown argument `{other}`")),
}
}
Ok(check)
}
crates/server-api/src/lib.rs
+803 -69
View File
File diff suppressed because it is too large Load Diff
crates/server-api/src/repository_openapi_typescript.rs
+226 -4
View File
@@ -18,6 +18,17 @@ pub const RUNTIME_GENERATOR_OPTIONS: &str =
"runtime-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output";
pub const RUNTIME_OUTPUT_PATH: &str = "web/workspace/src/lib/generated/runtime-api.ts";
pub const WORKER_GENERATOR_NAME: &str = "yoi-worker-openapi-typescript";
pub const WORKER_GENERATOR_VERSION: &str = "1";
pub const WORKER_GENERATOR_OPTIONS: &str =
"worker-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output";
pub const WORKER_OUTPUT_PATH: &str = "web/workspace/src/lib/generated/worker-launch-api.ts";
pub const COMPANION_GENERATOR_NAME: &str = "yoi-companion-openapi-typescript";
pub const COMPANION_GENERATOR_VERSION: &str = "1";
pub const COMPANION_GENERATOR_OPTIONS: &str =
"companion-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output";
pub const COMPANION_OUTPUT_PATH: &str = "web/workspace/src/lib/generated/companion-api.ts";
const MAX_SAFE_INTEGER: f64 = 9_007_199_254_740_991.0;
const MIN_SAFE_INTEGER: f64 = -9_007_199_254_740_991.0;
const SCHEMA_REF_PREFIX: &str = "#/components/schemas/";
@@ -166,6 +177,82 @@ pub fn generate_runtime_typescript(input: &str) -> Result<String, GenerationErro
Ok(output)
}
pub fn generate_worker_typescript(input: &str) -> Result<String, GenerationError> {
generate_domain_typescript(
input,
"Worker",
validate_worker_operations,
WORKER_GENERATOR_NAME,
WORKER_GENERATOR_VERSION,
WORKER_GENERATOR_OPTIONS,
WORKER_OUTPUT_PATH,
"generate_worker_launch_api_types",
)
}
pub fn generate_companion_typescript(input: &str) -> Result<String, GenerationError> {
generate_domain_typescript(
input,
"Companion",
validate_companion_operations,
COMPANION_GENERATOR_NAME,
COMPANION_GENERATOR_VERSION,
COMPANION_GENERATOR_OPTIONS,
COMPANION_OUTPUT_PATH,
"generate_companion_api_types",
)
}
#[allow(clippy::too_many_arguments)]
fn generate_domain_typescript(
input: &str,
domain: &str,
validate_operations: fn(&Map<String, Value>) -> Result<BTreeSet<String>, GenerationError>,
generator_name: &str,
generator_version: &str,
generator_options: &str,
output_path: &str,
example_name: &str,
) -> Result<String, GenerationError> {
let document: Value = serde_json::from_str(input)
.map_err(|error| GenerationError::invalid(format!("invalid OpenAPI JSON: {error}")))?;
let root = object(&document, "OpenAPI document")?;
if root.get("openapi").and_then(Value::as_str) != Some("3.1.0") {
return Err(GenerationError::invalid(format!(
"{domain} TypeScript generation requires OpenAPI 3.1.0"
)));
}
let info = object_field(root, "info", "OpenAPI document")?;
let contract_version = required_string(info, "version", "OpenAPI info")?;
let contract_digest = required_string(info, "x-yoi-source-digest", "OpenAPI info")?;
validate_contract_digest(contract_digest)?;
let schemas = object_field(
object_field(root, "components", "OpenAPI document")?,
"schemas",
"OpenAPI components",
)?;
let paths = object_field(root, "paths", "OpenAPI document")?;
let roots = validate_operations(paths)?;
let closure = schema_closure(schemas, &roots)?;
let mut output = format!(
"// @generated by {generator_name}@{generator_version}. Do not edit.\n\
// Input: {INPUT_PATH}\n\
// Input contract: {contract_version} ({contract_digest})\n\
// Options: {generator_options}\n\
// Output: {output_path}\n\
// Regenerate/check: cargo run -q -p server-api --example {example_name} -- --check\n\n"
);
for name in closure {
let schema = schemas
.get(&name)
.ok_or_else(|| GenerationError::invalid(format!("missing schema `{name}`")))?;
let declaration = render_schema(schema, &format!("schema `{name}`"))?;
output.push_str(&format!("export type {name} = {declaration};\n\n"));
}
output.pop();
Ok(output)
}
fn validate_contract_digest(digest: &str) -> Result<(), GenerationError> {
let Some(hex) = digest.strip_prefix("sha256:") else {
return Err(GenerationError::invalid(
@@ -180,6 +267,141 @@ fn validate_contract_digest(digest: &str) -> Result<(), GenerationError> {
Ok(())
}
fn validate_worker_operations(
paths: &Map<String, Value>,
) -> Result<BTreeSet<String>, GenerationError> {
let operations = [
OperationContract {
path: "/api/workers",
method: "get",
operation_id: "workspace_worker_list_alias",
request: None,
successes: &[("200", "WorkerListResponse")],
},
OperationContract {
path: "/api/workers",
method: "post",
operation_id: "workspace_worker_create_alias",
request: Some("CreateWorkspaceWorkerRequest"),
successes: &[("200", "BrowserCreateWorkerResponse")],
},
OperationContract {
path: "/api/w/{workspace_id}/orchestrator",
method: "get",
operation_id: "workspace_orchestrator_status",
request: None,
successes: &[("200", "BrowserWorkspaceOrchestratorResponse")],
},
OperationContract {
path: "/api/w/{workspace_id}/orchestrator",
method: "post",
operation_id: "workspace_orchestrator_start",
request: None,
successes: &[("200", "BrowserWorkspaceOrchestratorResponse")],
},
OperationContract {
path: "/api/w/{workspace_id}/workers",
method: "get",
operation_id: "workspace_worker_list",
request: None,
successes: &[("200", "WorkerListResponse")],
},
OperationContract {
path: "/api/w/{workspace_id}/workers",
method: "post",
operation_id: "workspace_worker_create",
request: Some("CreateWorkspaceWorkerRequest"),
successes: &[("200", "BrowserCreateWorkerResponse")],
},
OperationContract {
path: "/api/w/{workspace_id}/workers/{worker_ref}",
method: "get",
operation_id: "workspace_worker_detail",
request: None,
successes: &[("200", "WorkerSummary")],
},
OperationContract {
path: "/api/workers/launch-options",
method: "get",
operation_id: "workspace_worker_launch_options_alias",
request: None,
successes: &[("200", "WorkerLaunchOptionsResponse")],
},
OperationContract {
path: "/api/w/{workspace_id}/workers/launch-options",
method: "get",
operation_id: "workspace_worker_launch_options",
request: None,
successes: &[("200", "WorkerLaunchOptionsResponse")],
},
];
validate_operation_contracts(paths, &operations)
}
fn validate_companion_operations(
paths: &Map<String, Value>,
) -> Result<BTreeSet<String>, GenerationError> {
let operations = [
OperationContract {
path: "/api/companion/status",
method: "get",
operation_id: "companion_status_alias",
request: None,
successes: &[("200", "CompanionStatusResponse")],
},
OperationContract {
path: "/api/w/{workspace_id}/companion/status",
method: "get",
operation_id: "companion_status",
request: None,
successes: &[("200", "CompanionStatusResponse")],
},
OperationContract {
path: "/api/companion/transcript",
method: "get",
operation_id: "companion_transcript_alias",
request: None,
successes: &[("200", "CompanionTranscriptProjection")],
},
OperationContract {
path: "/api/w/{workspace_id}/companion/transcript",
method: "get",
operation_id: "companion_transcript",
request: None,
successes: &[("200", "CompanionTranscriptProjection")],
},
OperationContract {
path: "/api/companion/messages",
method: "post",
operation_id: "companion_message_alias",
request: Some("CompanionMessageRequest"),
successes: &[("200", "CompanionMessageResponse")],
},
OperationContract {
path: "/api/w/{workspace_id}/companion/messages",
method: "post",
operation_id: "companion_message",
request: Some("CompanionMessageRequest"),
successes: &[("200", "CompanionMessageResponse")],
},
OperationContract {
path: "/api/companion/cancel",
method: "post",
operation_id: "companion_cancel_alias",
request: Some("CompanionCancelRequest"),
successes: &[("200", "CompanionMessageResponse")],
},
OperationContract {
path: "/api/w/{workspace_id}/companion/cancel",
method: "post",
operation_id: "companion_cancel",
request: Some("CompanionCancelRequest"),
successes: &[("200", "CompanionMessageResponse")],
},
];
validate_operation_contracts(paths, &operations)
}
fn validate_repository_operations(
paths: &Map<String, Value>,
) -> Result<BTreeSet<String>, GenerationError> {
@@ -948,9 +1170,7 @@ fn literal_matches_type(kind: &str, value: &Value) -> bool {
fn tagged_object_discriminator(value: &Value) -> Option<(String, String)> {
let schema = value.as_object()?;
if schema.get("type")?.as_str()? != "object"
|| schema.get("additionalProperties") != Some(&Value::Bool(false))
{
if schema.get("type")?.as_str()? != "object" {
return None;
}
let required = schema.get("required")?.as_array()?;
@@ -1447,7 +1667,9 @@ mod tests {
)));
assert!(output.contains("export type RuntimeCleanupPlanResponse ="));
assert!(output.contains("export type RuntimeWorkerLifecycleResult ="));
assert!(output.contains("export type RuntimeWorkerSpawnAcceptanceRequirement = \"socket_ready\" | { run_accepted:"));
assert!(output.contains(
"export type RuntimeWorkerSpawnAcceptanceRequirement = { kind: \"socket_ready\"; } | { expected_segments: number; kind: \"run_accepted\"; };"
));
assert!(!output.contains("export type UnmigratedDomain ="));
let mut document: Value = serde_json::from_str(OPENAPI).unwrap();
crates/worker-runtime/src/resource.rs
+52 -41
View File
@@ -1,7 +1,7 @@
use crate::auth::BACKEND_RESOURCE_FETCH_PERMISSION;
use crate::identity::WorkerId;
use crate::profile_archive::{ProfileSourceArchive, ProfileSourceArchiveRef, sha256_hex};
use crate::workspace_request::{RuntimeWorkspaceRequest, RuntimeWorkspaceRequestClient};
use crate::workspace_request::RuntimeWorkspaceRequestClient;
use async_trait::async_trait;
use serde::{Deserialize, Serialize};
@@ -219,11 +219,6 @@ impl BackendResourceClient for HttpBackendResourceClient {
&self,
request: BackendResourceFetchRequest,
) -> Result<BackendResourceFetchResponse, BackendResourceError> {
let body = serde_json::to_vec(&request).map_err(|error| {
BackendResourceError::InvalidResponse {
message: error.to_string(),
}
})?;
let endpoint = reqwest::Url::parse(&self.endpoint).map_err(|error| {
BackendResourceError::Transport {
message: error.to_string(),
@@ -248,17 +243,17 @@ impl BackendResourceClient for HttpBackendResourceClient {
.to_string(),
}
})?;
if !endpoint_suffix.starts_with('/') {
let expected_path = format!(
"/api/runtime/v1/workspaces/{}/resources/fetch",
client.workspace_id()
);
if endpoint_suffix != expected_path {
return Err(BackendResourceError::Unauthorized {
message: "Workspace resource endpoint does not match its request client"
.to_string(),
});
}
let mut headers = reqwest::header::HeaderMap::new();
headers.insert(
reqwest::header::CONTENT_TYPE,
reqwest::header::HeaderValue::from_static("application/json"),
);
if let Some(token) = self.bearer_token.as_deref() {
let value = reqwest::header::HeaderValue::from_str(&format!("Bearer {token}"))
.map_err(|error| BackendResourceError::Transport {
@@ -266,41 +261,57 @@ impl BackendResourceClient for HttpBackendResourceClient {
})?;
headers.insert(reqwest::header::AUTHORIZATION, value);
}
let response = client
.execute(RuntimeWorkspaceRequest {
method: reqwest::Method::POST,
path_and_query: endpoint_suffix.to_string(),
body,
let request: server_api::RuntimeResourceFetchRequest =
serde_json::from_value(serde_json::to_value(&request).map_err(|error| {
BackendResourceError::InvalidResponse {
message: error.to_string(),
}
})?)
.map_err(|error| BackendResourceError::InvalidResponse {
message: error.to_string(),
})?;
let generated = client
.server_api_client(
BACKEND_RESOURCE_FETCH_PERMISSION,
None,
headers,
permission: BACKEND_RESOURCE_FETCH_PERMISSION.to_string(),
worker_id: None,
timeout: Some(self.request_timeout),
max_response_bytes: 8 * 1024 * 1024,
})
self.request_timeout,
8 * 1024 * 1024,
)
.map_err(|error| BackendResourceError::Transport {
message: error.to_string(),
})?;
match generated
.runtime_resource_fetch(client.workspace_id().to_string(), request)
.await
.map_err(|error| {
if error.is_timeout() {
BackendResourceError::Timeout
} else {
BackendResourceError::Transport {
{
Ok(response) => {
serde_json::from_value(serde_json::to_value(&response).map_err(|error| {
BackendResourceError::InvalidResponse {
message: error.to_string(),
}
}
})?;
if response.status.is_success() {
serde_json::from_slice::<BackendResourceFetchResponse>(&response.body).map_err(|err| {
BackendResourceError::InvalidResponse {
message: err.to_string(),
}
})
} else {
let status = response.status;
match serde_json::from_slice::<BackendResourceError>(&response.body) {
Ok(error) => Err(error),
Err(err) => Err(BackendResourceError::Transport {
message: format!("backend resource fetch failed with HTTP {status}: {err}"),
}),
})?)
.map_err(|error| BackendResourceError::InvalidResponse {
message: error.to_string(),
})
}
Err(server_api::client_support::ClientError::Public { error, .. }) => {
serde_json::from_value(serde_json::to_value(error).map_err(|error| {
BackendResourceError::InvalidResponse {
message: error.to_string(),
}
})?)
.map_err(|error| BackendResourceError::InvalidResponse {
message: error.to_string(),
})
.and_then(Err)
}
Err(server_api::client_support::ClientError::Failure(
server_api::client_support::ClientFailure::Timeout,
)) => Err(BackendResourceError::Timeout),
Err(error) => Err(BackendResourceError::Transport {
message: error.to_string(),
}),
}
}
}
crates/worker-runtime/src/worker_source.rs
+191 -68
View File
@@ -5,6 +5,7 @@ use crate::workspace_request::{RuntimeWorkspaceRequest, RuntimeWorkspaceRequestC
use worker::{
WorkspaceClient, WorkspaceClientError, WorkspacePromptCatalogResolution,
WorkspacePromptProjection, WorkspaceRequest, WorkspaceRequestMethod, WorkspaceResponse,
WorkspaceServerOperation,
};
use crate::auth::{
@@ -254,45 +255,53 @@ fn execute_remote_worker_remove_http(
fn execute_remote_worker_remove_http_blocking(
request: RemoteWorkerRemoveHttpRequest,
) -> Result<WorkspaceResponse, RuntimeWorkerMutationForwardError> {
let path = format!(
"/api/w/{}/workers/remove",
request.request_client.workspace_id()
);
let body = serde_json::to_vec(&serde_json::json!({
"target_runtime_id": request.target_runtime_id,
"target_worker_id": request.target_worker_id,
"reason": request.reason,
}))
.map_err(|error| RuntimeWorkerMutationForwardError::Transport(error.to_string()))?;
let mut headers = reqwest::header::HeaderMap::new();
headers.insert(
crate::auth::WORKER_MUTATION_SOURCE_PROOF_HEADER,
reqwest::header::HeaderValue::from_str(&request.token)
.map_err(|error| RuntimeWorkerMutationForwardError::Transport(error.to_string()))?,
);
headers.insert(
reqwest::header::CONTENT_TYPE,
reqwest::header::HeaderValue::from_static("application/json"),
);
let response = request
let client = request
.request_client
.execute_blocking(RuntimeWorkspaceRequest {
method: reqwest::Method::POST,
path_and_query: path,
body,
.server_api_client(
WORKSPACE_REQUEST_PERMISSION,
Some(request.source_worker_id),
headers,
permission: WORKSPACE_REQUEST_PERMISSION.to_string(),
worker_id: Some(request.source_worker_id),
timeout: Some(Duration::from_secs(5)),
max_response_bytes: 8 * 1024 * 1024,
})
Duration::from_secs(5),
8 * 1024 * 1024,
)
.map_err(|error| RuntimeWorkerMutationForwardError::Transport(error.to_string()))?;
let body = String::from_utf8(response.body)
let workspace_id = request.request_client.workspace_id().to_string();
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.map_err(|error| RuntimeWorkerMutationForwardError::Transport(error.to_string()))?;
Ok(WorkspaceResponse {
status: response.status.as_u16(),
body,
})
let result = runtime.block_on(client.workspace_worker_remove(
workspace_id,
server_api::WorkerRemoveRequest {
target_runtime_id: request.target_runtime_id,
target_worker_id: request.target_worker_id,
reason: request.reason,
},
));
match result {
Ok(response) => Ok(WorkspaceResponse {
status: 200,
body: serde_json::to_string(&response)
.map_err(|error| RuntimeWorkerMutationForwardError::Transport(error.to_string()))?,
}),
Err(server_api::client_support::ClientError::Public { status, error }) => {
Ok(WorkspaceResponse {
status: status.as_u16(),
body: serde_json::to_string(&error).map_err(|error| {
RuntimeWorkerMutationForwardError::Transport(error.to_string())
})?,
})
}
Err(error) => Err(RuntimeWorkerMutationForwardError::Transport(
error.to_string(),
)),
}
}
#[derive(Clone)]
@@ -432,6 +441,26 @@ impl std::fmt::Debug for RuntimeOwnedWorkspaceClient {
}
}
fn generated_workspace_response<T: serde::Serialize>(
response: Result<T, server_api::client_support::ClientError<server_api::RepositoryApiError>>,
) -> Result<WorkspaceResponse, WorkspaceClientError> {
match response {
Ok(response) => Ok(WorkspaceResponse {
status: 200,
body: serde_json::to_string(&response)
.map_err(|error| WorkspaceClientError::Request(error.to_string()))?,
}),
Err(server_api::client_support::ClientError::Public { status, error }) => {
Ok(WorkspaceResponse {
status: status.as_u16(),
body: serde_json::to_string(&error)
.map_err(|error| WorkspaceClientError::Request(error.to_string()))?,
})
}
Err(error) => Err(WorkspaceClientError::Request(error.to_string())),
}
}
impl WorkspaceClient for RuntimeOwnedWorkspaceClient {
fn workspace_id(&self) -> Option<&str> {
Some(self.request_client.workspace_id())
@@ -452,36 +481,143 @@ impl WorkspaceClient for RuntimeOwnedWorkspaceClient {
self.execute_with_permission(request, WORKSPACE_REQUEST_PERMISSION)
}
fn execute_server_operation(
&self,
operation: WorkspaceServerOperation,
) -> Result<WorkspaceResponse, WorkspaceClientError> {
let request_client = self.request_client.clone();
let workspace_id = self.workspace_id.clone();
let worker_id = self.worker_id.clone();
let timeout = self.request_timeout.unwrap_or(Duration::from_secs(5));
std::thread::Builder::new()
.name("yoi-workspace-server-api".to_string())
.spawn(move || {
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.map_err(|error| WorkspaceClientError::Request(error.to_string()))?;
let client = request_client
.server_api_client(
WORKSPACE_REQUEST_PERMISSION,
Some(worker_id),
reqwest::header::HeaderMap::new(),
timeout,
8 * 1024 * 1024,
)
.map_err(|error| WorkspaceClientError::Request(error.to_string()))?;
runtime.block_on(async move {
match operation {
WorkspaceServerOperation::WorkerControlList => {
generated_workspace_response(
client.worker_control_list(workspace_id).await,
)
}
WorkspaceServerOperation::WorkerControlSpawn(request) => {
generated_workspace_response(
client.worker_control_spawn(workspace_id, request).await,
)
}
WorkspaceServerOperation::WorkerControlInput {
runtime_id,
worker_id,
request,
} => generated_workspace_response(
client
.worker_control_input(workspace_id, runtime_id, worker_id, request)
.await,
),
WorkspaceServerOperation::WorkerControlCancel {
runtime_id,
worker_id,
request,
} => generated_workspace_response(
client
.worker_control_cancel(workspace_id, runtime_id, worker_id, request)
.await,
),
WorkspaceServerOperation::WorkerControlStop {
runtime_id,
worker_id,
request,
} => generated_workspace_response(
client
.worker_control_stop(workspace_id, runtime_id, worker_id, request)
.await,
),
WorkspaceServerOperation::WorkerControlRestore {
runtime_id,
worker_id,
} => generated_workspace_response(
client
.worker_control_restore(workspace_id, runtime_id, worker_id)
.await,
),
WorkspaceServerOperation::WorkerObservationSessions => {
generated_workspace_response(
client.worker_observation_sessions(workspace_id).await,
)
}
WorkspaceServerOperation::WorkerObservationCapture(request) => {
generated_workspace_response(
client
.worker_observation_capture(workspace_id, request)
.await,
)
}
}
})
})
.map_err(|error| WorkspaceClientError::Request(error.to_string()))?
.join()
.map_err(|_| {
WorkspaceClientError::Request(
"Workspace Server API request thread panicked".to_string(),
)
})?
}
fn list_workspace_workers(
&self,
request: worker::WorkspaceWorkerDiscoveryRequest,
) -> Result<server_api::WorkspaceWorkerDiscoveryPage, WorkspaceClientError> {
let mut path = format!(
"/api/w/{}/worker-discovery/workers?limit={}",
self.workspace_id, request.limit
);
if let Some(cursor) = request.cursor.as_deref() {
path.push_str("&cursor=");
path.push_str(&percent_encode_query(cursor));
}
if let Some(query) = request.query.as_deref() {
path.push_str("&query=");
path.push_str(&percent_encode_query(query));
}
let response = self.execute_with_permission(
WorkspaceRequest::get(path),
WORKSPACE_WORKER_DISCOVERY_PERMISSION,
)?;
if !(200..300).contains(&response.status) {
return Err(WorkspaceClientError::Request(format!(
"Workspace Worker discovery failed with HTTP {}: {}",
response.status, response.body
)));
}
serde_json::from_str(&response.body).map_err(|error| {
WorkspaceClientError::Request(format!(
"invalid Workspace Worker discovery response: {error}"
))
let request_client = self.request_client.clone();
let workspace_id = self.workspace_id.clone();
let worker_id = self.worker_id.clone();
let timeout = self.request_timeout.unwrap_or(Duration::from_secs(5));
let query = server_api::WorkspaceWorkerDiscoveryQuery {
cursor: request.cursor,
limit: Some(request.limit),
query: request.query,
};
let result = std::thread::Builder::new()
.name("yoi-worker-discovery-http".to_string())
.spawn(move || {
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.map_err(|error| error.to_string())?;
let client = request_client
.server_api_client(
WORKSPACE_WORKER_DISCOVERY_PERMISSION,
Some(worker_id),
reqwest::header::HeaderMap::new(),
timeout,
8 * 1024 * 1024,
)
.map_err(|error| error.to_string())?;
runtime
.block_on(client.workspace_worker_discovery(workspace_id, query))
.map_err(|error| error.to_string())
})
.map_err(|error| WorkspaceClientError::Request(error.to_string()))?
.join()
.map_err(|_| {
WorkspaceClientError::Request(
"Workspace Worker discovery HTTP thread panicked".to_string(),
)
})?;
result.map_err(|error| {
WorkspaceClientError::Request(format!("Workspace Worker discovery failed: {error}"))
})
}
@@ -587,19 +723,6 @@ impl WorkspaceClient for RuntimeOwnedWorkspaceClient {
}
}
fn percent_encode_query(value: &str) -> String {
let mut encoded = String::with_capacity(value.len());
for byte in value.bytes() {
if byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'.' | b'_' | b'~') {
encoded.push(char::from(byte));
} else {
use std::fmt::Write as _;
let _ = write!(encoded, "%{byte:02X}");
}
}
encoded
}
#[derive(Debug, thiserror::Error)]
pub enum RuntimeWorkerMutationForwardError {
#[error(transparent)]
crates/worker-runtime/src/workspace_request.rs
+78 -2
View File
@@ -62,9 +62,57 @@ impl RuntimeWorkspaceRequestError {
timeout,
}
}
}
pub(crate) fn is_timeout(&self) -> bool {
matches!(self, Self::Transport { timeout: true, .. })
#[derive(Clone)]
pub(crate) struct RuntimeWorkspaceRequestAuthorizer {
workspace_id: String,
runtime_id: String,
request_source: Option<(RuntimeRequestSourceSigner, String)>,
permission: String,
worker_id: Option<String>,
extra_headers: HeaderMap,
}
impl server_api::client_support::RequestAuthorizer for RuntimeWorkspaceRequestAuthorizer {
fn authorize(
&self,
request: server_api::client_support::AuthorizerRequest<'_>,
) -> Result<HeaderMap, server_api::client_support::AuthorizationError> {
let mut headers = self.extra_headers.clone();
headers.insert(
RUNTIME_ID_HEADER,
reqwest::header::HeaderValue::from_str(&self.runtime_id)
.map_err(|_| server_api::client_support::AuthorizationError::new())?,
);
if let Some(worker_id) = self.worker_id.as_deref() {
headers.insert(
WORKER_ID_HEADER,
reqwest::header::HeaderValue::from_str(worker_id)
.map_err(|_| server_api::client_support::AuthorizationError::new())?,
);
}
if let Some((signer, audience)) = self.request_source.as_ref() {
let proof = signer
.issue(
audience,
&self.workspace_id,
self.worker_id.as_deref(),
&self.permission,
request.method.as_str(),
request.path_and_query,
request.body,
unix_now_seconds(),
DEFAULT_REQUEST_PROOF_TTL_SECONDS,
)
.map_err(|_| server_api::client_support::AuthorizationError::new())?;
headers.insert(
RUNTIME_REQUEST_SOURCE_PROOF_HEADER,
reqwest::header::HeaderValue::from_str(&proof)
.map_err(|_| server_api::client_support::AuthorizationError::new())?,
);
}
Ok(headers)
}
}
@@ -117,6 +165,34 @@ impl RuntimeWorkspaceRequestClient {
&& self.base_url.trim_end_matches('/') == base_url.trim_end_matches('/')
}
pub(crate) fn server_api_client(
&self,
permission: impl Into<String>,
worker_id: Option<String>,
extra_headers: HeaderMap,
timeout: Duration,
response_body_limit: usize,
) -> Result<
server_api::ServerApiClient<RuntimeWorkspaceRequestAuthorizer>,
RuntimeWorkspaceRequestError,
> {
let authorizer = RuntimeWorkspaceRequestAuthorizer {
workspace_id: self.workspace_id.clone(),
runtime_id: self.runtime_id.clone(),
request_source: self.request_source.clone(),
permission: permission.into(),
worker_id,
extra_headers,
};
server_api::ServerApiClient::builder(&self.base_url)
.map_err(|error| RuntimeWorkspaceRequestError::InvalidRequest(error.to_string()))?
.authorizer(authorizer)
.request_timeout(timeout)
.response_body_limit(response_body_limit)
.build()
.map_err(|error| RuntimeWorkspaceRequestError::InvalidRequest(error.to_string()))
}
pub(crate) async fn execute(
&self,
request: RuntimeWorkspaceRequest,
crates/worker/src/feature/builtin/manage_worker.rs
+75 -85
View File
@@ -19,10 +19,7 @@ use crate::feature::{
ServiceDeclaration, ServiceId, ToolContribution, ToolDeclaration,
};
use crate::spawn::registry::{SpawnedWorkerRegistry, SubWorkerStopSummary};
use crate::worker::{
WorkspaceClient, WorkspaceClientError, WorkspaceRequest, WorkspaceRequestMethod,
WorkspaceResponse,
};
use crate::worker::{WorkspaceClient, WorkspaceClientError, WorkspaceResponse};
const FEATURE_ID: &str = "worker";
const FEATURE_NAME: &str = "Worker";
@@ -76,16 +73,22 @@ pub trait WorkerControlService: Send + Sync {
&self,
request: WorkerLifecycleSpawnRequest,
) -> Result<WorkspaceResponse, WorkspaceClientError>;
async fn restore_worker(
&self,
runtime_id: String,
worker_id: String,
) -> Result<WorkspaceResponse, WorkspaceClientError> {
let _ = (runtime_id, worker_id);
Err(WorkspaceClientError::Unavailable(
"Runtime Worker restore is unavailable".to_string(),
))
}
fn remove_runtime_worker(
&self,
runtime_id: &str,
worker_id: &str,
reason: &str,
) -> Result<WorkspaceResponse, WorkspaceClientError>;
async fn execute_runtime(
&self,
request: WorkspaceRequest,
) -> Result<WorkspaceResponse, WorkspaceClientError>;
async fn ensure_permission(
&self,
subject: &super::worker_observation::WorkerObservationSubjectRef,
@@ -144,11 +147,9 @@ impl WorkerControlService for WorkspaceWorkerControlService {
async fn list_workers(&self) -> Result<WorkspaceResponse, WorkspaceClientError> {
let response = if self.runtime_worker_control {
self.execute_runtime(WorkspaceRequest::get(format!(
"/api/w/{}/worker-control/workers",
self.workspace_id
)))
.await?
self.client.execute_server_operation(
crate::worker::WorkspaceServerOperation::WorkerControlList,
)?
} else {
WorkspaceResponse {
status: 200,
@@ -210,19 +211,17 @@ impl WorkerControlService for WorkspaceWorkerControlService {
WorkerControlInputKind::User => "user",
WorkerControlInputKind::Notify => "notify",
};
self.execute_runtime(WorkspaceRequest::json(
WorkspaceRequestMethod::Post,
format!(
"/api/w/{}/worker-control/workers/{runtime_id}/{worker_id}/input",
self.workspace_id
),
serde_json::json!({
"kind": kind,
"content": content,
})
.to_string(),
))
.await
self.client.execute_server_operation(
crate::worker::WorkspaceServerOperation::WorkerControlInput {
runtime_id,
worker_id,
request: server_api::RuntimeWorkerInputRequest {
kind: Some(kind.to_string()),
content,
segments: None,
},
},
)
}
WorkerControlSubject::SubWorker { name } => {
if kind != WorkerControlInputKind::User {
@@ -273,19 +272,29 @@ impl WorkerControlService for WorkspaceWorkerControlService {
runtime_id,
worker_id,
} => {
let action = match kind {
WorkerControlStopKind::Cancel => "cancel",
WorkerControlStopKind::Stop => "stop",
let operation = match kind {
WorkerControlStopKind::Cancel => {
crate::worker::WorkspaceServerOperation::WorkerControlCancel {
runtime_id,
worker_id,
request: server_api::RuntimeWorkerLifecycleRequest {
reason,
ticket_assignment: None,
},
}
}
WorkerControlStopKind::Stop => {
crate::worker::WorkspaceServerOperation::WorkerControlStop {
runtime_id,
worker_id,
request: server_api::RuntimeWorkerLifecycleRequest {
reason,
ticket_assignment: None,
},
}
}
};
self.execute_runtime(WorkspaceRequest::json(
WorkspaceRequestMethod::Post,
format!(
"/api/w/{}/worker-control/workers/{runtime_id}/{worker_id}/{action}",
self.workspace_id
),
serde_json::json!({ "reason": reason }).to_string(),
))
.await
self.client.execute_server_operation(operation)
}
WorkerControlSubject::SubWorker { name } => {
if kind != WorkerControlStopKind::Stop {
@@ -329,12 +338,29 @@ impl WorkerControlService for WorkspaceWorkerControlService {
) -> Result<WorkspaceResponse, WorkspaceClientError> {
WorkspaceWorkerLifecycleService {
client: self.client.clone(),
workspace_id: self.workspace_id.clone(),
}
.spawn(request)
.await
}
async fn restore_worker(
&self,
runtime_id: String,
worker_id: String,
) -> Result<WorkspaceResponse, WorkspaceClientError> {
if !self.runtime_worker_control {
return Err(WorkspaceClientError::Unavailable(
"Runtime Worker control is not enabled for this Worker".to_string(),
));
}
self.client.execute_server_operation(
crate::worker::WorkspaceServerOperation::WorkerControlRestore {
runtime_id,
worker_id,
},
)
}
fn remove_runtime_worker(
&self,
runtime_id: &str,
@@ -350,18 +376,6 @@ impl WorkerControlService for WorkspaceWorkerControlService {
.execute_worker_remove(runtime_id, worker_id, reason)
}
async fn execute_runtime(
&self,
request: WorkspaceRequest,
) -> Result<WorkspaceResponse, WorkspaceClientError> {
if !self.runtime_worker_control {
return Err(WorkspaceClientError::Request(
"Runtime Worker control is not enabled for this Worker".to_string(),
));
}
self.client.execute(request)
}
async fn ensure_permission(
&self,
subject: &super::worker_observation::WorkerObservationSubjectRef,
@@ -391,10 +405,9 @@ impl WorkerControlService for WorkspaceWorkerControlService {
"Runtime Worker control is not enabled for this Worker".to_string(),
));
}
let response = self.client.execute(WorkspaceRequest::get(format!(
"/api/w/{}/worker-control/workers",
self.workspace_id
)))?;
let response = self.client.execute_server_operation(
crate::worker::WorkspaceServerOperation::WorkerControlList,
)?;
if !response.is_success() {
return Err(WorkspaceClientError::Request(format!(
"Workspace control request returned {}: {}",
@@ -459,7 +472,6 @@ pub struct WorkerLifecycleSpawnRequest {
struct WorkspaceWorkerLifecycleService {
client: Arc<dyn WorkspaceClient>,
workspace_id: String,
}
fn workspace_worker_create_request(
@@ -505,12 +517,9 @@ impl WorkerLifecycleService for WorkspaceWorkerLifecycleService {
request: WorkerLifecycleSpawnRequest,
) -> Result<WorkspaceResponse, WorkspaceClientError> {
let body = workspace_worker_create_request(request)?;
self.client.execute(WorkspaceRequest::json(
WorkspaceRequestMethod::Post,
format!("/api/w/{}/worker-control/workers", self.workspace_id),
serde_json::to_string(&body)
.map_err(|error| WorkspaceClientError::Request(error.to_string()))?,
))
self.client.execute_server_operation(
crate::worker::WorkspaceServerOperation::WorkerControlSpawn(body),
)
}
}
@@ -639,20 +648,17 @@ impl FeatureModule for ManageWorkerFeature {
}
fn install(&self, context: &mut FeatureInstallContext<'_>) -> Result<(), FeatureInstallError> {
let workspace_id = self
.client
self.client
.workspace_id()
.filter(|value| !value.trim().is_empty())
.ok_or_else(|| {
FeatureInstallError::InvalidDescriptor(
"worker feature requires a Workspace id".to_string(),
)
})?
.to_string();
})?;
let lifecycle: Arc<dyn WorkerLifecycleService> =
Arc::new(WorkspaceWorkerLifecycleService {
client: self.client.clone(),
workspace_id: workspace_id.clone(),
});
context.services().provide(
ServiceDeclaration::new(
@@ -948,14 +954,7 @@ impl Tool for WorkspaceWorkerTool {
let input = parse::<WorkerTargetInput>(input_json, "WorkerRestore")?;
let (runtime_id, worker_id) = runtime_subject_ids(&input.subject, self.operation)?;
self.control
.execute_runtime(WorkspaceRequest::json(
WorkspaceRequestMethod::Post,
format!(
"/api/w/{}/worker-control/workers/{runtime_id}/{worker_id}/restore",
self.control.workspace_id()
),
"{}",
))
.restore_worker(runtime_id, worker_id)
.await
.map_err(control_tool_error)?
}
@@ -1191,7 +1190,7 @@ mod tests {
use std::sync::Mutex;
use super::*;
use crate::worker::{WorkspaceClientError, WorkspaceResponse};
use crate::worker::{WorkspaceClientError, WorkspaceRequest, WorkspaceResponse};
#[derive(Debug, Default)]
struct RecordingWorkspaceClient {
@@ -1361,15 +1360,6 @@ mod tests {
))
}
async fn execute_runtime(
&self,
_request: WorkspaceRequest,
) -> Result<WorkspaceResponse, WorkspaceClientError> {
Err(WorkspaceClientError::Unavailable(
"Runtime Worker control is disabled in this test".to_string(),
))
}
async fn ensure_permission(
&self,
_subject: &crate::feature::builtin::WorkerObservationSubjectRef,
crates/worker/src/feature/builtin/worker_observation.rs
+20 -13
View File
@@ -125,17 +125,16 @@ impl WorkerObservationProvider for WorkspaceClientWorkerObservationProvider {
async fn list_worker_sessions(
&self,
) -> Result<Vec<WorkerObservationSubject>, WorkerObservationError> {
let workspace_id = self.client.workspace_id().ok_or_else(|| {
let _workspace_id = self.client.workspace_id().ok_or_else(|| {
WorkerObservationError::Unavailable(
"Workspace observation requires a scoped Workspace client".to_string(),
)
})?;
let response = self
.client
.execute(crate::worker::WorkspaceRequest::get(format!(
"/api/w/{}/worker-observation/sessions",
workspace_id
)))
.execute_server_operation(
crate::worker::WorkspaceServerOperation::WorkerObservationSessions,
)
.map_err(workspace_client_error)?;
let body = workspace_response_body(response)?;
serde_json::from_str::<WorkspaceWorkerObservationListResponse>(&body)
@@ -147,20 +146,28 @@ impl WorkerObservationProvider for WorkspaceClientWorkerObservationProvider {
&self,
subject: &WorkerObservationSubjectRef,
) -> Result<WorkerSessionCapture, WorkerObservationError> {
let body = serde_json::to_string(subject)
.map_err(|error| WorkerObservationError::Unavailable(error.to_string()))?;
let workspace_id = self.client.workspace_id().ok_or_else(|| {
let request = match subject {
WorkerObservationSubjectRef::RuntimeWorker {
runtime_id,
worker_id,
} => server_api::WorkerObservationSubjectRef::RuntimeWorker {
runtime_id: runtime_id.clone(),
worker_id: worker_id.clone(),
},
WorkerObservationSubjectRef::SubWorker { name } => {
server_api::WorkerObservationSubjectRef::SubWorker { name: name.clone() }
}
};
let _workspace_id = self.client.workspace_id().ok_or_else(|| {
WorkerObservationError::Unavailable(
"Workspace observation requires a scoped Workspace client".to_string(),
)
})?;
let response = self
.client
.execute(crate::worker::WorkspaceRequest::json(
crate::worker::WorkspaceRequestMethod::Post,
format!("/api/w/{}/worker-observation/session", workspace_id),
body,
))
.execute_server_operation(
crate::worker::WorkspaceServerOperation::WorkerObservationCapture(request),
)
.map_err(workspace_client_error)?;
let body = workspace_response_body(response)?;
let response = serde_json::from_str::<WorkspaceWorkerObservationCaptureResponse>(&body)
crates/worker/src/lib.rs
+2 -2
View File
@@ -60,6 +60,6 @@ pub use worker::{
LocalWorkingDirectory, Worker, WorkerError, WorkerFilesystemAuthority, WorkerRunResult,
WorkerWorkspaceContext, WorkspaceClient, WorkspaceClientError, WorkspaceId, WorkspaceIdError,
WorkspacePromptCatalogResolution, WorkspaceRequest, WorkspaceRequestMethod, WorkspaceResponse,
WorkspaceWorkerDiscoveryRequest, apply_worker_manifest, marker_workspace_client,
unavailable_workspace_client,
WorkspaceServerOperation, WorkspaceWorkerDiscoveryRequest, apply_worker_manifest,
marker_workspace_client, unavailable_workspace_client,
};
crates/worker/src/worker.rs
+104
View File
@@ -544,6 +544,98 @@ pub struct WorkspaceWorkerDiscoveryRequest {
pub query: Option<String>,
}
#[derive(Debug, Clone)]
pub enum WorkspaceServerOperation {
WorkerControlList,
WorkerControlSpawn(server_api::CreateWorkspaceWorkerRequest),
WorkerControlInput {
runtime_id: String,
worker_id: String,
request: server_api::RuntimeWorkerInputRequest,
},
WorkerControlCancel {
runtime_id: String,
worker_id: String,
request: server_api::RuntimeWorkerLifecycleRequest,
},
WorkerControlStop {
runtime_id: String,
worker_id: String,
request: server_api::RuntimeWorkerLifecycleRequest,
},
WorkerControlRestore {
runtime_id: String,
worker_id: String,
},
WorkerObservationSessions,
WorkerObservationCapture(server_api::WorkerObservationSubjectRef),
}
fn workspace_server_json_request<T: serde::Serialize>(
path: String,
body: &T,
) -> Result<WorkspaceRequest, WorkspaceClientError> {
Ok(WorkspaceRequest::json(
WorkspaceRequestMethod::Post,
path,
serde_json::to_string(body)
.map_err(|error| WorkspaceClientError::Request(error.to_string()))?,
))
}
fn workspace_server_operation_request(
workspace_id: &str,
operation: WorkspaceServerOperation,
) -> Result<WorkspaceRequest, WorkspaceClientError> {
let base = format!("/api/w/{workspace_id}");
match operation {
WorkspaceServerOperation::WorkerControlList => Ok(WorkspaceRequest::get(format!(
"{base}/worker-control/workers"
))),
WorkspaceServerOperation::WorkerControlSpawn(request) => {
workspace_server_json_request(format!("{base}/worker-control/workers"), &request)
}
WorkspaceServerOperation::WorkerControlInput {
runtime_id,
worker_id,
request,
} => workspace_server_json_request(
format!("{base}/worker-control/workers/{runtime_id}/{worker_id}/input"),
&request,
),
WorkspaceServerOperation::WorkerControlCancel {
runtime_id,
worker_id,
request,
} => workspace_server_json_request(
format!("{base}/worker-control/workers/{runtime_id}/{worker_id}/cancel"),
&request,
),
WorkspaceServerOperation::WorkerControlStop {
runtime_id,
worker_id,
request,
} => workspace_server_json_request(
format!("{base}/worker-control/workers/{runtime_id}/{worker_id}/stop"),
&request,
),
WorkspaceServerOperation::WorkerControlRestore {
runtime_id,
worker_id,
} => Ok(WorkspaceRequest::json(
WorkspaceRequestMethod::Post,
format!("{base}/worker-control/workers/{runtime_id}/{worker_id}/restore"),
"".to_string(),
)),
WorkspaceServerOperation::WorkerObservationSessions => Ok(WorkspaceRequest::get(format!(
"{base}/worker-observation/sessions"
))),
WorkspaceServerOperation::WorkerObservationCapture(request) => {
workspace_server_json_request(format!("{base}/worker-observation/session"), &request)
}
}
}
/// Path-free Workspace operation authority injected by Runtime/host code.
///
/// Workers receive this trait object rather than a Backend URL. The concrete
@@ -556,6 +648,18 @@ pub trait WorkspaceClient: std::fmt::Debug + Send + Sync {
fn execute(&self, request: WorkspaceRequest)
-> Result<WorkspaceResponse, WorkspaceClientError>;
fn execute_server_operation(
&self,
operation: WorkspaceServerOperation,
) -> Result<WorkspaceResponse, WorkspaceClientError> {
let workspace_id = self.workspace_id().ok_or_else(|| {
WorkspaceClientError::Unavailable(
"Workspace Server operation requires Workspace identity".to_string(),
)
})?;
self.execute(workspace_server_operation_request(workspace_id, operation)?)
}
/// Lists Workspace-visible Workers through dedicated Runtime-owned source
/// proof. Implementations must not fall back to generic Workspace request
/// authority.
crates/workspace-server/src/server.rs
+834 -287
View File
File diff suppressed because it is too large Load Diff