feat(workspace-config): project grant-bound logical settings through WIP

This commit is contained in:
2026-10-06 06:56:51 +09:00
parent 7a96c42130
commit 0a471242cb
44 changed files with 8159 additions and 72 deletions
+74
View File
@@ -25,6 +25,80 @@ Does not own:
A Worker is runtime authority, not UI state. It should commit model-visible events through history/session paths and keep current Worker-name state in Worker metadata rather than in transient runtime files.
## Workspace config WIP Feature
Enable the opt-in surface in a Worker profile/manifest:
```toml
[feature.workspace_config]
enabled = true
```
This setting accepts **only** `enabled` and defaults to false. Controller
registration additionally requires Workspace identity, an available injected
Workspace Client and WIP mode. It creates no grant. An owner must explicitly
create a Backend grant bound to Workspace/Runtime/Worker and read-only or
read-write access; standalone/Tools-mode config compatibility tools do not exist.
`feature::builtin::workspace_config` registers `/workspace-config` through the
common structured invocation registry (T696). Its optional `access` argument is
`effective` (default), `read_only`, or `read_write`; explicit write never silently
downgrades. Feature preparation attaches before the request reaches the assistant
and reports the logical target and effective access. Native self-attach uses the
same `WorkspaceConfigFeature::attach` path. Backend alias attachment is idempotent
under the session lock; neither entry point blindly retries uncertain outcomes.
The canonical alias is `workspace-config`; the content entrance is WIP
`/workspace-config`. Attachment identity and conditional detach use the existing
T697 Backend ledger. No OS mount, Workdir session, Bash capability, local config
copy or second attachment inventory is created. The registry source is logical,
not a Runtime filesystem target.
The request-time provider maps `/workspace-config` to the Backend's empty root
path and descendants to canonical relative source paths, for example
`/workspace-config/main.dcdl` -> `main.dcdl`. Observations return structure,
Backend validators and current operations, **not bodies**. Files publish `read`;
read-write grants additionally publish `write`, unique-substring `edit`, and
`delete`. Directories and explicit absence observations publish `create` where
Backend metadata permits it. The root publishes `apply_changes` for atomic
create/update/delete/rename sets, allowing new imported
sources and their references to be saved together. Empty directory creation,
commands and generic POSIX operations are not exposed. The adapter uses the typed
`workers/self/workspace-config` GET/attach/observe/read/commit API; it never calls
an owner config API as a bypass.
Client-managed PATH validators are compared against freshly resolved Host
metadata. Each mutation captures the ROOT validator, whole-tree revision/digest,
canonical entrypoints and node state from that same observation. Backend resolves
the live grant/link and validates canonical changes/CAS under its execution lock.
No operation accepts LLM-supplied validators or `expected_digest` fields. Returned
post-call path validators are refreshed from the exact committed revision, not
substituted with root validators or later unrelated state. UI/Worker edits and
same-bytes replacement revisions reject stale calls. Old descriptors cannot
bypass read-only restrictions, detach, revocation or Workspace boundaries.
Bounds: depth 0..8, 256 observed nodes/changes, 512-byte relative paths, 256 KiB
aggregate changed text/read content, and 30-second retrieval/operation deadlines.
Path traversal and malformed parameters fail before content calls. Typed
`NotCommitted` failures remain distinct from `Unknown`; timeouts/lost mutation
responses are unknown and never automatically retried. Raw transport/provider
errors are not copied into model/UI output. Retrieval protocol failures are
reported as failures after recording their Client cache state, not as fresh
successful observations. Saving commits the canonical Workspace configuration
and existing projection/activation pipeline; it does **not** rewrite the resolved
manifests of already-running Workers.
Public integration seams: `WorkspaceConfigFeature::for_workspace`,
`workspace_config::wip::mount_workspace_config_wip`, and
`WipRuntime::from_mounts` plus `discover`/`inspect`/`call`. Worker tests exercise the
real Feature registry, native mount and WIP Client/Host through a typed in-process
contract router: deep discovery, slash/native attach, read/edit/write,
create/delete, atomic multi-source updates, grant restrictions/revocation,
detach, cross-Workspace isolation, concurrent/same-bytes stale observations,
invalid/oversize/save/unknown failures and no retries. Those contract-router tests
are not a substitute for Backend canonical persistence, semantic validation,
projection activation or full Web-to-Backend acceptance tests.
## See also
- [`../../docs/design/worker-session-state.md`](../../docs/design/worker-session-state.md)
+21
View File
@@ -1583,6 +1583,15 @@ where
);
}
let wip_mode = worker.manifest().worker.mode == manifest::WorkerMode::Wip;
let workspace_config_feature =
crate::feature::builtin::workspace_config::WorkspaceConfigFeature::configured(
worker.workspace_client_handle(),
feature_config.workspace_config.enabled,
wip_mode,
);
if let Some(module) = &workspace_config_feature {
feature_registry.add_module(module.clone());
}
let mut workdir_wip_feature = None;
if feature_config.manage_workdir.enabled
|| (wip_mode
@@ -1674,6 +1683,18 @@ 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 let Some(feature) = &workspace_config_feature {
crate::feature::builtin::workspace_config::wip::mount_workspace_config_wip(
&mut wip_mount_registry,
feature,
)
.map_err(|error| {
std::io::Error::new(
std::io::ErrorKind::InvalidInput,
format!("mount Workspace config WIP projection: {error}"),
)
})?;
}
if wip_mode && !backend_job_profile {
crate::checkout::mount_checkouts(
&mut wip_mount_registry,
+1
View File
@@ -24,6 +24,7 @@ pub(crate) mod subjektiv_session;
pub mod task;
pub mod ticket;
pub mod worker_observation;
pub mod workspace_config;
pub mod workspace_worker_discovery;
pub use task::{TaskFeature, task_tools_feature};
@@ -77,6 +77,7 @@ struct Provider {
permissions: Option<ToolPermissionConfig>,
catalog: bool,
manage: bool,
config_content: bool,
}
/// Register Object ownership once, then contribute independent management operations.
@@ -93,6 +94,9 @@ pub fn mount_workspace_workdir_wip(
permissions,
catalog,
manage,
config_content: registry
.routes()
.any(|route| route == crate::feature::builtin::workspace_config::CONTENT_ROOT),
};
for domain in [Domain::Repository, Domain::Workdir, Domain::Attachment] {
let namespace =
@@ -337,6 +341,20 @@ impl Provider {
key.as_str().unwrap_or_default()
));
}
if kind == "workspace_config"
&& source.get("content_path").and_then(Json::as_str)
== Some(crate::feature::builtin::workspace_config::CONTENT_ROOT)
&& matches!(
source.get("access").and_then(Json::as_str),
Some("read_only" | "read_write")
)
{
for field in ["access", "content_path", "purpose"] {
if let Some(value) = source.get(field) {
projected[field] = value.clone();
}
}
}
result.insert("source".into(), projected);
}
}
@@ -344,7 +362,29 @@ 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)));
}
if let Some(alias) = raw.get("alias").and_then(Json::as_str)
// Content references are resolved from current Backend authority,
// never inferred from the alias or an OS session. An old lifetime
// cannot refer to a replacement connection at the same root.
if self.config_content
&& raw.get("alias").and_then(Json::as_str)
== Some(crate::feature::builtin::workspace_config::ATTACHMENT_ALIAS)
&& let Ok(Some(current)) =
crate::feature::builtin::workspace_config::backend::WorkspaceConfigBackend::new(
self.backend.client.clone(),
)
.current()
&& current.connection_id == id
&& raw.get("working_directory_id").and_then(Json::as_str)
== Some(current.working_directory_id.as_str())
&& raw.get("alias").and_then(Json::as_str) == Some(current.alias.as_str())
{
result.insert("content_path".into(), json!(current.content_path));
result.insert("name".into(), json!(current.name));
result.insert("purpose".into(), json!(current.purpose));
result.insert("access".into(), json!(current.access));
}
if !result.contains_key("content_path")
&& 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())
@@ -1038,6 +1078,7 @@ pub(crate) mod tests {
pub(crate) attachments: Vec<Json>,
pub(crate) requests: Vec<WorkspaceRequest>,
pub(crate) denied: bool,
config_current: Option<Json>,
generation: usize,
}
impl Default for CatalogClient {
@@ -1051,6 +1092,7 @@ pub(crate) mod tests {
attachments: vec![],
requests: vec![],
denied: false,
config_current: None,
generation: 0,
}),
}
@@ -1088,6 +1130,9 @@ pub(crate) mod tests {
.map(|b| serde_json::from_str(b).unwrap())
.unwrap_or(Json::Null);
let value = match (request.method, path.split('?').next().unwrap()) {
(WorkspaceRequestMethod::Get, "workers/self/workspace-config") => {
state.config_current.clone().unwrap_or(Json::Null)
}
(WorkspaceRequestMethod::Get, "repositories") => {
json!({"items":state.repositories})
}
@@ -1236,6 +1281,72 @@ pub(crate) mod tests {
})
}
}
#[test]
fn config_content_reference_requires_enabled_mount_and_exact_current_lifetime() {
let client = Arc::new(CatalogClient::default());
let current = json!({"workspace_id":"test-workspace", "alias":"workspace-config", "connection_id":"config-lifetime", "working_directory_id":"config-workdir", "access":"read_only", "name":"Workspace configuration", "purpose":"Configuration editing", "content_path":"/workspace-config", "already_attached":true});
client.state.lock().unwrap().config_current = Some(current);
let raw = json!({"alias":"workspace-config", "connection_id":"config-lifetime", "working_directory_id":"config-workdir", "capabilities":{"bits":25}, "host_path":"/SECRET"});
let mut p = provider(client.clone(), true, true, None);
assert!(
p.projected(Domain::Attachment, &raw)
.ok()
.unwrap()
.get("content_path")
.is_none()
);
p.config_content = true;
let mut physical = raw.clone();
physical["alias"] = json!("checkout");
assert!(
p.projected(Domain::Attachment, &physical)
.ok()
.unwrap()
.get("content_path")
.is_none()
);
assert!(
client.state.lock().unwrap().requests.is_empty(),
"physical inventories must not perform a config request per attachment"
);
let projected = p.projected(Domain::Attachment, &raw).ok().unwrap();
assert_eq!(projected["content_path"], "/workspace-config");
assert_eq!(projected["access"], "read_only");
assert_eq!(projected["purpose"], "Configuration editing");
assert!(projected.get("checkout_path").is_none());
assert!(!projected.to_string().contains("SECRET"));
let mut old = raw.clone();
old["connection_id"] = json!("previous-lifetime");
assert!(
p.projected(Domain::Attachment, &old)
.ok()
.unwrap()
.get("content_path")
.is_none()
);
client.state.lock().unwrap().denied = true;
assert!(
p.projected(Domain::Attachment, &raw)
.ok()
.unwrap()
.get("content_path")
.is_none()
);
}
#[test]
fn config_workdir_source_preserves_safe_mapping_not_private_management_data() {
let p = provider(Arc::new(CatalogClient::default()), true, false, None);
let raw = json!({"working_directory_id":"config-workdir", "source":{"kind":"workspace_config", "access":"read_write", "content_path":"/workspace-config", "purpose":"Configuration editing", "grant_id":"PRIVATE", "host_path":"/SECRET"}});
let projected = p.projected(Domain::Workdir, &raw).ok().unwrap();
assert_eq!(
projected["source"],
json!({"kind":"workspace_config", "access":"read_write", "content_path":"/workspace-config", "purpose":"Configuration editing"})
);
assert!(!projected.to_string().contains("PRIVATE"));
assert!(!projected.to_string().contains("SECRET"));
}
fn provider(
client: Arc<CatalogClient>,
catalog: bool,
@@ -1246,6 +1357,7 @@ pub(crate) mod tests {
backend: ManageWorkdirFeature::new(client).backend(),
catalog,
manage,
config_content: false,
permissions,
}
}
@@ -1573,6 +1685,7 @@ mod lifetime_tests {
permissions: None,
catalog: true,
manage: true,
config_content: false,
};
let handler = CollectionHandler {
provider: p.clone(),
@@ -1668,6 +1781,7 @@ mod logical_capability_tests {
backend: ManageWorkdirFeature::new(client).backend(),
catalog: true,
manage: true,
config_content: false,
permissions: None,
};
let item = ItemHandler {
@@ -0,0 +1,333 @@
//! WIP-only, grant-bound Workspace configuration Feature. No local config store
//! or OS Workdir/session is created by this logical attachment.
pub mod backend;
mod content;
pub mod wip;
use std::sync::Arc;
use std::time::Duration;
use async_trait::async_trait;
use protocol::{
FeatureInvocation, FeatureInvocationDescriptor, FeatureInvocationIdentity,
FeatureInvocationResult, FeatureInvocationStatus, FeatureInvocationSyntax,
InvocationArgumentDescriptor, InvocationArgumentType, InvocationCompletion, InvocationValue,
};
use crate::feature::{
FeatureDescriptor, FeatureInstallContext, FeatureInstallError, FeatureInvocationContext,
FeatureInvocationHandler, FeatureInvocationHandlerError, FeatureModule,
};
use crate::worker::WorkspaceClient;
pub const FEATURE_ID: &str = "workspace-config";
pub const INVOCATION_ID: &str = "builtin:workspace-config/attach";
pub const ATTACHMENT_ALIAS: &str = "workspace-config";
pub const CONTENT_ROOT: &str = "/workspace-config";
const DEADLINE: Duration = Duration::from_secs(30);
/// Requested restriction, not an authorization grant. `None` selects the
/// effective Backend grant. Explicit read_write must never silently downgrade.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ConfigAccess {
ReadOnly,
ReadWrite,
}
impl ConfigAccess {
pub fn as_str(self) -> &'static str {
match self {
Self::ReadOnly => "read_only",
Self::ReadWrite => "read_write",
}
}
}
fn requested_access(value: Option<&str>) -> Result<Option<ConfigAccess>, ConfigAttachError> {
match value {
None | Some("effective") => Ok(None),
Some("read_only") => Ok(Some(ConfigAccess::ReadOnly)),
Some("read_write") => Ok(Some(ConfigAccess::ReadWrite)),
_ => Err(ConfigAttachError::InvalidRequest),
}
}
/// Adapter result from the *existing Backend attachment ledger*. No Worker-local
/// attachment registry or authorization state is created from this result.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ConfigAttachment {
pub connection_id: String,
pub access: ConfigAccess,
pub already_attached: bool,
}
/// Safe diagnostics only. Raw Backend/transport errors may contain host paths or
/// secrets and must be classified by the adapter, never copied into these results.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ConfigAttachError {
InvalidRequest,
Denied,
Unavailable,
OutcomeUnknown,
}
impl ConfigAttachError {
fn message(self) -> &'static str {
match self {
Self::InvalidRequest => "Invalid Workspace config attachment request",
Self::Denied => "Workspace config access denied; explicit access is not downgraded",
Self::Unavailable => "Workspace config attachment unavailable before effects",
Self::OutcomeUnknown => {
"Workspace config attachment outcome unknown; inspect current attachments before retrying"
}
}
}
fn invocation_error(self) -> FeatureInvocationHandlerError {
if self == Self::OutcomeUnknown {
FeatureInvocationHandlerError::outcome_unknown(self.message())
} else {
FeatureInvocationHandlerError::failed(self.message())
}
}
}
/// Narrow application adapter seam, not a config compatibility Tool/API.
///
/// The production implementation captures an identity-bound WorkspaceClient and
/// uses the shared typed Backend attach operation. Both validation (including
/// T696 replay/resume) and attach must consult current authority. Attach rechecks
/// authorization atomically with ledger mutation, enforces explicit access and
/// alias identity. The execution key identifies the host receipt, not a grant;
/// Backend alias attach is idempotent under its session lock. Never blind retry.
#[async_trait]
pub trait WorkspaceConfigAttachmentBackend: Send + Sync {
fn is_available(&self) -> bool;
fn validate_access(&self, access: Option<ConfigAccess>) -> Result<(), ConfigAttachError>;
async fn attach(
&self,
idempotency_key: &str,
access: Option<ConfigAccess>,
) -> Result<ConfigAttachment, ConfigAttachError>;
}
/// Shared slash/self-attach handler. Possession of Workdir tools is not a grant.
#[derive(Clone)]
pub struct WorkspaceConfigFeature {
client: Arc<dyn WorkspaceClient>,
backend: Arc<dyn WorkspaceConfigAttachmentBackend>,
content_backend: Option<Arc<backend::WorkspaceConfigBackend>>,
wip_mode: bool,
permits: Arc<tokio::sync::Semaphore>,
}
impl WorkspaceConfigFeature {
/// Host supplies the actual mode; a user invocation cannot enable WIP or
/// change the Workspace/security context captured by the adapter.
pub fn new(
client: Arc<dyn WorkspaceClient>,
backend: Arc<dyn WorkspaceConfigAttachmentBackend>,
wip_mode: bool,
) -> Self {
Self {
client,
backend,
content_backend: None,
wip_mode,
permits: Arc::new(tokio::sync::Semaphore::new(16)),
}
}
/// Controller registration gate: a flag cannot create Backend authority or
/// turn a Tools/standalone Worker into a Workspace WIP Worker.
pub fn configured(
client: Arc<dyn WorkspaceClient>,
enabled: bool,
wip_mode: bool,
) -> Option<Self> {
(enabled
&& wip_mode
&& client.is_available()
&& client
.workspace_id()
.is_some_and(|id| !id.is_empty() && !id.chars().any(char::is_control)))
.then(|| Self::for_workspace(client, true))
}
/// Production adapter captures the injected identity-bound Workspace client.
pub fn for_workspace(client: Arc<dyn WorkspaceClient>, wip_mode: bool) -> Self {
let backend = Arc::new(backend::WorkspaceConfigBackend::new(client.clone()));
let mut feature = Self::new(client, backend.clone(), wip_mode);
feature.content_backend = Some(backend);
feature
}
pub fn is_available(&self) -> bool {
self.wip_mode
&& self.client.is_available()
&& self.client.workspace_id().is_some_and(|id| !id.is_empty())
&& self.backend.is_available()
}
fn validate_access(&self, access: Option<ConfigAccess>) -> Result<(), ConfigAttachError> {
if !self.is_available() {
return Err(ConfigAttachError::Denied);
}
self.backend.validate_access(access)
}
/// The only attach execution path used by T696 and native WIP. No router
/// session, command capability, or attachment lifetime is synthesized here.
pub async fn attach(
&self,
idempotency_key: &str,
access: Option<ConfigAccess>,
) -> Result<ConfigAttachment, ConfigAttachError> {
if idempotency_key.is_empty() || idempotency_key.len() > 256 {
return Err(ConfigAttachError::InvalidRequest);
}
self.validate_access(access)?;
let _permit = tokio::time::timeout(DEADLINE, self.permits.acquire())
.await
.map_err(|_| ConfigAttachError::Unavailable)?
.map_err(|_| ConfigAttachError::Unavailable)?;
// Authority can change while queued. Backend still rechecks on execution.
self.validate_access(access)?;
let result = tokio::time::timeout(DEADLINE, self.backend.attach(idempotency_key, access))
.await
.map_err(|_| ConfigAttachError::OutcomeUnknown)??;
// The adapter must map only a safe lifetime identity from a validated
// Backend result. Invalid completion metadata cannot imply no effects.
if result.connection_id.is_empty()
|| result.connection_id.len() > 256
|| !result
.connection_id
.bytes()
.all(|b| b.is_ascii_alphanumeric() || matches!(b, b'-' | b'_'))
|| access.is_some_and(|requested| requested != result.access)
{
return Err(ConfigAttachError::OutcomeUnknown);
}
Ok(result)
}
}
pub fn workspace_config_invocation_descriptor() -> FeatureInvocationDescriptor {
FeatureInvocationDescriptor {
identity: FeatureInvocationIdentity(INVOCATION_ID.into()),
name: FEATURE_ID.into(),
aliases: Vec::new(),
display_name: "Workspace config".into(),
description: "Select this Workspace's logical config attachment before the request reaches the assistant. Backend grants determine access; no host filesystem or command access is added.".into(),
syntax: FeatureInvocationSyntax::Parenthesized,
arguments: vec![InvocationArgumentDescriptor {
name: "access".into(),
position: Some(0),
required: false,
value_type: InvocationArgumentType::Enum {
values: vec!["effective".into(), "read_only".into(), "read_write".into()],
},
completion: InvocationCompletion::None,
description: Some("Default: effective Backend grant. read_only restricts the connection; explicit read_write fails if unavailable.".into()),
}],
client_adapter: None,
}
}
impl FeatureModule for WorkspaceConfigFeature {
fn descriptor(&self) -> FeatureDescriptor {
let descriptor = FeatureDescriptor::builtin(FEATURE_ID, "Workspace config")
.with_description("WIP-only logical Workspace config attachment preparation");
if self.wip_mode && self.client.workspace_id().is_some_and(|id| !id.is_empty()) {
descriptor.with_chat_invocation(workspace_config_invocation_descriptor())
} else {
descriptor
}
}
fn install(&self, context: &mut FeatureInstallContext<'_>) -> Result<(), FeatureInstallError> {
if self.wip_mode && self.client.workspace_id().is_some_and(|id| !id.is_empty()) {
context
.chat_invocations()
.register(workspace_config_invocation_descriptor(), self.clone())?;
}
// No tools are declared/installed, including in non-WIP mode.
Ok(())
}
}
fn invocation_access(
invocation: &FeatureInvocation,
) -> Result<Option<ConfigAccess>, ConfigAttachError> {
if invocation.identity.0 != INVOCATION_ID
|| invocation.name != FEATURE_ID
|| invocation.arguments.len() > 1
{
return Err(ConfigAttachError::InvalidRequest);
}
match invocation.arguments.first() {
None => requested_access(None),
Some(argument) if argument.name == "access" => match &argument.value {
InvocationValue::String(value) => requested_access(Some(value)),
_ => Err(ConfigAttachError::InvalidRequest),
},
_ => Err(ConfigAttachError::InvalidRequest),
}
}
#[async_trait]
impl FeatureInvocationHandler for WorkspaceConfigFeature {
fn is_available(&self) -> bool {
WorkspaceConfigFeature::is_available(self)
}
fn validate(
&self,
invocation: &FeatureInvocation,
) -> Result<(), FeatureInvocationHandlerError> {
let access = invocation_access(invocation).map_err(ConfigAttachError::invocation_error)?;
self.validate_access(access)
.map_err(ConfigAttachError::invocation_error)
}
async fn invoke(
&self,
context: FeatureInvocationContext,
invocation: &FeatureInvocation,
) -> Result<FeatureInvocationResult, FeatureInvocationHandlerError> {
if context.invocation_id != invocation.invocation_id {
return Err(ConfigAttachError::InvalidRequest.invocation_error());
}
let access = invocation_access(invocation).map_err(ConfigAttachError::invocation_error)?;
let result = self
.attach(&context.invocation_id, access)
.await
.map_err(ConfigAttachError::invocation_error)?;
let action = if result.already_attached {
"Already attached"
} else {
"Attached"
};
Ok(FeatureInvocationResult {
invocation_id: context.invocation_id,
identity: invocation.identity.clone(),
status: FeatureInvocationStatus::Succeeded,
message: format!(
"{action}: {ATTACHMENT_ALIAS} (Workspace configuration; {}).",
result.access.as_str()
),
context: Some(format!(
"Workspace configuration is available as logical attachment `{ATTACHMENT_ALIAS}` at WIP `{CONTENT_ROOT}` with {} access. Discover structure through WIP; read and change bodies through published Operations. This attachment grants no OS paths or command execution. Backend validation, concurrency checks and activation rules apply; saving does not imply running Workers adopted new configuration.",
result.access.as_str()
)),
})
}
}
#[cfg(test)]
#[path = "workspace_config_tests.rs"]
mod tests;
#[cfg(test)]
mod production_tests;
@@ -0,0 +1,302 @@
//! Identity-bound typed Workspace transport. Never retries a dispatched request.
use std::sync::Arc;
use async_trait::async_trait;
use serde::{Serialize, de::DeserializeOwned};
use server_api::{
WorkspaceConfigAccess, WorkspaceConfigApiError, WorkspaceConfigAttachRequest,
WorkspaceConfigAttachment, WorkspaceConfigFailureClassification,
};
use wip_protocol::{ProtocolError, ProtocolErrorCode};
use super::{
ATTACHMENT_ALIAS, CONTENT_ROOT, ConfigAccess, ConfigAttachError, ConfigAttachment, DEADLINE,
WorkspaceConfigAttachmentBackend,
};
use crate::wip::WipOperationError;
use crate::worker::{
WorkspaceClient, WorkspaceClientError, WorkspaceRequest, WorkspaceRequestMethod,
};
pub const MAX_TEXT_BYTES: usize = 256 * 1024;
pub const MAX_PATH_BYTES: usize = 512;
pub const MAX_NODES: usize = 256;
pub const MAX_DEPTH: u32 = 8;
const MAX_RESPONSE_BYTES: usize = 2 * 1024 * 1024;
#[derive(Clone)]
pub struct WorkspaceConfigBackend {
client: Arc<dyn WorkspaceClient>,
}
impl WorkspaceConfigBackend {
pub fn new(client: Arc<dyn WorkspaceClient>) -> Self {
Self { client }
}
fn base(&self) -> Result<String, WipOperationError> {
let workspace = self
.client
.workspace_id()
.filter(|id| !id.is_empty())
.ok_or_else(|| {
failure(
ProtocolErrorCode::PermissionDenied,
"Workspace config authority unavailable",
)
})?;
let encoded: String = workspace
.bytes()
.map(|byte| {
if byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_') {
(byte as char).to_string()
} else {
format!("%{byte:02X}")
}
})
.collect();
Ok(format!("/api/w/{encoded}/workers/self/workspace-config"))
}
fn request<T: DeserializeOwned>(
&self,
request: WorkspaceRequest,
mutation: bool,
) -> Result<T, WipOperationError> {
let response = self
.client
.execute_with_timeout(request, DEADLINE)
.map_err(|error| {
if mutation && !matches!(error, WorkspaceClientError::InvalidPath(_)) {
unknown()
} else {
failure(
ProtocolErrorCode::Internal,
"Workspace config transport unavailable",
)
}
})?;
if response.body.len() > MAX_RESPONSE_BYTES {
return Err(if mutation {
unknown()
} else {
failure(
ProtocolErrorCode::ResourceLimitExceeded,
"Workspace config response exceeds limit",
)
});
}
if !response.is_success() {
if let Ok(error) = serde_json::from_str::<WorkspaceConfigApiError>(&response.body) {
if error.classification == WorkspaceConfigFailureClassification::Unknown {
return Err(unknown());
}
// Do not reflect arbitrary transport/provider text into model output.
let code = match error.status {
401 | 403 | 404 => ProtocolErrorCode::PermissionDenied,
409 => ProtocolErrorCode::ValidatorMismatch,
413 | 429 => ProtocolErrorCode::ResourceLimitExceeded,
400 | 422 => ProtocolErrorCode::InvalidArguments,
_ => ProtocolErrorCode::Internal,
};
return Err(failure(
code,
"Workspace config request rejected without commit",
));
}
return Err(if mutation {
unknown()
} else {
failure(
if matches!(response.status, 401 | 403 | 404) {
ProtocolErrorCode::PermissionDenied
} else {
ProtocolErrorCode::Internal
},
"Workspace config request denied or unavailable",
)
});
}
serde_json::from_str(&response.body).map_err(|_| {
if mutation {
unknown()
} else {
failure(
ProtocolErrorCode::Internal,
"Invalid Workspace config response",
)
}
})
}
/// GET checks the current grant even when the Worker has not attached yet.
pub fn current(&self) -> Result<Option<WorkspaceConfigAttachment>, WipOperationError> {
let result: Option<WorkspaceConfigAttachment> =
self.request(WorkspaceRequest::get(self.base()?), false)?;
if let Some(attachment) = &result {
self.validate_attachment(attachment)?;
}
Ok(result)
}
fn validate_attachment(
&self,
attachment: &WorkspaceConfigAttachment,
) -> Result<(), WipOperationError> {
if Some(attachment.workspace_id.as_str()) != self.client.workspace_id()
|| attachment.alias != ATTACHMENT_ALIAS
|| attachment.content_path != CONTENT_ROOT
|| attachment.connection_id.is_empty()
|| attachment.connection_id.len() > 256
|| !attachment
.connection_id
.bytes()
.all(|b| b.is_ascii_alphanumeric() || matches!(b, b'-' | b'_'))
{
return Err(failure(
ProtocolErrorCode::PermissionDenied,
"Invalid Workspace config attachment identity",
));
}
Ok(())
}
pub async fn post<T: DeserializeOwned + Send + 'static, R: Serialize>(
&self,
suffix: &str,
request: &R,
mutation: bool,
) -> Result<T, WipOperationError> {
let body = serde_json::to_string(request).map_err(|_| {
failure(
ProtocolErrorCode::InvalidArguments,
"Invalid Workspace config request",
)
})?;
if body.len() > MAX_RESPONSE_BYTES {
return Err(failure(
ProtocolErrorCode::ResourceLimitExceeded,
"Workspace config request exceeds limit",
));
}
let request = WorkspaceRequest::json(
WorkspaceRequestMethod::Post,
format!("{}{suffix}", self.base()?),
body,
);
let this = self.clone();
tokio::time::timeout(
DEADLINE,
tokio::task::spawn_blocking(move || this.request(request, mutation)),
)
.await
.map_err(|_| {
if mutation {
unknown()
} else {
failure(
ProtocolErrorCode::ResourceLimitExceeded,
"Workspace config request deadline exceeded",
)
}
})?
.map_err(|_| {
if mutation {
unknown()
} else {
failure(
ProtocolErrorCode::Internal,
"Workspace config transport unavailable",
)
}
})?
}
pub async fn current_async(
&self,
) -> Result<Option<WorkspaceConfigAttachment>, WipOperationError> {
let this = self.clone();
tokio::time::timeout(
DEADLINE,
tokio::task::spawn_blocking(move || this.current()),
)
.await
.map_err(|_| {
failure(
ProtocolErrorCode::ResourceLimitExceeded,
"Workspace config request deadline exceeded",
)
})?
.map_err(|_| {
failure(
ProtocolErrorCode::Internal,
"Workspace config transport unavailable",
)
})?
}
}
#[async_trait]
impl WorkspaceConfigAttachmentBackend for WorkspaceConfigBackend {
fn is_available(&self) -> bool {
self.client.is_available()
}
fn validate_access(&self, access: Option<ConfigAccess>) -> Result<(), ConfigAttachError> {
let current = self.current().map_err(attach_error)?;
if access == Some(ConfigAccess::ReadWrite)
&& current.is_some_and(|a| a.access == WorkspaceConfigAccess::ReadOnly)
{
return Err(ConfigAttachError::Denied);
}
Ok(())
}
async fn attach(
&self,
_execution_key: &str,
access: Option<ConfigAccess>,
) -> Result<ConfigAttachment, ConfigAttachError> {
// Alias attach is idempotent under the Backend session lock. The T696
// fenced receipt owns replay; no new idempotency protocol or blind retry.
let request = WorkspaceConfigAttachRequest {
alias: Some(ATTACHMENT_ALIAS.into()),
access: access.map(|a| match a {
ConfigAccess::ReadOnly => WorkspaceConfigAccess::ReadOnly,
ConfigAccess::ReadWrite => WorkspaceConfigAccess::ReadWrite,
}),
};
let attachment: WorkspaceConfigAttachment =
self.post("", &request, true).await.map_err(attach_error)?;
self.validate_attachment(&attachment)
.map_err(|_| ConfigAttachError::OutcomeUnknown)?;
Ok(ConfigAttachment {
connection_id: attachment.connection_id,
access: match attachment.access {
WorkspaceConfigAccess::ReadOnly => ConfigAccess::ReadOnly,
WorkspaceConfigAccess::ReadWrite => ConfigAccess::ReadWrite,
},
already_attached: attachment.already_attached,
})
}
}
fn attach_error(error: WipOperationError) -> ConfigAttachError {
match error {
WipOperationError::OutcomeUnknown(_) => ConfigAttachError::OutcomeUnknown,
WipOperationError::Protocol(error) => match error.code {
ProtocolErrorCode::PermissionDenied => ConfigAttachError::Denied,
ProtocolErrorCode::InvalidArguments => ConfigAttachError::InvalidRequest,
_ => ConfigAttachError::Unavailable,
},
_ => ConfigAttachError::Unavailable,
}
}
pub(super) fn failure(code: ProtocolErrorCode, message: &str) -> WipOperationError {
WipOperationError::Protocol(ProtocolError {
code,
message: message.into(),
})
}
pub(super) fn unknown() -> WipOperationError {
WipOperationError::OutcomeUnknown("Workspace config outcome unknown; refresh before any subsequent operation; no automatic retry".into())
}
@@ -0,0 +1,726 @@
//! Request-time logical subtree. Metadata is captured per resolution, never
//! cached as authority; authorization and canonical tree CAS stay in Backend.
use std::collections::{BTreeMap, BTreeSet};
use std::sync::Arc;
use async_trait::async_trait;
use serde::Deserialize;
use serde_json::json;
use server_api::{
ConfigCommitRequest, ConfigContentType, ConfigTreeChange, WorkspaceConfigAccess,
WorkspaceConfigAttachment, WorkspaceConfigCommitRequest, WorkspaceConfigCommitResponse,
WorkspaceConfigNode, WorkspaceConfigNodeKind, WorkspaceConfigObserveRequest,
WorkspaceConfigObserveResponse, WorkspaceConfigReadRequest, WorkspaceConfigReadResponse,
};
use sha2::{Digest, Sha256};
use wip_protocol::{
Documentation, INTERFACE_FORMAT_V1, InterfaceDescriptor, Object, OperationDeclaration,
ParameterDeclaration, ProtocolError, ProtocolErrorCode, ReturnDeclaration, TypeExpr, Value,
};
use super::backend::{
MAX_DEPTH, MAX_NODES, MAX_PATH_BYTES, MAX_TEXT_BYTES, WorkspaceConfigBackend, failure, unknown,
};
use super::{CONTENT_ROOT, WorkspaceConfigFeature};
use crate::wip::{
WipCallContext, WipOperationError, WipOperationHandler, WipOperationOutput, WipProjection,
WipProjectionKind, WipSubtreeProvider, contextual_reference, json_to_wip,
};
pub(super) struct ConfigProvider {
pub feature: WorkspaceConfigFeature,
pub backend: Arc<WorkspaceConfigBackend>,
}
/// Backend tree root is empty; source names are canonical relative config paths.
fn source_path(route: &str) -> Result<String, WipOperationError> {
if route == CONTENT_ROOT {
return Ok(String::new());
}
let path = route
.strip_prefix(&format!("{CONTENT_ROOT}/"))
.ok_or_else(invalid)?;
validate_source(path)?;
Ok(path.into())
}
fn validate_source(path: &str) -> Result<(), WipOperationError> {
if path.is_empty()
|| path.len() > MAX_PATH_BYTES
|| path.starts_with('/')
|| path.contains('\\')
|| path.chars().any(char::is_control)
|| path
.split('/')
.any(|s| s.is_empty() || s == "." || s == "..")
{
return Err(invalid());
}
Ok(())
}
fn route(path: &str) -> String {
if path.is_empty() {
CONTENT_ROOT.into()
} else {
format!("{CONTENT_ROOT}/{path}")
}
}
fn invalid() -> WipOperationError {
failure(
ProtocolErrorCode::InvalidArguments,
"Invalid Workspace config operation arguments",
)
}
fn protocol(error: WipOperationError) -> ProtocolError {
match error {
WipOperationError::Protocol(error) => error,
_ => ProtocolError {
code: ProtocolErrorCode::Internal,
message: "Workspace config metadata unavailable".into(),
},
}
}
/// A denied namespace is absent from observation, not a failure of the whole
/// Worldspace. Do not mask transport, validation or Backend failures as absence.
fn visible_metadata<T>(result: Result<T, WipOperationError>) -> Result<Option<T>, ProtocolError> {
match result {
Ok(value) => Ok(Some(value)),
Err(WipOperationError::Protocol(error))
if error.code == ProtocolErrorCode::PermissionDenied =>
{
Ok(None)
}
Err(error) => Err(protocol(error)),
}
}
fn same_snapshot(
snapshot: &WorkspaceConfigObserveResponse,
other: &WorkspaceConfigObserveResponse,
) -> Result<(), WipOperationError> {
if snapshot.connection_id != other.connection_id
|| snapshot.validator != other.validator
|| snapshot.revision != other.revision
|| snapshot.digest != other.digest
|| snapshot.entrypoints != other.entrypoints
{
return Err(failure(
ProtocolErrorCode::ValidatorMismatch,
"Workspace config observation is stale",
));
}
Ok(())
}
async fn observe(
backend: &WorkspaceConfigBackend,
connection: &str,
paths: Vec<String>,
depth: u32,
) -> Result<WorkspaceConfigObserveResponse, WipOperationError> {
if paths.len() > MAX_NODES || depth > MAX_DEPTH {
return Err(invalid());
}
for path in &paths {
if !path.is_empty() {
validate_source(path)?;
}
}
let result: WorkspaceConfigObserveResponse = backend
.post(
"/observe",
&WorkspaceConfigObserveRequest {
connection_id: connection.into(),
paths: paths.clone(),
depth,
},
false,
)
.await?;
if result.connection_id != connection
|| result.nodes.len() > MAX_NODES
|| result.validator.is_empty()
|| result.validator.len() > 4096
|| result.digest.is_empty()
{
return Err(failure(
ProtocolErrorCode::Internal,
"Invalid Workspace config metadata",
));
}
if result.entrypoints.is_empty() || result.entrypoints.len() > MAX_NODES {
return Err(failure(
ProtocolErrorCode::Internal,
"Invalid Workspace config entrypoints",
));
}
for path in &result.entrypoints {
validate_source(path)?;
}
let mut seen = BTreeSet::new();
for node in &result.nodes {
if !node.path.is_empty() {
validate_source(&node.path)?;
}
if !seen.insert(&node.path)
|| node.validator.is_empty()
|| node.validator.len() > 4096
|| !paths.iter().any(|parent| {
if node.path == *parent {
return true;
}
let suffix = if parent.is_empty() {
Some(node.path.as_str())
} else {
node.path.strip_prefix(&format!("{parent}/"))
};
suffix.is_some_and(|s| s.split('/').count() <= depth as usize)
})
{
return Err(failure(
ProtocolErrorCode::Internal,
"Invalid Workspace config metadata scope",
));
}
}
Ok(result)
}
#[async_trait]
impl WipSubtreeProvider for ConfigProvider {
fn max_depth(&self) -> u32 {
MAX_DEPTH
}
fn max_nodes(&self) -> usize {
MAX_NODES
}
async fn projection(&self, path: &str) -> Result<Option<WipProjection>, ProtocolError> {
let source = source_path(path).map_err(protocol)?;
let Some(attachment) = visible_metadata(self.backend.current_async().await)? else {
return Ok(None);
};
let Some(attachment) = attachment else {
// An authorized but unattached Worker sees only the self-attach entrance.
if path != CONTENT_ROOT {
return Ok(None);
}
let mut projection = super::wip::attach_projection(&self.feature);
projection.interface = contextual_reference(&projection.interface, path);
projection.object.interfaces = vec![projection.interface.clone()];
return Ok(Some(projection));
};
let Some(snapshot) = visible_metadata(
observe(
&self.backend,
&attachment.connection_id,
vec![source.clone()],
0,
)
.await,
)?
else {
return Ok(None);
};
let Some(node) = snapshot.nodes.iter().find(|n| n.path == source).cloned() else {
return Ok(None);
};
let operations = operations(&node, attachment.access, path == CONTENT_ROOT);
// Missing nodes are explicit absence observations for safe create, not files.
if operations.is_empty() && node.kind == WorkspaceConfigNodeKind::Missing {
return Ok(None);
}
let descriptor = descriptor(&operations);
let interface = contextual_reference("yoi.workspace-config/node/v1", path);
let interface_validator = Some(Sha256::digest(format!("{descriptor:?}")).to_vec());
Ok(Some(WipProjection {
route: path.into(),
capability: "workspace-config:content".into(),
kind: WipProjectionKind::Native,
object: Object {
name: path.rsplit('/').next().unwrap_or("workspace-config").into(),
description: Some(format!(
"Logical Workspace config {:?}; bodies are available only through Operations",
node.kind
)),
interfaces: vec![interface.clone()],
r#ref: None,
validator: Some(node.validator.as_bytes().to_vec()),
},
interface,
descriptor,
interface_validator,
handler: Arc::new(ConfigHandler {
feature: self.feature.clone(),
backend: self.backend.clone(),
attachment,
snapshot,
node,
operations,
}),
}))
}
async fn children(&self, path: &str) -> Result<Vec<String>, ProtocolError> {
let source = source_path(path).map_err(protocol)?;
let Some(Some(attachment)) = visible_metadata(self.backend.current_async().await)? else {
return Ok(Vec::new());
};
let Some(snapshot) = visible_metadata(
observe(
&self.backend,
&attachment.connection_id,
vec![source.clone()],
1,
)
.await,
)?
else {
return Ok(Vec::new());
};
Ok(snapshot
.nodes
.iter()
.filter(|n| n.path != source && n.kind != WorkspaceConfigNodeKind::Missing)
.map(|n| route(&n.path))
.collect())
}
}
fn operations(
node: &WorkspaceConfigNode,
access: WorkspaceConfigAccess,
root: bool,
) -> Vec<String> {
let mut names = Vec::new();
// Backend operations are a publication ceiling; never publish a write on RO.
for name in ["read", "write", "edit", "create", "delete", "apply_changes"] {
let kind_ok = match name {
"read" | "write" | "edit" | "delete" => node.kind == WorkspaceConfigNodeKind::File,
"apply_changes" => root,
"create" => node.kind != WorkspaceConfigNodeKind::File,
_ => false,
};
if kind_ok
&& (name == "read" || access == WorkspaceConfigAccess::ReadWrite)
&& node.operations.iter().any(|n| n == name)
{
names.push(name.into());
}
}
if root {
names.push("attach".into());
}
names
}
fn parameter(name: &str, required: bool, r#type: TypeExpr) -> ParameterDeclaration {
ParameterDeclaration {
name: name.into(),
required,
documentation: None,
r#type,
}
}
fn descriptor(names: &[String]) -> InterfaceDescriptor {
InterfaceDescriptor {
format: INTERFACE_FORMAT_V1.into(), types: Vec::new(),
documentation: Some(Documentation {
summary: "Logical Workspace configuration file operations".into(),
details: Some("Backend grants, semantic validation, atomic whole-tree CAS and activation apply. Validators and change digests are managed automatically. apply_changes atomically accepts create/update/delete/rename changes without validators or expected_digest fields. Saving updates configuration for future operation boundaries, not already-running Worker manifests. No OS or command access.".into()),
}),
operations: names.iter().map(|name| OperationDeclaration {
name: name.clone(), documentation: None,
parameters: match name.as_str() {
"attach" => vec![parameter("access", false, TypeExpr::String)],
"read" | "delete" => vec![],
"write" => vec![parameter("content", true, TypeExpr::String)],
"edit" => vec![parameter("old_string", true, TypeExpr::String), parameter("new_string", true, TypeExpr::String), parameter("replace_all", false, TypeExpr::Boolean)],
"create" => vec![parameter("path", false, TypeExpr::String), parameter("content", true, TypeExpr::String), parameter("content_type", false, TypeExpr::String)],
"apply_changes" => vec![parameter("changes", true, TypeExpr::Json)],
_ => vec![],
},
returns: ReturnDeclaration { documentation: None, r#type: TypeExpr::Json },
}).collect(),
}
}
struct ConfigHandler {
feature: WorkspaceConfigFeature,
backend: Arc<WorkspaceConfigBackend>,
attachment: WorkspaceConfigAttachment,
snapshot: WorkspaceConfigObserveResponse,
node: WorkspaceConfigNode,
operations: Vec<String>,
}
#[derive(Debug, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
enum Change {
Create {
path: String,
content: String,
#[serde(default)]
content_type: Option<ConfigContentType>,
},
Update {
path: String,
content: String,
},
Delete {
path: String,
},
Rename {
from: String,
to: String,
},
}
impl Change {
fn target(&self) -> &str {
match self {
Self::Create { path, .. } | Self::Update { path, .. } | Self::Delete { path } => path,
Self::Rename { from, .. } => from,
}
}
}
fn text(arguments: &BTreeMap<String, Value>, key: &str) -> Result<String, WipOperationError> {
match arguments.get(key) {
Some(Value::String(s)) if s.len() <= MAX_TEXT_BYTES => Ok(s.clone()),
_ => Err(invalid()),
}
}
fn content_type(value: Option<&Value>) -> Result<ConfigContentType, WipOperationError> {
match value {
None => Ok(ConfigContentType::Decodal),
Some(Value::String(value)) if value == "decodal" => Ok(ConfigContentType::Decodal),
Some(Value::String(value)) if value == "text" => Ok(ConfigContentType::Text),
_ => Err(invalid()),
}
}
impl ConfigHandler {
async fn read(&self) -> Result<WorkspaceConfigReadResponse, WipOperationError> {
let response: WorkspaceConfigReadResponse = self
.backend
.post(
"/read",
&WorkspaceConfigReadRequest {
connection_id: self.attachment.connection_id.clone(),
path: self.node.path.clone(),
validator: self.node.validator.clone(),
},
false,
)
.await?;
if response.path != self.node.path
|| response.validator != self.node.validator
|| Some(&response.digest) != self.node.digest.as_ref()
|| Some(&response.content_type) != self.node.content_type.as_ref()
|| response.content.len() > MAX_TEXT_BYTES
{
return Err(failure(
ProtocolErrorCode::Internal,
"Invalid Workspace config read response",
));
}
Ok(response)
}
async fn canonical_changes(
&self,
changes: Vec<Change>,
) -> Result<Vec<ConfigTreeChange>, WipOperationError> {
if changes.is_empty() || changes.len() > MAX_NODES {
return Err(invalid());
}
let mut paths = BTreeSet::new();
let mut text_bytes = 0usize;
for change in &changes {
validate_source(change.target())?;
if !paths.insert(change.target().to_owned()) {
return Err(invalid());
}
match change {
Change::Create { content, .. } | Change::Update { content, .. } => {
text_bytes = text_bytes.saturating_add(content.len());
}
Change::Rename { to, .. } => {
validate_source(to)?;
}
_ => {}
}
}
if text_bytes > MAX_TEXT_BYTES {
return Err(failure(
ProtocolErrorCode::ResourceLimitExceeded,
"Workspace config text exceeds limit",
));
}
// Resolve canonical per-file preconditions at the SAME captured tree revision.
let metadata = observe(
&self.backend,
&self.attachment.connection_id,
paths.into_iter().collect(),
0,
)
.await?;
same_snapshot(&self.snapshot, &metadata)?;
changes
.into_iter()
.map(|change| {
let node = metadata
.nodes
.iter()
.find(|n| n.path == change.target())
.ok_or_else(invalid)?;
let digest = || {
node.digest
.clone()
.filter(|_| node.kind == WorkspaceConfigNodeKind::File)
.ok_or_else(|| {
failure(
ProtocolErrorCode::ValidatorMismatch,
"Workspace config file no longer exists",
)
})
};
Ok(match change {
Change::Create {
path,
content,
content_type,
} => {
if node.kind != WorkspaceConfigNodeKind::Missing {
return Err(failure(
ProtocolErrorCode::ValidatorMismatch,
"Workspace config create target already exists",
));
}
ConfigTreeChange::Create {
path,
content,
content_type: content_type.unwrap_or(ConfigContentType::Decodal),
}
}
Change::Update { path, content } => ConfigTreeChange::Update {
path,
content,
expected_digest: digest()?,
},
Change::Delete { path } => ConfigTreeChange::Delete {
path,
expected_digest: digest()?,
},
Change::Rename { from, to } => ConfigTreeChange::Rename {
from,
to,
expected_digest: digest()?,
},
})
})
.collect()
}
}
#[async_trait]
impl WipOperationHandler for ConfigHandler {
async fn call(
&self,
operation: &str,
arguments: &BTreeMap<String, Value>,
context: WipCallContext,
) -> Result<WipOperationOutput, WipOperationError> {
let dispatched = std::sync::atomic::AtomicBool::new(false);
tokio::time::timeout(
super::DEADLINE,
self.execute(operation, arguments, context, &dispatched),
)
.await
.map_err(|_| {
if dispatched.load(std::sync::atomic::Ordering::Acquire) {
unknown()
} else {
failure(
ProtocolErrorCode::ResourceLimitExceeded,
"Workspace config operation deadline exceeded before effects",
)
}
})?
}
}
impl ConfigHandler {
async fn execute(
&self,
operation: &str,
arguments: &BTreeMap<String, Value>,
context: WipCallContext,
dispatched: &std::sync::atomic::AtomicBool,
) -> Result<WipOperationOutput, WipOperationError> {
if !self.operations.iter().any(|name| name == operation) {
return Err(failure(
ProtocolErrorCode::OperationNotFound,
"Workspace config operation is not published",
));
}
// Direct handler calls also enforce parameter shape; no model validators accepted.
let declaration = descriptor(&self.operations)
.operations
.into_iter()
.find(|op| op.name == operation)
.ok_or_else(invalid)?;
if arguments
.keys()
.any(|name| !declaration.parameters.iter().any(|p| p.name == *name))
{
return Err(invalid());
}
if operation == "attach" {
dispatched.store(true, std::sync::atomic::Ordering::Release);
return super::wip::attach_projection(&self.feature)
.handler
.call(operation, arguments, context)
.await;
}
if operation == "read" {
let result = self.read().await?;
// Validator goes to Client state, not the model operation result.
let value = json_to_wip(&json!({
"path": result.path, "content": result.content,
"content_type": result.content_type, "digest": result.digest,
}))
.map_err(|_| invalid())?;
return Ok(WipOperationOutput::native_with_validator(
value,
self.node.validator.as_bytes().to_vec(),
));
}
if self.attachment.access != WorkspaceConfigAccess::ReadWrite {
return Err(failure(
ProtocolErrorCode::PermissionDenied,
"Workspace config write denied",
));
}
let changes = match operation {
"write" => vec![Change::Update {
path: self.node.path.clone(),
content: text(arguments, "content")?,
}],
"edit" => {
let original = self.read().await?.content;
let old = text(arguments, "old_string")?;
let new = text(arguments, "new_string")?;
let all = match arguments.get("replace_all") {
None => false,
Some(Value::Boolean(all)) => *all,
_ => return Err(invalid()),
};
if old.is_empty() {
return Err(invalid());
}
let count = original.matches(&old).count();
if count == 0 || (!all && count != 1) {
return Err(invalid());
}
let count = if all { count } else { 1 };
let size = original
.len()
.checked_sub(count.saturating_mul(old.len()))
.and_then(|n| n.checked_add(count.checked_mul(new.len())?))
.filter(|n| *n <= MAX_TEXT_BYTES);
if size.is_none() {
return Err(failure(
ProtocolErrorCode::ResourceLimitExceeded,
"Workspace config edit exceeds text limit",
));
}
let content = if all {
original.replace(&old, &new)
} else {
original.replacen(&old, &new, 1)
};
vec![Change::Update {
path: self.node.path.clone(),
content,
}]
}
"delete" => vec![Change::Delete {
path: self.node.path.clone(),
}],
"create" => {
let path = match (self.node.kind, arguments.get("path")) {
(WorkspaceConfigNodeKind::Missing, None) => self.node.path.clone(),
(WorkspaceConfigNodeKind::Directory, Some(Value::String(path))) => {
validate_source(path)?;
if self.node.path.is_empty() {
path.clone()
} else {
format!("{}/{path}", self.node.path)
}
}
_ => return Err(invalid()),
};
vec![Change::Create {
path,
content: text(arguments, "content")?,
content_type: Some(content_type(arguments.get("content_type"))?),
}]
}
"apply_changes" => {
let Some(value) = arguments.get("changes") else {
return Err(invalid());
};
serde_json::from_value::<Vec<Change>>(
crate::wip::wip_to_json(value).map_err(|_| invalid())?,
)
.map_err(|_| invalid())?
}
_ => return Err(invalid()),
};
let changes = self.canonical_changes(changes).await?;
dispatched.store(true, std::sync::atomic::Ordering::Release);
let result: WorkspaceConfigCommitResponse = self
.backend
.post(
"/commit",
&WorkspaceConfigCommitRequest {
connection_id: self.attachment.connection_id.clone(),
validator: self.snapshot.validator.clone(),
request: ConfigCommitRequest {
base_revision: self.snapshot.revision,
base_digest: self.snapshot.digest.clone(),
changes,
entrypoints: self.snapshot.entrypoints.clone(),
},
},
true,
)
.await?;
// Commit response supplies ROOT state. Obtain the exact post-commit PATH
// validator, never substitute the root validator for a file validator.
let updated = observe(
&self.backend,
&self.attachment.connection_id,
vec![self.node.path.clone()],
0,
)
.await
.map_err(|_| unknown())?;
if updated.revision != result.revision
|| updated.digest != result.digest
|| updated.validator != result.validator
{
return Err(unknown());
}
let node = updated
.nodes
.iter()
.find(|n| n.path == self.node.path)
.ok_or_else(unknown)?;
let value = json_to_wip(&json!({"revision": result.revision, "digest": result.digest,
"activation": "Saved to canonical Workspace configuration; existing Worker manifests are not rewritten"})).map_err(|_| unknown())?;
Ok(WipOperationOutput::native_with_validator(
value,
node.validator.as_bytes().to_vec(),
))
}
}
@@ -0,0 +1,884 @@
//! Worker-side E2E across real Feature invocation, WIP Client/Host and typed
//! Workspace transport. Canonical persistence/projection tests live in Backend;
//! this deterministic router models its published auth/CAS/response contract.
use super::*;
use crate::feature::FeatureRegistryBuilder;
use crate::hook::HookRegistryBuilder;
use crate::wip::{WipMountRegistry, WipRuntime};
use crate::worker::{
WorkspaceClient, WorkspaceClientError, WorkspaceRequest, WorkspaceRequestMethod,
WorkspaceResponse,
};
use agen::tool::ToolExecutionContext;
use protocol::{FeatureInvocation, FeatureInvocationIdentity};
use serde::Serialize;
use serde_json::{Value as Json, json};
use server_api::*;
use sha2::{Digest, Sha256};
use std::collections::{BTreeMap, BTreeSet};
use std::sync::{Arc, Mutex};
#[derive(Debug)]
struct State {
granted: bool,
read_only: bool,
active: Option<WorkspaceConfigAttachment>,
next_connection: u64,
revision: u64,
entries: BTreeMap<String, (ConfigContentType, String)>,
commits: Vec<WorkspaceConfigCommitRequest>,
requests: Vec<WorkspaceRequest>,
fail_commit: bool,
fail_current: bool,
lose_commit_reply: bool,
race_commit: bool,
}
#[derive(Debug)]
struct Router {
workspace: &'static str,
state: Arc<Mutex<State>>,
}
impl Router {
fn new() -> Arc<Self> {
Arc::new(Self {
workspace: "workspace-1",
state: Arc::new(Mutex::new(State {
granted: true,
read_only: false,
active: None,
next_connection: 0,
revision: 1,
entries: BTreeMap::from([
(
"main.dcdl".into(),
(ConfigContentType::Decodal, "model = old".into()),
),
(
"profiles/team/deep/model.dcdl".into(),
(ConfigContentType::Decodal, "model = deep".into()),
),
]),
commits: Vec::new(),
requests: Vec::new(),
fail_commit: false,
fail_current: false,
lose_commit_reply: false,
race_commit: false,
})),
})
}
}
fn response<T: Serialize>(value: &T) -> WorkspaceResponse {
WorkspaceResponse {
status: 200,
body: serde_json::to_string(value).unwrap(),
}
}
fn denied(status: u16, classification: WorkspaceConfigFailureClassification) -> WorkspaceResponse {
// Deliberately unsafe provider text must never be propagated by the adapter.
WorkspaceResponse {
status,
body: serde_json::to_string(&WorkspaceConfigApiError {
status,
code: "test".into(),
message: "secret-token /private/host/config".into(),
classification,
})
.unwrap(),
}
}
fn digest(content: &str) -> String {
Sha256::digest(content.as_bytes())
.iter()
.map(|byte| format!("{byte:02x}"))
.collect()
}
impl State {
fn tree_digest(&self) -> String {
digest(&serde_json::to_string(&self.entries).unwrap())
}
fn validator(&self, path: &str) -> String {
format!(
"wc:{}:{}:{}:{}",
self.active.as_ref().unwrap().connection_id,
self.revision,
self.tree_digest(),
path
)
}
fn node(&self, path: &str, kind: WorkspaceConfigNodeKind) -> WorkspaceConfigNode {
let mut operations = match kind {
WorkspaceConfigNodeKind::File => vec!["read".into()],
_ => vec![],
};
if !self.read_only {
match kind {
WorkspaceConfigNodeKind::File => {
operations.extend(["write", "edit", "delete"].map(str::to_string))
}
WorkspaceConfigNodeKind::Missing => operations.push("create".into()),
WorkspaceConfigNodeKind::Directory => {
operations.push("create".into());
if path.is_empty() {
operations.push("apply_changes".into());
}
}
}
}
WorkspaceConfigNode {
path: path.into(),
kind,
validator: self.validator(path),
digest: self.entries.get(path).map(|(_, s)| digest(s)),
content_type: self.entries.get(path).map(|(t, _)| *t),
operations,
}
}
}
impl WorkspaceClient for Router {
fn workspace_id(&self) -> Option<&str> {
Some(self.workspace)
}
fn kind(&self) -> &str {
"typed-config-router"
}
fn is_available(&self) -> bool {
true
}
fn execute(
&self,
request: WorkspaceRequest,
) -> Result<WorkspaceResponse, WorkspaceClientError> {
let mut state = self.state.lock().unwrap();
state.requests.push(request.clone());
let prefix = "/api/w/workspace-1/workers/self/workspace-config";
if self.workspace != "workspace-1" || !request.path.starts_with(prefix) || !state.granted {
return Ok(denied(
403,
WorkspaceConfigFailureClassification::NotCommitted,
));
}
let suffix = request.path.strip_prefix(prefix).unwrap();
if suffix.is_empty() {
if request.method == WorkspaceRequestMethod::Get {
if state.fail_current {
return Ok(WorkspaceResponse {
status: 503,
body: "unsafe provider diagnostic /private/host/config".into(),
});
}
return Ok(response(&state.active));
}
let attach: WorkspaceConfigAttachRequest =
serde_json::from_str(request.body.as_ref().unwrap()).unwrap();
assert_eq!(attach.alias.as_deref(), Some(ATTACHMENT_ALIAS));
if attach.access == Some(WorkspaceConfigAccess::ReadWrite) && state.read_only {
return Ok(denied(
403,
WorkspaceConfigFailureClassification::NotCommitted,
));
}
let already_attached = state.active.is_some();
if !already_attached {
state.next_connection += 1;
}
let access = attach.access.unwrap_or(if state.read_only {
WorkspaceConfigAccess::ReadOnly
} else {
WorkspaceConfigAccess::ReadWrite
});
state.read_only |= access == WorkspaceConfigAccess::ReadOnly;
let attachment = WorkspaceConfigAttachment {
workspace_id: "workspace-1".into(),
connection_id: format!("connection-{}", state.next_connection),
alias: ATTACHMENT_ALIAS.into(),
working_directory_id: "logical-grant-1".into(),
access,
name: "Workspace configuration".into(),
purpose: "Canonical config".into(),
content_path: CONTENT_ROOT.into(),
already_attached,
};
state.active = Some(attachment.clone());
return Ok(response(&attachment));
}
let input: Json = serde_json::from_str(request.body.as_ref().unwrap()).unwrap();
if state.active.as_ref().map(|a| a.connection_id.as_str())
!= input["connection_id"].as_str()
{
return Ok(denied(
409,
WorkspaceConfigFailureClassification::NotCommitted,
));
}
match suffix {
"/observe" => {
let request: WorkspaceConfigObserveRequest = serde_json::from_value(input).unwrap();
assert!(request.depth <= 8 && request.paths.len() <= 256);
let mut kinds =
BTreeMap::from([(String::new(), WorkspaceConfigNodeKind::Directory)]);
for path in state.entries.keys() {
kinds.insert(path.clone(), WorkspaceConfigNodeKind::File);
let mut path = path.as_str();
while let Some((parent, _)) = path.rsplit_once('/') {
kinds
.entry(parent.into())
.or_insert(WorkspaceConfigNodeKind::Directory);
path = parent;
}
}
let mut selected = BTreeSet::new();
for path in &request.paths {
selected.insert(path.clone());
let prefix = if path.is_empty() {
String::new()
} else {
format!("{path}/")
};
for child in kinds.keys() {
if let Some(suffix) = child.strip_prefix(&prefix)
&& child != path
&& suffix.split('/').count() <= request.depth as usize
{
selected.insert(child.clone());
}
}
}
if selected.len() > 256 {
return Ok(denied(
413,
WorkspaceConfigFailureClassification::NotCommitted,
));
}
let nodes = selected
.into_iter()
.map(|p| {
state.node(
&p,
*kinds.get(&p).unwrap_or(&WorkspaceConfigNodeKind::Missing),
)
})
.collect();
Ok(response(&WorkspaceConfigObserveResponse {
connection_id: state.active.as_ref().unwrap().connection_id.clone(),
validator: state.validator(""),
revision: state.revision,
digest: state.tree_digest(),
entrypoints: vec!["main.dcdl".into()],
nodes,
}))
}
"/read" => {
let request: WorkspaceConfigReadRequest = serde_json::from_value(input).unwrap();
if request.validator != state.validator(&request.path) {
return Ok(denied(
409,
WorkspaceConfigFailureClassification::NotCommitted,
));
}
let Some((content_type, content)) = state.entries.get(&request.path) else {
return Ok(denied(
404,
WorkspaceConfigFailureClassification::NotCommitted,
));
};
Ok(response(&WorkspaceConfigReadResponse {
path: request.path.clone(),
content: content.clone(),
content_type: *content_type,
digest: digest(content),
validator: state.validator(&request.path),
}))
}
"/commit" => {
let request: WorkspaceConfigCommitRequest = serde_json::from_value(input).unwrap();
state.commits.push(request.clone());
if state.race_commit {
state.revision += 1;
state.race_commit = false;
}
if state.read_only {
return Ok(denied(
403,
WorkspaceConfigFailureClassification::NotCommitted,
));
}
assert_eq!(request.request.entrypoints, ["main.dcdl"]);
if request.validator != state.validator("")
|| request.request.base_revision != state.revision
|| request.request.base_digest != state.tree_digest()
{
return Ok(denied(
409,
WorkspaceConfigFailureClassification::NotCommitted,
));
}
if state.fail_commit {
return Ok(denied(
500,
WorkspaceConfigFailureClassification::NotCommitted,
));
}
let mut entries = state.entries.clone();
for change in request.request.changes {
match change {
ConfigTreeChange::Create {
path,
content_type,
content,
} => {
if entries.contains_key(&path) {
return Ok(denied(
409,
WorkspaceConfigFailureClassification::NotCommitted,
));
}
entries.insert(path, (content_type, content));
}
ConfigTreeChange::Update {
path,
expected_digest,
content,
} => {
let Some((kind, old)) = entries.get(&path) else {
return Ok(denied(
409,
WorkspaceConfigFailureClassification::NotCommitted,
));
};
if digest(old) != expected_digest {
return Ok(denied(
409,
WorkspaceConfigFailureClassification::NotCommitted,
));
}
entries.insert(path, (*kind, content));
}
ConfigTreeChange::Delete {
path,
expected_digest,
} => {
if entries
.get(&path)
.is_none_or(|(_, s)| digest(s) != expected_digest)
{
return Ok(denied(
409,
WorkspaceConfigFailureClassification::NotCommitted,
));
}
entries.remove(&path);
}
ConfigTreeChange::Rename {
from,
to,
expected_digest,
} => {
if entries
.get(&from)
.is_none_or(|(_, s)| digest(s) != expected_digest)
|| entries.contains_key(&to)
{
return Ok(denied(
409,
WorkspaceConfigFailureClassification::NotCommitted,
));
}
let entry = entries.remove(&from).unwrap();
entries.insert(to, entry);
}
}
}
if entries.values().any(|(_, s)| s == "INVALID") {
return Ok(denied(
400,
WorkspaceConfigFailureClassification::NotCommitted,
));
}
state.entries = entries;
state.revision += 1;
if state.lose_commit_reply {
state.lose_commit_reply = false;
return Err(WorkspaceClientError::Request(
"secret-token /private/host/config".into(),
));
}
Ok(response(&WorkspaceConfigCommitResponse {
validator: state.validator(""),
revision: state.revision,
digest: state.tree_digest(),
}))
}
_ => panic!("unregistered endpoint: {suffix}"),
}
}
}
fn invocation(key: &str) -> FeatureInvocation {
FeatureInvocation {
invocation_id: key.into(),
identity: FeatureInvocationIdentity(INVOCATION_ID.into()),
name: FEATURE_ID.into(),
arguments: vec![],
}
}
fn runtime(client: Arc<dyn WorkspaceClient>) -> (WorkspaceConfigFeature, WipRuntime) {
let feature = WorkspaceConfigFeature::for_workspace(client, true);
let mut mounts = WipMountRegistry::new();
wip::mount_workspace_config_wip(&mut mounts, &feature).unwrap();
let runtime = WipRuntime::from_mounts(mounts, "worker-1@workspace-1".into()).unwrap();
(feature, runtime)
}
fn interface(path: &str, attached: bool) -> String {
crate::wip::contextual_reference(
if attached {
"yoi.workspace-config/node/v1"
} else {
"yoi.workspace-config/attachment/v1"
},
path,
)
}
async fn observe_interface(runtime: &WipRuntime, path: &str, attached: bool) {
runtime.discover(path.into(), 0, true).await.unwrap();
runtime
.inspect(interface(path, attached), true)
.await
.unwrap();
}
async fn call(
runtime: &WipRuntime,
path: &str,
op: &str,
args: Json,
) -> Result<agen::tool::ToolOutput, agen::tool::ToolError> {
runtime
.call(
path.into(),
interface(path, true),
op.into(),
args,
ToolExecutionContext::direct(),
)
.await
}
#[tokio::test]
async fn workspace_config_production_slash_deep_discover_read_edit_and_atomic_import_changes() {
let client = Router::new();
let (feature, runtime) = runtime(client.clone());
let mut tools = Vec::new();
let report = FeatureRegistryBuilder::new()
.with_module(feature)
.install_into_pending(&mut tools, &mut HookRegistryBuilder::default());
assert!(tools.is_empty() && !report.has_errors());
assert_eq!(
report
.chat_invocations
.feature_completions("workspace-c")
.len(),
1
);
assert!(
report
.chat_invocations
.invoke(&invocation("slash-1"))
.await
.unwrap()
.message
.starts_with("Attached")
);
assert!(
report
.chat_invocations
.invoke(&invocation("slash-2"))
.await
.unwrap()
.message
.starts_with("Already attached")
);
let discovered = runtime
.discover(CONTENT_ROOT.into(), 5, true)
.await
.unwrap()
.content
.unwrap();
assert!(discovered.contains("profiles/team/deep/model.dcdl"));
assert!(!discovered.contains("model = deep"));
let path = "/workspace-config/profiles/team/deep/model.dcdl";
observe_interface(&runtime, path, true).await;
assert!(
call(&runtime, path, "read", json!({}))
.await
.unwrap()
.content
.unwrap()
.contains("model = deep")
);
call(
&runtime,
path,
"edit",
json!({"old_string":"deep","new_string":"changed"}),
)
.await
.unwrap();
assert_eq!(
client.state.lock().unwrap().entries["profiles/team/deep/model.dcdl"].1,
"model = changed"
);
// Client receives the new path validator automatically, no model tracking.
call(&runtime, path, "write", json!({"content":"model = second"}))
.await
.unwrap();
observe_interface(&runtime, CONTENT_ROOT, true).await;
call(
&runtime,
CONTENT_ROOT,
"apply_changes",
json!({"changes":[
{"kind":"create","path":"profiles/new.dcdl","content":"model = imported"},
{"kind":"update","path":"main.dcdl","content":"import profiles/new.dcdl"}
]}),
)
.await
.unwrap();
let state = client.state.lock().unwrap();
assert_eq!(state.entries["main.dcdl"].1, "import profiles/new.dcdl");
assert_eq!(state.entries["profiles/new.dcdl"].1, "model = imported");
assert_eq!(state.commits.last().unwrap().request.changes.len(), 2);
assert!(
state
.requests
.iter()
.all(|r| !r.path.contains("session") && !r.path.contains("bash"))
);
}
#[tokio::test]
async fn workspace_config_production_native_attach_create_delete_and_detach_end_use() {
let client = Router::new();
let (_, runtime) = runtime(client.clone());
observe_interface(&runtime, CONTENT_ROOT, false).await;
runtime
.call(
CONTENT_ROOT.into(),
interface(CONTENT_ROOT, false),
"attach".into(),
json!({}),
ToolExecutionContext::direct(),
)
.await
.unwrap();
let directory = "/workspace-config/profiles";
observe_interface(&runtime, directory, true).await;
call(
&runtime,
directory,
"create",
json!({"path":"notes.txt", "content_type":"text", "content":"configuration notes"}),
)
.await
.unwrap();
assert_eq!(
client.state.lock().unwrap().entries["profiles/notes.txt"].0,
ConfigContentType::Text
);
let path = "/workspace-config/new/nested/file.dcdl";
observe_interface(&runtime, path, true).await;
call(&runtime, path, "create", json!({"content":"new = true"}))
.await
.unwrap();
observe_interface(&runtime, path, true).await;
call(&runtime, path, "delete", json!({})).await.unwrap();
assert!(
!client
.state
.lock()
.unwrap()
.entries
.contains_key("new/nested/file.dcdl")
);
client.state.lock().unwrap().active = None;
assert!(
call(&runtime, "/workspace-config/main.dcdl", "read", json!({}))
.await
.is_err()
);
assert!(
runtime
.discover("/workspace-config/main.dcdl".into(), 0, true)
.await
.is_err()
);
}
#[tokio::test]
async fn workspace_config_production_revocation_ro_cross_workspace_and_cache_isolation() {
let client = Router::new();
let (feature, runtime) = runtime(client.clone());
feature.attach("grant", None).await.unwrap();
let path = "/workspace-config/main.dcdl";
observe_interface(&runtime, path, true).await;
client.state.lock().unwrap().granted = false;
assert!(call(&runtime, path, "read", json!({})).await.is_err());
assert!(feature.attach("revoked", None).await.is_err());
client.state.lock().unwrap().granted = true;
client.state.lock().unwrap().read_only = true;
client.state.lock().unwrap().active.as_mut().unwrap().access = WorkspaceConfigAccess::ReadOnly;
assert!(
feature
.attach("explicit-write", Some(ConfigAccess::ReadWrite))
.await
.is_err()
);
observe_interface(&runtime, path, true).await;
let descriptor = runtime
.inspect(interface(path, true), true)
.await
.unwrap()
.content
.unwrap();
let descriptor: Json = serde_json::from_str(&descriptor).unwrap();
let names = descriptor["descriptor"]["operations"]
.as_array()
.unwrap()
.iter()
.map(|op| op["name"].as_str().unwrap())
.collect::<Vec<_>>();
assert_eq!(names, ["read"]);
assert!(
call(&runtime, path, "write", json!({"content":"DENIED"}))
.await
.is_err()
);
let cross = Arc::new(Router {
workspace: "other-workspace",
state: client.state.clone(),
});
let (feature, other) = runtime_for_cross(cross);
assert!(feature.attach("cross", None).await.is_err());
assert!(other.discover(path.into(), 0, false).await.is_err());
assert!(client.state.lock().unwrap().commits.is_empty());
}
fn runtime_for_cross(client: Arc<dyn WorkspaceClient>) -> (WorkspaceConfigFeature, WipRuntime) {
runtime(client)
}
#[tokio::test]
async fn workspace_config_production_stale_same_bytes_and_commit_race_never_overwrite() {
let client = Router::new();
let (feature, runtime) = runtime(client.clone());
feature.attach("initial", None).await.unwrap();
let path = "/workspace-config/main.dcdl";
observe_interface(&runtime, path, true).await;
client.state.lock().unwrap().revision += 1; // Same bytes, different identity revision.
assert!(
call(&runtime, path, "write", json!({"content":"overwritten"}))
.await
.is_err()
);
assert!(client.state.lock().unwrap().commits.is_empty());
observe_interface(&runtime, path, true).await;
client.state.lock().unwrap().race_commit = true;
assert!(
call(&runtime, path, "write", json!({"content":"raced"}))
.await
.is_err()
);
assert_eq!(
client.state.lock().unwrap().entries["main.dcdl"].1,
"model = old"
);
assert_eq!(client.state.lock().unwrap().commits.len(), 1);
}
#[tokio::test]
async fn workspace_config_production_invalid_size_save_failure_unknown_are_safe_and_not_retried() {
let client = Router::new();
let (feature, runtime) = runtime(client.clone());
feature.attach("initial", None).await.unwrap();
let path = "/workspace-config/main.dcdl";
for content in ["INVALID".into(), "x".repeat(256 * 1024 + 1)] {
observe_interface(&runtime, path, true).await;
let error = call(&runtime, path, "write", json!({"content":content}))
.await
.unwrap_err()
.to_string();
assert!(!error.contains("secret-token") && !error.contains("/private/host"));
assert_eq!(
client.state.lock().unwrap().entries["main.dcdl"].1,
"model = old"
);
}
observe_interface(&runtime, path, true).await;
let before = client.state.lock().unwrap().commits.len();
assert!(
call(
&runtime,
path,
"write",
json!({"content":"valid","validator":"model-supplied"})
)
.await
.is_err()
);
assert_eq!(before, client.state.lock().unwrap().commits.len());
client.state.lock().unwrap().fail_commit = true;
assert!(
call(&runtime, path, "write", json!({"content":"valid"}))
.await
.is_err()
);
assert_eq!(
client.state.lock().unwrap().entries["main.dcdl"].1,
"model = old"
);
client.state.lock().unwrap().fail_commit = false;
observe_interface(&runtime, path, true).await;
client.state.lock().unwrap().lose_commit_reply = true;
let count = client.state.lock().unwrap().commits.len();
let error = call(
&runtime,
path,
"write",
json!({"content":"saved-but-reply-lost"}),
)
.await
.unwrap_err()
.to_string();
assert!(error.contains("unknown"));
assert!(!error.contains("secret-token") && !error.contains("/private/host"));
assert_eq!(client.state.lock().unwrap().commits.len(), count + 1);
assert_eq!(
client.state.lock().unwrap().entries["main.dcdl"].1,
"saved-but-reply-lost"
);
}
#[tokio::test]
async fn workspace_config_production_bounds_and_traversal_reject_before_content_calls() {
let client = Router::new();
let (feature, runtime) = runtime(client.clone());
feature.attach("initial", None).await.unwrap();
assert!(
runtime
.discover(CONTENT_ROOT.into(), 9, true)
.await
.is_err()
);
for path in [
"/workspace-config/../secret",
"/workspace-config//main.dcdl",
"/workspace-config/./main.dcdl",
] {
assert!(runtime.discover(path.into(), 0, true).await.is_err());
}
let path = format!("/workspace-config/{}", "a".repeat(513));
assert!(runtime.discover(path, 0, true).await.is_err());
assert!(client.state.lock().unwrap().commits.is_empty());
}
#[tokio::test]
async fn workspace_config_production_reattach_new_lifetime_rejects_old_same_bytes_cache() {
let client = Router::new();
let (feature, runtime) = runtime(client.clone());
let first = feature.attach("first", None).await.unwrap();
let path = "/workspace-config/main.dcdl";
observe_interface(&runtime, path, true).await;
client.state.lock().unwrap().active = None;
let second = feature.attach("second", None).await.unwrap();
assert_ne!(first.connection_id, second.connection_id);
assert!(
call(&runtime, path, "write", json!({"content":"bad"}))
.await
.is_err()
);
assert!(client.state.lock().unwrap().commits.is_empty());
observe_interface(&runtime, path, true).await;
call(&runtime, path, "write", json!({"content":"new-lifetime"}))
.await
.unwrap();
assert_eq!(
client.state.lock().unwrap().entries["main.dcdl"].1,
"new-lifetime"
);
}
#[tokio::test]
async fn workspace_config_production_cumulative_node_limit_and_edit_growth_are_bounded() {
let client = Router::new();
let (feature, runtime) = runtime(client.clone());
feature.attach("initial", None).await.unwrap();
let path = "/workspace-config/main.dcdl";
observe_interface(&runtime, path, true).await;
let before = client.state.lock().unwrap().commits.len();
assert!(
call(
&runtime,
path,
"edit",
json!({
"old_string":"o", "new_string":"x".repeat(256*1024), "replace_all":true
})
)
.await
.is_err()
);
assert_eq!(client.state.lock().unwrap().commits.len(), before);
{
let mut state = client.state.lock().unwrap();
for i in 0..90 {
state.entries.insert(
format!("groups/{i}/nested/f.dcdl"),
(ConfigContentType::Decodal, "x = true".into()),
);
}
}
// No individual depth-1 response exceeds the cap; the whole Host traversal
// must nevertheless stop before exceeding 256 Objects.
let error = runtime
.discover(CONTENT_ROOT.into(), 4, true)
.await
.unwrap_err();
assert!(error.to_string().contains("ResourceLimitExceeded"));
}
#[tokio::test]
async fn workspace_config_production_backend_failure_is_not_hidden_as_unauthorized_absence() {
let client = Router::new();
client.state.lock().unwrap().fail_current = true;
let (_, runtime) = runtime(client);
let error = runtime
.discover(CONTENT_ROOT.into(), 0, true)
.await
.unwrap_err();
let message = error.to_string();
assert!(message.contains("Internal"), "{message}");
assert!(!message.contains("/private/host/config"));
}
#[tokio::test]
async fn workspace_config_production_ungranted_root_does_not_poison_worldspace_discovery() {
let client = Router::new();
client.state.lock().unwrap().granted = false;
let (_, runtime) = runtime(client.clone());
let discovery = runtime.discover("/".into(), 2, true).await.unwrap();
let discovery: Json = serde_json::from_str(discovery.content.as_deref().unwrap()).unwrap();
assert_eq!(discovery["known_space"].as_array().unwrap().len(), 1);
assert_eq!(discovery["known_space"][0]["path"], "/");
assert!(
runtime
.discover(CONTENT_ROOT.into(), 0, true)
.await
.is_err()
);
assert!(
client
.state
.lock()
.unwrap()
.requests
.iter()
.all(|r| r.method == WorkspaceRequestMethod::Get)
);
}
@@ -0,0 +1,182 @@
//! Shared native self-attach entrance and production request-time config subtree.
use std::collections::BTreeMap;
use std::sync::Arc;
use async_trait::async_trait;
use serde_json::json;
use sha2::{Digest, Sha256};
use wip_protocol::{
Documentation, INTERFACE_FORMAT_V1, InterfaceDescriptor, Object, OperationDeclaration,
ParameterDeclaration, ProtocolError, ProtocolErrorCode, ReturnDeclaration, TypeExpr, Value,
};
use super::{
ATTACHMENT_ALIAS, CONTENT_ROOT, ConfigAttachError, WorkspaceConfigFeature, requested_access,
};
use crate::feature::builtin::manage_workdir::wip::encode_identity;
use crate::wip::{
WipCallContext, WipMountError, WipMountRegistry, WipOperationError, WipOperationHandler,
WipOperationOutput, WipProjection, WipProjectionKind, json_to_wip,
};
/// Mount the Feature's native attachment entrance and, for the production
/// adapter, the grant-bound config subtree. No static file inventory is created.
pub fn mount_workspace_config_wip(
registry: &mut WipMountRegistry,
feature: &WorkspaceConfigFeature,
) -> Result<(), WipMountError> {
mount_workspace_config_attach_wip(registry, feature)?;
if feature.wip_mode
&& feature
.client
.workspace_id()
.is_some_and(|id| !id.is_empty())
&& let Some(backend) = &feature.content_backend
{
registry.mount_subtree(crate::wip::WipSubtreeMount {
root: CONTENT_ROOT.into(),
provider: Arc::new(super::content::ConfigProvider {
feature: feature.clone(),
backend: backend.clone(),
}),
})?;
}
Ok(())
}
/// Shared root registration; no OS session or duplicate attachment inventory.
pub fn mount_workspace_config_attach_wip(
registry: &mut WipMountRegistry,
feature: &WorkspaceConfigFeature,
) -> Result<(), WipMountError> {
if !feature.wip_mode || feature.client.workspace_id().is_none_or(str::is_empty) {
return Ok(());
}
registry.allocate_namespace("workspace-config", "workspace-config")?;
registry.mount(attach_projection(feature))?;
Ok(())
}
pub(super) fn attach_projection(feature: &WorkspaceConfigFeature) -> WipProjection {
let interface = "yoi.workspace-config/attachment/v1";
let descriptor = InterfaceDescriptor {
format: INTERFACE_FORMAT_V1.into(),
documentation: Some(Documentation {
summary: "Attach this Workspace's logical configuration".into(),
details: Some("Uses the same Backend ledger operation as /workspace-config. Default access is the effective grant; explicit read_write is never downgraded. No host path, Workdir session or Bash capability is created.".into()),
}),
types: Vec::new(),
operations: vec![OperationDeclaration {
name: "attach".into(),
documentation: Some(Documentation {
summary: "Attach or resolve the existing workspace-config connection".into(),
details: None,
}),
parameters: vec![ParameterDeclaration {
name: "access".into(),
required: false,
documentation: None,
r#type: TypeExpr::String,
}],
returns: ReturnDeclaration {
documentation: None,
r#type: TypeExpr::Json,
},
}],
};
let interface_validator = Some(Sha256::digest(format!("{descriptor:?}")).to_vec());
WipProjection {
route: CONTENT_ROOT.into(),
capability: "workspace-config:attachment".into(),
kind: WipProjectionKind::Native,
object: Object {
name: ATTACHMENT_ALIAS.into(),
description: Some("Logical Workspace configuration attachment entrance".into()),
interfaces: vec![interface.into()],
r#ref: None,
// Attaching does not modify config content. The content subtree will
// supply its own Backend validators for file/directory Operations.
validator: None,
},
interface: interface.into(),
descriptor,
interface_validator,
handler: Arc::new(AttachHandler(feature.clone())),
}
}
struct AttachHandler(WorkspaceConfigFeature);
#[async_trait]
impl WipOperationHandler for AttachHandler {
fn is_visible(&self) -> bool {
self.0.is_available() && self.0.validate_access(None).is_ok()
}
fn operation_available(&self, operation: &str) -> bool {
operation == "attach" && self.0.validate_access(None).is_ok()
}
fn contextual_interface(&self) -> bool {
true
}
async fn call(
&self,
operation: &str,
arguments: &BTreeMap<String, Value>,
context: WipCallContext,
) -> Result<WipOperationOutput, WipOperationError> {
if operation != "attach" {
return Err(failure(
ProtocolErrorCode::OperationNotFound,
"Unknown config attachment operation",
));
}
if arguments.keys().any(|key| key != "access") {
return Err(map_error(ConfigAttachError::InvalidRequest));
}
let access = match arguments.get("access") {
None => requested_access(None),
Some(Value::String(value)) => requested_access(Some(value)),
_ => Err(ConfigAttachError::InvalidRequest),
}
.map_err(map_error)?;
// Host execution identity, never a model-supplied source or grant.
let key = format!("wip:{}", context.execution.execution_id());
let result = self.0.attach(&key, access).await.map_err(map_error)?;
let value = json_to_wip(&json!({
"alias": ATTACHMENT_ALIAS,
"purpose": "Workspace configuration",
"access": result.access.as_str(),
"connection_id": result.connection_id,
"attachment_path": format!("/workdir-attachments/{}", encode_identity(&result.connection_id)),
"content_path": CONTENT_ROOT,
"already_attached": result.already_attached,
})).map_err(|_| WipOperationError::OutcomeUnknown("Completed config attachment response unavailable".into()))?;
Ok(WipOperationOutput::native(value))
}
}
fn failure(code: ProtocolErrorCode, message: &str) -> WipOperationError {
WipOperationError::Protocol(ProtocolError {
code,
message: message.into(),
})
}
fn map_error(error: ConfigAttachError) -> WipOperationError {
match error {
ConfigAttachError::InvalidRequest => {
failure(ProtocolErrorCode::InvalidArguments, error.message())
}
ConfigAttachError::Denied => failure(ProtocolErrorCode::PermissionDenied, error.message()),
ConfigAttachError::Unavailable => {
failure(ProtocolErrorCode::ResourceLimitExceeded, error.message())
}
ConfigAttachError::OutcomeUnknown => {
WipOperationError::OutcomeUnknown(error.message().into())
}
}
}
@@ -0,0 +1,334 @@
//! Backend-independent preparation tests; not config persistence/E2E evidence.
use super::*;
use std::collections::BTreeMap;
use std::sync::Mutex;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use agen::tool::ToolExecutionContext;
use protocol::InvocationArgumentValue;
use wip_protocol::{ProtocolErrorCode, Value};
use crate::feature::{FeatureInvocationRegistry, FeatureRegistryBuilder};
use crate::hook::HookRegistryBuilder;
use crate::wip::{WipCallContext, WipMountRegistry, WipOperationError};
use crate::worker::{WorkspaceClientError, WorkspaceRequest, WorkspaceResponse};
#[derive(Debug)]
struct Client {
workspace: Option<&'static str>,
available: bool,
}
impl WorkspaceClient for Client {
fn workspace_id(&self) -> Option<&str> {
self.workspace
}
fn kind(&self) -> &str {
"test"
}
fn is_available(&self) -> bool {
self.available
}
fn execute(&self, _: WorkspaceRequest) -> Result<WorkspaceResponse, WorkspaceClientError> {
panic!("scaffolding must not invent Backend endpoints");
}
}
struct Backend {
granted: AtomicBool,
read_only: AtomicBool,
unknown: AtomicBool,
calls: AtomicUsize,
keys: Mutex<Vec<String>>,
active: Mutex<Option<ConfigAttachment>>,
}
impl Backend {
fn new() -> Self {
Self {
granted: AtomicBool::new(true),
read_only: AtomicBool::new(false),
unknown: AtomicBool::new(false),
calls: AtomicUsize::new(0),
keys: Mutex::new(Vec::new()),
active: Mutex::new(None),
}
}
}
#[async_trait]
impl WorkspaceConfigAttachmentBackend for Backend {
fn is_available(&self) -> bool {
self.granted.load(Ordering::SeqCst)
}
fn validate_access(&self, access: Option<ConfigAccess>) -> Result<(), ConfigAttachError> {
if !self.granted.load(Ordering::SeqCst)
|| (self.read_only.load(Ordering::SeqCst) && access == Some(ConfigAccess::ReadWrite))
{
return Err(ConfigAttachError::Denied);
}
Ok(())
}
async fn attach(
&self,
key: &str,
access: Option<ConfigAccess>,
) -> Result<ConfigAttachment, ConfigAttachError> {
self.validate_access(access)?;
self.calls.fetch_add(1, Ordering::SeqCst);
self.keys.lock().unwrap().push(key.into());
if self.unknown.load(Ordering::SeqCst) {
return Err(ConfigAttachError::OutcomeUnknown);
}
let mut active = self.active.lock().unwrap();
let already_attached = active.is_some();
let access = access.unwrap_or(if self.read_only.load(Ordering::SeqCst) {
ConfigAccess::ReadOnly
} else {
ConfigAccess::ReadWrite
});
let result = ConfigAttachment {
connection_id: "connection-1".into(),
access,
already_attached,
};
*active = Some(result.clone());
Ok(result)
}
}
fn feature(
backend: Arc<Backend>,
wip_mode: bool,
workspace: Option<&'static str>,
) -> WorkspaceConfigFeature {
WorkspaceConfigFeature::new(
Arc::new(Client {
workspace,
available: true,
}),
backend,
wip_mode,
)
}
fn registry(feature: WorkspaceConfigFeature) -> FeatureInvocationRegistry {
let mut tools = Vec::new();
let report = FeatureRegistryBuilder::new()
.with_module(feature)
.install_into_pending(&mut tools, &mut HookRegistryBuilder::default());
assert!(!report.has_errors());
assert!(tools.is_empty(), "no config compatibility tools");
report.chat_invocations
}
fn invocation(key: &str, access: Option<&str>) -> FeatureInvocation {
FeatureInvocation {
invocation_id: key.into(),
identity: FeatureInvocationIdentity(INVOCATION_ID.into()),
name: FEATURE_ID.into(),
arguments: access
.into_iter()
.map(|value| InvocationArgumentValue {
name: "access".into(),
value: InvocationValue::String(value.into()),
})
.collect(),
}
}
fn context() -> WipCallContext {
WipCallContext {
execution: ToolExecutionContext::direct(),
security_context: "test-worker".into(),
}
}
#[tokio::test]
async fn workspace_config_common_completion_selection_and_preparation_use_backend_ledger() {
let backend = Arc::new(Backend::new());
let registry = registry(feature(backend.clone(), true, Some("workspace-1")));
let completion = registry.feature_completions("workspace-c");
assert_eq!(completion.len(), 1);
let descriptor = completion[0].invocation.as_ref().unwrap();
assert_eq!(descriptor.identity.0, INVOCATION_ID);
assert!(descriptor.client_adapter.is_none());
assert_eq!(
registry
.argument_completions(&descriptor.identity, "access", "read_")
.await
.into_iter()
.map(|entry| entry.value)
.collect::<Vec<_>>(),
["read_only", "read_write"]
);
let first = registry
.invoke(&invocation("selected-1", None))
.await
.unwrap();
assert_eq!(first.status, FeatureInvocationStatus::Succeeded);
assert!(first.message.contains("read_write"));
assert!(first.context.unwrap().contains("WIP `/workspace-config`"));
let second = registry
.invoke(&invocation("selected-2", None))
.await
.unwrap();
assert!(second.message.starts_with("Already attached"));
assert_eq!(*backend.keys.lock().unwrap(), ["selected-1", "selected-2"]);
}
#[tokio::test]
async fn workspace_config_slash_and_wip_self_attach_share_handler_and_identity() {
let backend = Arc::new(Backend::new());
let feature = feature(backend.clone(), true, Some("workspace-1"));
let slash = registry(feature.clone());
slash.invoke(&invocation("slash-1", None)).await.unwrap();
let projection = wip::attach_projection(&feature);
assert_eq!(projection.route, CONTENT_ROOT);
assert_eq!(projection.descriptor.operations.len(), 1);
let result = projection
.handler
.call("attach", &BTreeMap::new(), context())
.await;
assert!(result.is_ok());
assert_eq!(backend.calls.load(Ordering::SeqCst), 2);
assert_eq!(
backend
.active
.lock()
.unwrap()
.as_ref()
.unwrap()
.connection_id,
"connection-1"
);
assert!(backend.keys.lock().unwrap()[1].starts_with("wip:"));
}
#[tokio::test]
async fn workspace_config_effective_access_and_explicit_read_write_do_not_silently_downgrade() {
let backend = Arc::new(Backend::new());
backend.read_only.store(true, Ordering::SeqCst);
let feature = feature(backend.clone(), true, Some("workspace-1"));
let registry = registry(feature.clone());
let result = registry
.invoke(&invocation("effective-1", None))
.await
.unwrap();
assert!(result.message.contains("read_only"));
assert!(
registry
.invoke(&invocation("write-1", Some("read_write")))
.await
.is_err()
);
assert_eq!(backend.calls.load(Ordering::SeqCst), 1);
let args = BTreeMap::from([("access".into(), Value::String("read_write".into()))]);
assert!(
matches!(wip::attach_projection(&feature).handler.call("attach", &args, context()).await,
Err(WipOperationError::Protocol(error)) if error.code == ProtocolErrorCode::PermissionDenied)
);
assert_eq!(backend.calls.load(Ordering::SeqCst), 1);
}
#[tokio::test]
async fn workspace_config_old_handler_and_replay_validation_recheck_revocation() {
let backend = Arc::new(Backend::new());
let feature = feature(backend.clone(), true, Some("workspace-1"));
let old = wip::attach_projection(&feature);
let registry = registry(feature);
let request = invocation("replay-1", None);
registry.invoke(&request).await.unwrap();
backend.granted.store(false, Ordering::SeqCst);
assert!(registry.feature_completions("").is_empty());
assert!(registry.validate(&request).is_err());
assert!(!old.handler.is_visible());
assert!(
matches!(old.handler.call("attach", &BTreeMap::new(), context()).await,
Err(WipOperationError::Protocol(error)) if error.code == ProtocolErrorCode::PermissionDenied)
);
assert_eq!(backend.calls.load(Ordering::SeqCst), 1);
}
#[tokio::test]
async fn workspace_config_unknown_attach_is_not_retried_and_never_reports_success() {
let backend = Arc::new(Backend::new());
backend.unknown.store(true, Ordering::SeqCst);
let feature = feature(backend.clone(), true, Some("workspace-1"));
let registry = registry(feature.clone());
let error = registry
.invoke(&invocation("unknown-1", None))
.await
.unwrap_err();
assert!(error.outcome_unknown);
assert_eq!(backend.calls.load(Ordering::SeqCst), 1);
assert!(matches!(
wip::attach_projection(&feature)
.handler
.call("attach", &BTreeMap::new(), context())
.await,
Err(WipOperationError::OutcomeUnknown(_))
));
assert_eq!(backend.calls.load(Ordering::SeqCst), 2);
}
#[tokio::test]
async fn workspace_config_non_wip_and_missing_workspace_have_no_fallback() {
for (wip_mode, workspace) in [(false, Some("workspace-1")), (true, None), (true, Some(""))] {
let backend = Arc::new(Backend::new());
let feature = feature(backend.clone(), wip_mode, workspace);
assert!(registry(feature.clone()).descriptors().is_empty());
let mut mounts = WipMountRegistry::new();
wip::mount_workspace_config_attach_wip(&mut mounts, &feature).unwrap();
assert_eq!(mounts.routes().count(), 0);
assert_eq!(
feature.attach("not-enabled", None).await,
Err(ConfigAttachError::Denied)
);
assert_eq!(backend.calls.load(Ordering::SeqCst), 0);
}
}
#[tokio::test]
async fn workspace_config_rejects_unstructured_authority_and_invalid_access_before_attach() {
let backend = Arc::new(Backend::new());
let feature = feature(backend.clone(), true, Some("workspace-1"));
let registry = registry(feature.clone());
assert!(
registry
.invoke(&invocation("bad-1", Some("admin")))
.await
.is_err()
);
let projection = wip::attach_projection(&feature);
let args = BTreeMap::from([("workspace_id".into(), Value::String("another".into()))]);
assert!(
matches!(projection.handler.call("attach", &args, context()).await,
Err(WipOperationError::Protocol(error)) if error.code == ProtocolErrorCode::InvalidArguments)
);
let mut bad = invocation("bad-2", None);
bad.arguments.push(InvocationArgumentValue {
name: "access".into(),
value: InvocationValue::String("true".into()),
});
assert!(registry.invoke(&bad).await.is_err());
assert_eq!(backend.calls.load(Ordering::SeqCst), 0);
}
#[test]
fn workspace_config_controller_gate_requires_configured_workspace_wip_and_not_a_grant() {
for enabled in [false, true] {
for wip in [false, true] {
for workspace in [None, Some(""), Some("workspace-1"), Some("bad\nworkspace")] {
for available in [false, true] {
let client = Arc::new(Client {
workspace,
available,
});
let expected = enabled && wip && available && workspace == Some("workspace-1");
// Registration needs identity/mode, but performs no grant or
// filesystem mutation. Runtime authorization is still Backend's.
assert_eq!(
WorkspaceConfigFeature::configured(client, enabled, wip).is_some(),
expected
);
}
}
}
}
}
+57 -9
View File
@@ -241,6 +241,13 @@ pub struct WipDynamicMount {
/// this resolves both files and directories and enumerates authorized children.
#[async_trait]
pub trait WipSubtreeProvider: Send + Sync {
/// Provider-specific observation bounds, capped by the Host global bounds.
fn max_depth(&self) -> u32 {
32
}
fn max_nodes(&self) -> usize {
1024
}
async fn projection(&self, path: &str) -> Result<Option<WipProjection>, ProtocolError>;
async fn children(&self, path: &str) -> Result<Vec<String>, ProtocolError>;
}
@@ -972,7 +979,7 @@ fn descriptor_validator(descriptor: &InterfaceDescriptor) -> Vec<u8> {
digest.finalize().to_vec()
}
fn contextual_reference(base: &str, path: &str) -> String {
pub(crate) fn contextual_reference(base: &str, path: &str) -> String {
let encoded: String = path
.as_bytes()
.iter()
@@ -1172,6 +1179,15 @@ impl WipHost {
Box<dyn std::future::Future<Output = Result<ObjectObservation, ProtocolError>> + Send + 'a>,
> {
Box::pin(async move {
if let Some(mount) = self.subtree(path) {
if depth > mount.provider.max_depth() {
return Err(protocol_error(
ProtocolErrorCode::ResourceLimitExceeded,
"observation exceeds provider depth limit",
));
}
*remaining = (*remaining).min(mount.provider.max_nodes());
}
if *remaining == 0 {
return Err(protocol_error(
ProtocolErrorCode::ResourceLimitExceeded,
@@ -1820,6 +1836,19 @@ pub struct WipRuntime {
}
impl WipRuntime {
/// Build an isolated Client/Host endpoint from native application mounts.
/// Useful to embedders and in-process integration tests without an LLM Engine.
pub fn from_mounts(
registry: WipMountRegistry,
security_context: String,
) -> Result<Self, String> {
Self::new(
WipHost::new(registry),
SecurityContext::new(security_context),
0,
)
}
fn new(
host: WipHost,
security_context: SecurityContext,
@@ -1874,7 +1903,7 @@ impl WipRuntime {
Ok(())
}
async fn discover(
pub async fn discover(
&self,
path: String,
depth: u32,
@@ -1899,10 +1928,11 @@ impl WipRuntime {
.fetch_add(1, Ordering::Relaxed);
let response = self.dispatch_retrieval(&prepared.request).await?;
let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner());
state
let completion = state
.client
.complete(prepared.id, response)
.map_err(client_tool_error)?;
check_retrieval_completion(completion)?;
}
let state = self.state.lock().unwrap_or_else(|error| error.into_inner());
let values = state
@@ -1926,7 +1956,7 @@ impl WipRuntime {
))
}
async fn inspect(&self, interface: String, refresh: bool) -> Result<ToolOutput, ToolError> {
pub async fn inspect(&self, interface: String, refresh: bool) -> Result<ToolOutput, ToolError> {
let prepared = {
let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner());
let session = state.session.clone();
@@ -1946,10 +1976,11 @@ impl WipRuntime {
.fetch_add(1, Ordering::Relaxed);
let response = self.dispatch_retrieval(&prepared.request).await?;
let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner());
state
let completion = state
.client
.complete(prepared.id, response)
.map_err(client_tool_error)?;
check_retrieval_completion(completion)?;
}
let state = self.state.lock().unwrap_or_else(|error| error.into_inner());
let observed = state
@@ -1968,7 +1999,7 @@ impl WipRuntime {
))
}
async fn call(
pub async fn call(
&self,
path: String,
interface: String,
@@ -2752,6 +2783,19 @@ fn tool_output_value(output: &ToolOutput) -> Value {
Value::Record(record)
}
fn check_retrieval_completion(completion: Completion) -> Result<(), ToolError> {
match completion {
Completion::Object(_) | Completion::Interface(_) => Ok(()),
Completion::ProtocolFailure(error) => Err(ToolError::ExecutionFailed(error.to_string())),
Completion::StaleResponseRejected(_) => Err(ToolError::ExecutionFailed(
"WIP retrieval superseded; refresh observations".into(),
)),
Completion::Call(_) => Err(ToolError::Internal(
"WIP retrieval returned a call result".into(),
)),
}
}
fn render_call_completion(completion: Completion) -> Result<ToolOutput, ToolError> {
let Completion::Call(record) = completion else {
return Err(ToolError::Internal(
@@ -4142,9 +4186,13 @@ mod tests {
.unwrap();
owner.visible.store(false, Ordering::SeqCst);
assert!(runtime.host.fetch_interface(&interface).is_err());
let inspected = runtime.inspect(interface.clone(), true).await.unwrap();
let inspected: Json = serde_json::from_str(inspected.content.as_deref().unwrap()).unwrap();
assert_eq!(inspected["state"], "error");
let error = runtime.inspect(interface.clone(), true).await.unwrap_err();
assert!(error.to_string().contains("InterfaceNotFound"));
{
let state = runtime.state.lock().unwrap();
let observed = state.client.interface(&state.session, &interface).unwrap();
assert!(matches!(observed.state, ObservationState::Error(_)));
}
assert!(
runtime
.call(