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

feat: abstract worker runtime spawn boundary

Browse Source
This commit is contained in:
Hare
2026-06-24 19:24:18 +09:00
parent a729d68600
commit 217a4828d7
7 changed files with 341 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/workspace-server/src/hosts.rs
+199
View File
@@ -63,6 +63,118 @@ pub struct WorkerImplementation {
pub pod_name: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct RuntimeList<T> {
pub items: Vec<T>,
pub diagnostics: Vec<RuntimeDiagnostic>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct WorkerLookupResult {
#[serde(skip_serializing_if = "Option::is_none")]
pub worker: Option<WorkerSummary>,
pub diagnostics: Vec<RuntimeDiagnostic>,
}
/// Browser-safe worker spawn request shape.
///
/// The request intentionally carries only workspace policy intents and stable
/// worker identifiers. Raw workspace roots, child cwd, executable path, and raw
/// profile selectors are resolved by the host/runtime service and never accepted
/// from Workspace API callers.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct WorkerSpawnRequest {
pub intent: WorkerSpawnIntent,
#[serde(skip_serializing_if = "Option::is_none")]
pub requested_worker_name: Option<String>,
pub acceptance: WorkerSpawnAcceptanceRequirement,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum WorkerSpawnIntent {
WorkspaceCompanion,
WorkspaceOrchestrator,
TicketRole {
ticket_id: String,
role: TicketWorkerRole,
},
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum TicketWorkerRole {
Intake,
Orchestrator,
Coder,
Reviewer,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum WorkerSpawnAcceptanceRequirement {
SocketReady,
RunAccepted { expected_segments: usize },
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct WorkerSpawnResult {
pub state: WorkerOperationState,
#[serde(skip_serializing_if = "Option::is_none")]
pub worker: Option<WorkerSummary>,
pub acceptance_evidence: Vec<WorkerSpawnAcceptanceEvidence>,
pub diagnostics: Vec<RuntimeDiagnostic>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum WorkerOperationState {
Accepted,
Unsupported,
Rejected,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct WorkerSpawnAcceptanceEvidence {
pub kind: String,
pub detail: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct WorkerStopRequest {
pub worker_id: String,
pub mode: WorkerStopMode,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum WorkerStopMode {
Graceful,
Force,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct WorkerStopResult {
pub state: WorkerOperationState,
pub diagnostics: Vec<RuntimeDiagnostic>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct WorkerProxyConnectPoint {
pub kind: String,
pub status: String,
pub diagnostics: Vec<RuntimeDiagnostic>,
}
pub trait WorkspaceWorkerRuntime: Send + Sync {
fn list_hosts(&self, limit: usize) -> RuntimeList<HostSummary>;
fn list_workers(&self, limit: usize) -> RuntimeList<WorkerSummary>;
fn worker(&self, worker_id: &str) -> WorkerLookupResult;
fn spawn_worker(&self, request: WorkerSpawnRequest) -> WorkerSpawnResult;
fn stop_worker(&self, request: WorkerStopRequest) -> WorkerStopResult;
fn proxy_connect_points(&self, worker_id: &str) -> Vec<WorkerProxyConnectPoint>;
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct LocalRuntimeBridge {
workspace_id: String,
@@ -247,6 +359,85 @@ impl LocalRuntimeBridge {
}
}
impl WorkspaceWorkerRuntime for LocalRuntimeBridge {
fn list_hosts(&self, limit: usize) -> RuntimeList<HostSummary> {
let (items, diagnostics) = LocalRuntimeBridge::list_hosts(self, limit);
RuntimeList { items, diagnostics }
}
fn list_workers(&self, limit: usize) -> RuntimeList<WorkerSummary> {
let (items, diagnostics) = LocalRuntimeBridge::list_workers(self, limit);
RuntimeList { items, diagnostics }
}
fn worker(&self, worker_id: &str) -> WorkerLookupResult {
let RuntimeList {
items,
mut diagnostics,
} = WorkspaceWorkerRuntime::list_workers(self, 200);
let worker = items
.into_iter()
.find(|worker| worker.worker_id == worker_id);
if worker.is_none() {
diagnostics.push(RuntimeDiagnostic::new(
"worker_not_found",
"info",
format!("worker '{worker_id}' was not found on the local runtime"),
));
}
truncate_diagnostics(&mut diagnostics);
WorkerLookupResult {
worker,
diagnostics,
}
}
fn spawn_worker(&self, request: WorkerSpawnRequest) -> WorkerSpawnResult {
let diagnostic = RuntimeDiagnostic::new(
"worker_spawn_resolver_pending",
"info",
format!(
"worker spawn intent '{}' was accepted as a typed request shape, but local launch resolution is not implemented by this ticket",
worker_spawn_intent_label(&request.intent)
),
);
WorkerSpawnResult {
state: WorkerOperationState::Unsupported,
worker: None,
acceptance_evidence: Vec::new(),
diagnostics: vec![diagnostic],
}
}
fn stop_worker(&self, request: WorkerStopRequest) -> WorkerStopResult {
WorkerStopResult {
state: WorkerOperationState::Unsupported,
diagnostics: vec![RuntimeDiagnostic::new(
"worker_stop_pending",
"info",
format!(
"worker stop for '{}' is reserved for the runtime service boundary and is not implemented by this ticket",
request.worker_id
),
)],
}
}
fn proxy_connect_points(&self, worker_id: &str) -> Vec<WorkerProxyConnectPoint> {
vec![WorkerProxyConnectPoint {
kind: "stream_proxy".to_string(),
status: "not_implemented".to_string(),
diagnostics: vec![RuntimeDiagnostic::new(
"worker_stream_proxy_pending",
"info",
format!(
"future stream/proxy connection point for '{worker_id}' is reserved without opening a protocol surface"
),
)],
}]
}
}
impl RuntimeDiagnostic {
pub fn new(
code: impl Into<String>,
@@ -393,6 +584,14 @@ fn safe_metadata_label(value: &str) -> Option<String> {
Some(value.to_string())
}
fn worker_spawn_intent_label(intent: &WorkerSpawnIntent) -> &'static str {
match intent {
WorkerSpawnIntent::WorkspaceCompanion => "workspace_companion",
WorkerSpawnIntent::WorkspaceOrchestrator => "workspace_orchestrator",
WorkerSpawnIntent::TicketRole { .. } => "ticket_role",
}
}
fn stable_local_host_id(workspace_id: &str) -> String {
format!("local-{}", sanitize_identifier(workspace_id, 96))
}
crates/workspace-server/src/server.rs
+25 -21
View File
@@ -10,7 +10,9 @@ use axum::{Json, Router};
use serde::{Deserialize, Serialize};
use tokio::net::TcpListener;
use crate::hosts::{HostSummary, LocalRuntimeBridge, RuntimeDiagnostic, WorkerSummary};
use crate::hosts::{
HostSummary, LocalRuntimeBridge, RuntimeDiagnostic, WorkerSummary, WorkspaceWorkerRuntime,
};
use crate::identity::WorkspaceIdentity;
use crate::records::{
LocalProjectRecordReader, ObjectiveDetail, ProjectRecordList, TicketDetail, TicketSummary,
@@ -61,6 +63,7 @@ pub struct WorkspaceApi {
config: ServerConfig,
store: Arc<dyn ControlPlaneStore>,
records: LocalProjectRecordReader,
runtime: Arc<dyn WorkspaceWorkerRuntime>,
}
impl WorkspaceApi {
@@ -74,10 +77,16 @@ impl WorkspaceApi {
updated_at: config.workspace_created_at.clone(),
})
.await?;
let runtime = Arc::new(LocalRuntimeBridge::new(
config.workspace_id.clone(),
config.workspace_root.clone(),
config.local_runtime_data_dir.clone(),
));
Ok(Self {
records: LocalProjectRecordReader::new(config.workspace_root.clone()),
config,
store,
runtime,
})
}
@@ -85,14 +94,6 @@ impl WorkspaceApi {
self.config.workspace_id.as_str()
}
fn local_runtime_bridge(&self) -> LocalRuntimeBridge {
LocalRuntimeBridge::new(
self.config.workspace_id.clone(),
self.config.workspace_root.clone(),
self.config.local_runtime_data_dir.clone(),
)
}
fn local_repository_reader(&self) -> LocalRepositoryReader {
LocalRepositoryReader::new(
self.config.workspace_root.clone(),
@@ -390,14 +391,13 @@ async fn list_hosts(
State(api): State<WorkspaceApi>,
) -> ApiResult<Json<RuntimeListResponse<HostSummary>>> {
let limit = api.config.max_records.min(200);
let bridge = api.local_runtime_bridge();
let (items, diagnostics) = bridge.list_hosts(limit);
let runtime_hosts = api.runtime.list_hosts(limit);
Ok(Json(RuntimeListResponse {
workspace_id: api.config.workspace_id,
limit,
items,
source: "local_pod_metadata".to_string(),
diagnostics,
items: runtime_hosts.items,
source: "worker_runtime".to_string(),
diagnostics: runtime_hosts.diagnostics,
}))
}
@@ -411,8 +411,13 @@ async fn list_host_workers(
State(api): State<WorkspaceApi>,
AxumPath(host_id): AxumPath<String>,
) -> ApiResult<Json<RuntimeListResponse<WorkerSummary>>> {
let bridge = api.local_runtime_bridge();
if host_id != bridge.host_id() {
let runtime_hosts = api.runtime.list_hosts(1);
let expected_host_id = runtime_hosts
.items
.first()
.map(|host| host.host_id.as_str())
.ok_or_else(|| Error::UnknownHost(host_id.clone()))?;
if host_id != expected_host_id {
return Err(Error::UnknownHost(host_id).into());
}
workers_response(api).map(Json)
@@ -420,14 +425,13 @@ async fn list_host_workers(
fn workers_response(api: WorkspaceApi) -> ApiResult<RuntimeListResponse<WorkerSummary>> {
let limit = api.config.max_records.min(200);
let bridge = api.local_runtime_bridge();
let (items, diagnostics) = bridge.list_workers(limit);
let runtime_workers = api.runtime.list_workers(limit);
Ok(RuntimeListResponse {
workspace_id: api.config.workspace_id,
limit,
items,
source: "local_pod_metadata".to_string(),
diagnostics,
items: runtime_workers.items,
source: "worker_runtime".to_string(),
diagnostics: runtime_workers.diagnostics,
})
}