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

workspace: remove observation cursor surface

Browse Source
This commit is contained in:
Keisuke Hirata 2026-07-11 09:31:49 +09:00
parent 54eaa44681
commit 10762a3c8b
No known key found for this signature in database
9 changed files with 172 additions and 377 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
.yoi/tickets/00001KX7838M2/item.md
View File

@ -1,8 +1,8 @@
--- ---
title: 'Remove worker observation WebSocket cursor surface' title: 'Remove worker observation WebSocket cursor surface'
state: 'inprogress' state: 'done'
created_at: '2026-07-11T00:09:06Z' created_at: '2026-07-11T00:09:06Z'
updated_at: '2026-07-11T00:09:33Z' updated_at: '2026-07-11T00:31:41Z'
assignee: null assignee: null
queued_by: 'yoi ticket' queued_by: 'yoi ticket'
queued_at: '2026-07-11T00:09:32Z' queued_at: '2026-07-11T00:09:32Z'

33
.yoi/tickets/00001KX7838M2/thread.md
View File

@ -39,4 +39,37 @@ Ticket を `yoi ticket` が queued にしました。
State changed to `inprogress`. State changed to `inprogress`.
---
<!-- event: implementation_report author: hare at: 2026-07-11T00:31:41Z -->
## Implementation report
Removed the backend-local worker observation cursor surface.
- Browser-facing worker observation envelopes no longer include `cursor`; they carry runtime/worker identity, runtime event id, and payload.
- Workspace-server observation proxy no longer owns cursor state, cursor validation, replay state, or store/open methods; it only maps upstream Runtime events to browser envelopes.
- Workspace-server WS attach ignores no cursor path because no cursor query is accepted/extracted.
- Backend Runtime client no longer sends cursor query params or deduplicates by backend cursor.
- Frontend Console no longer stores cursor/lastCursor and projects events by runtime event id.
Validation:
- cargo test -q -p yoi-workspace-server
- cargo test -q -p client
- cd web/workspace && deno task check
- cd web/workspace && deno task test
- cargo test -q
- git diff --check
- nix build .#yoi --no-link
---
<!-- event: state_changed author: "yoi ticket" at: 2026-07-11T00:31:41Z from: inprogress to: done reason: cli_state field: state -->
## State changed
State changed to `done`.
--- ---

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

@ -270,12 +270,10 @@ fn backend_command_from_method(method: &Method) -> BackendCommand {
} }
async fn observe_worker_events(target: BackendRuntimeTarget, tx: mpsc::UnboundedSender<Event>) { async fn observe_worker_events(target: BackendRuntimeTarget, tx: mpsc::UnboundedSender<Event>) {
let mut cursor: Option<String> = None;
let mut last_sequence = 0_u64;
let mut attempts = 0_usize; let mut attempts = 0_usize;
loop { loop {
let url = observation_ws_url(&target, cursor.as_deref()); let url = observation_ws_url(&target);
match connect_async(&url).await { match connect_async(&url).await {
Ok((mut ws, _)) => { Ok((mut ws, _)) => {
attempts = 0; attempts = 0;
@ -295,19 +293,6 @@ async fn observe_worker_events(target: BackendRuntimeTarget, tx: mpsc::Unbounded
))); )));
continue; continue;
} }
if let Some(sequence) = decode_backend_cursor(&envelope.cursor)
{
if sequence <= last_sequence {
continue;
}
last_sequence = sequence;
} else {
let _ = tx.send(diagnostic_event(format!(
"Backend observation cursor was malformed: {}",
envelope.cursor
)));
}
cursor = Some(envelope.cursor.clone());
let _ = tx.send(envelope.payload); let _ = tx.send(envelope.payload);
} }
Ok(ClientWorkerEventWsFrame::Diagnostic { diagnostic }) => { Ok(ClientWorkerEventWsFrame::Diagnostic { diagnostic }) => {
@ -316,11 +301,6 @@ async fn observe_worker_events(target: BackendRuntimeTarget, tx: mpsc::Unbounded
diagnostic.code, diagnostic.message diagnostic.code, diagnostic.message
); );
let _ = tx.send(diagnostic_event(message)); let _ = tx.send(diagnostic_event(message));
if diagnostic.code == "backend.cursor_unknown_or_expired" {
cursor = None;
last_sequence = 0;
break;
}
} }
Err(error) => { Err(error) => {
let _ = tx.send(diagnostic_event(format!( let _ = tx.send(diagnostic_event(format!(
@ -395,18 +375,13 @@ fn validate_target(target: &BackendRuntimeTarget) -> Result<(), BackendRuntimeCl
Ok(()) Ok(())
} }
fn observation_ws_url(target: &BackendRuntimeTarget, cursor: Option<&str>) -> String { fn observation_ws_url(target: &BackendRuntimeTarget) -> String {
let path = format!( let path = format!(
"/api/runtimes/{}/workers/{}/events/ws", "/api/runtimes/{}/workers/{}/events/ws",
path_segment_encode(&target.runtime_id), path_segment_encode(&target.runtime_id),
path_segment_encode(&target.worker_id) path_segment_encode(&target.worker_id)
); );
let mut url = join_base_and_path(&http_base_to_ws(&target.base_url), &path); join_base_and_path(&http_base_to_ws(&target.base_url), &path)
if let Some(cursor) = cursor {
url.push_str("?cursor=");
url.push_str(&query_value_encode(cursor));
}
url
} }
fn http_base_to_ws(base: &str) -> String { fn http_base_to_ws(base: &str) -> String {
@ -423,26 +398,12 @@ fn join_base_and_path(base: &str, path: &str) -> String {
format!("{}{}", base.trim_end_matches('/'), path) format!("{}{}", base.trim_end_matches('/'), path)
} }
fn decode_backend_cursor(cursor: &str) -> Option<u64> {
let encoded = cursor.strip_prefix("bo_")?;
if encoded.len() != 16 {
return None;
}
u64::from_str_radix(encoded, 16).ok()
}
fn path_segment_encode(input: &str) -> String { fn path_segment_encode(input: &str) -> String {
percent_encode(input, |byte| { percent_encode(input, |byte| {
byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'.' | b'_' | b'~') byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'.' | b'_' | b'~')
}) })
} }
fn query_value_encode(input: &str) -> String {
percent_encode(input, |byte| {
byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'.' | b'_' | b'~')
})
}
fn percent_encode(input: &str, keep: impl Fn(u8) -> bool) -> String { fn percent_encode(input: &str, keep: impl Fn(u8) -> bool) -> String {
let mut encoded = String::with_capacity(input.len()); let mut encoded = String::with_capacity(input.len());
for byte in input.bytes() { for byte in input.bytes() {
@ -507,7 +468,6 @@ enum ClientWorkerEventWsFrame {
#[derive(Debug, Deserialize)] #[derive(Debug, Deserialize)]
struct ClientWorkerEventWsEnvelope { struct ClientWorkerEventWsEnvelope {
cursor: String,
runtime_id: String, runtime_id: String,
worker_id: String, worker_id: String,
payload: Event, payload: Event,
@ -547,8 +507,8 @@ mod tests {
let target = let target =
BackendRuntimeTarget::new("http://127.0.0.1:8787/", "runtime/one", "worker one"); BackendRuntimeTarget::new("http://127.0.0.1:8787/", "runtime/one", "worker one");
assert_eq!( assert_eq!(
observation_ws_url(&target, Some("bo_0000000000000001")), observation_ws_url(&target),
"ws://127.0.0.1:8787/api/runtimes/runtime%2Fone/workers/worker%20one/events/ws?cursor=bo_0000000000000001" "ws://127.0.0.1:8787/api/runtimes/runtime%2Fone/workers/worker%20one/events/ws"
); );
} }
} }

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

@ -1,5 +1,5 @@
use std::collections::{BTreeMap, VecDeque}; use std::collections::{BTreeMap, VecDeque};
use std::sync::{Arc, Mutex}; use std::sync::Arc;
use worker_runtime::identity::WorkerRef; use worker_runtime::identity::WorkerRef;
use worker_runtime::observation::{WorkerObservationCursor, WorkerObservationEvent}; use worker_runtime::observation::{WorkerObservationCursor, WorkerObservationEvent};
@ -100,7 +100,7 @@ impl std::fmt::Debug for RuntimeObservationSource {
pub struct RuntimeObservationUpstreamEvent { pub struct RuntimeObservationUpstreamEvent {
pub runtime_id: String, pub runtime_id: String,
pub worker_id: String, pub worker_id: String,
pub runtime_cursor: String, pub runtime_event_id: String,
pub payload: protocol::Event, pub payload: protocol::Event,
} }
@ -116,11 +116,10 @@ pub enum ClientWorkerEventWsFrame {
}, },
} }
/// Backend-owned opaque event envelope. It intentionally omits Runtime endpoints, /// Backend-owned event envelope. It intentionally omits Runtime endpoints,
/// credentials, sockets and session paths. /// credentials, sockets and session paths.
#[derive(Clone, Debug, Serialize, Deserialize)] #[derive(Clone, Debug, Serialize, Deserialize)]
pub struct ClientWorkerEventWsEnvelope { pub struct ClientWorkerEventWsEnvelope {
pub cursor: String,
pub event_id: String, pub event_id: String,
pub runtime_id: String, pub runtime_id: String,
pub worker_id: String, pub worker_id: String,
@ -134,17 +133,10 @@ pub struct ClientWorkerEventWsDiagnostic {
pub message: String, pub message: String,
} }
#[derive(Clone, Debug, Default, Deserialize)]
pub struct ClientWorkerEventsWsQuery {
pub cursor: Option<String>,
}
#[derive(Clone, Debug, PartialEq, Eq)] #[derive(Clone, Debug, PartialEq, Eq)]
pub enum ObservationProxyError { pub enum ObservationProxyError {
RuntimeUnavailable(String), RuntimeUnavailable(String),
WorkerNotFound(String), WorkerNotFound(String),
CursorMalformed(String),
CursorUnknownOrExpired(String),
UpstreamDisconnect(String), UpstreamDisconnect(String),
MalformedFrame(String), MalformedFrame(String),
ObservationOnly, ObservationOnly,
@ -155,8 +147,6 @@ impl ObservationProxyError {
match self { match self {
ObservationProxyError::RuntimeUnavailable(_) => "backend.runtime_unavailable", ObservationProxyError::RuntimeUnavailable(_) => "backend.runtime_unavailable",
ObservationProxyError::WorkerNotFound(_) => "backend.worker_not_found", ObservationProxyError::WorkerNotFound(_) => "backend.worker_not_found",
ObservationProxyError::CursorMalformed(_) => "backend.cursor_malformed",
ObservationProxyError::CursorUnknownOrExpired(_) => "backend.cursor_unknown_or_expired",
ObservationProxyError::UpstreamDisconnect(_) => "backend.upstream_disconnect", ObservationProxyError::UpstreamDisconnect(_) => "backend.upstream_disconnect",
ObservationProxyError::MalformedFrame(_) => "backend.malformed_frame", ObservationProxyError::MalformedFrame(_) => "backend.malformed_frame",
ObservationProxyError::ObservationOnly => "backend.observation_only", ObservationProxyError::ObservationOnly => "backend.observation_only",
@ -167,8 +157,6 @@ impl ObservationProxyError {
match self { match self {
ObservationProxyError::RuntimeUnavailable(message) ObservationProxyError::RuntimeUnavailable(message)
| ObservationProxyError::WorkerNotFound(message) | ObservationProxyError::WorkerNotFound(message)
| ObservationProxyError::CursorMalformed(message)
| ObservationProxyError::CursorUnknownOrExpired(message)
| ObservationProxyError::UpstreamDisconnect(message) | ObservationProxyError::UpstreamDisconnect(message)
| ObservationProxyError::MalformedFrame(message) => message, | ObservationProxyError::MalformedFrame(message) => message,
ObservationProxyError::ObservationOnly => { ObservationProxyError::ObservationOnly => {
@ -193,46 +181,6 @@ impl ClientWorkerEventWsFrame {
} }
} }
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
pub struct BackendObservationCursor {
pub sequence: u64,
}
impl BackendObservationCursor {
pub fn new(sequence: u64) -> Self {
Self { sequence }
}
pub fn zero() -> Self {
Self { sequence: 0 }
}
pub fn encode(self) -> String {
format!("bo_{:016x}", self.sequence)
}
pub fn decode(value: &str) -> Option<Self> {
let encoded = value.strip_prefix("bo_")?;
if encoded.len() != 16 {
return None;
}
u64::from_str_radix(encoded, 16)
.ok()
.map(|sequence| Self { sequence })
}
}
#[derive(Debug, Default)]
struct BackendObservationState {
next_sequence: u64,
}
impl BackendObservationState {
fn new() -> Self {
Self { next_sequence: 1 }
}
}
#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)] #[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)]
struct ObservationKey { struct ObservationKey {
runtime_id: String, runtime_id: String,
@ -243,14 +191,12 @@ struct ObservationKey {
#[derive(Clone)] #[derive(Clone)]
pub struct BackendObservationProxy { pub struct BackendObservationProxy {
sources: Arc<BTreeMap<ObservationKey, RuntimeObservationSourceConfig>>, sources: Arc<BTreeMap<ObservationKey, RuntimeObservationSourceConfig>>,
state: Arc<Mutex<BackendObservationState>>,
} }
impl std::fmt::Debug for BackendObservationProxy { impl std::fmt::Debug for BackendObservationProxy {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("BackendObservationProxy") f.debug_struct("BackendObservationProxy")
.field("source_count", &self.sources.len()) .field("source_count", &self.sources.len())
.field("state", &"<omitted>")
.finish() .finish()
} }
} }
@ -271,7 +217,6 @@ impl BackendObservationProxy {
.collect(); .collect();
Self { Self {
sources: Arc::new(sources), sources: Arc::new(sources),
state: Arc::new(Mutex::new(BackendObservationState::new())),
} }
} }
@ -294,41 +239,13 @@ impl BackendObservationProxy {
}) })
} }
pub fn open( pub fn map_event(&self, event: RuntimeObservationUpstreamEvent) -> ClientWorkerEventWsEnvelope {
&self, ClientWorkerEventWsEnvelope {
_runtime_id: &str, event_id: event.runtime_event_id,
_worker_id: &str,
cursor: Option<&str>,
) -> Result<(), ObservationProxyError> {
if let Some(raw) = cursor {
BackendObservationCursor::decode(raw).ok_or_else(|| {
ObservationProxyError::CursorMalformed(format!(
"malformed backend observation cursor: {raw}"
))
})?;
}
Ok(())
}
pub fn store(
&self,
event: RuntimeObservationUpstreamEvent,
) -> Result<ClientWorkerEventWsEnvelope, ObservationProxyError> {
let mut state = self.state.lock().map_err(|_| {
ObservationProxyError::RuntimeUnavailable(
"backend observation state lock poisoned".into(),
)
})?;
let sequence = state.next_sequence;
state.next_sequence += 1;
let cursor = BackendObservationCursor::new(sequence).encode();
Ok(ClientWorkerEventWsEnvelope {
cursor: cursor.clone(),
event_id: cursor,
runtime_id: event.runtime_id, runtime_id: event.runtime_id,
worker_id: event.worker_id, worker_id: event.worker_id,
payload: event.payload, payload: event.payload,
}) }
} }
} }
@ -339,11 +256,6 @@ fn map_runtime_connect_error(error: TungsteniteError) -> ObservationProxyError {
"runtime worker observation endpoint returned 404 not found".into(), "runtime worker observation endpoint returned 404 not found".into(),
) )
} }
TungsteniteError::Http(response) if response.status() == StatusCode::BAD_REQUEST => {
ObservationProxyError::CursorMalformed(
"runtime worker observation endpoint rejected the request as malformed".into(),
)
}
TungsteniteError::Http(response) => ObservationProxyError::RuntimeUnavailable(format!( TungsteniteError::Http(response) => ObservationProxyError::RuntimeUnavailable(format!(
"runtime worker observation endpoint rejected WebSocket upgrade with status {}", "runtime worker observation endpoint rejected WebSocket upgrade with status {}",
response.status() response.status()
@ -357,10 +269,9 @@ fn map_runtime_connect_error(error: TungsteniteError) -> ObservationProxyError {
fn map_runtime_diagnostic(code: String, message: String) -> ObservationProxyError { fn map_runtime_diagnostic(code: String, message: String) -> ObservationProxyError {
match code.as_str() { match code.as_str() {
"runtime.worker_not_found" => ObservationProxyError::WorkerNotFound(message), "runtime.worker_not_found" => ObservationProxyError::WorkerNotFound(message),
"runtime.cursor_malformed" => ObservationProxyError::CursorMalformed(message), "runtime.cursor_malformed"
"runtime.cursor_unknown_or_expired" | "runtime.cursor_expired" => { | "runtime.cursor_unknown_or_expired"
ObservationProxyError::CursorUnknownOrExpired(message) | "runtime.cursor_expired" => ObservationProxyError::RuntimeUnavailable(message),
}
"runtime.unavailable" => ObservationProxyError::RuntimeUnavailable(message), "runtime.unavailable" => ObservationProxyError::RuntimeUnavailable(message),
"runtime.upstream_closed" | "runtime.websocket_error" => { "runtime.upstream_closed" | "runtime.websocket_error" => {
ObservationProxyError::UpstreamDisconnect(message) ObservationProxyError::UpstreamDisconnect(message)
@ -384,15 +295,8 @@ pub struct RuntimeWsObservationClient {
impl RuntimeWsObservationClient { impl RuntimeWsObservationClient {
pub async fn connect( pub async fn connect(
source: &RuntimeObservationSourceConfig, source: &RuntimeObservationSourceConfig,
runtime_cursor: Option<&str>,
) -> Result<Self, ObservationProxyError> { ) -> Result<Self, ObservationProxyError> {
let mut endpoint = source.endpoint.clone(); let endpoint = source.endpoint.clone();
if let Some(cursor) = runtime_cursor {
let separator = if endpoint.contains('?') { '&' } else { '?' };
endpoint.push(separator);
endpoint.push_str("cursor=");
endpoint.push_str(cursor);
}
let mut request = endpoint.into_client_request().map_err(|error| { let mut request = endpoint.into_client_request().map_err(|error| {
ObservationProxyError::RuntimeUnavailable(format!( ObservationProxyError::RuntimeUnavailable(format!(
"failed to build runtime WebSocket request: {error}" "failed to build runtime WebSocket request: {error}"
@ -481,7 +385,7 @@ impl RuntimeWsObservationClient {
RuntimeObservationUpstreamEvent { RuntimeObservationUpstreamEvent {
runtime_id: self.runtime_id.clone(), runtime_id: self.runtime_id.clone(),
worker_id: self.worker_id.clone(), worker_id: self.worker_id.clone(),
runtime_cursor: envelope.cursor, runtime_event_id: envelope.event_id,
payload: envelope.payload, payload: envelope.payload,
} }
} }
@ -493,18 +397,15 @@ pub enum RuntimeObservationClient {
} }
impl RuntimeObservationClient { impl RuntimeObservationClient {
pub async fn connect( pub async fn connect(source: &RuntimeObservationSource) -> Result<Self, ObservationProxyError> {
source: &RuntimeObservationSource,
runtime_cursor: Option<&str>,
) -> Result<Self, ObservationProxyError> {
match source { match source {
RuntimeObservationSource::RemoteWs(config) => { RuntimeObservationSource::RemoteWs(config) => {
RuntimeWsObservationClient::connect(config, runtime_cursor) RuntimeWsObservationClient::connect(config)
.await .await
.map(Self::RemoteWs) .map(Self::RemoteWs)
} }
RuntimeObservationSource::Embedded(source) => { RuntimeObservationSource::Embedded(source) => {
EmbeddedObservationClient::connect(source, runtime_cursor).map(Self::Embedded) EmbeddedObservationClient::connect(source).map(Self::Embedded)
} }
} }
} }
@ -529,17 +430,8 @@ pub struct EmbeddedObservationClient {
} }
impl EmbeddedObservationClient { impl EmbeddedObservationClient {
fn connect( fn connect(source: &EmbeddedRuntimeObservationSource) -> Result<Self, ObservationProxyError> {
source: &EmbeddedRuntimeObservationSource, let cursor = source
runtime_cursor: Option<&str>,
) -> Result<Self, ObservationProxyError> {
let cursor = match runtime_cursor {
Some(raw) => WorkerObservationCursor::decode(raw).ok_or_else(|| {
ObservationProxyError::CursorMalformed(
"embedded runtime cursor is malformed".into(),
)
})?,
None => source
.runtime .runtime
.worker_observation_cursor_now(&source.worker_ref) .worker_observation_cursor_now(&source.worker_ref)
.map_err(|err| { .map_err(|err| {
@ -547,8 +439,7 @@ impl EmbeddedObservationClient {
"embedded Worker '{}' is not observable: {err}", "embedded Worker '{}' is not observable: {err}",
source.worker_id source.worker_id
)) ))
})?, })?;
};
let receiver = source let receiver = source
.runtime .runtime
.subscribe_worker_observation() .subscribe_worker_observation()
@ -559,7 +450,6 @@ impl EmbeddedObservationClient {
)) ))
})?; })?;
let mut queued = VecDeque::new(); let mut queued = VecDeque::new();
if runtime_cursor.is_none() {
let snapshot = source let snapshot = source
.runtime .runtime
.worker_observation_snapshot(&source.worker_ref) .worker_observation_snapshot(&source.worker_ref)
@ -572,16 +462,15 @@ impl EmbeddedObservationClient {
queued.push_back(RuntimeObservationUpstreamEvent { queued.push_back(RuntimeObservationUpstreamEvent {
runtime_id: source.runtime_id.clone(), runtime_id: source.runtime_id.clone(),
worker_id: source.worker_id.clone(), worker_id: source.worker_id.clone(),
runtime_cursor: cursor.encode(), runtime_event_id: "snapshot".to_string(),
payload: snapshot, payload: snapshot,
}); });
}
for event in source for event in source
.runtime .runtime
.read_worker_observation_events(&source.worker_ref, cursor) .read_worker_observation_events(&source.worker_ref, cursor)
.map_err(|err| { .map_err(|err| {
ObservationProxyError::CursorUnknownOrExpired(format!( ObservationProxyError::RuntimeUnavailable(format!(
"embedded Worker '{}' cursor is unavailable: {err}", "embedded Worker '{}' observation cursor is unavailable: {err}",
source.worker_id source.worker_id
)) ))
})? })?
@ -606,9 +495,6 @@ impl EmbeddedObservationClient {
&mut self, &mut self,
) -> Result<RuntimeObservationUpstreamEvent, ObservationProxyError> { ) -> Result<RuntimeObservationUpstreamEvent, ObservationProxyError> {
if let Some(event) = self.queued.pop_front() { if let Some(event) = self.queued.pop_front() {
if let Some(cursor) = WorkerObservationCursor::decode(&event.runtime_cursor) {
self.cursor = cursor;
}
return Ok(event); return Ok(event);
} }
loop { loop {
@ -619,7 +505,7 @@ impl EmbeddedObservationClient {
{ {
self.cursor = self.cursor =
WorkerObservationCursor::decode(&event.cursor).ok_or_else(|| { WorkerObservationCursor::decode(&event.cursor).ok_or_else(|| {
ObservationProxyError::CursorMalformed( ObservationProxyError::RuntimeUnavailable(
"embedded runtime emitted a malformed cursor".into(), "embedded runtime emitted a malformed cursor".into(),
) )
})?; })?;
@ -627,7 +513,7 @@ impl EmbeddedObservationClient {
} }
Ok(_) => continue, Ok(_) => continue,
Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => { Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {
return Err(ObservationProxyError::CursorUnknownOrExpired( return Err(ObservationProxyError::RuntimeUnavailable(
"embedded runtime observation backlog was exceeded".into(), "embedded runtime observation backlog was exceeded".into(),
)); ));
} }
@ -648,7 +534,7 @@ impl EmbeddedObservationClient {
RuntimeObservationUpstreamEvent { RuntimeObservationUpstreamEvent {
runtime_id: runtime_id.to_string(), runtime_id: runtime_id.to_string(),
worker_id: worker_id.to_string(), worker_id: worker_id.to_string(),
runtime_cursor: event.cursor, runtime_event_id: event.cursor.clone(),
payload: event.payload, payload: event.payload,
} }
} }

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

@ -34,8 +34,8 @@ use crate::hosts::{
}; };
use crate::identity::WorkspaceIdentity; use crate::identity::WorkspaceIdentity;
use crate::observation::{ use crate::observation::{
BackendObservationProxy, ClientWorkerEventWsFrame, ClientWorkerEventsWsQuery, BackendObservationProxy, ClientWorkerEventWsFrame, ObservationProxyError,
ObservationProxyError, RuntimeObservationClient, RuntimeObservationSourceConfig, RuntimeObservationClient, RuntimeObservationSourceConfig,
}; };
use crate::profile_settings::{ use crate::profile_settings::{
CreateWorkspaceProfileSourceRequest, DeleteWorkspaceProfileSourceRequest, CreateWorkspaceProfileSourceRequest, DeleteWorkspaceProfileSourceRequest,
@ -2119,17 +2119,11 @@ async fn scoped_worker_observation_ws(
ws: WebSocketUpgrade, ws: WebSocketUpgrade,
State(api): State<WorkspaceApi>, State(api): State<WorkspaceApi>,
AxumPath(path): AxumPath<ScopedRuntimeWorkerPath>, AxumPath(path): AxumPath<ScopedRuntimeWorkerPath>,
Query(query): Query<ClientWorkerEventsWsQuery>,
) -> Response { ) -> Response {
if let Err(err) = validate_workspace_scope(&api, &path.workspace_id) { if let Err(err) = validate_workspace_scope(&api, &path.workspace_id) {
return err.into_response(); return err.into_response();
} }
worker_observation_ws( worker_observation_ws(State(api), AxumPath((path.runtime_id, path.worker_id)), ws)
State(api),
AxumPath((path.runtime_id, path.worker_id)),
Query(query),
ws,
)
.await .await
.into_response() .into_response()
} }
@ -2946,17 +2940,16 @@ async fn cancel_runtime_worker(
async fn worker_observation_ws( async fn worker_observation_ws(
State(api): State<WorkspaceApi>, State(api): State<WorkspaceApi>,
AxumPath((runtime_id, worker_id)): AxumPath<(String, String)>, AxumPath((runtime_id, worker_id)): AxumPath<(String, String)>,
Query(query): Query<ClientWorkerEventsWsQuery>,
ws: WebSocketUpgrade, ws: WebSocketUpgrade,
) -> impl IntoResponse { ) -> impl IntoResponse {
match api.observation_proxy.source(&runtime_id, &worker_id) { match api.observation_proxy.source(&runtime_id, &worker_id) {
Ok(source) => ws.on_upgrade(move |socket| { Ok(source) => ws.on_upgrade(move |socket| {
worker_observation_ws_session(api.observation_proxy, source, query, socket) worker_observation_ws_session(api.observation_proxy, source, socket)
}), }),
Err(ObservationProxyError::WorkerNotFound(_)) => { Err(ObservationProxyError::WorkerNotFound(_)) => {
match api.runtime.observation_source(&runtime_id, &worker_id) { match api.runtime.observation_source(&runtime_id, &worker_id) {
Ok(source) => ws.on_upgrade(move |socket| { Ok(source) => ws.on_upgrade(move |socket| {
worker_observation_ws_session(api.observation_proxy, source, query, socket) worker_observation_ws_session(api.observation_proxy, source, socket)
}), }),
Err(error) => ApiError::from(error.into_error()).into_response(), Err(error) => ApiError::from(error.into_error()).into_response(),
} }
@ -2978,20 +2971,9 @@ async fn worker_observation_ws(
async fn worker_observation_ws_session( async fn worker_observation_ws_session(
proxy: BackendObservationProxy, proxy: BackendObservationProxy,
source: crate::observation::RuntimeObservationSource, source: crate::observation::RuntimeObservationSource,
query: ClientWorkerEventsWsQuery,
mut socket: WebSocket, mut socket: WebSocket,
) { ) {
if let Err(error) = proxy.open( let mut upstream = match RuntimeObservationClient::connect(&source).await {
source.runtime_id(),
source.worker_id(),
query.cursor.as_deref(),
) {
let _ =
send_client_ws_frame(&mut socket, ClientWorkerEventWsFrame::diagnostic(error)).await;
return;
}
let mut upstream = match RuntimeObservationClient::connect(&source, None).await {
Ok(client) => client, Ok(client) => client,
Err(error) => { Err(error) => {
let _ = send_client_ws_frame(&mut socket, ClientWorkerEventWsFrame::diagnostic(error)) let _ = send_client_ws_frame(&mut socket, ClientWorkerEventWsFrame::diagnostic(error))
@ -3033,16 +3015,11 @@ async fn worker_observation_ws_session(
} }
upstream_event = upstream.next_event() => { upstream_event = upstream.next_event() => {
match upstream_event { match upstream_event {
Ok(event) => match proxy.store(event) { Ok(event) => {
Ok(envelope) => { let envelope = proxy.map_event(event);
if !send_client_ws_frame(&mut socket, ClientWorkerEventWsFrame::event(envelope)).await { if !send_client_ws_frame(&mut socket, ClientWorkerEventWsFrame::event(envelope)).await {
return; return;
} }
}
Err(error) => {
let _ = send_client_ws_frame(&mut socket, ClientWorkerEventWsFrame::diagnostic(error)).await;
return;
}
}, },
Err(error) => { Err(error) => {
let _ = send_client_ws_frame(&mut socket, ClientWorkerEventWsFrame::diagnostic(error)).await; let _ = send_client_ws_frame(&mut socket, ClientWorkerEventWsFrame::diagnostic(error)).await;
@ -7266,38 +7243,13 @@ mod tests {
ClientWorkerEventWsFrame::Event { envelope } if matches!(envelope.payload, protocol::Event::TextDone { .. }) ClientWorkerEventWsFrame::Event { envelope } if matches!(envelope.payload, protocol::Event::TextDone { .. })
)); ));
let (mut resumed, _) = connect_async(format!("{url}?cursor={}", live.cursor)) let (mut query_stream, _) = connect_async(format!("{url}?cursor=bad")).await.unwrap();
.await let query_snapshot = next_client_frame(&mut query_stream).await;
.unwrap();
let _snapshot = next_client_frame(&mut resumed).await;
runtime
.observe_worker_event(
&worker_ref,
protocol::Event::TextDone {
text: "done".into(),
},
)
.unwrap();
let resumed_event = next_client_frame(&mut resumed).await;
let ClientWorkerEventWsFrame::Event {
envelope: resumed_event,
} = resumed_event
else {
panic!("expected resumed live event");
};
assert_ne!(resumed_event.cursor, live.cursor);
assert!(matches!( assert!(matches!(
resumed_event.payload, query_snapshot,
protocol::Event::TextDone { .. } ClientWorkerEventWsFrame::Event { envelope } if matches!(envelope.payload, protocol::Event::Snapshot { .. })
)); ));
let (mut malformed, _) = connect_async(format!("{url}?cursor=bad")).await.unwrap();
let diagnostic = next_client_frame(&mut malformed).await;
let ClientWorkerEventWsFrame::Diagnostic { diagnostic } = diagnostic else {
panic!("expected malformed cursor diagnostic");
};
assert_eq!(diagnostic.code, "backend.cursor_malformed");
stream.send(Message::Text("{}".into())).await.unwrap(); stream.send(Message::Text("{}".into())).await.unwrap();
let mut saw_observation_only = false; let mut saw_observation_only = false;
for _ in 0..3 { for _ in 0..3 {
@ -7312,41 +7264,6 @@ mod tests {
assert!(saw_observation_only, "expected observation-only diagnostic"); assert!(saw_observation_only, "expected observation-only diagnostic");
} }
#[tokio::test]
async fn proxy_does_not_validate_backend_cursor_against_replay_history() {
let source = RuntimeObservationSourceConfig {
runtime_id: "runtime-a".into(),
worker_id: "worker-a".into(),
endpoint: "ws://127.0.0.1:9/not-used".into(),
bearer_token: None,
};
let (url, _dir) = spawn_workspace_proxy(source).await;
let (mut stream, _) = connect_async(format!("{url}?cursor=bo_ffffffffffffffff"))
.await
.unwrap();
let diagnostic = next_client_diagnostic(&mut stream).await;
assert_eq!(diagnostic.code, "backend.runtime_unavailable");
}
#[tokio::test]
async fn proxy_maps_runtime_cursor_diagnostic_to_typed_backend_diagnostic() {
let (_runtime, _worker_ref, endpoint) = spawn_runtime_worker().await;
let source = RuntimeObservationSourceConfig {
runtime_id: "runtime-a".into(),
worker_id: "worker-a".into(),
endpoint: format!("{endpoint}?cursor=wo_ffffffffffffffff"),
bearer_token: None,
};
let (url, _dir) = spawn_workspace_proxy(source).await;
let (mut stream, _) = connect_async(&url).await.unwrap();
assert!(matches!(
next_client_frame(&mut stream).await,
ClientWorkerEventWsFrame::Event { envelope } if matches!(envelope.payload, protocol::Event::Snapshot { .. })
));
let diagnostic = next_client_diagnostic(&mut stream).await;
assert_eq!(diagnostic.code, "backend.cursor_unknown_or_expired");
}
#[tokio::test] #[tokio::test]
async fn proxy_maps_runtime_worker_not_found_http_404_to_typed_backend_diagnostic() { async fn proxy_maps_runtime_worker_not_found_http_404_to_typed_backend_diagnostic() {
let (_runtime, _worker_ref, endpoint) = spawn_runtime_worker().await; let (_runtime, _worker_ref, endpoint) = spawn_runtime_worker().await;

52
web/workspace/src/lib/workspace-console/model.test.ts
View File

@ -56,28 +56,28 @@ Deno.test("segmentsToText preserves protocol segment semantics", () => {
Deno.test("projectConsole projects visible protocol rows", () => { Deno.test("projectConsole projects visible protocol rows", () => {
const projection = projectConsole([ const projection = projectConsole([
{ {
cursor: "10", eventId: "10",
event: { event: {
event: "user_message", event: "user_message",
data: { segments: [{ kind: "text", content: "input" }] }, data: { segments: [{ kind: "text", content: "input" }] },
} satisfies Event, } satisfies Event,
}, },
{ {
cursor: "11", eventId: "11",
event: { event: {
event: "text_delta", event: "text_delta",
data: { text: "stream" }, data: { text: "stream" },
} satisfies Event, } satisfies Event,
}, },
{ {
cursor: "12", eventId: "12",
event: { event: {
event: "thinking_done", event: "thinking_done",
data: { text: "reasoning" }, data: { text: "reasoning" },
} satisfies Event, } satisfies Event,
}, },
{ {
cursor: "13", eventId: "13",
event: { event: {
event: "tool_result", event: "tool_result",
data: { data: {
@ -89,14 +89,14 @@ Deno.test("projectConsole projects visible protocol rows", () => {
} satisfies Event, } satisfies Event,
}, },
{ {
cursor: "14", eventId: "14",
event: { event: {
event: "usage", event: "usage",
data: { input_tokens: 12, output_tokens: 5 }, data: { input_tokens: 12, output_tokens: 5 },
} satisfies Event, } satisfies Event,
}, },
{ {
cursor: "15", eventId: "15",
event: { event: {
event: "error", event: "error",
data: { code: "invalid_request", message: "bad frame" }, data: { code: "invalid_request", message: "bad frame" },
@ -141,35 +141,35 @@ Deno.test("projectConsole projects visible protocol rows", () => {
Deno.test("projectConsole groups tool call lifecycle into one Call block", () => { Deno.test("projectConsole groups tool call lifecycle into one Call block", () => {
const projection = projectConsole([ const projection = projectConsole([
{ {
cursor: "40", eventId: "40",
event: { event: {
event: "tool_call_start", event: "tool_call_start",
data: { id: "call-1", name: "Bash" }, data: { id: "call-1", name: "Bash" },
} satisfies Event, } satisfies Event,
}, },
{ {
cursor: "41", eventId: "41",
event: { event: {
event: "tool_call_args_delta", event: "tool_call_args_delta",
data: { id: "call-1", json: '{"command":"pw' }, data: { id: "call-1", json: '{"command":"pw' },
} satisfies Event, } satisfies Event,
}, },
{ {
cursor: "42", eventId: "42",
event: { event: {
event: "tool_call_args_delta", event: "tool_call_args_delta",
data: { id: "call-1", json: 'd"}' }, data: { id: "call-1", json: 'd"}' },
} satisfies Event, } satisfies Event,
}, },
{ {
cursor: "43", eventId: "43",
event: { event: {
event: "tool_call_done", event: "tool_call_done",
data: { id: "call-1", name: "Bash", arguments: '{"command":"pwd"}' }, data: { id: "call-1", name: "Bash", arguments: '{"command":"pwd"}' },
} satisfies Event, } satisfies Event,
}, },
{ {
cursor: "44", eventId: "44",
event: { event: {
event: "tool_result", event: "tool_result",
data: { data: {
@ -206,14 +206,14 @@ Deno.test("projectConsole groups tool call lifecycle into one Call block", () =>
Deno.test("projectConsole keeps streaming tool call updates in the same Call block", () => { Deno.test("projectConsole keeps streaming tool call updates in the same Call block", () => {
const projection = projectConsole([ const projection = projectConsole([
{ {
cursor: "45", eventId: "45",
event: { event: {
event: "tool_call_start", event: "tool_call_start",
data: { id: "call-2", name: "Read" }, data: { id: "call-2", name: "Read" },
} satisfies Event, } satisfies Event,
}, },
{ {
cursor: "46", eventId: "46",
event: { event: {
event: "tool_call_args_delta", event: "tool_call_args_delta",
data: { id: "call-2", json: '{"file_path":"/tmp/a.md"}' }, data: { id: "call-2", json: '{"file_path":"/tmp/a.md"}' },
@ -235,7 +235,7 @@ Deno.test("projectConsole keeps streaming tool call updates in the same Call blo
Deno.test("projectConsole aggregates Read calls without showing file content", () => { Deno.test("projectConsole aggregates Read calls without showing file content", () => {
const projection = projectConsole([ const projection = projectConsole([
{ {
cursor: "50", eventId: "50",
event: { event: {
event: "tool_call_done", event: "tool_call_done",
data: { data: {
@ -246,7 +246,7 @@ Deno.test("projectConsole aggregates Read calls without showing file content", (
} satisfies Event, } satisfies Event,
}, },
{ {
cursor: "51", eventId: "51",
event: { event: {
event: "tool_result", event: "tool_result",
data: { data: {
@ -258,7 +258,7 @@ Deno.test("projectConsole aggregates Read calls without showing file content", (
} satisfies Event, } satisfies Event,
}, },
{ {
cursor: "52", eventId: "52",
event: { event: {
event: "tool_call_done", event: "tool_call_done",
data: { data: {
@ -269,7 +269,7 @@ Deno.test("projectConsole aggregates Read calls without showing file content", (
} satisfies Event, } satisfies Event,
}, },
{ {
cursor: "53", eventId: "53",
event: { event: {
event: "tool_result", event: "tool_result",
data: { data: {
@ -307,7 +307,7 @@ Deno.test("projectConsole aggregates Read calls without showing file content", (
Deno.test("projectConsole renders Edit calls with structured diff lines", () => { Deno.test("projectConsole renders Edit calls with structured diff lines", () => {
const projection = projectConsole([ const projection = projectConsole([
{ {
cursor: "60", eventId: "60",
event: { event: {
event: "tool_call_done", event: "tool_call_done",
data: { data: {
@ -322,7 +322,7 @@ Deno.test("projectConsole renders Edit calls with structured diff lines", () =>
} satisfies Event, } satisfies Event,
}, },
{ {
cursor: "61", eventId: "61",
event: { event: {
event: "tool_result", event: "tool_result",
data: { data: {
@ -350,7 +350,7 @@ Deno.test("projectConsole renders Edit calls with structured diff lines", () =>
Deno.test("projectConsole preserves in-progress assistant protocol stream", () => { Deno.test("projectConsole preserves in-progress assistant protocol stream", () => {
const projection = projectConsole([ const projection = projectConsole([
{ {
cursor: "13", eventId: "13",
event: { event: "text_delta", data: { text: "new" } } satisfies Event, event: { event: "text_delta", data: { text: "new" } } satisfies Event,
}, },
]); ]);
@ -367,26 +367,26 @@ Deno.test("projectConsole preserves in-progress assistant protocol stream", () =
Deno.test("projectConsole keeps protocol lifecycle events out of the console surface", () => { Deno.test("projectConsole keeps protocol lifecycle events out of the console surface", () => {
const projection = projectConsole([ const projection = projectConsole([
{ {
cursor: "30", eventId: "30",
event: { event: "status", data: { status: "running" } } satisfies Event, event: { event: "status", data: { status: "running" } } satisfies Event,
}, },
{ {
cursor: "31", eventId: "31",
event: { event: "llm_call_end", data: { llm_call: 0 } } satisfies Event, event: { event: "llm_call_end", data: { llm_call: 0 } } satisfies Event,
}, },
{ {
cursor: "32", eventId: "32",
event: { event: {
event: "turn_end", event: "turn_end",
data: { turn: 0, result: "finished" }, data: { turn: 0, result: "finished" },
} satisfies Event, } satisfies Event,
}, },
{ {
cursor: "33", eventId: "33",
event: { event: "run_end", data: { result: "finished" } } satisfies Event, event: { event: "run_end", data: { result: "finished" } } satisfies Event,
}, },
{ {
cursor: "34", eventId: "34",
event: { event: {
event: "system_item", event: "system_item",
data: { item: { kind: "note", content: "internal" } }, data: { item: { kind: "note", content: "internal" } },
@ -401,7 +401,7 @@ Deno.test("projectConsole keeps protocol lifecycle events out of the console sur
Deno.test("projectConsole uses snapshot for state without rendering it as console output", () => { Deno.test("projectConsole uses snapshot for state without rendering it as console output", () => {
const projection = projectConsole([ const projection = projectConsole([
{ {
cursor: "20", eventId: "20",
event: { event: {
event: "snapshot", event: "snapshot",
data: { data: {

86
web/workspace/src/lib/workspace-console/model.ts
View File

@ -49,7 +49,7 @@ export type ConsoleLine = {
body: string; body: string;
detail?: string; detail?: string;
diff?: ConsoleDiffLine[]; diff?: ConsoleDiffLine[];
cursor?: string | null; eventId?: string | null;
source: "event"; source: "event";
streaming?: boolean; streaming?: boolean;
error?: boolean; error?: boolean;
@ -60,7 +60,7 @@ export type ConsoleProjection = {
lines: ConsoleLine[]; lines: ConsoleLine[];
status: string | null; status: string | null;
usage: string | null; usage: string | null;
lastCursor: string | null; lastEventId: string | null;
}; };
export type WorkerTarget = { export type WorkerTarget = {
@ -93,7 +93,7 @@ export function workerConsolePath(
); );
} }
export type ConsoleEventInput = { cursor: string; event: ProtocolEvent }; export type ConsoleEventInput = { eventId: string; event: ProtocolEvent };
export function projectConsole( export function projectConsole(
events: ConsoleEventInput[] = [], events: ConsoleEventInput[] = [],
@ -102,7 +102,7 @@ export function projectConsole(
lines: [], lines: [],
status: null, status: null,
usage: null, usage: null,
lastCursor: null, lastEventId: null,
}); });
return { return {
...projection, ...projection,
@ -112,13 +112,13 @@ export function projectConsole(
export function applyProtocolEvent( export function applyProtocolEvent(
projection: ConsoleProjection, projection: ConsoleProjection,
envelope: { cursor: string; event: ProtocolEvent }, envelope: { eventId: string; event: ProtocolEvent },
): ConsoleProjection { ): ConsoleProjection {
const next: ConsoleProjection = { const next: ConsoleProjection = {
lines: [...projection.lines], lines: [...projection.lines],
status: projection.status, status: projection.status,
usage: projection.usage, usage: projection.usage,
lastCursor: envelope.cursor, lastEventId: envelope.eventId,
}; };
const event = envelope.event; const event = envelope.event;
@ -126,7 +126,7 @@ export function applyProtocolEvent(
case "user_message": case "user_message":
next.lines.push( next.lines.push(
line( line(
envelope.cursor, envelope.eventId,
"user", "user",
"User", "User",
segmentsToText(event.data.segments), segmentsToText(event.data.segments),
@ -139,7 +139,7 @@ export function applyProtocolEvent(
case "text_delta": case "text_delta":
appendStreaming( appendStreaming(
next, next,
envelope.cursor, envelope.eventId,
"assistant", "assistant",
"assistant streaming", "assistant streaming",
event.data.text, event.data.text,
@ -149,20 +149,20 @@ export function applyProtocolEvent(
finalizeStreaming( finalizeStreaming(
next, next,
"assistant", "assistant",
envelope.cursor, envelope.eventId,
"assistant", "assistant",
event.data.text, event.data.text,
); );
break; break;
case "thinking_start": case "thinking_start":
next.lines.push( next.lines.push(
line(envelope.cursor, "thinking", "Thinking...", "", undefined, true), line(envelope.eventId, "thinking", "Thinking...", "", undefined, true),
); );
break; break;
case "thinking_delta": case "thinking_delta":
appendStreaming( appendStreaming(
next, next,
envelope.cursor, envelope.eventId,
"thinking", "thinking",
"Thinking...", "Thinking...",
event.data.text, event.data.text,
@ -172,22 +172,22 @@ export function applyProtocolEvent(
finalizeStreaming( finalizeStreaming(
next, next,
"thinking", "thinking",
envelope.cursor, envelope.eventId,
"Thought", "Thought",
event.data.text, event.data.text,
); );
break; break;
case "tool_call_start": case "tool_call_start":
upsertToolCall(next, envelope.cursor, event.data.id, { upsertToolCall(next, envelope.eventId, event.data.id, {
name: event.data.name, name: event.data.name,
state: "pending", state: "pending",
}); });
break; break;
case "tool_call_args_delta": case "tool_call_args_delta":
appendToolArgs(next, envelope.cursor, event.data.id, event.data.json); appendToolArgs(next, envelope.eventId, event.data.id, event.data.json);
break; break;
case "tool_call_done": case "tool_call_done":
upsertToolCall(next, envelope.cursor, event.data.id, { upsertToolCall(next, envelope.eventId, event.data.id, {
name: event.data.name, name: event.data.name,
arguments: event.data.arguments, arguments: event.data.arguments,
argsStream: event.data.arguments, argsStream: event.data.arguments,
@ -195,7 +195,7 @@ export function applyProtocolEvent(
}); });
break; break;
case "tool_result": case "tool_result":
attachToolResult(next, envelope.cursor, event.data.id, { attachToolResult(next, envelope.eventId, event.data.id, {
summary: event.data.summary, summary: event.data.summary,
output: event.data.output, output: event.data.output,
isError: event.data.is_error, isError: event.data.is_error,
@ -207,7 +207,7 @@ export function applyProtocolEvent(
case "error": case "error":
next.lines.push( next.lines.push(
line( line(
envelope.cursor, envelope.eventId,
"error", "error",
`error · ${event.data.code}`, `error · ${event.data.code}`,
event.data.message, event.data.message,
@ -220,7 +220,7 @@ export function applyProtocolEvent(
case "snapshot": case "snapshot":
next.status = event.data.status; next.status = event.data.status;
for (const block of event.data.in_flight?.blocks ?? []) { for (const block of event.data.in_flight?.blocks ?? []) {
next.lines.push(inFlightLine(envelope.cursor, block)); next.lines.push(inFlightLine(envelope.eventId, block));
} }
break; break;
case "status": case "status":
@ -251,7 +251,7 @@ export function applyProtocolEvent(
case "compact_failed": case "compact_failed":
next.lines.push( next.lines.push(
line( line(
envelope.cursor, envelope.eventId,
"error", "error",
"compact failed", "compact failed",
event.data.error, event.data.error,
@ -292,7 +292,7 @@ export function segmentsToText(segments: Segment[]): string {
} }
function line( function line(
cursor: string, eventId: string,
kind: ConsoleLineKind, kind: ConsoleLineKind,
title: string, title: string,
body: string, body: string,
@ -301,12 +301,12 @@ function line(
error = false, error = false,
): ConsoleLine { ): ConsoleLine {
return { return {
id: `event-${cursor}-${kind}-${slugify(title)}-${body.length}`, id: `event-${eventId}-${kind}-${slugify(title)}-${body.length}`,
kind, kind,
title, title,
body, body,
detail, detail,
cursor, eventId,
source: "event", source: "event",
streaming, streaming,
error, error,
@ -315,7 +315,7 @@ function line(
function appendStreaming( function appendStreaming(
projection: ConsoleProjection, projection: ConsoleProjection,
cursor: string, eventId: string,
kind: "assistant" | "thinking", kind: "assistant" | "thinking",
title: string, title: string,
delta: string, delta: string,
@ -325,16 +325,16 @@ function appendStreaming(
); );
if (existing) { if (existing) {
existing.body += delta; existing.body += delta;
existing.cursor = cursor; existing.eventId = eventId;
return; return;
} }
projection.lines.push(line(cursor, kind, title, delta, undefined, true)); projection.lines.push(line(eventId, kind, title, delta, undefined, true));
} }
function finalizeStreaming( function finalizeStreaming(
projection: ConsoleProjection, projection: ConsoleProjection,
kind: "assistant" | "thinking", kind: "assistant" | "thinking",
cursor: string, eventId: string,
title: string, title: string,
body: string, body: string,
): void { ): void {
@ -345,21 +345,21 @@ function finalizeStreaming(
existing.body = body || existing.body; existing.body = body || existing.body;
existing.streaming = false; existing.streaming = false;
existing.title = title; existing.title = title;
existing.cursor = cursor; existing.eventId = eventId;
return; return;
} }
projection.lines.push(line(cursor, kind, title, body)); projection.lines.push(line(eventId, kind, title, body));
} }
function upsertToolCall( function upsertToolCall(
projection: ConsoleProjection, projection: ConsoleProjection,
cursor: string, eventId: string,
id: string, id: string,
update: Partial<Omit<ToolCallView, "id">>, update: Partial<Omit<ToolCallView, "id">>,
): ConsoleLine { ): ConsoleLine {
let existing = findToolCallLine(projection, id); let existing = findToolCallLine(projection, id);
if (!existing) { if (!existing) {
existing = toolLine(cursor, { existing = toolLine(eventId, {
id, id,
name: update.name ?? "Tool", name: update.name ?? "Tool",
argsStream: update.argsStream ?? "", argsStream: update.argsStream ?? "",
@ -371,7 +371,7 @@ function upsertToolCall(
}); });
projection.lines.push(existing); projection.lines.push(existing);
} else { } else {
existing.cursor = cursor; existing.eventId = eventId;
existing.toolCall = { existing.toolCall = {
...existing.toolCall!, ...existing.toolCall!,
...update, ...update,
@ -385,11 +385,11 @@ function upsertToolCall(
function appendToolArgs( function appendToolArgs(
projection: ConsoleProjection, projection: ConsoleProjection,
cursor: string, eventId: string,
id: string, id: string,
delta: string, delta: string,
): void { ): void {
const existing = upsertToolCall(projection, cursor, id, { const existing = upsertToolCall(projection, eventId, id, {
state: "streaming_args", state: "streaming_args",
}); });
existing.toolCall!.argsStream += delta; existing.toolCall!.argsStream += delta;
@ -398,13 +398,13 @@ function appendToolArgs(
function attachToolResult( function attachToolResult(
projection: ConsoleProjection, projection: ConsoleProjection,
cursor: string, eventId: string,
id: string, id: string,
result: Pick<ToolCallView, "summary" | "output" | "isError">, result: Pick<ToolCallView, "summary" | "output" | "isError">,
): void { ): void {
const existing = findToolCallLine(projection, id); const existing = findToolCallLine(projection, id);
if (!existing) { if (!existing) {
const fallback = toolLine(cursor, { const fallback = toolLine(eventId, {
id, id,
name: "Tool", name: "Tool",
argsStream: "", argsStream: "",
@ -418,7 +418,7 @@ function attachToolResult(
projection.lines.push(fallback); projection.lines.push(fallback);
return; return;
} }
existing.cursor = cursor; existing.eventId = eventId;
existing.toolCall = { existing.toolCall = {
...existing.toolCall!, ...existing.toolCall!,
...result, ...result,
@ -436,14 +436,14 @@ function findToolCallLine(
); );
} }
function toolLine(cursor: string, toolCall: ToolCallView): ConsoleLine { function toolLine(eventId: string, toolCall: ToolCallView): ConsoleLine {
const item: ConsoleLine = { const item: ConsoleLine = {
id: `tool-call-${toolCall.id}`, id: `tool-call-${toolCall.id}`,
kind: "tool", kind: "tool",
title: `Call · ${toolCall.name}`, title: `Call · ${toolCall.name}`,
body: "", body: "",
detail: undefined, detail: undefined,
cursor, eventId,
source: "event", source: "event",
streaming: true, streaming: true,
error: false, error: false,
@ -531,7 +531,7 @@ function readAggregateLine(group: ConsoleLine[]): ConsoleLine {
title: "Call · Read", title: "Call · Read",
body, body,
detail: calls.map(readDetail).join("\n\n"), detail: calls.map(readDetail).join("\n\n"),
cursor: group.at(-1)?.cursor, eventId: group.at(-1)?.eventId,
source: "event", source: "event",
streaming: inProgress, streaming: inProgress,
error: hasError, error: hasError,
@ -760,11 +760,11 @@ function usageText(
} · cache ${data.cache_read_input_tokens ?? "unknown"}`; } · cache ${data.cache_read_input_tokens ?? "unknown"}`;
} }
function inFlightLine(cursor: string, block: InFlightBlock): ConsoleLine { function inFlightLine(eventId: string, block: InFlightBlock): ConsoleLine {
switch (block.kind) { switch (block.kind) {
case "text": case "text":
return line( return line(
cursor, eventId,
"in_flight", "in_flight",
"in-flight assistant text", "in-flight assistant text",
block.text, block.text,
@ -773,7 +773,7 @@ function inFlightLine(cursor: string, block: InFlightBlock): ConsoleLine {
); );
case "thinking": case "thinking":
return line( return line(
cursor, eventId,
"in_flight", "in_flight",
"in-flight thinking", "in-flight thinking",
block.text, block.text,
@ -781,7 +781,7 @@ function inFlightLine(cursor: string, block: InFlightBlock): ConsoleLine {
!block.finished, !block.finished,
); );
case "tool_call": case "tool_call":
return toolLine(cursor, { return toolLine(eventId, {
id: block.id, id: block.id,
name: block.name, name: block.name,
argsStream: block.args, argsStream: block.args,

1
web/workspace/src/lib/workspace-sidebar/types.ts
View File

@ -253,7 +253,6 @@ export type WorkerInputResult = {
}; };
export type ClientWorkerEventWsEnvelope = { export type ClientWorkerEventWsEnvelope = {
cursor: string;
event_id: string; event_id: string;
runtime_id: string; runtime_id: string;
worker_id: string; worker_id: string;

6
web/workspace/src/routes/w/[workspaceId]/runtimes/[runtimeId]/workers/[workerId]/console/+page.svelte
View File

@ -42,7 +42,7 @@
let consoleBodyElement: HTMLElement | null = null; let consoleBodyElement: HTMLElement | null = null;
let autoFollowConsole = $state(true); let autoFollowConsole = $state(true);
const CONSOLE_BOTTOM_THRESHOLD_PX = 48; const CONSOLE_BOTTOM_THRESHOLD_PX = 48;
let observedEvents = $state<Array<{ cursor: string; event: ClientWorkerEventWsFrame & { kind: 'event' } }>>([]); let observedEvents = $state<Array<{ eventId: string; event: ClientWorkerEventWsFrame & { kind: 'event' } }>>([]);
let seenObservationEventIds = new Set<string>(); let seenObservationEventIds = new Set<string>();
let nextReloadToken = 0; let nextReloadToken = 0;
let reloadToken = $state(0); let reloadToken = $state(0);
@ -55,7 +55,7 @@
const consoleTarget = $derived({ runtimeId, workerId }); const consoleTarget = $derived({ runtimeId, workerId });
const projection = $derived( const projection = $derived(
projectConsole(observedEvents.map((item) => ({ cursor: item.cursor, event: item.event.envelope.payload }))) projectConsole(observedEvents.map((item) => ({ eventId: item.eventId, event: item.event.envelope.payload })))
); );
const lines = $derived(projection.lines); const lines = $derived(projection.lines);
const diagnostics = $derived(mergeDiagnostics(worker?.diagnostics ?? [], streamDiagnostics)); const diagnostics = $derived(mergeDiagnostics(worker?.diagnostics ?? [], streamDiagnostics));
@ -185,7 +185,7 @@
observedEvents = [ observedEvents = [
...observedEvents, ...observedEvents,
{ {
cursor: frame.envelope.cursor, eventId: frame.envelope.event_id,
event: frame event: frame
} }
].slice(-500); ].slice(-500);