From 156f8ad044c60bcca2047889fdcbe450c7e3c457 Mon Sep 17 00:00:00 2001 From: Hare Date: Sat, 11 Jul 2026 03:55:44 +0900 Subject: [PATCH] runtime: remove worker transcript projection --- crates/client/src/backend_runtime.rs | 146 +------ crates/worker-runtime/src/catalog.rs | 10 +- crates/worker-runtime/src/execution.rs | 2 +- crates/worker-runtime/src/fs_store.rs | 129 +++---- crates/worker-runtime/src/http_server.rs | 55 +-- crates/worker-runtime/src/interaction.rs | 6 +- crates/worker-runtime/src/main.rs | 5 - crates/worker-runtime/src/management.rs | 4 +- crates/worker-runtime/src/observation.rs | 43 --- crates/worker-runtime/src/runtime.rs | 360 ++++++------------ crates/worker-runtime/src/worker_backend.rs | 20 +- crates/workspace-server/src/hosts.rs | 231 +---------- crates/workspace-server/src/server.rs | 101 ++--- web/workspace/src/app.css | 14 +- .../src/lib/workspace-console/model.test.ts | 218 ++++------- .../src/lib/workspace-console/model.ts | 106 +----- .../worker-console.ui.test.ts | 33 +- .../DecodalSourceEditor.svelte | 14 +- .../src/lib/workspace-sidebar/types.ts | 20 - .../workers/[workerId]/console/+page.svelte | 90 +++-- 20 files changed, 428 insertions(+), 1179 deletions(-) diff --git a/crates/client/src/backend_runtime.rs b/crates/client/src/backend_runtime.rs index 9636af32..ce06125b 100644 --- a/crates/client/src/backend_runtime.rs +++ b/crates/client/src/backend_runtime.rs @@ -3,13 +3,12 @@ use std::fmt; use std::time::Duration; use futures::StreamExt; -use protocol::{ErrorCode, Event, Greeting, InFlightSnapshot, Method, Segment, WorkerStatus}; +use protocol::{ErrorCode, Event, Method, Segment}; use serde::{Deserialize, Serialize}; use tokio::sync::mpsc; use tokio_tungstenite::connect_async; use tokio_tungstenite::tungstenite::Message as TungsteniteMessage; -const TRANSCRIPT_SNAPSHOT_LIMIT: usize = 512; const RECONNECT_DELAY: Duration = Duration::from_millis(500); const MAX_RECONNECT_ATTEMPTS: usize = 3; @@ -80,31 +79,10 @@ impl BackendRuntimeClient { let http = reqwest::Client::new(); let (tx, rx) = mpsc::unbounded_channel(); - let suppress_initial_snapshot = match load_initial_transcript(&http, &target).await { - Ok(events) => { - for event in events { - let _ = tx.send(event); - } - true - } - Err(error) => { - let _ = tx.send(diagnostic_event(format!( - "Backend initial transcript unavailable for {}: {error}", - target.display_label() - ))); - false - } - }; - let observation_target = target.clone(); let observation_tx = tx.clone(); let observation_task = tokio::spawn(async move { - observe_worker_events( - observation_target, - observation_tx, - suppress_initial_snapshot, - ) - .await; + observe_worker_events(observation_target, observation_tx).await; }); Ok(Self { @@ -291,80 +269,7 @@ fn backend_command_from_method(method: &Method) -> BackendCommand { } } -async fn load_initial_transcript( - http: &reqwest::Client, - target: &BackendRuntimeTarget, -) -> Result, BackendRuntimeClientError> { - let path = format!( - "/api/runtimes/{}/workers/{}/transcript?start=0&limit={TRANSCRIPT_SNAPSHOT_LIMIT}", - path_segment_encode(&target.runtime_id), - path_segment_encode(&target.worker_id) - ); - let response = http - .get(join_base_and_path(&target.base_url, &path)) - .send() - .await? - .error_for_status()?; - let transcript: WorkerTranscriptProjection = response.json().await?; - Ok(transcript_projection_to_events(target, transcript)) -} - -fn transcript_projection_to_events( - target: &BackendRuntimeTarget, - transcript: WorkerTranscriptProjection, -) -> Vec { - let mut events = vec![Event::Snapshot { - entries: Vec::new(), - greeting: Greeting { - worker_name: target.worker_id.clone(), - cwd: String::new(), - provider: "backend-runtime-api".to_string(), - model: target.runtime_id.clone(), - scope_summary: "Backend Runtime API worker observation".to_string(), - tools: Vec::new(), - context_window: 0, - context_tokens: 0, - }, - status: WorkerStatus::Idle, - in_flight: InFlightSnapshot { blocks: Vec::new() }, - }]; - - for item in transcript.items { - match item.role.as_str() { - "user" => events.push(Event::UserMessage { - segments: vec![Segment::text(item.content)], - }), - "assistant" => { - events.push(Event::TextDelta { - text: item.content.clone(), - }); - events.push(Event::TextDone { text: item.content }); - } - role => events.push(Event::Alert(protocol::Alert { - level: protocol::AlertLevel::Warn, - source: protocol::AlertSource::Worker, - message: format!( - "Backend transcript item with role `{role}` is not rendered as chat content" - ), - timestamp_ms: 0, - })), - } - } - - for diagnostic in transcript.diagnostics { - events.push(diagnostic_event(format!( - "Backend transcript diagnostic [{}]: {}", - diagnostic.code, diagnostic.message - ))); - } - events -} - -async fn observe_worker_events( - target: BackendRuntimeTarget, - tx: mpsc::UnboundedSender, - mut suppress_next_snapshot: bool, -) { +async fn observe_worker_events(target: BackendRuntimeTarget, tx: mpsc::UnboundedSender) { let mut cursor: Option = None; let mut last_sequence = 0_u64; let mut attempts = 0_usize; @@ -403,12 +308,6 @@ async fn observe_worker_events( ))); } cursor = Some(envelope.cursor.clone()); - if suppress_next_snapshot - && matches!(envelope.payload, Event::Snapshot { .. }) - { - suppress_next_snapshot = false; - continue; - } let _ = tx.send(envelope.payload); } Ok(ClientWorkerEventWsFrame::Diagnostic { diagnostic }) => { @@ -595,20 +494,6 @@ struct BackendDiagnostic { message: String, } -#[derive(Debug, Deserialize)] -struct WorkerTranscriptProjection { - #[serde(default)] - items: Vec, - #[serde(default)] - diagnostics: Vec, -} - -#[derive(Debug, Deserialize)] -struct WorkerTranscriptItem { - role: String, - content: String, -} - #[derive(Debug, Deserialize)] #[serde(tag = "kind", rename_all = "snake_case")] enum ClientWorkerEventWsFrame { @@ -666,29 +551,4 @@ mod tests { "ws://127.0.0.1:8787/api/runtimes/runtime%2Fone/workers/worker%20one/events/ws?cursor=bo_0000000000000001" ); } - - #[test] - fn transcript_projection_seeds_snapshot_and_chat_events() { - let target = BackendRuntimeTarget::new("http://backend", "runtime-a", "worker-b"); - let events = transcript_projection_to_events( - &target, - WorkerTranscriptProjection { - items: vec![ - WorkerTranscriptItem { - role: "user".to_string(), - content: "hi".to_string(), - }, - WorkerTranscriptItem { - role: "assistant".to_string(), - content: "hello".to_string(), - }, - ], - diagnostics: Vec::new(), - }, - ); - assert!(matches!(events[0], Event::Snapshot { .. })); - assert!(matches!(events[1], Event::UserMessage { .. })); - assert!(matches!(events[2], Event::TextDelta { .. })); - assert!(matches!(events[3], Event::TextDone { .. })); - } } diff --git a/crates/worker-runtime/src/catalog.rs b/crates/worker-runtime/src/catalog.rs index 7837a9ef..744f8dfd 100644 --- a/crates/worker-runtime/src/catalog.rs +++ b/crates/worker-runtime/src/catalog.rs @@ -170,10 +170,10 @@ pub struct WorkingDirectoryStatus { /// Browser/product launch semantics are resolved by a backend before this /// request is built. The request contains only durable Runtime identity inputs: /// a backend-decided profile selector, the Decodal profile source archive source -/// used to resolve that selector, optional initial user input committed with the -/// Worker catalog/transcript persistence, and an optional Runtime-owned working -/// directory binding. Browser-facing status for materialized working directories -/// is summarized without exposing raw host paths. +/// used to resolve that selector, optional initial user input committed as a +/// protocol observation event, and an optional Runtime-owned working directory +/// binding. Browser-facing status for materialized working directories is +/// summarized without exposing raw host paths. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct CreateWorkerRequest { pub profile: ProfileSelector, @@ -215,7 +215,6 @@ pub struct WorkerSummary { pub profile_source: ProfileSourceArchiveRef, #[serde(default, skip_serializing_if = "Option::is_none")] pub config_bundle: Option, - pub transcript_len: usize, pub last_event_id: u64, } @@ -231,7 +230,6 @@ pub struct WorkerDetail { pub profile_source: ProfileSourceArchiveRef, #[serde(default, skip_serializing_if = "Option::is_none")] pub config_bundle: Option, - pub transcript_len: usize, pub last_event_id: u64, } diff --git a/crates/worker-runtime/src/execution.rs b/crates/worker-runtime/src/execution.rs index e0fd79a6..1bbdf60b 100644 --- a/crates/worker-runtime/src/execution.rs +++ b/crates/worker-runtime/src/execution.rs @@ -305,7 +305,7 @@ pub enum WorkerExecutionSpawnResult { /// Backend boundary for Worker execution. /// -/// Runtime owns Worker catalog, transcript, observation, and lifecycle state. A +/// Runtime owns Worker catalog, protocol observation, and lifecycle state. A /// backend owns concrete execution. The default Runtime has no backend, so input /// to those Workers is rejected instead of producing providerless responses. pub trait WorkerExecutionBackend: Send + Sync + 'static { diff --git a/crates/worker-runtime/src/fs_store.rs b/crates/worker-runtime/src/fs_store.rs index 4578eb18..c94374b7 100644 --- a/crates/worker-runtime/src/fs_store.rs +++ b/crates/worker-runtime/src/fs_store.rs @@ -5,10 +5,9 @@ use crate::error::RuntimeError; use crate::execution::WorkerExecutionStatus; use crate::identity::{RuntimeId, WorkerId, WorkerRef}; use crate::management::{RuntimeBackendKind, RuntimeLimits, RuntimeStatus}; -use crate::observation::{ - EventCursor, RuntimeEvent, RuntimeEventBatch, TranscriptEntry, TranscriptProjection, - TranscriptQuery, -}; +#[cfg(feature = "ws-server")] +use crate::observation::WorkerObservationEvent; +use crate::observation::{EventCursor, RuntimeEvent, RuntimeEventBatch}; use serde::{Deserialize, Serialize}; use std::collections::BTreeMap; use std::fs::{self, File, OpenOptions}; @@ -22,7 +21,8 @@ const RUNTIME_FILE: &str = "runtime.json"; const EVENTS_FILE: &str = "events.jsonl"; const WORKERS_DIR: &str = "workers"; const WORKER_FILE: &str = "worker.json"; -const TRANSCRIPT_FILE: &str = "transcript.jsonl"; +#[cfg(feature = "ws-server")] +const OBSERVATIONS_FILE: &str = "observations.jsonl"; static NEXT_TMP_SEQUENCE: AtomicU64 = AtomicU64::new(1); @@ -119,42 +119,6 @@ impl FsRuntimeStore { }) } - /// Read a persisted Worker transcript directly from its Worker-scoped log. - pub fn read_transcript( - &self, - worker_ref: &WorkerRef, - query: TranscriptQuery, - max_limit: usize, - ) -> Result { - self.ensure_worker_ref(worker_ref)?; - if query.limit > max_limit { - return Err(RuntimeError::LimitTooLarge { - requested: query.limit, - max: max_limit, - }); - } - - let path = self.transcript_path(&worker_ref.worker_id); - let entries = read_json_lines::(&path, "read transcript")?; - let total_items = entries.len(); - let end = query.start.saturating_add(query.limit).min(total_items); - let items = if query.start >= total_items { - Vec::new() - } else { - entries[query.start..end].to_vec() - }; - let next_start = (end < total_items).then_some(end); - - Ok(TranscriptProjection { - worker_ref: worker_ref.clone(), - start: query.start, - limit: query.limit, - total_items, - items, - next_start, - }) - } - pub(crate) fn open_or_create( root: PathBuf, runtime_id: RuntimeId, @@ -220,7 +184,12 @@ impl FsRuntimeStore { &WorkerSnapshot::from_persisted(worker), "write worker snapshot", )?; - ensure_file_exists(&worker_dir.join(TRANSCRIPT_FILE), "create transcript log") + #[cfg(feature = "ws-server")] + ensure_file_exists( + &worker_dir.join(OBSERVATIONS_FILE), + "create observations log", + )?; + Ok(()) } pub(crate) fn append_event(&self, event: &RuntimeEvent) -> Result<(), RuntimeError> { @@ -230,15 +199,16 @@ impl FsRuntimeStore { append_json_line(&self.events_path(), event, "append event") } - pub(crate) fn append_transcript_entry( + #[cfg(feature = "ws-server")] + pub(crate) fn append_worker_observation_event( &self, - entry: &TranscriptEntry, + event: &WorkerObservationEvent, ) -> Result<(), RuntimeError> { - self.ensure_worker_ref(&entry.worker_ref)?; + self.ensure_worker_ref(&event.worker_ref)?; append_json_line( - &self.transcript_path(&entry.worker_ref.worker_id), - entry, - "append transcript", + &self.observations_path(&event.worker_ref.worker_id), + event, + "append worker observation", ) } @@ -278,6 +248,9 @@ impl FsRuntimeStore { })?; worker_dirs.sort_by_key(|entry| entry.path()); + #[cfg(feature = "ws-server")] + let mut observation_events = Vec::new(); + for entry in worker_dirs { let path = entry.path(); if !path.is_dir() { @@ -312,38 +285,44 @@ impl FsRuntimeStore { ); continue; } - let transcript = match read_json_lines::( - &path.join(TRANSCRIPT_FILE), - "read transcript", + #[cfg(feature = "ws-server")] + let worker_observations = match read_json_lines::( + &path.join(OBSERVATIONS_FILE), + "read worker observations", ) { - Ok(transcript) => transcript, + Ok(events) => events, Err(_error) => { record_worker_load_diagnostic( &mut snapshot, Some(worker_snapshot.worker_ref.clone()), - "ignored worker with unreadable transcript while loading runtime store", + "ignored worker with unreadable observations while loading runtime store", ); continue; } }; - let mut transcript_valid = true; - for entry in &transcript { - if self.ensure_worker_ref(&entry.worker_ref).is_err() - || entry.worker_ref.worker_id != worker_snapshot.worker_id + #[cfg(feature = "ws-server")] + let mut observations_valid = true; + #[cfg(feature = "ws-server")] + for event in &worker_observations { + if self.ensure_worker_ref(&event.worker_ref).is_err() + || event.worker_ref.worker_id != worker_snapshot.worker_id { - transcript_valid = false; + observations_valid = false; break; } } - if !transcript_valid { + #[cfg(feature = "ws-server")] + if !observations_valid { record_worker_load_diagnostic( &mut snapshot, Some(worker_snapshot.worker_ref.clone()), - "ignored worker with invalid transcript while loading runtime store", + "ignored worker with invalid observations while loading runtime store", ); continue; } - let worker = worker_snapshot.into_persisted(transcript); + #[cfg(feature = "ws-server")] + observation_events.extend(worker_observations); + let worker = worker_snapshot.into_persisted(); if workers.insert(worker.worker_id.clone(), worker).is_some() { record_worker_load_diagnostic( &mut snapshot, @@ -353,7 +332,15 @@ impl FsRuntimeStore { } } - Ok(snapshot.into_persisted(events, workers)) + #[cfg(feature = "ws-server")] + observation_events.sort_by_key(|event| event.sequence); + + Ok(snapshot.into_persisted( + events, + workers, + #[cfg(feature = "ws-server")] + observation_events, + )) } fn ensure_worker_ref(&self, worker_ref: &WorkerRef) -> Result<(), RuntimeError> { @@ -382,8 +369,9 @@ impl FsRuntimeStore { .join(encoded_component(worker_id.as_str())) } - fn transcript_path(&self, worker_id: &WorkerId) -> PathBuf { - self.worker_dir(worker_id).join(TRANSCRIPT_FILE) + #[cfg(feature = "ws-server")] + fn observations_path(&self, worker_id: &WorkerId) -> PathBuf { + self.worker_dir(worker_id).join(OBSERVATIONS_FILE) } } @@ -405,6 +393,8 @@ pub(crate) struct PersistedRuntimeState { pub(crate) workers: BTreeMap, pub(crate) config_bundles: BTreeMap, pub(crate) events: Vec, + #[cfg(feature = "ws-server")] + pub(crate) observation_events: Vec, pub(crate) diagnostics: Vec, } @@ -415,8 +405,6 @@ pub(crate) struct PersistedWorkerRecord { pub(crate) status: WorkerStatus, pub(crate) request: CreateWorkerRequest, pub(crate) execution: WorkerExecutionStatus, - pub(crate) transcript: Vec, - pub(crate) next_transcript_sequence: u64, pub(crate) last_event_id: u64, } @@ -504,6 +492,7 @@ impl RuntimeSnapshot { self, events: Vec, workers: BTreeMap, + #[cfg(feature = "ws-server")] observation_events: Vec, ) -> PersistedRuntimeState { PersistedRuntimeState { runtime_id: self.runtime_id, @@ -516,6 +505,8 @@ impl RuntimeSnapshot { workers, config_bundles: self.config_bundles, events, + #[cfg(feature = "ws-server")] + observation_events, diagnostics: self.diagnostics, } } @@ -530,7 +521,6 @@ struct WorkerSnapshot { request: CreateWorkerRequest, #[serde(default = "WorkerExecutionStatus::unconnected")] execution: WorkerExecutionStatus, - next_transcript_sequence: u64, last_event_id: u64, } @@ -543,7 +533,6 @@ impl WorkerSnapshot { status: worker.status, request: worker.request.clone(), execution: worker.execution.clone(), - next_transcript_sequence: worker.next_transcript_sequence, last_event_id: worker.last_event_id, } } @@ -582,15 +571,13 @@ impl WorkerSnapshot { Ok(()) } - fn into_persisted(self, transcript: Vec) -> PersistedWorkerRecord { + fn into_persisted(self) -> PersistedWorkerRecord { PersistedWorkerRecord { worker_ref: self.worker_ref, worker_id: self.worker_id, status: self.status, request: self.request, execution: self.execution, - transcript, - next_transcript_sequence: self.next_transcript_sequence, last_event_id: self.last_event_id, } } diff --git a/crates/worker-runtime/src/http_server.rs b/crates/worker-runtime/src/http_server.rs index d03b1a44..147a9f31 100644 --- a/crates/worker-runtime/src/http_server.rs +++ b/crates/worker-runtime/src/http_server.rs @@ -18,7 +18,6 @@ use crate::interaction::{WorkerInput, WorkerInteractionAck}; use crate::management::{RuntimeLimits, RuntimeSummary}; #[cfg(feature = "ws-server")] use crate::observation::WorkerObservationCursor; -use crate::observation::{TranscriptProjection, TranscriptQuery}; use axum::body::{Body, Bytes}; use axum::extract::rejection::{JsonRejection, QueryRejection}; #[cfg(feature = "ws-server")] @@ -151,11 +150,7 @@ pub fn runtime_http_router(runtime: Runtime, local_token: Option) -> Rou .route("/v1/workers/{worker_id}", get(get_worker)) .route("/v1/workers/{worker_id}/input", post(send_worker_input)) .route("/v1/workers/{worker_id}/stop", post(stop_worker)) - .route("/v1/workers/{worker_id}/cancel", post(cancel_worker)) - .route( - "/v1/workers/{worker_id}/transcript", - get(get_worker_transcript), - ); + .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)); @@ -243,12 +238,6 @@ pub struct RuntimeHttpWorkerLifecycleResponse { pub ack: WorkerLifecycleAck, } -/// `GET /v1/workers/{worker_id}/transcript` response. -#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] -pub struct RuntimeHttpTranscriptResponse { - pub transcript: TranscriptProjection, -} - /// Typed REST error response. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct RuntimeHttpErrorResponse { @@ -299,18 +288,6 @@ struct RuntimeWorkerEventsWsQuery { cursor: Option, } -#[derive(Clone, Debug, Deserialize)] -struct RuntimeHttpTranscriptQuery { - #[serde(default)] - start: usize, - #[serde(default = "default_transcript_limit")] - limit: usize, -} - -fn default_transcript_limit() -> usize { - 256 -} - #[cfg(feature = "ws-server")] impl RuntimeWorkerEventWsFrame { fn event( @@ -714,20 +691,6 @@ async fn cancel_worker( Ok(Json(RuntimeHttpWorkerLifecycleResponse { ack })) } -async fn get_worker_transcript( - State(state): State, - Path(worker_id): Path, - query: Result, QueryRejection>, -) -> RestResult { - let worker_ref = worker_ref_for(&state.runtime, worker_id)?; - let Query(query) = query.map_err(RuntimeHttpRestError::query_rejection)?; - let transcript = state - .runtime - .transcript_projection(&worker_ref, TranscriptQuery::new(query.start, query.limit)) - .map_err(RuntimeHttpRestError::runtime)?; - Ok(Json(RuntimeHttpTranscriptResponse { transcript })) -} - fn worker_ref_for(runtime: &Runtime, worker_id: String) -> Result { let worker_id = WorkerId::new(worker_id).ok_or_else(|| { RuntimeHttpRestError::new( @@ -1067,8 +1030,7 @@ mod tests { ) .await; assert_eq!(response.status(), StatusCode::OK); - let input_ack: RuntimeHttpWorkerInputResponse = read_json(response).await; - assert_eq!(input_ack.ack.transcript_sequence, 1); + let _input_ack: RuntimeHttpWorkerInputResponse = read_json(response).await; let response = empty_request( app.clone(), @@ -1077,21 +1039,15 @@ mod tests { ) .await; assert_eq!(response.status(), StatusCode::OK); - let detail: RuntimeHttpWorkerResponse = read_json(response).await; - assert_eq!(detail.worker.transcript_len, 1); + let _detail: RuntimeHttpWorkerResponse = read_json(response).await; let response = empty_request( app.clone(), Method::GET, - &format!( - "/v1/workers/{}/transcript?start=0&limit=1", - created.worker.worker_id - ), + &format!("/v1/workers/{}/transcript", created.worker.worker_id), ) .await; - assert_eq!(response.status(), StatusCode::OK); - let transcript: RuntimeHttpTranscriptResponse = read_json(response).await; - assert_eq!(transcript.transcript.items[0].content, "hello from backend"); + assert_eq!(response.status(), StatusCode::NOT_FOUND); let response = empty_request( app.clone(), @@ -1117,7 +1073,6 @@ mod tests { assert_eq!(response.status(), StatusCode::OK); let workers: RuntimeHttpWorkersResponse = read_json(response).await; assert_eq!(workers.workers.len(), 1); - assert_eq!(workers.workers[0].transcript_len, 1); let response = empty_request(app, Method::GET, "/v1/runtime").await; assert_eq!(response.status(), StatusCode::OK); diff --git a/crates/worker-runtime/src/interaction.rs b/crates/worker-runtime/src/interaction.rs index a9dd440d..60e12a0b 100644 --- a/crates/worker-runtime/src/interaction.rs +++ b/crates/worker-runtime/src/interaction.rs @@ -10,8 +10,7 @@ pub enum WorkerInputKind { System, } -/// Worker input request. v0 stores the input in an in-memory transcript and -/// does not execute providers/tools. +/// Worker input request accepted by a Runtime Worker. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct WorkerInput { pub kind: WorkerInputKind, @@ -34,11 +33,10 @@ impl WorkerInput { } } -/// Acknowledgement returned after input is accepted into the in-memory Worker. +/// Acknowledgement returned after input is accepted into the Worker. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct WorkerInteractionAck { pub worker_ref: WorkerRef, pub status: WorkerStatus, - pub transcript_sequence: u64, pub event_id: u64, } diff --git a/crates/worker-runtime/src/main.rs b/crates/worker-runtime/src/main.rs index 0cf435b8..e19cc79a 100644 --- a/crates/worker-runtime/src/main.rs +++ b/crates/worker-runtime/src/main.rs @@ -238,10 +238,6 @@ where } config.http.local_token = Some(value); } - "--max-transcript-projection-items" => { - config.http.limits.max_transcript_projection_items = - parse_usize_flag(&flag, take_value(&flag, inline_value, &mut args)?)?; - } "--max-event-batch-items" => { config.http.limits.max_event_batch_items = parse_usize_flag(&flag, take_value(&flag, inline_value, &mut args)?)?; @@ -436,7 +432,6 @@ Options:\n\ --fs-root Runtime catalog filesystem store root\n\ --local-token Minimal local bearer token placeholder\n\ --local-token-env Read local bearer token placeholder from env\n\ - --max-transcript-projection-items Override transcript projection limit\n\ --max-event-batch-items Override event batch limit\n\ -h, --help Show this help" } diff --git a/crates/worker-runtime/src/management.rs b/crates/worker-runtime/src/management.rs index 008355e5..f1a3549e 100644 --- a/crates/worker-runtime/src/management.rs +++ b/crates/worker-runtime/src/management.rs @@ -18,17 +18,15 @@ pub enum RuntimeStatus { Stopped, } -/// Guardrails for bounded observation/projection APIs. +/// Guardrails for bounded Runtime APIs. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct RuntimeLimits { - pub max_transcript_projection_items: usize, pub max_event_batch_items: usize, } impl Default for RuntimeLimits { fn default() -> Self { Self { - max_transcript_projection_items: 256, max_event_batch_items: 256, } } diff --git a/crates/worker-runtime/src/observation.rs b/crates/worker-runtime/src/observation.rs index d9c5bb00..1e39936e 100644 --- a/crates/worker-runtime/src/observation.rs +++ b/crates/worker-runtime/src/observation.rs @@ -1,49 +1,6 @@ use crate::identity::{RuntimeId, WorkerRef}; use serde::{Deserialize, Serialize}; -/// Transcript role used by bounded projection. -#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] -#[serde(rename_all = "snake_case")] -pub enum TranscriptRole { - User, - Assistant, - System, -} - -/// One projected transcript item. -#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] -pub struct TranscriptEntry { - pub sequence: u64, - pub worker_ref: WorkerRef, - pub role: TranscriptRole, - pub content: String, - pub event_id: u64, -} - -/// Bounded transcript query. -#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] -pub struct TranscriptQuery { - pub start: usize, - pub limit: usize, -} - -impl TranscriptQuery { - pub fn new(start: usize, limit: usize) -> Self { - Self { start, limit } - } -} - -/// Bounded transcript projection response. -#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] -pub struct TranscriptProjection { - pub worker_ref: WorkerRef, - pub start: usize, - pub limit: usize, - pub total_items: usize, - pub items: Vec, - pub next_start: Option, -} - /// Event cursor. `next_event_id` is the first event id that should be returned /// by the next poll. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index f3b069ff..4b6665ec 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -28,7 +28,7 @@ use crate::management::{ }; use crate::observation::{ EventCursor, EventSubscription, EventSubscriptionMode, RuntimeEvent, RuntimeEventBatch, - RuntimeEventKind, TranscriptEntry, TranscriptProjection, TranscriptQuery, TranscriptRole, + RuntimeEventKind, }; #[cfg(feature = "ws-server")] use crate::observation::{WorkerObservationCursor, WorkerObservationEvent}; @@ -333,19 +333,6 @@ impl Runtime { format!("worker {worker_id} created"), ); - let mut transcript = Vec::new(); - let mut next_transcript_sequence = 1; - if let Some(input) = request.initial_input.clone() { - transcript.push(TranscriptEntry { - sequence: next_transcript_sequence, - worker_ref: worker_ref.clone(), - role: TranscriptRole::User, - content: input.content, - event_id, - }); - next_transcript_sequence += 1; - } - let record = WorkerRecord { worker_ref: worker_ref.clone(), worker_id: worker_id.clone(), @@ -353,8 +340,6 @@ impl Runtime { request: request.clone(), execution: WorkerExecutionStatus::unconnected(), execution_handle: None, - transcript, - next_transcript_sequence, last_event_id: event_id, }; state.workers.insert(worker_id, record); @@ -392,7 +377,7 @@ impl Runtime { let state = self.lock()?; state.worker(&worker_ref)?.request.initial_input.clone() } { - let dispatch_result = backend.dispatch_input(&handle, initial_input); + let dispatch_result = backend.dispatch_input(&handle, initial_input.clone()); if !dispatch_result.is_accepted() { let _ = backend.stop_worker(&handle); self.rollback_failed_create(&worker_ref)?; @@ -404,7 +389,7 @@ impl Runtime { result: dispatch_result, }); } - self.commit_created_worker( + let detail = self.commit_created_worker( &worker_ref, handle, WorkerExecutionRunState::Busy, @@ -413,7 +398,9 @@ impl Runtime { WorkerExecutionOperation::Input, WorkerExecutionRunState::Busy, ), - ) + )?; + self.record_input_observation(&worker_ref, initial_input)?; + Ok(detail) } else { self.commit_created_worker( &worker_ref, @@ -442,7 +429,7 @@ impl Runtime { Ok(worker.detail(&state.runtime_id)) } - /// Accept input into a Worker transcript. + /// Accept input into a Worker. pub fn send_input( &self, worker_ref: &WorkerRef, @@ -502,14 +489,6 @@ impl Runtime { "worker input accepted", ); let worker = state.worker_mut(worker_ref)?; - - let input_content = input.content; - let role = match input.kind { - WorkerInputKind::User => TranscriptRole::User, - WorkerInputKind::System => TranscriptRole::System, - }; - let transcript_sequence = worker.next_transcript_sequence; - worker.next_transcript_sequence += 1; worker.last_event_id = event_id; worker.execution = WorkerExecutionStatus { backend: WorkerExecutionBackendKind::Connected, @@ -518,44 +497,24 @@ impl Runtime { working_directory: worker.execution.working_directory.clone(), last_result: Some(dispatch_result), }; - worker.transcript.push(TranscriptEntry { - sequence: transcript_sequence, - worker_ref: worker_ref.clone(), - role, - content: input_content.clone(), - event_id, - }); let status = worker.status; #[cfg(feature = "ws-server")] - { - let payload = match role { - TranscriptRole::User => protocol::Event::UserMessage { - segments: vec![protocol::Segment::Text { - content: input_content.clone(), - }], - }, - TranscriptRole::Assistant => protocol::Event::TextDone { - text: input_content.clone(), - }, - TranscriptRole::System => protocol::Event::SystemItem { - item: serde_json::json!({ - "kind": "embedded_worker_system_input", - "content": input_content.clone(), - }), - }, - }; - state.push_worker_observation_event(worker_ref.clone(), payload); - } + let observation = { + let payload = input_protocol_event(&input); + Some(state.push_worker_observation_event(worker_ref.clone(), payload)) + }; state.persist_runtime_snapshot()?; state.persist_worker(&worker_ref.worker_id)?; state.persist_event_by_id(event_id)?; - state.persist_transcript_entry(&worker_ref.worker_id, transcript_sequence)?; + #[cfg(feature = "ws-server")] + if let Some(observation) = observation.as_ref() { + state.persist_worker_observation_event(observation)?; + } Ok(WorkerInteractionAck { worker_ref: worker_ref.clone(), status, - transcript_sequence, event_id, }) } @@ -583,10 +542,6 @@ impl Runtime { }; state.persist_runtime_snapshot()?; state.persist_worker(&worker_ref.worker_id)?; - let worker = state.worker(worker_ref)?; - for entry in &worker.transcript { - state.persist_transcript_entry(&worker_ref.worker_id, entry.sequence)?; - } state.persist_event_by_id(detail.last_event_id)?; Ok(detail) } @@ -691,38 +646,6 @@ impl Runtime { ) } - /// Bounded transcript projection for a Worker. - pub fn transcript_projection( - &self, - worker_ref: &WorkerRef, - query: TranscriptQuery, - ) -> Result { - let state = self.lock()?; - if query.limit > state.limits.max_transcript_projection_items { - return Err(RuntimeError::LimitTooLarge { - requested: query.limit, - max: state.limits.max_transcript_projection_items, - }); - } - let worker = state.worker(worker_ref)?; - let total_items = worker.transcript.len(); - let end = query.start.saturating_add(query.limit).min(total_items); - let items = if query.start >= total_items { - Vec::new() - } else { - worker.transcript[query.start..end].to_vec() - }; - let next_start = (end < total_items).then_some(end); - Ok(TranscriptProjection { - worker_ref: worker_ref.clone(), - start: query.start, - limit: query.limit, - total_items, - items, - next_start, - }) - } - /// Cursor pointing to the beginning of Runtime events. pub fn event_cursor_from_start(&self) -> Result { let state = self.lock()?; @@ -880,16 +803,13 @@ impl Runtime { ) -> Result { let mut state = self.lock()?; state.ensure_worker_ref(worker_ref)?; - let transcript_sequence = state.project_protocol_event_to_transcript(worker_ref, &payload); let execution_state_changed = state.project_protocol_event_to_execution(worker_ref, &payload); let event = state.push_worker_observation_event(worker_ref.clone(), payload); - if transcript_sequence.is_some() || execution_state_changed { + if execution_state_changed { state.persist_worker(&worker_ref.worker_id)?; } - if let Some(sequence) = transcript_sequence { - state.persist_transcript_entry(&worker_ref.worker_id, sequence)?; - } + state.persist_worker_observation_event(&event)?; Ok(event) } @@ -898,6 +818,29 @@ impl Runtime { Ok(self.lock()?.diagnostics.clone()) } + #[cfg(feature = "ws-server")] + fn record_input_observation( + &self, + worker_ref: &WorkerRef, + input: WorkerInput, + ) -> Result<(), RuntimeError> { + let mut state = self.lock()?; + state.ensure_worker_ref(worker_ref)?; + let event = + state.push_worker_observation_event(worker_ref.clone(), input_protocol_event(&input)); + state.persist_worker_observation_event(&event)?; + Ok(()) + } + + #[cfg(not(feature = "ws-server"))] + fn record_input_observation( + &self, + _worker_ref: &WorkerRef, + _input: WorkerInput, + ) -> Result<(), RuntimeError> { + Ok(()) + } + fn transition_worker( &self, worker_ref: &WorkerRef, @@ -1106,13 +1049,22 @@ impl RuntimeState { request: worker.request, execution, execution_handle: None, - transcript: worker.transcript, - next_transcript_sequence: worker.next_transcript_sequence, last_event_id: worker.last_event_id, }, ); } + #[cfg(feature = "ws-server")] + let next_observation_sequence = persisted + .observation_events + .iter() + .map(|event| event.sequence) + .max() + .map(|sequence| sequence.saturating_add(1)) + .unwrap_or(1); + #[cfg(feature = "ws-server")] + let observation_events = persisted.observation_events.into_iter().collect(); + Ok(Self { runtime_id: persisted.runtime_id, display_name: persisted.display_name, @@ -1129,9 +1081,9 @@ impl RuntimeState { events: persisted.events, diagnostics, #[cfg(feature = "ws-server")] - next_observation_sequence: 1, + next_observation_sequence, #[cfg(feature = "ws-server")] - observation_events: VecDeque::new(), + observation_events, #[cfg(feature = "ws-server")] observation_tx: broadcast::channel(256).0, }) @@ -1154,6 +1106,8 @@ impl RuntimeState { .collect(), config_bundles: self.config_bundles.clone(), events: self.events.clone(), + #[cfg(feature = "ws-server")] + observation_events: Vec::new(), diagnostics: self.diagnostics.clone(), } } @@ -1206,36 +1160,25 @@ impl RuntimeState { Ok(()) } - #[cfg(feature = "fs-store")] - fn persist_transcript_entry( + #[cfg(all(feature = "fs-store", feature = "ws-server"))] + fn persist_worker_observation_event( &self, - worker_id: &WorkerId, - sequence: u64, + event: &WorkerObservationEvent, ) -> Result<(), RuntimeError> { if let Some(store) = self.fs_store() { - let worker = - self.workers - .get(worker_id) - .ok_or_else(|| RuntimeError::WorkerNotFound { - runtime_id: self.runtime_id.clone(), - worker_id: worker_id.clone(), - })?; - let entry = worker - .transcript - .iter() - .find(|entry| entry.sequence == sequence) - .ok_or_else(|| RuntimeError::StoreCorrupt { - operation: "persist transcript", - path: store.runtime_dir().to_path_buf(), - message: format!( - "transcript sequence {sequence} is missing from worker {worker_id}" - ), - })?; - store.append_transcript_entry(entry)?; + store.append_worker_observation_event(event)?; } Ok(()) } + #[cfg(all(not(feature = "fs-store"), feature = "ws-server"))] + fn persist_worker_observation_event( + &self, + _event: &WorkerObservationEvent, + ) -> Result<(), RuntimeError> { + Ok(()) + } + #[cfg(feature = "fs-store")] fn persist_workers(&self) -> Result<(), RuntimeError> { if self.fs_store().is_some() { @@ -1261,15 +1204,6 @@ impl RuntimeState { Ok(()) } - #[cfg(not(feature = "fs-store"))] - fn persist_transcript_entry( - &self, - _worker_id: &WorkerId, - _sequence: u64, - ) -> Result<(), RuntimeError> { - Ok(()) - } - #[cfg(not(feature = "fs-store"))] fn persist_workers(&self) -> Result<(), RuntimeError> { Ok(()) @@ -1421,66 +1355,6 @@ impl RuntimeState { event } - #[cfg(feature = "ws-server")] - fn append_worker_transcript_entry( - &mut self, - worker_ref: &WorkerRef, - role: TranscriptRole, - content: impl Into, - ) -> Option { - let content = content.into(); - if content.trim().is_empty() { - return None; - } - let event_id = self.last_event_id(); - let worker = self.workers.get_mut(&worker_ref.worker_id)?; - let sequence = worker.next_transcript_sequence; - worker.next_transcript_sequence += 1; - worker.transcript.push(TranscriptEntry { - sequence, - worker_ref: worker_ref.clone(), - role, - content, - event_id, - }); - Some(sequence) - } - - #[cfg(feature = "ws-server")] - fn project_protocol_event_to_transcript( - &mut self, - worker_ref: &WorkerRef, - event: &protocol::Event, - ) -> Option { - match event { - protocol::Event::TextDone { text, .. } => self.append_worker_transcript_entry( - worker_ref, - TranscriptRole::Assistant, - text.clone(), - ), - protocol::Event::Error { message, .. } => self.append_worker_transcript_entry( - worker_ref, - TranscriptRole::System, - format!("error: {message}"), - ), - protocol::Event::ToolResult { - id, - summary, - is_error, - .. - } => self.append_worker_transcript_entry( - worker_ref, - TranscriptRole::System, - format!( - "tool result {id}: {}{}", - if *is_error { "error: " } else { "" }, - summary - ), - ), - _ => None, - } - } - #[cfg(feature = "ws-server")] fn project_protocol_event_to_execution( &mut self, @@ -1535,8 +1409,6 @@ struct WorkerRecord { request: CreateWorkerRequest, execution: WorkerExecutionStatus, execution_handle: Option, - transcript: Vec, - next_transcript_sequence: u64, last_event_id: u64, } @@ -1551,7 +1423,6 @@ impl WorkerRecord { profile: self.request.profile.clone(), profile_source: self.request.profile_source.reference(), config_bundle: self.request.config_bundle.clone(), - transcript_len: self.transcript.len(), last_event_id: self.last_event_id, } } @@ -1566,7 +1437,6 @@ impl WorkerRecord { profile: self.request.profile.clone(), profile_source: self.request.profile_source.reference(), config_bundle: self.request.config_bundle.clone(), - transcript_len: self.transcript.len(), last_event_id: self.last_event_id, } } @@ -1579,8 +1449,6 @@ impl WorkerRecord { status: self.status, request: self.request.clone(), execution: self.execution.clone(), - transcript: self.transcript.clone(), - next_transcript_sequence: self.next_transcript_sequence, last_event_id: self.last_event_id, } } @@ -1630,6 +1498,23 @@ fn validate_worker_input(input: &WorkerInput) -> Result<(), RuntimeError> { Ok(()) } +#[cfg(feature = "ws-server")] +fn input_protocol_event(input: &WorkerInput) -> protocol::Event { + match input.kind { + WorkerInputKind::User => protocol::Event::UserMessage { + segments: vec![protocol::Segment::Text { + content: input.content.clone(), + }], + }, + WorkerInputKind::System => protocol::Event::SystemItem { + item: serde_json::json!({ + "kind": "embedded_worker_system_input", + "content": input.content.clone(), + }), + }, + } +} + #[cfg(test)] mod tests { use super::*; @@ -1965,12 +1850,12 @@ mod tests { )); let refreshed = runtime.worker_detail(&detail.worker_ref).unwrap(); assert_eq!(refreshed.execution.run_state, WorkerExecutionRunState::Busy); - assert_eq!( + #[cfg(feature = "ws-server")] + assert!( runtime - .transcript_projection(&detail.worker_ref, TranscriptQuery::new(0, 1)) + .read_worker_observation_events(&detail.worker_ref, WorkerObservationCursor::zero()) .unwrap() - .total_items, - 0 + .is_empty() ); } @@ -2049,12 +1934,12 @@ mod tests { ); } + #[cfg(feature = "ws-server")] #[test] - fn send_input_and_project_bounded_transcript() { + fn send_input_records_protocol_observations() { let runtime = Runtime::with_execution_backend( RuntimeOptions { limits: RuntimeLimits { - max_transcript_projection_items: 2, max_event_batch_items: 16, }, ..RuntimeOptions::default() @@ -2068,27 +1953,23 @@ mod tests { let first = runtime .send_input(&detail.worker_ref, WorkerInput::user("hello")) .unwrap(); - assert_eq!(first.transcript_sequence, 1); runtime .send_input(&detail.worker_ref, WorkerInput::system("note")) .unwrap(); - runtime - .send_input(&detail.worker_ref, WorkerInput::user("again")) - .unwrap(); - let projection = runtime - .transcript_projection(&detail.worker_ref, TranscriptQuery::new(0, 2)) + let observations = runtime + .read_worker_observation_events(&detail.worker_ref, WorkerObservationCursor::zero()) .unwrap(); - assert_eq!(projection.total_items, 3); - assert_eq!(projection.items.len(), 2); - assert_eq!(projection.items[0].content, "hello"); - assert_eq!(projection.items[1].role, TranscriptRole::System); - assert_eq!(projection.next_start, Some(2)); - - let err = runtime - .transcript_projection(&detail.worker_ref, TranscriptQuery::new(0, 3)) - .unwrap_err(); - assert!(matches!(err, RuntimeError::LimitTooLarge { .. })); + assert_eq!(first.event_id, 3); + assert_eq!(observations.len(), 2); + assert!(matches!( + observations[0].payload, + protocol::Event::UserMessage { .. } + )); + assert!(matches!( + observations[1].payload, + protocol::Event::SystemItem { .. } + )); } #[test] @@ -2253,7 +2134,7 @@ mod tests { #[cfg(feature = "fs-store")] #[test] - fn fs_store_restores_workers_events_and_transcripts() { + fn fs_store_restores_workers_events_and_protocol_observations() { let root = fs_store_root("restore"); let runtime_id = RuntimeId::new("runtime-fs-authority").unwrap(); let runtime = Runtime::with_fs_store_and_execution_backend( @@ -2262,7 +2143,6 @@ mod tests { runtime_id: Some(runtime_id.clone()), display_name: Some("filesystem runtime".to_string()), limits: RuntimeLimits { - max_transcript_projection_items: 2, max_event_batch_items: 2, }, }, @@ -2319,14 +2199,21 @@ mod tests { && diagnostic.worker_ref.as_ref() == Some(&worker.worker_ref) ) ); - assert_eq!(restored_worker.transcript_len, 2); - - let projection = restored - .transcript_projection(&worker.worker_ref, TranscriptQuery::new(0, 1)) - .unwrap(); - assert_eq!(projection.total_items, 2); - assert_eq!(projection.items[0].content, "first"); - assert_eq!(projection.next_start, Some(1)); + #[cfg(feature = "ws-server")] + { + let observations = restored + .read_worker_observation_events(&worker.worker_ref, WorkerObservationCursor::zero()) + .unwrap(); + assert_eq!(observations.len(), 2); + assert!(matches!( + observations[0].payload, + protocol::Event::UserMessage { .. } + )); + assert!(matches!( + observations[1].payload, + protocol::Event::SystemItem { .. } + )); + } let cursor = restored.event_cursor_from_start().unwrap(); let batch = restored.read_events(&cursor, 2).unwrap(); @@ -2337,11 +2224,6 @@ mod tests { let direct_events = store.read_events(&cursor, 2, 2).unwrap(); assert_eq!(direct_events.events, batch.events); - let direct_transcript = store - .read_transcript(&worker.worker_ref, TranscriptQuery::new(1, 1), 2) - .unwrap(); - assert_eq!(direct_transcript.items[0].content, "second"); - #[cfg(feature = "ws-server")] { let observation = restored @@ -2352,12 +2234,12 @@ mod tests { }, ) .unwrap(); - assert_eq!(observation.sequence, 1); + assert_eq!(observation.sequence, 3); let observations = restored .read_worker_observation_events(&worker.worker_ref, WorkerObservationCursor::zero()) .unwrap(); - assert_eq!(observations.len(), 1); - assert_eq!(observations[0].cursor, observation.cursor); + assert_eq!(observations.len(), 3); + assert_eq!(observations[2].cursor, observation.cursor); } let _ = std::fs::remove_dir_all(root); diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index e11a07b3..335600c8 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -820,7 +820,7 @@ mod tests { use crate::execution::WorkerExecutionContext; use crate::identity::RuntimeId; use crate::management::RuntimeOptions; - use crate::observation::{TranscriptQuery, TranscriptRole}; + use crate::observation::WorkerObservationCursor; use crate::working_directory::LocalGitWorktreeMaterializer; use async_trait::async_trait; use futures::Stream; @@ -1128,27 +1128,27 @@ mod tests { let deadline = std::time::Instant::now() + Duration::from_secs(5); loop { - let projection = runtime - .transcript_projection(&detail.worker_ref, TranscriptQuery::new(0, 10)) + let observations = runtime + .read_worker_observation_events(&detail.worker_ref, WorkerObservationCursor::zero()) .unwrap(); - if projection.items.iter().any(|item| { - item.role == TranscriptRole::Assistant && item.content == "hello from worker" + if observations.iter().any(|event| { + matches!( + &event.payload, + protocol::Event::TextDone { text } if text == "hello from worker" + ) }) { break; } assert!( std::time::Instant::now() < deadline, - "timed out waiting for assistant transcript projection" + "timed out waiting for assistant protocol observation" ); std::thread::sleep(Duration::from_millis(20)); } assert_eq!(client.captured.lock().unwrap().len(), 1); let observations = runtime - .read_worker_observation_events( - &detail.worker_ref, - crate::observation::WorkerObservationCursor::zero(), - ) + .read_worker_observation_events(&detail.worker_ref, WorkerObservationCursor::zero()) .unwrap(); assert!( observations diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index 987a3a3f..158a419f 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -19,9 +19,10 @@ use worker_runtime::catalog::{ WorkerStatus as EmbeddedWorkerStatus, WorkingDirectoryClaim, WorkingDirectoryRequest, WorkingDirectoryStatus, WorkingDirectorySummary, }; +use worker_runtime::config_bundle::{ConfigBundle, ConfigBundleAvailability, ConfigBundleSummary}; +#[cfg(test)] use worker_runtime::config_bundle::{ - ConfigBundle, ConfigBundleAvailability, ConfigBundleMetadata, ConfigBundleProvenance, - ConfigBundleSummary, ConfigProfileDescriptor, + ConfigBundleMetadata, ConfigBundleProvenance, ConfigProfileDescriptor, }; use worker_runtime::error::RuntimeError as EmbeddedRuntimeError; use worker_runtime::execution::{ @@ -30,10 +31,10 @@ use worker_runtime::execution::{ use worker_runtime::fs_store::FsRuntimeStoreOptions; use worker_runtime::http_server::{ RuntimeHttpConfigBundleAvailabilityResponse, RuntimeHttpConfigBundleSyncRequest, - RuntimeHttpErrorResponse, RuntimeHttpSummaryResponse, RuntimeHttpTranscriptResponse, - RuntimeHttpWorkerInputResponse, RuntimeHttpWorkerLifecycleRequest, - RuntimeHttpWorkerLifecycleResponse, RuntimeHttpWorkerResponse, RuntimeHttpWorkersResponse, - RuntimeHttpWorkingDirectoriesResponse, RuntimeHttpWorkingDirectoryResponse, + RuntimeHttpErrorResponse, RuntimeHttpSummaryResponse, RuntimeHttpWorkerInputResponse, + RuntimeHttpWorkerLifecycleRequest, RuntimeHttpWorkerLifecycleResponse, + RuntimeHttpWorkerResponse, RuntimeHttpWorkersResponse, RuntimeHttpWorkingDirectoriesResponse, + RuntimeHttpWorkingDirectoryResponse, }; use worker_runtime::identity::{ RuntimeId as EmbeddedRuntimeId, WorkerId as EmbeddedWorkerId, WorkerRef as EmbeddedWorkerRef, @@ -42,9 +43,6 @@ use worker_runtime::interaction::{ WorkerInput as EmbeddedWorkerInput, WorkerInputKind as EmbeddedWorkerInputKind, }; use worker_runtime::management::{RuntimeOptions as EmbeddedRuntimeOptions, RuntimeStatus}; -use worker_runtime::observation::{ - TranscriptProjection as EmbeddedTranscriptProjection, TranscriptQuery, TranscriptRole, -}; use worker_runtime::profile_archive::{ProfileSourceArchive, ProfileSourceArchiveInput}; const EMBEDDED_RUNTIME_ID: &str = "embedded-worker-runtime"; @@ -454,34 +452,10 @@ pub struct WorkerInputResult { pub runtime_id: String, pub worker_id: String, #[serde(skip_serializing_if = "Option::is_none")] - pub transcript_sequence: Option, - #[serde(skip_serializing_if = "Option::is_none")] pub event_id: Option, pub diagnostics: Vec, } -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] -pub struct WorkerTranscriptItem { - pub sequence: u64, - pub role: String, - pub content: String, - pub event_id: u64, -} - -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] -pub struct WorkerTranscriptProjection { - pub state: WorkerOperationState, - pub runtime_id: String, - pub worker_id: String, - pub start: usize, - pub limit: usize, - pub total_items: usize, - #[serde(skip_serializing_if = "Option::is_none")] - pub next_start: Option, - pub items: Vec, - pub diagnostics: Vec, -} - #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub struct WorkerProxyConnectPoint { pub kind: String, @@ -705,7 +679,6 @@ pub trait WorkspaceWorkerRuntime: Send + Sync { state: WorkerOperationState::Unsupported, runtime_id: self.runtime_id().to_string(), worker_id: worker_id.to_string(), - transcript_sequence: None, event_id: None, diagnostics: vec![diagnostic( "worker_input_pending", @@ -717,31 +690,6 @@ pub trait WorkspaceWorkerRuntime: Send + Sync { } } - fn transcript( - &self, - worker_id: &str, - start: usize, - limit: usize, - ) -> WorkerTranscriptProjection { - WorkerTranscriptProjection { - state: WorkerOperationState::Unsupported, - runtime_id: self.runtime_id().to_string(), - worker_id: worker_id.to_string(), - start, - limit, - total_items: 0, - next_start: None, - items: Vec::new(), - diagnostics: vec![diagnostic( - "worker_transcript_pending", - DiagnosticSeverity::Info, - format!( - "bounded transcript for '{worker_id}' is not implemented by this registry source" - ), - )], - } - } - fn proxy_connect_points(&self, worker_id: &str) -> Vec { vec![WorkerProxyConnectPoint { kind: "stream_proxy".to_string(), @@ -1050,27 +998,6 @@ impl RuntimeRegistry { Ok(runtime.send_input(worker_id, request)) } - pub fn transcript( - &self, - runtime_id: &str, - worker_id: &str, - start: usize, - limit: usize, - ) -> Result { - 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, - )); - } - Ok(runtime.transcript(worker_id, start, limit)) - } - pub fn stop_worker( &self, runtime_id: &str, @@ -1781,7 +1708,6 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime { state: WorkerOperationState::Accepted, runtime_id: self.runtime_id.clone(), worker_id: worker_id.to_string(), - transcript_sequence: Some(ack.transcript_sequence), event_id: Some(ack.event_id), diagnostics: Vec::new(), }, @@ -1792,42 +1718,6 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime { ), } } - - fn transcript( - &self, - worker_id: &str, - start: usize, - limit: usize, - ) -> WorkerTranscriptProjection { - let Some(worker_ref) = self.worker_ref(worker_id) else { - return embedded_transcript_rejected( - &self.runtime_id, - worker_id, - start, - limit, - diagnostic( - "embedded_worker_id_invalid", - DiagnosticSeverity::Warning, - "Worker id was empty and cannot be resolved".to_string(), - ), - ); - }; - match self - .runtime - .transcript_projection(&worker_ref, TranscriptQuery::new(start, limit)) - { - Ok(projection) => { - embedded_transcript_projection(&self.runtime_id, worker_id, projection) - } - Err(err) => embedded_transcript_rejected( - &self.runtime_id, - worker_id, - start, - limit, - embedded_runtime_diagnostic(&err), - ), - } - } } #[derive(Clone)] @@ -2444,31 +2334,12 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { state: WorkerOperationState::Accepted, runtime_id: self.runtime_id.clone(), worker_id: worker_id.to_string(), - transcript_sequence: Some(response.ack.transcript_sequence), event_id: Some(response.ack.event_id), diagnostics: Vec::new(), }, Err(diagnostic) => remote_input_rejected(&self.runtime_id, worker_id, diagnostic), } } - - fn transcript( - &self, - worker_id: &str, - start: usize, - limit: usize, - ) -> WorkerTranscriptProjection { - match self.get_json::(&format!( - "/v1/workers/{worker_id}/transcript?start={start}&limit={limit}" - )) { - Ok(response) => { - embedded_transcript_projection(&self.runtime_id, worker_id, response.transcript) - } - Err(diagnostic) => { - embedded_transcript_rejected(&self.runtime_id, worker_id, start, limit, diagnostic) - } - } - } } fn embedded_runtime_capabilities( @@ -2676,12 +2547,14 @@ fn default_profile_source_archive_http_source( }) } +#[cfg(test)] #[derive(Clone, Copy, Debug, Eq, PartialEq)] enum ProfileSourceArchiveTransport { Inline, BackendResourceHandle, } +#[cfg(test)] fn default_embedded_config_bundle( profile: &ProfileSelector, workspace_id: &str, @@ -2844,7 +2717,6 @@ fn embedded_input_rejected( state: WorkerOperationState::Rejected, runtime_id: runtime_id.to_string(), worker_id: worker_id.to_string(), - transcript_sequence: None, event_id: None, diagnostics: vec![diagnostic], } @@ -2859,7 +2731,6 @@ fn remote_input_rejected( state: WorkerOperationState::Rejected, runtime_id: runtime_id.to_string(), worker_id: worker_id.to_string(), - transcript_sequence: None, event_id: None, diagnostics: vec![diagnostic], } @@ -2893,61 +2764,6 @@ fn remote_lifecycle_rejected( } } -fn embedded_transcript_projection( - runtime_id: &str, - worker_id: &str, - projection: EmbeddedTranscriptProjection, -) -> WorkerTranscriptProjection { - WorkerTranscriptProjection { - state: WorkerOperationState::Accepted, - runtime_id: runtime_id.to_string(), - worker_id: worker_id.to_string(), - start: projection.start, - limit: projection.limit, - total_items: projection.total_items, - next_start: projection.next_start, - items: projection - .items - .into_iter() - .map(|item| WorkerTranscriptItem { - sequence: item.sequence, - role: embedded_transcript_role_label(item.role).to_string(), - content: item.content, - event_id: item.event_id, - }) - .collect(), - diagnostics: Vec::new(), - } -} - -fn embedded_transcript_rejected( - runtime_id: &str, - worker_id: &str, - start: usize, - limit: usize, - diagnostic: RuntimeDiagnostic, -) -> WorkerTranscriptProjection { - WorkerTranscriptProjection { - state: WorkerOperationState::Rejected, - runtime_id: runtime_id.to_string(), - worker_id: worker_id.to_string(), - start, - limit, - total_items: 0, - next_start: None, - items: Vec::new(), - diagnostics: vec![diagnostic], - } -} - -fn embedded_transcript_role_label(role: TranscriptRole) -> &'static str { - match role { - TranscriptRole::User => "user", - TranscriptRole::Assistant => "assistant", - TranscriptRole::System => "system", - } -} - fn embedded_runtime_diagnostic(error: &EmbeddedRuntimeError) -> RuntimeDiagnostic { match error { EmbeddedRuntimeError::RuntimeStopped { .. } => diagnostic( @@ -3814,7 +3630,7 @@ mod tests { } #[test] - fn embedded_runtime_with_execution_backend_routes_input_and_projects_transcript() { + fn embedded_runtime_with_execution_backend_routes_input_and_updates_status() { let runtime = EmbeddedWorkerRuntime::new_memory_with_execution_backend( "local:test", Arc::new(AcceptingExecutionBackend::default()), @@ -3841,13 +3657,7 @@ mod tests { .worker(&worker.worker_id) .worker .expect("worker detail"); - let transcript = runtime.transcript(&worker.worker_id, 0, 10); - if detail.status == "idle" - && transcript - .items - .iter() - .any(|entry| entry.role == "assistant" && entry.content == "echo: hello") - { + if detail.status == "idle" { assert!(detail.capabilities.can_accept_input); break; } @@ -3860,7 +3670,7 @@ mod tests { } #[test] - fn embedded_runtime_registers_routes_input_and_transcript_without_internal_leaks() { + fn embedded_runtime_registers_routes_input_without_internal_leaks() { let registry = RuntimeRegistry::for_workspace( EmbeddedWorkerRuntime::new_memory_with_execution_backend( "local:test", @@ -3933,18 +3743,11 @@ mod tests { assert_eq!(input.runtime_id, EMBEDDED_RUNTIME_ID); assert_eq!(input.worker_id, worker.worker_id); - let transcript = registry - .transcript(EMBEDDED_RUNTIME_ID, &worker.worker_id, 0, 10) + let detail = registry + .worker(EMBEDDED_RUNTIME_ID, &worker.worker_id) .unwrap(); - assert_eq!(transcript.state, WorkerOperationState::Accepted); - assert!( - transcript - .items - .iter() - .any(|entry| entry.role == "user" && entry.content == "hello embedded runtime") - ); - let json = serde_json::to_string(&(embedded_summary, worker, transcript)).unwrap(); + let json = serde_json::to_string(&(embedded_summary, worker, input, detail)).unwrap(); for forbidden in [ "/workspace/project", "metadata.json", @@ -3953,6 +3756,7 @@ mod tests { "token", "credential", "provider", + "transcript", "can_stream_events", "can_read_bounded_transcript", ] { @@ -4093,7 +3897,6 @@ mod tests { "ack": { "worker_ref": { "runtime_id": "remote:primary", "worker_id": "worker-remote-1" }, "status": "running", - "transcript_sequence": 7, "event_id": 8 } }) @@ -4157,7 +3960,6 @@ mod tests { ) .unwrap(); assert_eq!(input.state, WorkerOperationState::Accepted); - assert_eq!(input.transcript_sequence, Some(7)); assert_eq!(input.event_id, Some(8)); server.join().expect("mock remote server finished"); @@ -4482,7 +4284,6 @@ mod tests { "source_graph": { "source_count": 0, "total_source_bytes": 0, "entrypoints": {}, "import_count": 0 } }, "config_bundle": { "id": "remote-bundle", "digest": "remote-digest" }, - "transcript_len": 0, "last_event_id": 0 }) } diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 47c87934..e6041b2e 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -27,7 +27,7 @@ use crate::hosts::{ RuntimeRegistryUnregisterResult, RuntimeSummary, WorkerInputRequest, WorkerInputResult, WorkerLifecycleRequest, WorkerLifecycleResult, WorkerOperationState, WorkerSpawnAcceptanceRequirement, WorkerSpawnIntent, WorkerSpawnRequest, WorkerSpawnResult, - WorkerSpawnWorkingDirectoryRequest, WorkerSummary, WorkerTranscriptProjection, + WorkerSpawnWorkingDirectoryRequest, WorkerSummary, }; use crate::identity::WorkspaceIdentity; use crate::observation::{ @@ -492,14 +492,6 @@ pub fn build_router(api: WorkspaceApi) -> Router { "/api/w/{workspace_id}/runtimes/{runtime_id}/workers/{worker_id}/cancel", post(scoped_cancel_runtime_worker), ) - .route( - "/api/runtimes/{runtime_id}/workers/{worker_id}/transcript", - get(get_runtime_worker_transcript), - ) - .route( - "/api/w/{workspace_id}/runtimes/{runtime_id}/workers/{worker_id}/transcript", - get(scoped_get_runtime_worker_transcript), - ) .route( "/api/runtimes/{runtime_id}/workers/{worker_id}/events/ws", get(worker_observation_ws), @@ -1498,20 +1490,6 @@ async fn scoped_cancel_runtime_worker( .await } -async fn scoped_get_runtime_worker_transcript( - State(api): State, - AxumPath(path): AxumPath, - Query(query): Query, -) -> ApiResult> { - validate_workspace_scope(&api, &path.workspace_id)?; - get_runtime_worker_transcript( - State(api), - AxumPath((path.runtime_id, path.worker_id)), - Query(query), - ) - .await -} - async fn scoped_worker_observation_ws( ws: WebSocketUpgrade, State(api): State, @@ -2245,20 +2223,6 @@ async fn cancel_runtime_worker( Ok(Json(result)) } -async fn get_runtime_worker_transcript( - State(api): State, - AxumPath((runtime_id, worker_id)): AxumPath<(String, String)>, - Query(query): Query, -) -> ApiResult> { - let limit = query.limit.unwrap_or(api.config.max_records).min(200); - let start = query.start.unwrap_or(0); - let result = api - .runtime - .transcript(&runtime_id, &worker_id, start, limit) - .map_err(|err| err.into_error())?; - Ok(Json(result)) -} - async fn worker_observation_ws( State(api): State, AxumPath((runtime_id, worker_id)): AxumPath<(String, String)>, @@ -5243,8 +5207,7 @@ mod tests { } #[tokio::test] - async fn embedded_runtime_fs_store_restores_catalog_config_bundle_transcript_and_stale_execution() - { + async fn embedded_runtime_fs_store_restores_catalog_config_bundle_and_stale_execution() { let dir = tempfile::tempdir().unwrap(); let config = test_server_config(dir.path().join("workspace")); let store_root = config.embedded_runtime_store_root.clone(); @@ -5304,24 +5267,16 @@ mod tests { let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2); loop { - let transcript = api + let detail = api .runtime - .transcript("embedded-worker-runtime", &worker_id, 0, 10) - .expect("transcript"); - if transcript.items.iter().any(|item| { - item.role == "assistant" && item.content == "server companion echoed: persist me" - }) { - assert!( - transcript - .items - .iter() - .any(|item| item.role == "user" && item.content == "persist me") - ); + .worker("embedded-worker-runtime", &worker_id) + .expect("worker detail"); + if detail.status == "idle" { break; } assert!( std::time::Instant::now() < deadline, - "timed out waiting for deterministic transcript" + "timed out waiting for deterministic worker completion" ); tokio::time::sleep(std::time::Duration::from_millis(10)).await; } @@ -5359,20 +5314,6 @@ mod tests { .any(|summary| summary.id == bundle_id) ); - let restored_transcript = restored - .runtime - .transcript("embedded-worker-runtime", &worker_id, 0, 10) - .expect("restored transcript"); - assert!( - restored_transcript - .items - .iter() - .any(|item| item.role == "user" && item.content == "persist me") - ); - assert!(restored_transcript.items.iter().any(|item| { - item.role == "assistant" && item.content == "server companion echoed: persist me" - })); - let rejected_input = restored .runtime .send_input( @@ -5591,15 +5532,20 @@ mod tests { assert_eq!(accepted["worker_id"], worker_id); assert!(accepted["diagnostics"].as_array().unwrap().is_empty()); - let transcript = get_json( - app.clone(), - &format!("/api/runtimes/embedded-worker-runtime/workers/{worker_id}/transcript?start=0&limit=10"), - ) - .await; - assert_eq!(transcript["state"], "accepted"); - assert!(transcript["items"].as_array().unwrap().iter().any( - |item| item["role"] == "user" && item["content"] == "hello from browser-facing api" - )); + let transcript_route = app + .clone() + .oneshot( + Request::builder() + .method("GET") + .uri(format!( + "/api/runtimes/embedded-worker-runtime/workers/{worker_id}/transcript?start=0&limit=10" + )) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(transcript_route.status(), StatusCode::NOT_FOUND); let wrong_runtime = app .clone() @@ -5623,10 +5569,7 @@ mod tests { .unwrap(); assert_eq!(wrong_runtime.status(), StatusCode::NOT_FOUND); - let projected = format!( - "{}{}{}{}{}", - embedded_summary, spawned, worker, accepted, transcript - ); + let projected = format!("{}{}{}{}", embedded_summary, spawned, worker, accepted); for forbidden in [ dir.path().to_string_lossy().as_ref(), "metadata.json", diff --git a/web/workspace/src/app.css b/web/workspace/src/app.css index 94d5ca9a..55694a9c 100644 --- a/web/workspace/src/app.css +++ b/web/workspace/src/app.css @@ -967,13 +967,18 @@ } .worker-console-shell { - min-height: 100dvh; - padding-bottom: 0; + display: flex; + flex-direction: column; + min-height: 0; + height: calc(100dvh - (var(--space-6) * 2)); + overflow: hidden; } .worker-console-shell > .console-body { flex: 1 1 auto; min-height: 0; + overflow-y: auto; + overscroll-behavior: contain; } .console-header-actions { @@ -1018,12 +1023,10 @@ .console-log { display: grid; align-content: start; - gap: var(--space-2); - max-height: none; + gap: var(--space-3); min-height: 0; margin: 0; padding: 0; - overflow-y: auto; list-style: none; } @@ -1134,6 +1137,7 @@ position: sticky; bottom: 0; z-index: 2; + flex: 0 0 auto; display: grid; gap: var(--space-3); margin-inline: calc(-1 * var(--space-6)); diff --git a/web/workspace/src/lib/workspace-console/model.test.ts b/web/workspace/src/lib/workspace-console/model.test.ts index b1d92330..42f79356 100644 --- a/web/workspace/src/lib/workspace-console/model.test.ts +++ b/web/workspace/src/lib/workspace-console/model.test.ts @@ -53,71 +53,64 @@ Deno.test("segmentsToText preserves protocol segment semantics", () => { ); }); -Deno.test("projectConsole projects initial console output and live visible protocol rows", () => { - const projection = projectConsole( - [ - { - sequence: 1, - role: "user", - content: "transcript input", - event_id: 10, - }, - ], - [ - { - cursor: "11", - event: { - event: "text_delta", - data: { text: "stream" }, - } satisfies Event, - }, - { - cursor: "12", - event: { - event: "thinking_done", - data: { text: "reasoning" }, - } satisfies Event, - }, - { - cursor: "13", - event: { - event: "tool_result", - data: { - id: "tool-1", - summary: "read file", - output: "content", - is_error: false, - }, - } satisfies Event, - }, - { - cursor: "14", - event: { - event: "usage", - data: { input_tokens: 12, output_tokens: 5 }, - } satisfies Event, - }, - { - cursor: "15", - event: { - event: "error", - data: { code: "invalid_request", message: "bad frame" }, - } satisfies Event, - }, - ], - ); +Deno.test("projectConsole projects visible protocol rows", () => { + const projection = projectConsole([ + { + cursor: "10", + event: { + event: "user_message", + data: { segments: [{ kind: "text", content: "input" }] }, + } satisfies Event, + }, + { + cursor: "11", + event: { + event: "text_delta", + data: { text: "stream" }, + } satisfies Event, + }, + { + cursor: "12", + event: { + event: "thinking_done", + data: { text: "reasoning" }, + } satisfies Event, + }, + { + cursor: "13", + event: { + event: "tool_result", + data: { + id: "tool-1", + summary: "read file", + output: "content", + is_error: false, + }, + } satisfies Event, + }, + { + cursor: "14", + event: { + event: "usage", + data: { input_tokens: 12, output_tokens: 5 }, + } satisfies Event, + }, + { + cursor: "15", + event: { + event: "error", + data: { code: "invalid_request", message: "bad frame" }, + } satisfies Event, + }, + ]); assert( - projection.lines.some((line) => - line.source === "initial" && line.kind === "user" - ), - "initial user row expected", + projection.lines.some((line) => line.source === "event" && line.kind === "user"), + "user protocol row expected", ); assert( - projection.lines.some((line) => - line.source === "live" && line.kind === "assistant" - ), - "assistant live row expected", + projection.lines.some((line) => line.source === "event" && line.kind === "assistant"), + "assistant protocol row expected", ); assert( projection.lines.some((line) => line.kind === "thinking"), @@ -141,85 +134,25 @@ Deno.test("projectConsole projects initial console output and live visible proto ); }); -Deno.test("projectConsole suppresses replayed live conversation rows already present in initial transcript", () => { - const projection = projectConsole( - [ - { - sequence: 1, - role: "user", - content: "hello", - event_id: 10, - }, - { - sequence: 2, - role: "assistant", - content: "world", - event_id: 11, - }, - ], - [ - { - cursor: "12", - event: { - event: "user_message", - data: { segments: [{ kind: "text", content: "hello" }] }, - } satisfies Event, - }, - { - cursor: "13", - event: { event: "text_delta", data: { text: "wo" } } satisfies Event, - }, - { - cursor: "14", - event: { event: "text_delta", data: { text: "rld" } } satisfies Event, - }, - { - cursor: "15", - event: { event: "text_done", data: { text: "world" } } satisfies Event, - }, - { - cursor: "16", - event: { event: "status", data: { status: "idle" } } satisfies Event, - }, - ], - ); - - assertEquals( - projection.lines.map((line) => `${line.source}:${line.kind}:${line.body}`), - ["initial:user:hello", "initial:assistant:world"], - ); - assertEquals(projection.status, "idle"); -}); - -Deno.test("projectConsole preserves live assistant stream not yet present in initial transcript", () => { - const projection = projectConsole( - [ - { - sequence: 1, - role: "user", - content: "hello", - event_id: 10, - }, - ], - [ - { - cursor: "13", - event: { event: "text_delta", data: { text: "new" } } satisfies Event, - }, - ], - ); +Deno.test("projectConsole preserves in-progress assistant protocol stream", () => { + const projection = projectConsole([ + { + cursor: "13", + event: { event: "text_delta", data: { text: "new" } } satisfies Event, + }, + ]); assert( projection.lines.some((line) => - line.source === "live" && line.kind === "assistant" && + line.source === "event" && line.kind === "assistant" && line.body === "new" && line.streaming ), - "live in-progress assistant stream should remain visible", + "in-progress assistant stream should remain visible", ); }); Deno.test("projectConsole keeps protocol lifecycle events out of the console surface", () => { - const projection = projectConsole([], [ + const projection = projectConsole([ { cursor: "30", event: { event: "status", data: { status: "running" } } satisfies Event, @@ -250,7 +183,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", () => { - const projection = projectConsole([], [ + const projection = projectConsole([ { cursor: "20", event: { @@ -270,14 +203,7 @@ Deno.test("projectConsole uses snapshot for state without rendering it as consol status: "running", in_flight: { blocks: [ - { kind: "text", text: "unfinished answer", finished: false }, - { - kind: "tool_call", - id: "call-1", - name: "Read", - args: "{}", - state: "streaming_args", - }, + { kind: "text", text: "partial" }, ], }, }, @@ -285,13 +211,9 @@ Deno.test("projectConsole uses snapshot for state without rendering it as consol }, ]); - assert(projection.status === "running", "snapshot should update status"); - assert( - !projection.lines.some((line) => line.title.includes("snapshot")), - "snapshot should not render as a console row", - ); - assert( - projection.lines.filter((line) => line.kind === "in_flight").length === 2, - "in-flight rows expected", + assertEquals(projection.status, "running"); + assertEquals( + projection.lines.map((line) => `${line.kind}:${line.body}:${line.streaming}`), + ["in_flight:partial:true"], ); }); diff --git a/web/workspace/src/lib/workspace-console/model.ts b/web/workspace/src/lib/workspace-console/model.ts index ea1f3c37..3e41457b 100644 --- a/web/workspace/src/lib/workspace-console/model.ts +++ b/web/workspace/src/lib/workspace-console/model.ts @@ -3,7 +3,6 @@ import type { InFlightBlock, Segment, } from "$lib/generated/protocol"; -import type { WorkerTranscriptItem } from "$lib/workspace-sidebar/types"; import { workspaceRoute } from "$lib/workspace-api/http"; export type ConsoleLineKind = @@ -24,7 +23,7 @@ export type ConsoleLine = { body: string; detail?: string; cursor?: string | null; - source: "initial" | "live"; + source: "event"; streaming?: boolean; error?: boolean; }; @@ -60,103 +59,17 @@ export function workerConsolePath( return workerConsoleHref({ runtime_id: runtimeId, worker_id: workerId }, workspaceId); } -export function initialConsoleLines(items: WorkerTranscriptItem[]): ConsoleLine[] { - return items.map((item) => ({ - id: `initial-${item.event_id}-${item.sequence}`, - kind: initialRoleKind(item.role), - title: item.role, - body: item.content, - source: "initial", - })); -} +export type ConsoleEventInput = { cursor: string; event: ProtocolEvent }; -export function projectConsole( - initialItems: WorkerTranscriptItem[], - events: Array<{ cursor: string; event: ProtocolEvent }> = [], -): ConsoleProjection { - const visibleEvents = dedupeInitialTranscriptReplay(initialItems, events); - return visibleEvents.reduce(applyProtocolEvent, { - lines: initialConsoleLines(initialItems), +export function projectConsole(events: ConsoleEventInput[] = []): ConsoleProjection { + return events.reduce(applyProtocolEvent, { + lines: [], status: null, usage: null, lastCursor: null, }); } -type EventEnvelope = { cursor: string; event: ProtocolEvent }; - -function dedupeInitialTranscriptReplay( - initialItems: WorkerTranscriptItem[], - events: EventEnvelope[], -): EventEnvelope[] { - const remainingInitial = new Map(); - for (const item of initialItems) { - if (item.role !== "user" && item.role !== "assistant") continue; - const key = transcriptKey(item.role, item.content); - remainingInitial.set(key, (remainingInitial.get(key) ?? 0) + 1); - } - - const output: EventEnvelope[] = []; - let pendingAssistant: EventEnvelope[] = []; - let pendingAssistantText = ""; - - const flushPendingAssistant = () => { - if (pendingAssistant.length === 0) return; - output.push(...pendingAssistant); - pendingAssistant = []; - pendingAssistantText = ""; - }; - - for (const envelope of events) { - const event = envelope.event; - if (event.event === "user_message") { - flushPendingAssistant(); - const body = segmentsToText(event.data.segments); - if (consumeTranscriptKey(remainingInitial, "user", body)) continue; - output.push(envelope); - continue; - } - if (event.event === "text_delta") { - pendingAssistant.push(envelope); - pendingAssistantText += event.data.text; - continue; - } - if (event.event === "text_done") { - const body = event.data.text || pendingAssistantText; - if (!consumeTranscriptKey(remainingInitial, "assistant", body)) { - output.push(...pendingAssistant, envelope); - } - pendingAssistant = []; - pendingAssistantText = ""; - continue; - } - flushPendingAssistant(); - output.push(envelope); - } - flushPendingAssistant(); - return output; -} - -function consumeTranscriptKey( - remainingInitial: Map, - role: "user" | "assistant", - body: string, -): boolean { - const key = transcriptKey(role, body); - const remaining = remainingInitial.get(key) ?? 0; - if (remaining <= 0) return false; - if (remaining === 1) { - remainingInitial.delete(key); - } else { - remainingInitial.set(key, remaining - 1); - } - return true; -} - -function transcriptKey(role: "user" | "assistant", body: string): string { - return `${role}\0${body}`; -} - export function applyProtocolEvent( projection: ConsoleProjection, envelope: { cursor: string; event: ProtocolEvent }, @@ -353,13 +266,6 @@ export function segmentsToText(segments: Segment[]): string { .join("\n"); } -function initialRoleKind(role: string): ConsoleLineKind { - if (role === "user" || role === "assistant" || role === "system") { - return role; - } - return "system"; -} - function line( cursor: string, kind: ConsoleLineKind, @@ -376,7 +282,7 @@ function line( body, detail, cursor, - source: "live", + source: "event", streaming, error, }; diff --git a/web/workspace/src/lib/workspace-console/worker-console.ui.test.ts b/web/workspace/src/lib/workspace-console/worker-console.ui.test.ts index 63117750..292c0667 100644 --- a/web/workspace/src/lib/workspace-console/worker-console.ui.test.ts +++ b/web/workspace/src/lib/workspace-console/worker-console.ui.test.ts @@ -52,6 +52,35 @@ Deno.test("workspace Worker list lives on the dedicated Workers page", async () ); }); +Deno.test("Worker Console uses protocol observation events without transcript fetch", async () => { + const consolePage = await Deno.readTextFile( + new URL("./../../routes/w/[workspaceId]/runtimes/[runtimeId]/workers/[workerId]/console/+page.svelte", import.meta.url), + ); + + assert( + consolePage.includes("seenObservationEventIds") && + consolePage.includes("rememberObservationEvent(frame.envelope.event_id)") && + consolePage.includes("projectConsole(observedEvents.map") && + !consolePage.includes("/transcript") && + !consolePage.includes("WorkerTranscriptProjection"), + "Console should render protocol observation replay/live events directly and dedupe repeated frames by event id", + ); +}); + +Deno.test("Decodal source editor keeps imperative EditorView out of reactive state", async () => { + const editor = await Deno.readTextFile( + new URL("../workspace-settings/DecodalSourceEditor.svelte", import.meta.url), + ); + + assert( + editor.includes("let view: EditorView | null = null") && + !editor.includes("$state value)") && + editor.includes("untrack(() => onChange)"), + "CodeMirror EditorView must not be reactive state; otherwise mount cleanup can loop forever", + ); +}); + Deno.test("workspace Runtime management pages expose Runtimes and Runtime-owned workdirs", async () => { const sidebar = await Deno.readTextFile( new URL("../workspace-sidebar/WorkspaceSidebar.svelte", import.meta.url), @@ -138,9 +167,9 @@ Deno.test("Worker Console page is routed by runtime_id and worker_id through bac "Worker detail should use the scoped backend Worker detail API", ); assert( - consolePage.includes("/transcript?limit=200") && + !consolePage.includes("/transcript") && consolePage.includes("/events/ws") && consolePage.includes("/input"), - "Console should use bounded transcript, observation WS, and input APIs", + "Console should use protocol observation WS and input APIs without a transcript API", ); assert( !consolePage.includes("/api/companion"), diff --git a/web/workspace/src/lib/workspace-settings/DecodalSourceEditor.svelte b/web/workspace/src/lib/workspace-settings/DecodalSourceEditor.svelte index 90af4525..16db8214 100644 --- a/web/workspace/src/lib/workspace-settings/DecodalSourceEditor.svelte +++ b/web/workspace/src/lib/workspace-settings/DecodalSourceEditor.svelte @@ -1,4 +1,5 @@