feat: expose retained Bash streams through common WIP file contracts

This commit is contained in:
2026-10-20 03:13:46 +09:00
parent 6576850b7a
commit dd5384df3b
8 changed files with 1064 additions and 2 deletions
+5 -2
View File
@@ -172,6 +172,10 @@ async fn command_page(
} }
impl BashTool { 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<WorkdirSessionRouter>, output: BashOutputView) -> Self { pub fn new(router: Arc<WorkdirSessionRouter>, output: BashOutputView) -> Self {
Self { Self {
router, router,
@@ -499,10 +503,9 @@ pub(crate) fn routed_bash_tool(
output: BashOutputView, output: BashOutputView,
) -> ToolDefinition { ) -> ToolDefinition {
Arc::new(move || { Arc::new(move || {
let schema = schemars::schema_for!(BashParams);
let meta = ToolMeta::new("Bash") 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.") .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<dyn Tool> = Arc::new(BashTool::new(router.clone(), output.clone())); let tool: Arc<dyn Tool> = Arc::new(BashTool::new(router.clone(), output.clone()));
(meta, tool) (meta, tool)
}) })
+6
View File
@@ -183,6 +183,9 @@ impl BashOutputView {
// happens inside the runtime, so idle expiry must start here as well. // happens inside the runtime, so idle expiry must start here as well.
self.start_reaper(); self.start_reaper();
let mut store = self.store.lock().unwrap(); let mut store = self.store.lock().unwrap();
if !self.is_active() {
return Err("Bash output authority revoked".into());
}
store.expire(Instant::now()); store.expire(Instant::now());
while store.records.len() >= STORE_COUNT || store.reserved + COMMAND_BYTES > STORE_BYTES { while store.records.len() >= STORE_COUNT || store.reserved + COMMAND_BYTES > STORE_BYTES {
let Some(index) = store let Some(index) = store
@@ -220,6 +223,9 @@ impl BashOutputView {
return None; return None;
} }
let mut store = self.store.lock().unwrap(); let mut store = self.store.lock().unwrap();
if !self.is_active() {
return None;
}
store.expire(Instant::now()); store.expire(Instant::now());
store store
.records .records
+114
View File
@@ -31,11 +31,18 @@ enum WriteResponseFault {
LostBody, LostBody,
} }
#[derive(Clone, Copy)]
enum CommandResponseFault {
LostTerminal,
EmptyWaiting,
}
struct ProviderState { struct ProviderState {
session: WorkdirSessionHandle, session: WorkdirSessionHandle,
requests: Mutex<Vec<(String, String)>>, requests: Mutex<Vec<(String, String)>>,
operations: Mutex<Vec<WorkdirSessionOperation>>, operations: Mutex<Vec<WorkdirSessionOperation>>,
write_fault: Mutex<Option<WriteResponseFault>>, write_fault: Mutex<Option<WriteResponseFault>>,
command_fault: Mutex<Option<CommandResponseFault>>,
observe_refusal: Mutex<bool>, observe_refusal: Mutex<bool>,
} }
@@ -152,6 +159,21 @@ async fn operate(
source: Box::new(WorkdirError::OutOfScope("/private/provider/path".into())), 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 = let result =
dispatch_workdir_session_operation(state.session.as_ref(), request.operation).await; dispatch_workdir_session_operation(state.session.as_ref(), request.operation).await;
match result { match result {
@@ -159,6 +181,15 @@ async fn operate(
Ok(result) => { Ok(result) => {
let fault = if is_write { let fault = if is_write {
state.write_fault.lock().unwrap().take() 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 { } else {
None None
}; };
@@ -237,6 +268,7 @@ impl HttpFixture {
requests: Mutex::new(Vec::new()), requests: Mutex::new(Vec::new()),
operations: Mutex::new(Vec::new()), operations: Mutex::new(Vec::new()),
write_fault: Mutex::new(None), write_fault: Mutex::new(None),
command_fault: Mutex::new(None),
observe_refusal: Mutex::new(false), observe_refusal: Mutex::new(false),
}); });
let app = Router::new() let app = Router::new()
@@ -800,3 +832,85 @@ async fn remote_checkout_diagnostic_context_keeps_wip_refusal_classification_wit
"refusal was retried" "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;
}
+14
View File
@@ -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 { if wip_mode && let Some(feature) = &workdir_wip_feature {
crate::feature::builtin::manage_workdir::wip::mount_workspace_workdir_wip( crate::feature::builtin::manage_workdir::wip::mount_workspace_workdir_wip(
&mut wip_mount_registry, &mut wip_mount_registry,
+1
View File
@@ -5,6 +5,7 @@
//! an external plugin-loading surface. //! an external plugin-loading surface.
pub mod backend_job_result; pub mod backend_job_result;
pub(crate) mod bash;
pub mod chat_invocation; pub mod chat_invocation;
pub mod drive; pub mod drive;
pub mod flow_transition; pub mod flow_transition;
+451
View File
@@ -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<WorkdirSessionRouter>,
view: BashOutputView,
permissions: Option<ToolPermissionConfig>,
) -> 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<ToolPermissionConfig>,
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<Self>) -> 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<CommandStream>)> {
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<Provider> {
async fn publication(&self, path: &str) -> Result<Option<WipPublication>, 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<Vec<String>, ProtocolError> {
Ok(vec![])
}
}
struct Run {
provider: Arc<Provider>,
}
#[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<String, Value>,
ctx: WipCallContext,
) -> Result<WipOperationOutput, WipOperationError> {
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<Provider>,
id: String,
stream: Option<CommandStream>,
path: String,
validator: Vec<u8>,
}
impl File {
fn order_text(&self) -> Result<String, WipOperationError> {
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<Vec<u8>> {
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<String, Value>,
_: WipCallContext,
) -> Result<WipOperationOutput, WipOperationError> {
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,
}
}
+3
View File
@@ -7121,3 +7121,6 @@ mod tests {
assert_eq!(calls.load(Ordering::SeqCst), 0); assert_eq!(calls.load(Ordering::SeqCst), 0);
} }
} }
#[cfg(test)]
mod bash_output_tests;
+470
View File
@@ -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<WorkdirSessionRouter>,
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<WorkdirSessionRouter>, 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<Json, ToolError> {
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());
}