chore: merge develop into work/companion

# Conflicts:
#	crates/tui/src/app.rs
This commit is contained in:
2026-08-21 07:18:07 +09:00
61 changed files with 7110 additions and 523 deletions
+112 -47
View File
@@ -14,6 +14,8 @@ pub struct BackendRuntimeTarget {
/// Workspace Backend API root URL, for example `http://127.0.0.1:8787`.
/// This is intentionally the Backend endpoint, not a Runtime endpoint.
pub base_url: String,
/// Workspace identity used for every Worker lifecycle and protocol operation.
pub workspace_id: String,
/// Backend-owned Runtime identity used as path authority.
pub runtime_id: String,
/// Backend-owned Worker identity used as path authority.
@@ -23,11 +25,13 @@ pub struct BackendRuntimeTarget {
impl BackendRuntimeTarget {
pub fn new(
base_url: impl Into<String>,
workspace_id: impl Into<String>,
runtime_id: impl Into<String>,
worker_id: impl Into<String>,
) -> Self {
Self {
base_url: base_url.into(),
workspace_id: workspace_id.into(),
runtime_id: runtime_id.into(),
worker_id: worker_id.into(),
}
@@ -57,6 +61,36 @@ impl BackendRuntimeListTarget {
runtime_id,
}
}
pub fn select_workspace(&mut self, workspace_id: impl Into<String>) {
self.workspace_id = Some(workspace_id.into());
}
pub fn clear_workspace(&mut self) {
self.workspace_id = None;
}
pub fn workspace_id(&self) -> Option<&str> {
self.workspace_id.as_deref()
}
pub fn runtime_target(
&self,
runtime_id: impl Into<String>,
worker_id: impl Into<String>,
) -> Result<BackendRuntimeTarget, BackendRuntimeClientError> {
let workspace_id = self.workspace_id.clone().ok_or_else(|| {
BackendRuntimeClientError::InvalidTarget(
"workspace_id is required before selecting a Backend worker".to_string(),
)
})?;
Ok(BackendRuntimeTarget::new(
self.base_url.clone(),
workspace_id,
runtime_id,
worker_id,
))
}
}
#[derive(Debug, Clone, Deserialize)]
@@ -186,7 +220,13 @@ pub async fn list_backend_workers(
validate_list_target(target)?;
let http = reqwest::Client::new();
if let Some(runtime_id) = target.runtime_id.as_deref() {
let path = backend_runtime_workers_path(target.workspace_id.as_deref(), runtime_id);
let path = backend_runtime_workers_path(
target
.workspace_id
.as_deref()
.expect("validated Backend Workspace scope"),
runtime_id,
);
let url = join_base_and_path(&target.base_url, &path);
return Ok(http
.get(url)
@@ -197,7 +237,12 @@ pub async fn list_backend_workers(
.await?);
}
let runtime_path = backend_runtimes_path(target.workspace_id.as_deref());
let runtime_path = backend_runtimes_path(
target
.workspace_id
.as_deref()
.expect("validated Backend Workspace scope"),
);
let runtime_url = join_base_and_path(&target.base_url, &runtime_path);
let runtimes = http
.get(runtime_url)
@@ -210,8 +255,13 @@ pub async fn list_backend_workers(
let mut items = Vec::new();
let mut diagnostics = runtimes.diagnostics;
for runtime in runtimes.items {
let path =
backend_runtime_workers_path(target.workspace_id.as_deref(), &runtime.runtime_id);
let path = backend_runtime_workers_path(
target
.workspace_id
.as_deref()
.expect("validated Backend Workspace scope"),
&runtime.runtime_id,
);
let url = join_base_and_path(&target.base_url, &path);
match http
.get(url)
@@ -256,7 +306,13 @@ pub async fn list_backend_stopped_workers(
));
};
let http = reqwest::Client::new();
let path = backend_runtime_workers_path(target.workspace_id.as_deref(), runtime_id);
let path = backend_runtime_workers_path(
target
.workspace_id
.as_deref()
.expect("validated Backend Workspace scope"),
runtime_id,
);
let url = join_base_and_path(&target.base_url, &format!("{path}?status=stopped"));
Ok(http
.get(url)
@@ -272,7 +328,11 @@ pub async fn restore_backend_worker(
) -> Result<BackendWorkerRestoreResponse, BackendRuntimeClientError> {
validate_target(target)?;
let http = reqwest::Client::new();
let path = backend_runtime_worker_restore_path(None, &target.runtime_id, &target.worker_id);
let path = backend_runtime_worker_restore_path(
&target.workspace_id,
&target.runtime_id,
&target.worker_id,
);
let url = join_base_and_path(&target.base_url, &path);
Ok(http
.post(url)
@@ -440,6 +500,11 @@ fn validate_target(target: &BackendRuntimeTarget) -> Result<(), BackendRuntimeCl
"Backend API base URL must start with http:// or https://".to_string(),
));
}
if target.workspace_id.is_empty() {
return Err(BackendRuntimeClientError::InvalidTarget(
"workspace_id is required".to_string(),
));
}
if target.runtime_id.is_empty() {
return Err(BackendRuntimeClientError::InvalidTarget(
"runtime_id is required".to_string(),
@@ -466,10 +531,18 @@ fn validate_list_target(
"Backend API base URL must start with http:// or https://".to_string(),
));
}
if target.workspace_id.as_deref().is_some_and(str::is_empty) {
return Err(BackendRuntimeClientError::InvalidTarget(
"workspace_id must not be empty when provided".to_string(),
));
match target.workspace_id.as_deref() {
Some("") => {
return Err(BackendRuntimeClientError::InvalidTarget(
"workspace_id must not be empty".to_string(),
));
}
None => {
return Err(BackendRuntimeClientError::InvalidTarget(
"workspace selection is required before listing Backend workers".to_string(),
));
}
Some(_) => {}
}
if target.runtime_id.as_deref().is_some_and(str::is_empty) {
return Err(BackendRuntimeClientError::InvalidTarget(
@@ -479,47 +552,35 @@ fn validate_list_target(
Ok(())
}
fn backend_runtimes_path(workspace_id: Option<&str>) -> String {
match workspace_id {
Some(workspace_id) => format!("/api/w/{}/runtimes", path_segment_encode(workspace_id)),
None => "/api/runtimes".to_string(),
}
fn backend_runtimes_path(workspace_id: &str) -> String {
format!("/api/w/{}/runtimes", path_segment_encode(workspace_id))
}
fn backend_runtime_workers_path(workspace_id: Option<&str>, runtime_id: &str) -> String {
match workspace_id {
Some(workspace_id) => format!(
"/api/w/{}/runtimes/{}/workers",
path_segment_encode(workspace_id),
path_segment_encode(runtime_id)
),
None => format!("/api/runtimes/{}/workers", path_segment_encode(runtime_id)),
}
fn backend_runtime_workers_path(workspace_id: &str, runtime_id: &str) -> String {
format!(
"/api/w/{}/runtimes/{}/workers",
path_segment_encode(workspace_id),
path_segment_encode(runtime_id)
)
}
fn backend_runtime_worker_restore_path(
workspace_id: Option<&str>,
workspace_id: &str,
runtime_id: &str,
worker_id: &str,
) -> String {
match workspace_id {
Some(workspace_id) => format!(
"/api/w/{}/runtimes/{}/workers/{}/restore",
path_segment_encode(workspace_id),
path_segment_encode(runtime_id),
path_segment_encode(worker_id)
),
None => format!(
"/api/runtimes/{}/workers/{}/restore",
path_segment_encode(runtime_id),
path_segment_encode(worker_id)
),
}
format!(
"/api/w/{}/runtimes/{}/workers/{}/restore",
path_segment_encode(workspace_id),
path_segment_encode(runtime_id),
path_segment_encode(worker_id)
)
}
fn protocol_ws_url(target: &BackendRuntimeTarget) -> String {
let path = format!(
"/api/runtimes/{}/workers/{}/protocol/ws",
"/api/w/{}/runtimes/{}/workers/{}/protocol/ws",
path_segment_encode(&target.workspace_id),
path_segment_encode(&target.runtime_id),
path_segment_encode(&target.worker_id)
);
@@ -573,11 +634,15 @@ mod tests {
#[test]
fn protocol_url_uses_backend_runtime_worker_identity() {
let target =
BackendRuntimeTarget::new("http://127.0.0.1:8787/", "runtime/one", "worker one");
let target = BackendRuntimeTarget::new(
"http://127.0.0.1:8787/",
"workspace alpha",
"runtime/one",
"worker one",
);
assert_eq!(
protocol_ws_url(&target),
"ws://127.0.0.1:8787/api/runtimes/runtime%2Fone/workers/worker%20one/protocol/ws"
"ws://127.0.0.1:8787/api/w/workspace%20alpha/runtimes/runtime%2Fone/workers/worker%20one/protocol/ws"
);
}
@@ -622,8 +687,8 @@ mod tests {
}
#[test]
fn workers_path_can_be_workspace_scoped_for_status_queries() {
let path = backend_runtime_workers_path(Some("team main"), "runtime/one");
fn workers_path_requires_workspace_scope_for_status_queries() {
let path = backend_runtime_workers_path("team main", "runtime/one");
assert_eq!(
format!("{path}?status=stopped"),
"/api/w/team%20main/runtimes/runtime%2Fone/workers?status=stopped"
@@ -631,10 +696,10 @@ mod tests {
}
#[test]
fn restore_worker_path_uses_backend_runtime_worker_identity() {
fn restore_worker_path_requires_workspace_scope() {
assert_eq!(
backend_runtime_worker_restore_path(None, "runtime/one", "worker one"),
"/api/runtimes/runtime%2Fone/workers/worker%20one/restore"
backend_runtime_worker_restore_path("team main", "runtime/one", "worker one"),
"/api/w/team%20main/runtimes/runtime%2Fone/workers/worker%20one/restore"
);
}
}
+161
View File
@@ -0,0 +1,161 @@
use serde::{Deserialize, Serialize};
use std::fmt;
const DEFAULT_WORKSPACE_LIMIT: usize = 200;
#[derive(Debug, Clone, Deserialize, PartialEq, Eq)]
pub struct BackendWorkspace {
pub workspace_id: String,
pub owner_account_id: Option<String>,
pub display_name: String,
pub state: String,
pub created_at: String,
pub updated_at: String,
}
#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
#[serde(deny_unknown_fields)]
pub struct CreateBackendWorkspaceRequest {
pub operation_key: String,
pub display_name: String,
pub repository: CreateBackendWorkspaceRepository,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(deny_unknown_fields)]
pub struct CreateBackendWorkspaceRepository {
pub uri: String,
pub display_name: Option<String>,
pub default_ref: Option<String>,
}
#[derive(Debug, Clone, Deserialize, PartialEq, Eq)]
pub struct CreateBackendWorkspaceResponse {
pub workspace: BackendWorkspace,
pub repository: CreateBackendWorkspaceRepositoryRecord,
pub config_revision: u64,
pub request_fingerprint: String,
pub replayed: bool,
}
#[derive(Debug, Clone, Deserialize, PartialEq, Eq)]
pub struct CreateBackendWorkspaceRepositoryRecord {
pub workspace_id: String,
pub repository_id: String,
pub name: String,
pub kind: String,
pub uri: String,
pub default_ref: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct BackendWorkspaceCatalogTarget {
pub base_url: String,
}
impl BackendWorkspaceCatalogTarget {
pub fn new(base_url: impl Into<String>) -> Self {
Self {
base_url: base_url.into(),
}
}
}
#[derive(Debug)]
pub enum BackendWorkspaceClientError {
InvalidTarget(String),
RequestFailed { status: u16, message: String },
Http(reqwest::Error),
}
impl fmt::Display for BackendWorkspaceClientError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::InvalidTarget(message) => f.write_str(message),
Self::RequestFailed { status, message } => {
write!(f, "Backend request failed with HTTP {status}: {message}")
}
Self::Http(error) => write!(f, "{error}"),
}
}
}
impl std::error::Error for BackendWorkspaceClientError {}
impl From<reqwest::Error> for BackendWorkspaceClientError {
fn from(error: reqwest::Error) -> Self {
Self::Http(error)
}
}
pub async fn list_backend_workspaces(
target: &BackendWorkspaceCatalogTarget,
) -> Result<Vec<BackendWorkspace>, BackendWorkspaceClientError> {
validate_target(target)?;
let url = format!(
"{}/api/workspaces?limit={DEFAULT_WORKSPACE_LIMIT}",
target.base_url.trim_end_matches('/')
);
let response = reqwest::Client::new().get(url).send().await?;
let response = require_success(response).await?;
Ok(response.json::<Vec<BackendWorkspace>>().await?)
}
pub async fn create_backend_workspace(
target: &BackendWorkspaceCatalogTarget,
request: &CreateBackendWorkspaceRequest,
) -> Result<CreateBackendWorkspaceResponse, BackendWorkspaceClientError> {
validate_target(target)?;
let url = format!("{}/api/workspaces", target.base_url.trim_end_matches('/'));
let response = reqwest::Client::new()
.post(url)
.json(request)
.send()
.await?;
let response = require_success(response).await?;
Ok(response.json::<CreateBackendWorkspaceResponse>().await?)
}
async fn require_success(
response: reqwest::Response,
) -> Result<reqwest::Response, BackendWorkspaceClientError> {
if response.status().is_success() {
return Ok(response);
}
let status = response.status().as_u16();
let message = response.text().await.unwrap_or_default();
Err(BackendWorkspaceClientError::RequestFailed { status, message })
}
fn validate_target(
target: &BackendWorkspaceCatalogTarget,
) -> Result<(), BackendWorkspaceClientError> {
if !(target.base_url.starts_with("http://") || target.base_url.starts_with("https://")) {
return Err(BackendWorkspaceClientError::InvalidTarget(
"Backend API base URL must start with http:// or https://".to_string(),
));
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn create_request_keeps_operation_key_for_exact_retry() {
let request = CreateBackendWorkspaceRequest {
operation_key: "workspace-create-1".to_string(),
display_name: "Alpha".to_string(),
repository: CreateBackendWorkspaceRepository {
uri: "/srv/repos/alpha".to_string(),
display_name: Some("Main".to_string()),
default_ref: Some("develop".to_string()),
},
};
let retry = request.clone();
assert_eq!(retry.operation_key, "workspace-create-1");
assert_eq!(retry, request);
}
}
+6
View File
@@ -10,6 +10,7 @@
pub mod backend_auth;
pub mod backend_runtime;
pub mod backend_workspace;
pub mod runtime_command;
pub mod spawn;
pub mod target;
@@ -28,6 +29,11 @@ pub use backend_runtime::{
BackendWorkerWorkspaceSummary, BackendWorkingDirectorySummary, list_backend_stopped_workers,
list_backend_workers, restore_backend_worker,
};
pub use backend_workspace::{
BackendWorkspace, BackendWorkspaceCatalogTarget, BackendWorkspaceClientError,
CreateBackendWorkspaceRepository, CreateBackendWorkspaceRequest,
CreateBackendWorkspaceResponse, create_backend_workspace, list_backend_workspaces,
};
pub use runtime_command::WorkerRuntimeCommand;
pub use target::{
BackendTarget, Dashboard, LocalTarget, Target, TargetError, TargetKind, WorkerByName,
+30
View File
@@ -132,6 +132,12 @@ impl TargetError {
}
}
fn invalid(target: TargetKind, message: impl Into<String>) -> Self {
Self {
message: format!("invalid {target} target: {}", message.into()),
}
}
fn local_runtime_command(error: std::io::Error) -> Self {
Self {
message: format!("failed to resolve local Worker runtime command: {error}"),
@@ -260,9 +266,16 @@ impl Target for BackendTarget {
&self,
selector: WorkerConnectionSelector,
) -> Result<WorkerConnection, TargetError> {
let workspace_id = self.workspace_id.clone().ok_or_else(|| {
TargetError::invalid(
self.kind(),
"workspace selection is required before connecting to a Backend Worker",
)
})?;
Ok(WorkerConnection {
target: BackendRuntimeTarget::new(
self.base_url.clone(),
workspace_id,
selector.runtime_id,
selector.worker_id,
),
@@ -313,10 +326,27 @@ mod tests {
.unwrap();
assert_eq!(connection.target.base_url, "http://127.0.0.1:8787");
assert_eq!(connection.target.workspace_id, "workspace-a");
assert_eq!(connection.target.runtime_id, "runtime-a");
assert_eq!(connection.target.worker_id, "worker-b");
}
#[test]
fn backend_target_rejects_worker_connection_before_workspace_selection() {
let target = BackendTarget::new("http://127.0.0.1:8787", None::<String>);
let error =
match target.connect_worker(WorkerConnectionSelector::new("runtime-a", "worker-b")) {
Ok(_) => panic!("unscoped connection must fail"),
Err(error) => error,
};
assert!(
error
.to_string()
.contains("workspace selection is required")
);
}
#[test]
fn backend_target_rejects_local_worker_operations() {
let target = BackendTarget::new("http://127.0.0.1:8787", None::<String>);
+36
View File
@@ -0,0 +1,36 @@
use client::{
BackendTarget, CreateBackendWorkspaceRepository, CreateBackendWorkspaceRequest, Target,
WorkerConnectionSelector,
};
#[test]
fn workspace_creation_request_preserves_operation_key_for_retry() {
let request = CreateBackendWorkspaceRequest {
operation_key: "workspace-create-1".to_string(),
display_name: "Alpha".to_string(),
repository: CreateBackendWorkspaceRepository {
uri: "/srv/repos/alpha".to_string(),
display_name: Some("Main".to_string()),
default_ref: Some("develop".to_string()),
},
};
assert_eq!(request.clone(), request);
assert_eq!(request.operation_key, "workspace-create-1");
}
#[test]
fn backend_worker_connection_requires_explicit_workspace_scope() {
let target = BackendTarget::new("http://127.0.0.1:8787", None::<String>);
let error = match target.connect_worker(WorkerConnectionSelector::new("runtime-a", "worker-a"))
{
Ok(_) => panic!("unscoped Backend worker connection must fail"),
Err(error) => error,
};
assert!(
error
.to_string()
.contains("workspace selection is required")
);
}
+162 -3
View File
@@ -530,6 +530,15 @@ pub enum Event {
revision: u64,
event: Box<Event>,
},
/// Terminal removal fence for one parent-owned Internal Worker session.
///
/// Clients discard the matching child and descendants, then ignore later
/// nested events for this identity until an authoritative snapshot replaces
/// the projection.
InternalWorkerRemoved {
worker: InternalWorkerRef,
revision: u64,
},
/// Server-side segment log rotated to a fresh `SegmentStart`.
///
/// Fires on compaction and on auto-fork when the store head drifts
@@ -547,6 +556,12 @@ pub enum Event {
Status {
status: WorkerStatus,
},
/// Bounded, provider-owned command telemetry for the live Console. This is
/// intentionally not a history entry and is reconstructed from
/// `Snapshot.in_flight.commands` after reconnect.
Command {
event: CommandEvent,
},
/// Reply to `Method::ListCompletions`. Delivered only to the
/// requesting socket (not broadcast). `entries` is empty when no
/// candidates match or when the requested kind has no resolver
@@ -714,8 +729,79 @@ pub struct RewindSummary {
pub tool_side_effect_warning: bool,
}
/// Unfinished model output included in `Event::Snapshot` for clients that
/// attach while an LLM response is still streaming.
/// Live provider-owned command status. These values are operational Console
/// state only and are never appended to Worker history.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(rename_all = "snake_case")]
pub enum CommandStatus {
Running,
Completed,
Failed,
TimedOut,
Cancelled,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(rename_all = "snake_case")]
pub enum CommandStream {
Stdout,
Stderr,
}
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
pub struct CommandStreamSlice {
pub start_offset: u64,
pub end_offset: u64,
pub content: String,
pub truncated: bool,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
pub struct CommandSnapshot {
pub command_id: String,
pub tool_call_id: Option<String>,
pub status: CommandStatus,
pub started_at_ms: u64,
pub observed_at_ms: u64,
pub last_output_at_ms: Option<u64>,
pub stdout: CommandStreamSlice,
pub stderr: CommandStreamSlice,
pub exit_code: Option<i32>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum CommandEvent {
Started {
command_id: String,
tool_call_id: Option<String>,
observed_at_ms: u64,
},
Output {
command_id: String,
stream: CommandStream,
start_offset: u64,
end_offset: u64,
content: String,
observed_at_ms: u64,
},
Terminal {
command_id: String,
status: CommandStatus,
exit_code: Option<i32>,
stdout_end_offset: u64,
stderr_end_offset: u64,
observed_at_ms: u64,
},
}
/// Unfinished model output and active command state included in
/// `Event::Snapshot` for clients that attach while work is still streaming.
///
/// These blocks are presentation state only: they are reconstructed from the
/// active Worker controller and must not be treated as committed assistant
@@ -726,11 +812,13 @@ pub struct RewindSummary {
pub struct InFlightSnapshot {
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub blocks: Vec<InFlightBlock>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub commands: Vec<CommandSnapshot>,
}
impl InFlightSnapshot {
pub fn is_empty(&self) -> bool {
self.blocks.is_empty()
self.blocks.is_empty() && self.commands.is_empty()
}
}
@@ -1375,6 +1463,22 @@ mod tests {
state: InFlightToolCallState::StreamingArgs,
},
],
commands: vec![CommandSnapshot {
command_id: "command-1".into(),
tool_call_id: Some("call_1".into()),
status: CommandStatus::Running,
started_at_ms: 100,
observed_at_ms: 120,
last_output_at_ms: Some(120),
stdout: CommandStreamSlice {
start_offset: 4,
end_offset: 8,
content: "tail".into(),
truncated: true,
},
stderr: CommandStreamSlice::default(),
exit_code: None,
}],
},
internal_workers: Vec::new(),
};
@@ -1444,6 +1548,41 @@ mod tests {
));
}
#[test]
fn event_command_output_roundtrip_preserves_stream_and_offsets() {
let event = Event::Command {
event: CommandEvent::Output {
command_id: "command-1".into(),
stream: CommandStream::Stderr,
start_offset: 8,
end_offset: 12,
content: "warn".into(),
observed_at_ms: 42,
},
};
let json = serde_json::to_string(&event).unwrap();
let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
assert_eq!(parsed["event"], "command");
assert_eq!(parsed["data"]["event"]["kind"], "output");
assert_eq!(parsed["data"]["event"]["stream"], "stderr");
assert_eq!(parsed["data"]["event"]["start_offset"], 8);
assert_eq!(parsed["data"]["event"]["end_offset"], 12);
assert_eq!(parsed["data"]["event"]["observed_at_ms"], 42);
assert!(matches!(
serde_json::from_str::<Event>(&json).unwrap(),
Event::Command {
event: CommandEvent::Output {
command_id,
stream: CommandStream::Stderr,
start_offset: 8,
end_offset: 12,
content,
observed_at_ms: 42,
}
} if command_id == "command-1" && content == "warn"
));
}
#[test]
fn event_snapshot_legacy_without_status_defaults_to_idle() {
let json = r#"{"event":"snapshot","data":{"entries":[],"greeting":{"worker_name":"test","cwd":"/tmp","provider":"anthropic","model":"claude","scope_summary":"","tools":[]}}}"#;
@@ -1802,6 +1941,26 @@ mod tests {
}
}
#[test]
fn internal_worker_removal_roundtrip_preserves_terminal_fence() {
let event = Event::InternalWorkerRemoved {
worker: InternalWorkerRef {
session_id: "session-1".into(),
name: "research".into(),
parent_session_id: Some("parent-session".into()),
kind: InternalWorkerKind::SubWorker,
},
revision: 8,
};
let json = serde_json::to_string(&event).unwrap();
let decoded: Event = serde_json::from_str(&json).unwrap();
assert!(matches!(
decoded,
Event::InternalWorkerRemoved { worker, revision }
if worker.session_id == "session-1" && revision == 8
));
}
#[test]
fn legacy_snapshot_defaults_internal_workers_to_empty() {
let snapshot: Event = serde_json::from_value(serde_json::json!({
+8 -2
View File
@@ -3,8 +3,9 @@ use std::path::PathBuf;
use ts_rs::{Config, TS};
use crate::{
Alert, AlertLevel, AlertSource, CompletionEntry, CompletionKind, ErrorCode, Event, Greeting,
InFlightBlock, InFlightSnapshot, InFlightToolCallState, InternalWorkerKind, InternalWorkerRef,
Alert, AlertLevel, AlertSource, CommandEvent, CommandSnapshot, CommandStatus, CommandStream,
CommandStreamSlice, CompletionEntry, CompletionKind, ErrorCode, Event, Greeting, InFlightBlock,
InFlightSnapshot, InFlightToolCallState, InternalWorkerKind, InternalWorkerRef,
InternalWorkerSnapshot, InvokeKind, MemoryWorkerEvent, Method, Permission, RewindSummary,
RewindTarget, RewindTargetId, RunResult, ScopeRule, Segment, TurnResult, WorkerEvent,
WorkerStatus,
@@ -47,6 +48,11 @@ pub fn generated_protocol_types() -> String {
push_decl::<ErrorCode>(&cfg, &mut output);
push_decl::<Permission>(&cfg, &mut output);
push_decl::<InFlightToolCallState>(&cfg, &mut output);
push_decl::<CommandStatus>(&cfg, &mut output);
push_decl::<CommandStream>(&cfg, &mut output);
push_decl::<CommandStreamSlice>(&cfg, &mut output);
push_decl::<CommandSnapshot>(&cfg, &mut output);
push_decl::<CommandEvent>(&cfg, &mut output);
push_decl::<ScopeRule>(&cfg, &mut output);
push_decl::<CompletionEntry>(&cfg, &mut output);
push_decl::<RewindTargetId>(&cfg, &mut output);
+4 -6
View File
@@ -848,18 +848,16 @@ fn collect_foreign_key_diagnostics(
)
})
.collect::<BTreeSet<_>>();
// The Ticket component owns its required foreign keys, while an integrated host may
// strengthen Workspace/domain boundaries with additional references to host-owned
// tables. Reject missing component constraints, but do not treat those host extensions
// as Ticket schema drift.
for missing in expected.difference(&actual) {
push_diagnostic(
diagnostics,
format!("table {table:?} is missing foreign key {missing:?}"),
);
}
for unexpected in actual.difference(&expected) {
push_diagnostic(
diagnostics,
format!("table {table:?} has unexpected foreign key {unexpected:?}"),
);
}
}
fn collect_foreign_key_check_diagnostics(
+2 -1
View File
@@ -43,7 +43,7 @@ impl Tool for BashTool {
async fn execute(
&self,
input_json: &str,
_ctx: llm_engine::tool::ToolExecutionContext,
ctx: llm_engine::tool::ToolExecutionContext,
) -> Result<ToolOutput, ToolError> {
let params: BashParams = serde_json::from_str(input_json)
.map_err(|error| ToolError::InvalidArgument(format!("invalid Bash input: {error}")))?;
@@ -58,6 +58,7 @@ impl Tool for BashTool {
command: params.command,
timeout_secs,
output_limit: INLINE_BYTE_BUDGET,
tool_call_id: Some(ctx.call_id),
})
.await
.map_err(crate::ToolsError::from)?;
+10 -3
View File
@@ -72,13 +72,20 @@ impl Tool for EditTool {
})
.await
.map_err(ToolsError::from)?;
self.tracker.record_workdir_hash(&path, result.content_hash);
let replacements = result.replacements;
self.tracker.record_workdir_edit(
&path,
result.content_hash,
replacements,
params.new_string.lines().count(),
params.old_string.lines().count(),
);
let summary = format!(
"Edited {} ({} replacement{})",
path,
result.replacements,
if result.replacements == 1 { "" } else { "s" }
replacements,
if replacements == 1 { "" } else { "s" }
);
let preview = make_preview(&params.new_string, &params.new_string);
+1 -1
View File
@@ -27,7 +27,7 @@ pub use error::ToolsError;
pub use glob::glob_tool;
pub use grep::grep_tool;
pub use read::read_tool;
pub use tracker::Tracker;
pub use tracker::{ChangeStat, Tracker};
pub use view_image::view_image_tool;
pub use web::{web_fetch_tool, web_search_tool};
pub use write::write_tool;
+90 -2
View File
@@ -119,12 +119,22 @@ fn normalize_path_lexically(path: &Path) -> PathBuf {
normalized
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct ChangeStat {
pub added: u64,
pub deleted: u64,
}
#[derive(Debug, Default)]
struct Inner {
/// Hash of each file's last observed contents, keyed by canonical path.
hashes: HashMap<PathBuf, ContentHash>,
/// Line count paired with observations that included the file content.
line_counts: HashMap<PathBuf, usize>,
/// LRU list of touched files. Front = most recently touched.
recency: VecDeque<PathBuf>,
/// Successful Write/Edit mutations attributed to this session's tools.
change_stat: ChangeStat,
}
/// Canonical-path keyed tracker of file observations and their recency.
@@ -187,8 +197,27 @@ impl Tracker {
}
}
pub fn record_workdir_content(&self, path: &workdir::WorkdirPath, bytes: &[u8]) {
self.record_workdir_hash(path, hash_bytes(bytes));
pub fn record_workdir_content(&self, path: &workdir::WorkdirPath, content: &[u8]) {
let key = PathBuf::from(path.as_str());
let hash = hash_bytes(content);
let line_count = String::from_utf8_lossy(content).lines().count();
let mut inner = self.inner.lock().unwrap_or_else(|e| e.into_inner());
inner.line_counts.insert(key.clone(), line_count);
inner.hashes.insert(key.clone(), hash);
inner.recency.retain(|candidate| candidate != &key);
inner.recency.push_front(key);
if inner.recency.len() > RECENCY_CAPACITY {
inner.recency.pop_back();
}
}
pub fn observed_workdir_line_count(&self, path: &workdir::WorkdirPath) -> Option<usize> {
self.inner
.lock()
.unwrap_or_else(|e| e.into_inner())
.line_counts
.get(Path::new(path.as_str()))
.copied()
}
pub fn record_workdir_hash(&self, path: &workdir::WorkdirPath, hash: workdir::ContentHash) {
@@ -202,6 +231,50 @@ impl Tracker {
}
}
/// Record a successful, session-attributable source mutation.
///
/// Callers supply line counts derived from the exact replacement accepted
/// by a Write/Edit tool. Bash and external process mutations are excluded
/// because this tracker cannot attribute them to one tool operation
/// authoritatively.
pub fn record_change(&self, added: usize, deleted: usize) {
let mut inner = self.inner.lock().unwrap_or_else(|e| e.into_inner());
inner.change_stat.added = inner.change_stat.added.saturating_add(added as u64);
inner.change_stat.deleted = inner.change_stat.deleted.saturating_add(deleted as u64);
}
pub fn record_workdir_edit(
&self,
path: &workdir::WorkdirPath,
hash: workdir::ContentHash,
replacements: usize,
added_lines_per_replacement: usize,
deleted_lines_per_replacement: usize,
) {
let added = added_lines_per_replacement.saturating_mul(replacements);
let deleted = deleted_lines_per_replacement.saturating_mul(replacements);
let key = PathBuf::from(path.as_str());
let mut inner = self.inner.lock().unwrap_or_else(|e| e.into_inner());
inner.change_stat.added = inner.change_stat.added.saturating_add(added as u64);
inner.change_stat.deleted = inner.change_stat.deleted.saturating_add(deleted as u64);
if let Some(line_count) = inner.line_counts.get_mut(&key) {
*line_count = line_count.saturating_sub(deleted).saturating_add(added);
}
inner.hashes.insert(key.clone(), hash);
inner.recency.retain(|candidate| candidate != &key);
inner.recency.push_front(key);
if inner.recency.len() > RECENCY_CAPACITY {
inner.recency.pop_back();
}
}
pub fn change_stat(&self) -> ChangeStat {
self.inner
.lock()
.unwrap_or_else(|e| e.into_inner())
.change_stat
}
pub fn expected_workdir_hash(
&self,
path: &workdir::WorkdirPath,
@@ -458,6 +531,21 @@ mod tests {
}
}
#[test]
fn change_stat_saturates_and_accumulates_tracked_mutations() {
let tracker = Tracker::new();
tracker.record_change(7, 3);
tracker.record_change(5, 2);
assert_eq!(
tracker.change_stat(),
ChangeStat {
added: 12,
deleted: 5,
}
);
}
#[tokio::test]
async fn mutation_guard_blocks_equivalent_paths_until_drop() {
let dir = tempfile::tempdir().unwrap();
+3
View File
@@ -50,6 +50,7 @@ impl Tool for WriteTool {
Err(error) => return Err(ToolsError::from(error).into()),
};
let old_line_count = self.tracker.observed_workdir_line_count(&path).unwrap_or(0);
let outcome = self
.session
.write(WriteRequest {
@@ -60,6 +61,8 @@ impl Tool for WriteTool {
.await
.map_err(ToolsError::from)?;
self.tracker
.record_change(params.content.lines().count(), old_line_count);
self.tracker
.record_workdir_content(&path, params.content.as_bytes());
+125 -1
View File
@@ -1,4 +1,4 @@
use std::collections::VecDeque;
use std::collections::{HashMap, VecDeque};
use std::path::Path;
use std::time::{Duration, Instant};
@@ -292,6 +292,8 @@ pub struct App {
/// Selected Internal Worker transcript/task view. `None` is the parent (`main`)
/// view; the stable session identity survives projection reordering.
selected_internal_worker_session_id: Option<String>,
/// Terminal child-session fences, reset only by an authoritative snapshot.
removed_internal_workers: HashMap<String, u64>,
pub scroll: Scroll,
pub mode: Mode,
pub cache: FileCache,
@@ -371,6 +373,7 @@ impl App {
run_error_messages: Vec::new(),
internal_workers: Vec::new(),
selected_internal_worker_session_id: None,
removed_internal_workers: HashMap::new(),
scroll: Scroll::default(),
mode: Mode::Normal,
cache: FileCache::new(),
@@ -1410,10 +1413,16 @@ impl App {
revision,
event,
} => self.apply_internal_worker_event(worker, revision, *event),
Event::InternalWorkerRemoved { worker, revision } => {
self.remove_internal_worker(worker, revision)
}
Event::Status { status } => {
self.rewind_refresh_fence = false;
self.set_worker_status(status);
}
// Command telemetry is an operational Web Console surface. The
// TUI continues to render the final Bash ToolResult from history.
Event::Command { .. } => {}
Event::Completions { kind, entries } => {
// Apply only if the popup is still on the same
// (kind, prefix) the request was issued for; an
@@ -2112,6 +2121,7 @@ impl App {
if self.selected_internal_worker_index().is_none() {
self.selected_internal_worker_session_id = None;
}
self.removed_internal_workers.clear();
}
fn update_internal_worker_view_from_snapshot(
@@ -2175,6 +2185,12 @@ impl App {
revision: u64,
event: Event,
) {
if self
.removed_internal_workers
.contains_key(&worker.session_id)
{
return;
}
let index = self
.internal_workers
.iter()
@@ -2199,6 +2215,32 @@ impl App {
let _ = target.app.handle_worker_event(event);
}
fn remove_internal_worker(&mut self, worker: InternalWorkerRef, revision: u64) {
let session_id = worker.session_id;
let Some(index) = self
.internal_workers
.iter()
.position(|candidate| candidate.worker.session_id == session_id)
else {
if self.selected_internal_worker_session_id.as_deref() == Some(session_id.as_str()) {
self.selected_internal_worker_session_id = None;
}
self.removed_internal_workers
.entry(session_id)
.and_modify(|current| *current = (*current).max(revision))
.or_insert(revision);
return;
};
if revision <= self.internal_workers[index].revision {
return;
}
self.internal_workers.remove(index);
if self.selected_internal_worker_session_id.as_deref() == Some(session_id.as_str()) {
self.selected_internal_worker_session_id = None;
}
self.removed_internal_workers.insert(session_id, revision);
}
fn restore_snapshot(
&mut self,
entries: &[serde_json::Value],
@@ -3622,6 +3664,7 @@ mod completion_flow_tests {
finished: false,
},
],
commands: Vec::new(),
},
internal_workers: Vec::new(),
});
@@ -3876,6 +3919,87 @@ mod completion_flow_tests {
assert_eq!(app.task_pane_scroll, 3);
}
#[test]
fn terminal_internal_worker_removal_drops_descendants_and_fences_late_events() {
let mut app = App::new("parent".into());
let worker = InternalWorkerRef {
session_id: "child-session".into(),
name: "child".into(),
parent_session_id: Some("parent-session".into()),
kind: protocol::InternalWorkerKind::SubWorker,
};
let nested = InternalWorkerRef {
session_id: "grandchild-session".into(),
name: "grandchild".into(),
parent_session_id: Some("child-session".into()),
kind: protocol::InternalWorkerKind::SubWorker,
};
app.handle_worker_event(Event::InternalWorker {
worker: worker.clone(),
revision: 2,
event: Box::new(Event::InternalWorker {
worker: nested,
revision: 1,
event: Box::new(Event::TextDone {
text: "nested".into(),
}),
}),
});
assert_eq!(app.internal_workers.len(), 1);
assert_eq!(app.internal_workers[0].app.internal_workers.len(), 1);
app.cycle_worker_view();
assert_eq!(app.selected_worker_view().worker_name, "child");
app.handle_worker_event(Event::InternalWorkerRemoved {
worker: worker.clone(),
revision: 3,
});
app.handle_worker_event(Event::InternalWorker {
worker,
revision: 4,
event: Box::new(Event::TextDone {
text: "late".into(),
}),
});
assert!(app.internal_workers.is_empty());
assert_eq!(app.selected_worker_view().worker_name, "parent");
app.handle_worker_event(Event::Snapshot {
greeting: test_greeting(),
entries: Vec::new(),
status: WorkerStatus::Idle,
in_flight: Default::default(),
internal_workers: Vec::new(),
});
assert!(app.internal_workers.is_empty());
assert!(app.removed_internal_workers.is_empty());
}
#[test]
fn stale_internal_worker_removal_keeps_newer_projection() {
let mut app = App::new("parent".into());
let worker = InternalWorkerRef {
session_id: "child-session".into(),
name: "child".into(),
parent_session_id: Some("parent-session".into()),
kind: protocol::InternalWorkerKind::SubWorker,
};
app.handle_worker_event(Event::InternalWorker {
worker: worker.clone(),
revision: 4,
event: Box::new(Event::TextDone {
text: "current".into(),
}),
});
app.handle_worker_event(Event::InternalWorkerRemoved {
worker,
revision: 3,
});
assert_eq!(app.internal_workers.len(), 1);
assert_eq!(app.internal_workers[0].revision, 4);
}
#[test]
fn snapshot_authoritatively_replaces_internal_worker_views() {
let mut app = App::new("parent".into());
+96 -64
View File
@@ -3,8 +3,8 @@ use std::io;
use std::time::Duration;
use client::{
BackendRuntimeListTarget, BackendRuntimeTarget, BackendWorkerSummary,
list_backend_stopped_workers, list_backend_workers, restore_backend_worker,
BackendRuntimeListTarget, BackendWorkerSummary, list_backend_stopped_workers,
list_backend_workers, restore_backend_worker,
};
use crossterm::event::{self, Event as TermEvent, KeyCode, KeyEventKind, KeyModifiers};
use ratatui::backend::CrosstermBackend;
@@ -14,77 +14,94 @@ use ratatui::text::{Line, Span};
use ratatui::widgets::Paragraph;
use ratatui::{Frame, Terminal, TerminalOptions, Viewport};
use crate::backend_workspace_picker::select_backend_workspace;
use crate::console;
const MAX_ROWS: usize = 10;
const VIEWPORT_LINES: u16 = MAX_ROWS as u16 + 4;
pub(crate) async fn run(
target: BackendRuntimeListTarget,
mut target: BackendRuntimeListTarget,
include_stopped: bool,
) -> Result<(), Box<dyn Error>> {
let mut response = list_backend_workers(&target).await.map_err(|error| {
io::Error::other(format!(
"failed to list Backend runtime workers from {}: {error}",
target.base_url
))
})?;
if include_stopped {
match list_backend_stopped_workers(&target).await {
Ok(stopped) => {
response.items.extend(stopped.items);
response.diagnostics.extend(stopped.diagnostics);
loop {
if target.workspace_id().is_none() {
let workspace_id = select_backend_workspace(&target.base_url)
.await
.map_err(|error| io::Error::other(error.to_string()))?
.ok_or_else(|| io::Error::other("Backend workspace picker cancelled"))?;
target.select_workspace(workspace_id);
}
let mut response = list_backend_workers(&target).await.map_err(|error| {
io::Error::other(format!(
"failed to list Backend runtime workers from {}: {error}",
target.base_url
))
})?;
if include_stopped {
match list_backend_stopped_workers(&target).await {
Ok(stopped) => {
response.items.extend(stopped.items);
response.diagnostics.extend(stopped.diagnostics);
}
Err(error) => response.diagnostics.push(client::BackendDiagnostic {
code: "backend_stopped_workers_list_failed".to_string(),
severity: Some("error".to_string()),
message: error.to_string(),
}),
}
Err(error) => response.diagnostics.push(client::BackendDiagnostic {
code: "backend_stopped_workers_list_failed".to_string(),
severity: Some("error".to_string()),
message: error.to_string(),
}),
}
dedup_workers(&mut response.items);
if response.items.is_empty() {
let diagnostics = response
.diagnostics
.iter()
.map(|diagnostic| format!("{}: {}", diagnostic.code, diagnostic.message))
.collect::<Vec<_>>()
.join("; ");
let detail = if diagnostics.is_empty() {
"no backend diagnostics".to_string()
} else {
diagnostics
};
eprintln!(
"Backend returned no runtime workers for workspace {} ({detail}); choose another Workspace",
response.workspace_id
);
target.clear_workspace();
continue;
}
}
dedup_workers(&mut response.items);
if response.items.is_empty() {
let diagnostics = response
.diagnostics
.iter()
.map(|diagnostic| format!("{}: {}", diagnostic.code, diagnostic.message))
.collect::<Vec<_>>()
.join("; ");
let detail = if diagnostics.is_empty() {
"no backend diagnostics".to_string()
} else {
diagnostics
};
return Err(Box::new(io::Error::other(format!(
"Backend returned no runtime workers for workspace {} ({detail})",
response.workspace_id
))));
}
let selected = pick_worker(target.clone(), response.items)?;
let worker = if selected.state == "stopped" {
let restore_target = BackendRuntimeTarget::new(
target.base_url.clone(),
selected.runtime_id.clone(),
selected.worker_id.clone(),
);
restore_backend_worker(&restore_target)
.await
.map_err(|error| {
io::Error::other(format!(
"failed to restore Backend worker {}/{}: {error}",
selected.runtime_id, selected.worker_id
))
})?
.result
.worker
.unwrap_or(selected)
} else {
selected
};
let attach_target =
BackendRuntimeTarget::new(target.base_url, worker.runtime_id, worker.worker_id);
console::run_backend_runtime(attach_target).await
let selected = match pick_worker(target.clone(), response.items)? {
WorkerPickerResult::SwitchWorkspace => {
target.clear_workspace();
continue;
}
WorkerPickerResult::Selected(selected) => selected,
};
let worker = if selected.state == "stopped" {
let restore_target = target
.runtime_target(selected.runtime_id.clone(), selected.worker_id.clone())
.map_err(|error| io::Error::other(error.to_string()))?;
restore_backend_worker(&restore_target)
.await
.map_err(|error| {
io::Error::other(format!(
"failed to restore Backend worker {}/{}: {error}",
selected.runtime_id, selected.worker_id
))
})?
.result
.worker
.unwrap_or(selected)
} else {
selected
};
let attach_target = target
.runtime_target(worker.runtime_id, worker.worker_id)
.map_err(|error| io::Error::other(error.to_string()))?;
return console::run_backend_runtime(attach_target).await;
}
}
fn dedup_workers(workers: &mut Vec<BackendWorkerSummary>) {
@@ -92,10 +109,15 @@ fn dedup_workers(workers: &mut Vec<BackendWorkerSummary>) {
workers.retain(|worker| seen.insert((worker.runtime_id.clone(), worker.worker_id.clone())));
}
enum WorkerPickerResult {
Selected(BackendWorkerSummary),
SwitchWorkspace,
}
fn pick_worker(
target: BackendRuntimeListTarget,
mut workers: Vec<BackendWorkerSummary>,
) -> Result<BackendWorkerSummary, Box<dyn Error>> {
) -> Result<WorkerPickerResult, Box<dyn Error>> {
workers.sort_by(|a, b| {
a.runtime_id
.cmp(&b.runtime_id)
@@ -114,7 +136,13 @@ fn pick_worker(
Some(Action::Down) => state.next(),
Some(Action::Submit) => {
close_viewport(&mut terminal)?;
return Ok(state.selected_worker().clone());
return Ok(WorkerPickerResult::Selected(
state.selected_worker().clone(),
));
}
Some(Action::SwitchWorkspace) => {
close_viewport(&mut terminal)?;
return Ok(WorkerPickerResult::SwitchWorkspace);
}
Some(Action::Cancel) => {
close_viewport(&mut terminal)?;
@@ -181,6 +209,7 @@ enum Action {
Up,
Down,
Submit,
SwitchWorkspace,
Cancel,
}
@@ -197,6 +226,7 @@ fn poll_event() -> io::Result<Option<Action>> {
KeyCode::Char('k') if !ctrl => Some(Action::Up),
KeyCode::Char('j') if !ctrl => Some(Action::Down),
KeyCode::Enter => Some(Action::Submit),
KeyCode::Char('w') if !ctrl => Some(Action::SwitchWorkspace),
KeyCode::Esc => Some(Action::Cancel),
KeyCode::Char('c') if ctrl => Some(Action::Cancel),
_ => None,
@@ -239,6 +269,8 @@ fn draw(frame: &mut Frame<'_>, state: &BackendWorkerPickerState) {
Span::raw(" select "),
Span::styled("[enter]", Style::default().fg(Color::Green)),
Span::raw(" attach "),
Span::styled("[w]", Style::default().fg(Color::Cyan)),
Span::raw(" switch Workspace "),
Span::styled("[esc]", Style::default().fg(Color::Yellow)),
Span::raw(" cancel"),
])),
+240
View File
@@ -0,0 +1,240 @@
use client::{
BackendWorkspace, BackendWorkspaceCatalogTarget, CreateBackendWorkspaceRepository,
CreateBackendWorkspaceRequest, create_backend_workspace, list_backend_workspaces,
};
use crossterm::event::{self, Event, KeyCode, KeyEventKind};
use crossterm::terminal::{disable_raw_mode, enable_raw_mode};
use ratatui::Terminal;
use ratatui::backend::CrosstermBackend;
use ratatui::layout::{Constraint, Direction, Layout};
use ratatui::style::{Modifier, Style};
use ratatui::text::{Line, Span};
use ratatui::widgets::{Block, Borders, List, ListItem, ListState, Paragraph};
use std::error::Error;
use std::io::{self, IsTerminal, Write};
use std::time::{SystemTime, UNIX_EPOCH};
type PickerResult<T> = Result<T, Box<dyn Error>>;
pub(crate) async fn select_backend_workspace(base_url: &str) -> PickerResult<Option<String>> {
let target = BackendWorkspaceCatalogTarget::new(base_url);
let mut workspaces = Vec::new();
'catalog: loop {
let error = match list_backend_workspaces(&target).await {
Ok(items) => {
workspaces = items;
None
}
Err(fetch_error) => Some(format!("failed to refresh workspaces: {fetch_error}")),
};
match pick_workspace(&workspaces, error.as_deref())? {
WorkspacePickerAction::Select(index) => {
return Ok(workspaces.get(index).map(|item| item.workspace_id.clone()));
}
WorkspacePickerAction::Refresh => continue,
WorkspacePickerAction::Create => {
let Some(request) = prompt_create_request()? else {
continue;
};
loop {
match create_backend_workspace(&target, &request).await {
Ok(response) => return Ok(Some(response.workspace.workspace_id)),
Err(create_error) => {
let creation_error =
format!("workspace creation failed: {create_error}");
match pick_workspace(&workspaces, Some(&creation_error))? {
WorkspacePickerAction::Select(index) => {
return Ok(workspaces
.get(index)
.map(|item| item.workspace_id.clone()));
}
// Retry the exact request and operation key.
WorkspacePickerAction::Create => continue,
WorkspacePickerAction::Refresh => continue 'catalog,
WorkspacePickerAction::Cancel => return Ok(None),
}
}
}
}
}
WorkspacePickerAction::Cancel => return Ok(None),
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum WorkspacePickerAction {
Select(usize),
Create,
Refresh,
Cancel,
}
fn pick_workspace(
workspaces: &[BackendWorkspace],
error: Option<&str>,
) -> PickerResult<WorkspacePickerAction> {
if !io::stdin().is_terminal() || !io::stdout().is_terminal() {
return Err(
"Backend target has no configured workspace; an interactive terminal is required to choose one"
.into(),
);
}
let mut terminal = Terminal::new(CrosstermBackend::new(io::stdout()))?;
let mut selected = 0usize;
loop {
terminal.draw(|frame| {
let chunks = Layout::default()
.direction(Direction::Vertical)
.constraints([
Constraint::Length(3),
Constraint::Min(3),
Constraint::Length(if error.is_some() { 3 } else { 1 }),
])
.split(frame.area());
frame.render_widget(
Paragraph::new("Choose the Workspace for this Backend session")
.block(Block::default().title("Workspace").borders(Borders::ALL)),
chunks[0],
);
let rows = workspaces
.iter()
.map(|workspace| {
ListItem::new(Line::from(vec![
Span::styled(
workspace.display_name.clone(),
Style::default().add_modifier(Modifier::BOLD),
),
Span::raw(format!(" {} {}", workspace.workspace_id, workspace.state)),
]))
})
.collect::<Vec<_>>();
let rows = if rows.is_empty() {
vec![ListItem::new("No accessible Workspaces")]
} else {
rows
};
let mut state = ListState::default();
if !workspaces.is_empty() {
state.select(Some(selected));
}
frame.render_stateful_widget(
List::new(rows)
.block(Block::default().borders(Borders::ALL))
.highlight_symbol(""),
chunks[1],
&mut state,
);
let footer = error
.map(|message| {
format!(
"{message} [n] create/retry [r] refresh [Enter] select [Esc] cancel"
)
})
.unwrap_or_else(|| {
"[Enter] select [n] new [r] refresh [Esc] cancel".to_string()
});
frame.render_widget(Paragraph::new(footer), chunks[2]);
})?;
if let Event::Key(key) = event::read()?
&& key.kind == KeyEventKind::Press
{
match key.code {
KeyCode::Up if !workspaces.is_empty() => {
selected = selected.saturating_sub(1);
}
KeyCode::Down if !workspaces.is_empty() => {
selected = (selected + 1).min(workspaces.len() - 1);
}
KeyCode::Enter if !workspaces.is_empty() => {
terminal.clear()?;
return Ok(WorkspacePickerAction::Select(selected));
}
KeyCode::Char('n') => {
terminal.clear()?;
return Ok(WorkspacePickerAction::Create);
}
KeyCode::Char('r') => {
terminal.clear()?;
return Ok(WorkspacePickerAction::Refresh);
}
KeyCode::Esc | KeyCode::Char('q') => {
terminal.clear()?;
return Ok(WorkspacePickerAction::Cancel);
}
_ => {}
}
}
}
}
fn prompt_create_request() -> PickerResult<Option<CreateBackendWorkspaceRequest>> {
disable_raw_mode()?;
let result = prompt_create_request_inner();
enable_raw_mode()?;
result
}
fn prompt_create_request_inner() -> PickerResult<Option<CreateBackendWorkspaceRequest>> {
println!("Create Workspace (leave display name empty to cancel)");
let display_name = prompt_line("Workspace display name: ")?;
if display_name.is_empty() {
return Ok(None);
}
let uri = prompt_line("Initial repository absolute path/URI: ")?;
if uri.is_empty() {
println!("Repository path/URI is required.");
return Ok(None);
}
let repository_name = prompt_line("Repository display name [Main]: ")?;
let default_ref = prompt_line("Default ref [repository default]: ")?;
let operation_key = format!(
"tui-workspace-create-{}-{}",
std::process::id(),
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_nanos()
);
Ok(Some(CreateBackendWorkspaceRequest {
operation_key,
display_name,
repository: CreateBackendWorkspaceRepository {
uri,
display_name: Some(if repository_name.is_empty() {
"Main".to_string()
} else {
repository_name
}),
default_ref: (!default_ref.is_empty()).then_some(default_ref),
},
}))
}
fn prompt_line(prompt: &str) -> PickerResult<String> {
print!("{prompt}");
io::stdout().flush()?;
let mut value = String::new();
io::stdin().read_line(&mut value)?;
Ok(value.trim().to_string())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn picker_actions_distinguish_switch_refresh_create_and_cancel() {
assert_ne!(
WorkspacePickerAction::Create,
WorkspacePickerAction::Refresh
);
assert_ne!(
WorkspacePickerAction::Select(0),
WorkspacePickerAction::Cancel
);
}
}
+1
View File
@@ -1,5 +1,6 @@
mod app;
mod backend_worker_picker;
mod backend_workspace_picker;
mod block;
mod cache;
mod command;
+14
View File
@@ -16,6 +16,7 @@ use std::sync::Arc;
use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use tokio::sync::broadcast;
pub use delegation::{
AppliedWorkdirDelegation, ReadOnlyWorkdirSession, WorkdirDelegation,
@@ -192,6 +193,19 @@ pub trait WorkdirSession: std::fmt::Debug + Send + Sync {
request: CommandOutputRequest,
) -> Result<CommandOutput, WorkdirError>;
async fn cancel_command(&self, handle: CommandHandle) -> Result<(), WorkdirError>;
/// Subscribe to bounded provider-owned command telemetry. Implementations
/// that do not expose live command observation may keep the default.
fn subscribe_command_events(&self) -> Option<broadcast::Receiver<CommandEvent>> {
None
}
/// Return the bounded current command state used to recover from a lagged
/// provider subscription without replaying command output into history.
fn command_snapshot(&self) -> Vec<CommandSnapshot> {
Vec::new()
}
/// Terminal, idempotent release of this Worker-bound operation session.
async fn close(&self) -> Result<(), WorkdirError>;
}
+633 -55
View File
@@ -14,21 +14,22 @@ use std::io::Write as _;
use std::io::{Read as _, Seek as _, SeekFrom};
use std::path::{Path, PathBuf};
use std::process::Stdio;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::time::Duration;
use std::sync::{Arc, Mutex as StdMutex};
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use async_trait::async_trait;
use manifest::{Permission, Scope, ScopeConfig, ScopeRule, SharedScope};
use sha2::{Digest, Sha256};
use tokio::process::Command;
use tokio::sync::{Mutex, Notify};
use tokio::sync::{Mutex, Notify, broadcast, watch};
use tokio::task::JoinHandle;
use crate::{
CommandHandle, CommandOutput, CommandOutputRequest, CommandRequest, CommandStatus, EditRequest,
EditResult, GlobRequest, GlobResult, GrepRequest, GrepResult, ListRequest, ListResult,
ReadRequest, ReadResult, StatRequest, StatResult, Workdir, WorkdirDelegationPermission,
CommandEvent, CommandHandle, CommandOutput, CommandOutputRequest, CommandRequest,
CommandSnapshot, CommandStatus, CommandStream, CommandStreamSlice, EditRequest, EditResult,
GlobRequest, GlobResult, GrepRequest, GrepResult, ListRequest, ListResult, ReadRequest,
ReadResult, StatRequest, StatResult, Workdir, WorkdirDelegationPermission,
WorkdirDelegationRequest, WorkdirError, WorkdirPath, WorkdirSession,
WorkdirSessionCapabilities, WorkdirSessionCapability, WorkdirSessionHandle, WriteRequest,
WriteResult,
@@ -36,15 +37,172 @@ use crate::{
#[cfg(test)]
use crate::{EntryKind, WriteOutcome};
const COMMAND_EVENT_CHANNEL_CAPACITY: usize = 256;
const COMMAND_EVENT_CHUNK_BYTES: usize = 8 * 1024;
const COMMAND_SNAPSHOT_STREAM_BYTES: usize = 32 * 1024;
fn command_observed_at_ms() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_millis()
.try_into()
.unwrap_or(u64::MAX)
}
#[derive(Debug)]
enum LocalCommand {
Running {
task: JoinHandle<Result<CommandOutput, WorkdirError>>,
completion: Arc<Notify>,
cancel: watch::Sender<bool>,
},
Completed(CommandOutput),
}
#[derive(Debug, Clone)]
struct CommandTelemetry {
inner: Arc<CommandTelemetryInner>,
}
#[derive(Debug)]
struct CommandTelemetryInner {
snapshots: StdMutex<HashMap<String, CommandSnapshot>>,
events: broadcast::Sender<CommandEvent>,
}
impl CommandTelemetry {
fn new() -> Self {
let (events, _) = broadcast::channel(COMMAND_EVENT_CHANNEL_CAPACITY);
Self {
inner: Arc::new(CommandTelemetryInner {
snapshots: StdMutex::new(HashMap::new()),
events,
}),
}
}
fn subscribe(&self) -> broadcast::Receiver<CommandEvent> {
self.inner.events.subscribe()
}
fn snapshot(&self) -> Vec<CommandSnapshot> {
let mut snapshots = self
.inner
.snapshots
.lock()
.expect("command telemetry mutex poisoned")
.values()
.cloned()
.collect::<Vec<_>>();
snapshots.sort_by(|left, right| left.command_id.cmp(&right.command_id));
snapshots
}
fn started(&self, command_id: &str, tool_call_id: Option<String>) {
let observed_at_ms = command_observed_at_ms();
self.inner
.snapshots
.lock()
.expect("command telemetry mutex poisoned")
.insert(
command_id.to_string(),
CommandSnapshot {
command_id: command_id.to_string(),
tool_call_id: tool_call_id.clone(),
status: CommandStatus::Running,
started_at_ms: observed_at_ms,
observed_at_ms,
last_output_at_ms: None,
stdout: CommandStreamSlice::default(),
stderr: CommandStreamSlice::default(),
exit_code: None,
},
);
let _ = self.inner.events.send(CommandEvent::Started {
command_id: command_id.to_string(),
tool_call_id,
observed_at_ms,
});
}
fn output(&self, command_id: &str, stream: CommandStream, start_offset: u64, bytes: &[u8]) {
if bytes.is_empty() {
return;
}
let end_offset = start_offset.saturating_add(bytes.len() as u64);
let content = String::from_utf8_lossy(bytes).into_owned();
let observed_at_ms = command_observed_at_ms();
if let Some(snapshot) = self
.inner
.snapshots
.lock()
.expect("command telemetry mutex poisoned")
.get_mut(command_id)
{
snapshot.observed_at_ms = observed_at_ms;
snapshot.last_output_at_ms = Some(observed_at_ms);
let target = match stream {
CommandStream::Stdout => &mut snapshot.stdout,
CommandStream::Stderr => &mut snapshot.stderr,
};
target.end_offset = end_offset;
target.content.push_str(&content);
if target.content.len() > COMMAND_SNAPSHOT_STREAM_BYTES {
let mut cut = target.content.len() - COMMAND_SNAPSHOT_STREAM_BYTES;
while cut < target.content.len() && !target.content.is_char_boundary(cut) {
cut += 1;
}
target.content.drain(..cut);
target.start_offset = end_offset.saturating_sub(target.content.len() as u64);
target.truncated = true;
}
}
let _ = self.inner.events.send(CommandEvent::Output {
command_id: command_id.to_string(),
stream,
start_offset,
end_offset,
content,
observed_at_ms,
});
}
fn terminal(&self, command_id: &str, status: CommandStatus, exit_code: Option<i32>) {
let observed_at_ms = command_observed_at_ms();
let (stdout_end_offset, stderr_end_offset) = if let Some(snapshot) = self
.inner
.snapshots
.lock()
.expect("command telemetry mutex poisoned")
.get_mut(command_id)
{
snapshot.status = status;
snapshot.exit_code = exit_code;
snapshot.observed_at_ms = observed_at_ms;
(snapshot.stdout.end_offset, snapshot.stderr.end_offset)
} else {
(0, 0)
};
let _ = self.inner.events.send(CommandEvent::Terminal {
command_id: command_id.to_string(),
status,
exit_code,
stdout_end_offset,
stderr_end_offset,
observed_at_ms,
});
}
fn remove(&self, command_id: &str) {
self.inner
.snapshots
.lock()
.expect("command telemetry mutex poisoned")
.remove(command_id);
}
}
#[derive(Debug)]
struct ScopeAccess(Arc<Scope>);
@@ -69,6 +227,7 @@ struct LocalWorkdirSessionInner {
close_lock: Mutex<()>,
next_command_id: AtomicU64,
commands: Mutex<HashMap<String, LocalCommand>>,
command_telemetry: CommandTelemetry,
}
impl Drop for LocalWorkdirSessionInner {
@@ -171,6 +330,7 @@ impl LocalWorkdirSession {
close_lock: Mutex::new(()),
next_command_id: AtomicU64::new(1),
commands: Mutex::new(HashMap::new()),
command_telemetry: CommandTelemetry::new(),
}),
}
}
@@ -502,23 +662,35 @@ impl WorkdirSession for LocalWorkdirSession {
async fn start_command(&self, request: CommandRequest) -> Result<CommandHandle, WorkdirError> {
self.ensure_capability(WorkdirSessionCapability::Command)?;
self.ensure_open()?;
let id = self.inner.next_command_id.fetch_add(1, Ordering::Relaxed);
let handle = CommandHandle(format!("command-{id}"));
let cwd = self.inner.cwd.clone();
let completion = Arc::new(Notify::new());
let task_completion = Arc::clone(&completion);
let command_id = handle.0.clone();
let telemetry = self.inner.command_telemetry.clone();
let (cancel, cancel_rx) = watch::channel(false);
let task = tokio::spawn(async move {
let output = run_command(cwd, request).await;
let output = run_command(cwd, request, command_id, telemetry, cancel_rx).await;
task_completion.notify_one();
output
});
let mut commands = self.inner.commands.lock().await;
if let Err(error) = self.ensure_open() {
let _ = cancel.send(true);
task.abort();
completion.notify_one();
return Err(error);
}
commands.insert(handle.0.clone(), LocalCommand::Running { task, completion });
commands.insert(
handle.0.clone(),
LocalCommand::Running {
task,
completion,
cancel,
},
);
Ok(handle)
}
@@ -530,7 +702,13 @@ impl WorkdirSession for LocalWorkdirSession {
.ok_or_else(|| WorkdirError::UnknownCommand(handle.0.clone()))?;
Ok(match command {
LocalCommand::Running { task, .. } if !task.is_finished() => CommandStatus::Running,
LocalCommand::Running { .. } => CommandStatus::Completed,
LocalCommand::Running { .. } => self
.inner
.command_telemetry
.snapshot()
.into_iter()
.find(|snapshot| snapshot.command_id == handle.0)
.map_or(CommandStatus::Completed, |snapshot| snapshot.status),
LocalCommand::Completed(output) => output.status,
})
}
@@ -584,36 +762,75 @@ impl WorkdirSession for LocalWorkdirSession {
if !self.inner.closed.load(Ordering::Acquire) {
commands.insert(request.handle.0, LocalCommand::Completed(output));
}
} else {
self.inner.command_telemetry.remove(&request.handle.0);
}
Ok(page)
}
async fn cancel_command(&self, handle: CommandHandle) -> Result<(), WorkdirError> {
self.ensure_capability(WorkdirSessionCapability::Command)?;
let command = self
.inner
.commands
.lock()
.await
.remove(&handle.0)
.ok_or_else(|| WorkdirError::UnknownCommand(handle.0))?;
if let LocalCommand::Running { task, completion } = command {
task.abort();
completion.notify_one();
let cancel = {
let commands = self.inner.commands.lock().await;
let command = commands
.get(&handle.0)
.ok_or_else(|| WorkdirError::UnknownCommand(handle.0.clone()))?;
match command {
LocalCommand::Running { task, cancel, .. } if !task.is_finished() => {
Some(cancel.clone())
}
_ => None,
}
};
if let Some(cancel) = cancel {
let _ = cancel.send(true);
}
Ok(())
}
fn subscribe_command_events(&self) -> Option<broadcast::Receiver<CommandEvent>> {
self.inner
.capabilities
.supports(WorkdirSessionCapability::Command)
.then(|| self.inner.command_telemetry.subscribe())
}
fn command_snapshot(&self) -> Vec<CommandSnapshot> {
if self
.inner
.capabilities
.supports(WorkdirSessionCapability::Command)
{
self.inner.command_telemetry.snapshot()
} else {
Vec::new()
}
}
async fn close(&self) -> Result<(), WorkdirError> {
let _close_guard = self.inner.close_lock.lock().await;
if self.inner.closed.swap(true, Ordering::AcqRel) {
return Ok(());
}
let mut commands = self.inner.commands.lock().await;
for (_, command) in commands.drain() {
if let LocalCommand::Running { task, completion } = command {
task.abort();
completion.notify_one();
let commands = {
let mut commands = self.inner.commands.lock().await;
commands
.drain()
.map(|(_, command)| command)
.collect::<Vec<_>>()
};
for command in commands {
match command {
LocalCommand::Running {
task,
completion,
cancel,
} => {
let _ = cancel.send(true);
let _ = task.await;
completion.notify_one();
}
LocalCommand::Completed(_) => {}
}
}
Ok(())
@@ -680,7 +897,13 @@ fn sanitize_error(error: WorkdirError, logical: &WorkdirPath) -> WorkdirError {
}
}
async fn run_command(cwd: PathBuf, request: CommandRequest) -> Result<CommandOutput, WorkdirError> {
async fn run_command(
cwd: PathBuf,
request: CommandRequest,
command_id: String,
telemetry: CommandTelemetry,
mut cancel: watch::Receiver<bool>,
) -> Result<CommandOutput, WorkdirError> {
let stdout = tempfile::NamedTempFile::new().map_err(|error| WorkdirError::io(&cwd, error))?;
let stderr = tempfile::NamedTempFile::new().map_err(|error| WorkdirError::io(&cwd, error))?;
let stdout_path = stdout.into_temp_path();
@@ -690,7 +913,8 @@ async fn run_command(cwd: PathBuf, request: CommandRequest) -> Result<CommandOut
let stderr_file = std::fs::File::create(&stderr_path)
.map_err(|error| WorkdirError::io(&stderr_path, error))?;
let mut child = Command::new("bash")
telemetry.started(&command_id, request.tool_call_id.clone());
let mut child = match Command::new("bash")
.arg("-c")
.arg(&request.command)
.current_dir(&cwd)
@@ -699,45 +923,188 @@ async fn run_command(cwd: PathBuf, request: CommandRequest) -> Result<CommandOut
.stderr(Stdio::from(stderr_file))
.kill_on_drop(true)
.spawn()
.map_err(|error| WorkdirError::io(&cwd, error))?;
let timed_out = match tokio::time::timeout(
Duration::from_secs(request.timeout_secs.max(1)),
child.wait(),
)
.await
{
Ok(result) => {
let status = result.map_err(|error| WorkdirError::io(&cwd, error))?;
let (content, truncated) =
read_command_output_files(&stdout_path, &stderr_path, request.output_limit.max(1))?;
return Ok(CommandOutput {
status: CommandStatus::Completed,
exit_code: status.code(),
timed_out: false,
content,
next_cursor: None,
truncated,
});
}
Err(_) => {
let _ = child.kill().await;
true
Ok(child) => child,
Err(error) => {
telemetry.terminal(&command_id, CommandStatus::Failed, None);
return Err(WorkdirError::io(&cwd, error));
}
};
let mut stdout_reader =
std::fs::File::open(&stdout_path).map_err(|error| WorkdirError::io(&stdout_path, error))?;
let mut stderr_reader =
std::fs::File::open(&stderr_path).map_err(|error| WorkdirError::io(&stderr_path, error))?;
let mut stdout_decoder = CommandOutputDecoder::default();
let mut stderr_decoder = CommandOutputDecoder::default();
let mut timeout = Box::pin(tokio::time::sleep(Duration::from_secs(
request.timeout_secs.max(1),
)));
let mut interval = tokio::time::interval(Duration::from_millis(50));
interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
interval.tick().await;
let (status, exit_code) = loop {
tokio::select! {
exit = child.wait() => {
let exit = exit.map_err(|error| WorkdirError::io(&cwd, error))?;
break (
if exit.success() { CommandStatus::Completed } else { CommandStatus::Failed },
exit.code(),
);
}
_ = &mut timeout => {
let _ = child.start_kill();
let exit_code = child.wait().await.ok().and_then(|status| status.code());
break (CommandStatus::TimedOut, exit_code);
}
changed = cancel.changed() => {
if changed.is_err() || *cancel.borrow() {
let _ = child.start_kill();
let exit_code = child.wait().await.ok().and_then(|status| status.code());
break (CommandStatus::Cancelled, exit_code);
}
}
_ = interval.tick() => {
publish_available_output(
&mut stdout_reader,
&mut stdout_decoder,
&telemetry,
&command_id,
CommandStream::Stdout,
&stdout_path,
false,
)?;
publish_available_output(
&mut stderr_reader,
&mut stderr_decoder,
&telemetry,
&command_id,
CommandStream::Stderr,
&stderr_path,
false,
)?;
}
}
};
publish_available_output(
&mut stdout_reader,
&mut stdout_decoder,
&telemetry,
&command_id,
CommandStream::Stdout,
&stdout_path,
true,
)?;
publish_available_output(
&mut stderr_reader,
&mut stderr_decoder,
&telemetry,
&command_id,
CommandStream::Stderr,
&stderr_path,
true,
)?;
telemetry.terminal(&command_id, status, exit_code);
let (content, truncated) =
read_command_output_files(&stdout_path, &stderr_path, request.output_limit.max(1))?;
Ok(CommandOutput {
status: CommandStatus::Failed,
exit_code: None,
timed_out,
status,
exit_code,
timed_out: status == CommandStatus::TimedOut,
content,
next_cursor: None,
truncated,
})
}
#[derive(Debug, Default)]
struct CommandOutputDecoder {
read_offset: u64,
emitted_offset: u64,
pending: Vec<u8>,
}
fn publish_available_output(
file: &mut std::fs::File,
decoder: &mut CommandOutputDecoder,
telemetry: &CommandTelemetry,
command_id: &str,
stream: CommandStream,
path: &Path,
flush: bool,
) -> Result<(), WorkdirError> {
file.seek(SeekFrom::Start(decoder.read_offset))
.map_err(|error| WorkdirError::io(path, error))?;
loop {
let mut buffer = vec![0; COMMAND_EVENT_CHUNK_BYTES];
let read = file
.read(&mut buffer)
.map_err(|error| WorkdirError::io(path, error))?;
if read == 0 {
publish_decoded_output(decoder, telemetry, command_id, stream, flush);
return Ok(());
}
decoder.pending.extend_from_slice(&buffer[..read]);
decoder.read_offset = decoder.read_offset.saturating_add(read as u64);
publish_decoded_output(decoder, telemetry, command_id, stream, false);
if read < COMMAND_EVENT_CHUNK_BYTES {
if flush {
publish_decoded_output(decoder, telemetry, command_id, stream, true);
}
return Ok(());
}
}
}
fn publish_decoded_output(
decoder: &mut CommandOutputDecoder,
telemetry: &CommandTelemetry,
command_id: &str,
stream: CommandStream,
flush: bool,
) {
let prefix_len = if flush {
decoder.pending.len()
} else {
stable_utf8_prefix_len(&decoder.pending)
};
if prefix_len == 0 {
return;
}
telemetry.output(
command_id,
stream,
decoder.emitted_offset,
&decoder.pending[..prefix_len],
);
decoder.emitted_offset = decoder.emitted_offset.saturating_add(prefix_len as u64);
decoder.pending.drain(..prefix_len);
}
/// Return the byte prefix that can be decoded now without replacing a valid
/// UTF-8 scalar whose remaining bytes may arrive in a later file read. Definite
/// invalid sequences remain in the prefix and are rendered lossily, preserving
/// the existing arbitrary-byte output behavior.
fn stable_utf8_prefix_len(bytes: &[u8]) -> usize {
let mut inspected = 0;
while inspected < bytes.len() {
match std::str::from_utf8(&bytes[inspected..]) {
Ok(_) => return bytes.len(),
Err(error) => {
inspected += error.valid_up_to();
match error.error_len() {
Some(invalid_len) => inspected += invalid_len,
None => return inspected,
}
}
}
}
inspected
}
fn read_command_output_files(
stdout_path: &Path,
stderr_path: &Path,
@@ -1024,6 +1391,7 @@ mod tests {
command: "sleep 30".to_owned(),
timeout_secs: 60,
output_limit: 1024,
tool_call_id: None,
},
)
.await
@@ -1549,6 +1917,7 @@ mod tests {
command: "pwd && printf provider-command".into(),
timeout_secs: 5,
output_limit: 4096,
tool_call_id: None,
},
)
.await
@@ -1583,6 +1952,7 @@ mod tests {
command: "printf 'aéz'".into(),
timeout_secs: 5,
output_limit: 1024,
tool_call_id: None,
},
)
.await
@@ -1620,6 +1990,213 @@ mod tests {
));
}
#[test]
fn command_output_decoder_preserves_utf8_split_across_file_reads() {
let dir = TempDir::new().unwrap();
let path = dir.path().join("command.out");
let mut first_write = vec![b'a'; COMMAND_EVENT_CHUNK_BYTES - 1];
first_write.push(0xe2);
std::fs::write(&path, first_write).unwrap();
let telemetry = CommandTelemetry::new();
let mut events = telemetry.subscribe();
telemetry.started("command-utf8", None);
let mut decoder = CommandOutputDecoder::default();
let mut reader = std::fs::File::open(&path).unwrap();
publish_available_output(
&mut reader,
&mut decoder,
&telemetry,
"command-utf8",
CommandStream::Stdout,
&path,
false,
)
.unwrap();
assert_eq!(decoder.pending, vec![0xe2]);
let mut writer = std::fs::OpenOptions::new()
.append(true)
.open(&path)
.unwrap();
writer.write_all(&[0x82, 0xac]).unwrap();
writer.flush().unwrap();
publish_available_output(
&mut reader,
&mut decoder,
&telemetry,
"command-utf8",
CommandStream::Stdout,
&path,
false,
)
.unwrap();
let output = std::iter::from_fn(|| events.try_recv().ok())
.filter_map(|event| match event {
CommandEvent::Output {
stream: CommandStream::Stdout,
content,
..
} => Some(content),
_ => None,
})
.collect::<String>();
assert_eq!(output.len(), COMMAND_EVENT_CHUNK_BYTES - 1 + "".len());
assert!(output.ends_with('€'));
assert!(!output.contains('\u{fffd}'));
assert!(decoder.pending.is_empty());
}
#[tokio::test]
async fn provider_streams_bounded_command_lifecycle_and_distinct_output() {
let dir = TempDir::new().unwrap();
let workdir = make_fs(&dir);
let mut events = WorkdirSession::subscribe_command_events(&workdir)
.expect("local command observation must be available");
let handle = WorkdirSession::start_command(
&workdir,
CommandRequest {
command: "printf ready; printf warning >&2; sleep 0.2; printf done".into(),
timeout_secs: 5,
output_limit: 1024,
tool_call_id: Some("tool-7".into()),
},
)
.await
.unwrap();
let mut stdout = String::new();
let mut stdout_chunks = 0;
let mut stderr = String::new();
let mut terminal = None;
while terminal.is_none() {
let event = tokio::time::timeout(Duration::from_secs(2), events.recv())
.await
.expect("command telemetry should not stall")
.unwrap();
match event {
CommandEvent::Started {
command_id,
tool_call_id,
..
} => {
assert_eq!(command_id, handle.0);
assert_eq!(tool_call_id.as_deref(), Some("tool-7"));
}
CommandEvent::Output {
command_id,
stream,
content,
..
} => {
assert_eq!(command_id, handle.0);
match stream {
CommandStream::Stdout => {
stdout_chunks += 1;
stdout.push_str(&content);
}
CommandStream::Stderr => stderr.push_str(&content),
}
}
CommandEvent::Terminal {
command_id,
status,
exit_code,
stdout_end_offset,
stderr_end_offset,
observed_at_ms,
} => {
assert_eq!(command_id, handle.0);
terminal = Some((
status,
exit_code,
stdout_end_offset,
stderr_end_offset,
observed_at_ms,
));
}
}
}
let (status, exit_code, stdout_end_offset, stderr_end_offset, observed_at_ms) =
terminal.unwrap();
assert_eq!(status, CommandStatus::Completed);
assert_eq!(exit_code, Some(0));
assert_eq!(stdout_end_offset, "readydone".len() as u64);
assert_eq!(stderr_end_offset, "warning".len() as u64);
assert!(observed_at_ms > 0);
assert!(
stdout_chunks >= 2,
"long-running output should stream incrementally"
);
assert_eq!(stdout, "readydone");
assert_eq!(stderr, "warning");
let snapshot = WorkdirSession::command_snapshot(&workdir);
assert_eq!(snapshot.len(), 1);
assert_eq!(snapshot[0].status, CommandStatus::Completed);
assert_eq!(snapshot[0].stdout.content, "readydone");
assert_eq!(snapshot[0].stderr.content, "warning");
let output = WorkdirSession::command_output(
&workdir,
CommandOutputRequest {
handle,
cursor: 0,
limit: 1024,
wait: true,
},
)
.await
.unwrap();
assert_eq!(output.status, CommandStatus::Completed);
assert!(WorkdirSession::command_snapshot(&workdir).is_empty());
}
#[tokio::test]
async fn provider_distinguishes_timed_out_terminal_state() {
let dir = TempDir::new().unwrap();
let workdir = make_fs(&dir);
let mut events = WorkdirSession::subscribe_command_events(&workdir).unwrap();
let handle = WorkdirSession::start_command(
&workdir,
CommandRequest {
command: "sleep 30".into(),
timeout_secs: 1,
output_limit: 1024,
tool_call_id: None,
},
)
.await
.unwrap();
let output = WorkdirSession::command_output(
&workdir,
CommandOutputRequest {
handle: handle.clone(),
cursor: 0,
limit: 1024,
wait: true,
},
)
.await
.unwrap();
assert_eq!(output.status, CommandStatus::TimedOut);
assert!(output.timed_out);
let mut terminal = None;
while let Ok(event) = events.try_recv() {
if let CommandEvent::Terminal {
command_id,
status,
exit_code,
..
} = event
{
terminal = Some((command_id, status, exit_code));
}
}
assert_eq!(terminal, Some((handle.0, CommandStatus::TimedOut, None)));
}
#[tokio::test]
async fn provider_cancels_active_command() {
let dir = TempDir::new().unwrap();
@@ -1630,6 +2207,7 @@ mod tests {
command: "sleep 30".into(),
timeout_secs: 60,
output_limit: 1024,
tool_call_id: None,
},
)
.await
@@ -1658,12 +2236,12 @@ mod tests {
WorkdirSession::cancel_command(&workdir, handle.clone())
.await
.unwrap();
let waiter_error = tokio::time::timeout(Duration::from_secs(1), waiter)
let output = tokio::time::timeout(Duration::from_secs(1), waiter)
.await
.expect("cancel should wake command output waiters")
.unwrap()
.unwrap_err();
assert!(matches!(waiter_error, WorkdirError::UnknownCommand(_)));
.unwrap();
assert_eq!(output.status, CommandStatus::Cancelled);
assert!(matches!(
WorkdirSession::command_status(&workdir, handle).await,
Err(WorkdirError::UnknownCommand(_))
+61 -1
View File
@@ -9,6 +9,11 @@ pub struct CommandRequest {
pub command: String,
pub timeout_secs: u64,
pub output_limit: usize,
/// Optional caller-owned correlation id. Bash supplies its tool-call id so
/// user-facing command telemetry can update the corresponding Console row
/// without exposing provider/session handles.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub tool_call_id: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
@@ -24,8 +29,63 @@ pub struct CommandOutputRequest {
pub enum CommandStatus {
Running,
Completed,
Cancelled,
Failed,
TimedOut,
Cancelled,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CommandStream {
Stdout,
Stderr,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
pub struct CommandStreamSlice {
pub start_offset: u64,
pub end_offset: u64,
pub content: String,
pub truncated: bool,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct CommandSnapshot {
pub command_id: String,
pub tool_call_id: Option<String>,
pub status: CommandStatus,
pub started_at_ms: u64,
pub observed_at_ms: u64,
pub last_output_at_ms: Option<u64>,
pub stdout: CommandStreamSlice,
pub stderr: CommandStreamSlice,
pub exit_code: Option<i32>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum CommandEvent {
Started {
command_id: String,
tool_call_id: Option<String>,
observed_at_ms: u64,
},
Output {
command_id: String,
stream: CommandStream,
start_offset: u64,
end_offset: u64,
content: String,
observed_at_ms: u64,
},
Terminal {
command_id: String,
status: CommandStatus,
exit_code: Option<i32>,
stdout_end_offset: u64,
stderr_end_offset: u64,
observed_at_ms: u64,
},
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+8 -2
View File
@@ -1432,7 +1432,10 @@ impl Runtime {
context_tokens: 0,
},
status: protocol::WorkerStatus::Idle,
in_flight: protocol::InFlightSnapshot { blocks: Vec::new() },
in_flight: protocol::InFlightSnapshot {
blocks: Vec::new(),
commands: Vec::new(),
},
internal_workers: Vec::new(),
})
}
@@ -4088,7 +4091,10 @@ mod tests {
context_tokens: 64,
},
status: protocol::WorkerStatus::Running,
in_flight: protocol::InFlightSnapshot { blocks: Vec::new() },
in_flight: protocol::InFlightSnapshot {
blocks: Vec::new(),
commands: Vec::new(),
},
internal_workers: Vec::new(),
},
);
+130 -2
View File
@@ -28,8 +28,14 @@ use crate::worker::{
WorkerRunResult,
};
use protocol::{
AlertLevel, AlertSource, ErrorCode, Event, Method, RewindTargetId, RunResult, Segment,
TurnResult, WorkerStatus,
AlertLevel, AlertSource, CommandEvent as ProtocolCommandEvent,
CommandSnapshot as ProtocolCommandSnapshot, CommandStatus as ProtocolCommandStatus,
CommandStream as ProtocolCommandStream, CommandStreamSlice as ProtocolCommandStreamSlice,
ErrorCode, Event, Method, RewindTargetId, RunResult, Segment, TurnResult, WorkerStatus,
};
use workdir::{
CommandEvent as WorkdirCommandEvent, CommandSnapshot as WorkdirCommandSnapshot,
CommandStatus as WorkdirCommandStatus, CommandStream as WorkdirCommandStream, WorkdirSession,
};
// ---------------------------------------------------------------------------
@@ -424,6 +430,9 @@ impl WorkerController {
Some(method_tx.downgrade()),
)
.await?;
if let Some(session) = fs_for_view.as_ref() {
wire_workdir_command_events(session, &in_flight);
}
// Intake role Workers self-terminate only after a successful
// TicketIntakeReady turn has fully settled back to Idle. The request
@@ -498,6 +507,125 @@ impl WorkerController {
}
}
pub(crate) fn wire_workdir_command_events(
session: &Arc<dyn WorkdirSession>,
in_flight: &InFlightEvents,
) {
in_flight.replace_command_snapshot(protocol_command_snapshots(session.as_ref()));
let Some(mut events) = session.subscribe_command_events() else {
return;
};
// Keep only a weak reference in the observer task. Holding the session
// strongly here would keep its broadcast sender alive forever and prevent
// the receiver from observing closure during Worker teardown.
let session = Arc::downgrade(session);
let in_flight = in_flight.clone();
tokio::spawn(async move {
loop {
match events.recv().await {
Ok(event) => in_flight.publish_command_event(protocol_command_event(event)),
Err(broadcast::error::RecvError::Lagged(_)) => {
let Some(session) = session.upgrade() else {
break;
};
in_flight
.replace_command_snapshot(protocol_command_snapshots(session.as_ref()));
}
Err(broadcast::error::RecvError::Closed) => break,
}
}
});
}
fn protocol_command_snapshots(session: &dyn WorkdirSession) -> Vec<ProtocolCommandSnapshot> {
session
.command_snapshot()
.into_iter()
.map(protocol_command_snapshot)
.collect()
}
fn protocol_command_snapshot(snapshot: WorkdirCommandSnapshot) -> ProtocolCommandSnapshot {
ProtocolCommandSnapshot {
command_id: snapshot.command_id,
tool_call_id: snapshot.tool_call_id,
status: protocol_command_status(snapshot.status),
started_at_ms: snapshot.started_at_ms,
observed_at_ms: snapshot.observed_at_ms,
last_output_at_ms: snapshot.last_output_at_ms,
stdout: ProtocolCommandStreamSlice {
start_offset: snapshot.stdout.start_offset,
end_offset: snapshot.stdout.end_offset,
content: snapshot.stdout.content,
truncated: snapshot.stdout.truncated,
},
stderr: ProtocolCommandStreamSlice {
start_offset: snapshot.stderr.start_offset,
end_offset: snapshot.stderr.end_offset,
content: snapshot.stderr.content,
truncated: snapshot.stderr.truncated,
},
exit_code: snapshot.exit_code,
}
}
fn protocol_command_event(event: WorkdirCommandEvent) -> ProtocolCommandEvent {
match event {
WorkdirCommandEvent::Started {
command_id,
tool_call_id,
observed_at_ms,
} => ProtocolCommandEvent::Started {
command_id,
tool_call_id,
observed_at_ms,
},
WorkdirCommandEvent::Output {
command_id,
stream,
start_offset,
end_offset,
content,
observed_at_ms,
} => ProtocolCommandEvent::Output {
command_id,
stream: match stream {
WorkdirCommandStream::Stdout => ProtocolCommandStream::Stdout,
WorkdirCommandStream::Stderr => ProtocolCommandStream::Stderr,
},
start_offset,
end_offset,
content,
observed_at_ms,
},
WorkdirCommandEvent::Terminal {
command_id,
status,
exit_code,
stdout_end_offset,
stderr_end_offset,
observed_at_ms,
} => ProtocolCommandEvent::Terminal {
command_id,
status: protocol_command_status(status),
exit_code,
stdout_end_offset,
stderr_end_offset,
observed_at_ms,
},
}
}
fn protocol_command_status(status: WorkdirCommandStatus) -> ProtocolCommandStatus {
match status {
WorkdirCommandStatus::Running => ProtocolCommandStatus::Running,
WorkdirCommandStatus::Completed => ProtocolCommandStatus::Completed,
WorkdirCommandStatus::Failed => ProtocolCommandStatus::Failed,
WorkdirCommandStatus::TimedOut => ProtocolCommandStatus::TimedOut,
WorkdirCommandStatus::Cancelled => ProtocolCommandStatus::Cancelled,
}
}
/// Wire the per-event broadcast bridges on the Worker's Engine. Each callback
/// re-publishes a worker-level signal as a `protocol::Event` on `event_tx`
/// so subscribers (TUI, socket clients) get a single typed stream.
@@ -16,7 +16,7 @@ use crate::feature::{
FeatureDescriptor, FeatureInstallContext, FeatureInstallError, FeatureModule,
ServiceDeclaration, ServiceId, ToolContribution, ToolDeclaration,
};
use crate::spawn::registry::SpawnedWorkerRegistry;
use crate::spawn::registry::{SpawnedWorkerRegistry, SubWorkerStopSummary};
use crate::worker::{
WorkspaceClient, WorkspaceClientError, WorkspaceRequest, WorkspaceRequestMethod,
WorkspaceResponse,
@@ -137,14 +137,19 @@ impl WorkerControlService for WorkspaceWorkerControlService {
let registry = self.registry.as_ref().ok_or_else(|| {
WorkspaceClientError::Request("unknown Worker or permission not granted".to_string())
})?;
registry
let summary = registry
.remove_internal(name)
.await
.map_err(|error| WorkspaceClientError::Request(error.to_string()))?;
.map_err(|error| WorkspaceClientError::Request(error.to_string()))?
.ok_or_else(|| {
WorkspaceClientError::Request(
"unknown Worker or permission not granted".to_string(),
)
})?;
Ok(WorkspaceResponse {
status: 200,
body: serde_json::json!({ "subject": { "kind": "sub_worker", "name": name } })
.to_string(),
body: serde_json::to_string(&summary)
.map_err(|error| WorkspaceClientError::Request(error.to_string()))?,
})
}
@@ -829,6 +834,15 @@ fn tool_output(
response.status, response.body
)));
}
if operation == WorkerOperation::Stop
&& let Ok(summary) = serde_json::from_str::<SubWorkerStopSummary>(&response.body)
{
return Ok(ToolOutput {
summary: render_subworker_stop_summary(&summary),
content: Some(response.body),
attachments: Vec::new(),
});
}
Ok(ToolOutput {
summary: format!("{} completed", operation.tool_name()),
content: Some(response.body),
@@ -836,6 +850,37 @@ fn tool_output(
})
}
fn render_subworker_stop_summary(summary: &SubWorkerStopSummary) -> String {
let tools = if summary.tool_counts.is_empty() {
"No tool calls".to_string()
} else {
summary
.tool_counts
.iter()
.map(|tool| format!("{} {}", tool.count, tool.name))
.collect::<Vec<_>>()
.join(", ")
};
let elapsed = format_elapsed(summary.elapsed_ms);
let changes = summary
.change_stat
.as_ref()
.map(|stat| format!("+{}/-{} Changes · ", stat.added, stat.deleted))
.unwrap_or_default();
format!("SubWorkerStop - done\n {tools}\n {changes}{elapsed}",)
}
fn format_elapsed(elapsed_ms: u64) -> String {
let seconds = elapsed_ms / 1_000;
let minutes = seconds / 60;
let seconds = seconds % 60;
if minutes > 0 {
format!("{minutes}m {seconds}s")
} else {
format!("{seconds}s")
}
}
fn definition<I: JsonSchema + 'static>(
operation: WorkerOperation,
control: Arc<dyn WorkerControlService>,
@@ -1240,6 +1285,47 @@ mod tests {
assert!(client.removals.lock().unwrap().is_empty());
}
#[test]
fn subworker_stop_output_is_compact_and_keeps_typed_evidence() {
let summary = SubWorkerStopSummary {
session_id: "session-1".to_string(),
display_name: "research".to_string(),
outcome: crate::spawn::registry::SubWorkerFinalOutcome::Done,
elapsed_ms: 78_000,
tool_counts: vec![
crate::spawn::registry::SubWorkerToolCount {
name: "Read".to_string(),
count: 26,
},
crate::spawn::registry::SubWorkerToolCount {
name: "Grep".to_string(),
count: 5,
},
],
change_stat: Some(crate::spawn::registry::SubWorkerChangeStat {
added: 215,
deleted: 148,
source: "tracked_write_edit_tools".to_string(),
}),
};
let response = WorkspaceResponse {
status: 200,
body: serde_json::to_string(&summary).unwrap(),
};
let output = tool_output(WorkerOperation::Stop, response).unwrap();
assert_eq!(
output.summary,
"SubWorkerStop - done\n 26 Read, 5 Grep\n +215/-148 Changes · 1m 18s"
);
assert_eq!(
serde_json::from_str::<SubWorkerStopSummary>(output.content.as_deref().unwrap())
.unwrap(),
summary
);
}
#[test]
fn worker_inputs_reject_paths_and_parent_traversal() {
assert!(authority_id("https://runtime.example", "runtime_id").is_err());
@@ -1781,6 +1781,7 @@ provider = "github"
assert!(request.contains("\"title\":\"HTTP ticket\""));
let response_body = serde_json::to_string(&TicketRef {
id: "01TEST".to_string(),
human_key: None,
slug: "http-ticket".to_string(),
status: ticket::TicketStatus::Open,
})
+157 -2
View File
@@ -1,9 +1,14 @@
use std::sync::{Arc, Mutex, MutexGuard};
use protocol::{Event, InFlightBlock, InFlightSnapshot, InFlightToolCallState};
use protocol::{
CommandEvent, CommandSnapshot, CommandStatus, CommandStream, CommandStreamSlice, Event,
InFlightBlock, InFlightSnapshot, InFlightToolCallState,
};
use session_store::{LoggedContentPart, LoggedItem};
use tokio::sync::broadcast;
const COMMAND_SNAPSHOT_STREAM_BYTES: usize = 32 * 1024;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct InFlightBlockId(u64);
@@ -17,6 +22,7 @@ pub struct InFlightEvents {
pub(crate) struct InFlightInner {
next_block_id: u64,
blocks: Vec<TrackedBlock>,
commands: Vec<CommandSnapshot>,
}
#[derive(Debug, Clone)]
@@ -46,6 +52,7 @@ impl InFlightEvents {
inner: Arc::new(Mutex::new(InFlightInner {
next_block_id: 1,
blocks: Vec::new(),
commands: Vec::new(),
})),
event_tx,
}
@@ -201,6 +208,15 @@ impl InFlightEvents {
f()
}
pub(crate) fn publish_command_event(&self, event: CommandEvent) {
self.lock().apply_command_event(&event);
let _ = self.event_tx.send(Event::Command { event });
}
pub(crate) fn replace_command_snapshot(&self, commands: Vec<CommandSnapshot>) {
self.lock().commands = commands;
}
pub(crate) fn clear(&self) {
let mut inner = self.lock();
inner.clear();
@@ -224,6 +240,92 @@ impl InFlightInner {
.find(|block| block.block_id() == block_id)
}
fn apply_command_event(&mut self, event: &CommandEvent) {
match event {
CommandEvent::Started {
command_id,
tool_call_id,
observed_at_ms,
} => {
self.commands
.retain(|command| command.command_id != *command_id);
self.commands.push(CommandSnapshot {
command_id: command_id.clone(),
tool_call_id: tool_call_id.clone(),
status: CommandStatus::Running,
started_at_ms: *observed_at_ms,
observed_at_ms: *observed_at_ms,
last_output_at_ms: None,
stdout: CommandStreamSlice::default(),
stderr: CommandStreamSlice::default(),
exit_code: None,
});
}
CommandEvent::Output {
command_id,
stream,
start_offset,
end_offset,
content,
observed_at_ms,
} => {
let command = match self
.commands
.iter_mut()
.find(|command| command.command_id == *command_id)
{
Some(command) => command,
None => {
self.commands.push(CommandSnapshot {
command_id: command_id.clone(),
tool_call_id: None,
status: CommandStatus::Running,
started_at_ms: *observed_at_ms,
observed_at_ms: *observed_at_ms,
last_output_at_ms: Some(*observed_at_ms),
stdout: CommandStreamSlice::default(),
stderr: CommandStreamSlice::default(),
exit_code: None,
});
self.commands.last_mut().expect("command was inserted")
}
};
command.observed_at_ms = *observed_at_ms;
command.last_output_at_ms = Some(*observed_at_ms);
let target = match stream {
CommandStream::Stdout => &mut command.stdout,
CommandStream::Stderr => &mut command.stderr,
};
if target.end_offset != *start_offset {
target.content.clear();
target.start_offset = *start_offset;
target.truncated = *start_offset > 0;
}
target.content.push_str(content);
target.end_offset = *end_offset;
if target.content.len() > COMMAND_SNAPSHOT_STREAM_BYTES {
let mut cut = target.content.len() - COMMAND_SNAPSHOT_STREAM_BYTES;
while cut < target.content.len() && !target.content.is_char_boundary(cut) {
cut += 1;
}
target.content.drain(..cut);
target.start_offset = target
.end_offset
.saturating_sub(target.content.len() as u64);
target.truncated = true;
}
}
CommandEvent::Terminal { command_id, .. } => {
// Terminal state is delivered as a live protocol event. It is
// no longer in-flight snapshot state, and removing it here
// also prevents queued output from an aborted turn from
// surviving the subsequent terminal event after `clear()`.
self.commands
.retain(|command| command.command_id != *command_id);
}
}
}
fn clear_for_committed_item(&mut self, item: &LoggedItem) {
match item {
LoggedItem::Message { role, content }
@@ -273,14 +375,16 @@ impl InFlightInner {
.iter()
.filter_map(TrackedBlock::to_snapshot_block)
.collect(),
commands: self.commands.clone(),
}
}
fn clear(&mut self) -> bool {
if self.blocks.is_empty() {
if self.blocks.is_empty() && self.commands.is_empty() {
false
} else {
self.blocks.clear();
self.commands.clear();
true
}
}
@@ -583,6 +687,57 @@ mod tests {
);
}
#[test]
fn command_events_are_bounded_and_recoverable_from_snapshot() {
let (event_tx, _) = broadcast::channel(16);
let mut rx = event_tx.subscribe();
let in_flight = InFlightEvents::new(event_tx);
in_flight.publish_command_event(CommandEvent::Started {
command_id: "command-1".into(),
tool_call_id: Some("tool-1".into()),
observed_at_ms: 100,
});
in_flight.publish_command_event(CommandEvent::Output {
command_id: "command-1".into(),
stream: CommandStream::Stdout,
start_offset: 0,
end_offset: 5,
content: "ready".into(),
observed_at_ms: 110,
});
let guard = in_flight.snapshot_guard();
let snapshot = snapshot_from_guard(&guard);
assert_eq!(snapshot.commands.len(), 1);
assert_eq!(snapshot.commands[0].tool_call_id.as_deref(), Some("tool-1"));
assert_eq!(snapshot.commands[0].stdout.content, "ready");
assert_eq!(snapshot.commands[0].status, CommandStatus::Running);
drop(guard);
assert!(matches!(
rx.try_recv().unwrap(),
Event::Command {
event: CommandEvent::Started { .. }
}
));
assert!(matches!(
rx.try_recv().unwrap(),
Event::Command {
event: CommandEvent::Output { .. }
}
));
in_flight.publish_command_event(CommandEvent::Terminal {
command_id: "command-1".into(),
status: CommandStatus::TimedOut,
exit_code: None,
stdout_end_offset: 5,
stderr_end_offset: 0,
observed_at_ms: 200,
});
let guard = in_flight.snapshot_guard();
assert!(snapshot_from_guard(&guard).commands.is_empty());
}
#[test]
fn clear_discards_uncommitted_blocks_without_protocol_event() {
let (event_tx, _) = broadcast::channel(16);
+38 -2
View File
@@ -17,7 +17,7 @@ use session_store::{LogEntry, SegmentId, SessionId, Store, StoreError, TraceEntr
use tokio::sync::broadcast;
use uuid::Uuid;
use crate::controller::wire_event_bridges_on_engine;
use crate::controller::{wire_event_bridges_on_engine, wire_workdir_command_events};
use crate::feature::FeatureRegistryBuilder;
use crate::in_flight::{InFlightEvents, snapshot_from_guard};
use crate::ipc::alerter::Alerter;
@@ -294,6 +294,8 @@ pub(crate) struct InternalWorkerSessionHandle {
last_error: Arc<Mutex<Option<String>>>,
child_registry: Option<Arc<SpawnedWorkerRegistry>>,
sink: SegmentLogSink,
#[cfg(test)]
fail_stop: Arc<std::sync::atomic::AtomicBool>,
}
impl InternalWorkerSessionHandle {
@@ -319,6 +321,9 @@ impl InternalWorkerSessionHandle {
#[cfg(test)]
pub(crate) fn publish_test_entry(&self, entry: LogEntry) {
self.store
.append(self.session_id, self.segment_id, &entry)
.expect("append test Internal Worker entry");
self.sink.publish(entry);
}
@@ -419,7 +424,23 @@ impl InternalWorkerSessionHandle {
}
}
#[cfg(test)]
pub(crate) fn force_status(&self, status: InternalWorkerSessionStatus) {
self.status
.store(status.encode(), std::sync::atomic::Ordering::Release);
}
#[cfg(test)]
pub(crate) fn force_stop_failure(&self) {
self.fail_stop
.store(true, std::sync::atomic::Ordering::Release);
}
pub(crate) async fn stop(&self) -> Result<(), InternalWorkerSessionError> {
#[cfg(test)]
if self.fail_stop.load(std::sync::atomic::Ordering::Acquire) {
return Err(InternalWorkerSessionError::Unavailable);
}
let prior = self.status.swap(
InternalWorkerSessionStatus::Stopping.encode(),
std::sync::atomic::Ordering::AcqRel,
@@ -555,6 +576,9 @@ pub(crate) async fn prepare_internal_worker_session(
spawn_internal_log_event_bridge(sink.clone(), event_tx.clone());
let alerter = Alerter::new(event_tx.clone());
let in_flight = InFlightEvents::new(event_tx.clone());
if let Some(session) = worker.workdir_session() {
wire_workdir_command_events(session, &in_flight);
}
let actor_in_flight = in_flight.clone();
worker.attach_alerter(alerter.clone());
worker.attach_event_tx(event_tx.clone());
@@ -582,6 +606,8 @@ pub(crate) async fn prepare_internal_worker_session(
last_error: last_error.clone(),
child_registry,
sink,
#[cfg(test)]
fail_stop: Arc::new(std::sync::atomic::AtomicBool::new(false)),
};
tokio::spawn(async move {
@@ -890,8 +916,17 @@ pub(crate) fn test_internal_worker_session(
let session_id = session_store::new_session_id();
let segment_id = session_store::new_segment_id();
let (command_tx, mut command_rx) = tokio::sync::mpsc::channel(1);
tokio::spawn(async move { while command_rx.recv().await.is_some() {} });
let (event_tx, _) = broadcast::channel(256);
let command_event_tx = event_tx.clone();
tokio::spawn(async move {
while let Some(command) = command_rx.recv().await {
if let InternalWorkerSessionCommand::Stop(done_tx) = command {
let _ = command_event_tx.send(Event::Shutdown);
let _ = done_tx.send(());
break;
}
}
});
let sink = SegmentLogSink::new();
spawn_internal_log_event_bridge(sink.clone(), event_tx.clone());
let handle = InternalWorkerSessionHandle {
@@ -909,6 +944,7 @@ pub(crate) fn test_internal_worker_session(
last_error: Arc::new(Mutex::new(None)),
child_registry: None,
sink,
fail_stop: Arc::new(std::sync::atomic::AtomicBool::new(false)),
};
(handle, event_tx)
}
+18 -11
View File
@@ -170,20 +170,27 @@ impl Tool for SubWorkerStopTool {
) -> Result<ToolOutput, ToolError> {
let input: NameInput = serde_json::from_str(input_json)
.map_err(|e| ToolError::InvalidArgument(format!("invalid SubWorkerStop input: {e}")))?;
if let Some(record) = self.registry.get_internal(&input.name) {
record.session.stop().await.map_err(|error| {
ToolError::ExecutionFailed(format!("stop `{}`: {error}", input.name))
})?;
self.registry
.remove_internal(&input.name)
.await
.map_err(|error| ToolError::ExecutionFailed(error.to_string()))?;
if let Some(summary) = self
.registry
.remove_internal(&input.name)
.await
.map_err(|error| ToolError::ExecutionFailed(error.to_string()))?
{
return Ok(ToolOutput {
summary: format!(
"stopped worker `{}` and reclaimed delegated scope",
input.name
"SubWorkerStop - done\n {} tool kind{}\n {}ms",
summary.tool_counts.len(),
if summary.tool_counts.len() == 1 {
""
} else {
"s"
},
summary.elapsed_ms,
),
content: Some(
serde_json::to_string(&summary)
.map_err(|error| ToolError::ExecutionFailed(error.to_string()))?,
),
content: None,
attachments: Vec::new(),
});
}
+296 -8
View File
@@ -7,17 +7,20 @@
//! Parent registry drop closes all session handles and synchronously returns delegated Write deny
//! rules to the parent scope.
use std::collections::HashSet;
use std::collections::{BTreeMap, HashSet};
use std::io;
use std::sync::{
Arc, Mutex,
atomic::{AtomicBool, AtomicU64, Ordering},
};
use std::time::Instant;
use serde::{Deserialize, Serialize};
use manifest::{Permission, ScopeRule, SharedScope};
use protocol::{Event, InternalWorkerKind, InternalWorkerRef, InternalWorkerSnapshot};
use session_store::{
WorkerMetadataStore, WorkerReclaimedChild, WorkerSpawnedChild, WorkerStoreError,
LoggedItem, WorkerMetadataStore, WorkerReclaimedChild, WorkerSpawnedChild, WorkerStoreError,
};
use tokio::sync::broadcast;
use tracing::warn;
@@ -27,6 +30,39 @@ use crate::internal_worker::{InternalWorkerSessionHandle, InternalWorkerVisibili
use crate::runtime::dir::{RuntimeDir, SpawnedWorkerRecord};
use crate::runtime::worker_allocation;
const STOP_SUMMARY_TOOL_LIMIT: usize = 16;
const STOP_SUMMARY_TOOL_NAME_LIMIT: usize = 64;
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub(crate) enum SubWorkerFinalOutcome {
Done,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub(crate) struct SubWorkerToolCount {
pub name: String,
pub count: u64,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub(crate) struct SubWorkerChangeStat {
pub added: u64,
pub deleted: u64,
pub source: String,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub(crate) struct SubWorkerStopSummary {
pub session_id: String,
pub display_name: String,
pub outcome: SubWorkerFinalOutcome,
pub elapsed_ms: u64,
pub tool_counts: Vec<SubWorkerToolCount>,
#[serde(skip_serializing_if = "Option::is_none")]
pub change_stat: Option<SubWorkerChangeStat>,
}
#[derive(Clone)]
pub(crate) struct InternalSpawnedWorkerRecord {
pub worker_name: String,
@@ -35,8 +71,13 @@ pub(crate) struct InternalSpawnedWorkerRecord {
#[cfg(test)]
pub installed_tools: Arc<[String]>,
pub session: InternalWorkerSessionHandle,
change_tracker: Option<tools::Tracker>,
started_at: Instant,
stop_lock: Arc<tokio::sync::Mutex<()>>,
scope_reclaimed: Arc<AtomicBool>,
protocol_revision: Arc<AtomicU64>,
protocol_emit_lock: Arc<Mutex<()>>,
protocol_terminal: Arc<AtomicBool>,
forwarding_started: Arc<AtomicBool>,
}
@@ -47,6 +88,7 @@ impl InternalSpawnedWorkerRecord {
workdir_delegation: WorkdirDelegation,
#[cfg(test)] installed_tools: Vec<String>,
session: InternalWorkerSessionHandle,
change_tracker: Option<tools::Tracker>,
) -> Self {
Self {
worker_name,
@@ -55,12 +97,64 @@ impl InternalSpawnedWorkerRecord {
#[cfg(test)]
installed_tools: installed_tools.into(),
session,
change_tracker,
started_at: Instant::now(),
stop_lock: Arc::new(tokio::sync::Mutex::new(())),
scope_reclaimed: Arc::new(AtomicBool::new(false)),
protocol_revision: Arc::new(AtomicU64::new(0)),
protocol_emit_lock: Arc::new(Mutex::new(())),
protocol_terminal: Arc::new(AtomicBool::new(false)),
forwarding_started: Arc::new(AtomicBool::new(false)),
}
}
fn stop_summary(&self) -> SubWorkerStopSummary {
let mut counts = BTreeMap::<String, u64>::new();
for entry in self.session.entries() {
if let session_store::LogEntry::AssistantItem {
item: LoggedItem::ToolCall { name, .. },
..
} = entry
{
let count = counts.entry(bounded_tool_name(&name)).or_default();
*count = count.saturating_add(1);
}
}
let mut tool_counts = counts
.into_iter()
.map(|(name, count)| SubWorkerToolCount { name, count })
.collect::<Vec<_>>();
tool_counts.sort_by(|left, right| {
right
.count
.cmp(&left.count)
.then_with(|| left.name.cmp(&right.name))
});
tool_counts.truncate(STOP_SUMMARY_TOOL_LIMIT);
let change_stat = self.change_tracker.as_ref().and_then(|tracker| {
let stat = tracker.change_stat();
(stat.added > 0 || stat.deleted > 0).then(|| SubWorkerChangeStat {
added: stat.added,
deleted: stat.deleted,
source: "tracked_write_edit_tools".to_string(),
})
});
SubWorkerStopSummary {
session_id: self.session.session_id_string(),
display_name: self.worker_name.clone(),
outcome: SubWorkerFinalOutcome::Done,
elapsed_ms: self
.started_at
.elapsed()
.as_millis()
.min(u128::from(u64::MAX)) as u64,
tool_counts,
change_stat,
}
}
fn claim_scope_reclaim(&self) -> bool {
!self.scope_reclaimed.swap(true, Ordering::AcqRel)
}
@@ -277,12 +371,20 @@ impl SpawnedWorkerRegistry {
};
let worker = record.protocol_ref(Some(parent_session_id));
let protocol_revision = record.protocol_revision.clone();
let protocol_emit_lock = record.protocol_emit_lock.clone();
let protocol_terminal = record.protocol_terminal.clone();
let mut child_rx = record.session.subscribe_events();
tokio::spawn(async move {
loop {
match child_rx.recv().await {
Ok(event) => {
let shutdown = matches!(event, Event::Shutdown);
let _emit_guard = protocol_emit_lock
.lock()
.unwrap_or_else(|error| error.into_inner());
if protocol_terminal.load(Ordering::Acquire) {
break;
}
let revision = protocol_revision.fetch_add(1, Ordering::AcqRel) + 1;
let _ = parent_tx.send(Event::InternalWorker {
worker: worker.clone(),
@@ -294,6 +396,12 @@ impl SpawnedWorkerRegistry {
}
}
Err(broadcast::error::RecvError::Lagged(skipped)) => {
let _emit_guard = protocol_emit_lock
.lock()
.unwrap_or_else(|error| error.into_inner());
if protocol_terminal.load(Ordering::Acquire) {
break;
}
let revision = protocol_revision.fetch_add(1, Ordering::AcqRel) + 1;
let _ = parent_tx.send(Event::InternalWorker {
worker: worker.clone(),
@@ -385,13 +493,34 @@ impl SpawnedWorkerRegistry {
result
}
/// Stop one direct Internal SubWorker and discard its registry/scope state.
///
/// The child actor must acknowledge its stop before the registry is removed.
/// After scope reclamation and removal, `InternalWorkerRemoved` is published
/// exactly once as the parent-stream terminal fence. Callers only receive
/// `Done` after all authoritative cleanup succeeds.
pub(crate) async fn remove_internal(
&self,
worker_name: &str,
) -> io::Result<Option<InternalSpawnedWorkerRecord>> {
if let Some(record) = self.get_internal(worker_name) {
self.reclaim_record_scope(&record)?;
) -> io::Result<Option<SubWorkerStopSummary>> {
let Some(record) = self.get_internal(worker_name) else {
return Ok(None);
};
let _stop_guard = record.stop_lock.lock().await;
let still_registered = self.get_internal(worker_name).is_some_and(|current| {
current.session.session_id_string() == record.session.session_id_string()
});
if !still_registered {
return Ok(None);
}
record
.session
.stop()
.await
.map_err(|error| io::Error::other(error.to_string()))?;
let summary = record.stop_summary();
self.reclaim_record_scope(&record)?;
let removed =
{
let mut records = self.internal_records.lock().map_err(|_| {
@@ -402,14 +531,41 @@ impl SpawnedWorkerRegistry {
})?;
let removed = records
.iter()
.position(|record| record.worker_name == worker_name)
.position(|candidate| {
candidate.worker_name == worker_name
&& candidate.session.session_id_string()
== record.session.session_id_string()
})
.map(|index| records.remove(index));
if removed.is_some() {
names.remove(worker_name);
}
removed
};
Ok(removed)
if removed.is_some() {
self.publish_internal_removal(&record);
}
Ok(removed.map(|_| summary))
}
fn publish_internal_removal(&self, record: &InternalSpawnedWorkerRecord) {
if record.session.visibility() != InternalWorkerVisibility::ParentClient {
return;
}
let Some((parent_tx, parent_session_id)) = self.parent_protocol.lock().unwrap().clone()
else {
return;
};
let _emit_guard = record
.protocol_emit_lock
.lock()
.unwrap_or_else(|error| error.into_inner());
record.protocol_terminal.store(true, Ordering::Release);
let revision = record.protocol_revision.fetch_add(1, Ordering::AcqRel) + 1;
let _ = parent_tx.send(Event::InternalWorkerRemoved {
worker: record.protocol_ref(Some(parent_session_id)),
revision,
});
}
}
@@ -508,6 +664,17 @@ fn record_from_worker_state(child: &WorkerSpawnedChild) -> io::Result<SpawnedWor
})
}
fn bounded_tool_name(name: &str) -> String {
let mut bounded = name
.chars()
.take(STOP_SUMMARY_TOOL_NAME_LIMIT)
.collect::<String>();
if name.chars().count() > STOP_SUMMARY_TOOL_NAME_LIMIT {
bounded.push('…');
}
bounded
}
fn store_error_to_io(error: WorkerStoreError) -> io::Error {
io::Error::other(error)
}
@@ -520,7 +687,7 @@ mod tests {
use session_store::LogEntry;
use super::*;
use crate::internal_worker::test_internal_worker_session;
use crate::internal_worker::{InternalWorkerSessionStatus, test_internal_worker_session};
fn registry() -> Arc<SpawnedWorkerRegistry> {
let scope = Scope::from_config(&ScopeConfig {
@@ -577,6 +744,7 @@ mod tests {
delegation,
Vec::new(),
session,
None,
),
sender,
)
@@ -669,4 +837,124 @@ mod tests {
);
assert!(registry.internal_worker_snapshots().is_empty());
}
fn install_record(registry: &SpawnedWorkerRegistry, record: InternalSpawnedWorkerRecord) {
registry
.internal_names
.lock()
.unwrap()
.insert(record.worker_name.clone());
registry.internal_records.lock().unwrap().push(record);
}
#[tokio::test]
async fn stop_removes_internal_worker_and_returns_bounded_summary() {
let registry = registry();
let (parent_tx, mut parent_rx) = broadcast::channel(32);
registry.attach_parent_protocol(parent_tx, "parent-session".into());
let tracker = tools::Tracker::new();
tracker.record_change(12, 4);
let (mut record, _events) = record("child", InternalWorkerVisibility::ParentClient).await;
record.change_tracker = Some(tracker);
for (index, name) in ["Read", "Read", "Grep"].into_iter().enumerate() {
record.session.publish_test_entry(LogEntry::AssistantItem {
ts: index as u64,
item: LoggedItem::ToolCall {
call_id: format!("call-{index}"),
name: name.to_string(),
arguments: "{}".to_string(),
},
});
}
registry.start_protocol_forwarding(record.clone());
install_record(&registry, record);
let summary = registry.remove_internal("child").await.unwrap().unwrap();
assert_eq!(summary.display_name, "child");
assert_eq!(summary.outcome, SubWorkerFinalOutcome::Done);
assert_eq!(
summary.tool_counts,
vec![
SubWorkerToolCount {
name: "Read".to_string(),
count: 2,
},
SubWorkerToolCount {
name: "Grep".to_string(),
count: 1,
},
]
);
assert_eq!(
summary.change_stat,
Some(SubWorkerChangeStat {
added: 12,
deleted: 4,
source: "tracked_write_edit_tools".to_string(),
})
);
assert!(registry.get_internal("child").is_none());
let terminal_revision = loop {
if let Event::InternalWorkerRemoved { worker, revision } =
parent_rx.recv().await.unwrap()
{
assert_eq!(worker.session_id, summary.session_id);
assert!(revision > 0);
break revision;
}
};
assert!(registry.remove_internal("child").await.unwrap().is_none());
while let Ok(Ok(event)) =
tokio::time::timeout(Duration::from_millis(20), parent_rx.recv()).await
{
assert!(!matches!(event, Event::InternalWorkerRemoved { .. }));
if let Event::InternalWorker { revision, .. } = event {
assert!(revision > terminal_revision);
}
}
}
#[tokio::test]
async fn running_worker_is_stopped_before_removal() {
let registry = registry();
let (record, _events) = record("running", InternalWorkerVisibility::ParentClient).await;
record
.session
.force_status(InternalWorkerSessionStatus::Running);
install_record(&registry, record);
let summary = registry.remove_internal("running").await.unwrap().unwrap();
assert_eq!(summary.outcome, SubWorkerFinalOutcome::Done);
assert!(registry.get_internal("running").is_none());
}
#[tokio::test]
async fn stop_failure_keeps_registry_and_emits_no_removal() {
let registry = registry();
let (parent_tx, mut parent_rx) = broadcast::channel(8);
registry.attach_parent_protocol(parent_tx, "parent-session".into());
let (record, _events) = record("child", InternalWorkerVisibility::ParentClient).await;
record.session.force_stop_failure();
install_record(&registry, record);
let error = registry.remove_internal("child").await.unwrap_err();
assert!(error.to_string().contains("unavailable"));
assert!(registry.get_internal("child").is_some());
assert!(matches!(
parent_rx.try_recv(),
Err(broadcast::error::TryRecvError::Empty)
));
}
#[tokio::test]
async fn read_only_summary_omits_unavailable_change_stat() {
let tracker = tools::Tracker::new();
let (mut record, _events) = record("reader", InternalWorkerVisibility::ParentClient).await;
record.change_tracker = Some(tracker);
assert_eq!(record.stop_summary().change_stat, None);
}
}
+2
View File
@@ -481,6 +481,7 @@ impl Tool for SubWorkerSpawnTool {
.map_err(|error| {
ToolError::ExecutionFailed(format!("install Internal Worker features: {error}"))
})?;
let child_change_tracker = child.tracker().cloned();
#[cfg(test)]
let installed_tools = child
.engine()
@@ -587,6 +588,7 @@ impl Tool for SubWorkerSpawnTool {
#[cfg(test)]
installed_tools,
session.clone(),
child_change_tracker,
);
if let Err(error) = name_reservation.commit(record) {
let _ = session.stop().await;
+178 -2
View File
@@ -12,8 +12,8 @@ use llm_engine::tool::{Tool, ToolDefinition, ToolError, ToolMeta, ToolOutput};
use session_store::{CombinedStore, FsWorkerStore};
use session_store::{FsStore, LogEntry};
use workdir::{
CommandRequest, LocalWorkdirSession, Workdir, WorkdirError, WorkdirSessionCapabilities,
WorkdirSessionHandle,
CommandOutputRequest, CommandRequest, LocalWorkdirSession, Workdir, WorkdirError,
WorkdirSessionCapabilities, WorkdirSessionHandle,
};
use worker::{
@@ -232,6 +232,7 @@ async fn shutdown_closes_bound_workdir_session() {
command: "sleep 30".to_owned(),
timeout_secs: 60,
output_limit: 1024,
tool_call_id: None,
})
.await
.unwrap();
@@ -253,6 +254,180 @@ async fn shutdown_closes_bound_workdir_session() {
));
}
#[tokio::test]
async fn controller_projects_workdir_command_events_and_snapshot_state() {
let (mut worker, pwd) = make_worker_with_pwd(MockClient::new(simple_text_events())).await;
let session: WorkdirSessionHandle = Arc::new(LocalWorkdirSession::materialized_bound(
Workdir::new("controller-command-observation-workdir"),
pwd.clone(),
pwd,
worker.scope().clone(),
WorkdirSessionCapabilities::ALL,
));
worker.bind_workdir_session(Some(Arc::clone(&session)));
let handle = spawn_controller(worker).await;
let mut events = handle.subscribe();
let command = session
.start_command(CommandRequest {
command: "printf ready; sleep 0.3; printf done".to_owned(),
timeout_secs: 5,
output_limit: 1024,
tool_call_id: Some("tool-command-1".into()),
})
.await
.unwrap();
let mut saw_started = false;
let mut saw_output = false;
while !saw_output {
let event = tokio::time::timeout(std::time::Duration::from_secs(2), events.recv())
.await
.expect("command event should arrive")
.unwrap();
match event {
Event::Command {
event:
protocol::CommandEvent::Started {
command_id,
tool_call_id,
..
},
} => {
assert_eq!(command_id, command.0);
assert_eq!(tool_call_id.as_deref(), Some("tool-command-1"));
saw_started = true;
}
Event::Command {
event:
protocol::CommandEvent::Output {
command_id,
stream: protocol::CommandStream::Stdout,
content,
..
},
} if command_id == command.0 && content.contains("ready") => saw_output = true,
_ => {}
}
}
assert!(saw_started);
let Event::Snapshot { in_flight, .. } = handle.snapshot_event() else {
panic!("worker snapshot expected");
};
assert_eq!(in_flight.commands.len(), 1);
assert_eq!(in_flight.commands[0].command_id, command.0);
assert_eq!(in_flight.commands[0].stdout.content, "ready");
assert_eq!(
in_flight.commands[0].status,
protocol::CommandStatus::Running
);
let saw_terminal = drain_until(&mut events, std::time::Duration::from_secs(2), |event| {
matches!(
event,
Event::Command {
event: protocol::CommandEvent::Terminal {
command_id,
status: protocol::CommandStatus::Completed,
exit_code: Some(0),
..
}
} if command_id == &command.0
)
})
.await;
assert!(saw_terminal, "completed command event should arrive");
let output = session
.command_output(CommandOutputRequest {
handle: command,
cursor: 0,
limit: 1024,
wait: true,
})
.await
.unwrap();
assert_eq!(output.status, workdir::CommandStatus::Completed);
let (entries, _) = handle.sink.subscribe_with_snapshot();
let durable_history = serde_json::to_string(&entries).unwrap();
assert!(
!durable_history.contains("ready") && !durable_history.contains("done"),
"operational command chunks must not be appended to Worker history: {durable_history}"
);
handle.send(Method::Shutdown).await.unwrap();
}
#[tokio::test]
async fn controller_refreshes_command_snapshot_after_high_output_provider_lag() {
let (mut worker, pwd) = make_worker_with_pwd(MockClient::new(simple_text_events())).await;
let session: WorkdirSessionHandle = Arc::new(LocalWorkdirSession::materialized_bound(
Workdir::new("controller-command-lag-recovery-workdir"),
pwd.clone(),
pwd,
worker.scope().clone(),
WorkdirSessionCapabilities::ALL,
));
worker.bind_workdir_session(Some(Arc::clone(&session)));
let handle = spawn_controller(worker).await;
// Local command telemetry uses 8 KiB chunks and a 256-event channel. One
// synchronous file-poll burst with 300 chunks deterministically makes the
// worker-side receiver observe `Lagged` before this command terminates.
let command = session
.start_command(CommandRequest {
command: "dd if=/dev/zero bs=8192 count=300 2>/dev/null | tr '\\0' x; sleep 5"
.to_owned(),
timeout_secs: 10,
output_limit: 1024,
tool_call_id: Some("tool-high-output".into()),
})
.await
.unwrap();
let expected_end_offset = 300_u64 * 8192;
let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(3);
let recovered = loop {
let Event::Snapshot { in_flight, .. } = handle.snapshot_event() else {
panic!("worker snapshot expected");
};
if let Some(snapshot) = in_flight
.commands
.iter()
.find(|snapshot| snapshot.command_id == command.0)
&& snapshot.stdout.end_offset >= expected_end_offset
{
break snapshot.clone();
}
assert!(
tokio::time::Instant::now() < deadline,
"timed out waiting for lag recovery snapshot"
);
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
};
assert_eq!(recovered.tool_call_id.as_deref(), Some("tool-high-output"));
assert_eq!(recovered.status, protocol::CommandStatus::Running);
assert!(recovered.stdout.truncated);
assert!(recovered.stdout.start_offset > 0);
assert_eq!(recovered.stdout.end_offset, expected_end_offset);
assert!(recovered.stdout.content.len() <= 32 * 1024);
assert!(recovered.stdout.content.bytes().all(|byte| byte == b'x'));
session.cancel_command(command.clone()).await.unwrap();
let output = session
.command_output(CommandOutputRequest {
handle: command,
cursor: 0,
limit: 1024,
wait: true,
})
.await
.unwrap();
assert_eq!(output.status, workdir::CommandStatus::Cancelled);
handle.send(Method::Shutdown).await.unwrap();
}
#[tokio::test]
async fn controller_startup_failure_closes_bound_workdir_session() {
let (mut worker, pwd) = make_worker_with_pwd(MockClient::new(simple_text_events())).await;
@@ -279,6 +454,7 @@ async fn controller_startup_failure_closes_bound_workdir_session() {
command: "printf unreachable".to_owned(),
timeout_secs: 5,
output_limit: 1024,
tool_call_id: None,
})
.await,
Err(WorkdirError::Unavailable(_))
+1
View File
@@ -35,6 +35,7 @@ ticket.workspace = true
memory.workspace = true
merge-request.workspace = true
tokio = { workspace = true, features = ["fs", "macros", "net", "rt-multi-thread", "sync", "time"] }
tower.workspace = true
tokio-tungstenite.workspace = true
worker.workspace = true
workdir = { workspace = true, features = ["http-client"] }
+41 -6
View File
@@ -2843,12 +2843,6 @@ mod tests {
write_ticket(dir.path(), "00000000001J5", "Second ticket", "planning");
write_ticket(dir.path(), "00000000001J6", "Third ticket", "planning");
let db_path = dir.path().join("workspace.db");
SqliteTicketBackend::open(&db_path, "workspace-test")
.unwrap()
.import_from_local_backend(&ticket::LocalTicketBackend::new(
dir.path().join(".yoi/tickets"),
))
.unwrap();
let store = SqliteWorkspaceStore::open(&db_path).unwrap();
store
.upsert_workspace(&WorkspaceRecord {
@@ -2861,6 +2855,27 @@ mod tests {
})
.await
.unwrap();
SqliteTicketBackend::open(&db_path, "workspace-test")
.unwrap()
.import_from_local_backend(&ticket::LocalTicketBackend::new(
dir.path().join(".yoi/tickets"),
))
.unwrap();
rusqlite::Connection::open(&db_path)
.unwrap()
.execute_batch(
r#"
INSERT INTO workspace_resource_human_keys (
workspace_id, resource_kind, resource_id, sequence, human_key, allocated_at
) VALUES
('workspace-test', 'ticket', '00000000001J2', 1, 'T-1', '2026-01-01T00:00:00Z'),
('workspace-test', 'ticket', '00000000001J5', 2, 'T-2', '2026-01-01T00:00:00Z'),
('workspace-test', 'ticket', '00000000001J6', 3, 'T-3', '2026-01-01T00:00:00Z');
INSERT INTO workspace_resource_human_key_counters (workspace_id, resource_kind, next_sequence)
VALUES ('workspace-test', 'ticket', 4);
"#,
)
.unwrap();
store
.upsert_objective(&ObjectiveRecord {
workspace_id: "workspace-test".to_string(),
@@ -3216,6 +3231,26 @@ mod tests {
})
.await
.unwrap();
rusqlite::Connection::open(&db_path)
.unwrap()
.execute_batch(
r#"
INSERT INTO typed_tickets (
workspace_id, ticket_id, slug, title, status, kind, priority, body,
workflow_state, workflow_state_explicit
) VALUES
('workspace-test', '00000000001J2', 'ticket-j2', 'Ticket J2', 'open', 'task', 'normal', '', 'planning', 1),
('workspace-test', '00000000001J3', 'ticket-j3', 'Ticket J3', 'open', 'task', 'normal', '', 'planning', 1);
INSERT INTO workspace_resource_human_keys (
workspace_id, resource_kind, resource_id, sequence, human_key, allocated_at
) VALUES
('workspace-test', 'ticket', '00000000001J2', 1, 'T-1', '2026-01-01T00:00:00Z'),
('workspace-test', 'ticket', '00000000001J3', 2, 'T-2', '2026-01-01T00:00:00Z');
INSERT INTO workspace_resource_human_key_counters (workspace_id, resource_kind, next_sequence)
VALUES ('workspace-test', 'ticket', 3);
"#,
)
.unwrap();
let authority = SqliteWorkspaceAuthority::new(&db_path, "workspace-test").unwrap();
let created = authority
+8
View File
@@ -2447,6 +2447,8 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime {
#[derive(Clone)]
pub struct RemoteRuntimeConfig {
pub runtime_id: String,
/// Explicit Workspace assignment granted by Server authority.
pub workspace_id: Option<String>,
pub display_name: String,
pub base_url: String,
pub bearer_token: Option<String>,
@@ -2489,6 +2491,7 @@ impl RemoteRuntimeConfig {
) -> Self {
Self {
runtime_id: runtime_id.into(),
workspace_id: None,
display_name: display_name.into(),
base_url: base_url.into(),
bearer_token,
@@ -2501,6 +2504,11 @@ impl RemoteRuntimeConfig {
}
}
pub fn with_workspace_id(mut self, workspace_id: impl Into<String>) -> Self {
self.workspace_id = Some(workspace_id.into());
self
}
pub fn with_cached_capabilities(mut self, capabilities: RuntimeCapabilitySummary) -> Self {
self.cached_capabilities = capabilities;
self
+9 -1
View File
@@ -27,6 +27,7 @@ pub mod server;
pub mod skills;
pub mod store;
pub mod worker_source;
pub mod workspace_catalog;
mod workspace_subscription;
pub use authority::{
@@ -45,8 +46,15 @@ pub use repositories::{
ConfiguredRepository, GitCommitSummary, GitRemoteSummary, GitRepositorySummary,
RepositoryLogRead, RepositoryRegistryReader, RepositorySummary,
};
pub use server::{AuthConfig, ServerConfig, WorkspaceApi, build_router, serve};
pub use server::{
AuthConfig, ServerConfig, WorkspaceApi, WorkspaceServerApi, build_router,
build_workspace_server_router, serve, serve_workspace_catalog,
};
pub use store::{ControlPlaneStore, SqliteWorkspaceStore, WorkspaceRecord};
pub use workspace_catalog::{
InitialRepositoryIntent, WorkspaceCatalogService, WorkspaceCreateRequest,
WorkspaceCreateResponse,
};
use worker_runtime::identity::RuntimeWorkerRef;
+81 -54
View File
@@ -9,10 +9,11 @@ use serde::{Deserialize, Serialize};
use tokio::net::TcpListener;
use worker_runtime::auth::{RuntimeIdentityMaterial, decode_public_key};
use yoi_workspace_server::hosts::{RemoteRuntimeAuthConfig, RemoteRuntimeConfig};
use yoi_workspace_server::store::{RepositoryRecord, SqliteWorkspaceStore, TrustedRuntimeRecord};
use yoi_workspace_server::store::{SqliteWorkspaceStore, TrustedRuntimeRecord};
use yoi_workspace_server::{
BackendRuntimesConfigFile, ControlPlaneStore, ServerConfig, WORKSPACE_BACKEND_CONFIG_TEMPLATE,
WorkspaceBackendConfigFile, WorkspaceIdentity, WorkspaceRecord, serve,
BackendRuntimesConfigFile, ControlPlaneStore, InitialRepositoryIntent, ServerConfig,
WORKSPACE_BACKEND_CONFIG_TEMPLATE, WorkspaceBackendConfigFile, WorkspaceCatalogService,
WorkspaceCreateRequest, WorkspaceIdentity, WorkspaceRecord, serve_workspace_catalog,
};
#[derive(Debug)]
@@ -152,30 +153,21 @@ async fn run_init_with_database_path(
if let Some(parent) = database_path.parent() {
tokio::fs::create_dir_all(parent).await?;
}
let store = SqliteWorkspaceStore::open(&database_path)?;
store
.upsert_workspace(&WorkspaceRecord {
workspace_id: identity.workspace_id.clone(),
owner_account_id: None,
let store = Arc::new(SqliteWorkspaceStore::open(&database_path)?);
let service = WorkspaceCatalogService::new(store);
service.create_with_workspace_id(
WorkspaceCreateRequest {
operation_key: format!("cli-init:{}", identity.workspace_id),
display_name: identity.display_name.clone(),
state: "active".to_string(),
created_at: identity.created_at.clone(),
updated_at: identity.created_at.clone(),
})
.await?;
store.upsert_repository(&RepositoryRecord {
workspace_id: identity.workspace_id.clone(),
repository_id: "main".to_string(),
name: "Main repository".to_string(),
kind: "git".to_string(),
provider: Some("git".to_string()),
uri: options.workspace.display().to_string(),
default_ref: Some("HEAD".to_string()),
auth_ref_kind: None,
auth_ref_key: None,
created_at: identity.created_at.clone(),
updated_at: identity.created_at.clone(),
})?;
repository: InitialRepositoryIntent {
uri: options.workspace.display().to_string(),
display_name: Some("Main repository".to_string()),
default_ref: Some("HEAD".to_string()),
},
},
None,
Some(identity.workspace_id.clone()),
)?;
eprintln!(
"yoi-server: initialized workspace `{}` ({}) in server DB `{}`",
@@ -358,6 +350,7 @@ fn run_trust_runtime_command(args: Vec<String>) -> Result<(), Box<dyn std::error
match subcommand.as_str() {
"add" => {
let mut runtime_id = None;
let mut workspace_id = None;
let mut base_url = None;
let mut public_key = None;
let mut display_name = None;
@@ -368,6 +361,9 @@ fn run_trust_runtime_command(args: Vec<String>) -> Result<(), Box<dyn std::error
"--runtime-id" => {
runtime_id = Some(take_value(&flag, inline_value, &mut args)?)
}
"--workspace-id" => {
workspace_id = Some(take_value(&flag, inline_value, &mut args)?)
}
"--base-url" | "--endpoint" => {
base_url = Some(take_value(&flag, inline_value, &mut args)?)
}
@@ -390,15 +386,39 @@ fn run_trust_runtime_command(args: Vec<String>) -> Result<(), Box<dyn std::error
}
let runtime_id = runtime_id
.ok_or_else(|| CliError("trust-runtime add requires --runtime-id".to_string()))?;
let workspace_id = workspace_id
.ok_or_else(|| CliError("trust-runtime add requires --workspace-id".to_string()))?;
if !store
.list_workspaces()?
.iter()
.any(|workspace| workspace.workspace_id == workspace_id)
{
return Err(Box::new(CliError(format!(
"Workspace `{workspace_id}` is not registered"
))));
}
let base_url = base_url
.ok_or_else(|| CliError("trust-runtime add requires --base-url".to_string()))?;
let public_key = public_key
.ok_or_else(|| CliError("trust-runtime add requires --public-key".to_string()))?;
decode_public_key(&public_key)?;
ensure_trusted_runtime_replace_allowed(&store, &runtime_id, replace)?;
if let Some(existing) = store
.list_trusted_runtimes(true)?
.into_iter()
.find(|runtime| runtime.runtime_id == runtime_id)
{
if existing.workspace_id.as_deref() != Some(workspace_id.as_str()) {
return Err(Box::new(CliError(format!(
"runtime `{runtime_id}` is already assigned to Workspace `{}` and cannot be reparented",
existing.workspace_id.as_deref().unwrap_or("unassigned")
))));
}
}
let now = Utc::now().to_rfc3339();
store.upsert_trusted_runtime(&TrustedRuntimeRecord {
runtime_id: runtime_id.clone(),
workspace_id: Some(workspace_id.clone()),
display_name: display_name.unwrap_or_else(|| runtime_id.clone()),
base_url,
public_key,
@@ -437,8 +457,9 @@ fn run_trust_runtime_command(args: Vec<String>) -> Result<(), Box<dyn std::error
} else {
for runtime in records {
println!(
"runtime_id={} base_url={} public_key={} revoked_at={}",
"runtime_id={} workspace_id={} base_url={} public_key={} revoked_at={}",
runtime.runtime_id,
runtime.workspace_id.unwrap_or_default(),
runtime.base_url,
runtime.public_key,
runtime.revoked_at.unwrap_or_default()
@@ -582,12 +603,28 @@ async fn run_serve(options: ServeOptions) -> Result<(), Box<dyn std::error::Erro
}
let store = Arc::new(SqliteWorkspaceStore::open(&database_path)?);
let workspace = select_serve_workspace(store.as_ref())?;
let workspace_root = infer_workspace_root_from_repositories(store.as_ref(), &workspace)?;
let identity = WorkspaceIdentity {
workspace_id: workspace.workspace_id.clone(),
created_at: workspace.created_at.clone(),
display_name: workspace.display_name.clone(),
let workspaces = store.list_workspaces()?;
let (identity, workspace_root) = if let Some(workspace) = workspaces.first() {
(
WorkspaceIdentity {
workspace_id: workspace.workspace_id.clone(),
created_at: workspace.created_at.clone(),
display_name: workspace.display_name.clone(),
},
infer_workspace_root_from_repositories(store.as_ref(), workspace)?,
)
} else {
(
WorkspaceIdentity {
workspace_id: "00000000-0000-0000-0000-000000000000".to_string(),
created_at: Utc::now().to_rfc3339(),
display_name: "Server bootstrap".to_string(),
},
database_path
.parent()
.ok_or_else(|| CliError("server database path has no parent".to_string()))?
.to_path_buf(),
)
};
let runtime_config = BackendRuntimesConfigFile::load_default()?;
let mut resolved = WorkspaceBackendConfigFile::default().resolve_with_runtime_config(
@@ -601,6 +638,7 @@ async fn run_serve(options: ServeOptions) -> Result<(), Box<dyn std::error::Erro
if let Some(listen) = options.listen {
resolved = resolved.with_listen(listen);
}
resolved.server.allow_local_workspace_bootstrap = resolved.listen.ip().is_loopback();
let listener = TcpListener::bind(resolved.listen).await?;
let local_addr = listener.local_addr()?;
@@ -608,12 +646,12 @@ async fn run_serve(options: ServeOptions) -> Result<(), Box<dyn std::error::Erro
resolved = resolved.with_backend_base_url(format!("http://{local_addr}"));
}
eprintln!(
"yoi-server: serving workspace `{}` from server DB `{}` on http://{}",
workspace.workspace_id,
"yoi-server: serving {} workspace(s) from server DB `{}` on http://{}",
workspaces.len(),
database_path.display(),
local_addr
);
serve(resolved.server, store, listener).await?;
serve_workspace_catalog(resolved.server, store, listener).await?;
Ok(())
}
@@ -630,6 +668,9 @@ fn append_trusted_runtime_sources(
return Ok(());
};
for runtime in store.list_trusted_runtimes(false)? {
let Some(workspace_id) = runtime.workspace_id.clone() else {
continue;
};
let auth = RemoteRuntimeAuthConfig {
server_id: server_identity.identity.identity_id.clone(),
server_private_key: server_identity.identity.private_key.clone(),
@@ -640,6 +681,7 @@ fn append_trusted_runtime_sources(
runtime.base_url,
None,
)
.with_workspace_id(workspace_id)
.with_auth(auth);
remote_runtime_sources.retain(|existing| existing.runtime_id != runtime.runtime_id);
remote_runtime_sources.push(remote);
@@ -647,23 +689,6 @@ fn append_trusted_runtime_sources(
Ok(())
}
fn select_serve_workspace(store: &SqliteWorkspaceStore) -> Result<WorkspaceRecord, CliError> {
let workspaces = store
.list_workspaces()
.map_err(|error| CliError(format!("failed to list workspaces from server DB: {error}")))?;
match workspaces.as_slice() {
[] => Err(CliError(
"server DB has no workspace records; run `yoi-server init --workspace <PATH>`"
.to_string(),
)),
[workspace] => Ok(workspace.clone()),
_ => Err(CliError(format!(
"server DB contains {} workspaces; serve workspace selection is not implemented yet",
workspaces.len()
))),
}
}
fn infer_workspace_root_from_repositories(
store: &SqliteWorkspaceStore,
workspace: &WorkspaceRecord,
@@ -914,7 +939,7 @@ fn parse_listen(value: &str) -> Result<SocketAddr, CliError> {
fn print_help() {
println!(
"yoi-server\n\nUsage:\n yoi-server init [OPTIONS]\n yoi-server config <COMMAND> [OPTIONS]\n yoi-server identity init --server-id <SERVER_ID> [--replace]\n yoi-server identity show [--json]\n yoi-server trust-runtime add --runtime-id <RUNTIME_ID> --base-url <URL> --public-key <KEY> [--display-name <NAME>] [--replace]\n yoi-server trust-runtime list [--json] [--include-revoked]\n yoi-server trust-runtime revoke --runtime-id <RUNTIME_ID>\n yoi-server skills <COMMAND> [OPTIONS]\n yoi-server migrate --dry-run [--database <PATH>]
"yoi-server\n\nUsage:\n yoi-server init [OPTIONS]\n yoi-server config <COMMAND> [OPTIONS]\n yoi-server identity init --server-id <SERVER_ID> [--replace]\n yoi-server identity show [--json]\n yoi-server trust-runtime add --runtime-id <RUNTIME_ID> --workspace-id <WORKSPACE_ID> --base-url <URL> --public-key <KEY> [--display-name <NAME>] [--replace]\n yoi-server trust-runtime list [--json] [--include-revoked]\n yoi-server trust-runtime revoke --runtime-id <RUNTIME_ID>\n yoi-server skills <COMMAND> [OPTIONS]\n yoi-server migrate --dry-run [--database <PATH>]
yoi-server serve [OPTIONS]\n\nOptions:\n -h, --help Print help"
);
}
@@ -1038,6 +1063,7 @@ mod tests {
store
.upsert_trusted_runtime(&TrustedRuntimeRecord {
runtime_id: "runtime-a".to_string(),
workspace_id: None,
display_name: "Runtime A".to_string(),
base_url: "http://127.0.0.1:18080".to_string(),
public_key,
@@ -1059,6 +1085,7 @@ mod tests {
async fn init_creates_identity_local_config_and_server_records() {
let temp = tempfile::tempdir().unwrap();
let database_path = temp.path().join("data").join("server").join("server.db");
std::fs::create_dir(temp.path().join(".git")).unwrap();
run_init_with_database_path(
InitOptions {
workspace: temp.path().canonicalize().unwrap(),
+9
View File
@@ -1066,6 +1066,11 @@ mod tests {
) VALUES('w',?1,'r','one','builtin:coder','normal','created','rev1')",
[worker_id().to_string()],
)?;
c.execute(
"INSERT INTO typed_tickets (workspace_id, ticket_id, slug, title, status, kind, priority, body, workflow_state, workflow_state_explicit) \
VALUES ('w', 'ticket', 'ticket', 'Ticket', 'open', 'task', 'normal', '', 'planning', 1)",
[],
)?;
Ok(())
})
.unwrap();
@@ -1249,7 +1254,11 @@ mod tests {
fn purge_tombstone_commit_is_idempotent() {
let s = setup();
s.with_conn(|conn| {
conn.execute("INSERT INTO typed_tickets(workspace_id,ticket_id,slug,title,status,kind,priority,body,workflow_state,workflow_state_explicit) VALUES('w','ticket-old','ticket-old','Old Ticket','open','task','normal','','planning',1)", [])?;
conn.execute("INSERT INTO worker_registry(workspace_id,worker_id,runtime_id,display_name,profile,retention_state,created_at,updated_at) VALUES('w','1','r','old worker','builtin:coder','normal','created','rev1')", [])?;
conn.execute("INSERT INTO ticket_worker_assignments(workspace_id,ticket_id,assignment_id,runtime_id,worker_id,assigned_by,assigned_at) VALUES('w','ticket-old','assignment-old','r','1','test','t')", [])?;
conn.execute("DELETE FROM worker_registry WHERE workspace_id='w' AND runtime_id='r' AND worker_id='1'", [])?;
conn.execute("DELETE FROM typed_tickets WHERE workspace_id='w' AND ticket_id='ticket-old'", [])?;
Ok(())
}).unwrap();
let p = s.plan_worker_removal(&req(), &inv()).unwrap();
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,394 @@
use std::path::{Path, PathBuf};
use std::sync::Arc;
use chrono::{SecondsFormat, Utc};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use uuid::Uuid;
use crate::store::{
ControlPlaneStore, RepositoryRecord, WorkspaceBootstrapRecord, WorkspaceRecord,
};
use crate::{Error, Result};
const DEFAULT_REPOSITORY_ID: &str = "main";
const MAX_DISPLAY_NAME_BYTES: usize = 200;
const MAX_OPERATION_KEY_BYTES: usize = 200;
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(deny_unknown_fields)]
pub struct InitialRepositoryIntent {
pub uri: String,
#[serde(default)]
pub display_name: Option<String>,
#[serde(default)]
pub default_ref: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(deny_unknown_fields)]
pub struct WorkspaceCreateRequest {
pub operation_key: String,
pub display_name: String,
pub repository: InitialRepositoryIntent,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct WorkspaceCreateResponse {
pub workspace: WorkspaceRecord,
pub repository: RepositoryRecord,
pub config_revision: u64,
pub request_fingerprint: String,
pub replayed: bool,
}
#[derive(Clone)]
pub struct WorkspaceCatalogService {
store: Arc<dyn ControlPlaneStore>,
}
impl WorkspaceCatalogService {
pub fn new(store: Arc<dyn ControlPlaneStore>) -> Self {
Self { store }
}
pub fn list(
&self,
owner_account_id: Option<&str>,
limit: usize,
) -> Result<Vec<WorkspaceRecord>> {
let limit = limit.clamp(1, 200);
Ok(self
.store
.list_workspaces()?
.into_iter()
.filter(|workspace| {
workspace.owner_account_id.is_none()
|| owner_account_id
.is_some_and(|owner| workspace.owner_account_id.as_deref() == Some(owner))
})
.take(limit)
.collect())
}
pub fn create(
&self,
request: WorkspaceCreateRequest,
owner_account_id: Option<String>,
) -> Result<WorkspaceCreateResponse> {
self.create_internal(request, owner_account_id, None, false)
}
pub fn create_first_ownerless(
&self,
request: WorkspaceCreateRequest,
) -> Result<WorkspaceCreateResponse> {
self.create_internal(request, None, None, true)
}
pub fn create_with_workspace_id(
&self,
request: WorkspaceCreateRequest,
owner_account_id: Option<String>,
requested_workspace_id: Option<String>,
) -> Result<WorkspaceCreateResponse> {
self.create_internal(request, owner_account_id, requested_workspace_id, false)
}
fn create_internal(
&self,
request: WorkspaceCreateRequest,
owner_account_id: Option<String>,
requested_workspace_id: Option<String>,
require_empty_catalog: bool,
) -> Result<WorkspaceCreateResponse> {
let operation_key = normalize_required(
"operation_key",
request.operation_key,
MAX_OPERATION_KEY_BYTES,
)?;
let display_name =
normalize_required("display_name", request.display_name, MAX_DISPLAY_NAME_BYTES)?;
let repository_path = validate_repository_uri(&request.repository.uri)?;
let repository_uri = repository_path.to_string_lossy().into_owned();
let repository_name = request
.repository
.display_name
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
.unwrap_or("Main repository")
.to_string();
let default_ref = request
.repository
.default_ref
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
.unwrap_or("HEAD")
.to_string();
let requested_workspace_id = requested_workspace_id
.map(|value| {
Uuid::parse_str(value.trim())
.map(|id| id.to_string())
.map_err(|_| Error::InvalidInput("workspace_id must be a UUID".to_string()))
})
.transpose()?;
let workspace_id = requested_workspace_id
.clone()
.unwrap_or_else(|| Uuid::now_v7().to_string());
let fingerprint = workspace_create_fingerprint(
requested_workspace_id.as_deref(),
&display_name,
owner_account_id.as_deref(),
&repository_uri,
&repository_name,
&default_ref,
);
let now = Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true);
let result = self
.store
.create_workspace_bootstrap(&WorkspaceBootstrapRecord {
operation_key,
request_fingerprint: fingerprint.clone(),
require_empty_catalog,
workspace: WorkspaceRecord {
workspace_id: workspace_id.clone(),
owner_account_id,
display_name,
state: "active".to_string(),
created_at: now.clone(),
updated_at: now.clone(),
},
repository: RepositoryRecord {
workspace_id,
repository_id: DEFAULT_REPOSITORY_ID.to_string(),
name: repository_name,
kind: "git".to_string(),
provider: Some("git".to_string()),
uri: repository_uri,
default_ref: Some(default_ref),
auth_ref_kind: None,
auth_ref_key: None,
created_at: now.clone(),
updated_at: now,
},
})?;
Ok(WorkspaceCreateResponse {
workspace: result.workspace,
repository: result.repository,
config_revision: result.config_revision,
request_fingerprint: fingerprint,
replayed: result.replayed,
})
}
}
fn normalize_required(field: &str, value: String, max_bytes: usize) -> Result<String> {
let value = value.trim();
if value.is_empty() || value.len() > max_bytes {
return Err(Error::InvalidInput(format!(
"{field} must be between 1 and {max_bytes} bytes"
)));
}
Ok(value.to_string())
}
fn validate_repository_uri(uri: &str) -> Result<PathBuf> {
let uri = uri.trim();
if uri.is_empty() || uri.contains("://") {
return Err(Error::InvalidInput(
"initial repository uri must be an absolute server-local path".to_string(),
));
}
let path = Path::new(uri);
if !path.is_absolute() {
return Err(Error::InvalidInput(
"initial repository uri must be an absolute server-local path".to_string(),
));
}
let path = path.canonicalize().map_err(|error| {
Error::InvalidInput(format!("initial repository path is unavailable: {error}"))
})?;
if !path.is_dir() {
return Err(Error::InvalidInput(
"initial repository path must be a directory".to_string(),
));
}
let normal_git = path.join(".git").exists();
let bare_git = path.join("HEAD").is_file() && path.join("objects").is_dir();
if !normal_git && !bare_git {
return Err(Error::InvalidInput(
"initial repository path is not a Git repository".to_string(),
));
}
Ok(path)
}
fn workspace_create_fingerprint(
requested_workspace_id: Option<&str>,
display_name: &str,
owner_account_id: Option<&str>,
repository_uri: &str,
repository_name: &str,
default_ref: &str,
) -> String {
let payload = serde_json::json!({
"requested_workspace_id": requested_workspace_id,
"display_name": display_name,
"owner_account_id": owner_account_id,
"repository": {
"repository_id": DEFAULT_REPOSITORY_ID,
"uri": repository_uri,
"display_name": repository_name,
"default_ref": default_ref,
"kind": "git",
}
});
let mut hasher = Sha256::new();
hasher.update(serde_json::to_vec(&payload).expect("workspace fingerprint serializes"));
let digest = hasher.finalize();
let encoded = digest
.iter()
.map(|byte| format!("{byte:02x}"))
.collect::<String>();
format!("sha256:{encoded}")
}
#[cfg(test)]
mod tests {
use super::*;
use crate::store::SqliteWorkspaceStore;
fn git_repository() -> tempfile::TempDir {
let dir = tempfile::tempdir().unwrap();
std::fs::create_dir(dir.path().join(".git")).unwrap();
dir
}
#[tokio::test]
async fn create_is_atomic_and_exact_retries_converge() {
let store = Arc::new(SqliteWorkspaceStore::in_memory().unwrap());
let service = WorkspaceCatalogService::new(store.clone());
let repository = git_repository();
let request = WorkspaceCreateRequest {
operation_key: "request-1".to_string(),
display_name: "Workspace A".to_string(),
repository: InitialRepositoryIntent {
uri: repository.path().display().to_string(),
display_name: None,
default_ref: None,
},
};
let created = service.create(request.clone(), None).unwrap();
let replayed = service.create(request, None).unwrap();
assert!(!created.replayed);
assert!(replayed.replayed);
assert_eq!(
created.workspace.workspace_id,
replayed.workspace.workspace_id
);
assert_eq!(store.list_workspaces().unwrap().len(), 1);
assert_eq!(
store
.list_repositories(&created.workspace.workspace_id)
.unwrap()
.len(),
1
);
assert!(
store
.load_workspace_config(&created.workspace.workspace_id)
.unwrap()
.is_some()
);
}
#[test]
fn concurrent_ownerless_bootstrap_commits_exactly_one_workspace() {
let store = Arc::new(SqliteWorkspaceStore::in_memory().unwrap());
let service = WorkspaceCatalogService::new(store.clone());
let repository_a = git_repository();
let repository_b = git_repository();
let requests = [
WorkspaceCreateRequest {
operation_key: "bootstrap-a".to_string(),
display_name: "Workspace A".to_string(),
repository: InitialRepositoryIntent {
uri: repository_a.path().display().to_string(),
display_name: None,
default_ref: None,
},
},
WorkspaceCreateRequest {
operation_key: "bootstrap-b".to_string(),
display_name: "Workspace B".to_string(),
repository: InitialRepositoryIntent {
uri: repository_b.path().display().to_string(),
display_name: None,
default_ref: None,
},
},
];
let barrier = Arc::new(std::sync::Barrier::new(2));
let results = std::thread::scope(|scope| {
requests
.into_iter()
.map(|request| {
let service = service.clone();
let barrier = barrier.clone();
scope.spawn(move || {
barrier.wait();
service.create_first_ownerless(request)
})
})
.collect::<Vec<_>>()
.into_iter()
.map(|handle| handle.join().unwrap())
.collect::<Vec<_>>()
});
assert_eq!(results.iter().filter(|result| result.is_ok()).count(), 1);
assert_eq!(results.iter().filter(|result| result.is_err()).count(), 1);
assert_eq!(store.list_workspaces().unwrap().len(), 1);
let error = results
.into_iter()
.find_map(Result::err)
.unwrap()
.to_string();
assert!(error.contains("catalog is empty"), "{error}");
}
#[tokio::test]
async fn idempotency_key_reuse_with_different_payload_is_rejected() {
let store = Arc::new(SqliteWorkspaceStore::in_memory().unwrap());
let service = WorkspaceCatalogService::new(store);
let repository = git_repository();
let mut request = WorkspaceCreateRequest {
operation_key: "request-1".to_string(),
display_name: "Workspace A".to_string(),
repository: InitialRepositoryIntent {
uri: repository.path().display().to_string(),
display_name: None,
default_ref: None,
},
};
service.create(request.clone(), None).unwrap();
request.display_name = "Workspace B".to_string();
let error = service.create(request, None).unwrap_err().to_string();
assert!(error.contains("different input"), "{error}");
}
#[test]
fn repository_intent_rejects_remote_and_non_git_paths() {
let remote = validate_repository_uri("https://example.test/repo.git").unwrap_err();
assert!(remote.to_string().contains("server-local path"));
let dir = tempfile::tempdir().unwrap();
let non_git = validate_repository_uri(&dir.path().display().to_string()).unwrap_err();
assert!(non_git.to_string().contains("not a Git repository"));
}
}