feat: project attached checkouts with scoped native AI operations
This commit is contained in:
@@ -55,6 +55,8 @@ wip-http = { workspace = true }
|
||||
wip-protocol = { workspace = true }
|
||||
|
||||
[dev-dependencies]
|
||||
axum = { workspace = true }
|
||||
workdir = { workspace = true, features = ["http-client"] }
|
||||
dotenv = "0.15.0"
|
||||
futures = { workspace = true }
|
||||
serial_test = "3.4.0"
|
||||
|
||||
@@ -0,0 +1,764 @@
|
||||
//! Provider-owned request-time checkout projection. No filesystem path is opened
|
||||
//! by the Worker: every observation and operation goes through the live router.
|
||||
|
||||
use std::collections::BTreeMap;
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use agen::tool::ToolError;
|
||||
use async_trait::async_trait;
|
||||
use manifest::{ToolPermissionAction, ToolPermissionConfig};
|
||||
use serde_json::{Value as Json, json};
|
||||
use sha2::{Digest, Sha256};
|
||||
use wip_protocol::{
|
||||
Documentation, INTERFACE_FORMAT_V1, InterfaceDescriptor, Object, OperationDeclaration,
|
||||
ParameterDeclaration, ProtocolError, ProtocolErrorCode, ReturnDeclaration, TypeExpr, Value,
|
||||
};
|
||||
use workdir::{
|
||||
CheckoutObservation, EntryKind, ListRequest, WorkdirPath, WorkdirSessionCapability as Cap,
|
||||
WorkdirSessionRouter,
|
||||
};
|
||||
|
||||
use crate::feature::builtin::manage_workdir::wip::{decode_identity, encode_identity};
|
||||
use crate::permission::permission_action_for;
|
||||
use crate::wip::{
|
||||
WipCallContext, WipMountError, WipMountRegistry, WipOperationError, WipOperationHandler,
|
||||
WipOperationOutput, WipProjection, WipProjectionKind, WipSubtreeMount, WipSubtreeProvider,
|
||||
json_to_wip, wip_to_json,
|
||||
};
|
||||
|
||||
const ROOT: &str = "/checkouts";
|
||||
const MAX_NODES: usize = 1024;
|
||||
const DEADLINE: Duration = Duration::from_secs(30);
|
||||
|
||||
/// Return a stable, injective attachment-local entrance, never a host path.
|
||||
pub fn checkout_root(alias: &str) -> String {
|
||||
format!("{ROOT}/{}", encode_identity(alias))
|
||||
}
|
||||
|
||||
/// Register the native content namespace independently of inventory/lifecycle
|
||||
/// Features. Its capabilities come from current attachments, not Feature flags.
|
||||
pub fn mount_checkouts(
|
||||
registry: &mut WipMountRegistry,
|
||||
router: Arc<WorkdirSessionRouter>,
|
||||
tracker: tools::Tracker,
|
||||
permissions: Option<ToolPermissionConfig>,
|
||||
) -> Result<(), WipMountError> {
|
||||
registry.allocate_namespace("checkout", "checkouts")?;
|
||||
let provider = Arc::new(Provider {
|
||||
router,
|
||||
tracker,
|
||||
permissions,
|
||||
incarnation: uuid::Uuid::now_v7().to_string(),
|
||||
permits: Arc::new(tokio::sync::Semaphore::new(16)),
|
||||
deadline: DEADLINE,
|
||||
});
|
||||
registry.mount(provider.collection())?;
|
||||
registry.replace_compatibility_tools(ROOT, ["Read", "Edit", "Write", "Glob", "Grep"])?;
|
||||
registry.mount_subtree(WipSubtreeMount {
|
||||
root: ROOT.into(),
|
||||
provider: Arc::new(provider),
|
||||
})
|
||||
}
|
||||
|
||||
struct Provider {
|
||||
router: Arc<WorkdirSessionRouter>,
|
||||
tracker: tools::Tracker,
|
||||
permissions: Option<ToolPermissionConfig>,
|
||||
incarnation: String,
|
||||
permits: Arc<tokio::sync::Semaphore>,
|
||||
deadline: Duration,
|
||||
}
|
||||
|
||||
impl Provider {
|
||||
fn allowed(&self, tool: &str, input: &Json) -> bool {
|
||||
self.permissions
|
||||
.as_ref()
|
||||
.is_none_or(|p| permission_action_for(p, tool, input) == ToolPermissionAction::Allow)
|
||||
}
|
||||
fn may_allow(&self, tool: &str) -> bool {
|
||||
let Some(p) = &self.permissions else {
|
||||
return true;
|
||||
};
|
||||
for rule in p.rules.iter().filter(|r| r.tool.eq_ignore_ascii_case(tool)) {
|
||||
if rule.action == ToolPermissionAction::Allow {
|
||||
return true;
|
||||
}
|
||||
if rule.pattern == "*" {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
p.default_action == ToolPermissionAction::Allow
|
||||
}
|
||||
fn parse(&self, path: &str) -> Option<(String, WorkdirPath)> {
|
||||
wip_protocol::validate_path(path).ok()?;
|
||||
let rest = path.strip_prefix("/checkouts/")?;
|
||||
let (slug, tail) = rest.split_once('/').unwrap_or((rest, ""));
|
||||
if tail.contains('\0') {
|
||||
return None;
|
||||
}
|
||||
Some((decode_identity(slug)?, WorkdirPath::new(tail).ok()?))
|
||||
}
|
||||
fn wrap_validator(&self, alias: &str, generation: u64, workdir: &str, raw: &[u8]) -> Vec<u8> {
|
||||
let mut hash = Sha256::new();
|
||||
for value in [
|
||||
self.incarnation.as_bytes(),
|
||||
alias.as_bytes(),
|
||||
&generation.to_be_bytes(),
|
||||
workdir.as_bytes(),
|
||||
raw,
|
||||
] {
|
||||
hash.update(value.len().to_be_bytes());
|
||||
hash.update(value);
|
||||
}
|
||||
hash.finalize().to_vec()
|
||||
}
|
||||
fn collection(self: &Arc<Self>) -> WipProjection {
|
||||
let mut state = Sha256::new();
|
||||
state.update(&self.incarnation);
|
||||
for alias in self.router.aliases() {
|
||||
if let Ok(s) = self.router.resolve(Some(alias.as_str())) {
|
||||
state.update(alias.as_str());
|
||||
state.update(s.generation.to_be_bytes());
|
||||
state.update(s.session.workdir().id().as_str());
|
||||
state.update(format!("{:?}", s.session.capabilities()));
|
||||
}
|
||||
}
|
||||
self.project(
|
||||
ROOT,
|
||||
descriptor(vec![operation(
|
||||
"list",
|
||||
"List currently accessible checkouts",
|
||||
vec![],
|
||||
)]),
|
||||
state.finalize().to_vec(),
|
||||
"collection",
|
||||
Arc::new(Collection {
|
||||
provider: self.clone(),
|
||||
}),
|
||||
)
|
||||
}
|
||||
fn project(
|
||||
&self,
|
||||
path: &str,
|
||||
descriptor: InterfaceDescriptor,
|
||||
validator: Vec<u8>,
|
||||
kind: &str,
|
||||
handler: Arc<dyn WipOperationHandler>,
|
||||
) -> WipProjection {
|
||||
let encoded: String = path.as_bytes().iter().map(|b| format!("{b:02x}")).collect();
|
||||
let interface = format!("yoi.checkout/{}/{kind}/v1/@/{encoded}", self.incarnation);
|
||||
let interface_validator = Sha256::digest(format!("{descriptor:?}")).to_vec();
|
||||
WipProjection {
|
||||
route: path.into(),
|
||||
capability: format!("checkout:{kind}"),
|
||||
kind: WipProjectionKind::Native,
|
||||
object: Object {
|
||||
name: path.rsplit('/').next().unwrap_or_default().into(),
|
||||
description: Some(format!("Attached Workdir {kind}")),
|
||||
interfaces: vec![interface.clone()],
|
||||
r#ref: None,
|
||||
validator: Some(validator),
|
||||
},
|
||||
interface,
|
||||
descriptor,
|
||||
interface_validator: Some(interface_validator),
|
||||
handler,
|
||||
}
|
||||
}
|
||||
async fn node(self: &Arc<Self>, path: &str) -> Result<Option<WipProjection>, ProtocolError> {
|
||||
if path == ROOT {
|
||||
return Ok(Some(self.collection()));
|
||||
}
|
||||
let Some((alias, target)) = self.parse(path) else {
|
||||
return Ok(None);
|
||||
};
|
||||
let Ok(selected) = self.router.resolve(Some(&alias)) else {
|
||||
return Ok(None);
|
||||
};
|
||||
let observation = match selected.session.checkout_observe(target.clone()).await {
|
||||
Ok(o) => o,
|
||||
Err(
|
||||
workdir::WorkdirError::Denied(_)
|
||||
| workdir::WorkdirError::OutOfScope(_)
|
||||
| workdir::WorkdirError::NotFound(_)
|
||||
| workdir::WorkdirError::SessionClosed
|
||||
| workdir::WorkdirError::UnsupportedOperation(_)
|
||||
| workdir::WorkdirError::Unsupported(_),
|
||||
) => return Ok(None),
|
||||
Err(_) => {
|
||||
return Err(error(
|
||||
ProtocolErrorCode::Internal,
|
||||
"checkout observation unavailable",
|
||||
));
|
||||
}
|
||||
};
|
||||
if !matches!(observation.kind, EntryKind::File | EntryKind::Directory) {
|
||||
return Ok(None);
|
||||
}
|
||||
let permission_input = json!({"target_workdir":alias, "file_path":target.as_str()});
|
||||
if observation.kind == EntryKind::File && !self.allowed("Read", &permission_input) {
|
||||
return Ok(None);
|
||||
}
|
||||
let ops = self.operations(&observation);
|
||||
if ops.is_empty() {
|
||||
return Ok(None);
|
||||
}
|
||||
let kind = if observation.kind == EntryKind::File {
|
||||
"file"
|
||||
} else {
|
||||
"directory"
|
||||
};
|
||||
let workdir = selected.session.workdir().id().as_str().to_owned();
|
||||
let validator = self.wrap_validator(
|
||||
&alias,
|
||||
selected.generation,
|
||||
&workdir,
|
||||
&observation.validator,
|
||||
);
|
||||
let handler = Arc::new(FileHandler {
|
||||
provider: self.clone(),
|
||||
alias,
|
||||
generation: selected.generation,
|
||||
workdir,
|
||||
target,
|
||||
observation,
|
||||
});
|
||||
let family = format!("{kind}-g{}", selected.generation);
|
||||
Ok(Some(self.project(
|
||||
path,
|
||||
descriptor(ops),
|
||||
validator,
|
||||
&family,
|
||||
handler,
|
||||
)))
|
||||
}
|
||||
fn operations(&self, o: &CheckoutObservation) -> Vec<OperationDeclaration> {
|
||||
let mut ops = Vec::new();
|
||||
let has = |cap, name| o.capabilities.supports(cap) && self.may_allow(name);
|
||||
if o.kind == EntryKind::File {
|
||||
if has(Cap::Read, "Read") {
|
||||
ops.push(operation(
|
||||
"read",
|
||||
"Read text with line offsets and numbered output",
|
||||
vec![
|
||||
parameter("offset", false, TypeExpr::Integer),
|
||||
parameter("limit", false, TypeExpr::Integer),
|
||||
],
|
||||
));
|
||||
}
|
||||
if has(Cap::Edit, "Edit") {
|
||||
ops.push(operation(
|
||||
"edit",
|
||||
"Replace a unique string (requires prior Read)",
|
||||
vec![
|
||||
parameter("old_string", true, TypeExpr::String),
|
||||
parameter("new_string", true, TypeExpr::String),
|
||||
parameter("replace_all", false, TypeExpr::Boolean),
|
||||
],
|
||||
));
|
||||
}
|
||||
if has(Cap::Write, "Write") {
|
||||
ops.push(operation(
|
||||
"write",
|
||||
"Overwrite this existing file (requires prior Read)",
|
||||
vec![parameter("content", true, TypeExpr::String)],
|
||||
));
|
||||
}
|
||||
} else {
|
||||
if has(Cap::Glob, "Glob") {
|
||||
ops.push(operation(
|
||||
"glob",
|
||||
"Glob within this directory",
|
||||
vec![
|
||||
parameter("pattern", true, TypeExpr::String),
|
||||
parameter("path", false, TypeExpr::String),
|
||||
],
|
||||
));
|
||||
}
|
||||
if has(Cap::Grep, "Grep") {
|
||||
ops.push(operation(
|
||||
"grep",
|
||||
"Search within this directory with the existing Grep contract",
|
||||
grep_parameters(),
|
||||
));
|
||||
}
|
||||
if has(Cap::Write, "Write") {
|
||||
ops.push(operation("create_file", "Create a new relative file, creating permitted missing parents; never overwrite", vec![parameter("path",true,TypeExpr::String),parameter("content",true,TypeExpr::String)]));
|
||||
}
|
||||
}
|
||||
ops
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl WipSubtreeProvider for Arc<Provider> {
|
||||
async fn projection(&self, path: &str) -> Result<Option<WipProjection>, ProtocolError> {
|
||||
let _permit = tokio::time::timeout(self.deadline, self.permits.acquire())
|
||||
.await
|
||||
.map_err(|_| {
|
||||
error(
|
||||
ProtocolErrorCode::ResourceLimitExceeded,
|
||||
"checkout concurrency deadline",
|
||||
)
|
||||
})?
|
||||
.map_err(|_| error(ProtocolErrorCode::Internal, "checkout unavailable"))?;
|
||||
tokio::time::timeout(self.deadline, self.node(path))
|
||||
.await
|
||||
.map_err(|_| {
|
||||
error(
|
||||
ProtocolErrorCode::ResourceLimitExceeded,
|
||||
"checkout observation deadline",
|
||||
)
|
||||
})?
|
||||
}
|
||||
async fn children(&self, path: &str) -> Result<Vec<String>, ProtocolError> {
|
||||
let _permit = tokio::time::timeout(self.deadline, self.permits.acquire())
|
||||
.await
|
||||
.map_err(|_| {
|
||||
error(
|
||||
ProtocolErrorCode::ResourceLimitExceeded,
|
||||
"checkout capacity deadline",
|
||||
)
|
||||
})?
|
||||
.map_err(|_| error(ProtocolErrorCode::Internal, "checkout unavailable"))?;
|
||||
tokio::time::timeout(self.deadline, self.children_inner(path))
|
||||
.await
|
||||
.map_err(|_| {
|
||||
error(
|
||||
ProtocolErrorCode::ResourceLimitExceeded,
|
||||
"checkout enumeration deadline",
|
||||
)
|
||||
})?
|
||||
}
|
||||
}
|
||||
|
||||
impl Provider {
|
||||
async fn children_inner(self: &Arc<Self>, path: &str) -> Result<Vec<String>, ProtocolError> {
|
||||
if path == ROOT {
|
||||
let before = self.collection().object.validator.unwrap();
|
||||
if self.router.aliases().len() > MAX_NODES {
|
||||
return Err(error(
|
||||
ProtocolErrorCode::ResourceLimitExceeded,
|
||||
"too many checkouts",
|
||||
));
|
||||
}
|
||||
let mut paths = Vec::new();
|
||||
for alias in self.router.aliases() {
|
||||
let p = checkout_root(alias.as_str());
|
||||
if self.node(&p).await?.is_some() {
|
||||
paths.push(p);
|
||||
}
|
||||
}
|
||||
if before != self.collection().object.validator.unwrap() {
|
||||
return Err(error(
|
||||
ProtocolErrorCode::ValidatorMismatch,
|
||||
"attachment collection changed",
|
||||
));
|
||||
}
|
||||
return Ok(paths);
|
||||
}
|
||||
let Some((alias, target)) = self.parse(path) else {
|
||||
return Err(error(ProtocolErrorCode::NotFound, "checkout not published"));
|
||||
};
|
||||
let selected = self
|
||||
.router
|
||||
.resolve(Some(&alias))
|
||||
.map_err(|_| error(ProtocolErrorCode::NotFound, "attachment expired"))?;
|
||||
let o = selected
|
||||
.session
|
||||
.checkout_observe(target.clone())
|
||||
.await
|
||||
.map_err(|_| error(ProtocolErrorCode::NotFound, "checkout not published"))?;
|
||||
if o.kind != EntryKind::Directory {
|
||||
return Ok(Vec::new());
|
||||
}
|
||||
if !self.allowed(
|
||||
"Glob",
|
||||
&json!({"target_workdir":alias,"path":target.as_str(),"pattern":"*"}),
|
||||
) && !self.allowed(
|
||||
"Grep",
|
||||
&json!({"target_workdir":alias,"path":target.as_str(),"pattern":""}),
|
||||
) {
|
||||
return Ok(Vec::new());
|
||||
}
|
||||
let listing = selected
|
||||
.session
|
||||
.checkout_search(workdir::CheckoutSearchRequest::new(
|
||||
workdir::CheckoutSearchOperation::List(ListRequest {
|
||||
path: target,
|
||||
limit: MAX_NODES + 1,
|
||||
}),
|
||||
))
|
||||
.await
|
||||
.map_err(|_| error(ProtocolErrorCode::Internal, "checkout listing unavailable"))?;
|
||||
let workdir::CheckoutSearchResult::List(listing) = listing else {
|
||||
return Err(error(
|
||||
ProtocolErrorCode::Internal,
|
||||
"mismatched checkout listing",
|
||||
));
|
||||
};
|
||||
if listing.truncated || listing.entries.len() > MAX_NODES {
|
||||
return Err(error(
|
||||
ProtocolErrorCode::ResourceLimitExceeded,
|
||||
"directory observation exceeds node bound",
|
||||
));
|
||||
}
|
||||
let mut paths = Vec::new();
|
||||
for entry in listing.entries {
|
||||
let p = object_path(&alias, &entry.path);
|
||||
if self.node(&p).await?.is_some() {
|
||||
paths.push(p);
|
||||
}
|
||||
}
|
||||
if !self
|
||||
.router
|
||||
.resolve(Some(&alias))
|
||||
.is_ok_and(|current| current.generation == selected.generation)
|
||||
{
|
||||
return Err(error(
|
||||
ProtocolErrorCode::ValidatorMismatch,
|
||||
"attachment changed",
|
||||
));
|
||||
}
|
||||
Ok(paths)
|
||||
}
|
||||
}
|
||||
|
||||
struct Collection {
|
||||
provider: Arc<Provider>,
|
||||
}
|
||||
#[async_trait]
|
||||
impl WipOperationHandler for Collection {
|
||||
async fn call(
|
||||
&self,
|
||||
operation: &str,
|
||||
arguments: &BTreeMap<String, Value>,
|
||||
_context: WipCallContext,
|
||||
) -> Result<WipOperationOutput, WipOperationError> {
|
||||
if operation != "list" || !arguments.is_empty() {
|
||||
return Err(protocol(
|
||||
ProtocolErrorCode::InvalidArguments,
|
||||
"list accepts no arguments",
|
||||
));
|
||||
}
|
||||
let before = self.provider.collection().object.validator.unwrap();
|
||||
let paths = Arc::new(self.provider.clone())
|
||||
.children(ROOT)
|
||||
.await
|
||||
.map_err(WipOperationError::Protocol)?;
|
||||
let mut items = Vec::new();
|
||||
for path in paths {
|
||||
let (alias, _) = self.provider.parse(&path).expect("generated checkout path");
|
||||
let s = self.provider.router.resolve(Some(&alias)).map_err(|_| {
|
||||
protocol(ProtocolErrorCode::ValidatorMismatch, "attachment changed")
|
||||
})?;
|
||||
items.push(json!({"path":path,"slug":encode_identity(&alias),"alias":alias,
|
||||
"working_directory_id":s.session.workdir().id().as_str(),"generation":s.generation}));
|
||||
}
|
||||
let value = json_to_wip(&json!({"items":items}))
|
||||
.map_err(|_| protocol(ProtocolErrorCode::Internal, "invalid checkout result"))?;
|
||||
let after = self.provider.collection().object.validator.unwrap();
|
||||
if before != after {
|
||||
return Err(protocol(
|
||||
ProtocolErrorCode::ValidatorMismatch,
|
||||
"attachment collection changed",
|
||||
));
|
||||
}
|
||||
Ok(WipOperationOutput::native_with_validator(value, after))
|
||||
}
|
||||
}
|
||||
|
||||
struct FileHandler {
|
||||
provider: Arc<Provider>,
|
||||
alias: String,
|
||||
generation: u64,
|
||||
workdir: String,
|
||||
target: WorkdirPath,
|
||||
observation: CheckoutObservation,
|
||||
}
|
||||
impl FileHandler {
|
||||
fn check_connection(&self) -> Result<(), WipOperationError> {
|
||||
if self
|
||||
.provider
|
||||
.router
|
||||
.resolve(Some(&self.alias))
|
||||
.is_ok_and(|s| {
|
||||
s.generation == self.generation && s.session.workdir().id().as_str() == self.workdir
|
||||
})
|
||||
{
|
||||
Ok(())
|
||||
} else {
|
||||
Err(protocol(
|
||||
ProtocolErrorCode::ValidatorMismatch,
|
||||
"attachment changed while publishing result",
|
||||
))
|
||||
}
|
||||
}
|
||||
}
|
||||
#[async_trait]
|
||||
impl WipOperationHandler for FileHandler {
|
||||
async fn call(
|
||||
&self,
|
||||
operation: &str,
|
||||
arguments: &BTreeMap<String, Value>,
|
||||
context: WipCallContext,
|
||||
) -> Result<WipOperationOutput, WipOperationError> {
|
||||
let tool = match operation {
|
||||
"read" => "Read",
|
||||
"edit" => "Edit",
|
||||
"write" => "Write",
|
||||
"create_file" => "Create",
|
||||
"glob" => "Glob",
|
||||
"grep" => "Grep",
|
||||
_ => {
|
||||
return Err(protocol(
|
||||
ProtocolErrorCode::OperationNotFound,
|
||||
"unknown checkout operation",
|
||||
));
|
||||
}
|
||||
};
|
||||
let mut args = serde_json::Map::new();
|
||||
for (k, v) in arguments {
|
||||
args.insert(
|
||||
k.clone(),
|
||||
wip_to_json(v).map_err(|_| {
|
||||
protocol(
|
||||
ProtocolErrorCode::InvalidArguments,
|
||||
"invalid checkout argument",
|
||||
)
|
||||
})?,
|
||||
);
|
||||
}
|
||||
let mut permission_input = args.clone();
|
||||
permission_input.insert("target_workdir".into(), json!(self.alias));
|
||||
if tool == "Glob" || tool == "Grep" {
|
||||
let relative = args.get("path").and_then(Json::as_str).unwrap_or("");
|
||||
permission_input.insert("path".into(), json!(join_under(&self.target, relative)?));
|
||||
} else if tool == "Create" {
|
||||
let relative = args.get("path").and_then(Json::as_str).ok_or_else(|| {
|
||||
protocol(ProtocolErrorCode::InvalidArguments, "create requires path")
|
||||
})?;
|
||||
permission_input.remove("path");
|
||||
permission_input.insert(
|
||||
"file_path".into(),
|
||||
json!(join_under(&self.target, relative)?),
|
||||
);
|
||||
} else {
|
||||
permission_input.insert("file_path".into(), json!(self.target.as_str()));
|
||||
}
|
||||
if !self.provider.allowed(
|
||||
if tool == "Create" { "Write" } else { tool },
|
||||
&Json::Object(permission_input),
|
||||
) {
|
||||
return Err(protocol(
|
||||
ProtocolErrorCode::PermissionDenied,
|
||||
"file operation denied; approval-required policy fails closed",
|
||||
));
|
||||
}
|
||||
let _permit = tokio::time::timeout(self.provider.deadline, self.provider.permits.acquire())
|
||||
.await
|
||||
.map_err(|_| {
|
||||
protocol(
|
||||
ProtocolErrorCode::ResourceLimitExceeded,
|
||||
"checkout operation capacity deadline",
|
||||
)
|
||||
})?
|
||||
.map_err(|_| protocol(ProtocolErrorCode::Internal, "checkout unavailable"))?;
|
||||
let deadline = tokio::time::Instant::now() + self.provider.deadline;
|
||||
let result = tokio::time::timeout_at(
|
||||
deadline,
|
||||
tools::execute_checkout_tool(
|
||||
self.provider.router.clone(),
|
||||
self.provider.tracker.clone(),
|
||||
&self.alias,
|
||||
self.generation,
|
||||
self.target.clone(),
|
||||
self.observation.validator.clone(),
|
||||
tool,
|
||||
Json::Object(args),
|
||||
context.execution,
|
||||
),
|
||||
)
|
||||
.await
|
||||
.map_err(|_| {
|
||||
if matches!(tool, "Write" | "Edit" | "Create") {
|
||||
WipOperationError::OutcomeUnknown("file operation deadline after dispatch".into())
|
||||
} else {
|
||||
protocol(
|
||||
ProtocolErrorCode::ResourceLimitExceeded,
|
||||
"file operation deadline",
|
||||
)
|
||||
}
|
||||
})?
|
||||
.map_err(map_tool_error)?;
|
||||
let links = tokio::time::timeout_at(deadline, async {
|
||||
let mut links = Vec::new();
|
||||
self.check_connection()?;
|
||||
for path in &result.paths {
|
||||
let route = object_path(&self.alias, path);
|
||||
// References are validated only against the captured live connection.
|
||||
if self
|
||||
.provider
|
||||
.node(&route)
|
||||
.await
|
||||
.map_err(WipOperationError::Protocol)?
|
||||
.is_some()
|
||||
{
|
||||
links.push(json!({"path":route}));
|
||||
}
|
||||
self.check_connection()?;
|
||||
}
|
||||
Ok::<_, WipOperationError>(links)
|
||||
})
|
||||
.await
|
||||
.map_err(|_| {
|
||||
if matches!(tool, "Write" | "Edit" | "Create") {
|
||||
WipOperationError::OutcomeUnknown("completed operation response deadline".into())
|
||||
} else {
|
||||
protocol(
|
||||
ProtocolErrorCode::ResourceLimitExceeded,
|
||||
"checkout result deadline",
|
||||
)
|
||||
}
|
||||
})?
|
||||
.map_err(|e| {
|
||||
if matches!(tool, "Write" | "Edit" | "Create") {
|
||||
WipOperationError::OutcomeUnknown("completed operation response unavailable".into())
|
||||
} else {
|
||||
e
|
||||
}
|
||||
})?;
|
||||
let value = json_to_wip(
|
||||
&json!({"summary":result.output.summary,"content":result.output.content,"items":links}),
|
||||
)
|
||||
.map_err(|_| {
|
||||
WipOperationError::OutcomeUnknown("invalid completed file operation result".into())
|
||||
})?;
|
||||
if let Some(raw) = result.validator {
|
||||
Ok(WipOperationOutput::native_with_validator(
|
||||
value,
|
||||
self.provider
|
||||
.wrap_validator(&self.alias, self.generation, &self.workdir, &raw),
|
||||
))
|
||||
} else {
|
||||
Ok(WipOperationOutput::native(value))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn object_path(alias: &str, path: &WorkdirPath) -> String {
|
||||
if path.is_root() {
|
||||
checkout_root(alias)
|
||||
} else {
|
||||
format!("{}/{}", checkout_root(alias), path.as_str())
|
||||
}
|
||||
}
|
||||
fn join_under(parent: &WorkdirPath, relative: &str) -> Result<String, WipOperationError> {
|
||||
let child = WorkdirPath::new(relative).map_err(|_| {
|
||||
protocol(
|
||||
ProtocolErrorCode::InvalidArguments,
|
||||
"path must stay relative to this directory",
|
||||
)
|
||||
})?;
|
||||
if relative.starts_with('/') {
|
||||
return Err(protocol(
|
||||
ProtocolErrorCode::InvalidArguments,
|
||||
"absolute paths are not permitted",
|
||||
));
|
||||
}
|
||||
Ok(if parent.is_root() {
|
||||
child.as_str().into()
|
||||
} else if child.is_root() {
|
||||
parent.as_str().into()
|
||||
} else {
|
||||
format!("{}/{}", parent.as_str(), child.as_str())
|
||||
})
|
||||
}
|
||||
fn parameter(name: &str, required: bool, r#type: TypeExpr) -> ParameterDeclaration {
|
||||
ParameterDeclaration {
|
||||
name: name.into(),
|
||||
required,
|
||||
documentation: None,
|
||||
r#type,
|
||||
}
|
||||
}
|
||||
fn operation(
|
||||
name: &str,
|
||||
summary: &str,
|
||||
parameters: Vec<ParameterDeclaration>,
|
||||
) -> OperationDeclaration {
|
||||
OperationDeclaration {
|
||||
name: name.into(),
|
||||
documentation: Some(Documentation {
|
||||
summary: summary.into(),
|
||||
details: None,
|
||||
}),
|
||||
parameters,
|
||||
returns: ReturnDeclaration {
|
||||
documentation: None,
|
||||
r#type: TypeExpr::Json,
|
||||
},
|
||||
}
|
||||
}
|
||||
fn descriptor(operations: Vec<OperationDeclaration>) -> InterfaceDescriptor {
|
||||
InterfaceDescriptor{format:INTERFACE_FORMAT_V1.into(),documentation:Some(Documentation{summary:"AI-oriented attached Workdir operations".into(),details:Some("The route binds attachment and target. Line offsets are zero-based; text output numbers lines from one. Read-before-write and provider validators are both required. Paths are Worldspace references, never host paths.".into())}),types:Vec::new(),operations}
|
||||
}
|
||||
fn grep_parameters() -> Vec<ParameterDeclaration> {
|
||||
let mut p = vec![parameter("pattern", true, TypeExpr::String)];
|
||||
for name in ["path", "glob", "type", "output_mode"] {
|
||||
p.push(parameter(name, false, TypeExpr::String));
|
||||
}
|
||||
for name in ["-A", "-B", "-C", "head_limit", "offset"] {
|
||||
p.push(parameter(name, false, TypeExpr::Integer));
|
||||
}
|
||||
for name in ["case_insensitive", "multiline"] {
|
||||
p.push(parameter(name, false, TypeExpr::Boolean));
|
||||
}
|
||||
p
|
||||
}
|
||||
fn error(code: ProtocolErrorCode, message: &str) -> ProtocolError {
|
||||
ProtocolError {
|
||||
code,
|
||||
message: message.into(),
|
||||
}
|
||||
}
|
||||
fn protocol(code: ProtocolErrorCode, message: &str) -> WipOperationError {
|
||||
WipOperationError::Protocol(error(code, message))
|
||||
}
|
||||
fn map_tool_error(e: ToolError) -> WipOperationError {
|
||||
match e {
|
||||
ToolError::InvalidArgument(_) => protocol(
|
||||
ProtocolErrorCode::InvalidArguments,
|
||||
"invalid file arguments or read-before-write requirement",
|
||||
),
|
||||
ToolError::StructuredConflict { code, .. } => match code.as_str() {
|
||||
"checkout_denied" => protocol(
|
||||
ProtocolErrorCode::PermissionDenied,
|
||||
"file operation refused before effects",
|
||||
),
|
||||
"checkout_invalid" => protocol(
|
||||
ProtocolErrorCode::InvalidArguments,
|
||||
"invalid checkout operation or target",
|
||||
),
|
||||
"checkout_unavailable" => protocol(
|
||||
ProtocolErrorCode::ResourceLimitExceeded,
|
||||
"checkout provider declined before effects",
|
||||
),
|
||||
"checkout_stale" => protocol(
|
||||
ProtocolErrorCode::ValidatorMismatch,
|
||||
"file or attachment observation is stale; rediscover and reread",
|
||||
),
|
||||
_ => WipOperationError::OutcomeUnknown("unclassified file operation failure".into()),
|
||||
},
|
||||
ToolError::Cancelled(o) => WipOperationError::Cancelled(o),
|
||||
ToolError::Interrupted(o) => WipOperationError::Interrupted(o),
|
||||
_ => WipOperationError::OutcomeUnknown(
|
||||
"file operation failed; inspect current state before retry".into(),
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "checkout_race_tests.rs"]
|
||||
mod race_tests;
|
||||
@@ -0,0 +1,671 @@
|
||||
//! Real provider HTTP -> RemoteWorkdirSession -> router -> native WIP Client.
|
||||
//! This deliberately does not exercise the production Backend grant network.
|
||||
use super::*;
|
||||
use axum::{
|
||||
Json as AxumJson, Router,
|
||||
body::Body,
|
||||
extract::{Path, State},
|
||||
http::{HeaderMap, StatusCode, Uri},
|
||||
response::{IntoResponse, Response as AxumResponse},
|
||||
routing::{delete, post},
|
||||
};
|
||||
use std::time::Duration;
|
||||
use tempfile::TempDir;
|
||||
use workdir::http::{
|
||||
OpenWorkdirSessionRequest, OpenWorkdirSessionResponse, RemoteWorkdirSession, WorkdirSessionId,
|
||||
WorkdirSessionOperation, WorkdirSessionOperationRequest, WorkdirTransportError,
|
||||
WorkdirTransportErrorCode,
|
||||
};
|
||||
use workdir::{
|
||||
CheckoutOperation, CheckoutRequest, CheckoutSearchOperation, LocalWorkdirSession,
|
||||
WorkdirAttachmentAlias, WorkdirError, WorkdirPath, WorkdirSession, WorkdirSessionCapabilities,
|
||||
WorkdirSessionHandle, WorkdirSessionRouter, dispatch_workdir_session_operation,
|
||||
};
|
||||
|
||||
const TOKEN: &str = "checkout-http-test-private-bearer";
|
||||
const SESSION: &str = "http-test-session";
|
||||
|
||||
#[derive(Clone, Copy, Debug)]
|
||||
enum WriteResponseFault {
|
||||
Malformed,
|
||||
LostBody,
|
||||
}
|
||||
|
||||
struct ProviderState {
|
||||
session: WorkdirSessionHandle,
|
||||
requests: Mutex<Vec<(String, String)>>,
|
||||
operations: Mutex<Vec<WorkdirSessionOperation>>,
|
||||
write_fault: Mutex<Option<WriteResponseFault>>,
|
||||
}
|
||||
|
||||
impl ProviderState {
|
||||
fn authorize(&self, headers: &HeaderMap, method: &str, uri: &Uri) -> Result<(), AxumResponse> {
|
||||
self.requests
|
||||
.lock()
|
||||
.unwrap()
|
||||
.push((method.into(), uri.path().into()));
|
||||
let expected = format!("Bearer {TOKEN}");
|
||||
if headers.get("authorization").and_then(|h| h.to_str().ok()) != Some(expected.as_str()) {
|
||||
return Err((
|
||||
StatusCode::UNAUTHORIZED,
|
||||
AxumJson(WorkdirTransportError {
|
||||
code: WorkdirTransportErrorCode::Denied,
|
||||
message: "provider authorization denied".into(),
|
||||
}),
|
||||
)
|
||||
.into_response());
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn execute_calls(&self) -> usize {
|
||||
self.operations
|
||||
.lock()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.filter(|op| matches!(op, WorkdirSessionOperation::CheckoutExecute(_)))
|
||||
.count()
|
||||
}
|
||||
|
||||
fn write_calls(&self) -> usize {
|
||||
self.operations
|
||||
.lock()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.filter(|op| {
|
||||
matches!(
|
||||
op,
|
||||
WorkdirSessionOperation::CheckoutExecute(CheckoutRequest {
|
||||
operation: CheckoutOperation::Write { .. },
|
||||
..
|
||||
})
|
||||
)
|
||||
})
|
||||
.count()
|
||||
}
|
||||
}
|
||||
|
||||
fn provider_error(error: WorkdirError) -> AxumResponse {
|
||||
let error = WorkdirTransportError::from_workdir_error(&error);
|
||||
(
|
||||
StatusCode::from_u16(error.code.http_status()).unwrap(),
|
||||
AxumJson(error),
|
||||
)
|
||||
.into_response()
|
||||
}
|
||||
|
||||
async fn open_session(
|
||||
State(state): State<Arc<ProviderState>>,
|
||||
Path(id): Path<String>,
|
||||
uri: Uri,
|
||||
headers: HeaderMap,
|
||||
AxumJson(_request): AxumJson<OpenWorkdirSessionRequest>,
|
||||
) -> AxumResponse {
|
||||
if let Err(response) = state.authorize(&headers, "POST", &uri) {
|
||||
return response;
|
||||
}
|
||||
if id != state.session.workdir().id().as_str() {
|
||||
return provider_error(WorkdirError::NotFound("unknown workdir".into()));
|
||||
}
|
||||
AxumJson(OpenWorkdirSessionResponse {
|
||||
session_id: WorkdirSessionId::new(SESSION).unwrap(),
|
||||
workdir_id: state.session.workdir().id().clone(),
|
||||
capabilities: state.session.capabilities(),
|
||||
})
|
||||
.into_response()
|
||||
}
|
||||
|
||||
async fn operate(
|
||||
State(state): State<Arc<ProviderState>>,
|
||||
Path(id): Path<String>,
|
||||
uri: Uri,
|
||||
headers: HeaderMap,
|
||||
AxumJson(request): AxumJson<WorkdirSessionOperationRequest>,
|
||||
) -> AxumResponse {
|
||||
if let Err(response) = state.authorize(&headers, "POST", &uri) {
|
||||
return response;
|
||||
}
|
||||
if id != SESSION {
|
||||
return provider_error(WorkdirError::Unavailable("unknown session".into()));
|
||||
}
|
||||
let is_write = matches!(
|
||||
&request.operation,
|
||||
WorkdirSessionOperation::CheckoutExecute(CheckoutRequest {
|
||||
operation: CheckoutOperation::Write { .. },
|
||||
..
|
||||
})
|
||||
);
|
||||
state
|
||||
.operations
|
||||
.lock()
|
||||
.unwrap()
|
||||
.push(request.operation.clone());
|
||||
let result =
|
||||
dispatch_workdir_session_operation(state.session.as_ref(), request.operation).await;
|
||||
match result {
|
||||
// Inject only AFTER the real LocalWorkdirSession successfully committed.
|
||||
Ok(result) => {
|
||||
let fault = if is_write {
|
||||
state.write_fault.lock().unwrap().take()
|
||||
} else {
|
||||
None
|
||||
};
|
||||
match fault {
|
||||
Some(WriteResponseFault::Malformed) => (
|
||||
StatusCode::OK,
|
||||
[("content-type", "application/json")],
|
||||
"{broken-response",
|
||||
)
|
||||
.into_response(),
|
||||
Some(WriteResponseFault::LostBody) => {
|
||||
let body = Body::from_stream(futures::stream::once(async {
|
||||
Err::<Vec<u8>, _>(std::io::Error::other("response lost after commit"))
|
||||
}));
|
||||
AxumResponse::builder()
|
||||
.status(StatusCode::OK)
|
||||
.header("content-type", "application/json")
|
||||
.body(body)
|
||||
.unwrap()
|
||||
}
|
||||
None => AxumJson(result).into_response(),
|
||||
}
|
||||
}
|
||||
Err(error) => provider_error(error),
|
||||
}
|
||||
}
|
||||
|
||||
async fn close_session(
|
||||
State(state): State<Arc<ProviderState>>,
|
||||
Path(id): Path<String>,
|
||||
uri: Uri,
|
||||
headers: HeaderMap,
|
||||
) -> AxumResponse {
|
||||
if let Err(response) = state.authorize(&headers, "DELETE", &uri) {
|
||||
return response;
|
||||
}
|
||||
if id != SESSION {
|
||||
return provider_error(WorkdirError::Unavailable("unknown session".into()));
|
||||
}
|
||||
match state.session.close().await {
|
||||
Ok(()) => StatusCode::NO_CONTENT.into_response(),
|
||||
Err(error) => provider_error(error),
|
||||
}
|
||||
}
|
||||
|
||||
struct HttpFixture {
|
||||
dir: TempDir,
|
||||
state: Arc<ProviderState>,
|
||||
base_url: reqwest::Url,
|
||||
server: tokio::task::JoinHandle<()>,
|
||||
}
|
||||
|
||||
impl Drop for HttpFixture {
|
||||
fn drop(&mut self) {
|
||||
self.server.abort();
|
||||
}
|
||||
}
|
||||
|
||||
impl HttpFixture {
|
||||
async fn new(capabilities: WorkdirSessionCapabilities) -> Self {
|
||||
let dir = TempDir::new().unwrap();
|
||||
std::fs::create_dir_all(dir.path().join("src/deep")).unwrap();
|
||||
std::fs::write(
|
||||
dir.path().join("src/deep/a.txt"),
|
||||
"first\nneedle needle\nlast\n",
|
||||
)
|
||||
.unwrap();
|
||||
let session = Arc::new(LocalWorkdirSession::materialized(
|
||||
dir.path().to_path_buf(),
|
||||
dir.path().to_path_buf(),
|
||||
manifest::SharedScope::new(manifest::Scope::writable(dir.path()).unwrap()),
|
||||
capabilities,
|
||||
));
|
||||
let state = Arc::new(ProviderState {
|
||||
session,
|
||||
requests: Mutex::new(Vec::new()),
|
||||
operations: Mutex::new(Vec::new()),
|
||||
write_fault: Mutex::new(None),
|
||||
});
|
||||
let app = Router::new()
|
||||
.route(
|
||||
"/provider/v1/working-directories/{id}/sessions",
|
||||
post(open_session),
|
||||
)
|
||||
.route(
|
||||
"/provider/v1/workdir-sessions/{id}/operations",
|
||||
post(operate),
|
||||
)
|
||||
.route("/provider/v1/workdir-sessions/{id}", delete(close_session))
|
||||
.with_state(state.clone());
|
||||
// Binding before spawn makes readiness deterministic; no sleeps or polling.
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let base_url = format!("http://{}/provider/", listener.local_addr().unwrap())
|
||||
.parse()
|
||||
.unwrap();
|
||||
let server = tokio::spawn(async move {
|
||||
axum::serve(listener, app).await.unwrap();
|
||||
});
|
||||
Self {
|
||||
dir,
|
||||
state,
|
||||
base_url,
|
||||
server,
|
||||
}
|
||||
}
|
||||
|
||||
async fn open(&self, token: &str) -> Result<RemoteWorkdirSession, WorkdirError> {
|
||||
RemoteWorkdirSession::open(
|
||||
reqwest::Client::builder()
|
||||
.no_proxy()
|
||||
.timeout(Duration::from_secs(5))
|
||||
.build()
|
||||
.unwrap(),
|
||||
self.base_url.clone(),
|
||||
token.to_owned(),
|
||||
self.state.session.workdir().id().clone(),
|
||||
OpenWorkdirSessionRequest::default(),
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
fn public_output(&self, text: &str) {
|
||||
assert!(!text.contains(TOKEN), "bearer token leaked");
|
||||
assert!(
|
||||
!text.contains(self.dir.path().to_str().unwrap()),
|
||||
"host path leaked"
|
||||
);
|
||||
assert!(
|
||||
!text.contains("/provider/"),
|
||||
"transport prefix leaked into Worldspace"
|
||||
);
|
||||
}
|
||||
|
||||
fn assert_endpoints(&self) {
|
||||
let open = format!(
|
||||
"/provider/v1/working-directories/{}/sessions",
|
||||
self.state.session.workdir().id()
|
||||
);
|
||||
let operations = format!("/provider/v1/workdir-sessions/{SESSION}/operations");
|
||||
let close = format!("/provider/v1/workdir-sessions/{SESSION}");
|
||||
let requests = self.state.requests.lock().unwrap();
|
||||
assert!(requests.contains(&("POST".into(), open.clone())));
|
||||
assert!(requests.contains(&("POST".into(), operations.clone())));
|
||||
assert_eq!(
|
||||
requests
|
||||
.iter()
|
||||
.filter(|(method, path)| method == "DELETE" && path == &close)
|
||||
.count(),
|
||||
1
|
||||
);
|
||||
assert!(requests.iter().all(|(method, path)| (method == "POST"
|
||||
&& (path == &open || path == &operations))
|
||||
|| (method == "DELETE" && path == &close)));
|
||||
}
|
||||
}
|
||||
|
||||
fn native_runtime(remote: Arc<RemoteWorkdirSession>) -> (WipRuntime, Arc<WorkdirSessionRouter>) {
|
||||
let router = Arc::new(WorkdirSessionRouter::new());
|
||||
router
|
||||
.attach(WorkdirAttachmentAlias::new("main").unwrap(), remote)
|
||||
.unwrap();
|
||||
let mut registry = WipMountRegistry::new();
|
||||
crate::checkout::mount_checkouts(&mut registry, router.clone(), tools::Tracker::new(), None)
|
||||
.unwrap();
|
||||
let runtime = WipRuntime::new(
|
||||
WipHost::new(registry),
|
||||
SecurityContext::new("http-checkout-worker"),
|
||||
0,
|
||||
)
|
||||
.unwrap();
|
||||
(runtime, router)
|
||||
}
|
||||
|
||||
async fn observe_interface(runtime: &WipRuntime, path: &str) -> String {
|
||||
runtime.discover(path.into(), 0, true).await.unwrap();
|
||||
// Select from the real Client's Known Space, not a direct handler invocation.
|
||||
let interface = {
|
||||
let state = runtime.state.lock().unwrap();
|
||||
state
|
||||
.client
|
||||
.known_space(&state.session)
|
||||
.unwrap()
|
||||
.into_iter()
|
||||
.find(|o| o.path == path)
|
||||
.unwrap()
|
||||
.object
|
||||
.unwrap()
|
||||
.interfaces[0]
|
||||
.clone()
|
||||
};
|
||||
runtime.inspect(interface.clone(), true).await.unwrap();
|
||||
interface
|
||||
}
|
||||
|
||||
async fn native_call(runtime: &WipRuntime, path: &str, operation: &str, args: Json) -> Value {
|
||||
let interface = observe_interface(runtime, path).await;
|
||||
runtime
|
||||
.call(
|
||||
path.into(),
|
||||
interface,
|
||||
operation.into(),
|
||||
args,
|
||||
Default::default(),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let state = runtime.state.lock().unwrap();
|
||||
match &state
|
||||
.client
|
||||
.call_history(&state.session)
|
||||
.unwrap()
|
||||
.back()
|
||||
.unwrap()
|
||||
.outcome
|
||||
{
|
||||
CallOutcome::Success(response) => response.result.clone(),
|
||||
outcome => panic!("expected native Client success, got {outcome:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
fn field<'a>(value: &'a Value, name: &str) -> &'a Value {
|
||||
let Value::Record(fields) = value else {
|
||||
panic!("expected typed Record")
|
||||
};
|
||||
fields.get(name).unwrap()
|
||||
}
|
||||
|
||||
fn first_link(value: &Value) -> &str {
|
||||
let Value::List(items) = field(value, "items") else {
|
||||
panic!("expected typed link List")
|
||||
};
|
||||
assert_eq!(items.len(), 1);
|
||||
let Value::String(path) = field(&items[0], "path") else {
|
||||
panic!("expected typed link path")
|
||||
};
|
||||
path
|
||||
}
|
||||
|
||||
async fn watchdog(future: impl std::future::Future<Output = ()>) {
|
||||
tokio::time::timeout(Duration::from_secs(20), future)
|
||||
.await
|
||||
.expect("HTTP checkout test watchdog");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn checkout_http_native_search_links_read_edit_write_create_and_close() {
|
||||
watchdog(async {
|
||||
let fixture = HttpFixture::new(WorkdirSessionCapabilities::READ_WRITE).await;
|
||||
let remote = Arc::new(fixture.open(TOKEN).await.unwrap());
|
||||
let (runtime, router) = native_runtime(remote);
|
||||
let glob = native_call(
|
||||
&runtime,
|
||||
"/checkouts/main",
|
||||
"glob",
|
||||
json!({"pattern":"**/*.txt"}),
|
||||
)
|
||||
.await;
|
||||
let file = first_link(&glob).to_owned();
|
||||
assert_eq!(file, "/checkouts/main/src/deep/a.txt");
|
||||
let grep = native_call(
|
||||
&runtime,
|
||||
"/checkouts/main/src",
|
||||
"grep",
|
||||
json!({
|
||||
"pattern":"needle", "output_mode":"files_with_matches"
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
assert_eq!(first_link(&grep), file);
|
||||
let read = native_call(
|
||||
&runtime,
|
||||
first_link(&grep),
|
||||
"read",
|
||||
json!({"offset":1,"limit":1}),
|
||||
)
|
||||
.await;
|
||||
let Value::String(content) = field(&read, "content") else {
|
||||
panic!("expected line Read text")
|
||||
};
|
||||
assert_eq!(content, " 2\tneedle needle\n");
|
||||
let edit = native_call(
|
||||
&runtime,
|
||||
&file,
|
||||
"edit",
|
||||
json!({
|
||||
"old_string":"needle", "new_string":"done", "replace_all":true
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
assert_eq!(
|
||||
std::fs::read_to_string(fixture.dir.path().join("src/deep/a.txt")).unwrap(),
|
||||
"first\ndone done\nlast\n"
|
||||
);
|
||||
let write = native_call(&runtime, &file, "write", json!({"content":"saved\n"})).await;
|
||||
assert_eq!(
|
||||
std::fs::read_to_string(fixture.dir.path().join("src/deep/a.txt")).unwrap(),
|
||||
"saved\n"
|
||||
);
|
||||
let create = native_call(
|
||||
&runtime,
|
||||
"/checkouts/main/src",
|
||||
"create_file",
|
||||
json!({
|
||||
"path":"new/deeper.txt", "content":"created\n"
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
assert_eq!(
|
||||
std::fs::read_to_string(fixture.dir.path().join("src/new/deeper.txt")).unwrap(),
|
||||
"created\n"
|
||||
);
|
||||
let created = native_call(
|
||||
&runtime,
|
||||
"/checkouts/main/src/new/deeper.txt",
|
||||
"read",
|
||||
json!({}),
|
||||
)
|
||||
.await;
|
||||
for value in [&glob, &grep, &read, &edit, &write, &create, &created] {
|
||||
fixture.public_output(&serde_json::to_string(&wip_to_json(value).unwrap()).unwrap());
|
||||
}
|
||||
let operations = fixture.state.operations.lock().unwrap();
|
||||
assert!(
|
||||
operations
|
||||
.iter()
|
||||
.any(|op| matches!(op, WorkdirSessionOperation::CheckoutSearch(r)
|
||||
if matches!(r.operation, CheckoutSearchOperation::Glob(_))))
|
||||
);
|
||||
assert!(
|
||||
operations
|
||||
.iter()
|
||||
.any(|op| matches!(op, WorkdirSessionOperation::CheckoutSearch(r)
|
||||
if matches!(r.operation, CheckoutSearchOperation::Grep(_))))
|
||||
);
|
||||
assert!(operations.iter().any(
|
||||
|op| matches!(op, WorkdirSessionOperation::CheckoutExecute(r)
|
||||
if matches!(r.operation, CheckoutOperation::Edit { .. }))
|
||||
));
|
||||
assert!(operations.iter().any(
|
||||
|op| matches!(op, WorkdirSessionOperation::CheckoutExecute(r)
|
||||
if matches!(r.operation, CheckoutOperation::Create { .. }))
|
||||
));
|
||||
drop(operations);
|
||||
assert_eq!(fixture.state.write_calls(), 1);
|
||||
router
|
||||
.detach(&WorkdirAttachmentAlias::new("main").unwrap())
|
||||
.await
|
||||
.unwrap();
|
||||
fixture.assert_endpoints();
|
||||
})
|
||||
.await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn checkout_http_wrong_bearer_and_readonly_caps_fail_closed() {
|
||||
watchdog(async {
|
||||
let fixture = HttpFixture::new(WorkdirSessionCapabilities::READ_ONLY).await;
|
||||
let error = fixture.open("wrong-bearer").await.unwrap_err();
|
||||
assert!(matches!(error, WorkdirError::Denied(_)));
|
||||
fixture.public_output(&error.to_string());
|
||||
assert!(fixture.state.operations.lock().unwrap().is_empty());
|
||||
let remote = Arc::new(fixture.open(TOKEN).await.unwrap());
|
||||
assert_eq!(remote.capabilities(), WorkdirSessionCapabilities::READ_ONLY);
|
||||
let (runtime, router) = native_runtime(remote.clone());
|
||||
let file = "/checkouts/main/src/deep/a.txt";
|
||||
native_call(&runtime, file, "read", json!({})).await;
|
||||
let interface = observe_interface(&runtime, file).await;
|
||||
{
|
||||
let state = runtime.state.lock().unwrap();
|
||||
let observed = state.client.interface(&state.session, &interface).unwrap();
|
||||
let descriptor = observed.descriptor.as_ref().unwrap();
|
||||
assert_eq!(
|
||||
descriptor
|
||||
.operations
|
||||
.iter()
|
||||
.map(|op| op.name.as_str())
|
||||
.collect::<Vec<_>>(),
|
||||
["read"]
|
||||
);
|
||||
}
|
||||
let directory_interface = observe_interface(&runtime, "/checkouts/main").await;
|
||||
let before = fixture.state.execute_calls();
|
||||
for (path, interface, operation, args) in [
|
||||
(
|
||||
file,
|
||||
interface.clone(),
|
||||
"write",
|
||||
json!({"content":"forbidden"}),
|
||||
),
|
||||
(
|
||||
file,
|
||||
interface,
|
||||
"edit",
|
||||
json!({"old_string":"needle", "new_string":"forbidden"}),
|
||||
),
|
||||
(
|
||||
"/checkouts/main",
|
||||
directory_interface,
|
||||
"create_file",
|
||||
json!({"path":"forbidden.txt","content":"bad"}),
|
||||
),
|
||||
] {
|
||||
let error = runtime
|
||||
.call(
|
||||
path.into(),
|
||||
interface,
|
||||
operation.into(),
|
||||
args,
|
||||
Default::default(),
|
||||
)
|
||||
.await
|
||||
.unwrap_err();
|
||||
fixture.public_output(&error.to_string());
|
||||
}
|
||||
// Preflight may observe objects, but no rejected mutation reaches execution.
|
||||
assert_eq!(fixture.state.execute_calls(), before);
|
||||
// A caller bypassing WIP publication still cannot mutate at the provider.
|
||||
let parent = remote.checkout_observe(WorkdirPath::root()).await.unwrap();
|
||||
let error = remote
|
||||
.checkout_execute(CheckoutRequest {
|
||||
target: parent.path,
|
||||
validator: parent.validator,
|
||||
operation: CheckoutOperation::Create {
|
||||
path: WorkdirPath::new("forbidden.txt").unwrap(),
|
||||
content: b"bad".to_vec(),
|
||||
},
|
||||
})
|
||||
.await
|
||||
.unwrap_err();
|
||||
assert!(matches!(
|
||||
error,
|
||||
WorkdirError::Unsupported(_)
|
||||
| WorkdirError::UnsupportedOperation(_)
|
||||
| WorkdirError::ReadOnly(_)
|
||||
| WorkdirError::Denied(_)
|
||||
));
|
||||
fixture.public_output(&error.to_string());
|
||||
assert_eq!(
|
||||
std::fs::read_to_string(fixture.dir.path().join("src/deep/a.txt")).unwrap(),
|
||||
"first\nneedle needle\nlast\n"
|
||||
);
|
||||
assert!(!fixture.dir.path().join("forbidden.txt").exists());
|
||||
router
|
||||
.detach(&WorkdirAttachmentAlias::new("main").unwrap())
|
||||
.await
|
||||
.unwrap();
|
||||
fixture.assert_endpoints();
|
||||
})
|
||||
.await;
|
||||
}
|
||||
|
||||
async fn committed_write_unknown(fault: WriteResponseFault) {
|
||||
let fixture = HttpFixture::new(WorkdirSessionCapabilities::READ_WRITE).await;
|
||||
let remote = Arc::new(fixture.open(TOKEN).await.unwrap());
|
||||
let (runtime, router) = native_runtime(remote);
|
||||
let file = "/checkouts/main/src/deep/a.txt";
|
||||
native_call(&runtime, file, "read", json!({})).await;
|
||||
let interface = observe_interface(&runtime, file).await;
|
||||
*fixture.state.write_fault.lock().unwrap() = Some(fault);
|
||||
let error = runtime
|
||||
.call(
|
||||
file.into(),
|
||||
interface,
|
||||
"write".into(),
|
||||
json!({"content":"committed\n"}),
|
||||
Default::default(),
|
||||
)
|
||||
.await
|
||||
.unwrap_err();
|
||||
assert!(error.to_string().contains("outcome unknown"));
|
||||
assert!(error.to_string().contains("do not retry automatically"));
|
||||
fixture.public_output(&error.to_string());
|
||||
assert_eq!(
|
||||
runtime.audit().last().unwrap().outcome,
|
||||
WipAuditOutcome::OutcomeUnknown
|
||||
);
|
||||
{
|
||||
let state = runtime.state.lock().unwrap();
|
||||
assert!(matches!(
|
||||
state
|
||||
.client
|
||||
.call_history(&state.session)
|
||||
.unwrap()
|
||||
.back()
|
||||
.unwrap()
|
||||
.outcome,
|
||||
CallOutcome::Unknown { .. }
|
||||
));
|
||||
}
|
||||
assert_eq!(
|
||||
std::fs::read_to_string(fixture.dir.path().join("src/deep/a.txt")).unwrap(),
|
||||
"committed\n"
|
||||
);
|
||||
assert_eq!(
|
||||
fixture.state.write_calls(),
|
||||
1,
|
||||
"committed Write must not be automatically retried"
|
||||
);
|
||||
// Explicit inspection of the effect is safe; it must not replay the mutation.
|
||||
let read = native_call(&runtime, file, "read", json!({})).await;
|
||||
fixture.public_output(&serde_json::to_string(&wip_to_json(&read).unwrap()).unwrap());
|
||||
let Value::String(content) = field(&read, "content") else {
|
||||
panic!("expected Read text")
|
||||
};
|
||||
assert!(content.contains("committed"));
|
||||
assert_eq!(fixture.state.write_calls(), 1);
|
||||
router
|
||||
.detach(&WorkdirAttachmentAlias::new("main").unwrap())
|
||||
.await
|
||||
.unwrap();
|
||||
fixture.assert_endpoints();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn checkout_http_malformed_response_after_write_is_unknown_without_retry() {
|
||||
watchdog(committed_write_unknown(WriteResponseFault::Malformed)).await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn checkout_http_lost_response_after_write_is_unknown_without_retry() {
|
||||
watchdog(committed_write_unknown(WriteResponseFault::LostBody)).await;
|
||||
}
|
||||
@@ -0,0 +1,469 @@
|
||||
//! Deterministic attachment races at provider observation boundaries.
|
||||
//!
|
||||
//! These tests deliberately avoid spawned tasks, sleeps, and detach attempts
|
||||
//! inside a routed operation (which is busy and cannot be detached). Deadline
|
||||
//! tests use the provider's configurable deadline and a permanently pending
|
||||
//! observation, without Tokio's test-util clock or the production 30-second wait.
|
||||
use super::*;
|
||||
|
||||
use std::future::pending;
|
||||
use std::sync::Mutex;
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use tempfile::TempDir;
|
||||
use workdir::{
|
||||
CheckoutRequest, CheckoutResult, CommandHandle, CommandOutput, CommandOutputRequest,
|
||||
CommandRequest, CommandStatus, EditRequest, EditResult, GlobRequest, GlobResult, GrepRequest,
|
||||
GrepResult, ListResult, LocalWorkdirSession, ReadBytesRequest, ReadBytesResult, ReadRequest,
|
||||
ReadResult, StatRequest, StatResult, Workdir, WorkdirAttachmentAlias, WorkdirError,
|
||||
WorkdirScopeAuthorizationRequest, WorkdirScopeOverlapRequest, WorkdirSession,
|
||||
WorkdirSessionCapabilities, WorkdirSessionHandle, WriteRequest, WriteResult,
|
||||
};
|
||||
|
||||
type ObserveHook = Box<dyn FnOnce(&WorkdirPath) + Send>;
|
||||
|
||||
/// Shared local provider with a one-shot, synchronous observation boundary.
|
||||
/// Take the callback out of the lock before invoking it: it may call the router.
|
||||
struct HookedSession {
|
||||
local: LocalWorkdirSession,
|
||||
observe_hook: Mutex<Option<ObserveHook>>,
|
||||
stall_path: Mutex<Option<WorkdirPath>>,
|
||||
stalled_observations: AtomicUsize,
|
||||
}
|
||||
|
||||
impl std::fmt::Debug for HookedSession {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
f.debug_struct("HookedSession")
|
||||
.field("local", &self.local)
|
||||
.field("stall_path", &self.stall_path)
|
||||
.field("stalled_observations", &self.stalled_observations)
|
||||
.finish_non_exhaustive()
|
||||
}
|
||||
}
|
||||
|
||||
impl HookedSession {
|
||||
fn new(dir: &TempDir) -> Arc<Self> {
|
||||
Arc::new(Self {
|
||||
local: LocalWorkdirSession::materialized(
|
||||
dir.path().to_path_buf(),
|
||||
dir.path().to_path_buf(),
|
||||
manifest::SharedScope::new(manifest::Scope::writable(dir.path()).unwrap()),
|
||||
WorkdirSessionCapabilities::READ_WRITE,
|
||||
),
|
||||
observe_hook: Mutex::new(None),
|
||||
stall_path: Mutex::new(None),
|
||||
stalled_observations: AtomicUsize::new(0),
|
||||
})
|
||||
}
|
||||
|
||||
fn on_next_observe(&self, hook: impl FnOnce(&WorkdirPath) + Send + 'static) {
|
||||
let mut slot = self.observe_hook.lock().unwrap();
|
||||
assert!(slot.is_none(), "an observation hook is already armed");
|
||||
*slot = Some(Box::new(hook));
|
||||
}
|
||||
|
||||
fn stall_observation(&self, path: Option<WorkdirPath>) {
|
||||
*self.stall_path.lock().unwrap() = path;
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl WorkdirSession for HookedSession {
|
||||
fn workdir(&self) -> &Workdir {
|
||||
self.local.workdir()
|
||||
}
|
||||
|
||||
fn capabilities(&self) -> WorkdirSessionCapabilities {
|
||||
self.local.capabilities()
|
||||
}
|
||||
|
||||
async fn authorize_scope_path(
|
||||
&self,
|
||||
request: WorkdirScopeAuthorizationRequest,
|
||||
) -> Result<(), WorkdirError> {
|
||||
self.local.authorize_scope_path(request).await
|
||||
}
|
||||
|
||||
async fn scope_rules_overlap(
|
||||
&self,
|
||||
request: WorkdirScopeOverlapRequest,
|
||||
) -> Result<bool, WorkdirError> {
|
||||
self.local.scope_rules_overlap(request).await
|
||||
}
|
||||
|
||||
async fn checkout_observe(
|
||||
&self,
|
||||
path: WorkdirPath,
|
||||
) -> Result<CheckoutObservation, WorkdirError> {
|
||||
let hook = { self.observe_hook.lock().unwrap().take() };
|
||||
if let Some(hook) = hook {
|
||||
hook(&path);
|
||||
}
|
||||
let should_stall = {
|
||||
self.stall_path
|
||||
.lock()
|
||||
.unwrap()
|
||||
.as_ref()
|
||||
.is_some_and(|stall| stall.as_str() == path.as_str())
|
||||
};
|
||||
if should_stall {
|
||||
self.stalled_observations.fetch_add(1, Ordering::SeqCst);
|
||||
return pending().await;
|
||||
}
|
||||
self.local.checkout_observe(path).await
|
||||
}
|
||||
|
||||
async fn checkout_execute(
|
||||
&self,
|
||||
request: CheckoutRequest,
|
||||
) -> Result<CheckoutResult, WorkdirError> {
|
||||
self.local.checkout_execute(request).await
|
||||
}
|
||||
|
||||
async fn checkout_search(
|
||||
&self,
|
||||
request: workdir::CheckoutSearchRequest,
|
||||
) -> Result<workdir::CheckoutSearchResult, WorkdirError> {
|
||||
self.local.checkout_search(request).await
|
||||
}
|
||||
|
||||
async fn stat(&self, request: StatRequest) -> Result<StatResult, WorkdirError> {
|
||||
self.local.stat(request).await
|
||||
}
|
||||
|
||||
async fn read(&self, request: ReadRequest) -> Result<ReadResult, WorkdirError> {
|
||||
self.local.read(request).await
|
||||
}
|
||||
|
||||
async fn read_bytes(&self, request: ReadBytesRequest) -> Result<ReadBytesResult, WorkdirError> {
|
||||
self.local.read_bytes(request).await
|
||||
}
|
||||
|
||||
async fn write(&self, request: WriteRequest) -> Result<WriteResult, WorkdirError> {
|
||||
self.local.write(request).await
|
||||
}
|
||||
|
||||
async fn edit(&self, request: EditRequest) -> Result<EditResult, WorkdirError> {
|
||||
self.local.edit(request).await
|
||||
}
|
||||
|
||||
async fn list(&self, request: ListRequest) -> Result<ListResult, WorkdirError> {
|
||||
self.local.list(request).await
|
||||
}
|
||||
|
||||
async fn glob(&self, request: GlobRequest) -> Result<GlobResult, WorkdirError> {
|
||||
self.local.glob(request).await
|
||||
}
|
||||
|
||||
async fn grep(&self, request: GrepRequest) -> Result<GrepResult, WorkdirError> {
|
||||
self.local.grep(request).await
|
||||
}
|
||||
|
||||
async fn start_command(&self, request: CommandRequest) -> Result<CommandHandle, WorkdirError> {
|
||||
self.local.start_command(request).await
|
||||
}
|
||||
|
||||
async fn command_status(&self, handle: CommandHandle) -> Result<CommandStatus, WorkdirError> {
|
||||
self.local.command_status(handle).await
|
||||
}
|
||||
|
||||
async fn command_output(
|
||||
&self,
|
||||
request: CommandOutputRequest,
|
||||
) -> Result<CommandOutput, WorkdirError> {
|
||||
self.local.command_output(request).await
|
||||
}
|
||||
|
||||
async fn cancel_command(&self, handle: CommandHandle) -> Result<(), WorkdirError> {
|
||||
self.local.cancel_command(handle).await
|
||||
}
|
||||
|
||||
async fn close(&self) -> Result<(), WorkdirError> {
|
||||
self.local.close().await
|
||||
}
|
||||
}
|
||||
|
||||
fn provider(router: Arc<WorkdirSessionRouter>) -> Arc<Provider> {
|
||||
Arc::new(Provider {
|
||||
router,
|
||||
tracker: tools::Tracker::new(),
|
||||
permissions: None,
|
||||
incarnation: "checkout-race-tests".into(),
|
||||
permits: Arc::new(tokio::sync::Semaphore::new(16)),
|
||||
deadline: DEADLINE,
|
||||
})
|
||||
}
|
||||
|
||||
fn deadline_provider(router: Arc<WorkdirSessionRouter>) -> Arc<Provider> {
|
||||
let mut p = provider(router);
|
||||
let settings = Arc::get_mut(&mut p).unwrap();
|
||||
// Allow ample time for the tiny local search before stalling post-processing.
|
||||
// The outer test watchdog is intentionally much more generous than this.
|
||||
settings.deadline = Duration::from_millis(100);
|
||||
settings.permits = Arc::new(tokio::sync::Semaphore::new(1));
|
||||
p
|
||||
}
|
||||
|
||||
fn context() -> WipCallContext {
|
||||
WipCallContext {
|
||||
execution: Default::default(),
|
||||
security_context: "checkout-race-tests".into(),
|
||||
}
|
||||
}
|
||||
|
||||
fn assert_validator_mismatch<T>(result: Result<T, WipOperationError>) {
|
||||
match result {
|
||||
Err(WipOperationError::Protocol(e)) => {
|
||||
assert_eq!(e.code, ProtocolErrorCode::ValidatorMismatch);
|
||||
}
|
||||
Err(_) => panic!("expected ValidatorMismatch, received a non-protocol error"),
|
||||
Ok(_) => panic!("a changed attachment snapshot must not be published"),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn collection_list_rejects_alias_attached_during_observation_and_retries_coherently() {
|
||||
let main_dir = TempDir::new().unwrap();
|
||||
let extra_dir = TempDir::new().unwrap();
|
||||
let main = HookedSession::new(&main_dir);
|
||||
let extra: WorkdirSessionHandle = HookedSession::new(&extra_dir);
|
||||
let router = Arc::new(WorkdirSessionRouter::new());
|
||||
router
|
||||
.attach(WorkdirAttachmentAlias::new("main").unwrap(), main.clone())
|
||||
.unwrap();
|
||||
let p = provider(router.clone());
|
||||
let before = p.collection().object.validator;
|
||||
let hook_router = router.clone();
|
||||
main.on_next_observe(move |path| {
|
||||
assert!(path.is_root(), "collection enumeration observes the root");
|
||||
hook_router
|
||||
.attach(WorkdirAttachmentAlias::new("extra").unwrap(), extra)
|
||||
.unwrap();
|
||||
});
|
||||
|
||||
let collection = Collection {
|
||||
provider: p.clone(),
|
||||
};
|
||||
assert_validator_mismatch(collection.call("list", &BTreeMap::new(), context()).await);
|
||||
assert_ne!(before, p.collection().object.validator);
|
||||
assert!(router.resolve(Some("extra")).is_ok(), "the hook must run");
|
||||
|
||||
// The callback was consumed. A fresh snapshot includes both aliases, with
|
||||
// identity metadata from exactly the router state being enumerated.
|
||||
let output = match collection.call("list", &BTreeMap::new(), context()).await {
|
||||
Ok(output) => output,
|
||||
Err(_) => panic!("the stable retry must succeed"),
|
||||
};
|
||||
let value = wip_to_json(&output.value).unwrap();
|
||||
let items = value["items"].as_array().unwrap();
|
||||
assert_eq!(items.len(), 2);
|
||||
for alias in ["extra", "main"] {
|
||||
let item = items
|
||||
.iter()
|
||||
.find(|item| item["alias"] == json!(alias))
|
||||
.expect("each attached alias must appear in the stable snapshot");
|
||||
let selected = router.resolve(Some(alias)).unwrap();
|
||||
assert_eq!(item["alias"], json!(alias));
|
||||
assert_eq!(item["path"], json!(checkout_root(alias)));
|
||||
assert_eq!(item["slug"], json!(encode_identity(alias)));
|
||||
assert_eq!(item["generation"], json!(selected.generation));
|
||||
assert_eq!(
|
||||
item["working_directory_id"],
|
||||
json!(selected.session.workdir().id().as_str())
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn root_enumeration_rejects_alias_attached_during_observation() {
|
||||
let main_dir = TempDir::new().unwrap();
|
||||
let extra_dir = TempDir::new().unwrap();
|
||||
let main = HookedSession::new(&main_dir);
|
||||
let extra: WorkdirSessionHandle = HookedSession::new(&extra_dir);
|
||||
let router = Arc::new(WorkdirSessionRouter::new());
|
||||
router
|
||||
.attach(WorkdirAttachmentAlias::new("main").unwrap(), main.clone())
|
||||
.unwrap();
|
||||
let hook_router = router.clone();
|
||||
main.on_next_observe(move |_| {
|
||||
hook_router
|
||||
.attach(WorkdirAttachmentAlias::new("extra").unwrap(), extra)
|
||||
.unwrap();
|
||||
});
|
||||
let subtree = Arc::new(provider(router.clone()));
|
||||
|
||||
match subtree.children(ROOT).await {
|
||||
Err(e) => assert_eq!(e.code, ProtocolErrorCode::ValidatorMismatch),
|
||||
Ok(_) => panic!("enumeration must reject a mixed attachment snapshot"),
|
||||
}
|
||||
assert!(router.resolve(Some("extra")).is_ok(), "the hook must run");
|
||||
let mut paths = subtree.children(ROOT).await.unwrap();
|
||||
paths.sort();
|
||||
assert_eq!(paths, vec![checkout_root("extra"), checkout_root("main")]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn file_handler_connection_fences_detach_and_same_session_alias_reuse() {
|
||||
let dir = TempDir::new().unwrap();
|
||||
std::fs::write(dir.path().join("file.txt"), "unchanged\n").unwrap();
|
||||
let session: WorkdirSessionHandle = HookedSession::new(&dir);
|
||||
let router = Arc::new(WorkdirSessionRouter::new());
|
||||
let alias = WorkdirAttachmentAlias::new("main").unwrap();
|
||||
router.attach(alias.clone(), session.clone()).unwrap();
|
||||
let selected = router.resolve(Some("main")).unwrap();
|
||||
let target = WorkdirPath::new("file.txt").unwrap();
|
||||
let observation = selected
|
||||
.session
|
||||
.checkout_observe(target.clone())
|
||||
.await
|
||||
.unwrap();
|
||||
let mut handler = FileHandler {
|
||||
provider: provider(router.clone()),
|
||||
alias: "main".into(),
|
||||
generation: selected.generation,
|
||||
workdir: selected.session.workdir().id().as_str().into(),
|
||||
target,
|
||||
observation,
|
||||
};
|
||||
drop(selected);
|
||||
assert!(handler.check_connection().is_ok());
|
||||
|
||||
// No routed operation is active here. Reuse the exact same provider identity
|
||||
// to ensure generation, not changed workdir ID/content, causes rejection.
|
||||
// Detach may close the local session; no subsequent filesystem operation is
|
||||
// made on it. This test only exercises the router's connection identity.
|
||||
router.detach(&alias).await.unwrap();
|
||||
assert_validator_mismatch(handler.check_connection());
|
||||
router.attach(alias, session).unwrap();
|
||||
let replacement = router.resolve(Some("main")).unwrap();
|
||||
assert_ne!(replacement.generation, handler.generation);
|
||||
assert_eq!(replacement.session.workdir().id().as_str(), handler.workdir);
|
||||
assert_validator_mismatch(handler.check_connection());
|
||||
|
||||
// Positive control: the new generation is accepted with the same identity.
|
||||
handler.generation = replacement.generation;
|
||||
assert!(handler.check_connection().is_ok());
|
||||
}
|
||||
|
||||
fn assert_deadline<T>(result: Result<T, WipOperationError>, message: &str) {
|
||||
match result {
|
||||
Err(WipOperationError::Protocol(e)) => {
|
||||
assert_eq!(e.code, ProtocolErrorCode::ResourceLimitExceeded);
|
||||
assert_eq!(e.message, message);
|
||||
}
|
||||
Err(_) => panic!("expected a protocol deadline error"),
|
||||
Ok(_) => panic!("a stalled provider observation must time out"),
|
||||
}
|
||||
}
|
||||
|
||||
fn assert_permit_released(p: &Provider) {
|
||||
assert_eq!(p.permits.available_permits(), 1);
|
||||
let permit = p
|
||||
.permits
|
||||
.try_acquire()
|
||||
.expect("the timed-out operation must release its permit");
|
||||
drop(permit);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn collection_list_and_root_enumeration_bound_stalled_observation_and_release_permit() {
|
||||
let dir = TempDir::new().unwrap();
|
||||
let session = HookedSession::new(&dir);
|
||||
let router = Arc::new(WorkdirSessionRouter::new());
|
||||
router
|
||||
.attach(
|
||||
WorkdirAttachmentAlias::new("main").unwrap(),
|
||||
session.clone(),
|
||||
)
|
||||
.unwrap();
|
||||
session.stall_observation(Some(WorkdirPath::new("").unwrap()));
|
||||
let p = deadline_provider(router);
|
||||
let collection = Collection {
|
||||
provider: p.clone(),
|
||||
};
|
||||
|
||||
// The outer watchdog prevents a missing production timeout from hanging the
|
||||
// suite. The exact protocol message proves the inner enumeration timed out.
|
||||
let result = tokio::time::timeout(
|
||||
Duration::from_secs(2),
|
||||
collection.call("list", &BTreeMap::new(), context()),
|
||||
)
|
||||
.await
|
||||
.expect("Collection.list must be bounded by the provider deadline");
|
||||
assert_deadline(result, "checkout enumeration deadline");
|
||||
assert_eq!(session.stalled_observations.load(Ordering::SeqCst), 1);
|
||||
assert_permit_released(&p);
|
||||
|
||||
// Exercise the subtree entry point independently with the stall still armed.
|
||||
let subtree = Arc::new(p.clone());
|
||||
let result = tokio::time::timeout(Duration::from_secs(2), subtree.children(ROOT))
|
||||
.await
|
||||
.expect("root enumeration must be bounded by the provider deadline")
|
||||
.map_err(WipOperationError::Protocol);
|
||||
assert_deadline(result, "checkout enumeration deadline");
|
||||
assert_eq!(session.stalled_observations.load(Ordering::SeqCst), 2);
|
||||
assert_permit_released(&p);
|
||||
|
||||
// Removing the stall makes the same provider usable immediately, without
|
||||
// resetting its semaphore or waiting for any background task to finish.
|
||||
session.stall_observation(None);
|
||||
let paths = tokio::time::timeout(Duration::from_secs(2), subtree.children(ROOT))
|
||||
.await
|
||||
.expect("enumeration must remain usable after cancellation")
|
||||
.unwrap();
|
||||
assert_eq!(paths, vec![checkout_root("main")]);
|
||||
assert_permit_released(&p);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn grep_post_search_link_observation_is_deadline_bounded_and_releases_permit() {
|
||||
let dir = TempDir::new().unwrap();
|
||||
std::fs::write(dir.path().join("a.txt"), "needle\n").unwrap();
|
||||
let session = HookedSession::new(&dir);
|
||||
let router = Arc::new(WorkdirSessionRouter::new());
|
||||
router
|
||||
.attach(
|
||||
WorkdirAttachmentAlias::new("main").unwrap(),
|
||||
session.clone(),
|
||||
)
|
||||
.unwrap();
|
||||
let p = deadline_provider(router);
|
||||
let root = checkout_root("main");
|
||||
let projection = p.node(&root).await.unwrap().unwrap();
|
||||
|
||||
// Directory observations and the normal local grep/checkout execution can
|
||||
// complete. Only publishing the typed search result path gets stuck.
|
||||
session.stall_observation(Some(WorkdirPath::new("a.txt").unwrap()));
|
||||
let arguments = BTreeMap::from([
|
||||
("pattern".into(), Value::String("needle".into())),
|
||||
(
|
||||
"output_mode".into(),
|
||||
Value::String("files_with_matches".into()),
|
||||
),
|
||||
]);
|
||||
let result = tokio::time::timeout(
|
||||
Duration::from_secs(2),
|
||||
projection.handler.call("grep", &arguments, context()),
|
||||
)
|
||||
.await
|
||||
.expect("post-search link observations must share the operation deadline");
|
||||
assert_deadline(result, "checkout result deadline");
|
||||
assert_eq!(session.stalled_observations.load(Ordering::SeqCst), 1);
|
||||
assert_permit_released(&p);
|
||||
|
||||
// A successful retry both checks cancellation cleanup and proves that the
|
||||
// delegated search produces the typed path consumed by post-processing.
|
||||
session.stall_observation(None);
|
||||
let result = tokio::time::timeout(
|
||||
Duration::from_secs(2),
|
||||
projection.handler.call("grep", &arguments, context()),
|
||||
)
|
||||
.await
|
||||
.expect("grep must remain usable after cancelling result publication");
|
||||
let output = match result {
|
||||
Ok(output) => output,
|
||||
Err(_) => panic!("grep must succeed once the link observation is unstalled"),
|
||||
};
|
||||
let value = wip_to_json(&output.value).unwrap();
|
||||
assert_eq!(value["items"], json!([{"path": format!("{root}/a.txt")}]));
|
||||
assert_permit_released(&p);
|
||||
}
|
||||
@@ -0,0 +1,580 @@
|
||||
//! Exercises native checkout interactions through the real WIP Client/HTTP codec.
|
||||
use super::*;
|
||||
use tempfile::TempDir;
|
||||
use workdir::{
|
||||
LocalWorkdirSession, WorkdirAttachmentAlias, WorkdirSessionCapabilities, WorkdirSessionHandle,
|
||||
WorkdirSessionRouter,
|
||||
};
|
||||
|
||||
fn session(dir: &TempDir, read_only: bool) -> WorkdirSessionHandle {
|
||||
let caps = if read_only {
|
||||
WorkdirSessionCapabilities::READ_ONLY
|
||||
} else {
|
||||
WorkdirSessionCapabilities::READ_WRITE
|
||||
};
|
||||
Arc::new(LocalWorkdirSession::materialized(
|
||||
dir.path().to_path_buf(),
|
||||
dir.path().to_path_buf(),
|
||||
manifest::SharedScope::new(manifest::Scope::writable(dir.path()).unwrap()),
|
||||
caps,
|
||||
))
|
||||
}
|
||||
fn runtime(
|
||||
router: Arc<WorkdirSessionRouter>,
|
||||
permissions: Option<ToolPermissionConfig>,
|
||||
) -> WipRuntime {
|
||||
let mut registry = WipMountRegistry::new();
|
||||
crate::checkout::mount_checkouts(&mut registry, router, tools::Tracker::new(), permissions)
|
||||
.unwrap();
|
||||
WipRuntime::new(
|
||||
WipHost::new(registry),
|
||||
SecurityContext::new("checkout-test-worker"),
|
||||
0,
|
||||
)
|
||||
.unwrap()
|
||||
}
|
||||
async fn interface(r: &WipRuntime, path: &str) -> String {
|
||||
r.discover(path.into(), 0, true).await.unwrap();
|
||||
let p = r.host.projection_live(path).await.unwrap().unwrap();
|
||||
r.inspect(p.interface.clone(), true).await.unwrap();
|
||||
p.interface
|
||||
}
|
||||
async fn call(
|
||||
r: &WipRuntime,
|
||||
path: &str,
|
||||
operation: &str,
|
||||
args: Json,
|
||||
) -> Result<ToolOutput, ToolError> {
|
||||
let i = interface(r, path).await;
|
||||
r.call(path.into(), i, operation.into(), args, Default::default())
|
||||
.await
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn checkout_wip_native_roundtrip_search_read_edit_write_create() {
|
||||
let dir = TempDir::new().unwrap();
|
||||
std::fs::create_dir_all(dir.path().join("src/deep")).unwrap();
|
||||
std::fs::write(
|
||||
dir.path().join("src/deep/a.txt"),
|
||||
"first\nneedle needle\nlast\n",
|
||||
)
|
||||
.unwrap();
|
||||
let router = Arc::new(WorkdirSessionRouter::new());
|
||||
router
|
||||
.attach(
|
||||
WorkdirAttachmentAlias::new("main").unwrap(),
|
||||
session(&dir, false),
|
||||
)
|
||||
.unwrap();
|
||||
let r = runtime(router, None);
|
||||
let root = r.host.observe_live("/", 2).await.unwrap();
|
||||
assert_eq!(root.children.unwrap()[0].object.name, "checkouts");
|
||||
let file = "/checkouts/main/src/deep/a.txt";
|
||||
let glob = call(&r, "/checkouts/main", "glob", json!({"pattern":"**/*.txt"}))
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(glob.content.unwrap().contains(file));
|
||||
let grep = call(
|
||||
&r,
|
||||
"/checkouts/main/src",
|
||||
"grep",
|
||||
json!({"pattern":"needle","output_mode":"files_with_matches"}),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let grep_result: Json = serde_json::from_str(grep.content.as_deref().unwrap()).unwrap();
|
||||
assert_eq!(grep_result["items"][0]["path"], file);
|
||||
let followed = grep_result["items"][0]["path"].as_str().unwrap();
|
||||
assert_eq!(followed, file);
|
||||
assert!(
|
||||
call(
|
||||
&r,
|
||||
file,
|
||||
"edit",
|
||||
json!({"old_string":"needle","new_string":"done"})
|
||||
)
|
||||
.await
|
||||
.is_err()
|
||||
);
|
||||
let read = call(&r, file, "read", json!({"offset":1,"limit":1}))
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(read.content.unwrap().contains("needle needle"));
|
||||
assert!(
|
||||
call(
|
||||
&r,
|
||||
file,
|
||||
"edit",
|
||||
json!({"old_string":"needle","new_string":"done"})
|
||||
)
|
||||
.await
|
||||
.is_err()
|
||||
);
|
||||
call(
|
||||
&r,
|
||||
file,
|
||||
"edit",
|
||||
json!({"old_string":"needle","new_string":"done","replace_all":true}),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
std::fs::read_to_string(dir.path().join("src/deep/a.txt")).unwrap(),
|
||||
"first\ndone done\nlast\n"
|
||||
);
|
||||
call(&r, file, "write", json!({"content":"saved\n"}))
|
||||
.await
|
||||
.unwrap();
|
||||
let reread = call(&r, file, "read", json!({})).await.unwrap();
|
||||
assert!(reread.content.unwrap().contains("saved"));
|
||||
call(
|
||||
&r,
|
||||
"/checkouts/main/src",
|
||||
"create_file",
|
||||
json!({"path":"new/deeper.txt","content":"created\n"}),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
std::fs::read_to_string(dir.path().join("src/new/deeper.txt")).unwrap(),
|
||||
"created\n"
|
||||
);
|
||||
assert!(
|
||||
call(
|
||||
&r,
|
||||
"/checkouts/main/src",
|
||||
"create_file",
|
||||
json!({"path":"new/deeper.txt","content":"overwrite"})
|
||||
)
|
||||
.await
|
||||
.is_err()
|
||||
);
|
||||
call(&r, "/checkouts/main/src/new/deeper.txt", "read", json!({}))
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn checkout_wip_multiple_aliases_read_only_and_escape_rejection() {
|
||||
let a = TempDir::new().unwrap();
|
||||
let b = TempDir::new().unwrap();
|
||||
std::fs::write(a.path().join("a.txt"), "A").unwrap();
|
||||
std::fs::write(b.path().join("b.txt"), "B").unwrap();
|
||||
let router = Arc::new(WorkdirSessionRouter::new());
|
||||
router
|
||||
.attach(
|
||||
WorkdirAttachmentAlias::new("a").unwrap(),
|
||||
session(&a, false),
|
||||
)
|
||||
.unwrap();
|
||||
router
|
||||
.attach(WorkdirAttachmentAlias::new("b").unwrap(), session(&b, true))
|
||||
.unwrap();
|
||||
let r = runtime(router, None);
|
||||
let observed = r.host.observe_live("/checkouts", 2).await.unwrap();
|
||||
assert_eq!(observed.children.unwrap().len(), 2);
|
||||
let bfile = r
|
||||
.host
|
||||
.projection_live("/checkouts/b/b.txt")
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
bfile
|
||||
.descriptor
|
||||
.operations
|
||||
.iter()
|
||||
.map(|o| o.name.as_str())
|
||||
.collect::<Vec<_>>(),
|
||||
vec!["read"]
|
||||
);
|
||||
assert!(
|
||||
r.host
|
||||
.projection_live("/checkouts/b/a.txt")
|
||||
.await
|
||||
.unwrap()
|
||||
.is_none()
|
||||
);
|
||||
assert!(
|
||||
r.host
|
||||
.projection_live("/checkouts/a/../b/b.txt")
|
||||
.await
|
||||
.is_err()
|
||||
);
|
||||
assert!(
|
||||
call(
|
||||
&r,
|
||||
"/checkouts/a",
|
||||
"glob",
|
||||
json!({"pattern":"*","path":"../b"})
|
||||
)
|
||||
.await
|
||||
.is_err()
|
||||
);
|
||||
assert!(
|
||||
call(
|
||||
&r,
|
||||
"/checkouts/a",
|
||||
"create_file",
|
||||
json!({"path":"/outside.txt","content":"bad"})
|
||||
)
|
||||
.await
|
||||
.is_err()
|
||||
);
|
||||
assert!(
|
||||
call(
|
||||
&r,
|
||||
"/checkouts/a/a.txt",
|
||||
"read",
|
||||
json!({"target_workdir":"b"})
|
||||
)
|
||||
.await
|
||||
.is_err()
|
||||
);
|
||||
assert!(
|
||||
call(
|
||||
&r,
|
||||
"/checkouts/a/a.txt",
|
||||
"read",
|
||||
json!({"file_path":"b.txt"})
|
||||
)
|
||||
.await
|
||||
.is_err()
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn checkout_wip_detach_alias_reuse_and_restore_fence_old_references() {
|
||||
let a = TempDir::new().unwrap();
|
||||
let b = TempDir::new().unwrap();
|
||||
std::fs::write(a.path().join("file.txt"), "same").unwrap();
|
||||
std::fs::write(b.path().join("file.txt"), "same").unwrap();
|
||||
let router = Arc::new(WorkdirSessionRouter::new());
|
||||
let alias = WorkdirAttachmentAlias::new("main").unwrap();
|
||||
router.attach(alias.clone(), session(&a, false)).unwrap();
|
||||
let r = runtime(router.clone(), None);
|
||||
let path = "/checkouts/main/file.txt";
|
||||
call(&r, path, "read", json!({})).await.unwrap();
|
||||
let old = r.host.projection_live(path).await.unwrap().unwrap();
|
||||
router.detach(&alias).await.unwrap();
|
||||
assert!(r.host.projection_live(path).await.unwrap().is_none());
|
||||
assert!(r.host.fetch_interface_live(&old.interface).await.is_err());
|
||||
router.attach(alias, session(&b, false)).unwrap();
|
||||
let new = r.host.projection_live(path).await.unwrap().unwrap();
|
||||
assert_ne!(old.interface, new.interface);
|
||||
assert_ne!(old.object.validator, new.object.validator);
|
||||
assert!(r.host.fetch_interface_live(&old.interface).await.is_err());
|
||||
assert!(
|
||||
call(
|
||||
&r,
|
||||
path,
|
||||
"write",
|
||||
json!({"content":"must read new attachment"})
|
||||
)
|
||||
.await
|
||||
.is_err()
|
||||
);
|
||||
let restored = runtime(router, None);
|
||||
let next = restored.host.projection_live(path).await.unwrap().unwrap();
|
||||
assert_ne!(new.interface, next.interface);
|
||||
assert_ne!(new.object.validator, next.object.validator);
|
||||
assert!(
|
||||
restored
|
||||
.host
|
||||
.fetch_interface_live(&new.interface)
|
||||
.await
|
||||
.is_err()
|
||||
);
|
||||
assert_eq!(
|
||||
std::fs::read_to_string(b.path().join("file.txt")).unwrap(),
|
||||
"same"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn checkout_wip_provider_validator_rejects_same_content_inode_substitution() {
|
||||
let dir = TempDir::new().unwrap();
|
||||
std::fs::write(dir.path().join("file.txt"), "same").unwrap();
|
||||
let router = Arc::new(WorkdirSessionRouter::new());
|
||||
router
|
||||
.attach(
|
||||
WorkdirAttachmentAlias::new("main").unwrap(),
|
||||
session(&dir, false),
|
||||
)
|
||||
.unwrap();
|
||||
let r = runtime(router, None);
|
||||
let path = "/checkouts/main/file.txt";
|
||||
call(&r, path, "read", json!({})).await.unwrap();
|
||||
let p = r.host.projection_live(path).await.unwrap().unwrap();
|
||||
std::fs::write(dir.path().join("replacement"), "same").unwrap();
|
||||
std::fs::rename(dir.path().join("replacement"), dir.path().join("file.txt")).unwrap();
|
||||
// Call the captured handler directly: provider must reject even if Host validation was earlier.
|
||||
let out = p
|
||||
.handler
|
||||
.call(
|
||||
"write",
|
||||
&BTreeMap::from([("content".into(), Value::String("bad".into()))]),
|
||||
WipCallContext {
|
||||
execution: Default::default(),
|
||||
security_context: "checkout-test-worker".into(),
|
||||
},
|
||||
)
|
||||
.await;
|
||||
assert!(matches!(
|
||||
out,
|
||||
Err(WipOperationError::Protocol(ProtocolError {
|
||||
code: ProtocolErrorCode::ValidatorMismatch,
|
||||
..
|
||||
}))
|
||||
));
|
||||
assert_eq!(
|
||||
std::fs::read_to_string(dir.path().join("file.txt")).unwrap(),
|
||||
"same"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn checkout_wip_policy_is_not_feature_authority_and_compatibility_is_replaced() {
|
||||
let dir = TempDir::new().unwrap();
|
||||
std::fs::write(dir.path().join("file.txt"), "secret").unwrap();
|
||||
let router = Arc::new(WorkdirSessionRouter::new());
|
||||
router
|
||||
.attach(
|
||||
WorkdirAttachmentAlias::new("main").unwrap(),
|
||||
session(&dir, false),
|
||||
)
|
||||
.unwrap();
|
||||
let permissions = ToolPermissionConfig {
|
||||
default_action: ToolPermissionAction::Deny,
|
||||
rules: Vec::new(),
|
||||
};
|
||||
let r = runtime(router.clone(), Some(permissions));
|
||||
assert!(
|
||||
r.host
|
||||
.projection_live("/checkouts/main/file.txt")
|
||||
.await
|
||||
.unwrap()
|
||||
.is_none()
|
||||
);
|
||||
assert!(
|
||||
r.host
|
||||
.observe_live("/checkouts", 1)
|
||||
.await
|
||||
.unwrap()
|
||||
.children
|
||||
.unwrap()
|
||||
.is_empty()
|
||||
);
|
||||
let mut registry = WipMountRegistry::new();
|
||||
crate::checkout::mount_checkouts(&mut registry, router.clone(), tools::Tracker::new(), None)
|
||||
.unwrap();
|
||||
for def in tools::routed_builtin_tools(router, tools::Tracker::new(), dir.path().join("bash")) {
|
||||
let (meta, tool) = def();
|
||||
let p = compatibility_projection(meta, tool, None).unwrap();
|
||||
registry.mount(p).unwrap();
|
||||
}
|
||||
for tool in ["Read", "Edit", "Write", "Glob", "Grep"] {
|
||||
assert!(!registry.routes().any(|p| p == format!("/tools/{tool}")));
|
||||
}
|
||||
assert!(registry.routes().any(|p| p == "/tools/Bash"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn checkout_wip_bounded_observation_and_deep_subtree_collision() {
|
||||
let dir = TempDir::new().unwrap();
|
||||
let router = Arc::new(WorkdirSessionRouter::new());
|
||||
router
|
||||
.attach(
|
||||
WorkdirAttachmentAlias::new("main").unwrap(),
|
||||
session(&dir, false),
|
||||
)
|
||||
.unwrap();
|
||||
let mut registry = WipMountRegistry::new();
|
||||
crate::checkout::mount_checkouts(&mut registry, router, tools::Tracker::new(), None).unwrap();
|
||||
let placeholder = registry.mounts["/checkouts"].resolved();
|
||||
let mut collision = placeholder.clone();
|
||||
collision.route = "/checkouts/main/a".into();
|
||||
collision.object.name = "a".into();
|
||||
assert!(matches!(
|
||||
registry.mount(collision),
|
||||
Err(WipMountError::RouteCollision { .. })
|
||||
));
|
||||
let host = WipHost::new(registry);
|
||||
assert_eq!(
|
||||
host.observe_live("/checkouts", 33).await.unwrap_err().code,
|
||||
ProtocolErrorCode::ResourceLimitExceeded
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn checkout_wip_post_operation_validators_support_cached_chains() {
|
||||
let dir = TempDir::new().unwrap();
|
||||
std::fs::write(dir.path().join("file.txt"), "old\n").unwrap();
|
||||
let router = Arc::new(WorkdirSessionRouter::new());
|
||||
router
|
||||
.attach(
|
||||
WorkdirAttachmentAlias::new("main").unwrap(),
|
||||
session(&dir, false),
|
||||
)
|
||||
.unwrap();
|
||||
let r = runtime(router, None);
|
||||
let path = "/checkouts/main/file.txt";
|
||||
let i = interface(&r, path).await;
|
||||
// No rediscovery or Interface refresh between any of these operations.
|
||||
for (op, args) in [
|
||||
("read", json!({})),
|
||||
("edit", json!({"old_string":"old","new_string":"edited"})),
|
||||
("write", json!({"content":"saved\n"})),
|
||||
("read", json!({})),
|
||||
] {
|
||||
r.call(path.into(), i.clone(), op.into(), args, Default::default())
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
assert_eq!(
|
||||
std::fs::read_to_string(dir.path().join("file.txt")).unwrap(),
|
||||
"saved\n"
|
||||
);
|
||||
let parent = "/checkouts/main";
|
||||
let i = interface(&r, parent).await;
|
||||
for name in ["one.txt", "nested/two.txt"] {
|
||||
r.call(
|
||||
parent.into(),
|
||||
i.clone(),
|
||||
"create_file".into(),
|
||||
json!({"path":name,"content":"new"}),
|
||||
Default::default(),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
assert_eq!(
|
||||
std::fs::read_to_string(dir.path().join("nested/two.txt")).unwrap(),
|
||||
"new"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn checkout_wip_scoped_child_search_and_revoke_cannot_expose_siblings() {
|
||||
use workdir::{
|
||||
WorkdirPath, WorkdirToolBroker, WorkdirToolScope, WorkdirToolScopePermission as Permission,
|
||||
WorkdirToolScopeRule,
|
||||
};
|
||||
let dir = TempDir::new().unwrap();
|
||||
std::fs::create_dir(dir.path().join("child")).unwrap();
|
||||
std::fs::write(dir.path().join("child/a.txt"), "visible needle").unwrap();
|
||||
std::fs::write(dir.path().join("private-sibling-name.txt"), "hidden needle").unwrap();
|
||||
let broker = Arc::new(WorkdirToolBroker::new(session(&dir, false)));
|
||||
let lease = broker
|
||||
.scope(WorkdirToolScope {
|
||||
rules: vec![WorkdirToolScopeRule {
|
||||
target: WorkdirPath::new("child").unwrap(),
|
||||
permission: Permission::Write,
|
||||
recursive: true,
|
||||
symlink_policy: Default::default(),
|
||||
}],
|
||||
cwd: WorkdirPath::new("child").unwrap(),
|
||||
command: false,
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
let parent_router = Arc::new(WorkdirSessionRouter::new());
|
||||
parent_router
|
||||
.attach(
|
||||
WorkdirAttachmentAlias::new("parent").unwrap(),
|
||||
broker.tool_session(),
|
||||
)
|
||||
.unwrap();
|
||||
let parent = runtime(parent_router, None);
|
||||
let parent_file = parent
|
||||
.host
|
||||
.projection_live("/checkouts/parent/child/a.txt")
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
assert!(
|
||||
!parent_file
|
||||
.descriptor
|
||||
.operations
|
||||
.iter()
|
||||
.any(|op| op.name == "write" || op.name == "edit")
|
||||
);
|
||||
let child_router = Arc::new(WorkdirSessionRouter::new());
|
||||
child_router
|
||||
.attach(
|
||||
WorkdirAttachmentAlias::new("child").unwrap(),
|
||||
lease.tool_session(),
|
||||
)
|
||||
.unwrap();
|
||||
let child = runtime(child_router, None);
|
||||
let observed = child
|
||||
.host
|
||||
.observe_live("/checkouts/child", 2)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(observed.children.unwrap().len(), 1);
|
||||
for (op, args) in [
|
||||
("glob", json!({"pattern":"**/*"})),
|
||||
(
|
||||
"grep",
|
||||
json!({"pattern":"needle","output_mode":"files_with_matches"}),
|
||||
),
|
||||
] {
|
||||
let output = call(&child, "/checkouts/child", op, args)
|
||||
.await
|
||||
.unwrap()
|
||||
.content
|
||||
.unwrap();
|
||||
assert!(output.contains("/checkouts/child/a.txt"));
|
||||
assert!(!output.contains("private-sibling-name"));
|
||||
assert!(!output.contains("hidden needle"));
|
||||
}
|
||||
assert!(
|
||||
child
|
||||
.host
|
||||
.projection_live("/checkouts/child/private-sibling-name.txt")
|
||||
.await
|
||||
.unwrap()
|
||||
.is_none()
|
||||
);
|
||||
let file = "/checkouts/child/a.txt";
|
||||
let i = interface(&child, file).await;
|
||||
child
|
||||
.call(
|
||||
file.into(),
|
||||
i.clone(),
|
||||
"read".into(),
|
||||
json!({}),
|
||||
Default::default(),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
child
|
||||
.call(
|
||||
file.into(),
|
||||
i.clone(),
|
||||
"write".into(),
|
||||
json!({"content":"changed"}),
|
||||
Default::default(),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
lease.close().await.unwrap();
|
||||
assert!(child.host.projection_live(file).await.unwrap().is_none());
|
||||
assert!(
|
||||
child
|
||||
.call(
|
||||
file.into(),
|
||||
i,
|
||||
"write".into(),
|
||||
json!({"content":"after revoke"}),
|
||||
Default::default()
|
||||
)
|
||||
.await
|
||||
.is_err()
|
||||
);
|
||||
assert_eq!(
|
||||
std::fs::read_to_string(dir.path().join("child/a.txt")).unwrap(),
|
||||
"changed"
|
||||
);
|
||||
}
|
||||
@@ -1597,6 +1597,20 @@ 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 && !backend_job_profile {
|
||||
crate::checkout::mount_checkouts(
|
||||
&mut wip_mount_registry,
|
||||
worker.workdir_sessions(),
|
||||
tracker.clone(),
|
||||
wip_permissions.clone(),
|
||||
)
|
||||
.map_err(|error| {
|
||||
std::io::Error::new(
|
||||
std::io::ErrorKind::InvalidInput,
|
||||
format!("mount checkout WIP projection: {error}"),
|
||||
)
|
||||
})?;
|
||||
}
|
||||
if wip_mode && let Some(feature) = &workdir_wip_feature {
|
||||
crate::feature::builtin::manage_workdir::wip::mount_workspace_workdir_wip(
|
||||
&mut wip_mount_registry,
|
||||
|
||||
@@ -353,6 +353,102 @@ impl WorkdirSession for WorkspaceAttachedWorkdirSession {
|
||||
}
|
||||
}
|
||||
|
||||
async fn checkout_observe(
|
||||
&self,
|
||||
path: workdir::WorkdirPath,
|
||||
) -> Result<workdir::CheckoutObservation, WorkdirError> {
|
||||
match self.operate(WorkdirSessionOperation::CheckoutObserve(path.clone()))? {
|
||||
WorkdirSessionOperationResult::CheckoutObserve(mut result)
|
||||
if result.path == path
|
||||
&& !result.validator.is_empty()
|
||||
&& result.validator.len() <= 256 =>
|
||||
{
|
||||
result.capabilities = result.capabilities.intersection(self.capabilities);
|
||||
Ok(result)
|
||||
}
|
||||
_ => Err(Self::mismatch("checkout_observe")),
|
||||
}
|
||||
}
|
||||
|
||||
async fn checkout_execute(
|
||||
&self,
|
||||
request: workdir::CheckoutRequest,
|
||||
) -> Result<workdir::CheckoutResult, WorkdirError> {
|
||||
let cap = request.operation.capability();
|
||||
if !self.capabilities.supports(cap) {
|
||||
return Err(WorkdirError::Unsupported(cap));
|
||||
}
|
||||
let target = request.target.clone();
|
||||
let mutation = cap != workdir::WorkdirSessionCapability::Read;
|
||||
match self.operate(WorkdirSessionOperation::CheckoutExecute(request)) {
|
||||
Ok(WorkdirSessionOperationResult::CheckoutExecute(mut result))
|
||||
if result.observation.path == target
|
||||
&& !result.observation.validator.is_empty()
|
||||
&& result.observation.validator.len() <= 256
|
||||
&& matches!(
|
||||
(&result.output, cap),
|
||||
(
|
||||
workdir::CheckoutOutput::Read(_),
|
||||
workdir::WorkdirSessionCapability::Read
|
||||
) | (
|
||||
workdir::CheckoutOutput::Write(_),
|
||||
workdir::WorkdirSessionCapability::Write
|
||||
) | (
|
||||
workdir::CheckoutOutput::Edit(_),
|
||||
workdir::WorkdirSessionCapability::Edit
|
||||
)
|
||||
) =>
|
||||
{
|
||||
result.observation.capabilities = result
|
||||
.observation
|
||||
.capabilities
|
||||
.intersection(self.capabilities);
|
||||
Ok(result)
|
||||
}
|
||||
Ok(_) if mutation => Err(WorkdirError::OutcomeUnknown(
|
||||
"mismatched checkout mutation response".into(),
|
||||
)),
|
||||
Ok(_) => Err(Self::mismatch("checkout_execute")),
|
||||
Err(WorkdirError::Transport(_) | WorkdirError::Unavailable(_)) if mutation => {
|
||||
Err(WorkdirError::OutcomeUnknown(
|
||||
"checkout mutation response lost; inspect effects before retry".into(),
|
||||
))
|
||||
}
|
||||
Err(error) => Err(error),
|
||||
}
|
||||
}
|
||||
|
||||
async fn checkout_search(
|
||||
&self,
|
||||
request: workdir::CheckoutSearchRequest,
|
||||
) -> Result<workdir::CheckoutSearchResult, WorkdirError> {
|
||||
use workdir::{
|
||||
CheckoutSearchOperation as Op, CheckoutSearchResult as SearchResult,
|
||||
WorkdirSessionCapability as Cap,
|
||||
};
|
||||
let cap = match &request.operation {
|
||||
Op::List(_) => Cap::Read,
|
||||
Op::Glob(_) => Cap::Glob,
|
||||
Op::Grep(_) => Cap::Grep,
|
||||
};
|
||||
if !self.capabilities.supports(cap) {
|
||||
return Err(WorkdirError::Unsupported(cap));
|
||||
}
|
||||
match self.operate(WorkdirSessionOperation::CheckoutSearch(request))? {
|
||||
WorkdirSessionOperationResult::CheckoutSearch(result)
|
||||
if matches!(
|
||||
(&result, cap),
|
||||
(SearchResult::List(_), Cap::Read)
|
||||
| (SearchResult::Glob(_), Cap::Glob)
|
||||
| (SearchResult::Grep(_), Cap::Grep)
|
||||
) =>
|
||||
{
|
||||
Ok(result)
|
||||
}
|
||||
_ => Err(Self::mismatch("checkout_search")),
|
||||
}
|
||||
}
|
||||
|
||||
async fn stat(&self, request: StatRequest) -> Result<StatResult, WorkdirError> {
|
||||
match self.operate(WorkdirSessionOperation::Stat(request))? {
|
||||
WorkdirSessionOperationResult::Stat(result) => Ok(result),
|
||||
@@ -1414,6 +1510,56 @@ mod tests {
|
||||
assert!(body.get("session_id").is_none());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn attached_checkout_mutations_preserve_known_rejection_and_lost_outcomes() {
|
||||
let client = Arc::new(RecordingWorkspaceClient::new(vec![
|
||||
response(json!({"operation":"write","result":{"bytes_written":3,"created":false}})),
|
||||
WorkspaceResponse {
|
||||
status: 200,
|
||||
body: "invalid payload".into(),
|
||||
},
|
||||
error_response(
|
||||
409,
|
||||
workdir::http::WorkdirTransportErrorCode::Conflict,
|
||||
"stale",
|
||||
),
|
||||
]));
|
||||
let session = WorkspaceAttachedWorkdirSession::handle(client.clone(), "checkout");
|
||||
let request = workdir::CheckoutRequest {
|
||||
target: workdir::WorkdirPath::new("file.txt").unwrap(),
|
||||
validator: vec![1],
|
||||
operation: workdir::CheckoutOperation::Write {
|
||||
content: b"new".to_vec(),
|
||||
expected_hash: [0; 32],
|
||||
},
|
||||
};
|
||||
for _ in 0..2 {
|
||||
assert!(matches!(
|
||||
session.checkout_execute(request.clone()).await,
|
||||
Err(WorkdirError::OutcomeUnknown(_))
|
||||
));
|
||||
}
|
||||
assert!(matches!(
|
||||
session.checkout_execute(request.clone()).await,
|
||||
Err(WorkdirError::Conflict(_))
|
||||
));
|
||||
// Missing response simulates the Workspace client losing the provider result.
|
||||
assert!(matches!(
|
||||
session.checkout_execute(request).await,
|
||||
Err(WorkdirError::OutcomeUnknown(_))
|
||||
));
|
||||
let calls = client.requests();
|
||||
assert_eq!(calls.len(), 4);
|
||||
for call in calls {
|
||||
let body: serde_json::Value =
|
||||
serde_json::from_str(call.body.as_deref().unwrap()).unwrap();
|
||||
assert_eq!(body["target_workdir"], "checkout");
|
||||
assert_eq!(body["operation"]["operation"], "checkout_execute");
|
||||
assert_eq!(body["operation"]["request"]["target"], "file.txt");
|
||||
assert!(body.get("runtime_id").is_none());
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn attached_session_does_not_send_worker_local_bash_spill_path() {
|
||||
let client = Arc::new(RecordingWorkspaceClient::new(vec![response(json!({
|
||||
|
||||
@@ -344,7 +344,21 @@ impl Provider {
|
||||
if let Some(id) = raw.get("working_directory_id").and_then(Json::as_str) {
|
||||
result.insert("workdir_path".into(), json!(item_path(Domain::Workdir, id)));
|
||||
}
|
||||
// Content projection belongs to T-699/T-695. No unconditional FS entrance.
|
||||
if let Some(alias) = raw.get("alias").and_then(Json::as_str)
|
||||
&& let Ok(selected) = self.backend.session_router.resolve(Some(alias))
|
||||
&& raw.get("working_directory_id").and_then(Json::as_str)
|
||||
== Some(selected.session.workdir().id().as_str())
|
||||
&& selected
|
||||
.session
|
||||
.capabilities()
|
||||
.supports(workdir::WorkdirSessionCapability::Read)
|
||||
{
|
||||
result.insert(
|
||||
"checkout_path".into(),
|
||||
json!(crate::checkout::checkout_root(alias)),
|
||||
);
|
||||
result.insert("checkout_slug".into(), json!(encode_identity(alias)));
|
||||
}
|
||||
}
|
||||
result.insert("path".into(), json!(item_path(domain, id)));
|
||||
Ok(Json::Object(result))
|
||||
@@ -735,6 +749,17 @@ impl WipOperationHandler for ItemHandler {
|
||||
let mut result = raw;
|
||||
result["workdir_path"] = json!(item_path(Domain::Workdir, &self.id));
|
||||
result["attachments_path"] = json!(ATTACHMENTS);
|
||||
if let Some(alias) = input.get("alias").and_then(Json::as_str)
|
||||
&& self
|
||||
.provider
|
||||
.backend
|
||||
.session_router
|
||||
.resolve(Some(alias))
|
||||
.is_ok()
|
||||
{
|
||||
result["checkout_path"] = json!(crate::checkout::checkout_root(alias));
|
||||
result["checkout_slug"] = json!(encode_identity(alias));
|
||||
}
|
||||
// Backend attachment list supplies the durable lifetime identity.
|
||||
result
|
||||
} else {
|
||||
@@ -866,7 +891,7 @@ fn page_limit(input: &Json) -> Result<usize, WipOperationError> {
|
||||
// Readable ASCII identities remain unchanged. Every other UTF-8 identity uses a
|
||||
// canonical ~hex segment; the escape marker is itself always escaped. This is
|
||||
// injective, does not exclude existing keys, and never decodes a second route.
|
||||
fn encode_identity(id: &str) -> String {
|
||||
pub(crate) fn encode_identity(id: &str) -> String {
|
||||
if !id.is_empty()
|
||||
&& id
|
||||
.bytes()
|
||||
@@ -883,7 +908,7 @@ fn encode_identity(id: &str) -> String {
|
||||
)
|
||||
}
|
||||
}
|
||||
fn decode_identity(segment: &str) -> Option<String> {
|
||||
pub(crate) fn decode_identity(segment: &str) -> Option<String> {
|
||||
let id = if let Some(hex) = segment.strip_prefix('~') {
|
||||
if hex.len() % 2 != 0 || hex.is_empty() || !hex.bytes().all(|b| b.is_ascii_hexdigit()) {
|
||||
return None;
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
pub mod bootstrap;
|
||||
pub mod checkout;
|
||||
pub mod compact;
|
||||
pub mod controller;
|
||||
pub mod discovery;
|
||||
|
||||
+369
-14
@@ -42,6 +42,13 @@ use wip_protocol::{
|
||||
|
||||
use crate::permission::permission_action_for;
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "checkout_http_tests.rs"]
|
||||
mod checkout_http_tests;
|
||||
#[cfg(test)]
|
||||
#[path = "checkout_wip_tests.rs"]
|
||||
mod checkout_wip_tests;
|
||||
|
||||
const WIP_ENDPOINT: &str = "https://worker.wip.invalid/v1/";
|
||||
const WIP_ROOT: &str = "/";
|
||||
const WIP_TOOLS_ROOT: &str = "/tools";
|
||||
@@ -92,6 +99,8 @@ pub struct WipCallContext {
|
||||
pub struct WipOperationOutput {
|
||||
pub value: Value,
|
||||
tool_output: Option<ToolOutput>,
|
||||
// Exact post-operation provider state; never substitute a later path stat.
|
||||
validator: Option<Vec<u8>>,
|
||||
}
|
||||
|
||||
impl WipOperationOutput {
|
||||
@@ -99,6 +108,15 @@ impl WipOperationOutput {
|
||||
Self {
|
||||
value,
|
||||
tool_output: None,
|
||||
validator: None,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn native_with_validator(value: Value, validator: Vec<u8>) -> Self {
|
||||
Self {
|
||||
value,
|
||||
tool_output: None,
|
||||
validator: Some(validator),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -106,6 +124,7 @@ impl WipOperationOutput {
|
||||
Self {
|
||||
value,
|
||||
tool_output: Some(output),
|
||||
validator: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -218,6 +237,19 @@ pub struct WipDynamicMount {
|
||||
pub resolver: Arc<dyn WipDynamicItemResolver>,
|
||||
}
|
||||
|
||||
/// Explicit arbitrary-depth request-time subtree provider. Unlike item families,
|
||||
/// this resolves both files and directories and enumerates authorized children.
|
||||
#[async_trait]
|
||||
pub trait WipSubtreeProvider: Send + Sync {
|
||||
async fn projection(&self, path: &str) -> Result<Option<WipProjection>, ProtocolError>;
|
||||
async fn children(&self, path: &str) -> Result<Vec<String>, ProtocolError>;
|
||||
}
|
||||
|
||||
pub struct WipSubtreeMount {
|
||||
pub root: String,
|
||||
pub provider: Arc<dyn WipSubtreeProvider>,
|
||||
}
|
||||
|
||||
pub trait WipDynamicOperationResolver: Send + Sync {
|
||||
fn handler(&self, item_reference: &str) -> Arc<dyn WipOperationHandler>;
|
||||
}
|
||||
@@ -327,6 +359,7 @@ impl WipOperationHandler for OperationDispatchHandler {
|
||||
pub struct WipMountRegistry {
|
||||
mounts: BTreeMap<String, MountedProjection>,
|
||||
dynamic_mounts: Vec<WipDynamicMount>,
|
||||
subtree_mounts: Vec<WipSubtreeMount>,
|
||||
dynamic_operation_contributions: BTreeMap<String, Vec<MountedDynamicOperationContribution>>,
|
||||
namespaces: BTreeMap<String, String>,
|
||||
replaced_compatibility_capabilities: BTreeMap<String, String>,
|
||||
@@ -395,6 +428,11 @@ impl WipMountRegistry {
|
||||
projection: WipProjection,
|
||||
) -> Result<WipMountDisposition, WipMountError> {
|
||||
validate_projection(&projection)?;
|
||||
if projection.kind == WipProjectionKind::Compatibility
|
||||
&& self.replaces_compatibility(&projection.capability)
|
||||
{
|
||||
return Ok(WipMountDisposition::NativeAlreadySelected);
|
||||
}
|
||||
if projection.kind == WipProjectionKind::Compatibility && !is_tool_route(&projection.route)
|
||||
{
|
||||
return Err(WipMountError::InvalidProjection {
|
||||
@@ -424,6 +462,15 @@ impl WipMountRegistry {
|
||||
});
|
||||
}
|
||||
}
|
||||
if self.subtree_mounts.iter().any(|mount| {
|
||||
projection.route == mount.root
|
||||
|| projection.route.starts_with(&format!("{}/", mount.root))
|
||||
}) {
|
||||
return Err(WipMountError::RouteCollision {
|
||||
route: projection.route,
|
||||
existing: "request-time subtree".into(),
|
||||
});
|
||||
}
|
||||
if self.dynamic_mounts.iter().any(|dynamic| {
|
||||
projection
|
||||
.route
|
||||
@@ -571,6 +618,16 @@ impl WipMountRegistry {
|
||||
message: error.to_string(),
|
||||
}
|
||||
})?;
|
||||
if self.subtree_mounts.iter().any(|m| {
|
||||
m.root == mount.collection_route
|
||||
|| m.root.starts_with(&format!("{}/", mount.collection_route))
|
||||
|| mount.collection_route.starts_with(&format!("{}/", m.root))
|
||||
}) {
|
||||
return Err(WipMountError::RouteCollision {
|
||||
route: mount.collection_route,
|
||||
existing: "request-time subtree".into(),
|
||||
});
|
||||
}
|
||||
let Some(collection) = self.mounts.get(&mount.collection_route) else {
|
||||
return Err(WipMountError::InvalidProjection {
|
||||
route: mount.collection_route,
|
||||
@@ -635,6 +692,42 @@ impl WipMountRegistry {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Delegate an entire descendant namespace to one request-time provider.
|
||||
/// The root Object remains registry-owned; nested/static/item collisions fail.
|
||||
pub fn mount_subtree(&mut self, mount: WipSubtreeMount) -> Result<(), WipMountError> {
|
||||
let Some(root) = self.mounts.get(&mount.root) else {
|
||||
return Err(WipMountError::OperationTargetNotFound { route: mount.root });
|
||||
};
|
||||
if root.projection.kind != WipProjectionKind::Native {
|
||||
return Err(WipMountError::InvalidProjection {
|
||||
route: mount.root,
|
||||
message: "subtree requires a native root Object".into(),
|
||||
});
|
||||
}
|
||||
let overlaps = |path: &str| {
|
||||
path == mount.root
|
||||
|| path.starts_with(&format!("{}/", mount.root))
|
||||
|| mount.root.starts_with(&format!("{path}/"))
|
||||
};
|
||||
if self.subtree_mounts.iter().any(|m| overlaps(&m.root))
|
||||
|| self
|
||||
.dynamic_mounts
|
||||
.iter()
|
||||
.any(|m| overlaps(&m.collection_route))
|
||||
|| self
|
||||
.mounts
|
||||
.keys()
|
||||
.any(|p| p != &mount.root && p.starts_with(&format!("{}/", mount.root)))
|
||||
{
|
||||
return Err(WipMountError::RouteCollision {
|
||||
route: mount.root,
|
||||
existing: "subtree/static/item provider".into(),
|
||||
});
|
||||
}
|
||||
self.subtree_mounts.push(mount);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Add operations to every item resolved by one existing dynamic Object
|
||||
/// family. The family keeps exactly one Object resolver; each Feature only
|
||||
/// supplies handlers for its own disjoint operation names.
|
||||
@@ -793,6 +886,10 @@ impl WipMountRegistry {
|
||||
}
|
||||
}
|
||||
for capability in capabilities {
|
||||
self.mounts.retain(|_, mounted| {
|
||||
mounted.projection.kind != WipProjectionKind::Compatibility
|
||||
|| mounted.projection.capability != capability
|
||||
});
|
||||
self.replaced_compatibility_capabilities
|
||||
.insert(capability, owner_route.to_string());
|
||||
}
|
||||
@@ -964,6 +1061,163 @@ impl WipHost {
|
||||
}
|
||||
}
|
||||
|
||||
fn subtree(&self, path: &str) -> Option<&WipSubtreeMount> {
|
||||
self.registry
|
||||
.subtree_mounts
|
||||
.iter()
|
||||
.find(|m| path == m.root || path.starts_with(&format!("{}/", m.root)))
|
||||
}
|
||||
|
||||
async fn projection_live(&self, path: &str) -> Result<Option<WipProjection>, ProtocolError> {
|
||||
wip_protocol::validate_path(path).map_err(|_| unpublished_path_error(path))?;
|
||||
if let Some(mount) = self.subtree(path) {
|
||||
let Some(projection) = mount.provider.projection(path).await? else {
|
||||
return Ok(None);
|
||||
};
|
||||
validate_projection(&projection).map_err(|_| {
|
||||
protocol_error(
|
||||
ProtocolErrorCode::Internal,
|
||||
"subtree returned an invalid Object",
|
||||
)
|
||||
})?;
|
||||
let owner = &self.registry.mounts[&mount.root].projection.capability;
|
||||
if projection.route != path
|
||||
|| projection.kind != WipProjectionKind::Native
|
||||
|| projection.capability.split(':').next() != owner.split(':').next()
|
||||
{
|
||||
return Err(protocol_error(
|
||||
ProtocolErrorCode::Internal,
|
||||
"subtree changed route or ownership",
|
||||
));
|
||||
}
|
||||
return Ok(Some(projection));
|
||||
}
|
||||
Ok(self.projection(path))
|
||||
}
|
||||
|
||||
async fn descriptor_live(
|
||||
&self,
|
||||
reference: &str,
|
||||
) -> Result<Option<(InterfaceDescriptor, Option<Vec<u8>>)>, ProtocolError> {
|
||||
if let Some((_, encoded)) = reference.rsplit_once("/@/")
|
||||
&& let Some(path) = decode_contextual_path(encoded)
|
||||
&& self.subtree(&path).is_some()
|
||||
{
|
||||
return Ok(self
|
||||
.projection_live(&path)
|
||||
.await?
|
||||
.filter(|p| p.interface == reference)
|
||||
.map(|p| (p.descriptor, p.interface_validator)));
|
||||
}
|
||||
// Never expose the static registration placeholder descriptor of a subtree.
|
||||
if self
|
||||
.registry
|
||||
.subtree_mounts
|
||||
.iter()
|
||||
.any(|m| self.registry.mounts[&m.root].projection.interface == reference)
|
||||
{
|
||||
return Ok(None);
|
||||
}
|
||||
Ok(self.descriptor(reference))
|
||||
}
|
||||
|
||||
async fn fetch_interface_live(
|
||||
&self,
|
||||
reference: &str,
|
||||
) -> Result<FetchInterfaceResponse, ProtocolError> {
|
||||
let (descriptor, validator) = self.descriptor_live(reference).await?.ok_or_else(|| {
|
||||
protocol_error(
|
||||
ProtocolErrorCode::InterfaceNotFound,
|
||||
"interface is not published",
|
||||
)
|
||||
})?;
|
||||
Ok(FetchInterfaceResponse {
|
||||
interface: reference.into(),
|
||||
descriptor,
|
||||
validator,
|
||||
})
|
||||
}
|
||||
|
||||
async fn observe_live(
|
||||
&self,
|
||||
path: &str,
|
||||
depth: u32,
|
||||
) -> Result<ObjectObservation, ProtocolError> {
|
||||
if depth > 32 {
|
||||
return Err(protocol_error(
|
||||
ProtocolErrorCode::ResourceLimitExceeded,
|
||||
"observation depth exceeds 32",
|
||||
));
|
||||
}
|
||||
let mut remaining = 1024usize;
|
||||
tokio::time::timeout(
|
||||
std::time::Duration::from_secs(30),
|
||||
self.observe_live_node(path, depth, &mut remaining),
|
||||
)
|
||||
.await
|
||||
.map_err(|_| {
|
||||
protocol_error(
|
||||
ProtocolErrorCode::ResourceLimitExceeded,
|
||||
"observation deadline exceeded",
|
||||
)
|
||||
})?
|
||||
}
|
||||
|
||||
fn observe_live_node<'a>(
|
||||
&'a self,
|
||||
path: &'a str,
|
||||
depth: u32,
|
||||
remaining: &'a mut usize,
|
||||
) -> std::pin::Pin<
|
||||
Box<dyn std::future::Future<Output = Result<ObjectObservation, ProtocolError>> + Send + 'a>,
|
||||
> {
|
||||
Box::pin(async move {
|
||||
if *remaining == 0 {
|
||||
return Err(protocol_error(
|
||||
ProtocolErrorCode::ResourceLimitExceeded,
|
||||
"observation exceeds 1024 nodes",
|
||||
));
|
||||
}
|
||||
*remaining -= 1;
|
||||
let object = if self.subtree(path).is_some() {
|
||||
self.projection_live(path).await?.map(|p| p.object)
|
||||
} else {
|
||||
self.object_at(path)
|
||||
}
|
||||
.ok_or_else(|| unpublished_path_error(path))?;
|
||||
let children = if depth == 0 {
|
||||
None
|
||||
} else {
|
||||
let paths = if let Some(mount) = self.subtree(path) {
|
||||
mount.provider.children(path).await?
|
||||
} else {
|
||||
self.children(path)
|
||||
};
|
||||
let mut values = Vec::new();
|
||||
let mut seen = BTreeSet::new();
|
||||
for child in paths {
|
||||
let prefix = if path == "/" {
|
||||
"/".to_string()
|
||||
} else {
|
||||
format!("{path}/")
|
||||
};
|
||||
let suffix = child.strip_prefix(&prefix).ok_or_else(|| {
|
||||
protocol_error(ProtocolErrorCode::Internal, "subtree child outside parent")
|
||||
})?;
|
||||
if suffix.is_empty() || suffix.contains('/') || !seen.insert(child.clone()) {
|
||||
return Err(protocol_error(
|
||||
ProtocolErrorCode::Internal,
|
||||
"invalid/duplicate subtree child",
|
||||
));
|
||||
}
|
||||
values.push(self.observe_live_node(&child, depth - 1, remaining).await?);
|
||||
}
|
||||
Some(values)
|
||||
};
|
||||
Ok(ObjectObservation { object, children })
|
||||
})
|
||||
}
|
||||
|
||||
fn descriptor(&self, reference: &str) -> Option<(InterfaceDescriptor, Option<Vec<u8>>)> {
|
||||
for (path, mounted) in &self.registry.mounts {
|
||||
if mounted.projection.interface == reference {
|
||||
@@ -1168,6 +1422,7 @@ impl WipHost {
|
||||
children.into_iter().collect()
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
fn observe(&self, path: &str, depth: u32) -> Result<ObjectObservation, ProtocolError> {
|
||||
let object = self
|
||||
.object_at(path)
|
||||
@@ -1184,6 +1439,7 @@ impl WipHost {
|
||||
Ok(ObjectObservation { object, children })
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
fn fetch_interface(&self, reference: &str) -> Result<FetchInterfaceResponse, ProtocolError> {
|
||||
let (descriptor, validator) = self.descriptor(reference).ok_or_else(|| {
|
||||
protocol_error(
|
||||
@@ -1204,9 +1460,13 @@ impl WipHost {
|
||||
request: CallOperationRequest,
|
||||
context: WipCallContext,
|
||||
) -> Result<WipOperationOutput, WipOperationError> {
|
||||
let projection = self.projection(&request.target.path).ok_or_else(|| {
|
||||
WipOperationError::Protocol(unpublished_path_error(&request.target.path))
|
||||
})?;
|
||||
let projection = self
|
||||
.projection_live(&request.target.path)
|
||||
.await
|
||||
.map_err(WipOperationError::Protocol)?
|
||||
.ok_or_else(|| {
|
||||
WipOperationError::Protocol(unpublished_path_error(&request.target.path))
|
||||
})?;
|
||||
self.call_projection(projection, request, context).await
|
||||
}
|
||||
|
||||
@@ -1716,7 +1976,12 @@ impl WipRuntime {
|
||||
arguments: Json,
|
||||
execution: ToolExecutionContext,
|
||||
) -> Result<ToolOutput, ToolError> {
|
||||
let arguments = match self.host.projection(&path) {
|
||||
let arguments = match self
|
||||
.host
|
||||
.projection_live(&path)
|
||||
.await
|
||||
.map_err(|e| ToolError::InvalidArgument(e.message))?
|
||||
{
|
||||
Some(projection) if projection.kind == WipProjectionKind::Compatibility => {
|
||||
BTreeMap::from([(
|
||||
"input".to_string(),
|
||||
@@ -1847,7 +2112,11 @@ impl WipRuntime {
|
||||
Route::Observe => {
|
||||
let request_value = decode_observe_request(request.body(), self.wire_limits)
|
||||
.map_err(|error| ToolError::Internal(error.to_string()))?;
|
||||
match self.host.observe(&request_value.path, request_value.depth) {
|
||||
match self
|
||||
.host
|
||||
.observe_live(&request_value.path, request_value.depth)
|
||||
.await
|
||||
{
|
||||
Ok(value) => encode_observe_response(&request_value, &value, self.wire_limits),
|
||||
Err(error) => encode_protocol_error_response(
|
||||
ProtocolInteraction::Observe,
|
||||
@@ -1861,7 +2130,11 @@ impl WipRuntime {
|
||||
let request_value =
|
||||
decode_fetch_interface_request(request.body(), self.wire_limits)
|
||||
.map_err(|error| ToolError::Internal(error.to_string()))?;
|
||||
match self.host.fetch_interface(&request_value.interface) {
|
||||
match self
|
||||
.host
|
||||
.fetch_interface_live(&request_value.interface)
|
||||
.await
|
||||
{
|
||||
Ok(value) => {
|
||||
encode_fetch_interface_response(&request_value, &value, self.wire_limits)
|
||||
}
|
||||
@@ -1895,7 +2168,9 @@ impl WipRuntime {
|
||||
.map_err(|error| ToolError::InvalidArgument(error.to_string()))?;
|
||||
let descriptor = self
|
||||
.host
|
||||
.descriptor(&metadata.interface.reference)
|
||||
.descriptor_live(&metadata.interface.reference)
|
||||
.await
|
||||
.map_err(|e| ToolError::InvalidArgument(e.message))?
|
||||
.map(|(descriptor, _)| descriptor.clone())
|
||||
.ok_or_else(|| ToolError::InvalidArgument("interface is not published".into()))?;
|
||||
let request_value = metadata
|
||||
@@ -1905,7 +2180,12 @@ impl WipRuntime {
|
||||
execution,
|
||||
security_context: self.security_context.as_str().to_string(),
|
||||
};
|
||||
let result = match self.host.projection(&request_value.target.path) {
|
||||
let result = match self
|
||||
.host
|
||||
.projection_live(&request_value.target.path)
|
||||
.await
|
||||
.map_err(|e| ToolError::InvalidArgument(e.message))?
|
||||
{
|
||||
Some(projection) => {
|
||||
// Pin the exact resolved instance used for dispatch. A dynamic
|
||||
// resolver may return a fresh stateful handler on each lookup.
|
||||
@@ -1934,14 +2214,37 @@ impl WipRuntime {
|
||||
&descriptor,
|
||||
&CallOperationResponse {
|
||||
result: output.value,
|
||||
validator: self
|
||||
.host
|
||||
.projection(&request_value.target.path)
|
||||
.and_then(|projection| projection.object.validator.clone()),
|
||||
validator: match output.validator {
|
||||
Some(validator) => Some(validator),
|
||||
None if self.host.subtree(&request_value.target.path).is_some() => None,
|
||||
None => self
|
||||
.host
|
||||
.projection(&request_value.target.path)
|
||||
.and_then(|projection| projection.object.validator.clone()),
|
||||
},
|
||||
},
|
||||
self.wire_limits,
|
||||
)
|
||||
.map_err(|error| ToolError::Internal(error.to_string()))?;
|
||||
);
|
||||
let response = match response {
|
||||
Ok(response) => response,
|
||||
Err(_) => {
|
||||
// Dispatch has completed: losing its result is not a pre-effect refusal.
|
||||
let message = "completed operation result could not be encoded; inspect effects before retry";
|
||||
let response = encode_protocol_error_response(
|
||||
ProtocolInteraction::CallOperation,
|
||||
&protocol_error(ProtocolErrorCode::OperationOutcomeUnknown, message),
|
||||
self.wire_limits,
|
||||
)
|
||||
.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}"
|
||||
)))),
|
||||
WipAuditOutcome::OutcomeUnknown,
|
||||
));
|
||||
}
|
||||
};
|
||||
Ok((
|
||||
response,
|
||||
output.tool_output.map(Ok),
|
||||
@@ -2800,6 +3103,58 @@ mod tests {
|
||||
));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn completed_native_result_encoding_failure_is_unknown_and_not_retried() {
|
||||
struct OversizedResult(Arc<AtomicUsize>);
|
||||
#[async_trait]
|
||||
impl WipOperationHandler for OversizedResult {
|
||||
async fn call(
|
||||
&self,
|
||||
_: &str,
|
||||
_: &BTreeMap<String, Value>,
|
||||
_: WipCallContext,
|
||||
) -> Result<WipOperationOutput, WipOperationError> {
|
||||
self.0.fetch_add(1, Ordering::SeqCst);
|
||||
Ok(WipOperationOutput::native(Value::String(
|
||||
"x".repeat(3 * 1024 * 1024),
|
||||
)))
|
||||
}
|
||||
}
|
||||
let calls = Arc::new(AtomicUsize::new(0));
|
||||
let mut p = contribution_projection(Arc::new(OversizedResult(calls.clone())));
|
||||
p.descriptor.operations[0].returns.r#type = TypeExpr::String;
|
||||
p.interface_validator = Some(descriptor_validator(&p.descriptor));
|
||||
let mut registry = WipMountRegistry::new();
|
||||
registry.allocate_namespace("asset", "assets").unwrap();
|
||||
registry.mount(p).unwrap();
|
||||
let runtime = WipRuntime::new(
|
||||
WipHost::new(registry),
|
||||
SecurityContext::new("encoding-test"),
|
||||
0,
|
||||
)
|
||||
.unwrap();
|
||||
runtime
|
||||
.discover("/assets/A-1".into(), 0, true)
|
||||
.await
|
||||
.unwrap();
|
||||
runtime
|
||||
.inspect("test.asset/item/v1".into(), true)
|
||||
.await
|
||||
.unwrap();
|
||||
let error = runtime
|
||||
.call(
|
||||
"/assets/A-1".into(),
|
||||
"test.asset/item/v1".into(),
|
||||
"read".into(),
|
||||
Json::Object(Default::default()),
|
||||
Default::default(),
|
||||
)
|
||||
.await
|
||||
.unwrap_err();
|
||||
assert!(error.to_string().contains("outcome unknown"), "{error}");
|
||||
assert_eq!(calls.load(Ordering::SeqCst), 1);
|
||||
}
|
||||
|
||||
struct ContributionHandler {
|
||||
calls: Arc<AtomicUsize>,
|
||||
allowed: Arc<AtomicBool>,
|
||||
|
||||
Reference in New Issue
Block a user