From dd5384df3b410f7b30b89f417d3e7a55b5d36c6e Mon Sep 17 00:00:00 2001 From: Hare Date: Tue, 20 Oct 2026 03:13:46 +0900 Subject: [PATCH] feat: expose retained Bash streams through common WIP file contracts --- crates/tools/src/bash.rs | 7 +- crates/tools/src/bash_output.rs | 6 + crates/worker/src/checkout_http_tests.rs | 114 +++++ crates/worker/src/controller.rs | 14 + crates/worker/src/feature/builtin.rs | 1 + crates/worker/src/feature/builtin/bash.rs | 451 ++++++++++++++++++++ crates/worker/src/wip.rs | 3 + crates/worker/src/wip/bash_output_tests.rs | 470 +++++++++++++++++++++ 8 files changed, 1064 insertions(+), 2 deletions(-) create mode 100644 crates/worker/src/feature/builtin/bash.rs create mode 100644 crates/worker/src/wip/bash_output_tests.rs diff --git a/crates/tools/src/bash.rs b/crates/tools/src/bash.rs index ae911bb71..74dcee471 100644 --- a/crates/tools/src/bash.rs +++ b/crates/tools/src/bash.rs @@ -172,6 +172,10 @@ async fn command_page( } impl BashTool { + pub fn input_schema() -> serde_json::Value { + serde_json::to_value(schemars::schema_for!(BashParams)).expect("Bash schema serialization") + } + pub fn new(router: Arc, output: BashOutputView) -> Self { Self { router, @@ -499,10 +503,9 @@ pub(crate) fn routed_bash_tool( output: BashOutputView, ) -> ToolDefinition { Arc::new(move || { - let schema = schemars::schema_for!(BashParams); let meta = ToolMeta::new("Bash") .description("Execute a shell command in one selected Workdir. Workdir owns launch, lossless bounded byte transport, timeout and cancel; Bash owns preview, quota-limited ephemeral output and read-only virtual references. This is not a sandbox.") - .input_schema(serde_json::to_value(schema).expect("Bash schema serialization")); + .input_schema(BashTool::input_schema()); let tool: Arc = Arc::new(BashTool::new(router.clone(), output.clone())); (meta, tool) }) diff --git a/crates/tools/src/bash_output.rs b/crates/tools/src/bash_output.rs index 3f31fcb44..538b50433 100644 --- a/crates/tools/src/bash_output.rs +++ b/crates/tools/src/bash_output.rs @@ -183,6 +183,9 @@ impl BashOutputView { // happens inside the runtime, so idle expiry must start here as well. self.start_reaper(); let mut store = self.store.lock().unwrap(); + if !self.is_active() { + return Err("Bash output authority revoked".into()); + } store.expire(Instant::now()); while store.records.len() >= STORE_COUNT || store.reserved + COMMAND_BYTES > STORE_BYTES { let Some(index) = store @@ -220,6 +223,9 @@ impl BashOutputView { return None; } let mut store = self.store.lock().unwrap(); + if !self.is_active() { + return None; + } store.expire(Instant::now()); store .records diff --git a/crates/worker/src/checkout_http_tests.rs b/crates/worker/src/checkout_http_tests.rs index b3d85bfc6..f9202f202 100644 --- a/crates/worker/src/checkout_http_tests.rs +++ b/crates/worker/src/checkout_http_tests.rs @@ -31,11 +31,18 @@ enum WriteResponseFault { LostBody, } +#[derive(Clone, Copy)] +enum CommandResponseFault { + LostTerminal, + EmptyWaiting, +} + struct ProviderState { session: WorkdirSessionHandle, requests: Mutex>, operations: Mutex>, write_fault: Mutex>, + command_fault: Mutex>, observe_refusal: Mutex, } @@ -152,6 +159,21 @@ async fn operate( source: Box::new(WorkdirError::OutOfScope("/private/provider/path".into())), }); } + if let WorkdirSessionOperation::CommandOutput(output) = &request.operation { + if matches!( + *state.command_fault.lock().unwrap(), + Some(CommandResponseFault::EmptyWaiting) + ) { + return AxumJson(workdir::http::WorkdirSessionOperationResult::CommandOutput( + workdir::CommandOutput { + chunks: vec![], + next_cursor: output.cursor, + terminal: None, + }, + )) + .into_response(); + } + } let result = dispatch_workdir_session_operation(state.session.as_ref(), request.operation).await; match result { @@ -159,6 +181,15 @@ async fn operate( Ok(result) => { let fault = if is_write { state.write_fault.lock().unwrap().take() + } else if matches!(&result, workdir::http::WorkdirSessionOperationResult::CommandOutput(output) if output.terminal.is_some()) + { + let mut command_fault = state.command_fault.lock().unwrap(); + if matches!(*command_fault, Some(CommandResponseFault::LostTerminal)) { + command_fault.take(); + Some(WriteResponseFault::LostBody) + } else { + None + } } else { None }; @@ -237,6 +268,7 @@ impl HttpFixture { requests: Mutex::new(Vec::new()), operations: Mutex::new(Vec::new()), write_fault: Mutex::new(None), + command_fault: Mutex::new(None), observe_refusal: Mutex::new(false), }); let app = Router::new() @@ -800,3 +832,85 @@ async fn remote_checkout_diagnostic_context_keeps_wip_refusal_classification_wit "refusal was retried" ); } + +// Full collection/projection integration: production HTTP client, local process pipes, +// Bash collector/store, WIP codec and common file dispatch. Only HTTP response loss is injected. +#[tokio::test] +async fn remote_native_bash_retains_streams_quota_and_replays_lost_terminal_through_common_files() { + watchdog(async { + let f = HttpFixture::new(WorkdirSessionCapabilities::ALL).await; + let remote = Arc::new(f.open(TOKEN).await.unwrap()); + let router = Arc::new(WorkdirSessionRouter::new()); + router.attach(WorkdirAttachmentAlias::new("main").unwrap(), remote.clone()).unwrap(); + let mut registry = WipMountRegistry::new(); + crate::checkout::mount_checkouts(&mut registry, router.clone(), tools::Tracker::new(), None).unwrap(); + let view = tools::bash_output::BashOutputView::new(); + crate::feature::builtin::bash::mount(&mut registry, router.clone(), view, None).unwrap(); + let runtime = WipRuntime::new(WipHost::new(registry), SecurityContext::new("remote-bash"), 0).unwrap(); + *f.state.command_fault.lock().unwrap() = Some(CommandResponseFault::LostTerminal); + let run = runtime.invoke("/bash-output".into(), contextual_reference("yoi.bash/command/v1", "/bash-output"), "run".into(), json!({"input":{"command":r"printf 'first\nneedle\n'; printf '%200000s' ''; printf '\377€' >&2; exit 7","cwd":"."}}), Default::default()).await.unwrap(); + f.public_output(run.content.as_ref().unwrap()); + let result: Json = serde_json::from_str(&run.content.unwrap()).unwrap(); + assert_eq!(result["metadata"]["stdout_saved"], 200013); + assert_eq!(result["metadata"]["stderr_saved"], 4); + assert_eq!(result["metadata"]["terminal"]["exit_code"], 7); + assert_eq!(result["metadata"]["terminal"]["transport_complete"], true); + assert_eq!(result["metadata"]["storage_truncated"], false); + assert_eq!(result["preview_omitted"], true); + for name in ["stdout_entry", "stderr_entry", "order_entry"] { + runtime.inspect(result[name].as_str().unwrap().into(), false).await.unwrap(); + } + let stdout = result["stdout_entry"].as_str().unwrap(); + let search = runtime.invoke(stdout.into(), crate::file_operation::Contract::SearchText.reference(), "search".into(), json!({"pattern":"needle","limit":2}), Default::default()).await.unwrap(); + assert!(search.content.unwrap().contains("needle")); + let stderr = runtime.invoke(result["stderr_entry"].as_str().unwrap().into(), crate::file_operation::Contract::ReadBytes.reference(), "read".into(), json!({"limit":4}), Default::default()).await.unwrap(); + let stderr: Json = serde_json::from_str(&stderr.content.unwrap()).unwrap(); + use base64::Engine as _; + assert_eq!(base64::engine::general_purpose::STANDARD.decode(stderr["data"].as_str().unwrap()).unwrap(), [vec![255], "€".as_bytes().to_vec()].concat()); + let run = runtime.invoke("/bash-output".into(), contextual_reference("yoi.bash/command/v1", "/bash-output"), "run".into(), json!({"input":{"command":"printf '%4194303s' ''; printf '€'","cwd":"."}}), Default::default()).await.unwrap(); + let result: Json = serde_json::from_str(&run.content.unwrap()).unwrap(); + assert_eq!(result["metadata"]["stdout_saved"], 4194304); + assert_eq!(result["metadata"]["stdout_seen"], 4194306); + assert_eq!(result["metadata"]["transport_truncation"], "byte_budget"); + assert_eq!(result["metadata"]["storage_truncated"], true); + let tail = runtime.invoke(result["stdout_entry"].as_str().unwrap().into(), crate::file_operation::Contract::ReadBytes.reference(), "read".into(), json!({"offset":4194303,"limit":4}), Default::default()).await.unwrap(); + let tail: Json = serde_json::from_str(&tail.content.unwrap()).unwrap(); + assert_eq!(base64::engine::general_purpose::STANDARD.decode(tail["data"].as_str().unwrap()).unwrap(), vec![0xe2]); + assert_eq!(tail["eof"], true); + assert_eq!(f.state.operations.lock().unwrap().iter().filter(|op| matches!(op, WorkdirSessionOperation::CommandStart(_))).count(), 2); + assert!(f.state.command_fault.lock().unwrap().is_none()); + router.close_all().await.unwrap(); + f.assert_endpoints(); + }).await; +} + +#[tokio::test] +async fn waiting_remote_command_rejects_empty_response_without_http_busy_loop() { + watchdog(async { + let f = HttpFixture::new(WorkdirSessionCapabilities::ALL).await; + let remote = f.open(TOKEN).await.unwrap(); + *f.state.command_fault.lock().unwrap() = Some(CommandResponseFault::EmptyWaiting); + let error = remote + .command_output(workdir::CommandOutputRequest { + handle: workdir::CommandHandle("unused".into()), + cursor: 0, + limit: 1024, + wait: true, + }) + .await + .unwrap_err(); + assert!(matches!(error, WorkdirError::Unavailable(_))); + assert_eq!( + f.state + .operations + .lock() + .unwrap() + .iter() + .filter(|op| matches!(op, WorkdirSessionOperation::CommandOutput(_))) + .count(), + 1 + ); + remote.close().await.unwrap(); + }) + .await; +} diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index 9f233f95f..f51ab9505 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -1861,6 +1861,20 @@ where ) })?; } + if wip_mode { + crate::feature::builtin::bash::mount( + &mut wip_mount_registry, + worker.workdir_sessions(), + bash_output.clone(), + wip_permissions.clone(), + ) + .map_err(|error| { + std::io::Error::new( + std::io::ErrorKind::InvalidInput, + format!("mount Bash output WIP projection: {error}"), + ) + })?; + } if wip_mode && let Some(feature) = &workdir_wip_feature { crate::feature::builtin::manage_workdir::wip::mount_workspace_workdir_wip( &mut wip_mount_registry, diff --git a/crates/worker/src/feature/builtin.rs b/crates/worker/src/feature/builtin.rs index a68b8f992..529cfa8a0 100644 --- a/crates/worker/src/feature/builtin.rs +++ b/crates/worker/src/feature/builtin.rs @@ -5,6 +5,7 @@ //! an external plugin-loading surface. pub mod backend_job_result; +pub(crate) mod bash; pub mod chat_invocation; pub mod drive; pub mod flow_transition; diff --git a/crates/worker/src/feature/builtin/bash.rs b/crates/worker/src/feature/builtin/bash.rs new file mode 100644 index 000000000..ecd94428a --- /dev/null +++ b/crates/worker/src/feature/builtin/bash.rs @@ -0,0 +1,451 @@ +//! Bash Feature projection. The view is instance-owned and shared only by explicit parent +//! routing. No output is indexable; committed result entries directly address immutable files. +//! Root namespace knowledge grants nothing: every publication and read checks the current view. + +use crate::{ + file_operation::{self, Contract}, + permission::permission_action_for, + wip::{ + WipCallContext, WipMountError, WipMountRegistry, WipOperationError, WipOperationHandler, + WipOperationOutput, WipProjection, WipProjectionKind, WipPublication, WipSubtreeMount, + WipSubtreeProvider, contextual_reference, json_to_wip, wip_to_json, + }, +}; +use agen::tool::{Tool, ToolError}; +use async_trait::async_trait; +use manifest::{ToolPermissionAction, ToolPermissionConfig}; +use serde_json::{Value as Json, json}; +use sha2::{Digest, Sha256}; +use std::{collections::BTreeMap, sync::Arc}; +use tools::{BashTool, bash_output::BashOutputView}; +use wip_protocol::{ + Documentation, FieldDeclaration, INTERFACE_FORMAT_V1, InterfaceDescriptor, Object, + OperationDeclaration, ParameterDeclaration, ProtocolError, ProtocolErrorCode, + ReturnDeclaration, TypeExpr, Value, +}; +use workdir::{CommandStream, WorkdirSessionRouter}; + +const ROOT: &str = "/bash-output"; + +pub(crate) fn mount( + registry: &mut WipMountRegistry, + router: Arc, + view: BashOutputView, + permissions: Option, +) -> Result<(), WipMountError> { + registry.allocate_namespace("bash", "bash-output")?; + let provider = Arc::new(Provider { + service: BashTool::new(router, view.clone()), + view, + permissions, + incarnation: uuid::Uuid::now_v7().to_string(), + }); + registry.mount(provider.root())?; + registry.replace_compatibility_tools(ROOT, ["Bash"])?; + registry.mount_subtree(WipSubtreeMount { + root: ROOT.into(), + provider: Arc::new(provider), + }) +} + +struct Provider { + service: BashTool, + view: BashOutputView, + permissions: Option, + incarnation: String, +} +impl Provider { + fn allowed(&self, tool: &str, args: &Json) -> bool { + self.view.is_active() + && self + .permissions + .as_ref() + .is_none_or(|p| permission_action_for(p, tool, args) == ToolPermissionAction::Allow) + } + fn may_allow(&self, tool: &str) -> bool { + if !self.view.is_active() { + return false; + } + let Some(p) = &self.permissions else { + return true; + }; + for rule in p.rules.iter().filter(|r| r.tool.eq_ignore_ascii_case(tool)) { + if rule.action == ToolPermissionAction::Allow { + return true; + } + if rule.pattern == "*" { + return false; + } + } + p.default_action == ToolPermissionAction::Allow + } + fn root(self: &Arc) -> WipProjection { + let interface = contextual_reference("yoi.bash/command/v1", ROOT); + let descriptor = InterfaceDescriptor { + format: INTERFACE_FORMAT_V1.into(), + documentation: Some(Documentation { + summary: "Run Bash with bounded ephemeral output; retained bytes are accessed via result entries, not listed".into(), + details: None, + }), + types: vec![], + operations: vec![OperationDeclaration { + name: "run".into(), + documentation: None, + parameters: vec![ParameterDeclaration { + name: "input".into(), + required: true, + documentation: Some(Documentation { + summary: "Bash input, validated by the existing Tool argument contract".into(), + details: Some(BashTool::input_schema().to_string()), + }), + r#type: TypeExpr::Json, + }], + returns: ReturnDeclaration { + documentation: None, + r#type: TypeExpr::Record { + fields: vec![ + field("summary", TypeExpr::String), + field("stdout_preview", TypeExpr::String), + field("stderr_preview", TypeExpr::String), + field("preview_omitted", TypeExpr::Boolean), + field("metadata", TypeExpr::Json), + field("stdout_entry", TypeExpr::Entry), + field("stderr_entry", TypeExpr::Entry), + field("order_entry", TypeExpr::Entry), + ], + }, + }, + }], + }; + WipProjection { + route: ROOT.into(), + capability: "bash:command".into(), + kind: WipProjectionKind::Native, + object: Object { + name: "bash-output".into(), + description: Some("Bash command and nonindexable retained output".into()), + interfaces: vec![interface.clone()], + r#ref: Some(format!("bash:{}", self.incarnation)), + validator: Some(self.incarnation.as_bytes().to_vec()), + }, + interface, + interface_validator: Some(Sha256::digest(format!("{descriptor:?}")).to_vec()), + descriptor, + handler: Arc::new(Run { + provider: self.clone(), + }), + } + } + fn parse(&self, path: &str) -> Option<(String, Option)> { + let rest = path.strip_prefix("/bash-output/")?; + let (id, stream) = rest.split_once('/')?; + if uuid::Uuid::parse_str(id).ok()?.to_string() != id { + return None; + } + let stream = match stream { + "stdout" => Some(CommandStream::Stdout), + "stderr" => Some(CommandStream::Stderr), + "order" => None, + _ => return None, + }; + Some((id.into(), stream)) + } +} + +#[async_trait] +impl WipSubtreeProvider for Arc { + async fn publication(&self, path: &str) -> Result, ProtocolError> { + if !self.view.is_active() { + return Ok(None); + } + if path == ROOT { + let projection = self.root(); + return Ok(Some(WipPublication { + scope: Some(projection.object.clone()), + projection, + })); + } + let Some((id, stream)) = self.parse(path) else { + return Ok(None); + }; + let Some(metadata) = self.view.metadata(&id) else { + return Ok(None); + }; + if !self.may_allow("Read") && !self.may_allow("Grep") { + return Ok(None); + } + let mut contracts = vec![]; + if self.may_allow("Read") { + contracts.extend([Contract::ReadLines, Contract::ReadBytes]); + } + if self.may_allow("Grep") { + contracts.push(Contract::SearchText); + } + let primary = contracts[0]; + let interface = primary.reference(); + let descriptor = primary.descriptor(); + let validator = Sha256::digest(format!("{}:{id}:{stream:?}", self.incarnation)).to_vec(); + let mut projection = WipProjection { + route: path.into(), + capability: "bash:output".into(), + kind: WipProjectionKind::Native, + object: Object { + name: path.rsplit('/').next().unwrap().into(), + description: Some(format!( + "Read-only output component (stdout={} bytes, stderr={} bytes); text is UTF-8 lossy; storage_truncated={}", + metadata.stdout_saved, metadata.stderr_saved, metadata.storage_truncated + )), + interfaces: vec![interface.clone()], + r#ref: Some(format!("bash:{}:{id}:{stream:?}", self.incarnation)), + validator: Some(validator.clone()), + }, + interface, + descriptor: descriptor.clone(), + interface_validator: Some(primary.validator()), + handler: Arc::new(File { + provider: self.clone(), + id, + stream, + path: path.into(), + validator, + }), + }; + file_operation::expose(&mut projection, contracts); + Ok(Some(WipPublication { + projection, + scope: None, + })) + } + async fn children(&self, _: &str) -> Result, ProtocolError> { + Ok(vec![]) + } +} + +struct Run { + provider: Arc, +} +#[async_trait] +impl WipOperationHandler for Run { + fn is_visible(&self) -> bool { + self.provider.view.is_active() + } + fn operation_available(&self, op: &str) -> bool { + op == "run" && self.provider.may_allow("Bash") + } + async fn call( + &self, + op: &str, + args: &BTreeMap, + ctx: WipCallContext, + ) -> Result { + if op != "run" { + return Err(fail( + ProtocolErrorCode::OperationNotFound, + "unknown Bash operation", + )); + } + let input = args + .get("input") + .ok_or_else(|| fail(ProtocolErrorCode::InvalidArguments, "input is required"))?; + let input = + wip_to_json(input).map_err(|e| fail(ProtocolErrorCode::InvalidArguments, &e))?; + if !self.provider.allowed("Bash", &input) { + return Err(fail( + ProtocolErrorCode::PermissionDenied, + "Bash not permitted", + )); + } + let result = self + .provider + .service + .execute_result(&input.to_string(), ctx.execution) + .await + .map_err(tool_error)?; + if result.cancelled { + return Err(WipOperationError::Cancelled(result.tool_output())); + } + if result.metadata.terminal.status == Some(workdir::CommandStatus::TimedOut) { + return Err(WipOperationError::Interrupted(result.tool_output())); + } + // Entry fields are typed by the native Descriptor. No display-string parsing is involved. + Ok(WipOperationOutput::native(Value::Record(BTreeMap::from([ + ("summary".into(), Value::String(result.summary.clone())), + ( + "stdout_preview".into(), + Value::String(result.stdout_preview.clone()), + ), + ( + "stderr_preview".into(), + Value::String(result.stderr_preview.clone()), + ), + ( + "preview_omitted".into(), + Value::Boolean(result.preview_omitted), + ), + ("stdout_entry".into(), Value::String(result.stdout_entry())), + ("stderr_entry".into(), Value::String(result.stderr_entry())), + ("order_entry".into(), Value::String(result.order_entry())), + ( + "metadata".into(), + json_to_wip(&serde_json::to_value(result.metadata).unwrap()) + .map_err(|e| fail(ProtocolErrorCode::Internal, &e))?, + ), + ])))) + } + async fn cancel(&self, ctx: &agen::tool::ToolExecutionContext) -> Result<(), ToolError> { + self.provider.service.cancel_execution(ctx).await + } +} + +struct File { + provider: Arc, + id: String, + stream: Option, + path: String, + validator: Vec, +} +impl File { + fn order_text(&self) -> Result { + use std::fmt::Write; + let ranges = self + .provider + .view + .order(&self.id) + .ok_or_else(|| fail(ProtocolErrorCode::NotFound, "output expired or revoked"))?; + let mut text = String::new(); + for r in ranges { + writeln!(&mut text, "{:?} {} {}", r.stream, r.start, r.end).unwrap(); + } + Ok(text) + } +} +#[async_trait] +impl WipOperationHandler for File { + fn is_visible(&self) -> bool { + self.provider.view.metadata(&self.id).is_some() + } + fn object_validator(&self) -> Option> { + self.is_visible().then(|| self.validator.clone()) + } + fn operation_available(&self, op: &str) -> bool { + match op { + "read_lines" | "read_bytes" | "read" => self.provider.may_allow("Read"), + "search_text" | "search" => self.provider.may_allow("Grep"), + _ => false, + } + } + async fn call( + &self, + op: &str, + args: &BTreeMap, + _: WipCallContext, + ) -> Result { + if !self.operation_available(op) { + return Err(fail( + ProtocolErrorCode::PermissionDenied, + "output operation not permitted", + )); + } + let mut input = wip_to_json(&Value::Record(args.clone())) + .map_err(|e| fail(ProtocolErrorCode::InvalidArguments, &e))?; + let tool = if op == "search_text" { + input["path"] = json!(self.path); + "Grep" + } else { + input["file_path"] = json!(self.path); + "Read" + }; + if !self.provider.allowed(tool, &input) { + return Err(fail( + ProtocolErrorCode::PermissionDenied, + "output operation not permitted", + )); + } + let metadata = self + .provider + .view + .metadata(&self.id) + .ok_or_else(|| fail(ProtocolErrorCode::NotFound, "output expired or revoked"))?; + if op == "read_bytes" { + let (offset, limit) = file_operation::byte_range(args)?; + let (bytes, total) = if let Some(stream) = self.stream { + let total = if stream == CommandStream::Stdout { + metadata.stdout_saved + } else { + metadata.stderr_saved + }; + if offset > total as u64 { + return Err(fail( + ProtocolErrorCode::InvalidArguments, + "byte offset exceeds output length", + )); + } + let bytes = self + .provider + .view + .read(&self.id, stream, offset as usize, limit) + .ok_or_else(|| { + fail(ProtocolErrorCode::NotFound, "output expired or revoked") + })?; + (bytes, total) + } else { + let text = self.order_text()?; + if offset > text.len() as u64 { + return Err(fail( + ProtocolErrorCode::InvalidArguments, + "byte offset exceeds output length", + )); + } + ( + text.as_bytes()[offset as usize..(offset as usize + limit).min(text.len())] + .to_vec(), + text.len(), + ) + }; + let eof = offset + bytes.len() as u64 == total as u64; + return Ok(WipOperationOutput::native_with_validator( + file_operation::bytes_result(args, &bytes, eof)?, + self.validator.clone(), + )); + } + let (offset, limit) = file_operation::window(args, op == "search_text")?; + let value = if let Some(stream) = self.stream { + let (text, more) = self + .provider + .view + .line_window(&self.id, stream, offset, limit) + .ok_or_else(|| fail(ProtocolErrorCode::NotFound, "output expired or revoked"))? + .map_err(|e| fail(ProtocolErrorCode::ResourceLimitExceeded, &e))?; + file_operation::page_result(op, args, &text, offset, more)? + } else { + file_operation::text_result(op, args, &self.order_text()?)? + }; + Ok(WipOperationOutput::native_with_validator( + value, + self.validator.clone(), + )) + } +} + +fn fail(code: ProtocolErrorCode, message: &str) -> WipOperationError { + WipOperationError::Protocol(ProtocolError { + code, + message: message.into(), + }) +} +fn tool_error(error: ToolError) -> WipOperationError { + match error { + ToolError::InvalidArgument(_) => fail(ProtocolErrorCode::InvalidArguments, "invalid Bash arguments"), + ToolError::Cancelled(output) => WipOperationError::Cancelled(output), + ToolError::Interrupted(output) => WipOperationError::Interrupted(output), + _ => WipOperationError::OutcomeUnknown("Bash execution failed; effects or process termination may be unconfirmed; do not automatically restart the command".into()), + } +} +fn field(name: &str, r#type: TypeExpr) -> FieldDeclaration { + FieldDeclaration { + name: name.into(), + required: true, + documentation: None, + r#type, + } +} diff --git a/crates/worker/src/wip.rs b/crates/worker/src/wip.rs index 2929aba0b..a0951f187 100644 --- a/crates/worker/src/wip.rs +++ b/crates/worker/src/wip.rs @@ -7121,3 +7121,6 @@ mod tests { assert_eq!(calls.load(Ordering::SeqCst), 0); } } + +#[cfg(test)] +mod bash_output_tests; diff --git a/crates/worker/src/wip/bash_output_tests.rs b/crates/worker/src/wip/bash_output_tests.rs new file mode 100644 index 000000000..39eac727e --- /dev/null +++ b/crates/worker/src/wip/bash_output_tests.rs @@ -0,0 +1,470 @@ +//! Native Bash output integration through real WIP Client/Host codec and local process pipes. +//! No telemetry/storage fake: only the model client is absent. Runtime HTTP transport is tested +//! by worker-runtime's command_transport_tests. Each fixture owns its Workdir and ephemeral view. +use super::*; +use crate::file_operation::Contract; +use tempfile::TempDir; +use tools::bash_output::BashOutputView; +use workdir::{ + LocalWorkdirSession, WorkdirAttachmentAlias, WorkdirSessionHandle, WorkdirSessionRouter, +}; + +struct Fixture { + _dir: TempDir, + router: Arc, + view: BashOutputView, + runtime: WipRuntime, +} +impl Fixture { + fn new() -> Self { + let dir = TempDir::new().unwrap(); + std::fs::write(dir.path().join("real.txt"), "first\n雪 needle\nlast\n").unwrap(); + let session: WorkdirSessionHandle = Arc::new(LocalWorkdirSession::new( + manifest::Scope::writable(dir.path()).unwrap(), + dir.path().into(), + )); + let router = Arc::new(WorkdirSessionRouter::new()); + router + .attach(WorkdirAttachmentAlias::new("main").unwrap(), session) + .unwrap(); + let view = BashOutputView::new(); + let runtime = runtime(router.clone(), view.clone()); + Self { + _dir: dir, + router, + view, + runtime, + } + } + async fn run(&self, command: &str) -> Json { + value( + self.runtime + .invoke( + "/bash-output".into(), + contextual_reference("yoi.bash/command/v1", "/bash-output"), + "run".into(), + json!({"input":{"command":command,"cwd":"."}}), + Default::default(), + ) + .await + .unwrap(), + ) + } +} +fn runtime(router: Arc, view: BashOutputView) -> WipRuntime { + let mut registry = WipMountRegistry::new(); + crate::checkout::mount_checkouts(&mut registry, router.clone(), tools::Tracker::new(), None) + .unwrap(); + crate::feature::builtin::bash::mount(&mut registry, router, view, None).unwrap(); + WipRuntime::new(WipHost::new(registry), SecurityContext::new("bash-test"), 0).unwrap() +} +fn value(output: ToolOutput) -> Json { + serde_json::from_str(&output.content.unwrap()).unwrap() +} +async fn invoke( + runtime: &WipRuntime, + path: &str, + contract: Contract, + operation: &str, + args: Json, +) -> Result { + runtime + .invoke( + path.into(), + contract.reference(), + operation.into(), + args, + Default::default(), + ) + .await + .map(value) +} + +#[tokio::test] +async fn native_bash_entries_reuse_checkout_contract_for_direct_inspect_read_search() { + let f = Fixture::new(); + let result = f + .run("printf 'first\n雪 needle\nlast\n'; printf error >&2; exit 3") + .await; + assert_eq!(result["metadata"]["terminal"]["exit_code"], 3); + let stdout = result["stdout_entry"].as_str().unwrap(); + assert!(stdout.starts_with("/bash-output/")); + // The native result Descriptor gives these coordinates TypeExpr::Entry; Object inspection + // contains capabilities and metadata, not the process output body. + let projection = f + .runtime + .host + .projection_live("/bash-output") + .await + .unwrap() + .unwrap(); + let TypeExpr::Record { fields } = &projection.descriptor.operations[0].returns.r#type else { + panic!("native record"); + }; + assert_eq!( + fields + .iter() + .find(|v| v.name == "stdout_entry") + .unwrap() + .r#type, + TypeExpr::Entry + ); + let inspect = f.runtime.inspect(stdout.into(), false).await.unwrap(); + assert!(!inspect.content.unwrap().contains("雪 needle")); + for path in ["/checkouts/main/real.txt", stdout] { + let read = invoke( + &f.runtime, + path, + Contract::ReadLines, + "read", + json!({"offset":1,"limit":1}), + ) + .await + .unwrap(); + assert!(read["content"].as_str().unwrap().contains("雪 needle")); + assert_eq!(read["next_offset"], 2); + let search = invoke( + &f.runtime, + path, + Contract::SearchText, + "search", + json!({"offset":0,"limit":2,"pattern":"needle"}), + ) + .await + .unwrap(); + assert_eq!(search["matches"][0]["line"], 2); + assert_eq!(search["next_offset"], 2); + assert_eq!(search["truncated"], true); + } + for name in ["stdout_entry", "stderr_entry", "order_entry"] { + assert_eq!( + fields.iter().find(|v| v.name == name).unwrap().r#type, + TypeExpr::Entry + ); + f.runtime + .inspect(result[name].as_str().unwrap().into(), false) + .await + .unwrap(); + } + let stderr = invoke( + &f.runtime, + result["stderr_entry"].as_str().unwrap(), + Contract::ReadLines, + "read", + json!({}), + ) + .await + .unwrap(); + assert!(stderr["content"].as_str().unwrap().contains("error")); + let order = invoke( + &f.runtime, + result["order_entry"].as_str().unwrap(), + Contract::ReadLines, + "read", + json!({}), + ) + .await + .unwrap(); + assert!(order["content"].as_str().unwrap().contains("Stdout")); + assert!(order["content"].as_str().unwrap().contains("Stderr")); + assert_eq!(result["metadata"]["ordering"], "provider_read_observation"); + let tree = value( + f.runtime + .tree("/bash-output".into(), 8, true) + .await + .unwrap(), + ); + assert!(!tree.to_string().contains(stdout)); + assert!( + invoke( + &f.runtime, + stdout, + Contract::Write, + "write", + json!({"content":"no"}) + ) + .await + .is_err() + ); + assert!( + invoke( + &f.runtime, + stdout, + Contract::Edit, + "edit", + json!({"old_string":"first","new_string":"no"}) + ) + .await + .is_err() + ); +} + +#[tokio::test] +async fn common_byte_reader_handles_long_lines_raw_invalid_utf8_boundaries_and_eof() { + let f = Fixture::new(); + let result = f.run("printf '%100000s' ''; printf '€\\377'").await; + let path = result["stdout_entry"].as_str().unwrap(); + assert!( + invoke( + &f.runtime, + path, + Contract::ReadLines, + "read", + json!({"limit":1}) + ) + .await + .is_err() + ); + for (offset, expected) in [ + (99999, vec![b' ', 0xe2]), + (100001, vec![0x82, 0xac]), + (100003, vec![0xff]), + ] { + let read = invoke( + &f.runtime, + path, + Contract::ReadBytes, + "read", + json!({"offset":offset,"limit":2}), + ) + .await + .unwrap(); + let bytes = base64::engine::general_purpose::STANDARD + .decode(read["data"].as_str().unwrap()) + .unwrap(); + assert_eq!(bytes, expected); + assert_eq!(read["next_offset"], offset + expected.len() as u64); + } + let eof = invoke( + &f.runtime, + path, + Contract::ReadBytes, + "read", + json!({"offset":100004,"limit":2}), + ) + .await + .unwrap(); + assert_eq!(eof["data"], ""); + assert_eq!(eof["eof"], true); + assert!( + invoke( + &f.runtime, + path, + Contract::ReadBytes, + "read", + json!({"offset":100005}) + ) + .await + .is_err() + ); + assert!( + invoke( + &f.runtime, + path, + Contract::ReadBytes, + "read", + json!({"limit":32769}) + ) + .await + .is_err() + ); +} + +#[tokio::test] +async fn publication_and_invoke_recheck_child_authority_and_root_teardown_invalidates_old_refs() { + let f = Fixture::new(); + let lease = Arc::new(std::sync::atomic::AtomicBool::new(true)); + let child = f.view.child({ + let lease = lease.clone(); + Arc::new(move || lease.load(std::sync::atomic::Ordering::Acquire)) + }); + let child_runtime = runtime(f.router.clone(), child.clone()); + let result = value( + child_runtime + .invoke( + "/bash-output".into(), + contextual_reference("yoi.bash/command/v1", "/bash-output"), + "run".into(), + json!({"input":{"command":"printf child-evidence","cwd":"."}}), + Default::default(), + ) + .await + .unwrap(), + ); + let path = result["stdout_entry"].as_str().unwrap(); + child_runtime.inspect(path.into(), false).await.unwrap(); + let sibling = runtime(f.router.clone(), f.view.child(Arc::new(|| true))); + let unrelated = runtime(f.router.clone(), BashOutputView::new()); + assert!(sibling.inspect(path.into(), false).await.is_err()); + assert!(unrelated.inspect(path.into(), false).await.is_err()); + lease.store(false, std::sync::atomic::Ordering::Release); + assert!( + invoke(&child_runtime, path, Contract::ReadBytes, "read", json!({})) + .await + .is_err() + ); + assert!( + invoke( + &child_runtime, + path, + Contract::SearchText, + "search", + json!({"pattern":"child"}) + ) + .await + .is_err() + ); + assert!( + invoke(&f.runtime, path, Contract::ReadBytes, "read", json!({})) + .await + .is_ok() + ); + f.view.revoke(); + assert!( + invoke(&f.runtime, path, Contract::ReadBytes, "read", json!({})) + .await + .is_err() + ); + let restored = runtime(f.router.clone(), BashOutputView::new()); + assert!(restored.inspect(path.into(), false).await.is_err()); +} + +#[tokio::test] +async fn native_timeout_and_cancellation_keep_terminal_evidence_instead_of_success() { + let f = Fixture::new(); + let session = f.router.only_session().unwrap(); + let mut events = session.subscribe_command_events().unwrap(); + let runtime = Arc::new(f.runtime); + let context = agen::tool::ToolExecutionContext::new("native-cancel", "native-attempt", 0); + let running = runtime.clone(); + let executing_context = context.clone(); + let execution = tokio::spawn(async move { + running + .invoke( + "/bash-output".into(), + contextual_reference("yoi.bash/command/v1", "/bash-output"), + "run".into(), + json!({"input":{"command":"printf before; while :; do :; done","cwd":"."}}), + executing_context, + ) + .await + }); + tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + if matches!( + events.recv().await.unwrap(), + workdir::CommandEvent::Output { .. } + ) { + break; + } + } + }) + .await + .unwrap(); + runtime.cancel(&context).await.unwrap(); + let ToolError::Cancelled(output) = + tokio::time::timeout(std::time::Duration::from_secs(5), execution) + .await + .unwrap() + .unwrap() + .unwrap_err() + else { + panic!("expected cancelled terminal output") + }; + let body: Json = serde_json::from_str(&output.content.unwrap()).unwrap(); + let id = body["output"]["id"].as_str().unwrap(); + assert_eq!(body["output"]["terminal"]["transport_complete"], true); + assert_eq!(body["output"]["terminal"]["status"], "cancelled"); + assert_eq!( + f.view + .read(id, workdir::CommandStream::Stdout, 0, 20) + .unwrap(), + b"before" + ); + assert!( + invoke( + &runtime, + body["stdout_entry"].as_str().unwrap(), + Contract::ReadBytes, + "read", + json!({}) + ) + .await + .is_ok() + ); + let error = runtime.invoke("/bash-output".into(), contextual_reference("yoi.bash/command/v1", "/bash-output"), "run".into(), json!({"input":{"command":"printf timeout; while :; do :; done","cwd":".","timeout":1}}), Default::default()).await.unwrap_err(); + let ToolError::Interrupted(output) = error else { + panic!("expected timeout evidence") + }; + let body: Json = serde_json::from_str(&output.content.unwrap()).unwrap(); + assert_eq!(body["output"]["terminal"]["status"], "timed_out"); + assert_eq!(body["output"]["terminal"]["transport_complete"], true); + assert!( + f.view + .metadata(body["output"]["id"].as_str().unwrap()) + .is_some() + ); + session.close().await.unwrap(); +} + +#[tokio::test] +async fn common_read_and_search_permissions_are_independent_from_bash_execution() { + let f = Fixture::new(); + let result = f.run("printf needle").await; + let path = result["stdout_entry"].as_str().unwrap(); + let mut registry = WipMountRegistry::new(); + crate::feature::builtin::bash::mount( + &mut registry, + f.router.clone(), + f.view.clone(), + Some(ToolPermissionConfig { + default_action: ToolPermissionAction::Deny, + rules: vec![manifest::ToolPermissionRule { + tool: "Grep".into(), + pattern: "*".into(), + action: ToolPermissionAction::Allow, + }], + }), + ) + .unwrap(); + let restricted = WipRuntime::new( + WipHost::new(registry), + SecurityContext::new("restricted"), + 0, + ) + .unwrap(); + restricted.inspect(path.into(), false).await.unwrap(); + assert!( + invoke(&restricted, path, Contract::ReadBytes, "read", json!({})) + .await + .is_err() + ); + assert!( + invoke(&restricted, path, Contract::ReadLines, "read", json!({})) + .await + .is_err() + ); + let found = invoke( + &restricted, + path, + Contract::SearchText, + "search", + json!({"pattern":"needle"}), + ) + .await + .unwrap(); + assert!(found["matches"].as_array().unwrap().len() == 1); + assert!( + restricted + .invoke( + "/bash-output".into(), + contextual_reference("yoi.bash/command/v1", "/bash-output"), + "run".into(), + json!({"input":{"command":"touch denied","cwd":"."}}), + Default::default() + ) + .await + .is_err() + ); + assert!(!f._dir.path().join("denied").exists()); +}