feat: integrate authorized Worker Drive operations

Merge approved T-723 source e94af40d into develop3a2994cd. Resolve overlapping configuration WIP test migration by retaining source typed result assertions and target live Tree/Inspect publication checks. Runtime implementation merged without conflicts. Root check; full Manifest, Tools, Worker935+110, Runtime349+34, Server710+11, FS and Workdir suites; Workdir HTTP; API83/TypeScript91; WIP0.2.0 resolution and fmt/diff checks passed.
This commit is contained in:
2026-10-09 16:11:07 +09:00
35 changed files with 5886 additions and 116 deletions
+161
View File
@@ -1631,6 +1631,15 @@ where
);
}
let wip_mode = worker.manifest().worker.mode == manifest::WorkerMode::Wip;
let drive_feature = crate::feature::builtin::drive::DriveFeature::configured(
worker.workspace_client_handle(),
feature_config.drive.enabled,
worker.workdir_sessions(),
)
.map(|feature| feature.with_permissions(worker.manifest().permissions.clone()));
if let Some(module) = &drive_feature {
feature_registry.add_module(module.clone());
}
let workspace_config_feature =
crate::feature::builtin::workspace_config::WorkspaceConfigFeature::configured(
worker.workspace_client_handle(),
@@ -1731,6 +1740,15 @@ where
let host_worker_observation_provider = worker.worker_observation_provider();
let wip_permissions = worker.manifest().permissions.clone();
let mut wip_mount_registry = crate::wip::WipMountRegistry::new();
if wip_mode && let Some(feature) = &drive_feature {
crate::feature::builtin::drive::wip::mount_drive_wip(&mut wip_mount_registry, feature)
.map_err(|error| {
std::io::Error::new(
std::io::ErrorKind::InvalidInput,
format!("mount Drive WIP projection: {error}"),
)
})?;
}
if let Some(feature) = &workspace_config_feature {
crate::feature::builtin::workspace_config::wip::mount_workspace_config_wip(
&mut wip_mount_registry,
@@ -3969,6 +3987,149 @@ mod tests {
}
}
#[derive(Debug)]
struct DriveWorkspaceHost {
available: bool,
workspace: Option<&'static str>,
}
impl crate::worker::WorkspaceClient for DriveWorkspaceHost {
fn workspace_id(&self) -> Option<&str> {
self.workspace
}
fn kind(&self) -> &str {
"drive-test"
}
fn is_available(&self) -> bool {
self.available
}
fn execute(
&self,
request: crate::worker::WorkspaceRequest,
) -> Result<crate::worker::WorkspaceResponse, crate::worker::WorkspaceClientError>
{
// Native discovery must not enumerate or perform any mutation.
assert_eq!(request.path, "/api/w/workspace-1/drive/root");
Ok(crate::worker::WorkspaceResponse { status: 200, body: serde_json::json!({
"entry": {"workspace_id": "workspace-1", "node_id": "1"},
"parent": null, "name": "", "kind": "folder", "revision": "1",
"size": null, "content_type": null, "updated_by": "test",
"updated_at": "2026-10-08T00:00:00Z",
"latest_url": "/api/w/workspace-1/drive/download?entry_workspace_id=workspace-1&id=1"
}).to_string() })
}
}
#[tokio::test]
async fn drive_controller_wiring_requires_activation_and_workspace_host_in_both_modes() {
use crate::feature::FeatureModule as _;
for mode in [manifest::WorkerMode::Tools, manifest::WorkerMode::Wip] {
for (enabled, available, workspace) in [
(false, true, Some("workspace-1")),
(true, false, Some("workspace-1")),
(true, true, None),
(true, true, Some("workspace-1")),
] {
let root = TempDir::new().unwrap();
let mut manifest = manifest::WorkerManifest::from_toml(
"[worker]\nname = 'drive-wiring'\n[model]\nscheme = 'anthropic'\nmodel_id = 'test'\n[engine]\n[scope]\n"
).unwrap();
manifest.worker.mode = mode;
manifest.feature.drive.enabled = enabled;
// Isolate Drive discovery from the independent Workspace catalog.
manifest.feature.workdir_catalog.enabled = false;
let client = ScriptedClient::new(vec![finished_response()]);
let requests = client.requests.clone();
let engine = Engine::<_, agen::state::Mutable, crate::SessionHistoryMetadata>::new_annotated(client);
let store = session_store::CombinedStore::new(
session_store::FsStore::new(root.path().join("sessions")).unwrap(),
session_store::FsWorkerStore::new(root.path().join("workers")).unwrap(),
);
let host: Arc<dyn crate::worker::WorkspaceClient> =
Arc::new(DriveWorkspaceHost {
available,
workspace,
});
let context = crate::worker::WorkerWorkspaceContext::with_client(
workspace.map(|id| crate::worker::WorkspaceId::new(id).unwrap()),
host.clone(),
);
let mut worker = Worker::new(
manifest,
engine,
store,
context,
crate::worker::WorkerFilesystemAuthority::local(
root.path().into(),
root.path().into(),
),
manifest::Scope::writable(root.path()).unwrap(),
)
.await
.unwrap();
worker.set_system_prompt_template(
crate::SystemPromptTemplate::parse(
"default",
crate::PromptCatalogSource::builtins_only(),
)
.unwrap(),
);
let descriptor = crate::feature::builtin::drive::DriveFeature::new(
host,
worker.workdir_sessions(),
)
.descriptor();
register_worker_tools(
&mut worker,
root.path().join("bash-output"),
root.path().join("runtime"),
SpawnedWorkerRegistry::new_for_internal_services(),
None,
None,
)
.await
.unwrap();
let configured = enabled && available && workspace.is_some();
let tools = worker.engine().tool_server_handle();
tools.flush_pending();
for tool in &descriptor.tools {
assert_eq!(
tools.get_tool(&tool.name).is_some(),
configured && mode == manifest::WorkerMode::Tools,
"{mode:?}, enabled={enabled}, available={available}, workspace={workspace:?}, tool={}",
tool.name
);
}
if mode == manifest::WorkerMode::Wip {
let (_, tree) = tools.get_tool("Tree").unwrap();
let output = tree
.execute(
r#"{"path":"/drive","depth":1}"#,
ToolExecutionContext::new("drive-wiring-tree", "batch", 0),
)
.await;
assert_eq!(output.is_ok(), configured, "{output:?}");
if let Ok(output) = output {
let tree: serde_json::Value =
serde_json::from_str(&output.content.unwrap()).unwrap();
assert_eq!(tree["path"], "/drive");
// A self-scoped entrance is an enumeration boundary,
// not an observed empty Workspace hierarchy.
assert_eq!(tree["children"], serde_json::Value::Null);
}
}
worker.run_text("Inspect enabled features").await.unwrap();
let requests = requests.lock().unwrap();
let prompt = requests[0].system_prompt.as_deref().unwrap_or("");
assert_eq!(
prompt.contains("## Workspace Drive"),
configured,
"{mode:?}, enabled={enabled}, available={available}, workspace={workspace:?}"
);
}
}
}
struct LargeOutputTool;
#[async_trait::async_trait]
+1
View File
@@ -6,6 +6,7 @@
pub mod backend_job_result;
pub mod chat_invocation;
pub mod drive;
pub mod flow_transition;
pub mod manage_workdir;
pub mod manage_worker;
+718
View File
@@ -0,0 +1,718 @@
//! Worker Drive operations use only the injected Workspace API. Observations
//! below are bounded preconditions, never a hierarchy replica or a grant cache.
pub mod backend;
pub(crate) mod delegation;
#[cfg(test)]
mod tests;
pub mod wip;
use crate::{
feature::{
FeatureDescriptor, FeatureInstallContext, FeatureInstallError, FeatureModule,
ToolContribution, ToolDeclaration,
},
worker::WorkspaceClient,
};
use agen::tool::{
Attachment, ImageAttachment, Tool, ToolDefinition, ToolError, ToolExecutionContext, ToolMeta,
ToolOutput,
};
use async_trait::async_trait;
use backend::{DriveBackend, DriveError};
use schemars::JsonSchema;
use serde::{Deserialize, de::DeserializeOwned};
use serde_json::{Value, json};
use server_api::*;
use sha2::{Digest, Sha256};
use std::{
collections::{HashMap, VecDeque},
sync::{Arc, Mutex},
};
use workdir::{WorkdirPath, WorkdirSessionCapability, WorkdirSessionRouter};
pub const FEATURE_ID: &str = "drive";
const MAX_OBSERVATIONS: usize = 256;
#[derive(Default)]
struct Observations {
entries: HashMap<String, DriveEntry>,
order: VecDeque<String>,
}
impl Observations {
fn insert(&mut self, entry: DriveEntry) {
let key = entry.entry.node_id.clone();
self.order.retain(|old| old != &key);
self.order.push_back(key.clone());
self.entries.insert(key, entry);
while self.order.len() > MAX_OBSERVATIONS {
if let Some(old) = self.order.pop_front() {
self.entries.remove(&old);
}
}
}
fn remove(&mut self, key: &str) {
self.entries.remove(key);
self.order.retain(|old| old != key);
}
}
#[derive(Clone)]
pub struct DriveFeature {
pub(super) backend: DriveBackend,
pub(super) router: Arc<WorkdirSessionRouter>,
observations: Arc<Mutex<Observations>>,
nonce: Arc<str>,
pub(super) permissions: Option<manifest::ToolPermissionConfig>,
}
impl DriveFeature {
pub fn configured(
client: Arc<dyn WorkspaceClient>,
enabled: bool,
router: Arc<WorkdirSessionRouter>,
) -> Option<Self> {
(enabled
&& client.is_available()
&& client
.workspace_id()
.is_some_and(|id| !id.is_empty() && !id.chars().any(char::is_control)))
.then(|| Self::new(client, router))
}
pub fn new(client: Arc<dyn WorkspaceClient>, router: Arc<WorkdirSessionRouter>) -> Self {
Self {
backend: DriveBackend::new(client),
router,
observations: Arc::new(Mutex::new(Observations::default())),
nonce: uuid::Uuid::now_v7().to_string().into(),
permissions: None,
}
}
pub fn with_permissions(mut self, permissions: Option<manifest::ToolPermissionConfig>) -> Self {
self.permissions = permissions;
self
}
pub fn tools(&self) -> Vec<(&'static str, ToolDefinition)> {
SPECS
.iter()
.map(|spec| {
let feature = self.clone();
let spec = *spec;
let definition: ToolDefinition = Arc::new(move || {
(
ToolMeta::new(spec.name)
.description(spec.description)
.input_schema((spec.schema)()),
Arc::new(DriveTool {
feature: feature.clone(),
name: spec.name,
}) as Arc<dyn Tool>,
)
});
(spec.name, definition)
})
.collect()
}
fn observe(&self, entry: DriveEntry) -> Result<(), DriveError> {
self.backend.validate_entry(&entry)?;
self.observations
.lock()
.map_err(|_| DriveError::Unavailable)?
.insert(entry);
Ok(())
}
fn observed(
&self,
id: &DriveEntryRef,
pinned: Option<&DriveEntry>,
) -> Result<DriveEntry, DriveError> {
self.backend.validate_ref(id)?;
if let Some(pinned) = pinned {
if pinned.entry != *id {
return Err(DriveError::Denied);
}
return Ok(pinned.clone());
}
self.observations.lock().map_err(|_| DriveError::Unavailable)?.entries.get(&id.node_id).cloned().ok_or_else(|| DriveError::Invalid("read or inspect metadata for this entry before mutation; observations are bounded and not persisted authority".into()))
}
fn request_id(&self, context: &ToolExecutionContext) -> String {
digest(format!("{}:{}", self.nonce, context.execution_id()).as_bytes())
}
pub(super) async fn execute(
&self,
name: &str,
input: &str,
context: &ToolExecutionContext,
pinned: Option<&DriveEntry>,
) -> Result<DriveOutput, DriveError> {
self.backend.workspace()?;
let request_id = self.request_id(context);
let mut attachments = Vec::new();
let mut mutated_id = None;
let value = match name {
"DriveRoot" => {
let _: EmptyArgs = decode(input)?;
let entry = self.backend.root().await?;
self.observe(entry.clone())?;
entry_value(entry)
}
"DriveMetadata" => {
let args: EntryArgs = decode(input)?;
let entry = self.backend.metadata(&args.entry).await?;
self.observe(entry.clone())?;
entry_value(entry)
}
"DriveList" => {
let args: ListArgs = decode(input)?;
if args
.limit
.is_some_and(|n| n == 0 || n > DRIVE_PAGE_MAX_LIMIT)
{
return Err(DriveError::Limit);
}
let parent = match args.parent {
Some(parent) => parent,
None => self.backend.root().await?.entry,
};
let response = self.backend.list(&parent, args.limit, args.after).await?;
page_value(response)
}
"DriveSearch" => {
let args: DriveSearchQuery = decode(input)?;
if args.query.len() > 512 || args.after.as_ref().is_some_and(|s| s.len() > 2048) {
return Err(DriveError::Limit);
}
page_value(self.backend.search(args).await?)
}
"DriveRead" => {
let args: ReadArgs = decode(input)?;
if args.max_bytes == 0 || args.max_bytes > DRIVE_CHUNK_MAX_BYTES {
return Err(DriveError::Limit);
}
let read = self.backend.read(&args.entry, args.max_bytes).await?;
if pinned.is_some_and(|entry| entry.revision != read.entry.revision) {
return Err(DriveError::Conflict);
}
self.observe(read.entry.clone())?;
json!({"entry": entry_value(read.entry), "text": read.text, "truncated":read.truncated})
}
"DriveViewImage" => {
let args: EntryArgs = decode(input)?;
let entry = if let Some(entry) = pinned {
self.observed(&args.entry, Some(entry))?
} else {
self.backend.metadata(&args.entry).await?
};
let bytes = self
.backend
.bytes(&entry, tools::view_image::MAX_IMAGE_BYTES)
.await?;
let media = tools::view_image::detect_image_mime(&bytes).ok_or_else(|| {
DriveError::Invalid("supported image bytes required: PNG/JPEG/GIF/WebP".into())
})?;
let size = bytes.len();
attachments.push(Attachment::Image(ImageAttachment::new(
media,
Arc::<[u8]>::from(bytes),
)));
self.observe(entry.clone())?;
json!({"entry":entry_value(entry), "attached":true, "mime_type":media, "bytes":size})
}
"DriveCreateFolder" => {
let args: FolderArgs = decode(input)?;
self.backend.validate_ref(&args.parent)?;
mutation_value(
self.backend
.mutate(DriveMutationRequest {
request_id,
mutation: DriveMutation::CreateFolder {
parent: args.parent,
name: args.name,
},
})
.await?,
)
}
"DriveCreateText" => {
let args: CreateArgs = decode(input)?;
self.backend.validate_ref(&args.parent)?;
let text = checked_text(args.content)?;
mutation_value(
self.backend
.mutate(DriveMutationRequest {
request_id,
mutation: DriveMutation::CreateText {
parent: args.parent,
name: args.name,
text,
content_type: args.content_type,
},
})
.await?,
)
}
"DriveWrite" => {
let args: WriteArgs = decode(input)?;
let entry = self.observed(&args.entry, pinned)?;
let text = checked_text(args.content)?;
mutated_id = Some(entry.entry.node_id.clone());
mutation_value(
self.backend
.mutate(DriveMutationRequest {
request_id,
mutation: DriveMutation::UpdateText {
id: entry.entry,
expected_revision: entry.revision,
text,
content_type: entry.content_type.unwrap_or_else(default_media),
},
})
.await?,
)
}
"DriveEdit" => {
let args: EditArgs = decode(input)?;
let entry = self.observed(&args.entry, pinned)?;
let edit = fs_operation::text::EditArgs {
old_string: args.old_string,
new_string: args.new_string,
replace_all: args.replace_all,
};
edit.validate(text_limits()).map_err(text_error)?;
let read = self
.backend
.read(&entry.entry, DRIVE_TEXT_MAX_BYTES as u32)
.await?;
if read.entry.revision != entry.revision {
return Err(DriveError::Conflict);
}
if read.truncated {
return Err(DriveError::Limit);
}
let text = fs_operation::text::edit(&read.text, &edit, text_limits())
.map_err(text_error)?
.content;
mutated_id = Some(entry.entry.node_id.clone());
mutation_value(
self.backend
.mutate(DriveMutationRequest {
request_id,
mutation: DriveMutation::UpdateText {
id: entry.entry,
expected_revision: entry.revision,
text,
content_type: entry.content_type.unwrap_or_else(default_media),
},
})
.await?,
)
}
"DriveRelocate" => {
let args: RelocateArgs = decode(input)?;
self.backend.validate_ref(&args.parent)?;
let entry = self.observed(&args.entry, pinned)?;
mutated_id = Some(entry.entry.node_id.clone());
mutation_value(
self.backend
.mutate(DriveMutationRequest {
request_id,
mutation: DriveMutation::Relocate {
id: entry.entry,
expected_revision: entry.revision,
parent: args.parent,
name: args.name,
},
})
.await?,
)
}
"DriveDelete" => {
let args: EntryArgs = decode(input)?;
let entry = self.observed(&args.entry, pinned)?;
mutated_id = Some(entry.entry.node_id.clone());
mutation_value(
self.backend
.mutate(DriveMutationRequest {
request_id,
mutation: DriveMutation::Delete {
id: entry.entry,
expected_revision: entry.revision,
},
})
.await?,
)
}
"DriveSaveWorkdir" => {
let args: SaveArgs = decode(input)?;
self.backend.validate_ref(&args.parent)?;
// Drive API rechecks write at publication; source uses the selected
// Workdir read broker, with no OS-path or command fallback.
let selected = self
.router
.resolve(Some(&args.target_workdir))
.map_err(|_| DriveError::Denied)?;
if !selected
.session
.capabilities()
.supports(WorkdirSessionCapability::Read)
{
return Err(DriveError::Denied);
}
if args.path.starts_with('/')
|| args.path.contains('\\')
|| args.path.split('/').any(|part| part == "..")
{
return Err(DriveError::Invalid(
"logical Workdir-relative path required".into(),
));
}
let path = WorkdirPath::new(&args.path).map_err(|_| {
DriveError::Invalid("logical Workdir-relative path required".into())
})?;
let bytes = tools::view_image::read_workdir_bytes(
selected.session.as_ref(),
path,
DRIVE_FILE_MAX_BYTES as usize,
)
.await
.map_err(|_| {
DriveError::Invalid(
"Workdir source denied, changed, oversized or unavailable".into(),
)
})?;
let query = DriveUploadQuery {
operation: DriveUploadOperation::Create,
request_id,
entry_workspace_id: args.parent.workspace_id,
parent_id: Some(args.parent.node_id),
name: Some(args.name),
id: None,
expected_revision: None,
content_type: args.content_type,
size: bytes.len() as u32,
sha256: digest(&bytes),
};
mutation_value(self.backend.upload(query, bytes).await?)
}
"DriveRequestStatus" => {
let args: StatusArgs = decode(input)?;
serde_json::to_value(self.backend.status(&args.request_id).await?)
.map_err(|_| DriveError::Unavailable)?
}
_ => return Err(DriveError::Invalid("unknown operation".into())),
};
// A successful mutation does not silently substitute its new revision
// into a prior observation. The caller observes explicitly next time.
if let Some(id) = mutated_id {
self.observations
.lock()
.map_err(|_| DriveError::Unavailable)?
.remove(&id);
}
Ok(DriveOutput { value, attachments })
}
}
pub(super) struct DriveOutput {
pub value: Value,
pub attachments: Vec<Attachment>,
}
impl DriveOutput {
fn tool(self, name: &str) -> Result<ToolOutput, ToolError> {
Ok(ToolOutput {
summary: format!("{name} completed through Workspace Drive API"),
content: Some(
serde_json::to_string(&self.value)
.map_err(|_| ToolError::Internal("Drive result encoding failed".into()))?,
),
attachments: self.attachments,
})
}
}
struct DriveTool {
feature: DriveFeature,
name: &'static str,
}
#[async_trait]
impl Tool for DriveTool {
async fn execute(
&self,
input: &str,
context: ToolExecutionContext,
) -> Result<ToolOutput, ToolError> {
self.feature
.execute(self.name, input, &context, None)
.await
.map_err(tool_error)?
.tool(self.name)
}
}
fn tool_error(error: DriveError) -> ToolError {
match error {
DriveError::Conflict => ToolError::StructuredConflict {
code: "drive_conflict".into(),
message: error.to_string(),
},
DriveError::Invalid(_) | DriveError::Limit => ToolError::InvalidArgument(error.to_string()),
_ => ToolError::ExecutionFailed(error.to_string()),
}
}
impl FeatureModule for DriveFeature {
fn descriptor(&self) -> FeatureDescriptor {
let mut descriptor = FeatureDescriptor::builtin(FEATURE_ID,"Workspace Drive").with_description("Explicit Backend-authorized Drive documents, images and artifact publication; no Workdir required").with_instruction(drive_instruction());
for spec in SPECS {
descriptor = descriptor.with_tool(ToolDeclaration::new(spec.name, spec.description));
}
descriptor
}
fn install(&self, context: &mut FeatureInstallContext<'_>) -> Result<(), FeatureInstallError> {
context
.instructions()
.register(crate::feature::FeatureInstructionContribution::new(
drive_instruction(),
))?;
for (name, definition) in self.tools() {
context
.tools()
.register(ToolContribution::new(name, definition))?;
}
Ok(())
}
}
fn drive_instruction() -> crate::feature::FeatureInstructionDeclaration {
crate::feature::FeatureInstructionDeclaration::new(
crate::feature::FeatureInstructionId::builtin("drive-workflow"),
"common.drive",
"Authorized Drive operation guidance",
)
.expect("static Drive instruction")
}
#[derive(Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)]
struct EmptyArgs {}
#[derive(Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)]
struct EntryArgs {
entry: DriveEntryRef,
}
#[derive(Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)]
struct ListArgs {
#[serde(default)]
parent: Option<DriveEntryRef>,
#[serde(default)]
#[schemars(range(min = 1, max = 200))]
limit: Option<u32>,
#[serde(default)]
after: Option<String>,
}
#[derive(Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)]
struct ReadArgs {
entry: DriveEntryRef,
#[serde(default = "default_read")]
#[schemars(range(min = 1, max = 65536))]
max_bytes: u32,
}
fn default_read() -> u32 {
DRIVE_CHUNK_MAX_BYTES
}
#[derive(Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)]
struct FolderArgs {
parent: DriveEntryRef,
name: String,
}
#[derive(Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)]
struct CreateArgs {
parent: DriveEntryRef,
name: String,
content: String,
#[serde(default = "default_media")]
content_type: String,
}
#[derive(Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)]
struct WriteArgs {
entry: DriveEntryRef,
content: String,
}
#[derive(Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)]
struct EditArgs {
entry: DriveEntryRef,
old_string: String,
new_string: String,
#[serde(default)]
replace_all: bool,
}
#[derive(Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)]
struct RelocateArgs {
entry: DriveEntryRef,
parent: DriveEntryRef,
name: String,
}
#[derive(Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)]
struct SaveArgs {
target_workdir: String,
path: String,
parent: DriveEntryRef,
name: String,
#[serde(default = "default_binary_media")]
content_type: String,
}
fn default_binary_media() -> String {
"application/octet-stream".into()
}
#[derive(Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)]
struct StatusArgs {
request_id: String,
}
fn default_media() -> String {
"text/markdown".into()
}
fn decode<T: DeserializeOwned>(input: &str) -> Result<T, DriveError> {
if input.len() > 512 * 1024 {
return Err(DriveError::Limit);
}
serde_json::from_str(input)
.map_err(|_| DriveError::Invalid("input shape or bound rejected".into()))
}
fn text_limits() -> fs_operation::text::TextLimits {
fs_operation::text::TextLimits {
max_input_bytes: Some(DRIVE_TEXT_MAX_BYTES),
max_output_bytes: Some(DRIVE_TEXT_MAX_BYTES),
max_replacements: None,
}
}
fn checked_text(text: String) -> Result<String, DriveError> {
fs_operation::text::write(text, text_limits())
.map(|out| out.content)
.map_err(text_error)
}
fn text_error(error: fs_operation::text::TextError) -> DriveError {
DriveError::Invalid(error.to_string())
}
fn digest(bytes: &[u8]) -> String {
Sha256::digest(bytes)
.iter()
.map(|b| format!("{b:02x}"))
.collect()
}
pub(super) fn entry_path(entry: &DriveEntryRef) -> String {
format!(
"/drive/{}/{}",
crate::feature::builtin::manage_workdir::wip::encode_identity(&entry.workspace_id),
entry.node_id
)
}
fn entry_value(entry: DriveEntry) -> Value {
json!({"path":entry_path(&entry.entry),"metadata":entry})
}
fn page_value(page: DriveListResponse) -> Value {
json!({"entries":page.entries.into_iter().map(entry_value).collect::<Vec<_>>(),"next_after":page.next_after})
}
fn mutation_value(response: DriveMutationResponse) -> Value {
json!({"request_id":response.request_id,"entry":response.entry.map(entry_value)})
}
#[derive(Clone, Copy)]
pub(super) struct Spec {
pub name: &'static str,
pub operation: &'static str,
pub description: &'static str,
pub schema: fn() -> Value,
}
fn schema<T: JsonSchema>() -> Value {
serde_json::to_value(schemars::schema_for!(T)).expect("static Drive argument schema")
}
pub(super) const SPECS: &[Spec] = &[
Spec {
name: "DriveRoot",
operation: "root",
description: "Discover the authorized Workspace Drive root. Profile activation is not a grant; no Workdir is required.",
schema: schema::<EmptyArgs>,
},
Spec {
name: "DriveMetadata",
operation: "metadata",
description: "Inspect one Workspace-bound stable entry ID and observe its revision for mutation.",
schema: schema::<EntryArgs>,
},
Spec {
name: "DriveList",
operation: "list",
description: "List a bounded folder page (default root), up to 200 entries. Results do not expand the WIP tree or observe revisions for writes.",
schema: schema::<ListArgs>,
},
Spec {
name: "DriveSearch",
operation: "search",
description: "Search names and optionally bounded text through Backend paged search; no embedding/index. Use the returned cursor to continue.",
schema: schema::<DriveSearchQuery>,
},
Spec {
name: "DriveRead",
operation: "read",
description: "Read up to 64 KiB UTF-8 text and observe entry revision. Past Session text remains an immutable observation.",
schema: schema::<ReadArgs>,
},
Spec {
name: "DriveCreateFolder",
operation: "create_folder",
description: "Create a folder through the explicit Drive write grant. Same-name conflicts are not overwritten.",
schema: schema::<FolderArgs>,
},
Spec {
name: "DriveCreateText",
operation: "create_text",
description: "Create bounded UTF-8 text (default Markdown), <=64 KiB. Success means DB publication committed.",
schema: schema::<CreateArgs>,
},
Spec {
name: "DriveWrite",
operation: "write",
description: "Replace <=64 KiB text using a previously read/metadata-observed revision. Conflicts are not retried with latest revision.",
schema: schema::<WriteArgs>,
},
Spec {
name: "DriveEdit",
operation: "edit",
description: "Partial text edit using previously observed revision and shared old_string rules. Require nonempty unique match unless replace_all. Truncated preimages cannot be edited.",
schema: schema::<EditArgs>,
},
Spec {
name: "DriveRelocate",
operation: "relocate",
description: "Rename and/or move an observed entry with revision precondition; stable ID and URL remain unchanged.",
schema: schema::<RelocateArgs>,
},
Spec {
name: "DriveDelete",
operation: "delete",
description: "Delete one observed entry with revision precondition. A recreated name has a different ID; stale observations cannot target it.",
schema: schema::<EntryArgs>,
},
Spec {
name: "DriveViewImage",
operation: "view_image",
description: "Attach revision-fixed PNG/JPEG/GIF/WebP bytes (<=10 MiB) through durable ToolOutput history/capture; a URL alone is not image viewing.",
schema: schema::<EntryArgs>,
},
Spec {
name: "DriveSaveWorkdir",
operation: "save_workdir",
description: "Publish a <=16 MiB artifact using selected Workdir alias and logical path plus Drive parent/name. Both source read and destination write are required. No Bash/raw host path/base64 input.",
schema: schema::<SaveArgs>,
},
Spec {
name: "DriveRequestStatus",
operation: "request_status",
description: "Query the same mutation request ID after OutcomeUnknown. Uncommitted snapshot is not proof of failure; never blindly recreate the request.",
schema: schema::<StatusArgs>,
},
];
@@ -0,0 +1,470 @@
//! Identity-bound Drive API adapter. The Backend owns grants, hierarchy, CAS and
//! receipts. In particular, a lost mutation response is queried, never resent.
use std::{sync::Arc, time::Duration};
use crate::worker::{
WorkspaceBinaryRequest, WorkspaceClient, WorkspaceRequest, WorkspaceRequestMethod,
};
use serde::{Serialize, de::DeserializeOwned};
use server_api::*;
pub(super) const DEADLINE: Duration = Duration::from_secs(30);
const RESPONSE_LIMIT: usize = 2 * 1024 * 1024;
#[derive(Clone)]
pub struct DriveBackend {
client: Arc<dyn WorkspaceClient>,
}
#[derive(Debug, Clone, thiserror::Error)]
pub enum DriveError {
#[error("Drive access denied")]
Denied,
#[error("Drive resource not found")]
NotFound,
#[error("Drive revision or name conflict; observe explicitly before another operation")]
Conflict,
#[error("Invalid Drive arguments: {0}")]
Invalid(String),
#[error("Drive request exceeds a limit")]
Limit,
#[error("Drive storage or transport unavailable")]
Unavailable,
#[error(
"Drive mutation outcome unknown; request_id={0}; query request_status; do not blindly repeat"
)]
OutcomeUnknown(String),
}
impl DriveBackend {
pub fn new(client: Arc<dyn WorkspaceClient>) -> Self {
Self { client }
}
pub fn workspace(&self) -> Result<&str, DriveError> {
self.client
.workspace_id()
.filter(|id| !id.is_empty() && self.client.is_available())
.ok_or(DriveError::Denied)
}
fn base(&self) -> Result<String, DriveError> {
Ok(format!("/api/w/{}/drive", encode(self.workspace()?)))
}
pub(super) fn validate_entry(&self, entry: &DriveEntry) -> Result<(), DriveError> {
self.validate_ref(&entry.entry)?;
if let Some(parent) = &entry.parent {
self.validate_ref(parent)?;
}
if !decimal(&entry.revision) || entry.size.is_some_and(|n| n > DRIVE_FILE_MAX_BYTES) {
return Err(DriveError::Unavailable);
}
Ok(())
}
pub(super) fn validate_ref(&self, entry: &DriveEntryRef) -> Result<(), DriveError> {
if entry.workspace_id != self.workspace()? {
return Err(DriveError::Denied);
}
if !decimal(&entry.node_id) {
return Err(DriveError::Invalid(
"canonical decimal node ID required".into(),
));
}
Ok(())
}
async fn json<T: DeserializeOwned + Send + 'static>(
&self,
request: WorkspaceRequest,
mutation: bool,
request_id: &str,
) -> Result<T, DriveError> {
let client = self.client.clone();
let response = tokio::time::timeout(
DEADLINE,
tokio::task::spawn_blocking(move || client.execute_with_timeout(request, DEADLINE)),
)
.await;
let unknown = || {
if mutation {
DriveError::OutcomeUnknown(request_id.into())
} else {
DriveError::Unavailable
}
};
let response = response
.map_err(|_| unknown())?
.map_err(|_| unknown())?
.map_err(|_| unknown())?;
if response.body.len() > RESPONSE_LIMIT {
return Err(if mutation {
unknown()
} else {
DriveError::Limit
});
}
if !response.is_success() {
return Err(classify(
response.status,
response.body.as_bytes(),
mutation,
request_id,
));
}
serde_json::from_str(&response.body).map_err(|_| unknown())
}
async fn get<T: DeserializeOwned + Send + 'static, Q: Serialize>(
&self,
suffix: &str,
query: &Q,
) -> Result<T, DriveError> {
let query = query_string(query)?;
self.json(
WorkspaceRequest::get(format!("{}{suffix}{query}", self.base()?)),
false,
"",
)
.await
}
pub async fn root(&self) -> Result<DriveEntry, DriveError> {
let entry: DriveEntry = self
.json(
WorkspaceRequest::get(format!("{}/root", self.base()?)),
false,
"",
)
.await?;
self.validate_entry(&entry)?;
if entry.kind != DriveEntryKind::Folder || entry.parent.is_some() {
return Err(DriveError::Unavailable);
}
Ok(entry)
}
pub async fn metadata(&self, entry: &DriveEntryRef) -> Result<DriveEntry, DriveError> {
self.validate_ref(entry)?;
let result: DriveEntry = self
.get(
"/metadata",
&DriveEntryQuery {
entry_workspace_id: entry.workspace_id.clone(),
id: entry.node_id.clone(),
},
)
.await?;
self.validate_entry(&result)?;
if result.entry != *entry {
return Err(DriveError::Unavailable);
}
Ok(result)
}
pub async fn list(
&self,
parent: &DriveEntryRef,
limit: Option<u32>,
after: Option<String>,
) -> Result<DriveListResponse, DriveError> {
self.validate_ref(parent)?;
let result: DriveListResponse = self
.get(
"/list",
&DriveListQuery {
entry_workspace_id: parent.workspace_id.clone(),
id: parent.node_id.clone(),
limit,
after,
},
)
.await?;
self.validate_page(
&result,
limit
.unwrap_or(DRIVE_PAGE_MAX_LIMIT)
.min(DRIVE_PAGE_MAX_LIMIT),
)?;
if result
.entries
.iter()
.any(|entry| entry.parent.as_ref() != Some(parent))
{
return Err(DriveError::Unavailable);
}
Ok(result)
}
pub async fn search(&self, query: DriveSearchQuery) -> Result<DriveListResponse, DriveError> {
let result: DriveListResponse = self.get("/search", &query).await?;
self.validate_page(
&result,
query
.limit
.unwrap_or(DRIVE_SEARCH_MAX_LIMIT)
.min(DRIVE_SEARCH_MAX_LIMIT),
)?;
Ok(result)
}
fn validate_page(&self, page: &DriveListResponse, limit: u32) -> Result<(), DriveError> {
if page.entries.len() > limit as usize {
return Err(DriveError::Limit);
}
for entry in &page.entries {
self.validate_entry(entry)?;
}
Ok(())
}
pub async fn read(
&self,
entry: &DriveEntryRef,
max_bytes: u32,
) -> Result<DriveReadTextResponse, DriveError> {
self.validate_ref(entry)?;
let result: DriveReadTextResponse = self
.get(
"/read-text",
&DriveReadTextQuery {
entry_workspace_id: entry.workspace_id.clone(),
id: entry.node_id.clone(),
max_bytes,
},
)
.await?;
self.validate_entry(&result.entry)?;
if result.entry.entry != *entry || result.text.len() > max_bytes as usize {
return Err(DriveError::Unavailable);
}
Ok(result)
}
pub async fn bytes(&self, entry: &DriveEntry, limit: usize) -> Result<Vec<u8>, DriveError> {
self.validate_entry(entry)?;
let total = entry
.size
.ok_or_else(|| DriveError::Invalid("file required".into()))?;
if total as usize > limit {
return Err(DriveError::Limit);
}
let mut bytes = Vec::with_capacity(total as usize);
// Even an empty file gets a revision-fixed read and current authorization.
loop {
let offset = bytes.len() as u32;
let length = total.saturating_sub(offset).clamp(1, DRIVE_CHUNK_MAX_BYTES);
let query = query_string(&DriveReadChunkQuery {
entry_workspace_id: entry.entry.workspace_id.clone(),
id: entry.entry.node_id.clone(),
expected_revision: entry.revision.clone(),
offset,
length,
})?;
let request = WorkspaceBinaryRequest {
method: WorkspaceRequestMethod::Get,
path: format!("{}/read-chunk{query}", self.base()?),
body: None,
max_response_bytes: DRIVE_CHUNK_MAX_BYTES as usize,
};
let client = self.client.clone();
let response = tokio::time::timeout(
DEADLINE,
tokio::task::spawn_blocking(move || {
client.execute_binary_with_timeout(request, DEADLINE)
}),
)
.await
.map_err(|_| DriveError::Unavailable)?
.map_err(|_| DriveError::Unavailable)?
.map_err(|_| DriveError::Unavailable)?;
if !(200..300).contains(&response.status) {
return Err(classify(response.status, &response.body, false, ""));
}
let expected = total.saturating_sub(offset).min(length) as usize;
if response.body.len() != expected {
return Err(DriveError::Unavailable);
}
bytes.extend_from_slice(&response.body);
if bytes.len() == total as usize {
return Ok(bytes);
}
}
}
async fn reconcile(
&self,
id: &str,
result: Result<DriveMutationResponse, DriveError>,
) -> Result<DriveMutationResponse, DriveError> {
// A parseable success can still be an invalid completion. Classify it
// before reconciliation so it also gets one original-request inquiry.
let result = result.and_then(|response| {
self.validate_completion(id, &response)
.map_err(|_| DriveError::OutcomeUnknown(id.into()))?;
Ok(response)
});
match result {
Err(DriveError::OutcomeUnknown(_)) => match self.status(id).await {
Ok(DriveRequestStatusResponse {
state: DriveRequestState::Committed,
response: Some(response),
..
}) => Ok(response), // status validates both IDs and entry metadata
_ => Err(DriveError::OutcomeUnknown(id.into())),
},
other => other,
}
}
fn validate_completion(
&self,
id: &str,
response: &DriveMutationResponse,
) -> Result<(), DriveError> {
if response.request_id != id {
return Err(DriveError::Unavailable);
}
if let Some(entry) = &response.entry {
self.validate_entry(entry)?;
}
Ok(())
}
pub async fn mutate(
&self,
request: DriveMutationRequest,
) -> Result<DriveMutationResponse, DriveError> {
let body = serde_json::to_string(&request)
.map_err(|_| DriveError::Invalid("invalid mutation".into()))?;
if body.len() > 512 * 1024 {
return Err(DriveError::Limit);
}
let result = self
.json(
WorkspaceRequest::json(
WorkspaceRequestMethod::Post,
format!("{}/mutate", self.base()?),
body,
),
true,
&request.request_id,
)
.await;
self.reconcile(&request.request_id, result).await
}
pub async fn upload(
&self,
query: DriveUploadQuery,
bytes: Vec<u8>,
) -> Result<DriveMutationResponse, DriveError> {
if bytes.len() > DRIVE_FILE_MAX_BYTES as usize {
return Err(DriveError::Limit);
}
let path = format!("{}/upload{}", self.base()?, query_string(&query)?);
let request = WorkspaceBinaryRequest {
method: WorkspaceRequestMethod::Put,
path,
body: Some(bytes),
max_response_bytes: RESPONSE_LIMIT,
};
let client = self.client.clone();
let result = tokio::time::timeout(
DEADLINE,
tokio::task::spawn_blocking(move || {
client.execute_binary_with_timeout(request, DEADLINE)
}),
)
.await;
let result = match result {
Ok(Ok(Ok(response))) if (200..300).contains(&response.status) => {
serde_json::from_slice(&response.body)
.map_err(|_| DriveError::OutcomeUnknown(query.request_id.clone()))
}
Ok(Ok(Ok(response))) => Err(classify(
response.status,
&response.body,
true,
&query.request_id,
)),
_ => Err(DriveError::OutcomeUnknown(query.request_id.clone())),
};
self.reconcile(&query.request_id, result).await
}
pub async fn status(&self, request_id: &str) -> Result<DriveRequestStatusResponse, DriveError> {
if request_id.is_empty() || request_id.len() > 256 {
return Err(DriveError::Invalid("invalid request ID".into()));
}
let response: DriveRequestStatusResponse = self
.json(
WorkspaceRequest::get(format!("{}/requests/{}", self.base()?, encode(request_id))),
false,
"",
)
.await?;
if response.request_id != request_id {
return Err(DriveError::Unavailable);
}
if let Some(result) = &response.response {
self.validate_completion(request_id, result)?;
}
Ok(response)
}
}
pub(super) fn decimal(value: &str) -> bool {
value
.as_bytes()
.first()
.is_some_and(|c| matches!(c, b'1'..=b'9'))
&& value.bytes().all(|c| c.is_ascii_digit())
&& value.parse::<i64>().is_ok_and(|id| id > 0)
}
pub(super) fn encode(value: &str) -> String {
value
.bytes()
.map(|b| {
if b.is_ascii_alphanumeric() || matches!(b, b'-' | b'_') {
(b as char).to_string()
} else {
format!("%{b:02X}")
}
})
.collect()
}
fn query_string<T: Serialize>(value: &T) -> Result<String, DriveError> {
let value =
serde_json::to_value(value).map_err(|_| DriveError::Invalid("invalid query".into()))?;
let object = value
.as_object()
.ok_or_else(|| DriveError::Invalid("invalid query".into()))?;
let pairs: Vec<_> = object
.iter()
.filter(|(_, v)| !v.is_null())
.map(|(k, v)| {
format!(
"{}={}",
encode(k),
encode(
v.as_str()
.map(str::to_owned)
.unwrap_or_else(|| v.to_string())
.as_str()
)
)
})
.collect();
Ok(if pairs.is_empty() {
String::new()
} else {
format!("?{}", pairs.join("&"))
})
}
fn classify(status: u16, body: &[u8], mutation: bool, id: &str) -> DriveError {
if let Ok(error) = serde_json::from_slice::<DriveApiError>(body) {
if error.classification == DriveFailureClassification::Unknown {
return DriveError::OutcomeUnknown(id.into());
}
return match error.code {
DriveApiErrorCode::Denied => DriveError::Denied,
DriveApiErrorCode::NotFound => DriveError::NotFound,
DriveApiErrorCode::Conflict => DriveError::Conflict,
DriveApiErrorCode::Invalid => DriveError::Invalid("Backend rejected arguments".into()),
DriveApiErrorCode::Limit => DriveError::Limit,
DriveApiErrorCode::StorageUnavailable => DriveError::Unavailable,
DriveApiErrorCode::OutcomeUnknown => DriveError::OutcomeUnknown(id.into()),
};
}
if matches!(status, 401 | 403) {
DriveError::Denied
} else if mutation {
DriveError::OutcomeUnknown(id.into())
} else {
DriveError::Unavailable
}
}
@@ -0,0 +1,477 @@
//! Fail-closed Drive boundary for parent-owned Internal SubWorkers, including Reviewers.
//!
//! Internal children have no Backend Drive identity. Explicit internal Drive delegation is not
//! supported until the Backend can approve scope bound to that child identity. A parent's current
//! grant, enabled profile, or Reviewer attestation is not a child Drive grant. Registered Runtime
//! Workers must use their own identity-bound, current Backend grants; this wrapper creates neither
//! identities nor grants and keeps no local authorization cache.
use std::{sync::Arc, time::Duration};
use crate::worker::{
ReviewerContext, WorkerWorkspaceContext, WorkspaceBinaryRequest, WorkspaceBinaryResponse,
WorkspaceClient, WorkspaceClientError, WorkspacePromptCatalogResolution, WorkspaceRequest,
WorkspaceResponse, WorkspaceServerOperation, WorkspaceWorkerDiscoveryRequest,
};
/// Removes Drive authority without changing the selected client's other authority or behavior.
/// Install outside any Reviewer client so its trusted review submission attestation is preserved.
#[derive(Debug)]
pub(crate) struct InternalChildWorkspaceClient {
inner: Arc<dyn WorkspaceClient>,
}
impl InternalChildWorkspaceClient {
pub(crate) fn new(inner: Arc<dyn WorkspaceClient>) -> Self {
Self { inner }
}
fn check_path(path: &str) -> Result<(), WorkspaceClientError> {
// Classify the route, not the method: even readonly Drive access must not inherit the
// parent's identity. Decode route segments and normalize dot segments before checking,
// so transport/router normalization cannot turn an allowed alias into a Drive request.
let path = path.split(['?', '#']).next().unwrap_or(path);
let mut decoded = Vec::with_capacity(path.len());
// HTTP URL parsers discard literal tabs/newlines before route normalization.
let bytes: Vec<_> = path
.bytes()
.filter(|byte| !matches!(byte, b'\t' | b'\r' | b'\n'))
.collect();
let mut index = 0;
while index < bytes.len() {
if bytes[index] == b'%'
&& index + 2 < bytes.len()
&& let (Some(high), Some(low)) =
(hex_digit(bytes[index + 1]), hex_digit(bytes[index + 2]))
{
decoded.push(high * 16 + low);
index += 3;
} else {
decoded.push(bytes[index]);
index += 1;
}
}
let decoded = String::from_utf8_lossy(&decoded).replace('\\', "/");
let mut segments = Vec::new();
for segment in decoded.split('/') {
match segment {
"" | "." => {}
".." => {
segments.pop();
}
segment => segments.push(segment),
}
}
if matches!(segments.as_slice(), ["api", "w", _, "drive", ..]) {
return Err(WorkspaceClientError::Unavailable(
"Internal SubWorker Drive access denied: no identity-bound Backend-approved child scope; explicit internal Drive delegation is not supported".into(),
));
}
Ok(())
}
}
fn hex_digit(byte: u8) -> Option<u8> {
match byte {
b'0'..=b'9' => Some(byte - b'0'),
b'a'..=b'f' => Some(byte - b'a' + 10),
b'A'..=b'F' => Some(byte - b'A' + 10),
_ => None,
}
}
/// Keep Workspace identity and prompt context, but never inherit Drive authority.
pub(crate) fn restrict_internal_child_drive(
context: WorkerWorkspaceContext,
) -> WorkerWorkspaceContext {
WorkerWorkspaceContext::with_client(
context.workspace_id().cloned(),
Arc::new(InternalChildWorkspaceClient::new(context.client_handle())),
)
}
impl WorkspaceClient for InternalChildWorkspaceClient {
fn workspace_id(&self) -> Option<&str> {
self.inner.workspace_id()
}
fn kind(&self) -> &str {
self.inner.kind()
}
fn is_available(&self) -> bool {
self.inner.is_available()
}
fn execute(
&self,
request: WorkspaceRequest,
) -> Result<WorkspaceResponse, WorkspaceClientError> {
Self::check_path(&request.path)?;
self.inner.execute(request)
}
fn execute_with_timeout(
&self,
request: WorkspaceRequest,
timeout: Duration,
) -> Result<WorkspaceResponse, WorkspaceClientError> {
Self::check_path(&request.path)?;
self.inner.execute_with_timeout(request, timeout)
}
fn execute_binary(
&self,
request: WorkspaceBinaryRequest,
) -> Result<WorkspaceBinaryResponse, WorkspaceClientError> {
Self::check_path(&request.path)?;
self.inner.execute_binary(request)
}
fn execute_binary_with_timeout(
&self,
request: WorkspaceBinaryRequest,
timeout: Duration,
) -> Result<WorkspaceBinaryResponse, WorkspaceClientError> {
Self::check_path(&request.path)?;
self.inner.execute_binary_with_timeout(request, timeout)
}
fn execute_server_operation(
&self,
operation: WorkspaceServerOperation,
) -> Result<WorkspaceResponse, WorkspaceClientError> {
self.inner.execute_server_operation(operation)
}
fn current_prompt_projection(
&self,
minimum_revision: Option<u64>,
) -> Result<Option<WorkspacePromptCatalogResolution>, WorkspaceClientError> {
self.inner.current_prompt_projection(minimum_revision)
}
fn list_workspace_workers(
&self,
request: WorkspaceWorkerDiscoveryRequest,
) -> Result<server_api::WorkspaceWorkerDiscoveryPage, WorkspaceClientError> {
self.inner.list_workspace_workers(request)
}
fn execute_worker_remove(
&self,
target_runtime_id: &str,
target_worker_id: &str,
reason: &str,
) -> Result<WorkspaceResponse, WorkspaceClientError> {
self.inner
.execute_worker_remove(target_runtime_id, target_worker_id, reason)
}
fn reviewer_context(&self) -> Option<&ReviewerContext> {
self.inner.reviewer_context()
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::worker::{ReviewerChildWorkspaceClient, WorkspaceRequestMethod};
use std::sync::Mutex;
#[derive(Debug, PartialEq, Eq)]
enum Call {
Json(WorkspaceRequest, Option<Duration>),
Binary(WorkspaceBinaryRequest, Option<Duration>),
Projection(Option<u64>),
Discovery(WorkspaceWorkerDiscoveryRequest),
Remove(String, String, String),
Server,
}
#[derive(Debug, Default)]
struct RecordingClient {
calls: Mutex<Vec<Call>>,
}
impl RecordingClient {
fn response(&self, call: Call) -> Result<WorkspaceResponse, WorkspaceClientError> {
self.calls.lock().unwrap().push(call);
Ok(WorkspaceResponse {
status: 202,
body: "forwarded".into(),
})
}
fn binary_response(
&self,
call: Call,
) -> Result<WorkspaceBinaryResponse, WorkspaceClientError> {
self.calls.lock().unwrap().push(call);
Ok(WorkspaceBinaryResponse {
status: 206,
body: vec![0, 255],
})
}
}
impl WorkspaceClient for RecordingClient {
fn workspace_id(&self) -> Option<&str> {
Some("ws")
}
fn kind(&self) -> &str {
"recording"
}
fn is_available(&self) -> bool {
true
}
fn execute(
&self,
request: WorkspaceRequest,
) -> Result<WorkspaceResponse, WorkspaceClientError> {
self.response(Call::Json(request, None))
}
fn execute_with_timeout(
&self,
request: WorkspaceRequest,
timeout: Duration,
) -> Result<WorkspaceResponse, WorkspaceClientError> {
self.response(Call::Json(request, Some(timeout)))
}
fn execute_binary(
&self,
request: WorkspaceBinaryRequest,
) -> Result<WorkspaceBinaryResponse, WorkspaceClientError> {
self.binary_response(Call::Binary(request, None))
}
fn execute_binary_with_timeout(
&self,
request: WorkspaceBinaryRequest,
timeout: Duration,
) -> Result<WorkspaceBinaryResponse, WorkspaceClientError> {
self.binary_response(Call::Binary(request, Some(timeout)))
}
fn current_prompt_projection(
&self,
revision: Option<u64>,
) -> Result<Option<WorkspacePromptCatalogResolution>, WorkspaceClientError> {
self.calls.lock().unwrap().push(Call::Projection(revision));
Err(WorkspaceClientError::Request("projection marker".into()))
}
fn list_workspace_workers(
&self,
request: WorkspaceWorkerDiscoveryRequest,
) -> Result<server_api::WorkspaceWorkerDiscoveryPage, WorkspaceClientError> {
self.calls.lock().unwrap().push(Call::Discovery(request));
Err(WorkspaceClientError::Request("discovery marker".into()))
}
fn execute_worker_remove(
&self,
runtime: &str,
worker: &str,
reason: &str,
) -> Result<WorkspaceResponse, WorkspaceClientError> {
self.response(Call::Remove(runtime.into(), worker.into(), reason.into()))
}
fn execute_server_operation(
&self,
_: WorkspaceServerOperation,
) -> Result<WorkspaceResponse, WorkspaceClientError> {
self.response(Call::Server)
}
}
fn binary(method: WorkspaceRequestMethod, path: &str) -> WorkspaceBinaryRequest {
WorkspaceBinaryRequest {
method,
path: path.into(),
body: Some(vec![0, 128, 255]),
max_response_bytes: 17,
}
}
#[test]
fn internal_children_deny_all_drive_transports_without_parent_side_effects() {
let inner = Arc::new(RecordingClient::default());
for reviewer in [false, true] {
let selected: Arc<dyn WorkspaceClient> = if reviewer {
Arc::new(ReviewerChildWorkspaceClient::new(
inner.clone(),
ReviewerContext {
ticket_id: "T1".into(),
merge_request_id: "MR1".into(),
},
"attestation".into(),
))
} else {
inner.clone()
};
let client = InternalChildWorkspaceClient::new(selected);
for path in [
"/api/w/ws/drive",
"/api/w/ws/drive?limit=1",
"/api/w/other/drive/root",
"/api/w/ws/drive/read-chunk?id=1",
"/api/w/ws/drive/upload",
"/api/w/ws/drive/requests/request-id",
"/api/w/ws/%64rive/root",
"/api/w/ws/tickets/../drive/root",
"/api/w/ws/tickets/%2e%2e/drive/root",
"/api/w/ws%2fdrive/root",
"/api/w/ws/dr\tive/root",
"/api/w/ws/dri\r\nve/root",
] {
for method in [
WorkspaceRequestMethod::Get,
WorkspaceRequestMethod::Post,
WorkspaceRequestMethod::Put,
WorkspaceRequestMethod::Patch,
WorkspaceRequestMethod::Delete,
] {
let request = WorkspaceRequest::json(method, path, "{}");
for result in [
client.execute(request.clone()),
client.execute_with_timeout(request, Duration::from_secs(2)),
] {
assert!(
matches!(result, Err(WorkspaceClientError::Unavailable(message)) if message.contains("Drive access denied")),
"path={path}, reviewer={reviewer}"
);
}
for result in [
client.execute_binary(binary(method, path)),
client.execute_binary_with_timeout(
binary(method, path),
Duration::from_secs(2),
),
] {
assert!(
matches!(result, Err(WorkspaceClientError::Unavailable(message)) if message.contains("Drive access denied")),
"path={path}, reviewer={reviewer}"
);
}
}
}
}
assert!(inner.calls.lock().unwrap().is_empty());
}
#[test]
fn non_drive_requests_preserve_payloads_deadlines_and_special_operations() {
let inner = Arc::new(RecordingClient::default());
let client = InternalChildWorkspaceClient::new(inner.clone());
let request = WorkspaceRequest::json(
WorkspaceRequestMethod::Post,
"/api/w/ws/tickets/query",
"{\"query\":\"drive\"}",
);
let bytes = binary(WorkspaceRequestMethod::Put, "/api/w/ws/files/content");
let timeout = Duration::from_millis(37);
let expected = WorkspaceResponse {
status: 202,
body: "forwarded".into(),
};
assert_eq!(client.execute(request.clone()).unwrap(), expected);
assert_eq!(
client
.execute_with_timeout(request.clone(), timeout)
.unwrap(),
expected
);
let expected_bytes = WorkspaceBinaryResponse {
status: 206,
body: vec![0, 255],
};
assert_eq!(
client.execute_binary(bytes.clone()).unwrap(),
expected_bytes
);
assert_eq!(
client
.execute_binary_with_timeout(bytes.clone(), timeout)
.unwrap(),
expected_bytes
);
assert!(
matches!(client.current_prompt_projection(Some(42)), Err(WorkspaceClientError::Request(message)) if message == "projection marker")
);
let discovery = WorkspaceWorkerDiscoveryRequest {
cursor: Some("next".into()),
limit: 3,
query: Some("coder".into()),
};
assert!(
matches!(client.list_workspace_workers(discovery.clone()), Err(WorkspaceClientError::Request(message)) if message == "discovery marker")
);
assert_eq!(
client
.execute_worker_remove("runtime", "worker", "reason")
.unwrap(),
expected
);
assert_eq!(
client
.execute_server_operation(WorkspaceServerOperation::WorkerControlList)
.unwrap(),
expected
);
assert_eq!(client.workspace_id(), Some("ws"));
assert_eq!(client.kind(), "recording");
assert!(client.is_available());
assert_eq!(
*inner.calls.lock().unwrap(),
vec![
Call::Json(request.clone(), None),
Call::Json(request, Some(timeout)),
Call::Binary(bytes.clone(), None),
Call::Binary(bytes, Some(timeout)),
Call::Projection(Some(42)),
Call::Discovery(discovery),
Call::Remove("runtime".into(), "worker".into(), "reason".into()),
Call::Server,
]
);
}
#[test]
fn drive_restriction_preserves_reviewer_context_and_review_attestation() {
let inner = Arc::new(RecordingClient::default());
let context = ReviewerContext {
ticket_id: "T1".into(),
merge_request_id: "MR1".into(),
};
let client =
InternalChildWorkspaceClient::new(Arc::new(ReviewerChildWorkspaceClient::new(
inner.clone(),
context.clone(),
"trusted-token".into(),
)));
assert_eq!(client.reviewer_context(), Some(&context));
assert_eq!(client.kind(), "runtime-reviewer-child");
let request = WorkspaceRequest::json(
WorkspaceRequestMethod::Post,
"/api/w/ws/merge-requests/MR1/reviews",
"{\"verdict\":\"approve\"}",
);
// Reviewer timeout execution uses its existing JSON attestation path, not raw parent access.
client
.execute_with_timeout(request, Duration::from_secs(1))
.unwrap();
let calls = inner.calls.lock().unwrap();
let [Call::Json(request, None)] = calls.as_slice() else {
panic!("unexpected calls: {calls:?}")
};
let body: serde_json::Value = serde_json::from_str(request.body.as_ref().unwrap()).unwrap();
assert_eq!(body["capability_token"], "trusted-token");
assert_eq!(body["verdict"], "approve");
drop(calls);
let count = inner.calls.lock().unwrap().len();
assert!(
client
.execute(WorkspaceRequest::json(
WorkspaceRequestMethod::Delete,
"/api/w/ws/tickets/T1",
"{}"
))
.is_err()
);
assert_eq!(inner.calls.lock().unwrap().len(), count);
}
}
@@ -0,0 +1,651 @@
use super::*;
use crate::worker::{WorkspaceClientError, WorkspaceRequest, WorkspaceResponse};
use serde::Serialize;
use std::collections::VecDeque;
#[derive(Debug)]
struct Client {
responses: Mutex<VecDeque<Result<WorkspaceResponse, WorkspaceClientError>>>,
requests: Mutex<Vec<WorkspaceRequest>>,
}
impl Client {
fn new(responses: Vec<WorkspaceResponse>) -> Arc<Self> {
Arc::new(Self {
responses: Mutex::new(responses.into_iter().map(Ok).collect()),
requests: Mutex::new(Vec::new()),
})
}
}
impl WorkspaceClient for Client {
fn workspace_id(&self) -> Option<&str> {
Some("workspace")
}
fn kind(&self) -> &str {
"drive-test"
}
fn is_available(&self) -> bool {
true
}
fn execute(
&self,
request: WorkspaceRequest,
) -> Result<WorkspaceResponse, WorkspaceClientError> {
self.requests.lock().unwrap().push(request);
self.responses
.lock()
.unwrap()
.pop_front()
.expect("unexpected Backend call")
}
}
fn response(value: impl Serialize) -> WorkspaceResponse {
WorkspaceResponse {
status: 200,
body: serde_json::to_string(&value).unwrap(),
}
}
fn entry(id: &str, revision: &str) -> DriveEntry {
DriveEntry {
entry: DriveEntryRef {
workspace_id: "workspace".into(),
node_id: id.into(),
},
parent: Some(DriveEntryRef {
workspace_id: "workspace".into(),
node_id: "1".into(),
}),
name: "note.md".into(),
kind: DriveEntryKind::File,
revision: revision.into(),
size: Some(5),
content_type: Some("text/markdown".into()),
updated_by: "worker".into(),
updated_at: "now".into(),
latest_url: format!("/api/w/workspace/drive/download?id={id}"),
}
}
fn feature(client: Arc<Client>) -> DriveFeature {
DriveFeature::new(client, Arc::new(WorkdirSessionRouter::new()))
}
async fn execute(
feature: &DriveFeature,
name: &str,
input: Value,
) -> Result<DriveOutput, DriveError> {
feature
.execute(
name,
&input.to_string(),
&ToolExecutionContext::new(name, "batch", 0),
None,
)
.await
}
#[test]
fn drive_feature_activation_never_creates_authority_and_schemas_hide_revisions() {
let client = Client::new(Vec::new());
let router = Arc::new(WorkdirSessionRouter::new());
assert!(DriveFeature::configured(client.clone(), false, router.clone()).is_none());
assert!(
DriveFeature::configured(
crate::worker::unavailable_workspace_client(None, "standalone"),
true,
router
)
.is_none()
);
let feature = feature(client);
for (name, definition) in feature.tools() {
let (meta, _) = definition();
assert_eq!(name, meta.name);
let schema = meta.input_schema.to_string();
assert!(!schema.contains("expected_revision"), "{name}");
assert!(!schema.contains("grant_id"), "{name}");
}
assert_eq!(feature.descriptor().instructions.len(), 1);
}
#[tokio::test]
async fn foreign_or_unobserved_entry_cannot_dispatch_mutation() {
let client = Client::new(Vec::new());
let feature = feature(client.clone());
let own = json!({"workspace_id":"workspace","node_id":"9007199254740993"});
assert!(matches!(
execute(&feature, "DriveWrite", json!({"entry":own,"content":"new"})).await,
Err(DriveError::Invalid(_))
));
assert!(matches!(
execute(
&feature,
"DriveDelete",
json!({"entry":{"workspace_id":"other","node_id":"1"}})
)
.await,
Err(DriveError::Denied)
));
for id in [json!(1), json!("01"), json!("9223372036854775808")] {
assert!(matches!(
execute(
&feature,
"DriveMetadata",
json!({"entry":{"workspace_id":"workspace","node_id":id}})
)
.await,
Err(DriveError::Invalid(_))
));
}
assert!(client.requests.lock().unwrap().is_empty());
}
#[tokio::test]
async fn stale_edit_returns_conflict_without_upgrading_revision_or_dispatching_write() {
let old = entry("2", "1");
let current = entry("2", "2");
let client = Client::new(vec![response(DriveReadTextResponse {
entry: current,
text: "hello".into(),
truncated: false,
})]);
let feature = feature(client.clone());
feature.observe(old.clone()).unwrap();
let result = execute(
&feature,
"DriveEdit",
json!({"entry":old.entry,"old_string":"hello","new_string":"new"}),
)
.await;
assert!(matches!(result, Err(DriveError::Conflict)));
assert_eq!(client.requests.lock().unwrap().len(), 1);
assert_eq!(feature.observed(&old.entry, None).unwrap().revision, "1");
}
#[tokio::test]
async fn shared_edit_rules_reject_ambiguous_and_truncated_preimages_without_mutation() {
for (text, truncated) in [("hello hello", false), ("hello", true)] {
let old = entry("2", "1");
let client = Client::new(vec![response(DriveReadTextResponse {
entry: old.clone(),
text: text.into(),
truncated,
})]);
let feature = feature(client.clone());
feature.observe(old.clone()).unwrap();
assert!(
execute(
&feature,
"DriveEdit",
json!({"entry":old.entry,"old_string":"hello","new_string":"new"})
)
.await
.is_err()
);
assert!(
client
.requests
.lock()
.unwrap()
.iter()
.all(|r| r.method == crate::worker::WorkspaceRequestMethod::Get)
);
}
}
#[tokio::test]
async fn successful_mutation_uses_observed_decimal_revision_and_requires_next_explicit_observation()
{
let old = entry("9007199254740993", "9007199254740993");
let updated = entry("9007199254740993", "9007199254740994");
let client = Client::new(Vec::new());
let feature = feature(client.clone());
feature.observe(old.clone()).unwrap();
let context = ToolExecutionContext::new("write", "batch", 0);
let id = feature.request_id(&context);
client
.responses
.lock()
.unwrap()
.push_back(Ok(response(DriveMutationResponse {
request_id: id,
entry: Some(updated),
})));
feature
.execute(
"DriveWrite",
&json!({"entry":old.entry,"content":"new"}).to_string(),
&context,
None,
)
.await
.unwrap();
let requests = client.requests.lock().unwrap();
let body: Value = serde_json::from_str(requests[0].body.as_ref().unwrap()).unwrap();
assert_eq!(body["mutation"]["expected_revision"], "9007199254740993");
assert_eq!(body["mutation"]["id"]["node_id"], "9007199254740993");
drop(requests);
assert!(feature.observed(&old.entry, None).is_err());
}
#[tokio::test]
async fn lost_mutation_response_queries_same_receipt_without_resending() {
let old = entry("2", "1");
let client = Client::new(Vec::new());
let feature = feature(client.clone());
feature.observe(old.clone()).unwrap();
let context = ToolExecutionContext::new("write", "batch", 0);
let id = feature.request_id(&context);
client.responses.lock().unwrap().extend([
Err(WorkspaceClientError::Request("lost".into())),
Ok(response(DriveRequestStatusResponse {
request_id: id.clone(),
state: DriveRequestState::Committed,
response: Some(DriveMutationResponse {
request_id: id.clone(),
entry: Some(entry("2", "2")),
}),
})),
]);
feature
.execute(
"DriveWrite",
&json!({"entry":old.entry,"content":"new"}).to_string(),
&context,
None,
)
.await
.unwrap();
let requests = client.requests.lock().unwrap();
assert_eq!(requests.len(), 2);
assert!(requests[1].path.ends_with(&format!("/requests/{id}")));
assert_eq!(
requests
.iter()
.filter(|r| r.method == crate::worker::WorkspaceRequestMethod::Post)
.count(),
1
);
}
#[tokio::test]
async fn uncommitted_receipt_snapshot_does_not_turn_unknown_into_failure_or_retry() {
let old = entry("2", "1");
let client = Client::new(Vec::new());
let feature = feature(client.clone());
feature.observe(old.clone()).unwrap();
let context = ToolExecutionContext::new("write", "batch", 0);
let id = feature.request_id(&context);
client.responses.lock().unwrap().extend([
Err(WorkspaceClientError::Request("lost".into())),
Ok(response(DriveRequestStatusResponse {
request_id: id.clone(),
state: DriveRequestState::Uncommitted,
response: None,
})),
]);
let result = feature
.execute(
"DriveWrite",
&json!({"entry":old.entry,"content":"new"}).to_string(),
&context,
None,
)
.await;
assert!(matches!(result,Err(DriveError::OutcomeUnknown(ref observed)) if observed==&id));
assert_eq!(client.requests.lock().unwrap().len(), 2);
}
fn invalid_completions(id: &str) -> Vec<DriveMutationResponse> {
let mut invalid_entry = entry("2", "2");
invalid_entry.revision = "not-a-revision".into();
vec![
DriveMutationResponse {
request_id: "another-request".into(),
entry: Some(entry("2", "2")),
},
DriveMutationResponse {
request_id: id.into(),
entry: Some(invalid_entry),
},
]
}
#[tokio::test]
async fn invalid_mutation_completion_recovers_from_original_receipt_without_resending() {
let context = ToolExecutionContext::new("write", "batch", 0);
let client = Client::new(vec![]);
let feature = feature(client.clone());
let id = feature.request_id(&context);
for completion in invalid_completions(&id) {
let committed = DriveMutationResponse {
request_id: id.clone(),
entry: Some(entry("2", "2")),
};
client.requests.lock().unwrap().clear();
client.responses.lock().unwrap().extend([
Ok(response(completion)),
Ok(response(DriveRequestStatusResponse {
request_id: id.clone(),
state: DriveRequestState::Committed,
response: Some(committed.clone()),
})),
]);
feature.observe(entry("2", "1")).unwrap();
let output = feature
.execute(
"DriveWrite",
&json!({"entry":entry("2", "1").entry,"content":"new"}).to_string(),
&context,
None,
)
.await
.unwrap();
assert_eq!(output.value["request_id"], id);
assert_eq!(output.value["entry"]["metadata"]["revision"], "2");
let requests = client.requests.lock().unwrap();
assert_eq!(requests.len(), 2);
assert_eq!(
requests[0].method,
crate::worker::WorkspaceRequestMethod::Post
);
assert_eq!(
requests[1].method,
crate::worker::WorkspaceRequestMethod::Get
);
assert!(requests[1].path.ends_with(&format!("/requests/{id}")));
let body: Value = serde_json::from_str(requests[0].body.as_ref().unwrap()).unwrap();
assert_eq!(body["request_id"], id);
assert_eq!(body["mutation"]["expected_revision"], "1");
}
}
#[tokio::test]
async fn invalid_completion_and_unresolved_or_invalid_receipt_stay_unknown_after_one_query() {
let context = ToolExecutionContext::new("write", "batch", 0);
let client = Client::new(vec![]);
let feature = feature(client.clone());
let id = feature.request_id(&context);
let committed = DriveMutationResponse {
request_id: id.clone(),
entry: Some(entry("2", "2")),
};
let mut receipts = vec![
response(DriveRequestStatusResponse {
request_id: id.clone(),
state: DriveRequestState::Uncommitted,
response: None,
}),
response(DriveRequestStatusResponse {
request_id: id.clone(),
state: DriveRequestState::Committed,
response: None,
}),
response(DriveRequestStatusResponse {
request_id: "another-request".into(),
state: DriveRequestState::Committed,
response: Some(committed),
}),
WorkspaceResponse {
status: 200,
body: "invalid-json".into(),
},
];
receipts.extend(invalid_completions(&id).into_iter().map(|completion| {
response(DriveRequestStatusResponse {
request_id: id.clone(),
state: DriveRequestState::Committed,
response: Some(completion),
})
}));
for completion in invalid_completions(&id) {
for receipt in &receipts {
client.requests.lock().unwrap().clear();
client
.responses
.lock()
.unwrap()
.extend([Ok(response(&completion)), Ok(receipt.clone())]);
feature.observe(entry("2", "1")).unwrap();
let result = feature
.execute(
"DriveWrite",
&json!({"entry":entry("2", "1").entry,"content":"new"}).to_string(),
&context,
None,
)
.await;
assert!(
matches!(result, Err(DriveError::OutcomeUnknown(ref observed)) if observed == &id)
);
let requests = client.requests.lock().unwrap();
assert_eq!(requests.len(), 2);
assert_eq!(
requests[0].method,
crate::worker::WorkspaceRequestMethod::Post
);
assert_eq!(
requests[1].method,
crate::worker::WorkspaceRequestMethod::Get
);
assert!(requests[1].path.ends_with(&format!("/requests/{id}")));
}
}
}
#[tokio::test]
async fn observation_eviction_never_substitutes_another_entry() {
let feature = feature(Client::new(Vec::new()));
let first = entry("2", "1");
feature.observe(first.clone()).unwrap();
for id in 3..=258 {
feature.observe(entry(&id.to_string(), "1")).unwrap();
}
assert!(feature.observed(&first.entry, None).is_err());
let state = feature.observations.lock().unwrap();
assert_eq!(state.entries.len(), MAX_OBSERVATIONS);
assert_eq!(state.order.len(), MAX_OBSERVATIONS);
}
#[tokio::test]
async fn backend_error_details_are_not_reflected_and_conflict_is_not_retried() {
let client = Client::new(vec![WorkspaceResponse {
status: 409,
body: serde_json::to_string(&DriveApiError {
code: DriveApiErrorCode::Conflict,
classification: DriveFailureClassification::NotCommitted,
message: "/secret/provider/path".into(),
})
.unwrap(),
}]);
let feature = feature(client.clone());
let old = entry("2", "1");
feature.observe(old.clone()).unwrap();
let result = execute(&feature, "DriveDelete", json!({"entry":old.entry}))
.await
.err()
.unwrap();
assert!(matches!(result, DriveError::Conflict));
assert!(!result.to_string().contains("secret"));
assert_eq!(client.requests.lock().unwrap().len(), 1);
}
#[tokio::test]
async fn drive_native_entry_is_directly_resolvable_but_never_indexable_and_binds_identity() {
use crate::wip::WipSubtreeProvider;
let entry = entry("9007199254740993", "3");
let client = Client::new(vec![response(entry.clone())]);
let feature = feature(client.clone());
let mut mounts = crate::wip::WipMountRegistry::new();
wip::mount_drive_wip(&mut mounts, &feature).unwrap();
let provider = wip::DriveProvider(feature);
let path = entry_path(&entry.entry);
let publication = provider.publication(&path).await.unwrap().unwrap();
assert_eq!(publication.projection.route, path);
assert_eq!(
publication.scope.as_ref().unwrap().r#ref,
publication.projection.object.r#ref
);
assert_eq!(publication.projection.object.validator, Some(b"3".to_vec()));
assert!(provider.children("/drive").await.unwrap().is_empty());
assert!(
provider
.publication("/drive/other/9007199254740993")
.await
.unwrap()
.is_none()
);
let operations = &publication.projection.descriptor.operations;
assert!(operations.iter().any(|op| op.name == "view_image"));
assert!(
!operations
.iter()
.flat_map(|op| &op.parameters)
.any(|parameter| parameter.name == "expected_revision")
);
assert_eq!(client.requests.lock().unwrap().len(), 1);
}
#[derive(Debug)]
struct ChunkClient {
json: Arc<Client>,
chunks: Mutex<VecDeque<crate::worker::WorkspaceBinaryResponse>>,
requests: Mutex<Vec<crate::worker::WorkspaceBinaryRequest>>,
}
impl WorkspaceClient for ChunkClient {
fn workspace_id(&self) -> Option<&str> {
Some("workspace")
}
fn kind(&self) -> &str {
"drive-chunk-test"
}
fn is_available(&self) -> bool {
true
}
fn execute(
&self,
request: WorkspaceRequest,
) -> Result<WorkspaceResponse, WorkspaceClientError> {
self.json.execute(request)
}
fn execute_binary(
&self,
request: crate::worker::WorkspaceBinaryRequest,
) -> Result<crate::worker::WorkspaceBinaryResponse, WorkspaceClientError> {
self.requests.lock().unwrap().push(request);
Ok(self
.chunks
.lock()
.unwrap()
.pop_front()
.expect("unexpected binary read"))
}
}
fn chunk_client(
metadata: DriveEntry,
chunks: Vec<crate::worker::WorkspaceBinaryResponse>,
) -> Arc<ChunkClient> {
Arc::new(ChunkClient {
json: Client::new(vec![response(metadata)]),
chunks: Mutex::new(chunks.into()),
requests: Mutex::new(Vec::new()),
})
}
fn chunk(bytes: Vec<u8>) -> crate::worker::WorkspaceBinaryResponse {
crate::worker::WorkspaceBinaryResponse {
status: 200,
body: bytes,
}
}
#[tokio::test]
async fn multi_chunk_image_uses_one_revision_and_never_exposes_binary_in_text_content() {
let mut metadata = entry("2", "9007199254740993");
let mut bytes = b"\x89PNG\r\n\x1a\n".to_vec();
bytes.resize(DRIVE_CHUNK_MAX_BYTES as usize + 13, 42);
metadata.size = Some(bytes.len() as u32);
let client = chunk_client(
metadata.clone(),
vec![
chunk(bytes[..DRIVE_CHUNK_MAX_BYTES as usize].to_vec()),
chunk(bytes[DRIVE_CHUNK_MAX_BYTES as usize..].to_vec()),
],
);
let feature = DriveFeature::new(client.clone(), Arc::new(WorkdirSessionRouter::new()));
let output = execute(&feature, "DriveViewImage", json!({"entry":metadata.entry}))
.await
.unwrap();
let Attachment::Image(image) = &output.attachments[0];
assert_eq!(image.data(), bytes);
assert!(output.value.to_string().len() < 2048);
let requests = client.requests.lock().unwrap();
assert_eq!(requests.len(), 2);
for request in &*requests {
assert!(request.path.contains("expected_revision=9007199254740993"));
assert_eq!(request.max_response_bytes, DRIVE_CHUNK_MAX_BYTES as usize);
}
assert!(requests[0].path.split('&').any(|pair| pair == "offset=0"));
assert!(
requests[1]
.path
.split('&')
.any(|pair| pair == "offset=65536")
);
}
#[tokio::test]
async fn expired_revision_or_partial_chunk_aborts_image_without_rebinding_or_retrying() {
for next in [
crate::worker::WorkspaceBinaryResponse {
status: 409,
body: serde_json::to_vec(&DriveApiError::new(DriveApiErrorCode::Conflict)).unwrap(),
},
chunk(vec![1]), // second range requires 13 bytes, not a successful prefix
] {
let mut metadata = entry("2", "3");
metadata.size = Some(DRIVE_CHUNK_MAX_BYTES + 13);
let mut first = vec![0; DRIVE_CHUNK_MAX_BYTES as usize];
first[..8].copy_from_slice(b"\x89PNG\r\n\x1a\n");
let client = chunk_client(metadata.clone(), vec![chunk(first), next]);
let feature = DriveFeature::new(client.clone(), Arc::new(WorkdirSessionRouter::new()));
assert!(
execute(&feature, "DriveViewImage", json!({"entry":metadata.entry}))
.await
.is_err()
);
let requests = client.requests.lock().unwrap();
assert_eq!(
requests.len(),
2,
"must not retry an expired/truncated image"
);
assert!(
requests
.iter()
.all(|request| request.path.contains("expected_revision=3&"))
);
assert_eq!(
client.json.requests.lock().unwrap().len(),
1,
"must not fetch a new revision"
);
}
}
#[tokio::test]
async fn oversized_images_and_unknown_arguments_fail_before_binary_transfer() {
let mut metadata = entry("2", "1");
metadata.size = Some(tools::view_image::MAX_IMAGE_BYTES as u32 + 1);
let client = chunk_client(metadata.clone(), vec![]);
let feature = DriveFeature::new(client.clone(), Arc::new(WorkdirSessionRouter::new()));
assert!(matches!(
execute(&feature, "DriveViewImage", json!({"entry":metadata.entry})).await,
Err(DriveError::Limit)
));
assert!(client.requests.lock().unwrap().is_empty());
assert!(
execute(
&feature,
"DriveRead",
json!({"entry":metadata.entry,"expected_revision":"1"})
)
.await
.is_err()
);
assert_eq!(client.json.requests.lock().unwrap().len(), 1);
}
@@ -0,0 +1,363 @@
//! Native non-indexable Drive entries. Publication resolves just one ID through
//! current Backend authorization; the Client owns identity/revision validators.
use super::{
DriveFeature, SPECS, Spec,
backend::{DriveError, decimal},
digest, entry_path,
};
use crate::wip::{
WipCallContext, WipMountError, WipMountRegistry, WipOperationError, WipOperationHandler,
WipOperationOutput, WipProjection, WipProjectionKind, WipPublication, WipSubtreeMount,
WipSubtreeProvider, contextual_reference, json_to_wip, wip_to_json,
};
use async_trait::async_trait;
use serde_json::json;
use server_api::{DriveEntry, DriveEntryKind, DriveEntryRef};
use std::{collections::BTreeMap, sync::Arc};
use wip_protocol::{
Documentation, INTERFACE_FORMAT_V1, InterfaceDescriptor, Object, OperationDeclaration,
ParameterDeclaration, ProtocolError, ProtocolErrorCode, ReturnDeclaration, TypeExpr, Value,
};
const ROOT: &str = "/drive";
const INTERFACE: &str = "yoi.drive/entry/v1";
pub fn mount_drive_wip(
registry: &mut WipMountRegistry,
feature: &DriveFeature,
) -> Result<(), WipMountError> {
registry.allocate_namespace("drive", "drive")?;
// Registration is only a namespace entrance. The subtree provides the
// current target/scope/descriptor/handler in a single self-scoped snapshot.
let names = root_specs();
let descriptor = descriptor(&names, true);
registry.mount(WipProjection {
route: ROOT.into(),
capability: "drive:content".into(),
kind: WipProjectionKind::Native,
object: Object {
name: "drive".into(),
description: Some(
"Authorized Workspace Drive; discover entries via bounded list/search, not tree"
.into(),
),
interfaces: vec![contextual_reference(INTERFACE, ROOT)],
r#ref: None,
validator: None,
},
interface: contextual_reference(INTERFACE, ROOT),
interface_validator: Some(digest(format!("{descriptor:?}").as_bytes()).into_bytes()),
descriptor,
handler: Arc::new(Unresolved),
})?;
registry.mount_subtree(WipSubtreeMount {
root: ROOT.into(),
provider: Arc::new(DriveProvider(feature.clone())),
})?;
registry.replace_compatibility_tools(ROOT, SPECS.iter().map(|spec| spec.name))?;
Ok(())
}
struct Unresolved;
#[async_trait]
impl WipOperationHandler for Unresolved {
async fn call(
&self,
_: &str,
_: &BTreeMap<String, Value>,
_: WipCallContext,
) -> Result<WipOperationOutput, WipOperationError> {
Err(map_error(DriveError::Denied))
}
}
pub(super) struct DriveProvider(pub(super) DriveFeature);
#[async_trait]
impl WipSubtreeProvider for DriveProvider {
// Arbitrary tree depth is safe: there are deliberately no indexable entries.
fn max_depth(&self) -> u32 {
32
}
fn max_nodes(&self) -> usize {
1
}
async fn publication(&self, path: &str) -> Result<Option<WipPublication>, ProtocolError> {
let result = if path == ROOT {
self.0.backend.root().await
} else {
let Some(entry) = parse_entry(&self.0, path).map_err(protocol)? else {
return Ok(None);
};
self.0.backend.metadata(&entry).await
};
let entry = match result {
Ok(entry) => entry,
Err(DriveError::Denied | DriveError::NotFound) => return Ok(None),
Err(error) => return Err(protocol(error)),
};
let root = path == ROOT;
let specs = if root {
root_specs()
} else {
entry_specs(&entry)
};
let descriptor = descriptor(&specs, root);
let reference = contextual_reference(INTERFACE, path);
let validator = entry.revision.as_bytes().to_vec();
let object = Object {
name: if root {
"drive".into()
} else {
entry.entry.node_id.clone()
},
description: Some(format!(
"Workspace Drive {:?} {:?}; stable ID {}, bodies only through bounded operations",
entry.kind, entry.name, entry.entry.node_id
)),
interfaces: vec![reference.clone()],
r#ref: Some(format!(
"drive:{}:{}",
entry.entry.workspace_id, entry.entry.node_id
)),
validator: Some(validator),
};
let projection = WipProjection {
route: path.into(),
capability: "drive:content".into(),
kind: WipProjectionKind::Native,
object,
interface: reference,
interface_validator: Some(digest(format!("{descriptor:?}").as_bytes()).into_bytes()),
descriptor,
handler: Arc::new(DriveHandler {
feature: self.0.clone(),
entry,
specs,
root,
}),
};
WipPublication::self_scoped(projection).map(Some)
}
async fn children(&self, _: &str) -> Result<Vec<String>, ProtocolError> {
Ok(Vec::new())
}
}
fn root_specs() -> Vec<Spec> {
SPECS
.iter()
.copied()
.filter(|spec| {
matches!(
spec.operation,
"root"
| "metadata"
| "list"
| "search"
| "create_folder"
| "create_text"
| "save_workdir"
| "request_status"
)
})
.collect()
}
fn entry_specs(entry: &DriveEntry) -> Vec<Spec> {
SPECS
.iter()
.copied()
.filter(|spec| match spec.operation {
"metadata" => true,
"relocate" | "delete" => entry.parent.is_some(),
"list" | "create_folder" | "create_text" | "save_workdir" => {
entry.kind == DriveEntryKind::Folder
}
"read" | "write" | "edit" | "view_image" => entry.kind == DriveEntryKind::File,
_ => false,
})
.collect()
}
fn parameter(name: &str, required: bool, r#type: TypeExpr) -> ParameterDeclaration {
ParameterDeclaration {
name: name.into(),
required,
documentation: None,
r#type,
}
}
fn descriptor(specs: &[Spec], _root: bool) -> InterfaceDescriptor {
InterfaceDescriptor {format:INTERFACE_FORMAT_V1.into(),documentation:Some(Documentation {summary:"Backend-authorized Workspace Drive operations".into(),details:Some("Entry IDs are Workspace-bound. The Host/Client manage revisions; never supply validators or retry conflicts/outcome-unknown with a new revision. Read-only grants reject mutations at Backend. List/search return paths for direct Inspect, never tree inventories. Image bytes attach to durable ToolOutput; URLs are not image viewing. Workdir source read and Drive destination write are independent.".into())}),types:Vec::new(),operations:specs.iter().map(|spec|OperationDeclaration {name:spec.operation.into(),documentation:Some(Documentation {summary:spec.description.into(),details:None}),parameters:match spec.operation {
"list"=>vec![parameter("limit",false,TypeExpr::Integer),parameter("after",false,TypeExpr::String)],
"search"=>vec![parameter("query",true,TypeExpr::String),parameter("include_text",false,TypeExpr::Boolean),parameter("limit",false,TypeExpr::Integer),parameter("after",false,TypeExpr::String)],
"read"=>vec![parameter("max_bytes",false,TypeExpr::Integer)],
"write"|"edit"=>crate::file_operation::parameters(spec.operation,false).expect("shared text arguments"),
"create_text"=>vec![parameter("name",true,TypeExpr::String),parameter("content",true,TypeExpr::String),parameter("content_type",false,TypeExpr::String)],
"create_folder"=>vec![parameter("name",true,TypeExpr::String)],
"relocate"=>vec![parameter("parent",true,TypeExpr::Entry),parameter("name",true,TypeExpr::String)],
"save_workdir"=>vec![parameter("target_workdir",true,TypeExpr::String),parameter("path",true,TypeExpr::String),parameter("name",true,TypeExpr::String),parameter("content_type",false,TypeExpr::String)],
"request_status"=>vec![parameter("request_id",true,TypeExpr::String)],
_=>vec![],
},returns:ReturnDeclaration {documentation:None,r#type:TypeExpr::Json}}).collect()}
}
struct DriveHandler {
feature: DriveFeature,
entry: DriveEntry,
specs: Vec<Spec>,
root: bool,
}
#[async_trait]
impl WipOperationHandler for DriveHandler {
async fn call(
&self,
operation: &str,
arguments: &BTreeMap<String, Value>,
context: WipCallContext,
) -> Result<WipOperationOutput, WipOperationError> {
let spec = self
.specs
.iter()
.find(|spec| spec.operation == operation)
.ok_or_else(|| {
failure(
ProtocolErrorCode::OperationNotFound,
"Drive operation is not published",
)
})?;
let declaration = descriptor(&self.specs, self.root)
.operations
.into_iter()
.find(|op| op.name == operation)
.unwrap();
if arguments
.keys()
.any(|key| !declaration.parameters.iter().any(|p| p.name == *key))
{
return Err(map_error(DriveError::Invalid(
"unrecognized arguments; revision metadata is Client-managed".into(),
)));
}
let mut input = wip_to_json(&Value::Record(arguments.clone()))
.map_err(|_| map_error(DriveError::Invalid("invalid WIP arguments".into())))?;
let fields = input
.as_object_mut()
.ok_or_else(|| map_error(DriveError::Invalid("named arguments required".into())))?;
match operation {
"metadata" | "read" | "write" | "edit" | "delete" | "view_image" => {
fields.insert("entry".into(), json!(self.entry.entry));
}
"list" | "create_folder" | "create_text" | "save_workdir" => {
fields.insert("parent".into(), json!(self.entry.entry));
}
"relocate" => {
let path = match arguments.get("parent") {
Some(Value::String(path)) => path.as_str(),
_ => {
return Err(map_error(DriveError::Invalid(
"parent must be a Drive entry".into(),
)));
}
};
let parent = if path == ROOT {
self.feature.backend.root().await.map_err(map_error)?.entry
} else {
parse_entry(&self.feature, path)
.map_err(map_error)?
.ok_or_else(|| map_error(DriveError::Denied))?
};
fields.insert("parent".into(), json!(parent));
fields.insert("entry".into(), json!(self.entry.entry));
}
_ => {}
}
if let Some(permissions) = &self.feature.permissions {
if !matches!(
crate::permission::permission_action_for(permissions, spec.name, &input),
manifest::ToolPermissionAction::Allow
) {
return Err(failure(
ProtocolErrorCode::PermissionDenied,
"Drive operation denied by tool permission; approval unavailable in native invocation",
));
}
}
let output = self
.feature
.execute(
spec.name,
&input.to_string(),
&context.execution,
Some(&self.entry),
)
.await
.map_err(map_error)?;
let value = json_to_wip(&output.value).map_err(|_| {
WipOperationError::OutcomeUnknown(
"Completed Drive result unavailable; query request_status before another mutation"
.into(),
)
})?;
// Attachments use the existing ToolOutput side channel and durable Engine
// capture, not transient request injection or URL-only success.
let result = if matches!(operation, "write" | "edit" | "relocate") {
let revision = output.value["entry"]["metadata"]["revision"]
.as_str()
.ok_or_else(|| {
WipOperationError::OutcomeUnknown("Committed Drive revision unavailable".into())
})?;
WipOperationOutput::native_with_validator(value, revision.as_bytes().to_vec())
} else {
WipOperationOutput::native(value)
};
Ok(result.with_attachments(output.attachments))
}
}
fn parse_entry(feature: &DriveFeature, path: &str) -> Result<Option<DriveEntryRef>, DriveError> {
let workspace = feature.backend.workspace()?;
let prefix = format!(
"/drive/{}/",
crate::feature::builtin::manage_workdir::wip::encode_identity(workspace)
);
let Some(id) = path.strip_prefix(&prefix) else {
return Ok(None);
};
if !decimal(id) {
return Ok(None);
}
let entry = DriveEntryRef {
workspace_id: workspace.into(),
node_id: id.into(),
};
if entry_path(&entry) != path {
return Ok(None);
}
Ok(Some(entry))
}
fn failure(code: ProtocolErrorCode, message: &str) -> WipOperationError {
WipOperationError::Protocol(ProtocolError {
code,
message: message.into(),
})
}
fn map_error(error: DriveError) -> WipOperationError {
let code = match &error {
DriveError::Denied => ProtocolErrorCode::PermissionDenied,
DriveError::NotFound => ProtocolErrorCode::NotFound,
DriveError::Conflict => ProtocolErrorCode::ValidatorMismatch,
DriveError::Invalid(_) => ProtocolErrorCode::InvalidArguments,
DriveError::Limit => ProtocolErrorCode::ResourceLimitExceeded,
DriveError::Unavailable => ProtocolErrorCode::Internal,
DriveError::OutcomeUnknown(_) => {
return WipOperationError::OutcomeUnknown(error.to_string());
}
};
WipOperationError::Protocol(ProtocolError {
code,
message: error.to_string(),
})
}
fn protocol(error: DriveError) -> ProtocolError {
match map_error(error) {
WipOperationError::Protocol(error) => error,
_ => ProtocolError {
code: ProtocolErrorCode::Internal,
message: "Drive observation unavailable".into(),
},
}
}
+2
View File
@@ -284,6 +284,7 @@ fn validate_profile_with_grant(
manage_workdir,
workdir_catalog,
workspace_config,
drive,
ticket,
merge_request,
orchestration,
@@ -315,6 +316,7 @@ fn validate_profile_with_grant(
("feature.objective", objective.enabled),
("feature.manage_workdir", manage_workdir.enabled),
("feature.workspace_config", workspace_config.enabled),
("feature.drive", drive.enabled),
(
"feature.ticket",
ticket.enabled || ticket.authoring || ticket.thread || ticket.intake || ticket.workflow,
+5 -5
View File
@@ -65,9 +65,9 @@ pub use session_history::{
pub use shared_state::WorkerSharedState;
pub use worker::{
LocalWorkingDirectory, SubjektivSessionAttributionLifecycle, Worker, WorkerError,
WorkerFilesystemAuthority, WorkerRunResult, WorkerWorkspaceContext, WorkspaceClient,
WorkspaceClientError, WorkspaceId, WorkspaceIdError, WorkspacePromptCatalogResolution,
WorkspaceRequest, WorkspaceRequestMethod, WorkspaceResponse, WorkspaceServerOperation,
WorkspaceWorkerDiscoveryRequest, apply_worker_manifest, marker_workspace_client,
unavailable_workspace_client,
WorkerFilesystemAuthority, WorkerRunResult, WorkerWorkspaceContext, WorkspaceBinaryRequest,
WorkspaceBinaryResponse, WorkspaceClient, WorkspaceClientError, WorkspaceId, WorkspaceIdError,
WorkspacePromptCatalogResolution, WorkspaceRequest, WorkspaceRequestMethod, WorkspaceResponse,
WorkspaceServerOperation, WorkspaceWorkerDiscoveryRequest, apply_worker_manifest,
marker_workspace_client, unavailable_workspace_client,
};
+190
View File
@@ -28,6 +28,7 @@ use workdir::{
use crate::PromptCatalogSource;
use crate::controller::register_worker_tools;
use crate::feature::builtin::drive::delegation::restrict_internal_child_drive;
use crate::internal_worker::{
EphemeralSessionStore, InternalWorkerSessionStatus, InternalWorkerVisibility,
prepare_internal_worker_session,
@@ -327,6 +328,9 @@ pub struct SubWorkerSpawnTool {
/// Compact selector list shared by tool description and diagnostics.
available_profiles: AvailableProfiles,
internal_client_override: Option<Box<dyn agen::llm_client::LlmClient>>,
#[cfg(test)]
installed_child_workspace_client:
std::sync::Mutex<Option<Arc<dyn crate::worker::WorkspaceClient>>>,
}
impl SubWorkerSpawnTool {
@@ -362,6 +366,8 @@ impl SubWorkerSpawnTool {
prompt_loader,
available_profiles,
internal_client_override: None,
#[cfg(test)]
installed_child_workspace_client: std::sync::Mutex::new(None),
}
}
}
@@ -482,6 +488,9 @@ impl Tool for SubWorkerSpawnTool {
// Delegated children stay bound to their scoped session and cannot use
// Workspace attachment tools to replace it with parent-level authority.
child_manifest.feature.manage_workdir.enabled = false;
// Internal children (including Reviewers) have no Backend Drive identity or approved
// child scope. Profiles and the parent's current Drive grant cannot delegate it.
child_manifest.feature.drive.enabled = false;
let reviewer_capability = input.review.as_ref().map(|review| {
(
review.ticket_id.clone(),
@@ -523,6 +532,12 @@ impl Tool for SubWorkerSpawnTool {
} else {
self.workspace_context.clone()
};
let child_workspace_context = restrict_internal_child_drive(child_workspace_context);
#[cfg(test)]
{
*self.installed_child_workspace_client.lock().unwrap() =
Some(child_workspace_context.client_handle());
}
let store = EphemeralSessionStore::default();
let filesystem_authority = WorkerFilesystemAuthority::None;
let mut child = Worker::<Box<dyn agen::llm_client::LlmClient>, EphemeralSessionStore>::from_internal_manifest_with_context(
@@ -1576,6 +1591,181 @@ enabled = false
assert!(spawner_scope.snapshot().is_writable(&workspace_root));
}
#[derive(Debug, Default)]
struct DriveAllowingParentClient {
requests: Mutex<Vec<WorkspaceRequest>>,
}
impl WorkspaceClient for DriveAllowingParentClient {
fn workspace_id(&self) -> Option<&str> {
Some("workspace-test")
}
fn kind(&self) -> &str {
"drive-allowing-parent"
}
fn is_available(&self) -> bool {
true
}
fn execute(
&self,
request: WorkspaceRequest,
) -> Result<WorkspaceResponse, WorkspaceClientError> {
self.requests.lock().unwrap().push(request);
Ok(WorkspaceResponse {
status: 200,
body: "{}".into(),
})
}
}
#[tokio::test]
async fn internal_spawn_profiles_never_inherit_parent_drive_tools_or_client_authority() {
for (profile, review) in [
("inherit", false),
("default", false),
("builtin:reviewer", true),
] {
let runtime = TempDir::new().unwrap();
let workspace_root = runtime.path().join("project");
let bash_output_dir = runtime.path().join("bash-output");
let drive_profile =
format!("{INTERNAL_REVIEWER_PROFILE}\n[feature.drive]\nenabled = true\n");
let available_profiles = write_project_profile_registry(
&workspace_root,
Some("reviewer"),
&[("reviewer", "reviewer.toml", &drive_profile)],
);
let mut manifest = parent_manifest(&workspace_root, None);
manifest.feature.drive.enabled = true;
manifest
.scope
.allow
.push(abs_rule(&bash_output_dir, Permission::Read));
let spawner_scope = SharedScope::new(Scope::from_config(&manifest.scope).unwrap());
let registry =
SpawnedWorkerRegistry::new_internal("parent".into(), spawner_scope.clone());
let parent_client = Arc::new(DriveAllowingParentClient::default());
let drive_request = WorkspaceRequest::get("/api/w/workspace-test/drive/root");
assert!(
parent_client
.execute(drive_request.clone())
.unwrap()
.is_success()
);
let workdir_tool_broker = workdir::WorkdirToolBrokerRouter::from_single(
workdir::WorkdirAttachmentAlias::new("workdir").unwrap(),
Arc::new(workdir::LocalWorkdirSession::materialized_bound(
workdir::Workdir::new("test-workdir"),
workspace_root.clone(),
workspace_root.clone(),
spawner_scope.clone(),
workdir::WorkdirSessionCapabilities::ALL,
)),
)
.unwrap();
let calls = Arc::new(AtomicUsize::new(0));
let tool = SubWorkerSpawnTool::new(
"parent".into(),
crate::worker::WorkerWorkspaceContext::with_client(
Some(WorkspaceId::new("workspace-test").unwrap()),
parent_client.clone(),
),
ParentNotificationTarget::Durable(Arc::new(|_| {})),
runtime.path().to_path_buf(),
bash_output_dir,
workspace_root.clone(),
Some(workdir_tool_broker),
registry.clone(),
manifest.clone(),
PromptCatalogSource::builtins_only(),
available_profiles,
)
.with_internal_client(Box::new(ScriptedInternalClient {
calls: calls.clone(),
parent_scope: spawner_scope,
delegated_path: workspace_root,
observed_parent_write_revoked: Arc::new(AtomicBool::new(false)),
observed_instruction_override: Arc::new(AtomicBool::new(false)),
fail_requests: Arc::new(AtomicBool::new(false)),
}));
let mut input = serde_json::json!({
"name": "child", "profile": profile, "instruction": "role.reviewer",
"task": "review", "scope": [{"target": ".", "permission": "write"}],
"command": true,
});
if review {
input["review"] = serde_json::json!({"ticket_id": "T1", "merge_request_id": "MR1"});
}
tool.execute(
&input.to_string(),
agen::tool::ToolExecutionContext::direct(),
)
.await
.unwrap_or_else(|error| panic!("profile={profile}: {error:?}"));
let record = registry.get_internal("child").unwrap();
assert_eq!(
record.session.wait_until_idle().await,
InternalWorkerSessionStatus::Idle
);
assert_eq!(calls.load(Ordering::SeqCst), 1);
assert!(
manifest.feature.drive.enabled,
"parent must keep Drive enabled"
);
assert!(
!record
.installed_tools
.iter()
.any(|name| name.starts_with("Drive")),
"profile={profile}, installed tools={:?}",
record.installed_tools
);
let child_client = tool
.installed_child_workspace_client
.lock()
.unwrap()
.clone()
.unwrap();
let before = parent_client.requests.lock().unwrap().len();
assert!(
matches!(child_client.execute(drive_request), Err(WorkspaceClientError::Unavailable(message)) if message.contains("Drive access denied")),
"profile={profile}"
);
assert_eq!(parent_client.requests.lock().unwrap().len(), before);
if review {
assert_eq!(
child_client.reviewer_context(),
Some(&ReviewerContext {
ticket_id: "T1".into(),
merge_request_id: "MR1".into(),
})
);
child_client
.execute(WorkspaceRequest::json(
WorkspaceRequestMethod::Post,
"/api/w/workspace-test/merge-requests/MR1/reviews",
"{\"verdict\":\"approve\"}",
))
.unwrap();
let requests = parent_client.requests.lock().unwrap();
let review_body: serde_json::Value =
serde_json::from_str(requests.last().unwrap().body.as_ref().unwrap()).unwrap();
assert!(
review_body["capability_token"]
.as_str()
.is_some_and(|token| !token.is_empty())
);
assert!(
requests
.iter()
.any(|request| request.path.ends_with("/review-capabilities"))
);
}
record.session.stop().await.unwrap();
registry.remove_internal("child").await.unwrap();
}
}
#[tokio::test]
async fn remote_subworker_spawn_keeps_worker_host_output_path_out_of_workdir_scope() {
let runtime = TempDir::new().unwrap();
+334 -25
View File
@@ -14,7 +14,9 @@ use std::sync::{Arc, Mutex};
use agen::Engine;
use agen::llm_client::client::LlmClient;
use agen::state::Mutable;
use agen::tool::{Tool, ToolDefinition, ToolError, ToolExecutionContext, ToolMeta, ToolOutput};
use agen::tool::{
Attachment, Tool, ToolDefinition, ToolError, ToolExecutionContext, ToolMeta, ToolOutput,
};
use async_trait::async_trait;
use base64::Engine as _;
use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
@@ -99,8 +101,8 @@ pub struct WipCallContext {
pub security_context: String,
}
/// Successful operation value. Compatibility projections can retain an ordinary
/// ToolOutput so images and pruning semantics remain on the existing Engine path.
/// Successful operation value. Native attachments and compatibility ToolOutputs
/// use the existing Engine side channel, separate from the validated WIP value.
pub struct WipOperationOutput {
pub value: Value,
tool_output: Option<ToolOutput>,
@@ -125,6 +127,20 @@ impl WipOperationOutput {
}
}
/// Attach model-visible binary content without replacing the native result
/// or its exact post-operation validator. Invoke returns these attachments
/// through ToolOutput so Engine capture/history can persist their bytes.
pub fn with_attachments(mut self, attachments: Vec<Attachment>) -> Self {
self.tool_output
.get_or_insert_with(|| ToolOutput {
summary: String::new(),
content: None,
attachments: Vec::new(),
})
.attachments = attachments;
self
}
fn compatibility(value: Value, output: ToolOutput) -> Self {
Self {
value,
@@ -134,6 +150,13 @@ impl WipOperationOutput {
}
}
enum WipToolOutput {
/// Only binary attachments accompany the validated native completion.
Native(ToolOutput),
/// Existing compatibility/error output retains its summary and pruning policy.
Passthrough(Result<ToolOutput, ToolError>),
}
pub enum WipOperationError {
Protocol(ProtocolError),
Cancelled(ToolOutput),
@@ -2255,15 +2278,22 @@ impl WipRuntime {
guard.disarm();
let completion = completion.map_err(client_tool_error)?;
self.push_audit(prepared.id, path, operation, audit_outcome);
if let Some(result) = passthrough {
if result.is_ok() {
self.metrics
.operation_successes
.fetch_add(1, Ordering::Relaxed);
let mut attachments = Vec::new();
if let Some(passthrough) = passthrough {
match passthrough {
WipToolOutput::Passthrough(result) => {
if result.is_ok() {
self.metrics
.operation_successes
.fetch_add(1, Ordering::Relaxed);
}
return result;
}
WipToolOutput::Native(output) => attachments = output.attachments,
}
return result;
}
render_call_completion(completion).map(|output| {
render_call_completion(completion).map(|mut output| {
output.attachments = attachments;
self.metrics
.operation_successes
.fetch_add(1, Ordering::Relaxed);
@@ -2366,14 +2396,7 @@ impl WipRuntime {
&self,
request: &Request<Vec<u8>>,
execution: ToolExecutionContext,
) -> Result<
(
Response<Vec<u8>>,
Option<Result<ToolOutput, ToolError>>,
WipAuditOutcome,
),
ToolError,
> {
) -> Result<(Response<Vec<u8>>, Option<WipToolOutput>, WipAuditOutcome), ToolError> {
let rejected = |error: ProtocolError| -> Result<_, ToolError> {
Ok((
encode_protocol_error_response(
@@ -2413,6 +2436,7 @@ impl WipRuntime {
return rejected(error);
}
let projection = projection.projection;
let kind = projection.kind;
let descriptor = projection.descriptor.clone();
let target_validator = projection.object.validator.clone();
let request_value =
@@ -2471,8 +2495,10 @@ impl WipRuntime {
.map_err(|error| ToolError::Internal(error.to_string()))?;
return Ok((
response,
Some(Err(ToolError::ExecutionFailed(format!(
"WIP operation outcome unknown; do not retry automatically: {message}"
Some(WipToolOutput::Passthrough(Err(ToolError::ExecutionFailed(
format!(
"WIP operation outcome unknown; do not retry automatically: {message}"
),
)))),
WipAuditOutcome::OutcomeUnknown,
));
@@ -2480,7 +2506,10 @@ impl WipRuntime {
};
Ok((
response,
output.tool_output.map(Ok),
output.tool_output.map(|output| match kind {
WipProjectionKind::Native => WipToolOutput::Native(output),
WipProjectionKind::Compatibility => WipToolOutput::Passthrough(Ok(output)),
}),
WipAuditOutcome::Success,
))
}
@@ -2505,7 +2534,9 @@ impl WipRuntime {
.map_err(|encode| ToolError::Internal(encode.to_string()))?;
Ok((
response,
Some(Err(ToolError::Cancelled(output))),
Some(WipToolOutput::Passthrough(Err(ToolError::Cancelled(
output,
)))),
WipAuditOutcome::Cancelled,
))
}
@@ -2521,7 +2552,9 @@ impl WipRuntime {
.map_err(|encode| ToolError::Internal(encode.to_string()))?;
Ok((
response,
Some(Err(ToolError::Interrupted(output))),
Some(WipToolOutput::Passthrough(Err(ToolError::Interrupted(
output,
)))),
WipAuditOutcome::Interrupted,
))
}
@@ -2534,8 +2567,10 @@ impl WipRuntime {
.map_err(|encode| ToolError::Internal(encode.to_string()))?;
Ok((
response,
Some(Err(ToolError::ExecutionFailed(format!(
"WIP operation outcome unknown; do not retry automatically: {message}"
Some(WipToolOutput::Passthrough(Err(ToolError::ExecutionFailed(
format!(
"WIP operation outcome unknown; do not retry automatically: {message}"
),
)))),
WipAuditOutcome::OutcomeUnknown,
))
@@ -5100,6 +5135,280 @@ mod tests {
}
}
struct NativeImage {
invalid_result: bool,
}
#[async_trait]
impl WipOperationHandler for NativeImage {
async fn call(
&self,
_operation: &str,
_arguments: &BTreeMap<String, Value>,
_context: WipCallContext,
) -> Result<WipOperationOutput, WipOperationError> {
let value = if self.invalid_result {
Value::Integer(1)
} else {
Value::String("Attached native image".into())
};
Ok(
WipOperationOutput::native_with_validator(value, vec![7]).with_attachments(vec![
Attachment::Image(agen::tool::ImageAttachment::new(
"image/png",
include_bytes!("../../tools/tests/fixtures/view-image/valid.png").to_vec(),
)),
]),
)
}
}
fn native_image_registry(invalid_result: bool) -> WipMountRegistry {
let mut registry = WipMountRegistry::new();
registry.allocate_namespace("native", "native").unwrap();
registry
.mount(WipProjection {
route: "/native/image".into(),
capability: "native:image".into(),
kind: WipProjectionKind::Native,
object: Object {
name: "image".into(),
description: None,
interfaces: vec![root_reference("yoi.native/image/v1")],
r#ref: Some("native:image".into()),
validator: Some(vec![1]),
},
interface: root_reference("yoi.native/image/v1"),
descriptor: InterfaceDescriptor {
format: INTERFACE_FORMAT_V1.into(),
documentation: None,
types: Vec::new(),
operations: vec![OperationDeclaration {
name: "view".into(),
documentation: None,
parameters: Vec::new(),
returns: ReturnDeclaration {
documentation: None,
r#type: TypeExpr::String,
},
}],
},
interface_validator: Some(vec![1]),
handler: Arc::new(NativeImage { invalid_result }),
})
.unwrap();
registry
}
fn native_image_invoke() -> String {
json!({
"path":"/native/image",
"interface":{"scope":"/","name":"yoi.native/image/v1"},
"operation":"view",
"arguments":{}
})
.to_string()
}
#[tokio::test]
async fn native_invoke_returns_image_bytes_and_preserves_result_and_validator() {
let mut engine = Engine::<_, Mutable, ()>::new_annotated(DummyClient);
let runtime = install_wip_mode_with_mounts(
&mut engine,
None,
"worker-a".into(),
native_image_registry(false),
)
.unwrap();
let tools = engine.tool_server_handle();
tools.flush_pending();
tools
.call_tool(
"Inspect",
r#"{"path":"/native/image"}"#,
ToolExecutionContext::direct(),
)
.await
.unwrap();
let output = tools
.call_tool(
"Invoke",
&native_image_invoke(),
ToolExecutionContext::direct(),
)
.await
.unwrap();
assert_eq!(output.summary, "WIP operation completed");
assert_eq!(
serde_json::from_str::<Json>(output.content.as_deref().unwrap()).unwrap(),
json!("Attached native image")
);
let [Attachment::Image(image)] = output.attachments.as_slice() else {
panic!("Invoke must return an image, not a URL")
};
assert_eq!(image.mime_type(), "image/png");
assert_eq!(
image.data(),
include_bytes!("../../tools/tests/fixtures/view-image/valid.png")
);
let state = runtime.state.lock().unwrap();
let calls = state.client.call_history(&state.session).unwrap();
assert!(
matches!(&calls.back().unwrap().outcome, CallOutcome::Success(response) if response.result == Value::String("Attached native image".into()) && response.validator == Some(vec![7]))
);
}
#[tokio::test]
async fn native_attachments_do_not_bypass_return_value_validation() {
let mut engine = Engine::<_, Mutable, ()>::new_annotated(DummyClient);
install_wip_mode_with_mounts(
&mut engine,
None,
"worker-a".into(),
native_image_registry(true),
)
.unwrap();
let tools = engine.tool_server_handle();
tools.flush_pending();
tools
.call_tool(
"Inspect",
r#"{"path":"/native/image"}"#,
ToolExecutionContext::direct(),
)
.await
.unwrap();
let error = tools
.call_tool(
"Invoke",
&native_image_invoke(),
ToolExecutionContext::direct(),
)
.await
.unwrap_err();
assert!(error.to_string().contains("outcome unknown"), "{error}");
}
#[derive(Clone)]
struct NativeImageClient {
events: Arc<Mutex<VecDeque<Vec<agen::llm_client::event::Event>>>>,
requests: Arc<Mutex<Vec<LlmRequest>>>,
}
#[async_trait]
impl LlmClient for NativeImageClient {
fn clone_boxed(&self) -> Box<dyn LlmClient> {
Box::new(self.clone())
}
async fn stream(&self, request: LlmRequest) -> Result<ResponseStream, ClientError> {
self.requests.lock().unwrap().push(request);
let events = self
.events
.lock()
.unwrap()
.pop_front()
.expect("unexpected model request");
Ok(Box::pin(stream::iter(events.into_iter().map(Ok))))
}
}
#[tokio::test]
async fn native_invoke_image_survives_engine_capture_and_durable_history_replay() {
use agen::llm_client::event::{Event, ResponseStatus, StatusEvent};
let completed = || {
Event::Status(StatusEvent {
status: ResponseStatus::Completed,
})
};
let client = NativeImageClient {
events: Arc::new(Mutex::new(VecDeque::from([
vec![
Event::tool_use_start(0, "native-image-call", "Invoke"),
Event::tool_input_delta(0, &native_image_invoke()),
Event::tool_use_stop(0),
completed(),
],
vec![completed()],
]))),
requests: Arc::new(Mutex::new(Vec::new())),
};
let mut engine = Engine::new(client.clone());
install_wip_mode_with_mounts(
&mut engine,
None,
"worker-a".into(),
native_image_registry(false),
)
.unwrap();
let tools = engine.tool_server_handle();
tools.flush_pending();
tools
.call_tool(
"Inspect",
r#"{"path":"/native/image"}"#,
ToolExecutionContext::direct(),
)
.await
.unwrap();
let captured = Arc::new(Mutex::new(Vec::new()));
let sink = Arc::clone(&captured);
engine.on_history_append(move |item| {
sink.lock().unwrap().push(item.clone());
Ok(())
});
let mut history = agen::History::new();
assert!(matches!(
engine
.run(&mut history, "view the native image")
.await
.result,
agen::EngineRunExit::Finished
));
let saved = history.items().find(|item| matches!(item, agen::Item::ToolResult { call_id, .. } if call_id == "native-image-call")).unwrap();
let agen::Item::ToolResult {
attachments,
is_error,
content,
..
} = saved
else {
unreachable!()
};
assert!(!is_error);
let [Attachment::Image(image)] = attachments.as_slice() else {
panic!("durable Invoke result lost image")
};
assert_eq!(
image.data(),
include_bytes!("../../tools/tests/fixtures/view-image/valid.png")
);
assert_eq!(
serde_json::from_str::<Json>(content.as_deref().unwrap()).unwrap(),
json!("Attached native image")
);
assert!(captured.lock().unwrap().iter().any(|item| item == saved));
assert!(
client.requests.lock().unwrap()[1]
.items
.iter()
.any(|item| item == saved)
);
let log = session_store::LogEntry::AnnotatedToolResult {
ts: 1,
entry: session_store::LoggedHistoryEntry {
item: session_store::LoggedItem::from(saved),
metadata: session_store::LoggedSessionHistoryMetadata {
entry_id: session_store::LoggedSessionHistoryEntryId::new(),
origin: session_store::LoggedSessionHistoryOrigin::LegacyUnknown,
derivation: None,
},
},
};
let serialized = serde_json::to_string(&log).unwrap();
let restored: session_store::LogEntry = serde_json::from_str(&serialized).unwrap();
assert_eq!(&session_store::collect_state(&[restored]).history[0], saved);
}
#[tokio::test]
async fn native_projection_preserves_descriptor_and_decodes_typed_arguments_end_to_end() {
let mut registry = WipMountRegistry::new();
+181
View File
@@ -639,6 +639,27 @@ impl WorkspaceRequest {
}
}
/// Raw-byte request through a host-bound Workspace client, separate from JSON requests.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WorkspaceBinaryRequest {
pub method: WorkspaceRequestMethod,
pub path: String,
pub body: Option<Vec<u8>>,
pub max_response_bytes: usize,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WorkspaceBinaryResponse {
pub status: u16,
pub body: Vec<u8>,
}
impl WorkspaceBinaryResponse {
pub fn is_success(&self) -> bool {
(200..300).contains(&self.status)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum WorkspaceRequestMethod {
Get,
@@ -859,6 +880,24 @@ pub trait WorkspaceClient: std::fmt::Debug + Send + Sync {
self.execute(request)
}
/// Binary transport is unavailable unless a host-bound implementation opts in.
fn execute_binary(
&self,
_request: WorkspaceBinaryRequest,
) -> Result<WorkspaceBinaryResponse, WorkspaceClientError> {
Err(WorkspaceClientError::Unavailable(
"Workspace binary transport is unavailable".to_string(),
))
}
fn execute_binary_with_timeout(
&self,
request: WorkspaceBinaryRequest,
_timeout: Duration,
) -> Result<WorkspaceBinaryResponse, WorkspaceClientError> {
self.execute_binary(request)
}
fn execute_server_operation(
&self,
operation: WorkspaceServerOperation,
@@ -961,6 +1000,31 @@ impl WorkspaceClient for ReviewerChildWorkspaceClient {
self.inner.current_prompt_projection(minimum_revision)
}
fn execute_binary(
&self,
request: WorkspaceBinaryRequest,
) -> Result<WorkspaceBinaryResponse, WorkspaceClientError> {
if request.method != WorkspaceRequestMethod::Get {
return Err(WorkspaceClientError::Unavailable(
"Reviewer child binary Workspace authority is read-only".to_string(),
));
}
self.inner.execute_binary(request)
}
fn execute_binary_with_timeout(
&self,
request: WorkspaceBinaryRequest,
timeout: Duration,
) -> Result<WorkspaceBinaryResponse, WorkspaceClientError> {
if request.method != WorkspaceRequestMethod::Get {
return Err(WorkspaceClientError::Unavailable(
"Reviewer child binary Workspace authority is read-only".to_string(),
));
}
self.inner.execute_binary_with_timeout(request, timeout)
}
fn execute(
&self,
mut request: WorkspaceRequest,
@@ -1157,6 +1221,123 @@ mod reviewer_client_tests {
)
}
#[test]
fn binary_transport_defaults_to_denial_without_json_fallback() {
let client = RecordingWorkspaceClient::default();
let request = WorkspaceBinaryRequest {
method: WorkspaceRequestMethod::Get,
path: "/api/w/ws/drive/entries/a/content".into(),
body: None,
max_response_bytes: 64 * 1024,
};
assert!(matches!(
client.execute_binary(request.clone()),
Err(WorkspaceClientError::Unavailable(_))
));
assert!(matches!(
client.execute_binary_with_timeout(request, Duration::from_secs(1)),
Err(WorkspaceClientError::Unavailable(_))
));
assert!(client.requests.lock().unwrap().is_empty());
}
#[derive(Debug, Default)]
struct BinaryRecordingClient {
requests: Mutex<Vec<(WorkspaceBinaryRequest, Option<Duration>)>>,
}
impl WorkspaceClient for BinaryRecordingClient {
fn workspace_id(&self) -> Option<&str> {
Some("ws")
}
fn kind(&self) -> &str {
"binary-recording"
}
fn is_available(&self) -> bool {
true
}
fn execute(&self, _: WorkspaceRequest) -> Result<WorkspaceResponse, WorkspaceClientError> {
panic!("binary request must not use JSON fallback")
}
fn execute_binary(
&self,
request: WorkspaceBinaryRequest,
) -> Result<WorkspaceBinaryResponse, WorkspaceClientError> {
self.requests.lock().unwrap().push((request, None));
Ok(WorkspaceBinaryResponse {
status: 200,
body: vec![0, 255],
})
}
fn execute_binary_with_timeout(
&self,
request: WorkspaceBinaryRequest,
timeout: Duration,
) -> Result<WorkspaceBinaryResponse, WorkspaceClientError> {
self.requests.lock().unwrap().push((request, Some(timeout)));
Ok(WorkspaceBinaryResponse {
status: 200,
body: vec![0, 255],
})
}
}
#[test]
fn reviewer_binary_get_preserves_request_limit_and_timeout() {
let inner = Arc::new(BinaryRecordingClient::default());
let client = reviewer_client(inner.clone());
let request = WorkspaceBinaryRequest {
method: WorkspaceRequestMethod::Get,
path: "/api/w/ws/drive/entries/a/content?offset=2&limit=4".into(),
body: None,
max_response_bytes: 4,
};
assert_eq!(
client.execute_binary(request.clone()).unwrap().body,
vec![0, 255]
);
let timeout = Duration::from_secs(2);
assert_eq!(
client
.execute_binary_with_timeout(request.clone(), timeout)
.unwrap()
.body,
vec![0, 255]
);
assert_eq!(
*inner.requests.lock().unwrap(),
vec![(request.clone(), None), (request, Some(timeout))]
);
}
#[test]
fn reviewer_binary_mutations_are_denied_even_for_attested_review_path() {
let inner = Arc::new(BinaryRecordingClient::default());
let client = reviewer_client(inner.clone());
for method in [
WorkspaceRequestMethod::Post,
WorkspaceRequestMethod::Put,
WorkspaceRequestMethod::Patch,
WorkspaceRequestMethod::Delete,
] {
let request = WorkspaceBinaryRequest {
method,
path: "/api/w/ws/merge-requests/MR1/reviews".into(),
body: Some(vec![0, 255]),
max_response_bytes: 4,
};
assert!(matches!(
client.execute_binary(request.clone()),
Err(WorkspaceClientError::Unavailable(_))
));
assert!(matches!(
client.execute_binary_with_timeout(request, Duration::from_secs(1)),
Err(WorkspaceClientError::Unavailable(_))
));
}
assert!(inner.requests.lock().unwrap().is_empty());
}
#[test]
fn reviewer_child_client_allows_typed_ticket_reads() {
let inner = Arc::new(RecordingWorkspaceClient::default());