diff --git a/.yoi/tickets/00001KY1ZKCPC/item.md b/.yoi/tickets/00001KY1ZKCPC/item.md index 4ecbd35e..af474a6f 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-22T03:51:04Z' +updated_at: '2026-07-22T05:03:17Z' assignee: null readiness: 'draft' --- diff --git a/.yoi/tickets/00001KY1ZKCPC/thread.md b/.yoi/tickets/00001KY1ZKCPC/thread.md index c2bd896e..1c160d9a 100644 --- a/.yoi/tickets/00001KY1ZKCPC/thread.md +++ b/.yoi/tickets/00001KY1ZKCPC/thread.md @@ -301,3 +301,43 @@ Verification: - `git diff --check` --- + + + +## Implementation report + +Implementation update for shared protocol frame/pump semantics: + +- Added Method/Event-specific JSON codec helpers in `protocol::stream`: + - `encode_method` / `decode_method` + - `encode_event` / `decode_event` + - `JsonLineReader::next_method()` + - `JsonLineWriter::write_event()` +- Updated local IPC to use the shared codec helpers and the shared `worker::ipc::protocol_session` semantics. +- Updated runtime WS and workspace-server embedded protocol handling to use the same `decode_method` / `encode_event` helpers instead of direct `serde_json` calls. +- Updated Backend runtime client and remote observation client to use `encode_method` / `decode_event` for protocol frames. + +Resulting split: +- Shared: + - JSON shape for `protocol::Method` / `protocol::Event` + - Method/Event encode/decode + - WorkerHandle subscription semantics + - live `LogEntry -> Event` mapping + - direct-response method semantics such as `ListCompletions` +- Transport-specific: + - Unix IPC uses JSONL line boundary. + - WebSocket uses one text frame per JSON value. + +This makes WS effectively a frame-boundary adapter around the same JSON protocol frames rather than a separate protocol implementation. + +Verification: +- `nix develop -c cargo fmt -- --check` +- `nix develop -c cargo check -p protocol -p worker -p worker-runtime -p yoi-workspace-server -p client -p tui` +- `nix develop -c cargo test -p worker --lib ipc::protocol_session` +- `nix develop -c cargo test -p worker --lib ipc::server` +- `nix develop -c cargo test -p worker-runtime --features ws-server --lib protocol_ws` +- `nix develop -c cargo test -p yoi-workspace-server protocol_ws --lib` +- `nix develop -c cargo test -p client backend_runtime --lib` +- `git diff --check` + +--- diff --git a/crates/client/src/backend_runtime.rs b/crates/client/src/backend_runtime.rs index 633ab6aa..47f0aceb 100644 --- a/crates/client/src/backend_runtime.rs +++ b/crates/client/src/backend_runtime.rs @@ -1,4 +1,5 @@ use futures::{SinkExt, StreamExt}; +use protocol::stream::{decode_event, encode_method}; use protocol::{ErrorCode, Event, Method}; use serde::Deserialize; use std::collections::VecDeque; @@ -317,7 +318,7 @@ async fn run_worker_protocol_transport( let Some(method) = maybe_method else { break; }; - match serde_json::to_string(&method) { + match encode_method(&method) { Ok(text) => { if let Err(error) = sink.send(TungsteniteMessage::Text(text.into())).await { let _ = tx.send(diagnostic_event(format!( @@ -338,7 +339,7 @@ async fn run_worker_protocol_transport( frame = stream.next() => { match frame { Some(Ok(TungsteniteMessage::Text(text))) => { - match serde_json::from_str::(&text) { + match decode_event(&text) { Ok(event) => { let _ = tx.send(event); } diff --git a/crates/protocol/src/stream.rs b/crates/protocol/src/stream.rs index cf612168..d3744ad4 100644 --- a/crates/protocol/src/stream.rs +++ b/crates/protocol/src/stream.rs @@ -4,6 +4,24 @@ use serde::Serialize; use serde::de::DeserializeOwned; use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncWrite, AsyncWriteExt, BufReader}; +use crate::{Event, Method}; + +pub fn encode_method(method: &Method) -> Result { + serde_json::to_string(method) +} + +pub fn decode_method(input: &str) -> Result { + serde_json::from_str(input.trim()) +} + +pub fn encode_event(event: &Event) -> Result { + serde_json::to_string(event) +} + +pub fn decode_event(input: &str) -> Result { + serde_json::from_str(input.trim()) +} + /// JSONL line reader over an async byte stream. /// /// Wraps the inner reader and deserialises each non‑empty line as `T`. @@ -41,6 +59,10 @@ impl JsonLineReader { return Ok(Some(value)); } } + + pub async fn next_method(&mut self) -> Result, io::Error> { + self.next::().await + } } /// JSONL line writer over an async byte stream. @@ -62,4 +84,8 @@ impl JsonLineWriter { self.inner.flush().await?; Ok(()) } + + pub async fn write_event(&mut self, event: &Event) -> Result<(), io::Error> { + self.write(event).await + } } diff --git a/crates/worker-runtime/src/http_server.rs b/crates/worker-runtime/src/http_server.rs index 252d94a0..f99c1800 100644 --- a/crates/worker-runtime/src/http_server.rs +++ b/crates/worker-runtime/src/http_server.rs @@ -30,6 +30,8 @@ use axum::routing::{get, post}; use axum::{Json, Router}; #[cfg(feature = "ws-server")] use futures::StreamExt; +#[cfg(feature = "ws-server")] +use protocol::stream::{decode_method, encode_event}; use serde::{Deserialize, Serialize}; use std::fmt; use std::net::SocketAddr; @@ -521,7 +523,7 @@ async fn worker_protocol_ws_session( tokio::select! { inbound = socket.next() => { match inbound { - Some(Ok(WsMessage::Text(text))) => match serde_json::from_str::(&text) { + Some(Ok(WsMessage::Text(text))) => match decode_method(&text) { Ok(method) => match runtime.send_protocol_method(&worker_ref, method) { Ok(events) => { for event in events { @@ -587,13 +589,13 @@ async fn worker_protocol_ws_session( #[cfg(feature = "ws-server")] async fn send_protocol_event(socket: &mut WebSocket, event: &protocol::Event) -> bool { - match serde_json::to_string(event) { + match encode_event(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 { + let Ok(text) = encode_event(&fallback) else { return false; }; socket.send(WsMessage::Text(text.into())).await.is_ok() diff --git a/crates/worker/src/ipc/server.rs b/crates/worker/src/ipc/server.rs index 27f0676d..62044ecf 100644 --- a/crates/worker/src/ipc/server.rs +++ b/crates/worker/src/ipc/server.rs @@ -10,7 +10,7 @@ use crate::controller::WorkerHandle; use crate::ipc::protocol_session::{ dispatch_worker_protocol_method, live_log_entry_event, subscribe_worker_protocol_session, }; -use protocol::{ErrorCode, Event, Method}; +use protocol::{ErrorCode, Event}; /// Unix socket server for Worker Protocol. /// @@ -84,14 +84,14 @@ async fn handle_connection(stream: tokio::net::UnixStream, handle: WorkerHandle) // keeps the snapshot/live boundary gap-free. let mut streams = subscribe_worker_protocol_session(&handle); for alert in streams.alert_snapshot { - if writer.write(&Event::Alert(alert)).await.is_err() { + if writer.write_event(&Event::Alert(alert)).await.is_err() { return; } } // Send the typed snapshot up front so late attachers can // reconstruct view state without an extra round trip. - if writer.write(&streams.snapshot_event).await.is_err() { + if writer.write_event(&streams.snapshot_event).await.is_err() { return; } @@ -105,7 +105,7 @@ async fn handle_connection(stream: tokio::net::UnixStream, handle: WorkerHandle) match entry { Ok(entry) => { if let Some(event) = live_log_entry_event(entry) { - if writer.write(&event).await.is_err() { + if writer.write_event(&event).await.is_err() { break; } } @@ -123,7 +123,7 @@ async fn handle_connection(stream: tokio::net::UnixStream, handle: WorkerHandle) event = streams.events.recv() => { match event { Ok(event) => { - if writer.write(&event).await.is_err() { + if writer.write_event(&event).await.is_err() { break; } } @@ -131,11 +131,11 @@ async fn handle_connection(stream: tokio::net::UnixStream, handle: WorkerHandle) } } // Client methods → handle or forward to controller - method = reader.next::() => { + method = reader.next_method() => { match method { Ok(Some(method)) => { if let Some(response) = dispatch_worker_protocol_method(&handle, method).await - && writer.write(&response).await.is_err() + && writer.write_event(&response).await.is_err() { break; } diff --git a/crates/workspace-server/src/observation.rs b/crates/workspace-server/src/observation.rs index 53f7d126..756108a6 100644 --- a/crates/workspace-server/src/observation.rs +++ b/crates/workspace-server/src/observation.rs @@ -6,6 +6,7 @@ use worker_runtime::observation::{WorkerObservationCursor, WorkerObservationEven use axum::http::StatusCode; use futures::{SinkExt, StreamExt}; +use protocol::stream::decode_event; use serde::{Deserialize, Serialize}; use tokio_tungstenite::connect_async; use tokio_tungstenite::tungstenite::client::IntoClientRequest; @@ -284,7 +285,7 @@ impl RuntimeWsObservationClient { )); } }; - let payload: protocol::Event = serde_json::from_str(&text).map_err(|error| { + let payload: protocol::Event = decode_event(&text).map_err(|error| { ObservationProxyError::MalformedFrame(format!( "failed to decode runtime protocol event frame: {error}" )) diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 3ffe726a..0d2992ed 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -15,6 +15,7 @@ use futures::{SinkExt, StreamExt}; use memory::backend::{ MemoryBackendHttpResponse, MemoryBackendOperation, execute_memory_backend_operation, }; +use protocol::stream::{decode_method, encode_event}; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; use ticket::{ @@ -3561,7 +3562,7 @@ async fn embedded_worker_protocol_ws_session( tokio::select! { inbound = socket.next() => { match inbound { - Some(Ok(WsMessage::Text(text))) => match serde_json::from_str::(&text) { + Some(Ok(WsMessage::Text(text))) => match decode_method(&text) { Ok(method) => match source.runtime.send_protocol_method(&source.worker_ref, method) { Ok(events) => { for event in events { @@ -3617,13 +3618,13 @@ async fn embedded_worker_protocol_ws_session( } } async fn send_protocol_event(socket: &mut WebSocket, event: &protocol::Event) -> bool { - match serde_json::to_string(event) { + match encode_event(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 { + let Ok(text) = encode_event(&fallback) else { return false; }; socket.send(WsMessage::Text(text.into())).await.is_ok()