From e298856c856d310b45d052a0e07450b6aa333cf1 Mon Sep 17 00:00:00 2001 From: Hare Date: Tue, 14 Jul 2026 07:55:51 +0900 Subject: [PATCH] feat: add web console commands --- crates/manifest/src/profile.rs | 26 +++ crates/tui/src/app.rs | 57 +++++ crates/worker-runtime/src/execution.rs | 18 ++ crates/worker-runtime/src/http_server.rs | 36 +++ crates/worker-runtime/src/interaction.rs | 14 ++ crates/worker-runtime/src/runtime.rs | 50 ++++- crates/worker-runtime/src/worker_backend.rs | 63 ++++-- crates/worker/src/controller.rs | 38 ++++ crates/worker/src/ipc/server.rs | 32 +-- crates/worker/src/worker.rs | 102 +++++++-- crates/workspace-server/src/hosts.rs | 166 +++++++++++++- crates/workspace-server/src/server.rs | 45 +++- resources/profiles/default.dcdl | 7 + web/workspace/deno.json | 2 +- web/workspace/src/app.css | 9 + .../lib/workspace-console/chat-submit.test.ts | 28 ++- .../src/lib/workspace-console/chat-submit.ts | 2 +- .../composer-command.test.ts | 64 ++++++ .../lib/workspace-console/composer-command.ts | 208 ++++++++++++++++++ .../composer-completion.test.ts | 54 +++++ .../workspace-console/composer-completion.ts | 85 +++++++ .../src/lib/workspace-console/model.test.ts | 125 +++++++++++ .../src/lib/workspace-console/model.ts | 119 +++++++++- .../workers/[workerId]/console/+page.svelte | 117 +++++++++- 24 files changed, 1365 insertions(+), 102 deletions(-) create mode 100644 web/workspace/src/lib/workspace-console/composer-command.test.ts create mode 100644 web/workspace/src/lib/workspace-console/composer-command.ts create mode 100644 web/workspace/src/lib/workspace-console/composer-completion.test.ts create mode 100644 web/workspace/src/lib/workspace-console/composer-completion.ts diff --git a/crates/manifest/src/profile.rs b/crates/manifest/src/profile.rs index 953db338..460adb38 100644 --- a/crates/manifest/src/profile.rs +++ b/crates/manifest/src/profile.rs @@ -970,6 +970,12 @@ fn builtin_default_profile_artifact() -> serde_json::Value { "model": { "ref": "codex-oauth/gpt-5.5" }, "session": { "record_event_trace": true }, "engine": { "reasoning": "high" }, + "compaction": { + "kind": "tokens", + "threshold": 240000, + "request_threshold": 270000, + "worker_context_max_tokens": 100000 + }, "feature": { "task": { "enabled": true }, "memory": { "enabled": true }, @@ -1441,6 +1447,22 @@ mod tests { assert!(companion.delegation_scope.allow.is_empty()); assert_eq!(companion.model.ref_.as_deref(), Some("codex-oauth/gpt-5.5")); assert!(companion.web.is_some()); + assert_eq!( + companion.compaction.as_ref().unwrap().threshold, + Some(240000) + ); + assert_eq!( + companion.compaction.as_ref().unwrap().request_threshold, + Some(270000) + ); + assert_eq!( + companion + .compaction + .as_ref() + .unwrap() + .worker_context_max_tokens, + 100000 + ); let intake = resolve("intake"); assert!(intake.feature.task.enabled); @@ -1450,6 +1472,7 @@ mod tests { assert!(intake.delegation_scope.allow.is_empty()); assert_eq!(intake.model.ref_.as_deref(), Some("codex-oauth/gpt-5.5")); assert!(intake.web.is_some()); + assert!(intake.compaction.is_some()); assert!(!intake.feature.ticket_orchestration.enabled); let orchestrator = resolve("orchestrator"); @@ -1464,6 +1487,7 @@ mod tests { Some("codex-oauth/gpt-5.5") ); assert!(orchestrator.web.is_some()); + assert!(orchestrator.compaction.is_some()); let coder = resolve("coder"); assert!(coder.feature.task.enabled); @@ -1472,6 +1496,7 @@ mod tests { assert!(coder.delegation_scope.allow.is_empty()); assert_eq!(coder.model.ref_.as_deref(), Some("codex-oauth/gpt-5.5")); assert!(coder.web.is_some()); + assert!(coder.compaction.is_some()); let reviewer = resolve("reviewer"); assert!(reviewer.feature.task.enabled); @@ -1481,6 +1506,7 @@ mod tests { assert!(reviewer.delegation_scope.allow.is_empty()); assert_eq!(reviewer.model.ref_.as_deref(), Some("codex-oauth/gpt-5.5")); assert!(reviewer.web.is_some()); + assert!(reviewer.compaction.is_some()); } #[test] diff --git a/crates/tui/src/app.rs b/crates/tui/src/app.rs index 57558434..e40c4b47 100644 --- a/crates/tui/src/app.rs +++ b/crates/tui/src/app.rs @@ -2082,6 +2082,11 @@ impl App { let value = serde_json::to_value(&item).expect("SystemItem is Serialize"); self.apply_system_item(&value); } + session_store::LogEntry::Extension { + domain, payload, .. + } if domain == "yoi.compaction" => { + self.apply_compaction_extension(&payload); + } // Non-history-bearing variants don't affect the block view. _ => {} } @@ -2092,6 +2097,58 @@ impl App { /// Kind-based routing replaces the old free-text `[Notification]` / /// `[File: …]` parsing path: each kind maps directly to a typed /// block (`Block::Notify`, `Block::WorkerEvent`, …). + fn apply_compaction_extension(&mut self, payload: &serde_json::Value) { + if payload.get("kind").and_then(|value| value.as_str()) != Some("compaction_block") { + return; + } + match payload.get("state").and_then(|value| value.as_str()) { + Some("running") => { + if self.last_streaming_compact_mut().is_none() { + self.blocks.push(Block::Compact(CompactEvent::Streaming { + started_at: Instant::now(), + })); + } + } + Some("done") => { + let new_segment_id = payload + .get("new_segment_id") + .and_then(|value| value.as_str()) + .and_then(|value| value.parse::().ok()) + .unwrap_or_else(uuid::Uuid::nil); + if let Some(evt) = self.last_streaming_compact_mut() { + *evt = CompactEvent::Done { + new_segment_id, + elapsed_secs: None, + }; + } else { + self.blocks.push(Block::Compact(CompactEvent::Done { + new_segment_id, + elapsed_secs: None, + })); + } + } + Some("failed") => { + let error = payload + .get("error") + .and_then(|value| value.as_str()) + .unwrap_or("compact failed") + .to_string(); + if let Some(evt) = self.last_streaming_compact_mut() { + *evt = CompactEvent::Failed { + error, + elapsed_secs: None, + }; + } else { + self.blocks.push(Block::Compact(CompactEvent::Failed { + error, + elapsed_secs: None, + })); + } + } + _ => {} + } + } + fn apply_system_item(&mut self, value: &serde_json::Value) { let Ok(item) = serde_json::from_value::(value.clone()) else { // Unknown / forward-compat shape: fall back to rendering the diff --git a/crates/worker-runtime/src/execution.rs b/crates/worker-runtime/src/execution.rs index faa452aa..d655e54b 100644 --- a/crates/worker-runtime/src/execution.rs +++ b/crates/worker-runtime/src/execution.rs @@ -398,6 +398,15 @@ pub trait WorkerExecutionBackend: Send + Sync + 'static { fn worker_snapshot(&self, _handle: &WorkerExecutionHandle) -> Option { None } + + fn worker_completions( + &self, + _handle: &WorkerExecutionHandle, + _kind: protocol::CompletionKind, + _prefix: &str, + ) -> Vec { + Vec::new() + } } #[derive(Clone)] @@ -485,6 +494,15 @@ impl WorkerExecutionBackendRef { ) -> Option { self.backend.worker_snapshot(handle) } + + pub(crate) fn worker_completions( + &self, + handle: &WorkerExecutionHandle, + kind: protocol::CompletionKind, + prefix: &str, + ) -> Vec { + self.backend.worker_completions(handle, kind, prefix) + } } impl fmt::Debug for WorkerExecutionBackendRef { diff --git a/crates/worker-runtime/src/http_server.rs b/crates/worker-runtime/src/http_server.rs index c543ac08..a108faba 100644 --- a/crates/worker-runtime/src/http_server.rs +++ b/crates/worker-runtime/src/http_server.rs @@ -146,6 +146,10 @@ pub fn runtime_http_router(runtime: Runtime, local_token: Option) -> Rou get(get_worker).delete(delete_worker), ) .route("/v1/workers/{worker_id}/input", post(send_worker_input)) + .route( + "/v1/workers/{worker_id}/completions", + post(worker_completions), + ) .route("/v1/workers/{worker_id}/stop", post(stop_worker)) .route("/v1/workers/{worker_id}/cancel", post(cancel_worker)); @@ -228,6 +232,20 @@ pub struct RuntimeHttpWorkerInputResponse { pub ack: WorkerInteractionAck, } +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct RuntimeHttpWorkerCompletionsRequest { + pub kind: protocol::CompletionKind, + #[serde(default)] + pub prefix: String, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct RuntimeHttpWorkerCompletionsResponse { + pub kind: protocol::CompletionKind, + pub prefix: String, + pub entries: Vec, +} + /// Worker lifecycle request body used by stop/cancel endpoints. #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct RuntimeHttpWorkerLifecycleRequest { @@ -678,6 +696,24 @@ async fn send_worker_input( Ok(Json(RuntimeHttpWorkerInputResponse { ack })) } +async fn worker_completions( + State(state): State, + Path(worker_id): Path, + body: Result, JsonRejection>, +) -> RestResult { + let worker_ref = worker_ref_for(&state.runtime, worker_id)?; + let Json(request) = body.map_err(RuntimeHttpRestError::json_rejection)?; + let entries = state + .runtime + .worker_completions(&worker_ref, request.kind, &request.prefix) + .map_err(RuntimeHttpRestError::runtime)?; + Ok(Json(RuntimeHttpWorkerCompletionsResponse { + kind: request.kind, + prefix: request.prefix, + entries, + })) +} + async fn stop_worker( State(state): State, Path(worker_id): Path, diff --git a/crates/worker-runtime/src/interaction.rs b/crates/worker-runtime/src/interaction.rs index 60e12a0b..225544c7 100644 --- a/crates/worker-runtime/src/interaction.rs +++ b/crates/worker-runtime/src/interaction.rs @@ -1,5 +1,6 @@ use crate::catalog::WorkerStatus; use crate::identity::WorkerRef; +use protocol::Segment; use serde::{Deserialize, Serialize}; /// Input kind accepted by the embedded interaction API. @@ -8,6 +9,15 @@ use serde::{Deserialize, Serialize}; pub enum WorkerInputKind { User, System, + Compact, + ListRewindTargets, + RegisterPeer, +} + +impl WorkerInputKind { + pub fn is_empty_content_allowed(&self) -> bool { + matches!(self, Self::Compact | Self::ListRewindTargets) + } } /// Worker input request accepted by a Runtime Worker. @@ -15,6 +25,8 @@ pub enum WorkerInputKind { pub struct WorkerInput { pub kind: WorkerInputKind, pub content: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub segments: Option>, } impl WorkerInput { @@ -22,6 +34,7 @@ impl WorkerInput { Self { kind: WorkerInputKind::User, content: content.into(), + segments: None, } } @@ -29,6 +42,7 @@ impl WorkerInput { Self { kind: WorkerInputKind::System, content: content.into(), + segments: None, } } } diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index f3866fd3..e5c057a7 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -537,6 +537,28 @@ impl Runtime { }) } + /// Return live completion entries for the Worker composer. + pub fn worker_completions( + &self, + worker_ref: &WorkerRef, + kind: protocol::CompletionKind, + prefix: &str, + ) -> Result, RuntimeError> { + let (backend, handle) = { + let state = self.lock()?; + state.ensure_worker_ref(worker_ref)?; + let worker = state.worker(worker_ref)?; + ( + state.execution_backend.clone(), + worker.execution_handle.clone(), + ) + }; + let Some((backend, handle)) = backend.zip(handle) else { + return Ok(Vec::new()); + }; + Ok(backend.worker_completions(&handle, kind, prefix)) + } + fn commit_created_worker( &self, worker_ref: &WorkerRef, @@ -1730,11 +1752,20 @@ fn validate_create_worker_request(request: &CreateWorkerRequest) -> Result<(), R } fn validate_worker_input(input: &WorkerInput) -> Result<(), RuntimeError> { - if input.content.trim().is_empty() { + if !input.kind.is_empty_content_allowed() && input.content.trim().is_empty() { return Err(RuntimeError::InvalidRequest( "worker input content must not be empty".to_string(), )); } + if input + .segments + .as_ref() + .is_some_and(|segments| segments.is_empty()) + { + return Err(RuntimeError::InvalidRequest( + "worker input segments must not be empty".to_string(), + )); + } Ok(()) } @@ -1742,9 +1773,11 @@ fn validate_worker_input(input: &WorkerInput) -> Result<(), RuntimeError> { 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(), - }], + segments: input.segments.clone().unwrap_or_else(|| { + vec![protocol::Segment::Text { + content: input.content.clone(), + }] + }), }, WorkerInputKind::System => protocol::Event::SystemItem { item: serde_json::json!({ @@ -1752,6 +1785,15 @@ fn input_protocol_event(input: &WorkerInput) -> protocol::Event { "content": input.content.clone(), }), }, + WorkerInputKind::Compact + | WorkerInputKind::ListRewindTargets + | WorkerInputKind::RegisterPeer => protocol::Event::SystemItem { + item: serde_json::json!({ + "kind": "embedded_worker_command_input", + "command": input.kind, + "content": input.content.clone(), + }), + }, } } diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index 2a5d08c4..e473eae7 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -996,31 +996,38 @@ where ); } - let WorkerInputKind::User = input.kind else { - busy.store(false, Ordering::SeqCst); - return WorkerExecutionResult::unsupported( - WorkerExecutionOperation::Input, - "runtime adapter currently dispatches user input only", - ); + let method = match input.kind { + WorkerInputKind::User => Method::Run { + input: input + .segments + .unwrap_or_else(|| vec![Segment::text(input.content.trim().to_string())]), + }, + WorkerInputKind::System => Method::Notify { + message: input.content, + auto_run: true, + }, + WorkerInputKind::Compact => Method::Compact, + WorkerInputKind::ListRewindTargets => Method::ListRewindTargets, + WorkerInputKind::RegisterPeer => Method::RegisterPeer { + name: input.content.trim().to_string(), + }, }; - let content = input.content.trim().to_string(); - if content.is_empty() { - busy.store(false, Ordering::SeqCst); - return WorkerExecutionResult::rejected( - WorkerExecutionOperation::Input, - "runtime adapter rejects empty user input", - ); - } + let accepted_run_state = match method { + Method::Run { .. } | Method::Notify { .. } | Method::Compact => { + WorkerExecutionRunState::Busy + } + _ => WorkerExecutionRunState::Idle, + }; + let accepted_is_idle = accepted_run_state == WorkerExecutionRunState::Idle; let result = self.send_method( WorkerExecutionOperation::Input, worker, - Method::Run { - input: vec![Segment::text(content)], - }, - WorkerExecutionRunState::Busy, + method, + accepted_run_state, ); - if result.outcome != crate::execution::WorkerExecutionOutcome::Accepted { + if accepted_is_idle || result.outcome != crate::execution::WorkerExecutionOutcome::Accepted + { busy.store(false, Ordering::SeqCst); } result @@ -1093,6 +1100,24 @@ where .get(handle.worker_ref()) .map(|execution| execution.handle.snapshot_event()) } + + fn worker_completions( + &self, + handle: &WorkerExecutionHandle, + kind: protocol::CompletionKind, + prefix: &str, + ) -> Vec { + if handle.backend_id() != self.backend_id() { + return Vec::new(); + } + let Ok(workers) = self.workers.lock() else { + return Vec::new(); + }; + workers + .get(handle.worker_ref()) + .map(|execution| execution.handle.completion_entries(kind, prefix)) + .unwrap_or_default() + } } #[cfg(test)] diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index ad5cdae6..891e7b3e 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -91,6 +91,44 @@ impl WorkerHandle { (event, entry_rx) } + pub fn completion_entries( + &self, + kind: protocol::CompletionKind, + prefix: &str, + ) -> Vec { + match kind { + protocol::CompletionKind::File => self + .shared_state + .fs_view() + .map(|view| view.list_file_completions(prefix)) + .unwrap_or_default() + .into_iter() + .map(|c| protocol::CompletionEntry { + value: c.path, + is_dir: c.is_dir, + }) + .collect(), + protocol::CompletionKind::Knowledge => self + .shared_state + .list_knowledge_completions(prefix) + .into_iter() + .map(|c| protocol::CompletionEntry { + value: c.slug, + is_dir: false, + }) + .collect(), + protocol::CompletionKind::Workflow => self + .shared_state + .list_workflow_completions(prefix) + .into_iter() + .map(|c| protocol::CompletionEntry { + value: c.slug, + is_dir: false, + }) + .collect(), + } + } + /// Broadcast an event to all listeners (including socket clients). pub fn send_event(&self, event: Event) -> Result> { self.event_tx.send(event) diff --git a/crates/worker/src/ipc/server.rs b/crates/worker/src/ipc/server.rs index d366f029..885ddfc7 100644 --- a/crates/worker/src/ipc/server.rs +++ b/crates/worker/src/ipc/server.rs @@ -165,37 +165,7 @@ async fn handle_connection(stream: tokio::net::UnixStream, handle: WorkerHandle) method = reader.next::() => { match method { Ok(Some(Method::ListCompletions { kind, prefix })) => { - let entries = match kind { - protocol::CompletionKind::File => handle - .shared_state - .fs_view() - .map(|view| view.list_file_completions(&prefix)) - .unwrap_or_default() - .into_iter() - .map(|c| protocol::CompletionEntry { - value: c.path, - is_dir: c.is_dir, - }) - .collect(), - protocol::CompletionKind::Knowledge => handle - .shared_state - .list_knowledge_completions(&prefix) - .into_iter() - .map(|c| protocol::CompletionEntry { - value: c.slug, - is_dir: false, - }) - .collect(), - protocol::CompletionKind::Workflow => handle - .shared_state - .list_workflow_completions(&prefix) - .into_iter() - .map(|c| protocol::CompletionEntry { - value: c.slug, - is_dir: false, - }) - .collect(), - }; + let entries = handle.completion_entries(kind, &prefix); if writer .write(&Event::Completions { kind, entries }) .await diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index 8ad93cb5..f2b9d69b 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -38,6 +38,9 @@ use crate::hook::{ PreToolCall, }; use crate::in_flight::InFlightEvents; + +const COMPACTION_EXTENSION_DOMAIN: &str = "yoi.compaction"; +const COMPACTION_BLOCK_ID: &str = "compact"; use crate::ipc::alerter::Alerter; use crate::ipc::interceptor::WorkerInterceptor; use crate::ipc::notify_buffer::NotifyBuffer; @@ -2309,6 +2312,55 @@ impl Worker { .map_err(WorkerError::Engine) } + fn persist_compaction_block( + &mut self, + state: &str, + message: &str, + error: Option<&str>, + new_segment_id: Option, + ) -> Result<(), WorkerError> { + let payload = serde_json::json!({ + "kind": "compaction_block", + "schema_version": 1, + "block_id": COMPACTION_BLOCK_ID, + "state": state, + "message": message, + "error": error, + "new_segment_id": new_segment_id.map(|id| id.to_string()), + }); + Ok(self.commit_entry(LogEntry::Extension { + ts: segment_log::now_millis(), + domain: COMPACTION_EXTENSION_DOMAIN.into(), + payload, + })?) + } + + fn persist_and_send_compact_start(&mut self) -> Result<(), WorkerError> { + self.persist_compaction_block("running", "Compacting…", None, None)?; + self.send_event(Event::CompactStart); + Ok(()) + } + + fn persist_and_send_compact_done( + &mut self, + new_segment_id: SegmentId, + ) -> Result<(), WorkerError> { + self.persist_compaction_block("done", "Compacted.", None, Some(new_segment_id))?; + self.send_event(Event::CompactDone { new_segment_id }); + Ok(()) + } + + fn persist_and_send_compact_failed(&mut self, error: String) -> Result<(), WorkerError> { + self.persist_compaction_block( + "failed", + &format!("Compact failed: {error}"), + Some(error.as_str()), + None, + )?; + self.send_event(Event::CompactFailed { error }); + Ok(()) + } + /// Perform compaction after a `compact_needed` abort and resume execution. /// /// Uses `Box::pin` for the recursive `resume()` call to break the @@ -2334,14 +2386,14 @@ impl Worker { .map(|s| s.retained_tokens()) .unwrap_or(manifest::defaults::COMPACT_RETAINED_TOKENS); - self.send_event(Event::CompactStart); + self.persist_and_send_compact_start()?; match self.compact(retained).await { Ok(new_segment_id) => { info!( new_segment_id = %new_segment_id, "Compaction succeeded, resuming execution" ); - self.send_event(Event::CompactDone { new_segment_id }); + self.persist_and_send_compact_done(new_segment_id)?; if let Some(ref state) = self.compact_state { state.record_compact_success(); } @@ -2349,9 +2401,7 @@ impl Worker { } Err(e) => { warn!(error = %e, "Compaction failed during run"); - self.send_event(Event::CompactFailed { - error: e.to_string(), - }); + self.persist_and_send_compact_failed(e.to_string())?; self.alert( AlertLevel::Error, AlertSource::Compactor, @@ -2384,21 +2434,45 @@ impl Worker { } let retained = state.retained_tokens(); - self.send_event(Event::CompactStart); + if let Err(err) = self.persist_and_send_compact_start() { + warn!(error = %err, "failed to persist proactive compact start"); + self.alert( + AlertLevel::Warn, + AlertSource::Compactor, + format!("pre-run compaction not started: failed to persist status block: {err}"), + ); + return; + } match self.compact(retained).await { Ok(new_segment_id) => { info!( new_segment_id = %new_segment_id, "Proactive pre-run compaction succeeded" ); - self.send_event(Event::CompactDone { new_segment_id }); + if let Err(err) = self.persist_and_send_compact_done(new_segment_id) { + warn!(error = %err, "failed to persist proactive compact completion"); + self.alert( + AlertLevel::Warn, + AlertSource::Compactor, + format!( + "pre-run compaction completed but status block was not persisted: {err}" + ), + ); + } state.record_compact_success(); } Err(e) => { warn!(error = %e, "Proactive pre-run compaction failed"); - self.send_event(Event::CompactFailed { - error: e.to_string(), - }); + if let Err(err) = self.persist_and_send_compact_failed(e.to_string()) { + warn!(error = %err, "failed to persist proactive compact failure"); + self.alert( + AlertLevel::Warn, + AlertSource::Compactor, + format!( + "pre-run compaction failed and status block was not persisted: {err}" + ), + ); + } self.alert( AlertLevel::Warn, AlertSource::Compactor, @@ -2456,11 +2530,11 @@ impl Worker { } self.join_memory_task().await; - self.send_event(Event::CompactStart); + self.persist_and_send_compact_start()?; match self.compact(retained).await { Ok(new_segment_id) => { info!(new_segment_id = %new_segment_id, "Manual compaction succeeded"); - self.send_event(Event::CompactDone { new_segment_id }); + self.persist_and_send_compact_done(new_segment_id)?; if let Some(ref state) = state { state.record_compact_success(); } @@ -2468,9 +2542,7 @@ impl Worker { } Err(e) => { warn!(error = %e, "Manual compaction failed"); - self.send_event(Event::CompactFailed { - error: e.to_string(), - }); + self.persist_and_send_compact_failed(e.to_string())?; self.alert( AlertLevel::Error, AlertSource::Compactor, diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index 12a60b57..629de1c8 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -31,7 +31,8 @@ use worker_runtime::execution::{ use worker_runtime::fs_store::FsRuntimeStoreOptions; use worker_runtime::http_server::{ RuntimeHttpConfigBundleAvailabilityResponse, RuntimeHttpConfigBundleSyncRequest, - RuntimeHttpErrorResponse, RuntimeHttpSummaryResponse, RuntimeHttpWorkerDeleteResponse, + RuntimeHttpErrorResponse, RuntimeHttpSummaryResponse, RuntimeHttpWorkerCompletionsRequest, + RuntimeHttpWorkerCompletionsResponse, RuntimeHttpWorkerDeleteResponse, RuntimeHttpWorkerInputResponse, RuntimeHttpWorkerLifecycleRequest, RuntimeHttpWorkerLifecycleResponse, RuntimeHttpWorkerResponse, RuntimeHttpWorkersResponse, RuntimeHttpWorkingDirectoriesResponse, RuntimeHttpWorkingDirectoryResponse, @@ -436,6 +437,9 @@ pub struct WorkerLifecycleResult { pub enum WorkerInputKind { User, System, + Compact, + ListRewindTargets, + RegisterPeer, } #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] @@ -452,6 +456,25 @@ pub struct WorkerInputRequest { #[serde(default = "default_worker_input_kind")] pub kind: WorkerInputKind, pub content: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub segments: Option>, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct WorkerCompletionsRequest { + pub kind: protocol::CompletionKind, + #[serde(default)] + pub prefix: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct WorkerCompletionsResult { + pub runtime_id: String, + pub worker_id: String, + pub kind: protocol::CompletionKind, + pub prefix: String, + pub entries: Vec, + pub diagnostics: Vec, } #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] @@ -712,6 +735,25 @@ pub trait WorkspaceWorkerRuntime: Send + Sync { } } + fn worker_completions( + &self, + worker_id: &str, + request: WorkerCompletionsRequest, + ) -> WorkerCompletionsResult { + WorkerCompletionsResult { + runtime_id: self.runtime_id().to_string(), + worker_id: worker_id.to_string(), + kind: request.kind, + prefix: request.prefix, + entries: Vec::new(), + diagnostics: vec![diagnostic( + "worker_completions_unsupported", + DiagnosticSeverity::Info, + format!("runtime does not implement completions for worker '{worker_id}'"), + )], + } + } + fn proxy_connect_points(&self, worker_id: &str) -> Vec { vec![WorkerProxyConnectPoint { kind: "stream_proxy".to_string(), @@ -1020,6 +1062,26 @@ impl RuntimeRegistry { Ok(runtime.send_input(worker_id, request)) } + pub fn worker_completions( + &self, + runtime_id: &str, + worker_id: &str, + request: WorkerCompletionsRequest, + ) -> 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.worker_completions(worker_id, request)) + } + pub fn stop_worker( &self, runtime_id: &str, @@ -1750,8 +1812,12 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime { kind: match request.kind { WorkerInputKind::User => EmbeddedWorkerInputKind::User, WorkerInputKind::System => EmbeddedWorkerInputKind::System, + WorkerInputKind::Compact => EmbeddedWorkerInputKind::Compact, + WorkerInputKind::ListRewindTargets => EmbeddedWorkerInputKind::ListRewindTargets, + WorkerInputKind::RegisterPeer => EmbeddedWorkerInputKind::RegisterPeer, }, content: request.content, + segments: request.segments, }; match self.runtime.send_input(&worker_ref, input) { Ok(ack) => WorkerInputResult { @@ -1768,6 +1834,64 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime { ), } } + + fn worker_completions( + &self, + worker_id: &str, + request: WorkerCompletionsRequest, + ) -> WorkerCompletionsResult { + if !self.execution_enabled { + return WorkerCompletionsResult { + runtime_id: self.runtime_id.clone(), + worker_id: worker_id.to_string(), + kind: request.kind, + prefix: request.prefix, + entries: Vec::new(), + diagnostics: vec![diagnostic( + "embedded_worker_execution_unavailable", + DiagnosticSeverity::Info, + format!( + "worker completions for '{worker_id}' require an embedded execution backend" + ), + )], + }; + } + let Some(worker_ref) = self.worker_ref(worker_id) else { + return WorkerCompletionsResult { + runtime_id: self.runtime_id.clone(), + worker_id: worker_id.to_string(), + kind: request.kind, + prefix: request.prefix, + entries: Vec::new(), + diagnostics: vec![diagnostic( + "embedded_worker_id_invalid", + DiagnosticSeverity::Warning, + "Worker id was empty and cannot be resolved".to_string(), + )], + }; + }; + match self + .runtime + .worker_completions(&worker_ref, request.kind, &request.prefix) + { + Ok(entries) => WorkerCompletionsResult { + runtime_id: self.runtime_id.clone(), + worker_id: worker_id.to_string(), + kind: request.kind, + prefix: request.prefix, + entries, + diagnostics: Vec::new(), + }, + Err(error) => WorkerCompletionsResult { + runtime_id: self.runtime_id.clone(), + worker_id: worker_id.to_string(), + kind: request.kind, + prefix: request.prefix, + entries: Vec::new(), + diagnostics: vec![embedded_runtime_diagnostic(&error)], + }, + } + } } #[derive(Clone)] @@ -2397,8 +2521,12 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { kind: match request.kind { WorkerInputKind::User => EmbeddedWorkerInputKind::User, WorkerInputKind::System => EmbeddedWorkerInputKind::System, + WorkerInputKind::Compact => EmbeddedWorkerInputKind::Compact, + WorkerInputKind::ListRewindTargets => EmbeddedWorkerInputKind::ListRewindTargets, + WorkerInputKind::RegisterPeer => EmbeddedWorkerInputKind::RegisterPeer, }, content: request.content, + segments: request.segments, }; match self.post_json::<_, RuntimeHttpWorkerInputResponse>( &format!("/v1/workers/{worker_id}/input"), @@ -2414,6 +2542,38 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { Err(diagnostic) => remote_input_rejected(&self.runtime_id, worker_id, diagnostic), } } + + fn worker_completions( + &self, + worker_id: &str, + request: WorkerCompletionsRequest, + ) -> WorkerCompletionsResult { + let request = RuntimeHttpWorkerCompletionsRequest { + kind: request.kind, + prefix: request.prefix, + }; + match self.post_json::<_, RuntimeHttpWorkerCompletionsResponse>( + &format!("/v1/workers/{worker_id}/completions"), + &request, + ) { + Ok(response) => WorkerCompletionsResult { + runtime_id: self.runtime_id.clone(), + worker_id: worker_id.to_string(), + kind: response.kind, + prefix: response.prefix, + entries: response.entries, + diagnostics: Vec::new(), + }, + Err(diagnostic) => WorkerCompletionsResult { + runtime_id: self.runtime_id.clone(), + worker_id: worker_id.to_string(), + kind: request.kind, + prefix: request.prefix, + entries: Vec::new(), + diagnostics: vec![diagnostic], + }, + } + } } fn embedded_runtime_capabilities( @@ -3676,6 +3836,7 @@ mod tests { request.initial_input = Some(EmbeddedWorkerInput { kind: EmbeddedWorkerInputKind::System, content: "system/role instruction belongs in profile".to_string(), + segments: None, }); let spawned = runtime.spawn_worker(request); @@ -3707,6 +3868,7 @@ mod tests { WorkerInputRequest { kind: WorkerInputKind::User, content: "hello".to_string(), + segments: None, }, ); assert_eq!(input.state, WorkerOperationState::Accepted); @@ -3792,6 +3954,7 @@ mod tests { WorkerInputRequest { kind: WorkerInputKind::User, content: "hello embedded runtime".to_string(), + segments: None, }, ) .unwrap(); @@ -4007,6 +4170,7 @@ mod tests { WorkerInputRequest { kind: WorkerInputKind::User, content: "hello remote".to_string(), + segments: None, }, ) .unwrap(); diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index dbfd0a12..a859941a 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -28,10 +28,10 @@ use crate::hosts::{ ConfigBundleCheckResult, ConfigBundleSyncResult, DiagnosticSeverity, EmbeddedWorkerRuntime, HostSummary, RemoteRuntimeConfig, RemoteWorkerRuntime, RuntimeDiagnostic, RuntimeRegistry, RuntimeRegistryError, RuntimeRegistryUnregisterResult, RuntimeSummary, WorkerCapabilitySummary, - WorkerImplementationSummary, WorkerInputRequest, WorkerInputResult, WorkerLifecycleRequest, - WorkerLifecycleResult, WorkerOperationState, WorkerSpawnAcceptanceRequirement, - WorkerSpawnIntent, WorkerSpawnRequest, WorkerSpawnResult, WorkerSpawnWorkingDirectoryRequest, - WorkerSummary, WorkerWorkspaceSummary, + WorkerCompletionsRequest, WorkerCompletionsResult, WorkerImplementationSummary, + WorkerInputRequest, WorkerInputResult, WorkerLifecycleRequest, WorkerLifecycleResult, + WorkerOperationState, WorkerSpawnAcceptanceRequirement, WorkerSpawnIntent, WorkerSpawnRequest, + WorkerSpawnResult, WorkerSpawnWorkingDirectoryRequest, WorkerSummary, WorkerWorkspaceSummary, }; use crate::identity::WorkspaceIdentity; use crate::observation::{ @@ -496,6 +496,14 @@ pub fn build_router(api: WorkspaceApi) -> Router { "/api/w/{workspace_id}/runtimes/{runtime_id}/workers/{worker_id}/input", post(scoped_send_runtime_worker_input), ) + .route( + "/api/runtimes/{runtime_id}/workers/{worker_id}/completions", + post(runtime_worker_completions), + ) + .route( + "/api/w/{workspace_id}/runtimes/{runtime_id}/workers/{worker_id}/completions", + post(scoped_runtime_worker_completions), + ) .route( "/api/runtimes/{runtime_id}/workers/{worker_id}/stop", post(stop_runtime_worker), @@ -2252,6 +2260,20 @@ async fn scoped_send_runtime_worker_input( .await } +async fn scoped_runtime_worker_completions( + State(api): State, + AxumPath(path): AxumPath, + Json(request): Json, +) -> ApiResult> { + validate_workspace_scope(&api, &path.workspace_id)?; + runtime_worker_completions( + State(api), + AxumPath((path.runtime_id, path.worker_id)), + Json(request), + ) + .await +} + async fn scoped_stop_runtime_worker( State(api): State, AxumPath(path): AxumPath, @@ -2786,6 +2808,7 @@ async fn create_workspace_worker( Some(EmbeddedWorkerInput { kind: EmbeddedWorkerInputKind::User, content: initial_text, + segments: None, }) }; let selected_working_directory_id = request @@ -3137,6 +3160,18 @@ async fn send_runtime_worker_input( Ok(Json(result)) } +async fn runtime_worker_completions( + State(api): State, + AxumPath((runtime_id, worker_id)): AxumPath<(String, String)>, + Json(request): Json, +) -> ApiResult> { + let result = api + .runtime + .worker_completions(&runtime_id, &worker_id, request) + .map_err(|err| err.into_error())?; + Ok(Json(result)) +} + async fn stop_runtime_worker( State(api): State, AxumPath((runtime_id, worker_id)): AxumPath<(String, String)>, @@ -7237,6 +7272,7 @@ mod tests { WorkerInputRequest { kind: WorkerInputKind::User, content: "persist me".to_string(), + segments: None, }, ) .expect("send input"); @@ -7298,6 +7334,7 @@ mod tests { WorkerInputRequest { kind: WorkerInputKind::User, content: "should not be routed to stale handle".to_string(), + segments: None, }, ) .expect("stale worker input is projected as an operation result"); diff --git a/resources/profiles/default.dcdl b/resources/profiles/default.dcdl index 4fb756e4..df896714 100644 --- a/resources/profiles/default.dcdl +++ b/resources/profiles/default.dcdl @@ -14,6 +14,13 @@ engine = { reasoning = "high"; }; +compaction = { + kind = "tokens"; + threshold = 240000; + request_threshold = 270000; + worker_context_max_tokens = 100000; +}; + feature = { task = { enabled = true; }; memory = { enabled = true; }; diff --git a/web/workspace/deno.json b/web/workspace/deno.json index 66a8f851..4142ec5f 100644 --- a/web/workspace/deno.json +++ b/web/workspace/deno.json @@ -6,7 +6,7 @@ "dev": "deno run -A npm:vite@7.2.7 dev", "dev:backend": "cd ../.. && cargo run -p yoi-workspace-server -- serve --workspace . --db .yoi/workspace.db --listen 127.0.0.1:8787", "check": "deno run -A npm:@sveltejs/kit@2.49.4 sync && deno run -A npm:svelte-check@4.3.4 --tsconfig ./tsconfig.json", - "test": "deno test --allow-read=src --allow-env=VSCODE_TEXTMATE_DEBUG src/lib/workspace-api/http.test.ts src/lib/workspace-console/chat-submit.test.ts src/lib/workspace-console/markdown.test.ts src/lib/workspace-console/model.test.ts src/lib/workspace-console/worker-console.ui.test.ts src/lib/workspace-settings/model.test.ts src/lib/workspace-sidebar/workers.test.ts src/lib/workspace-sidebar/worker-launch.test.ts src/lib/workspace-sidebar/repository-nav.test.ts", + "test": "deno test --allow-read=src --allow-env=VSCODE_TEXTMATE_DEBUG src/lib/workspace-api/http.test.ts src/lib/workspace-console/chat-submit.test.ts src/lib/workspace-console/composer-command.test.ts src/lib/workspace-console/composer-completion.test.ts src/lib/workspace-console/markdown.test.ts src/lib/workspace-console/model.test.ts src/lib/workspace-console/worker-console.ui.test.ts src/lib/workspace-settings/model.test.ts src/lib/workspace-sidebar/workers.test.ts src/lib/workspace-sidebar/worker-launch.test.ts src/lib/workspace-sidebar/repository-nav.test.ts", "build": "deno run -A npm:vite@7.2.7 build", "preview": "deno run -A npm:vite@7.2.7 preview" }, diff --git a/web/workspace/src/app.css b/web/workspace/src/app.css index 6119f8c2..1616ef6a 100644 --- a/web/workspace/src/app.css +++ b/web/workspace/src/app.css @@ -1383,6 +1383,15 @@ color: var(--text-muted); } +.composer-completions { + display: flex; + flex-wrap: wrap; + gap: var(--space-2); + align-items: center; + color: var(--text-muted); + font-size: 0.86rem; +} + @media (max-width: 960px) { .console-header { flex-direction: column; diff --git a/web/workspace/src/lib/workspace-console/chat-submit.test.ts b/web/workspace/src/lib/workspace-console/chat-submit.test.ts index 1b45d169..87aa7455 100644 --- a/web/workspace/src/lib/workspace-console/chat-submit.test.ts +++ b/web/workspace/src/lib/workspace-console/chat-submit.test.ts @@ -18,20 +18,30 @@ function assertEquals(actual: T, expected: T): void { } } -Deno.test("shouldSubmitChatKey defaults to Cmd+Enter submit behavior", () => { +Deno.test("shouldSubmitChatKey supports platform-auto submit modifier", () => { + assert( + shouldSubmitChatKey( + { key: "Enter", ctrlKey: true }, + { mode: "mod-enter", modKey: "auto", enabled: true }, + ), + "Ctrl+Enter should submit on non-Apple platforms", + ); + assertEquals( + shouldSubmitChatKey( + { key: "Enter" }, + { mode: "mod-enter", modKey: "auto", enabled: true }, + ), + false, + ); +}); + +Deno.test("shouldSubmitChatKey still supports explicit Cmd+Enter behavior", () => { assert( shouldSubmitChatKey( { key: "Enter", metaKey: true }, { mode: "mod-enter", modKey: "meta", enabled: true }, ), - "Cmd+Enter should submit by default", - ); - assertEquals( - shouldSubmitChatKey( - { key: "Enter" }, - { mode: "mod-enter", modKey: "meta", enabled: true }, - ), - false, + "Cmd+Enter should submit when meta is selected", ); assertEquals( shouldSubmitChatKey( diff --git a/web/workspace/src/lib/workspace-console/chat-submit.ts b/web/workspace/src/lib/workspace-console/chat-submit.ts index 070a58a0..88c4a1d9 100644 --- a/web/workspace/src/lib/workspace-console/chat-submit.ts +++ b/web/workspace/src/lib/workspace-console/chat-submit.ts @@ -26,7 +26,7 @@ export type ChatSubmitKeyEventLike = { function normalizeOptions(options: ChatSubmitOptions): NormalizedChatSubmitOptions { return { mode: "mod-enter", - modKey: "meta", + modKey: "auto", allowEmptySubmit: false, stopPropagation: false, enabled: true, diff --git a/web/workspace/src/lib/workspace-console/composer-command.test.ts b/web/workspace/src/lib/workspace-console/composer-command.test.ts new file mode 100644 index 00000000..d9ee89b9 --- /dev/null +++ b/web/workspace/src/lib/workspace-console/composer-command.test.ts @@ -0,0 +1,64 @@ +import { buildComposerRequest, parseSigilSegments } from "./composer-command.ts"; + +declare const Deno: { + test(name: string, fn: () => void): void; +}; + +function assert(condition: unknown, message: string): asserts condition { + if (!condition) { + throw new Error(message); + } +} + +function assertEquals(actual: T, expected: T): void { + const actualJson = JSON.stringify(actual); + const expectedJson = JSON.stringify(expected); + if (actualJson !== expectedJson) { + throw new Error(`Expected ${expectedJson}, got ${actualJson}`); + } +} + +Deno.test("parseSigilSegments converts TUI-style references", () => { + assertEquals(parseSigilSegments("read @src/main.rs then #memory and /workflow"), [ + { kind: "text", content: "read " }, + { kind: "file_ref", path: "src/main.rs" }, + { kind: "text", content: " then " }, + { kind: "knowledge_ref", slug: "memory" }, + { kind: "text", content: " and " }, + { kind: "workflow_invoke", slug: "workflow" }, + ]); +}); + +Deno.test("buildComposerRequest sends user segments when sigils are present", () => { + const result = buildComposerRequest("inspect @README.md"); + assert(result.ok, "request should be accepted"); + assertEquals(result.request?.kind, "user"); + assertEquals(result.request?.content, "inspect @README.md"); + assertEquals(result.request?.segments, [ + { kind: "text", content: "inspect " }, + { kind: "file_ref", path: "README.md" }, + ]); +}); + +Deno.test("buildComposerRequest parses colon commands", () => { + const compact = buildComposerRequest(":compact"); + assert(compact.ok, "compact should be accepted"); + assertEquals(compact.request, { kind: "compact", content: "" }); + + const peer = buildComposerRequest(":peer companion"); + assert(peer.ok, "peer should be accepted"); + assertEquals(peer.request, { kind: "register_peer", content: "companion" }); + + const help = buildComposerRequest(":help compact"); + assert(help.ok, "help should be accepted"); + assertEquals(help.request, undefined); + assert(help.notice?.includes(":compact"), "help should return a local notice"); +}); + +Deno.test("buildComposerRequest rejects invalid colon commands", () => { + const unknown = buildComposerRequest(":does-not-exist"); + assert(!unknown.ok, "unknown command should be rejected"); + + const invalidPeer = buildComposerRequest(":peer"); + assert(!invalidPeer.ok, "invalid peer command should be rejected"); +}); diff --git a/web/workspace/src/lib/workspace-console/composer-command.ts b/web/workspace/src/lib/workspace-console/composer-command.ts new file mode 100644 index 00000000..ffd0ec3b --- /dev/null +++ b/web/workspace/src/lib/workspace-console/composer-command.ts @@ -0,0 +1,208 @@ +import type { Segment } from "$lib/generated/protocol"; + +export type WorkerConsoleInputKind = + | "user" + | "system" + | "compact" + | "list_rewind_targets" + | "register_peer"; + +export type WorkerConsoleInputRequest = { + kind: WorkerConsoleInputKind; + content: string; + segments?: Segment[]; +}; + +export type ComposerCommandResult = + | { ok: true; request?: WorkerConsoleInputRequest; notice?: string } + | { ok: false; message: string }; + +type CommandSpec = { + usage: string; + description: string; +}; + +const COMMANDS: Record = { + help: { + usage: ":help [command]", + description: "Show available Web Console commands or details for one command.", + }, + "?": { + usage: ":? [command]", + description: "Alias for :help.", + }, + noop: { + usage: ":noop", + description: "Validate command dispatch without side effects.", + }, + compact: { + usage: ":compact", + description: "Request immediate Worker context compaction.", + }, + rewind: { + usage: ":rewind", + description: "Ask the Worker for rewind targets.", + }, + rollback: { + usage: ":rollback", + description: "Alias for :rewind.", + }, + peer: { + usage: ":peer ", + description: "Register another existing Worker as a reciprocal metadata peer.", + }, + system: { + usage: ":system ", + description: "Send an agent-visible system notification to the Worker.", + }, +}; + +export function buildComposerRequest(value: string): ComposerCommandResult { + const content = value.trim(); + if (!content) { + return { ok: false, message: "Input is empty." }; + } + if (content.startsWith(":")) { + return buildColonCommand(content.slice(1)); + } + const segments = parseSigilSegments(content); + return { + ok: true, + request: { + kind: "user", + content, + segments: segments.some((segment) => segment.kind !== "text") ? segments : undefined, + }, + }; +} + +function buildColonCommand(commandLine: string): ComposerCommandResult { + const [name = "", ...argv] = commandLine.trim().split(/\s+/).filter(Boolean); + if (!name) { + return { + ok: false, + message: "Empty command. Type :help for available commands.", + }; + } + switch (name) { + case "help": + case "?": + return helpCommand(argv); + case "noop": + if (argv.length > 0) { + return invalidUsage("noop"); + } + return { ok: true, notice: "noop: no action" }; + case "compact": + if (argv.length > 0) { + return invalidUsage("compact"); + } + return { ok: true, request: { kind: "compact", content: "" }, notice: "compact requested" }; + case "rewind": + case "rollback": + if (argv.length > 0) { + return invalidUsage("rewind"); + } + return { ok: true, request: { kind: "list_rewind_targets", content: "" }, notice: "rewind targets requested" }; + case "peer": + if (argv.length !== 1) { + return invalidUsage("peer"); + } + return { + ok: true, + request: { kind: "register_peer", content: argv[0] }, + notice: `peer metadata registration requested with \`${argv[0]}\``, + }; + case "system": { + const message = commandLine.trim().slice(name.length).trimStart(); + if (!message) { + return invalidUsage("system"); + } + return { ok: true, request: { kind: "system", content: message } }; + } + default: + return { + ok: false, + message: `Unknown command: ${name}. Type :help for available commands.`, + }; + } +} + +function helpCommand(argv: string[]): ComposerCommandResult { + if (argv.length > 1) { + return invalidUsage("help"); + } + const name = argv[0]; + if (name) { + const spec = COMMANDS[name]; + if (!spec) { + return { + ok: false, + message: `Unknown command: ${name}. Type :help for available commands.`, + }; + } + return { + ok: true, + notice: `command: ${name} — usage: ${spec.usage}. ${spec.description}`, + }; + } + const list = ["help", "noop", "compact", "rewind", "peer", "system"] + .map((command) => `${command} (${COMMANDS[command].usage})`) + .join(", "); + return { + ok: true, + notice: `available commands: ${list}`, + }; +} + +function invalidUsage(name: string): ComposerCommandResult { + return { ok: false, message: `Invalid arguments. Usage: ${COMMANDS[name].usage}` }; +} + +export function parseSigilSegments(input: string): Segment[] { + const segments: Segment[] = []; + const pattern = /(^|\s)([@#/])([^\s]+)/g; + let cursor = 0; + let match: RegExpExecArray | null; + while ((match = pattern.exec(input)) !== null) { + const leading = match[1] ?? ""; + const sigil = match[2]; + const value = match[3]; + const atomStart = match.index + leading.length; + if (atomStart > cursor) { + segments.push({ kind: "text", content: input.slice(cursor, atomStart) }); + } + segments.push(sigilSegment(sigil, value)); + cursor = atomStart + sigil.length + value.length; + } + if (cursor < input.length) { + segments.push({ kind: "text", content: input.slice(cursor) }); + } + return coalesceTextSegments(segments.length > 0 ? segments : [{ kind: "text", content: input }]); +} + +function sigilSegment(sigil: string, value: string): Segment { + switch (sigil) { + case "@": + return { kind: "file_ref", path: value }; + case "#": + return { kind: "knowledge_ref", slug: value }; + case "/": + return { kind: "workflow_invoke", slug: value }; + default: + return { kind: "text", content: `${sigil}${value}` }; + } +} + +function coalesceTextSegments(segments: Segment[]): Segment[] { + const coalesced: Segment[] = []; + for (const segment of segments) { + const last = coalesced.at(-1); + if (segment.kind === "text" && last?.kind === "text") { + last.content += segment.content; + } else { + coalesced.push(segment); + } + } + return coalesced; +} diff --git a/web/workspace/src/lib/workspace-console/composer-completion.test.ts b/web/workspace/src/lib/workspace-console/composer-completion.test.ts new file mode 100644 index 00000000..8249b133 --- /dev/null +++ b/web/workspace/src/lib/workspace-console/composer-completion.test.ts @@ -0,0 +1,54 @@ +import { + applyCompletion, + completionTokenAt, + localCommandCompletions, +} from "./composer-completion.ts"; + +declare const Deno: { + test(name: string, fn: () => void): void; +}; + +function assert(condition: unknown, message: string): asserts condition { + if (!condition) { + throw new Error(message); + } +} + +function assertEquals(actual: T, expected: T): void { + const actualJson = JSON.stringify(actual); + const expectedJson = JSON.stringify(expected); + if (actualJson !== expectedJson) { + throw new Error(`Expected ${expectedJson}, got ${actualJson}`); + } +} + +Deno.test("completionTokenAt detects TUI-style sigils before the cursor", () => { + assertEquals(completionTokenAt("open @src/ma", "open @src/ma".length), { + sigil: "@", + kind: "file", + start: 5, + end: 12, + prefix: "src/ma", + }); + assertEquals(completionTokenAt(":comp", 5)?.kind, "command"); + assertEquals(completionTokenAt("ask #mem", 8)?.kind, "knowledge"); + assertEquals(completionTokenAt("run /work", 9)?.kind, "workflow"); +}); + +Deno.test("applyCompletion replaces the active token and advances the cursor", () => { + const value = "open @src/ma please"; + const token = completionTokenAt(value, "open @src/ma".length); + assert(token, "token should exist"); + assertEquals(applyCompletion(value, token, { value: "src/main.rs" }), { + value: "open @src/main.rs please", + cursor: "open @src/main.rs ".length, + }); + assertEquals(applyCompletion(value, token, { value: "src", is_dir: true }), { + value: "open @src/ please", + cursor: "open @src/".length, + }); +}); + +Deno.test("localCommandCompletions filters colon commands", () => { + assertEquals(localCommandCompletions("com").map((entry) => entry.value), ["compact"]); +}); diff --git a/web/workspace/src/lib/workspace-console/composer-completion.ts b/web/workspace/src/lib/workspace-console/composer-completion.ts new file mode 100644 index 00000000..5a214b0c --- /dev/null +++ b/web/workspace/src/lib/workspace-console/composer-completion.ts @@ -0,0 +1,85 @@ +export type ComposerCompletionKind = "command" | "file" | "knowledge" | "workflow"; + +export type ComposerCompletionToken = { + sigil: ":" | "@" | "#" | "/"; + kind: ComposerCompletionKind; + start: number; + end: number; + prefix: string; +}; + +export type ComposerCompletionEntry = { + value: string; + is_dir?: boolean; + description?: string; +}; + +export type CompletionApplyResult = { + value: string; + cursor: number; +}; + +export const COLON_COMMAND_COMPLETIONS: ComposerCompletionEntry[] = [ + { value: "help", description: "Show commands" }, + { value: "noop", description: "No-op" }, + { value: "compact", description: "Compact Worker context" }, + { value: "rewind", description: "List rewind targets" }, + { value: "rollback", description: "Alias for rewind" }, + { value: "peer", description: "Register metadata peer" }, + { value: "system", description: "Send system notification" }, +]; + +export function completionTokenAt( + value: string, + cursor: number, +): ComposerCompletionToken | null { + const boundedCursor = Math.max(0, Math.min(cursor, value.length)); + const before = value.slice(0, boundedCursor); + const match = /(^|\s)([:@#/])([^\s]*)$/.exec(before); + if (!match) { + return null; + } + const sigil = match[2] as ComposerCompletionToken["sigil"]; + const prefix = match[3] ?? ""; + const tokenStart = before.length - prefix.length - sigil.length; + return { + sigil, + kind: completionKindForSigil(sigil), + start: tokenStart, + end: boundedCursor, + prefix, + }; +} + +export function localCommandCompletions(prefix: string): ComposerCompletionEntry[] { + const normalized = prefix.toLowerCase(); + return COLON_COMMAND_COMPLETIONS.filter((entry) => + entry.value.toLowerCase().startsWith(normalized) + ); +} + +export function applyCompletion( + value: string, + token: ComposerCompletionToken, + entry: ComposerCompletionEntry, +): CompletionApplyResult { + const suffix = entry.is_dir ? "/" : " "; + const replacement = `${token.sigil}${entry.value}${suffix}`; + const restStart = !entry.is_dir && value[token.end] === " " ? token.end + 1 : token.end; + const next = `${value.slice(0, token.start)}${replacement}${value.slice(restStart)}`; + const cursor = token.start + replacement.length; + return { value: next, cursor }; +} + +function completionKindForSigil(sigil: ComposerCompletionToken["sigil"]): ComposerCompletionKind { + switch (sigil) { + case ":": + return "command"; + case "@": + return "file"; + case "#": + return "knowledge"; + case "/": + return "workflow"; + } +} diff --git a/web/workspace/src/lib/workspace-console/model.test.ts b/web/workspace/src/lib/workspace-console/model.test.ts index 89cf674d..9e8d8da3 100644 --- a/web/workspace/src/lib/workspace-console/model.test.ts +++ b/web/workspace/src/lib/workspace-console/model.test.ts @@ -349,6 +349,118 @@ Deno.test("projectConsole keeps Grep error detail in the body", () => { ); }); +Deno.test("projectConsole renders alert events", () => { + const projection = projectConsole([ + { + eventId: "alert-1", + event: { + event: "alert", + data: { + level: "warn", + source: "compactor", + message: "manual compaction skipped", + timestamp_ms: 1, + }, + } satisfies Event, + }, + { + eventId: "alert-2", + event: { + event: "alert", + data: { + level: "error", + source: "engine", + message: "provider failed", + timestamp_ms: 2, + }, + } satisfies Event, + }, + ]); + + assertEquals(projection.lines.length, 2); + assertEquals(projection.lines[0].kind, "status"); + assertEquals(projection.lines[0].title, "Alert · compactor"); + assertEquals(projection.lines[0].body, "manual compaction skipped"); + assertEquals(projection.lines[0].error, false); + assertEquals(projection.lines[1].kind, "error"); + assertEquals(projection.lines[1].title, "Alert · engine"); + assertEquals(projection.lines[1].body, "provider failed"); + assertEquals(projection.lines[1].error, true); +}); + +Deno.test("projectConsole shows compact progress as a status block", () => { + const projection = projectConsole([ + { + eventId: "compact-1", + event: { event: "compact_start" } satisfies Event, + }, + ]); + + assertEquals(projection.lines.length, 1); + assertEquals(projection.lines[0].id, "status-compact"); + assertEquals(projection.lines[0].kind, "status"); + assertEquals(projection.lines[0].body, "Compacting…"); + assertEquals(projection.lines[0].streaming, true); + + const completed = projectConsole([ + { + eventId: "compact-1", + event: { event: "compact_start" } satisfies Event, + }, + { + eventId: "compact-2", + event: { + event: "compact_done", + data: { new_segment_id: "00000000-0000-0000-0000-000000000001" }, + } satisfies Event, + }, + ]); + + assertEquals(completed.lines.length, 1); + assertEquals(completed.lines[0].id, "status-compact"); + assertEquals(completed.lines[0].body, "Compacted."); + assertEquals(completed.lines[0].streaming, false); +}); + +Deno.test("createConsoleProjector updates only compact status block", () => { + const projector = createConsoleProjector(); + let projection = projector.append([ + { + eventId: "compact-identity-1", + event: { + event: "user_message", + data: { segments: [{ kind: "text", content: "hello" }] }, + } satisfies Event, + }, + { + eventId: "compact-identity-2", + event: { event: "compact_start" } satisfies Event, + }, + ]); + const userLine = projection.lines[0]; + const compactLine = projection.lines[1]; + + projection = projector.append([ + { + eventId: "compact-identity-3", + event: { + event: "compact_done", + data: { new_segment_id: "00000000-0000-0000-0000-000000000001" }, + } satisfies Event, + }, + ]); + + assert( + projection.lines[0] === userLine, + "unrelated message line should keep object identity", + ); + assert( + projection.lines[1] !== compactLine, + "compact status line should update object identity", + ); + assertEquals(projection.lines[1].body, "Compacted."); +}); + Deno.test("projectConsole keeps streaming tool call updates in the same Call block", () => { const projection = projectConsole([ { @@ -680,6 +792,18 @@ Deno.test("projectConsole renders snapshot entries and in-flight output", () => is_error: false, }, }, + { + kind: "extension", + ts: 6, + domain: "yoi.compaction", + payload: { + kind: "compaction_block", + schema_version: 1, + block_id: "compact", + state: "running", + message: "Compacting…", + }, + }, ], greeting: { worker_name: "Worker", @@ -710,6 +834,7 @@ Deno.test("projectConsole renders snapshot entries and in-flight output", () => "user:new user:false", "assistant:assistant reply:false", "tool:Read — 1 file read\n /tmp/a.md:false", + "status:Compacting…:true", "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 f7d6271c..1070671c 100644 --- a/web/workspace/src/lib/workspace-console/model.ts +++ b/web/workspace/src/lib/workspace-console/model.ts @@ -1,5 +1,6 @@ import type { Event as ProtocolEvent, + Alert, InFlightBlock, InFlightToolCallState, Segment, @@ -262,7 +263,10 @@ export function applyProtocolEvent( case "llm_retry": case "llm_continuation": case "run_end": + break; case "alert": + appendAlertLine(next, envelope.eventId, event.data); + break; case "memory_worker": case "segment_rotated": case "completions": @@ -271,22 +275,23 @@ export function applyProtocolEvent( case "workers_listed": case "worker_restored": case "peer_registered": - case "compact_start": - case "compact_done": // These are protocol/status/control events. TUI Console does not append // them to the conversation surface; browser Console should not either. break; + case "compact_start": + upsertStatusLine(next, "compact", envelope.eventId, "Compacting…", true); + break; + case "compact_done": + upsertStatusLine(next, "compact", envelope.eventId, "Compacted.", false); + break; case "compact_failed": - next.lines.push( - line( - envelope.eventId, - "error", - "compact failed", - event.data.error, - undefined, - false, - true, - ), + upsertStatusLine( + next, + "compact", + envelope.eventId, + `Compact failed: ${event.data.error}`, + false, + true, ); break; case "shutdown": @@ -341,6 +346,52 @@ function line( }; } +function upsertStatusLine( + projection: ConsoleProjection, + id: string, + eventId: string, + body: string, + streaming: boolean, + error = false, +): void { + const lineId = `status-${id}`; + const index = projection.lines.findIndex((item) => item.id === lineId); + const item: ConsoleLine = { + id: lineId, + kind: error ? "error" : "status", + title: error ? "Status error" : "Status", + body, + eventId, + source: "event", + streaming, + error, + }; + if (index >= 0) { + projection.lines[index] = item; + } else { + projection.lines.push(item); + } +} + +function appendAlertLine( + projection: ConsoleProjection, + eventId: string, + alert: Alert, +): void { + const isError = alert.level === "error"; + projection.lines.push( + line( + eventId, + isError ? "error" : "status", + `Alert · ${alert.source}`, + alert.message, + undefined, + false, + isError, + ), + ); +} + function findLastLineIndex( lines: ConsoleLine[], predicate: (line: ConsoleLine) => boolean, @@ -943,11 +994,55 @@ function applyLogEntry( case "tool_result": applyLoggedItem(projection, eventId, entry["item"]); break; + case "extension": + applyExtensionEntry(projection, eventId, entry); + break; default: break; } } +function applyExtensionEntry( + projection: ConsoleProjection, + eventId: string, + entry: Record, +) { + if (entry["domain"] !== "yoi.compaction") { + return; + } + const payload = entry["payload"]; + if (!isRecord(payload) || payload["kind"] !== "compaction_block") { + return; + } + const blockId = stringField(payload, "block_id") || "compact"; + const state = stringField(payload, "state") || "running"; + const message = stringField(payload, "message") || compactMessageForState(state, payload); + upsertStatusLine( + projection, + blockId, + eventId, + message, + state === "running", + state === "failed", + ); +} + +function compactMessageForState( + state: string, + payload: Record, +): string { + switch (state) { + case "done": + return "Compacted."; + case "failed": { + const error = stringField(payload, "error"); + return error ? `Compact failed: ${error}` : "Compact failed."; + } + default: + return "Compacting…"; + } +} + function applyLoggedItem( projection: ConsoleProjection, eventId: string, diff --git a/web/workspace/src/routes/w/[workspaceId]/runtimes/[runtimeId]/workers/[workerId]/console/+page.svelte b/web/workspace/src/routes/w/[workspaceId]/runtimes/[runtimeId]/workers/[workerId]/console/+page.svelte index 8459bc7b..7db7d851 100644 --- a/web/workspace/src/routes/w/[workspaceId]/runtimes/[runtimeId]/workers/[workerId]/console/+page.svelte +++ b/web/workspace/src/routes/w/[workspaceId]/runtimes/[runtimeId]/workers/[workerId]/console/+page.svelte @@ -2,6 +2,14 @@ import { tick } from 'svelte'; import ConsoleLineItem from '$lib/workspace-console/ConsoleLineItem.svelte'; import { chatSubmit } from '$lib/workspace-console/chat-submit'; + import { buildComposerRequest } from '$lib/workspace-console/composer-command'; + import { + applyCompletion, + completionTokenAt, + localCommandCompletions, + type ComposerCompletionEntry, + type ComposerCompletionToken + } from '$lib/workspace-console/composer-completion'; import { fitTextarea } from '$lib/workspace-console/textarea-fit'; import { createConsoleProjector, @@ -35,12 +43,24 @@ return workspaceApiPath(workspaceId, path); } + type WorkerCompletionsResult = { + kind: 'file' | 'knowledge' | 'workflow'; + prefix: string; + entries: ComposerCompletionEntry[]; + diagnostics: Diagnostic[]; + }; + let worker = $state(null); let liveWorkerState = $state(null); let workerError = $state(null); let draft = $state(''); + let completionEntries = $state([]); + let completionToken = $state(null); + let completionBusy = $state(false); + let completionError = $state(null); let sending = $state(false); let sendError = $state(null); + let composerNotice = $state(null); let streamState = $state<'connecting' | 'open' | 'closed' | 'error'>('connecting'); let streamDiagnostics = $state([]); let workerDetailsOpen = $state(false); @@ -200,9 +220,80 @@ return true; } + async function applyComposerCompletion(event: KeyboardEvent) { + const target = event.currentTarget; + if (!(target instanceof HTMLTextAreaElement)) { + return; + } + const token = completionTokenAt(draft, target.selectionStart ?? draft.length); + completionToken = token; + completionError = null; + if (!token) { + completionEntries = []; + return; + } + + completionBusy = true; + try { + const entries = await resolveCompletionEntries(token); + completionEntries = entries; + if (entries.length === 0) { + completionError = `No completions for ${token.sigil}${token.prefix}`; + return; + } + const applied = applyCompletion(draft, token, entries[0]); + draft = applied.value; + await tick(); + target.setSelectionRange(applied.cursor, applied.cursor); + composerNotice = entries.length > 1 + ? `Completed ${token.sigil}${entries[0].value}; ${entries.length - 1} more candidate(s)` + : null; + } catch (error) { + completionError = error instanceof Error ? error.message : String(error); + } finally { + completionBusy = false; + } + } + + async function resolveCompletionEntries( + token: ComposerCompletionToken + ): Promise { + if (token.kind === 'command') { + return localCommandCompletions(token.prefix); + } + const result = await postJson( + workerApiPath( + `/runtimes/${encodeURIComponent(runtimeId)}/workers/${encodeURIComponent(workerId)}/completions` + ), + { kind: token.kind, prefix: token.prefix } + ); + if (result.diagnostics.length > 0 && result.entries.length === 0) { + throw new Error(diagnosticsToText(result.diagnostics)); + } + return result.entries; + } + + function handleComposerKeydown(event: KeyboardEvent) { + if (event.key !== 'Tab') { + return; + } + event.preventDefault(); + void applyComposerCompletion(event); + } + async function submitDraft(value = draft) { - const content = value.trim(); - if (!content || sending || !inputReady) { + const command = buildComposerRequest(value); + if (!command.ok) { + composerNotice = null; + sendError = command.message; + return; + } + composerNotice = command.notice ?? null; + if (!command.request) { + draft = ''; + return; + } + if (sending || !inputReady) { return; } @@ -211,7 +302,7 @@ try { const result = await postJson( workerApiPath(`/runtimes/${encodeURIComponent(runtimeId)}/workers/${encodeURIComponent(workerId)}/input`), - { kind: 'user', content } + command.request ); if (result.state === 'accepted') { draft = ''; @@ -470,18 +561,34 @@ + {#if completionBusy || completionError || completionEntries.length > 0} +
+ {#if completionBusy} + completing… + {:else if completionError} + {completionError} + {:else} + Tab: {completionToken?.sigil}{completionEntries[0]?.value} + {#if completionEntries.length > 1} + {completionEntries.length - 1} more + {/if} + {/if} +
+ {/if}
- {#if sendError}

{sendError}

{/if} + {#if composerNotice}

{composerNotice}

{/if} + {#if sendError}

{sendError}

{/if}