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

fix: share protocol frame codec

Browse Source
This commit is contained in:
Keisuke Hirata 2026-07-22 14:03:33 +09:00
parent 945716d010
commit b9f4c06c68
No known key found for this signature in database
8 changed files with 88 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.yoi/tickets/00001KY1ZKCPC/item.md
View File

@ -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'
---

40
.yoi/tickets/00001KY1ZKCPC/thread.md
View File

@ -301,3 +301,43 @@ Verification:
- `git diff --check`
---
<!-- event: implementation_report author: assistant at: 2026-07-22T05:03:17Z -->
## 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`
---

5
crates/client/src/backend_runtime.rs
View File

@ -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::<Event>(&text) {
match decode_event(&text) {
Ok(event) => {
let _ = tx.send(event);
}

26
crates/protocol/src/stream.rs
View File

@ -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<String, serde_json::Error> {
serde_json::to_string(method)
}
pub fn decode_method(input: &str) -> Result<Method, serde_json::Error> {
serde_json::from_str(input.trim())
}
pub fn encode_event(event: &Event) -> Result<String, serde_json::Error> {
serde_json::to_string(event)
}
pub fn decode_event(input: &str) -> Result<Event, serde_json::Error> {
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<R: AsyncBufRead + Unpin> JsonLineReader<R> {
return Ok(Some(value));
}
}
pub async fn next_method(&mut self) -> Result<Option<Method>, io::Error> {
self.next::<Method>().await
}
}
/// JSONL line writer over an async byte stream.
@ -62,4 +84,8 @@ impl<W: AsyncWrite + Unpin> JsonLineWriter<W> {
self.inner.flush().await?;
Ok(())
}
pub async fn write_event(&mut self, event: &Event) -> Result<(), io::Error> {
self.write(event).await
}
}

8
crates/worker-runtime/src/http_server.rs
View File

@ -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::<protocol::Method>(&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()

14
crates/worker/src/ipc/server.rs
View File

@ -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>() => {
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;
}

3
crates/workspace-server/src/observation.rs
View File

@ -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}"
))

7
crates/workspace-server/src/server.rs
View File

@ -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::<protocol::Method>(&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()