diff --git a/.yoi/tickets/00001KY1ZKCPC/item.md b/.yoi/tickets/00001KY1ZKCPC/item.md index 14f40f7b..0b8a3f3a 100644 --- a/.yoi/tickets/00001KY1ZKCPC/item.md +++ b/.yoi/tickets/00001KY1ZKCPC/item.md @@ -2,7 +2,7 @@ title: 'Backend runtime経由の操作をprotocol transportへ統一しTUI同等にする' state: 'planning' created_at: '2026-07-21T09:20:07Z' -updated_at: '2026-07-21T09:47:10Z' +updated_at: '2026-07-21T10:33:12Z' assignee: null readiness: 'draft' --- diff --git a/.yoi/tickets/00001KY1ZKCPC/thread.md b/.yoi/tickets/00001KY1ZKCPC/thread.md index a4b665dd..8ab81de0 100644 --- a/.yoi/tickets/00001KY1ZKCPC/thread.md +++ b/.yoi/tickets/00001KY1ZKCPC/thread.md @@ -21,3 +21,77 @@ Ticket item updated: title, body. Ticket item updated: title, body. --- + + + +## Plan + +Implementation started. + +Plan for this pass: +- Inspect local Worker protocol transport and current Backend/frontend HTTP operation mappings. +- Add a Backend/runtime protocol command transport that accepts existing `protocol::Method` payloads rather than per-operation HTTP endpoints. +- Route TUI Backend Runtime client through that protocol transport for run/notify/completion/cancel/pause/resume/compact/rewind operations. +- Route browser Console operations through the same protocol transport and stop using HTTP operation endpoints as the normal path. +- Keep existing HTTP endpoints only as compatibility where needed, not as frontend/TUI authority. + +--- + + + +## Implementation report + +Implementation update: + +Implemented the first full protocol-transport pass for Backend/runtime Worker controls. + +Backend/runtime: +- Added `Runtime::send_protocol_method(...)` in `worker-runtime`. + - `Method::ListCompletions` returns direct `Event::Completions` for same-command-connection semantics. + - Other methods are delivered unchanged to the execution backend. +- Added `WorkerExecutionBackend::dispatch_method(...)` and concrete support in `WorkerRuntimeExecutionBackend`. + - This accepts existing `protocol::Method` values instead of translating through operation-specific HTTP request DTOs. +- Added runtime command WebSocket: + - `/v1/workers/{worker_id}/protocol/ws` + - accepts raw `protocol::Method` JSON frames + - returns raw direct `protocol::Event` JSON frames where needed +- Added workspace-server command WebSocket: + - `/api/runtimes/{runtime_id}/workers/{worker_id}/protocol/ws` + - `/api/w/{workspace_id}/runtimes/{runtime_id}/workers/{worker_id}/protocol/ws` +- Added `RuntimeRegistry::send_protocol_method(...)` and embedded runtime forwarding. + +TUI client: +- Reworked `BackendRuntimeClient` so `send(&Method)` no longer maps to HTTP `/input`, lifecycle, or unsupported diagnostics. +- It now opens a command WS to `/protocol/ws` and sends cloned raw `protocol::Method` values. +- Direct command response events are merged into the existing event queue alongside observation WS events. +- This removes the previous unsupported handling for pause/resume/compact/completions/rewind at the TUI client transport layer. + +Browser Console: +- Added command WS connection to `/protocol/ws` while keeping `/events/ws` as observation stream. +- Removed normal operation use of HTTP `POST /input` and `POST /completions` from the Console page. +- Composer send now maps to protocol methods: + - user input -> `Method::Run` with protocol `Segment[]` + - system input -> `Method::Notify { auto_run: true }` + - `:compact` -> `Method::Compact` + - `:rewind` -> `Method::ListRewindTargets` + - `:peer` -> `Method::RegisterPeer` +- Completion lookup now sends `Method::ListCompletions` over command WS and waits for direct `Event::Completions`. +- Added frontend controls for cancel / pause / resume / compact / rewind target request / rewind to target. + +Compatibility note: +- Existing HTTP operation endpoints remain in backend/server as compatibility for now, but the TUI Backend client and browser Console normal path now use protocol transport. + +Verification run under `nix develop`: +- `cargo fmt -- --check` +- `cargo check -p worker-runtime -p yoi-workspace-server -p client -p tui` +- `cargo test -p client backend_runtime --lib` +- `cargo test -p yoi-workspace-server registry_worker_list_can_be_scoped_by_runtime_id --lib` +- `cargo test -p worker-runtime --features ws-server --lib worker_backend` +- `git diff --check` + +Frontend check note: +- `web/workspace/package.json` currently has no check script and local frontend dependencies are not installed in this worktree. +- Attempted temporary `npm exec --package svelte-check --package typescript -- svelte-check`, but the tool bootstrap failed before checking project code. +- The Svelte page was updated against the generated protocol types in `web/workspace/src/lib/generated/protocol.ts`, but a frontend-native check still needs to be run in an environment with the frontend deps installed. + +--- diff --git a/crates/client/src/backend_runtime.rs b/crates/client/src/backend_runtime.rs index 7fda54cf..e4762109 100644 --- a/crates/client/src/backend_runtime.rs +++ b/crates/client/src/backend_runtime.rs @@ -2,9 +2,9 @@ use std::collections::VecDeque; use std::fmt; use std::time::Duration; -use futures::StreamExt; -use protocol::{ErrorCode, Event, Method, Segment}; -use serde::{Deserialize, Serialize}; +use futures::{SinkExt, StreamExt}; +use protocol::{ErrorCode, Event, Method}; +use serde::Deserialize; use tokio::sync::mpsc; use tokio_tungstenite::connect_async; use tokio_tungstenite::tungstenite::Message as TungsteniteMessage; @@ -160,10 +160,11 @@ pub struct BackendWorkerSummary { #[derive(Debug)] pub struct BackendRuntimeClient { target: BackendRuntimeTarget, - http: reqwest::Client, + command_tx: mpsc::UnboundedSender, events: mpsc::UnboundedReceiver, diagnostics: VecDeque, _observation_task: tokio::task::JoinHandle<()>, + _command_task: tokio::task::JoinHandle<()>, } #[derive(Debug)] @@ -258,21 +259,28 @@ pub async fn list_backend_workers( impl BackendRuntimeClient { pub async fn connect(target: BackendRuntimeTarget) -> Result { validate_target(&target)?; - let http = reqwest::Client::new(); - let (tx, rx) = mpsc::unbounded_channel(); + let (event_tx, rx) = mpsc::unbounded_channel(); + let (command_tx, command_rx) = mpsc::unbounded_channel(); let observation_target = target.clone(); - let observation_tx = tx.clone(); + let observation_tx = event_tx.clone(); let observation_task = tokio::spawn(async move { observe_worker_events(observation_target, observation_tx).await; }); + let command_target = target.clone(); + let command_event_tx = event_tx.clone(); + let command_task = tokio::spawn(async move { + run_worker_protocol_commands(command_target, command_rx, command_event_tx).await; + }); + Ok(Self { target, - http, + command_tx, events: rx, diagnostics: VecDeque::new(), _observation_task: observation_task, + _command_task: command_task, }) } @@ -291,163 +299,106 @@ impl BackendRuntimeClient { } pub async fn send(&mut self, method: &Method) -> Result<(), BackendRuntimeClientError> { - match backend_command_from_method(method) { - BackendCommand::Input { kind, content } => { - let url = self.worker_api_url("input"); - match self - .http - .post(url) - .json(&WorkerInputRequest { kind, content }) - .send() - .await - .and_then(|response| response.error_for_status()) - { - Ok(response) => match response.json::().await { - Ok(result) => self.enqueue_operation_diagnostics( - "input", - result.state, - result.diagnostics, - ), - Err(error) => self.enqueue_diagnostic(format!( - "Backend runtime input response could not be decoded for {}: {error}", - self.target.display_label() - )), - }, - Err(error) => self.enqueue_diagnostic(format!( - "Backend runtime input failed for {}: {error}", - self.target.display_label() - )), - } - } - BackendCommand::Lifecycle { action, reason } => { - let url = self.worker_api_url(action); - match self - .http - .post(url) - .json(&WorkerLifecycleRequest { reason }) - .send() - .await - .and_then(|response| response.error_for_status()) - { - Ok(response) => match response.json::().await { - Ok(result) => self.enqueue_operation_diagnostics( - action, - result.state, - result.diagnostics, - ), - Err(error) => self.enqueue_diagnostic(format!( - "Backend runtime {action} response could not be decoded for {}: {error}", - self.target.display_label() - )), - }, - Err(error) => self.enqueue_diagnostic(format!( - "Backend runtime {action} failed for {}: {error}", - self.target.display_label() - )), - } - } - BackendCommand::Unsupported(message) => { - self.enqueue_diagnostic(message); - } - } - Ok(()) - } - - fn worker_api_url(&self, suffix: &str) -> String { - let path = format!( - "/api/runtimes/{}/workers/{}/{}", - path_segment_encode(&self.target.runtime_id), - path_segment_encode(&self.target.worker_id), - suffix - ); - join_base_and_path(&self.target.base_url, &path) - } - - fn enqueue_operation_diagnostics( - &mut self, - operation: &str, - state: String, - diagnostics: Vec, - ) { - if state != "accepted" { - self.enqueue_diagnostic(format!( - "Backend runtime {operation} was {state} for {}", + self.command_tx.send(method.clone()).map_err(|_| { + BackendRuntimeClientError::InvalidTarget(format!( + "Backend protocol command stream is closed for {}", self.target.display_label() - )); - } - for diagnostic in diagnostics { - self.enqueue_diagnostic(format!( - "Backend runtime {operation} diagnostic [{}]: {}", - diagnostic.code, diagnostic.message - )); - } - } - - fn enqueue_diagnostic(&mut self, message: impl Into) { - self.diagnostics.push_back(diagnostic_event(message)); + )) + })?; + Ok(()) } } impl Drop for BackendRuntimeClient { fn drop(&mut self) { self._observation_task.abort(); + self._command_task.abort(); } } -#[derive(Debug, PartialEq, Eq)] -enum BackendCommand { - Input { - kind: WorkerInputKind, - content: String, - }, - Lifecycle { - action: &'static str, - reason: Option, - }, - Unsupported(String), -} - -fn backend_command_from_method(method: &Method) -> BackendCommand { - match method { - Method::Run { input } => BackendCommand::Input { - kind: WorkerInputKind::User, - content: Segment::flatten_to_text(input), - }, - Method::Notify { message, .. } => BackendCommand::Input { - kind: WorkerInputKind::System, - content: message.clone(), - }, - Method::Cancel => BackendCommand::Lifecycle { - action: "cancel", - reason: Some("requested from TUI Backend Runtime API client".to_string()), - }, - Method::Shutdown => BackendCommand::Lifecycle { - action: "stop", - reason: Some("requested from TUI Backend Runtime API client".to_string()), - }, - Method::Pause => BackendCommand::Unsupported( - "Backend Runtime API does not expose pause/resume for the TUI client yet; command was not sent".to_string(), - ), - Method::Resume => BackendCommand::Unsupported( - "Backend Runtime API does not expose resume for the TUI client yet; command was not sent".to_string(), - ), - Method::Compact => BackendCommand::Unsupported( - "Backend Runtime API does not expose compaction for the TUI client yet; command was not sent".to_string(), - ), - Method::ListCompletions { .. } => BackendCommand::Unsupported( - "Backend Runtime API does not expose completion lookup for the TUI client yet".to_string(), - ), - Method::ListRewindTargets | Method::RewindTo { .. } => BackendCommand::Unsupported( - "Backend Runtime API does not expose rewind controls for the TUI client yet; command was not sent".to_string(), - ), - Method::ListWorkers | Method::RestoreWorker { .. } | Method::RegisterPeer { .. } => { - BackendCommand::Unsupported( - "Backend Runtime API worker-management controls are not available from this Console connection".to_string(), - ) +async fn run_worker_protocol_commands( + target: BackendRuntimeTarget, + mut commands: mpsc::UnboundedReceiver, + tx: mpsc::UnboundedSender, +) { + let url = protocol_ws_url(&target); + match connect_async(&url).await { + Ok((ws, _)) => { + let (mut sink, mut stream) = ws.split(); + loop { + tokio::select! { + maybe_method = commands.recv() => { + let Some(method) = maybe_method else { + break; + }; + match serde_json::to_string(&method) { + Ok(text) => { + if let Err(error) = sink.send(TungsteniteMessage::Text(text.into())).await { + let _ = tx.send(diagnostic_event(format!( + "Backend protocol command send failed for {}: {error}", + target.display_label() + ))); + break; + } + } + Err(error) => { + let _ = tx.send(diagnostic_event(format!( + "Backend protocol command could not serialize method for {}: {error}", + target.display_label() + ))); + } + } + } + frame = stream.next() => { + match frame { + Some(Ok(TungsteniteMessage::Text(text))) => { + match serde_json::from_str::(&text) { + Ok(event) => { + let _ = tx.send(event); + } + Err(error) => { + let _ = tx.send(diagnostic_event(format!( + "Backend protocol response was not valid Event JSON for {}: {error}", + target.display_label() + ))); + } + } + } + Some(Ok(TungsteniteMessage::Close(_))) | None => { + let _ = tx.send(diagnostic_event(format!( + "Backend protocol command stream closed for {}", + target.display_label() + ))); + break; + } + Some(Ok(TungsteniteMessage::Ping(_))) + | Some(Ok(TungsteniteMessage::Pong(_))) + | Some(Ok(TungsteniteMessage::Binary(_))) + | Some(Ok(TungsteniteMessage::Frame(_))) => {} + Some(Err(error)) => { + let _ = tx.send(diagnostic_event(format!( + "Backend protocol WebSocket error for {}: {error}", + target.display_label() + ))); + break; + } + } + } + } + } + } + Err(error) => { + let _ = tx.send(diagnostic_event(format!( + "Backend protocol WebSocket connect failed for {}: {error}", + target.display_label() + ))); + while commands.recv().await.is_some() { + let _ = tx.send(diagnostic_event(format!( + "Backend protocol command was not sent because command stream is unavailable for {}", + target.display_label() + ))); + } } - Method::WorkerEvent(_) => BackendCommand::Unsupported( - "Backend Runtime API does not accept child Worker lifecycle events from this Console connection".to_string(), - ), } } @@ -610,6 +561,15 @@ fn observation_ws_url(target: &BackendRuntimeTarget) -> String { join_base_and_path(&http_base_to_ws(&target.base_url), &path) } +fn protocol_ws_url(target: &BackendRuntimeTarget) -> String { + let path = format!( + "/api/runtimes/{}/workers/{}/protocol/ws", + path_segment_encode(&target.runtime_id), + path_segment_encode(&target.worker_id) + ); + join_base_and_path(&http_base_to_ws(&target.base_url), &path) +} + fn http_base_to_ws(base: &str) -> String { if let Some(rest) = base.strip_prefix("https://") { format!("wss://{rest}") @@ -643,38 +603,6 @@ fn percent_encode(input: &str, keep: impl Fn(u8) -> bool) -> String { encoded } -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] -#[serde(rename_all = "snake_case")] -enum WorkerInputKind { - User, - System, -} - -#[derive(Debug, Serialize)] -struct WorkerInputRequest { - kind: WorkerInputKind, - content: String, -} - -#[derive(Debug, Serialize)] -struct WorkerLifecycleRequest { - reason: Option, -} - -#[derive(Debug, Deserialize)] -struct WorkerInputResult { - state: String, - #[serde(default)] - diagnostics: Vec, -} - -#[derive(Debug, Deserialize)] -struct WorkerLifecycleResult { - state: String, - #[serde(default)] - diagnostics: Vec, -} - #[derive(Debug, Clone, Deserialize, PartialEq, Eq)] pub struct BackendDiagnostic { pub code: String, @@ -712,47 +640,16 @@ mod tests { use super::*; #[test] - fn backend_command_maps_run_to_user_input_without_runtime_endpoint() { - let method = Method::Run { - input: vec![ - Segment::text("hello"), - Segment::FileRef { - path: "src/lib.rs".into(), - }, - ], - }; - assert_eq!( - backend_command_from_method(&method), - BackendCommand::Input { - kind: WorkerInputKind::User, - content: "hello@src/lib.rs".to_string(), - } - ); - } - - #[test] - fn backend_worker_list_paths_use_scoped_workspace_when_available() { - assert_eq!( - backend_runtimes_path(Some("workspace/one")), - "/api/w/workspace%2Fone/runtimes" - ); - assert_eq!( - backend_runtime_workers_path(Some("workspace/one"), "runtime one"), - "/api/w/workspace%2Fone/runtimes/runtime%20one/workers" - ); - assert_eq!( - backend_runtime_workers_path(None, "runtime one"), - "/api/runtimes/runtime%20one/workers" - ); - } - - #[test] - fn observation_url_uses_backend_runtime_worker_identity() { + fn command_and_observation_urls_use_backend_protocol_paths() { let target = BackendRuntimeTarget::new("http://127.0.0.1:8787/", "runtime/one", "worker one"); assert_eq!( observation_ws_url(&target), "ws://127.0.0.1:8787/api/runtimes/runtime%2Fone/workers/worker%20one/events/ws" ); + assert_eq!( + protocol_ws_url(&target), + "ws://127.0.0.1:8787/api/runtimes/runtime%2Fone/workers/worker%20one/protocol/ws" + ); } } diff --git a/crates/worker-runtime/src/execution.rs b/crates/worker-runtime/src/execution.rs index d655e54b..74dac542 100644 --- a/crates/worker-runtime/src/execution.rs +++ b/crates/worker-runtime/src/execution.rs @@ -6,6 +6,7 @@ use crate::interaction::WorkerInput; #[cfg(feature = "ws-server")] use crate::observation::WorkerObservationEvent; use crate::working_directory::{WorkingDirectoryBinding, WorkingDirectoryDiagnostic}; +use protocol::Method; use serde::{Deserialize, Serialize}; use std::fmt; use std::sync::Arc; @@ -63,6 +64,7 @@ pub enum WorkerExecutionOperation { Spawn, Restore, Input, + ProtocolMethod, Stop, Cancel, } @@ -380,6 +382,17 @@ pub trait WorkerExecutionBackend: Send + Sync + 'static { input: WorkerInput, ) -> WorkerExecutionResult; + fn dispatch_method( + &self, + _handle: &WorkerExecutionHandle, + _method: Method, + ) -> WorkerExecutionResult { + WorkerExecutionResult::unsupported( + WorkerExecutionOperation::ProtocolMethod, + "execution backend does not support direct Worker protocol methods", + ) + } + fn stop_worker(&self, _handle: &WorkerExecutionHandle) -> WorkerExecutionResult { WorkerExecutionResult::unsupported( WorkerExecutionOperation::Stop, @@ -479,6 +492,14 @@ impl WorkerExecutionBackendRef { self.backend.dispatch_input(handle, input) } + pub(crate) fn dispatch_method( + &self, + handle: &WorkerExecutionHandle, + method: Method, + ) -> WorkerExecutionResult { + self.backend.dispatch_method(handle, method) + } + pub(crate) fn stop_worker(&self, handle: &WorkerExecutionHandle) -> WorkerExecutionResult { self.backend.stop_worker(handle) } diff --git a/crates/worker-runtime/src/http_server.rs b/crates/worker-runtime/src/http_server.rs index 6d8a5e23..cc49d3a2 100644 --- a/crates/worker-runtime/src/http_server.rs +++ b/crates/worker-runtime/src/http_server.rs @@ -154,7 +154,12 @@ pub fn runtime_http_router(runtime: Runtime, local_token: Option) -> Rou .route("/v1/workers/{worker_id}/cancel", post(cancel_worker)); #[cfg(feature = "ws-server")] - let router = router.route("/v1/workers/{worker_id}/events/ws", get(worker_events_ws)); + let router = router + .route("/v1/workers/{worker_id}/events/ws", get(worker_events_ws)) + .route( + "/v1/workers/{worker_id}/protocol/ws", + get(worker_protocol_ws), + ); router .with_state(state.clone()) @@ -506,6 +511,94 @@ async fn create_worker( Ok(Json(RuntimeHttpWorkerResponse { worker })) } +#[cfg(feature = "ws-server")] +async fn worker_protocol_ws( + State(state): State, + Path(worker_id): Path, + ws: WebSocketUpgrade, +) -> Result { + let worker_ref = worker_ref_for(&state.runtime, worker_id)?; + state + .runtime + .worker_detail(&worker_ref) + .map_err(RuntimeHttpRestError::runtime)?; + Ok(ws + .on_upgrade(move |socket| worker_protocol_ws_session(state.runtime, worker_ref, socket)) + .into_response()) +} + +#[cfg(feature = "ws-server")] +async fn worker_protocol_ws_session( + runtime: Runtime, + worker_ref: WorkerRef, + mut socket: WebSocket, +) { + while let Some(frame) = socket.next().await { + match frame { + Ok(WsMessage::Text(text)) => match serde_json::from_str::(&text) { + Ok(method) => match runtime.send_protocol_method(&worker_ref, method) { + Ok(events) => { + for event in events { + if !send_protocol_event(&mut socket, &event).await { + return; + } + } + } + Err(error) => { + let event = protocol_error_event(error.to_string()); + if !send_protocol_event(&mut socket, &event).await { + return; + } + } + }, + Err(error) => { + let event = + protocol_error_event(format!("malformed protocol method frame: {error}")); + if !send_protocol_event(&mut socket, &event).await { + return; + } + } + }, + Ok(WsMessage::Close(_)) => return, + Ok(WsMessage::Ping(payload)) => { + if socket.send(WsMessage::Pong(payload)).await.is_err() { + return; + } + } + Ok(WsMessage::Pong(_)) | Ok(WsMessage::Binary(_)) => {} + Err(error) => { + let event = protocol_error_event(format!("protocol WebSocket error: {error}")); + let _ = send_protocol_event(&mut socket, &event).await; + return; + } + } + } +} + +#[cfg(feature = "ws-server")] +async fn send_protocol_event(socket: &mut WebSocket, event: &protocol::Event) -> bool { + match serde_json::to_string(event) { + Ok(text) => socket.send(WsMessage::Text(text.into())).await.is_ok(), + Err(error) => { + let fallback = protocol_error_event(format!( + "failed to serialize protocol response event: {error}" + )); + let Ok(text) = serde_json::to_string(&fallback) else { + return false; + }; + socket.send(WsMessage::Text(text.into())).await.is_ok() + } + } +} + +#[cfg(feature = "ws-server")] +fn protocol_error_event(message: impl Into) -> protocol::Event { + protocol::Event::Error { + code: protocol::ErrorCode::Internal, + message: message.into(), + } +} + #[cfg(feature = "ws-server")] async fn worker_events_ws( State(state): State, diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index 9bd05f32..5c9bc38b 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -35,6 +35,7 @@ use crate::observation::{ }; #[cfg(feature = "ws-server")] use crate::observation::{WorkerObservationCursor, WorkerObservationEvent}; +use protocol::{Event, Method}; use std::collections::BTreeMap; #[cfg(feature = "ws-server")] use std::collections::VecDeque; @@ -559,6 +560,71 @@ impl Runtime { Ok(backend.worker_completions(&handle, kind, prefix)) } + /// Accept a protocol method for a Worker through a Backend/runtime transport. + /// + /// Most methods are delivered to the execution backend unchanged. Methods with + /// direct same-connection replies in the local socket protocol return those + /// events from this function so WebSocket transports can write them back to the + /// requesting client without rebroadcasting them. + pub fn send_protocol_method( + &self, + worker_ref: &WorkerRef, + method: Method, + ) -> Result, RuntimeError> { + if let Method::ListCompletions { kind, prefix } = method { + let entries = self.worker_completions(worker_ref, kind, &prefix)?; + return Ok(vec![Event::Completions { kind, entries }]); + } + + let (backend, handle) = { + let mut state = self.lock()?; + state.ensure_running()?; + state.ensure_worker_ref(worker_ref)?; + let worker = state.worker(worker_ref)?; + if !worker.status.is_active() { + return Err(RuntimeError::InvalidRequest(format!( + "worker {} is not running", + worker_ref.worker_id + ))); + } + let backend = state.execution_backend.clone(); + let handle = worker.execution_handle.clone(); + match (backend, handle) { + (Some(backend), Some(handle)) => (backend, handle), + _ => { + let result = WorkerExecutionResult::rejected( + WorkerExecutionOperation::ProtocolMethod, + "worker has no execution backend", + ); + let worker = state.worker_mut(worker_ref)?; + let mut execution = WorkerExecutionStatus::unconnected().with_result(result); + execution.binding = worker.execution.binding.clone(); + worker.execution = execution; + state.persist_worker(&worker_ref.worker_id)?; + return Err(RuntimeError::WorkerExecutionUnavailable { + worker_id: worker_ref.worker_id.clone(), + message: "worker has no execution backend".to_string(), + }); + } + } + }; + + let dispatch_result = backend.dispatch_method(&handle, method); + if !dispatch_result.is_accepted() { + self.record_execution_result(worker_ref, dispatch_result.clone())?; + return Err(RuntimeError::WorkerExecutionRejected { + worker_id: worker_ref.worker_id.clone(), + operation: dispatch_result.operation, + outcome: dispatch_result.outcome, + message: dispatch_result.message_or_default(), + result: dispatch_result, + }); + } + + self.record_execution_result(worker_ref, dispatch_result)?; + Ok(Vec::new()) + } + fn commit_created_worker( &self, worker_ref: &WorkerRef, @@ -641,7 +707,8 @@ impl Runtime { WorkerExecutionOperation::Cancel => backend.cancel_worker(&handle), WorkerExecutionOperation::Spawn | WorkerExecutionOperation::Restore - | WorkerExecutionOperation::Input => return Ok(()), + | WorkerExecutionOperation::Input + | WorkerExecutionOperation::ProtocolMethod => return Ok(()), }; if result.is_accepted() { self.record_execution_result(worker_ref, result)?; diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index bddec8fb..35f12934 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -722,6 +722,27 @@ impl Drop for WorkerRuntimeExecutionBackend { } } +fn method_starts_turn(method: &Method) -> bool { + matches!( + method, + Method::Run { .. } + | Method::Notify { auto_run: true, .. } + | Method::Resume + | Method::Compact + ) +} + +fn accepted_run_state_for_method(method: &Method) -> WorkerExecutionRunState { + match method { + Method::Run { .. } + | Method::Notify { auto_run: true, .. } + | Method::Resume + | Method::Compact => WorkerExecutionRunState::Busy, + Method::Shutdown => WorkerExecutionRunState::Stopped, + _ => WorkerExecutionRunState::Idle, + } +} + impl WorkerExecutionBackend for WorkerRuntimeExecutionBackend where F: RuntimeWorkerFactory, @@ -1039,6 +1060,48 @@ where result } + fn dispatch_method( + &self, + handle: &WorkerExecutionHandle, + method: Method, + ) -> WorkerExecutionResult { + let (worker, busy) = match self.get_execution(handle) { + Ok(execution) => execution, + Err(mut result) => { + result.operation = WorkerExecutionOperation::ProtocolMethod; + return result; + } + }; + + let starts_turn = method_starts_turn(&method); + if starts_turn + && (worker.shared_state.get_status() != WorkerStatus::Idle + || busy + .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst) + .is_err()) + { + return WorkerExecutionResult::busy( + WorkerExecutionOperation::ProtocolMethod, + "Worker is already running; runtime adapter v0 does not queue protocol methods", + ); + } + + let accepted_run_state = accepted_run_state_for_method(&method); + let accepted_is_idle = accepted_run_state == WorkerExecutionRunState::Idle; + let result = self.send_method( + WorkerExecutionOperation::ProtocolMethod, + worker, + method, + accepted_run_state, + ); + if (starts_turn && accepted_is_idle) + || (starts_turn && result.outcome != crate::execution::WorkerExecutionOutcome::Accepted) + { + busy.store(false, Ordering::SeqCst); + } + result + } + fn stop_worker(&self, handle: &WorkerExecutionHandle) -> WorkerExecutionResult { if handle.backend_id() != self.backend_id() { return WorkerExecutionResult::rejected( diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index b62231ff..b9a19eae 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -514,6 +514,21 @@ pub enum RuntimeRegistryError { } impl RuntimeRegistryError { + pub fn message(&self) -> String { + match self { + Self::InvalidIdentifier { kind, value } => { + format!("invalid {kind} identifier `{value}`") + } + Self::UnknownRuntime(runtime_id) => format!("unknown runtime `{runtime_id}`"), + Self::UnknownHost(host_id) => format!("unknown host `{host_id}`"), + Self::UnknownWorker { + runtime_id, + worker_id, + } => format!("unknown worker `{worker_id}` in runtime `{runtime_id}`"), + Self::RuntimeOperationFailed { message, .. } => message.clone(), + } + } + pub fn into_error(self) -> Error { match self { Self::InvalidIdentifier { kind, value } => Error::InvalidRuntimeIdentifier { @@ -658,6 +673,18 @@ pub trait WorkspaceWorkerRuntime: Send + Sync { } } + fn send_protocol_method( + &self, + _worker_id: &str, + _method: protocol::Method, + ) -> Result, RuntimeRegistryError> { + Err(RuntimeRegistryError::RuntimeOperationFailed { + runtime_id: self.runtime_id().to_string(), + code: "worker_protocol_method_unsupported".to_string(), + message: "runtime does not support Worker protocol command transport".to_string(), + }) + } + fn stop_worker( &self, worker_id: &str, @@ -1062,6 +1089,26 @@ impl RuntimeRegistry { Ok(runtime.list_config_bundles()) } + pub fn send_protocol_method( + &self, + runtime_id: &str, + worker_id: &str, + method: protocol::Method, + ) -> Result, RuntimeRegistryError> { + validate_backend_identifier("runtime_id", runtime_id)?; + validate_backend_identifier("worker_id", worker_id)?; + let runtime = self.runtime(runtime_id)?; + let lookup = runtime.worker(worker_id); + if lookup.worker.is_none() { + return Err(operation_failed_or_unknown_worker( + runtime_id, + worker_id, + lookup.diagnostics, + )); + } + runtime.send_protocol_method(worker_id, method) + } + pub fn send_input( &self, runtime_id: &str, @@ -1807,6 +1854,35 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime { )) } + fn send_protocol_method( + &self, + worker_id: &str, + method: protocol::Method, + ) -> Result, RuntimeRegistryError> { + if !self.execution_enabled { + return Err(RuntimeRegistryError::RuntimeOperationFailed { + runtime_id: self.runtime_id.clone(), + code: "embedded_worker_execution_unavailable".to_string(), + message: format!( + "worker protocol command for '{worker_id}' requires an embedded execution backend" + ), + }); + } + let Some(worker_ref) = self.worker_ref(worker_id) else { + return Err(RuntimeRegistryError::UnknownWorker { + runtime_id: self.runtime_id.clone(), + worker_id: worker_id.to_string(), + }); + }; + self.runtime + .send_protocol_method(&worker_ref, method) + .map_err(|error| RuntimeRegistryError::RuntimeOperationFailed { + runtime_id: self.runtime_id.clone(), + code: "embedded_worker_protocol_command_failed".to_string(), + message: error.to_string(), + }) + } + fn send_input(&self, worker_id: &str, request: WorkerInputRequest) -> WorkerInputResult { if !self.execution_enabled { return embedded_input_rejected( diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index f67c8486..15fefb06 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -555,6 +555,14 @@ pub fn build_router(api: WorkspaceApi) -> Router { "/api/w/{workspace_id}/runtimes/{runtime_id}/workers/{worker_id}/events/ws", get(scoped_worker_observation_ws), ) + .route( + "/api/runtimes/{runtime_id}/workers/{worker_id}/protocol/ws", + get(worker_protocol_ws), + ) + .route( + "/api/w/{workspace_id}/runtimes/{runtime_id}/workers/{worker_id}/protocol/ws", + get(scoped_worker_protocol_ws), + ) .route("/api/hosts/{host_id}/workers", get(list_host_workers)) .route( "/api/w/{workspace_id}/hosts/{host_id}/workers", @@ -2457,6 +2465,19 @@ async fn scoped_worker_observation_ws( .into_response() } +async fn scoped_worker_protocol_ws( + ws: WebSocketUpgrade, + State(api): State, + AxumPath(path): AxumPath, +) -> Response { + if let Err(err) = validate_workspace_scope(&api, &path.workspace_id) { + return err.into_response(); + } + worker_protocol_ws(State(api), AxumPath((path.runtime_id, path.worker_id)), ws) + .await + .into_response() +} + async fn scoped_list_host_workers( State(api): State, AxumPath(path): AxumPath, @@ -3388,6 +3409,89 @@ async fn cancel_runtime_worker( Ok(Json(result)) } +async fn worker_protocol_ws( + State(api): State, + AxumPath((runtime_id, worker_id)): AxumPath<(String, String)>, + ws: WebSocketUpgrade, +) -> impl IntoResponse { + match api.runtime.worker(&runtime_id, &worker_id) { + Ok(_) => ws.on_upgrade(move |socket| { + worker_protocol_ws_session(api.runtime, runtime_id, worker_id, socket) + }), + Err(error) => ApiError::from(error.into_error()).into_response(), + } +} + +async fn worker_protocol_ws_session( + runtime: Arc, + runtime_id: String, + worker_id: String, + mut socket: WebSocket, +) { + while let Some(frame) = socket.next().await { + match frame { + Ok(WsMessage::Text(text)) => match serde_json::from_str::(&text) { + Ok(method) => match runtime.send_protocol_method(&runtime_id, &worker_id, method) { + Ok(events) => { + for event in events { + if !send_protocol_event(&mut socket, &event).await { + return; + } + } + } + Err(error) => { + let event = protocol_error_event(error.message()); + if !send_protocol_event(&mut socket, &event).await { + return; + } + } + }, + Err(error) => { + let event = + protocol_error_event(format!("malformed protocol method frame: {error}")); + if !send_protocol_event(&mut socket, &event).await { + return; + } + } + }, + Ok(WsMessage::Close(_)) => return, + Ok(WsMessage::Ping(payload)) => { + if socket.send(WsMessage::Pong(payload)).await.is_err() { + return; + } + } + Ok(WsMessage::Pong(_)) | Ok(WsMessage::Binary(_)) => {} + Err(error) => { + let event = protocol_error_event(format!("protocol WebSocket error: {error}")); + let _ = send_protocol_event(&mut socket, &event).await; + return; + } + } + } +} + +async fn send_protocol_event(socket: &mut WebSocket, event: &protocol::Event) -> bool { + match serde_json::to_string(event) { + Ok(text) => socket.send(WsMessage::Text(text.into())).await.is_ok(), + Err(error) => { + let fallback = protocol_error_event(format!( + "failed to serialize protocol response event: {error}" + )); + let Ok(text) = serde_json::to_string(&fallback) else { + return false; + }; + socket.send(WsMessage::Text(text.into())).await.is_ok() + } + } +} + +fn protocol_error_event(message: impl Into) -> protocol::Event { + protocol::Event::Error { + code: protocol::ErrorCode::Internal, + message: message.into(), + } +} + async fn worker_observation_ws( State(api): State, AxumPath((runtime_id, worker_id)): AxumPath<(String, String)>, diff --git a/web/workspace/src/routes/w/[workspaceId]/runtimes/[runtimeId]/workers/[workerId]/console/+page.svelte b/web/workspace/src/routes/w/[workspaceId]/runtimes/[runtimeId]/workers/[workerId]/console/+page.svelte index c18c0415..b2485897 100644 --- a/web/workspace/src/routes/w/[workspaceId]/runtimes/[runtimeId]/workers/[workerId]/console/+page.svelte +++ b/web/workspace/src/routes/w/[workspaceId]/runtimes/[runtimeId]/workers/[workerId]/console/+page.svelte @@ -3,7 +3,10 @@ import ConsoleLineItem from "$lib/workspace/console/ConsoleLineItem.svelte"; import ConsoleTimeline from "$lib/workspace/console/ConsoleTimeline.svelte"; import { chatSubmit } from "$lib/workspace/console/chat-submit"; - import { buildComposerRequest } from "$lib/workspace/console/composer-command"; + import { + buildComposerRequest, + type WorkerConsoleInputRequest, + } from "$lib/workspace/console/composer-command"; import { applyCompletion, completionTokenAt, @@ -18,12 +21,12 @@ type ConsoleLine, type ConsoleProjection, } from "$lib/workspace/console/model"; + import type { Event as ProtocolEvent, Method as ProtocolMethod, RewindTarget, Segment } from "$lib/generated/protocol"; import { workspaceApiPath } from "$lib/workspace/api/http"; import type { ClientWorkerEventWsFrame, Diagnostic, Worker, - WorkerInputResult, PodProtocolEvent, } from "$lib/workspace/sidebar/types"; @@ -45,13 +48,6 @@ return workspaceApiPath(workspaceId, path); } - type WorkerCompletionsResult = { - kind: "file"; - prefix: string; - entries: ComposerCompletionEntry[]; - diagnostics: Diagnostic[]; - }; - type TimelineKind = "turn" | "assistant"; type TimelineMark = { @@ -98,10 +94,22 @@ let completionError = $state(null); let sending = $state(false); let sendError = $state(null); + let rewindTargets = $state([]); + let rewindHeadEntries = $state(0); + let controlNotice = $state(null); let composerNotice = $state(null); let streamState = $state<"connecting" | "open" | "closed" | "error">( "connecting", ); + let commandState = $state<"connecting" | "open" | "closed" | "error">( + "connecting", + ); + let commandSocket: WebSocket | null = null; + let pendingCompletionRequest: { + resolve: (entries: ComposerCompletionEntry[]) => void; + reject: (error: Error) => void; + timeout: number; + } | null = null; let streamDiagnostics = $state([]); let workerDetailsOpen = $state(false); let timelineOpen = $state(false); @@ -162,37 +170,6 @@ return response.json() as Promise; } - async function postJson( - path: string, - body: unknown, - timeoutMs = 30_000, - ): Promise { - const controller = new AbortController(); - const timeout = window.setTimeout(() => controller.abort(), timeoutMs); - try { - const response = await fetch(path, { - method: "POST", - headers: { "content-type": "application/json" }, - body: JSON.stringify(body), - signal: controller.signal, - }); - if (!response.ok) { - let detail = ""; - try { - detail = await response.text(); - } catch { - detail = ""; - } - throw new Error( - `POST ${path} failed: ${response.status}${detail ? ` ${detail}` : ""}`, - ); - } - return response.json() as Promise; - } finally { - window.clearTimeout(timeout); - } - } - async function loadWorker(target: ConsoleTarget) { workerError = null; try { @@ -284,14 +261,16 @@ } const observedAtMs = Date.now(); eventObservedAtById.set(frame.envelope.event_id, observedAtMs); + const payload = frame.envelope.payload; pendingObservationEvents.push({ eventId: frame.envelope.event_id, - event: frame.envelope.payload, + event: payload, observedAtMs, }); - pendingObservedStates.push( - workerStateFromProtocolEvent(frame.envelope.payload), - ); + if (payload.event === "rewind_targets") { + handleProtocolCommandEvent(payload as ProtocolEvent); + } + pendingObservedStates.push(workerStateFromProtocolEvent(payload)); scheduleObservationFlush(); } @@ -354,16 +333,29 @@ if (token.kind === "command") { return localCommandCompletions(token.prefix); } - const result = await postJson( - workerApiPath( - `/runtimes/${encodeURIComponent(runtimeId)}/workers/${encodeURIComponent(workerId)}/completions`, - ), - { kind: token.kind, prefix: token.prefix }, - ); - if (result.diagnostics.length > 0 && result.entries.length === 0) { - throw new Error(diagnosticsToText(result.diagnostics)); - } - return result.entries; + return new Promise((resolve, reject) => { + if (pendingCompletionRequest) { + rejectPendingCompletion( + new Error("Superseded by a newer completion request."), + ); + } + const timeout = window.setTimeout(() => { + rejectPendingCompletion( + new Error("Worker completion request timed out."), + ); + }, 30_000); + pendingCompletionRequest = { resolve, reject, timeout }; + try { + sendProtocolMethod({ + method: "list_completions", + params: { kind: token.kind, prefix: token.prefix }, + }); + } catch (error) { + rejectPendingCompletion( + error instanceof Error ? error : new Error(String(error)), + ); + } + }); } function handleComposerKeydown(event: KeyboardEvent) { @@ -401,6 +393,63 @@ composerTextareaElement?.focus(); } + function sendControl(method: ProtocolMethod, label: string) { + try { + sendProtocolMethod(method); + controlNotice = `${label} sent through Worker protocol.`; + } catch (error) { + controlNotice = null; + sendError = error instanceof Error ? error.message : String(error); + } + } + + function requestRewindTargets() { + sendControl({ method: "list_rewind_targets" }, "Rewind target request"); + } + + function rewindTo(target: RewindTarget) { + sendControl( + { + method: "rewind_to", + params: { + target: target.id, + expected_head_entries: rewindHeadEntries, + }, + }, + `Rewind to ${target.preview || "target"}`, + ); + } + + function composerRequestToProtocolMethod( + request: WorkerConsoleInputRequest, + ): ProtocolMethod { + switch (request.kind) { + case "user": + return { + method: "run", + params: { + input: request.segments ?? [ + { kind: "text", content: request.content }, + ], + }, + }; + case "system": + return { + method: "notify", + params: { message: request.content, auto_run: true }, + }; + case "compact": + return { method: "compact" }; + case "list_rewind_targets": + return { method: "list_rewind_targets" }; + case "register_peer": + return { + method: "register_peer", + params: { name: request.content }, + }; + } + } + async function submitDraft(value = draft) { const command = buildComposerRequest(value); if (!command.ok) { @@ -420,20 +469,13 @@ sending = true; sendError = null; try { - const result = await postJson( - workerApiPath( - `/runtimes/${encodeURIComponent(runtimeId)}/workers/${encodeURIComponent(workerId)}/input`, - ), - command.request, - ); - if (result.state === "accepted") { - draft = ""; + const method = composerRequestToProtocolMethod(command.request); + sendProtocolMethod(method); + draft = ""; + if (method.method === "run" || method.method === "notify") { liveWorkerState = "running"; - } else { - sendError = - diagnosticsToText(result.diagnostics) || - `Input was ${result.state}.`; } + composerNotice = "Sent through Worker protocol."; } catch (error) { sendError = error instanceof Error ? error.message : String(error); } finally { @@ -533,6 +575,138 @@ return () => ws.close(); } + function connectProtocolCommands( + targetWorker: Worker | null, + token: number, + target: ConsoleTarget, + ) { + if (!targetWorker) { + commandState = "closed"; + return; + } + commandState = "connecting"; + const protocol = window.location.protocol === "https:" ? "wss:" : "ws:"; + const wsPath = workerApiPath( + `/runtimes/${encodeURIComponent(target.runtimeId)}/workers/${encodeURIComponent( + target.workerId, + )}/protocol/ws`, + ); + const ws = new WebSocket( + `${protocol}//${window.location.host}${wsPath}`, + ); + commandSocket = ws; + + ws.onopen = () => { + if (token === reloadToken) { + commandState = "open"; + } + }; + ws.onmessage = (message) => { + if (token !== reloadToken) { + return; + } + try { + handleProtocolCommandEvent( + JSON.parse(String(message.data)) as ProtocolEvent, + ); + } catch (error) { + streamDiagnostics = [ + ...streamDiagnostics, + { + code: "worker_protocol_command_frame_invalid", + severity: "warning", + message: + error instanceof Error ? error.message : String(error), + }, + ]; + } + }; + ws.onerror = () => { + if (token === reloadToken) { + commandState = "error"; + streamDiagnostics = [ + ...streamDiagnostics, + { + code: "worker_protocol_command_ws_error", + severity: "error", + message: "Worker protocol command WebSocket failed.", + }, + ]; + } + }; + ws.onclose = () => { + if (commandSocket === ws) { + commandSocket = null; + } + if (token === reloadToken && commandState !== "error") { + commandState = "closed"; + } + rejectPendingCompletion( + new Error("Worker protocol command WebSocket closed."), + ); + }; + + return () => { + if (commandSocket === ws) { + commandSocket = null; + } + ws.close(); + }; + } + + function sendProtocolMethod(method: ProtocolMethod) { + if (!commandSocket || commandSocket.readyState !== WebSocket.OPEN) { + throw new Error("Worker protocol command WebSocket is not open."); + } + commandSocket.send(JSON.stringify(method)); + } + + function handleProtocolCommandEvent(event: ProtocolEvent) { + if (event.event === "completions") { + const pending = pendingCompletionRequest; + if (!pending) { + return; + } + pendingCompletionRequest = null; + window.clearTimeout(pending.timeout); + pending.resolve(event.data.entries); + return; + } + if (event.event === "rewind_targets") { + rewindHeadEntries = event.data.head_entries; + rewindTargets = event.data.targets; + controlNotice = + event.data.targets.length === 0 + ? "No rewind targets are available." + : `Loaded ${event.data.targets.length} rewind target(s).`; + return; + } + if (event.event === "error") { + const error = new Error(event.data.message); + if (pendingCompletionRequest) { + rejectPendingCompletion(error); + } + streamDiagnostics = [ + ...streamDiagnostics, + { + code: event.data.code, + severity: "error", + message: event.data.message, + }, + ]; + } + } + + function rejectPendingCompletion(error: Error) { + const pending = pendingCompletionRequest; + if (!pending) { + return; + } + pendingCompletionRequest = null; + window.clearTimeout(pending.timeout); + pending.reject(error); + } + function mergeDiagnostics(...groups: Diagnostic[][]): Diagnostic[] { return groups.flat(); } @@ -989,6 +1163,7 @@ }); $effect(() => connectObservation(worker, reloadToken, consoleTarget)); + $effect(() => connectProtocolCommands(worker, reloadToken, consoleTarget)); @@ -1009,8 +1184,48 @@ class="console-status-pill" class:warn={streamState !== "open"} > - {workerState} · stream {streamState} + {workerState} · stream {streamState} · command {commandState} + + + + + + {/each} + + + {/if} +
@@ -1267,6 +1508,35 @@ color: var(--warning); } + .console-notice { + margin: 0; + color: var(--text-muted); + font-size: 0.86rem; + } + + .rewind-targets { + display: flex; + align-items: center; + gap: var(--space-3); + padding: var(--space-3); + } + + .rewind-targets h3 { + margin: 0; + font-size: 0.9rem; + } + + .rewind-target-list { + display: flex; + flex-wrap: wrap; + gap: var(--space-2); + } + + .rewind-target-list span { + margin-left: 0.5rem; + color: var(--text-muted); + } + .console-body { --console-timeline-width: 12rem; --console-timeline-fold-width: 2.25rem;