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

feat: add web console commands

Browse Source
This commit is contained in:
Hare
2026-07-14 07:55:51 +09:00
parent 520c10d807
commit e298856c85
24 changed files with 1365 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/manifest/src/profile.rs
+26
View File
@@ -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]
crates/tui/src/app.rs
+57
View File
@@ -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::<uuid::Uuid>().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::<session_store::SystemItem>(value.clone()) else {
// Unknown / forward-compat shape: fall back to rendering the
crates/worker-runtime/src/execution.rs
+18
View File
@@ -398,6 +398,15 @@ pub trait WorkerExecutionBackend: Send + Sync + 'static {
fn worker_snapshot(&self, _handle: &WorkerExecutionHandle) -> Option<protocol::Event> {
None
}
fn worker_completions(
&self,
_handle: &WorkerExecutionHandle,
_kind: protocol::CompletionKind,
_prefix: &str,
) -> Vec<protocol::CompletionEntry> {
Vec::new()
}
}
#[derive(Clone)]
@@ -485,6 +494,15 @@ impl WorkerExecutionBackendRef {
) -> Option<protocol::Event> {
self.backend.worker_snapshot(handle)
}
pub(crate) fn worker_completions(
&self,
handle: &WorkerExecutionHandle,
kind: protocol::CompletionKind,
prefix: &str,
) -> Vec<protocol::CompletionEntry> {
self.backend.worker_completions(handle, kind, prefix)
}
}
impl fmt::Debug for WorkerExecutionBackendRef {
crates/worker-runtime/src/http_server.rs
+36
View File
@@ -146,6 +146,10 @@ pub fn runtime_http_router(runtime: Runtime, local_token: Option<String>) -> 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<protocol::CompletionEntry>,
}
/// 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<RuntimeHttpState>,
Path(worker_id): Path<String>,
body: Result<Json<RuntimeHttpWorkerCompletionsRequest>, JsonRejection>,
) -> RestResult<RuntimeHttpWorkerCompletionsResponse> {
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<RuntimeHttpState>,
Path(worker_id): Path<String>,
crates/worker-runtime/src/interaction.rs
+14
View File
@@ -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<Vec<Segment>>,
}
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,
}
}
}
crates/worker-runtime/src/runtime.rs
+46 -4
View File
@@ -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<Vec<protocol::CompletionEntry>, 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(),
}),
},
}
}
crates/worker-runtime/src/worker_backend.rs
+44 -19
View File
@@ -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<protocol::CompletionEntry> {
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)]
crates/worker/src/controller.rs
+38
View File
@@ -91,6 +91,44 @@ impl WorkerHandle {
(event, entry_rx)
}
pub fn completion_entries(
&self,
kind: protocol::CompletionKind,
prefix: &str,
) -> Vec<protocol::CompletionEntry> {
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<usize, broadcast::error::SendError<Event>> {
self.event_tx.send(event)
crates/worker/src/ipc/server.rs
+1 -31
View File
@@ -165,37 +165,7 @@ async fn handle_connection(stream: tokio::net::UnixStream, handle: WorkerHandle)
method = reader.next::<Method>() => {
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
crates/worker/src/worker.rs
+87 -15
View File
@@ -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<C: LlmClient, St: Store> Worker<C, St> {
.map_err(WorkerError::Engine)
}
fn persist_compaction_block(
&mut self,
state: &str,
message: &str,
error: Option<&str>,
new_segment_id: Option<SegmentId>,
) -> 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<C: LlmClient, St: Store> Worker<C, St> {
.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<C: LlmClient, St: Store> Worker<C, St> {
}
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<C: LlmClient, St: Store> Worker<C, St> {
}
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<C: LlmClient, St: Store> Worker<C, St> {
}
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<C: LlmClient, St: Store> Worker<C, St> {
}
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,
crates/workspace-server/src/hosts.rs
+165 -1
View File
@@ -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<Vec<protocol::Segment>>,
}
#[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<protocol::CompletionEntry>,
pub diagnostics: Vec<RuntimeDiagnostic>,
}
#[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<WorkerProxyConnectPoint> {
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<WorkerCompletionsResult, RuntimeRegistryError> {
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();
crates/workspace-server/src/server.rs
+41 -4
View File
@@ -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<WorkspaceApi>,
AxumPath(path): AxumPath<ScopedRuntimeWorkerPath>,
Json(request): Json<WorkerCompletionsRequest>,
) -> ApiResult<Json<WorkerCompletionsResult>> {
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<WorkspaceApi>,
AxumPath(path): AxumPath<ScopedRuntimeWorkerPath>,
@@ -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<WorkspaceApi>,
AxumPath((runtime_id, worker_id)): AxumPath<(String, String)>,
Json(request): Json<WorkerCompletionsRequest>,
) -> ApiResult<Json<WorkerCompletionsResult>> {
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<WorkspaceApi>,
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");