From 4a05aad508222ba123403323e0307f4c3f487a9c Mon Sep 17 00:00:00 2001 From: Hare Date: Sun, 4 Oct 2026 00:26:15 +0900 Subject: [PATCH] feat: add explicit Worker WIP mode --- Cargo.lock | 220 ++- Cargo.toml | 5 + crates/manifest/Cargo.toml | 1 + crates/manifest/src/config.rs | 35 +- crates/manifest/src/lib.rs | 15 + crates/manifest/src/profile.rs | 23 +- crates/worker/Cargo.toml | 6 + crates/worker/src/controller.rs | 17 + crates/worker/src/internal_worker.rs | 6 + crates/worker/src/lib.rs | 1 + crates/worker/src/permission.rs | 31 +- crates/worker/src/spawn/tool.rs | 7 +- crates/worker/src/wip.rs | 2087 ++++++++++++++++++++++++++ docs/README.md | 1 + docs/design/worker-wip-mode.md | 109 ++ 15 files changed, 2546 insertions(+), 18 deletions(-) create mode 100644 crates/worker/src/wip.rs create mode 100644 docs/design/worker-wip-mode.md diff --git a/Cargo.lock b/Cargo.lock index 1afc18527..ed23c579d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -88,6 +88,20 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "ahash" +version = "0.8.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a15f179cd60c4584b8a8c596927aadc462e27f2ca70c04e0071964a73ba7a75" +dependencies = [ + "cfg-if", + "getrandom 0.3.4", + "once_cell", + "serde", + "version_check", + "zerocopy", +] + [[package]] name = "aho-corasick" version = "1.1.4" @@ -438,7 +452,16 @@ version = "0.5.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0700ddab506f33b20a03b13996eccd309a48e5ff77d0d95926aa0210fb4e95f1" dependencies = [ - "bit-vec", + "bit-vec 0.6.3", +] + +[[package]] +name = "bit-set" +version = "0.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "08807e080ed7f9d5433fa9b275196cfc35414f66a0c79d864dc51a0d825231a3" +dependencies = [ + "bit-vec 0.8.0", ] [[package]] @@ -447,6 +470,12 @@ version = "0.6.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "349f9b6a179ed607305526ca489b34ad0a41aed5f7980fa90eb03160b69598fb" +[[package]] +name = "bit-vec" +version = "0.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5e764a1d40d510daf35e07be9eb06e75770908c27d411ee6c92109c9840eaaf7" + [[package]] name = "bitflags" version = "1.3.2" @@ -496,6 +525,12 @@ dependencies = [ "cipher", ] +[[package]] +name = "borrow-or-share" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc0b364ead1874514c8c2855ab558056ebfeb775653e7ae45ff72f28f8f3166c" + [[package]] name = "bstr" version = "1.12.1" @@ -1249,6 +1284,15 @@ dependencies = [ "zeroize", ] +[[package]] +name = "email_address" +version = "0.2.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e079f19b08ca6239f47f8ba8509c11cf3ea30095831f7fed61441475edd8c449" +dependencies = [ + "serde", +] + [[package]] name = "encoding_rs" version = "0.8.35" @@ -1330,10 +1374,21 @@ version = "0.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b95f7c0680e4142284cf8b22c14a476e87d61b004a3a0861872b32ef7ead40a2" dependencies = [ - "bit-set", + "bit-set 0.5.3", "regex", ] +[[package]] +name = "fancy-regex" +version = "0.16.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "998b056554fbe42e03ae0e152895cd1a7e1002aec800fdc6635d20270260c46f" +dependencies = [ + "bit-set 0.8.0", + "regex-automata", + "regex-syntax", +] + [[package]] name = "fastrand" version = "2.3.0" @@ -1418,6 +1473,17 @@ dependencies = [ "thiserror 2.0.18", ] +[[package]] +name = "fluent-uri" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1918b65d96df47d3591bed19c5cca17e3fa5d0707318e4b5ef2eae01764df7e5" +dependencies = [ + "borrow-or-share", + "ref-cast", + "serde", +] + [[package]] name = "fnv" version = "1.0.7" @@ -1460,6 +1526,16 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "fraction" +version = "0.15.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e076045bb43dac435333ed5f04caf35c7463631d0dae2deb2638d94dd0a5b872" +dependencies = [ + "lazy_static", + "num", +] + [[package]] name = "fs-operation" version = "0.1.0" @@ -2278,6 +2354,32 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "jsonschema" +version = "0.33.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d46662859bc5f60a145b75f4632fbadc84e829e45df6c5de74cfc8e05acb96b5" +dependencies = [ + "ahash", + "base64 0.22.1", + "bytecount", + "email_address", + "fancy-regex 0.16.2", + "fraction", + "idna", + "itoa", + "num-cmp", + "num-traits", + "once_cell", + "percent-encoding", + "referencing", + "regex", + "regex-syntax", + "serde", + "serde_json", + "uuid-simd", +] + [[package]] name = "kasuari" version = "0.4.12" @@ -2730,6 +2832,20 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "num" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "35bd024e8b2ff75562e5f34e7f4905839deb4b22955ef5e73d2fea1b9813cb23" +dependencies = [ + "num-bigint", + "num-complex", + "num-integer", + "num-iter", + "num-rational", + "num-traits", +] + [[package]] name = "num-bigint" version = "0.4.8" @@ -2756,6 +2872,21 @@ dependencies = [ "zeroize", ] +[[package]] +name = "num-cmp" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63335b2e2c34fae2fb0aa2cecfd9f0832a1e24b3b32ecec612c3426d46dc8aaa" + +[[package]] +name = "num-complex" +version = "0.4.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "73f88a1307638156682bada9d7604135552957b7818057dcef22705b4d509495" +dependencies = [ + "num-traits", +] + [[package]] name = "num-conv" version = "0.2.1" @@ -2792,6 +2923,17 @@ dependencies = [ "num-traits", ] +[[package]] +name = "num-rational" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f83d14da390562dca69fc84082e73e548e1ad308d24accdedd2720017cb37824" +dependencies = [ + "num-bigint", + "num-integer", + "num-traits", +] + [[package]] name = "num-traits" version = "0.2.19" @@ -2925,6 +3067,12 @@ dependencies = [ "num-traits", ] +[[package]] +name = "outref" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1a80800c0488c3a21695ea981a54918fbb37abf04f4d0720c453632255e2ff0e" + [[package]] name = "p256" version = "0.13.2" @@ -3628,6 +3776,20 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "referencing" +version = "0.33.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9e9c261f7ce75418b3beadfb3f0eb1299fe8eb9640deba45ffa2cb783098697d" +dependencies = [ + "ahash", + "fluent-uri", + "once_cell", + "parking_lot", + "percent-encoding", + "serde_json", +] + [[package]] name = "regex" version = "1.12.3" @@ -4674,7 +4836,7 @@ dependencies = [ "anyhow", "base64 0.22.1", "bitflags 2.11.0", - "fancy-regex", + "fancy-regex 0.11.0", "filedescriptor", "finl_unicode", "fixedbitset", @@ -5409,6 +5571,17 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "uuid-simd" +version = "0.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "23b082222b4f6619906941c17eb2297fff4c2fb96cb60164170522942a200bd8" +dependencies = [ + "outref", + "uuid", + "vsimd", +] + [[package]] name = "valuable" version = "0.1.1" @@ -5427,6 +5600,12 @@ version = "0.9.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" +[[package]] +name = "vsimd" +version = "0.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c3082ca00d5a5ef149bb8b555a72ae84c9c59f7250f013ac822ac2e49b19c64" + [[package]] name = "vtparse" version = "0.6.2" @@ -6063,6 +6242,36 @@ dependencies = [ "memchr", ] +[[package]] +name = "wip-client" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "73a82416411af69b7eb4ad0ed9a954df062769d3a03b3a1c59b0b4bf617f3534" +dependencies = [ + "wip-http", + "wip-protocol", +] + +[[package]] +name = "wip-http" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0ec0d32d1e56e52fb9ce925db808354a3432b3ebbd112fab8c443c9b865c58c6" +dependencies = [ + "base64 0.22.1", + "http", + "serde", + "serde_json", + "url", + "wip-protocol", +] + +[[package]] +name = "wip-protocol" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "275ffaebf544a077de7578fd3674c062a098a72c5a904cf66e934f76bd9efbab" + [[package]] name = "wiremock" version = "0.6.5" @@ -6210,7 +6419,9 @@ dependencies = [ "fs4", "futures", "futures-util", + "http", "include_dir", + "jsonschema", "libc", "manifest", "mcp", @@ -6237,6 +6448,9 @@ dependencies = [ "tracing", "tungstenite 0.28.0", "uuid", + "wip-client", + "wip-http", + "wip-protocol", "workdir", ] diff --git a/Cargo.toml b/Cargo.toml index ff5061333..b04334dea 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -73,6 +73,7 @@ default-members = [ [workspace.package] edition = "2024" license = "MIT" +rust-version = "1.88" [profile.dev] debug = "line-tables-only" @@ -124,6 +125,7 @@ decodal-language-service = "0.4.0" decodal-language-tools = "0.4.0" fs4 = "0.13" futures = "0.3" +http = "1" libc = "0.2" schemars = "1.2" serde = "1.0" @@ -144,5 +146,8 @@ tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] } url = "2.5" uuid = "1.23" +wip-client = "0.1.0" +wip-http = "0.1.0" +wip-protocol = "0.1.0" zeroize = "1" webauthn-rs = { version = "0.5.2", features = ["danger-allow-state-serialisation", "danger-credential-internals"] } diff --git a/crates/manifest/Cargo.toml b/crates/manifest/Cargo.toml index f41f4b20f..6f6fe2275 100644 --- a/crates/manifest/Cargo.toml +++ b/crates/manifest/Cargo.toml @@ -3,6 +3,7 @@ name = "manifest" version = "0.1.0" edition.workspace = true license.workspace = true +rust-version.workspace = true [dependencies] arc-swap = "1" diff --git a/crates/manifest/src/config.rs b/crates/manifest/src/config.rs index d4e135111..910eacfe1 100644 --- a/crates/manifest/src/config.rs +++ b/crates/manifest/src/config.rs @@ -590,6 +590,8 @@ impl From for FeatureConfigPartial { pub struct WorkerMetaConfig { #[serde(default)] pub name: Option, + #[serde(default)] + pub mode: Option, } #[derive(Debug, Clone, Default, Serialize, Deserialize)] @@ -925,6 +927,7 @@ impl WorkerMetaConfig { fn merge(self, upper: Self) -> Self { Self { name: upper.name.or(self.name), + mode: upper.mode.or(self.mode), } } } @@ -1356,7 +1359,10 @@ impl TryFrom for WorkerManifest { validate_mcp_config(&cfg.mcp)?; Ok(WorkerManifest { - worker: WorkerMeta { name }, + worker: WorkerMeta { + name, + mode: cfg.worker.mode.unwrap_or_default(), + }, model: cfg.model, engine, scope: cfg.scope, @@ -1391,6 +1397,7 @@ mod tests { WorkerManifestConfig { worker: WorkerMetaConfig { name: Some("test".into()), + mode: None, }, model: ModelManifest { scheme: Some(SchemeKind::Anthropic), @@ -1498,10 +1505,31 @@ mod tests { fn resolve_minimal_succeeds() { let manifest: WorkerManifest = minimal_valid().try_into().unwrap(); assert_eq!(manifest.worker.name, "test"); + assert_eq!(manifest.worker.mode, crate::WorkerMode::Tools); assert_eq!(manifest.model.scheme, Some(SchemeKind::Anthropic)); assert!(manifest.permissions.is_none()); } + #[test] + fn worker_mode_is_explicit_opt_in_and_persists_in_snapshots() { + let mut configured = minimal_valid(); + configured.worker.mode = Some(crate::WorkerMode::Wip); + let manifest: WorkerManifest = configured.try_into().unwrap(); + assert_eq!(manifest.worker.mode, crate::WorkerMode::Wip); + + let snapshot = crate::write_persisted_worker_manifest_snapshot(&manifest).unwrap(); + let restored = crate::read_persisted_worker_manifest_snapshot(snapshot.clone()).unwrap(); + assert_eq!(restored.worker.mode, crate::WorkerMode::Wip); + + let mut legacy = snapshot; + legacy["manifest"]["worker"] + .as_object_mut() + .unwrap() + .remove("mode"); + let restored = crate::read_persisted_worker_manifest_snapshot(legacy).unwrap(); + assert_eq!(restored.worker.mode, crate::WorkerMode::Tools); + } + #[test] fn resolve_mcp_stdio_config_preserves_explicit_policy() { let mut cfg = minimal_valid(); @@ -1784,6 +1812,7 @@ mod tests { let lower = WorkerManifestConfig { worker: WorkerMetaConfig { name: Some("lower".into()), + mode: None, }, model: ModelManifest { model_id: Some("lower-model".into()), @@ -1794,6 +1823,7 @@ mod tests { let upper = WorkerManifestConfig { worker: WorkerMetaConfig { name: Some("upper".into()), + mode: None, }, ..Default::default() }; @@ -2248,6 +2278,7 @@ enabled = false .merge(WorkerManifestConfig { worker: WorkerMetaConfig { name: Some("feature-test".into()), + mode: None, }, model: ModelManifest { scheme: Some(SchemeKind::Anthropic), @@ -2351,6 +2382,7 @@ enabled = true .merge(WorkerManifestConfig { worker: WorkerMetaConfig { name: Some("feature-merge-test".into()), + mode: None, }, model: ModelManifest { scheme: Some(SchemeKind::Anthropic), @@ -2428,6 +2460,7 @@ permission = "write" let overlay = WorkerManifestConfig { worker: WorkerMetaConfig { name: Some("x".into()), + mode: None, }, model: ModelManifest { scheme: Some(SchemeKind::Anthropic), diff --git a/crates/manifest/src/lib.rs b/crates/manifest/src/lib.rs index 77ee30896..62634b34f 100644 --- a/crates/manifest/src/lib.rs +++ b/crates/manifest/src/lib.rs @@ -670,6 +670,21 @@ pub struct WorkspaceMemorySettingsSnapshot { #[derive(Debug, Clone, Serialize, Deserialize)] pub struct WorkerMeta { pub name: String, + /// Model-facing capability surface. Existing manifests default to ordinary tools. + #[serde(default)] + pub mode: WorkerMode, +} + +/// Model-facing Worker capability surface. +/// +/// WIP is explicit opt-in: it replaces individual model-visible tools with the +/// discover/inspect/call Worldspace client while retaining the same host authority. +#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum WorkerMode { + #[default] + Tools, + Wip, } /// When the engine may begin executing provider-confirmed tool calls. diff --git a/crates/manifest/src/profile.rs b/crates/manifest/src/profile.rs index 3ba5c3e1b..c71fe2e7b 100644 --- a/crates/manifest/src/profile.rs +++ b/crates/manifest/src/profile.rs @@ -626,6 +626,7 @@ fn resolve_profile_value( let config = WorkerManifestConfig { worker: WorkerMetaConfig { name: Some(worker_name), + mode: profile.worker.and_then(|worker| worker.mode), }, model: profile.model.unwrap_or_default(), engine: profile.engine.unwrap_or_default(), @@ -668,6 +669,8 @@ struct ProfileConfig { #[serde(default)] description: Option, #[serde(default)] + worker: Option, + #[serde(default)] model: Option, #[serde(default)] engine: Option, @@ -691,6 +694,13 @@ struct ProfileConfig { skills: Option, } +#[derive(Debug, Default, Deserialize)] +#[serde(deny_unknown_fields)] +struct ProfileWorkerConfig { + #[serde(default)] + mode: Option, +} + #[derive(Debug, Clone, Deserialize)] #[serde(untagged)] enum ProfileScopeConfig { @@ -864,8 +874,15 @@ fn reject_manifest_shaped_profile(value: &serde_json::Value) -> Result<(), Profi ))); } } - if map.contains_key("worker") { - return Err(ProfileError::InvalidProfile("field `worker` is runtime-bound and is not allowed in reusable Profiles; pass the Worker name via CLI/TUI runtime inputs".into())); + if let Some(worker) = map.get("worker") { + let Some(worker) = worker.as_object() else { + return Err(ProfileError::InvalidProfile( + "field `worker` must be an object containing only reusable `mode` policy".into(), + )); + }; + if worker.contains_key("name") { + return Err(ProfileError::InvalidProfile("field `worker.name` is runtime-bound and is not allowed in reusable Profiles; pass the Worker name via CLI/TUI runtime inputs".into())); + } } if let Some(scope) = map.get("scope").and_then(|v| v.as_object()) { for key in ["allow", "deny"] { @@ -1691,6 +1708,7 @@ mod tests { r#" slug = "coder" scope = "workspace_read" +worker = { mode = "wip" } [model] scheme = "anthropic" @@ -1710,6 +1728,7 @@ reasoning = "high" ) .unwrap(); assert_eq!(resolved.manifest.worker.name, "runtime-worker"); + assert_eq!(resolved.manifest.worker.mode, crate::WorkerMode::Wip); assert_eq!(resolved.manifest.model.scheme, Some(SchemeKind::Anthropic)); assert_eq!( resolved.manifest.engine.reasoning, diff --git a/crates/worker/Cargo.toml b/crates/worker/Cargo.toml index dacab4226..250229187 100644 --- a/crates/worker/Cargo.toml +++ b/crates/worker/Cargo.toml @@ -3,6 +3,7 @@ name = "worker" version = "0.1.0" edition.workspace = true license.workspace = true +rust-version.workspace = true autobins = false [features] @@ -34,6 +35,7 @@ chrono = "0.4" config-source = { path = "../config-source" } include_dir = "0.7.4" fs4 = { workspace = true, features = ["sync"] } +http = { workspace = true } flow = { path = "../flow" } fs-operation = { workspace = true } libc = { workspace = true } @@ -46,6 +48,10 @@ arc-swap = "1.9.1" tungstenite = { version = "0.28.0", default-features = false, features = ["handshake", "native-tls", "url"] } tokio-tungstenite = { version = "0.28.0", default-features = false, features = ["native-tls", "connect"] } futures-util = { version = "0.3", features = ["sink"] } +jsonschema = { version = "0.33.0", default-features = false } +wip-client = { workspace = true } +wip-http = { workspace = true } +wip-protocol = { workspace = true } [dev-dependencies] dotenv = "0.15.0" diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index 87e84f9c0..5632bdd1a 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -1631,6 +1631,23 @@ where if let Some((resident_summary, system_prompt_override)) = feature_prompt_contribution { worker.install_system_prompt_contribution(resident_summary, system_prompt_override); } + if worker.manifest().worker.mode == manifest::WorkerMode::Wip { + let permissions = worker.manifest().permissions.clone(); + let security_context = format!( + "workspace={};worker={};session={}", + worker.workspace_id().map_or("standalone", |id| id.as_str()), + worker.manifest().worker.name, + worker.session_id() + ); + crate::wip::install_wip_mode(worker.engine_mut(), permissions, security_context).map_err( + |error| { + std::io::Error::new( + std::io::ErrorKind::InvalidInput, + format!("install WIP Worker surface: {error}"), + ) + }, + )?; + } worker.attach_tracker(tracker); Ok(workdir_for_view) } diff --git a/crates/worker/src/internal_worker.rs b/crates/worker/src/internal_worker.rs index b431f2bf9..47642a2e1 100644 --- a/crates/worker/src/internal_worker.rs +++ b/crates/worker/src/internal_worker.rs @@ -123,6 +123,12 @@ where authority, } = spec; + // Feature-owned private Internal Workers use their explicitly supplied tool + // composition. They are not model-facing direct SubWorkers and must not + // inherit a parent's WIP surface (which is finalized only by the normal + // controller path); keeping Tool mode also preserves manifest permission + // hooks for these private jobs. + manifest.worker.mode = manifest::WorkerMode::Tools; // Internal identities are run-scoped and never enter the public Runtime Worker catalog. manifest.worker.name = format!("internal-{}-{}", identity.kind, identity.run_id); // Internal jobs only receive the explicitly supplied Feature set below. A diff --git a/crates/worker/src/lib.rs b/crates/worker/src/lib.rs index a5b88a79f..4bf979d92 100644 --- a/crates/worker/src/lib.rs +++ b/crates/worker/src/lib.rs @@ -20,6 +20,7 @@ pub mod shared_state; mod shutdown_after_idle; pub mod skill; pub mod spawn; +pub mod wip; mod internal_worker; mod interrupt_prep; diff --git a/crates/worker/src/permission.rs b/crates/worker/src/permission.rs index 21a9c7b54..d7af4ce33 100644 --- a/crates/worker/src/permission.rs +++ b/crates/worker/src/permission.rs @@ -21,16 +21,7 @@ impl PermissionHook { } fn action_for(&self, input: &ToolCallSummary) -> ToolPermissionAction { - let target = permission_target(&input.arguments); - self.config - .rules - .iter() - .find(|rule| { - rule.tool.eq_ignore_ascii_case(&input.tool_name) - && wildcard_match(&rule.pattern, &target) - }) - .map(|rule| rule.action) - .unwrap_or(self.config.default_action) + permission_action_for(&self.config, &input.tool_name, &input.arguments) } } @@ -39,7 +30,9 @@ impl Worker { let Some(permissions) = self.manifest().permissions.clone() else { return; }; - self.add_pre_tool_call_hook(PermissionHook::new(permissions)); + if self.manifest().worker.mode == manifest::WorkerMode::Tools { + self.add_pre_tool_call_hook(PermissionHook::new(permissions)); + } } } @@ -59,6 +52,22 @@ impl Hook for PermissionHook { } } +pub(crate) fn permission_action_for( + config: &ToolPermissionConfig, + tool_name: &str, + arguments: &Value, +) -> ToolPermissionAction { + let target = permission_target(arguments); + config + .rules + .iter() + .find(|rule| { + rule.tool.eq_ignore_ascii_case(tool_name) && wildcard_match(&rule.pattern, &target) + }) + .map(|rule| rule.action) + .unwrap_or(config.default_action) +} + fn permission_denied_message(input: &ToolCallSummary) -> String { format!( "permission denied: tool `{}` arguments matched the manifest permission policy", diff --git a/crates/worker/src/spawn/tool.rs b/crates/worker/src/spawn/tool.rs index 55db47c2a..97cf919a1 100644 --- a/crates/worker/src/spawn/tool.rs +++ b/crates/worker/src/spawn/tool.rs @@ -883,6 +883,7 @@ fn build_spawn_config_json( let config = WorkerManifestConfig { worker: WorkerMetaConfig { name: Some(name.to_string()), + mode: None, }, model: model.clone(), engine: EngineManifestConfig { @@ -957,6 +958,7 @@ fn manifest_to_reusable_config(manifest: &WorkerManifest) -> WorkerManifestConfi WorkerManifestConfig { worker: WorkerMetaConfig { name: Some(manifest.worker.name.clone()), + mode: Some(manifest.worker.mode), }, model: manifest.model.clone(), engine: EngineManifestConfig { @@ -1871,6 +1873,7 @@ enabled = false WorkerManifestConfig { worker: WorkerMetaConfig { name: Some("parent".into()), + mode: None, }, model: ModelManifest { scheme: Some(SchemeKind::Anthropic), @@ -2188,7 +2191,8 @@ max_tokens = 3333 let delegated = tmp.path().join("delegated"); std::fs::create_dir_all(&parent_deny).unwrap(); std::fs::create_dir_all(&delegated).unwrap(); - let parent = parent_manifest(&parent_root, Some(&parent_deny)); + let mut parent = parent_manifest(&parent_root, Some(&parent_deny)); + parent.worker.mode = manifest::WorkerMode::Wip; let scope = vec![abs_rule(&delegated, Permission::Read)]; let available = AvailableProfiles { registry: None, @@ -2206,6 +2210,7 @@ max_tokens = 3333 ); assert_eq!(config.worker.name.as_deref(), Some("inherited-child")); + assert_eq!(config.worker.mode, Some(manifest::WorkerMode::Wip)); assert_eq!(config.model.model_id.as_deref(), Some("parent-model")); assert_eq!(config.engine.instruction.as_deref(), Some("default")); assert_eq!(config.engine.language.as_deref(), Some("Parentish")); diff --git a/crates/worker/src/wip.rs b/crates/worker/src/wip.rs new file mode 100644 index 000000000..e8e7aadf9 --- /dev/null +++ b/crates/worker/src/wip.rs @@ -0,0 +1,2087 @@ +//! Explicit-opt-in Web Interface Protocol (WIP) Worker surface. +//! +//! The adapter intentionally keeps authority in the existing Worker host. In WIP +//! mode ordinary tools are mounted below `/tools`, removed from the model-visible +//! tool list, and invoked through three stateful discover/inspect/call tools. The +//! published WIP crates own protocol validation, HTTP encoding, Known Space, and +//! operation lifecycle state; this module owns only the in-process transport and +//! the compatibility projection into existing async `Tool` implementations. + +use std::collections::{BTreeMap, BTreeSet, HashMap, VecDeque}; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; + +use agen::Engine; +use agen::llm_client::client::LlmClient; +use agen::state::Mutable; +use agen::tool::{Tool, ToolDefinition, ToolError, ToolExecutionContext, ToolMeta, ToolOutput}; +use async_trait::async_trait; +use http::{Request, Response}; +use manifest::{ToolPermissionAction, ToolPermissionConfig}; +use serde::Deserialize; +use serde_json::{Value as Json, json}; +use sha2::{Digest, Sha256}; +use thiserror::Error; +use wip_client::{ + CallOutcome, Client, ClientLimits, Completion, DispatchedTransportFailure, ObservationState, + RequestId, SecurityContext, SessionId, +}; +use wip_http::{ + Endpoint, Limits, Route, decode_call_operation_metadata, decode_fetch_interface_request, + decode_observe_request, encode_call_operation_response, encode_fetch_interface_response, + encode_observe_response, encode_protocol_error_response, +}; +use wip_protocol::{ + CallOperationRequest, CallOperationResponse, Documentation, FetchInterfaceResponse, + INTERFACE_FORMAT_V1, InterfaceDescriptor, Object, ObjectObservation, OperationDeclaration, + ParameterDeclaration, ProtocolError, ProtocolErrorCode, ProtocolInteraction, ReturnDeclaration, + TypeExpr, Value, +}; + +use crate::permission::permission_action_for; + +const WIP_ENDPOINT: &str = "https://worker.wip.invalid/v1/"; +const WIP_ROOT: &str = "/"; +const WIP_TOOLS_ROOT: &str = "/tools"; +const MAX_AUDIT_RECORDS: usize = 128; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum WipProjectionKind { + Compatibility, + Native, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum WipMountDisposition { + Mounted, + ReplacedCompatibility, + NativeAlreadySelected, +} + +#[derive(Debug, Error, PartialEq, Eq)] +pub enum WipMountError { + #[error("invalid WIP route `{route}`: {message}")] + InvalidRoute { route: String, message: String }, + #[error("WIP route `{route}` is already owned by capability `{existing}`")] + RouteCollision { route: String, existing: String }, + #[error("WIP interface `{interface}` has conflicting descriptors")] + InterfaceCollision { interface: String }, + #[error("invalid WIP projection for `{route}`: {message}")] + InvalidProjection { route: String, message: String }, +} + +/// Runtime-neutral call context supplied to native WIP projections. +#[derive(Debug, Clone)] +pub struct WipCallContext { + pub execution: ToolExecutionContext, + pub security_context: String, +} + +/// Successful operation value. Compatibility projections can retain an ordinary +/// ToolOutput so images and pruning semantics remain on the existing Engine path. +pub struct WipOperationOutput { + pub value: Value, + tool_output: Option, +} + +impl WipOperationOutput { + pub fn native(value: Value) -> Self { + Self { + value, + tool_output: None, + } + } + + fn compatibility(value: Value, output: ToolOutput) -> Self { + Self { + value, + tool_output: Some(output), + } + } +} + +pub enum WipOperationError { + Protocol(ProtocolError), + Cancelled(ToolOutput), + Interrupted(ToolOutput), + OutcomeUnknown(String), +} + +/// Async extension point used by future native Feature projections. Host-side +/// routing and validators are resolved before this trait is called. +#[async_trait] +pub trait WipOperationHandler: Send + Sync { + async fn call( + &self, + operation: &str, + arguments: &BTreeMap, + context: WipCallContext, + ) -> Result; + + async fn cancel(&self, _context: &ToolExecutionContext) -> Result<(), ToolError> { + Ok(()) + } +} + +/// One route contribution. Native projections replace compatibility projections +/// only when both declare the same semantic capability; unrelated collisions fail. +pub struct WipProjection { + pub route: String, + pub capability: String, + pub kind: WipProjectionKind, + pub object: Object, + pub interface: String, + pub descriptor: InterfaceDescriptor, + pub interface_validator: Option>, + pub handler: Arc, +} + +struct MountedProjection { + projection: WipProjection, +} + +#[derive(Default)] +pub struct WipMountRegistry { + mounts: BTreeMap, +} + +impl WipMountRegistry { + pub fn new() -> Self { + Self::default() + } + + pub fn mount( + &mut self, + projection: WipProjection, + ) -> Result { + validate_projection(&projection)?; + if let Some(existing) = self.mounts.get(&projection.route) { + if existing.projection.capability != projection.capability { + return Err(WipMountError::RouteCollision { + route: projection.route, + existing: existing.projection.capability.clone(), + }); + } + match (existing.projection.kind, projection.kind) { + (WipProjectionKind::Compatibility, WipProjectionKind::Native) => { + self.ensure_interface_available(&projection, Some(&existing.projection.route))?; + self.mounts + .insert(projection.route.clone(), MountedProjection { projection }); + return Ok(WipMountDisposition::ReplacedCompatibility); + } + (WipProjectionKind::Native, WipProjectionKind::Compatibility) => { + return Ok(WipMountDisposition::NativeAlreadySelected); + } + _ => { + return Err(WipMountError::RouteCollision { + route: projection.route, + existing: existing.projection.capability.clone(), + }); + } + } + } + self.ensure_interface_available(&projection, None)?; + let route = projection.route.clone(); + self.mounts.insert(route, MountedProjection { projection }); + Ok(WipMountDisposition::Mounted) + } + + fn ensure_interface_available( + &self, + candidate: &WipProjection, + replacing_route: Option<&str>, + ) -> Result<(), WipMountError> { + if self.mounts.iter().any(|(route, mounted)| { + Some(route.as_str()) != replacing_route + && mounted.projection.interface == candidate.interface + && mounted.projection.descriptor != candidate.descriptor + }) { + return Err(WipMountError::InterfaceCollision { + interface: candidate.interface.clone(), + }); + } + Ok(()) + } + + pub fn routes(&self) -> impl Iterator { + self.mounts.keys().map(String::as_str) + } +} + +fn validate_projection(projection: &WipProjection) -> Result<(), WipMountError> { + wip_protocol::validate_path(&projection.route).map_err(|error| { + WipMountError::InvalidRoute { + route: projection.route.clone(), + message: error.to_string(), + } + })?; + if projection.route == WIP_ROOT { + return Err(WipMountError::InvalidRoute { + route: projection.route.clone(), + message: "the Host reserves the Worldspace root".into(), + }); + } + projection + .object + .validate_at_path(&projection.route) + .map_err(|error| WipMountError::InvalidProjection { + route: projection.route.clone(), + message: error.to_string(), + })?; + if projection.object.interfaces != [projection.interface.clone()] { + return Err(WipMountError::InvalidProjection { + route: projection.route.clone(), + message: "projected object must expose exactly its declared interface".into(), + }); + } + projection + .descriptor + .validate() + .map_err(|error| WipMountError::InvalidProjection { + route: projection.route.clone(), + message: error.to_string(), + }) +} + +struct WipHost { + registry: WipMountRegistry, + generation: Vec, +} + +impl WipHost { + fn new(registry: WipMountRegistry) -> Self { + let mut digest = Sha256::new(); + for route in registry.routes() { + digest.update(route.as_bytes()); + digest.update([0]); + } + Self { + registry, + generation: digest.finalize().to_vec(), + } + } + + fn descriptor(&self, reference: &str) -> Option<(&InterfaceDescriptor, Option<&[u8]>)> { + self.registry + .mounts + .values() + .find(|mounted| mounted.projection.interface == reference) + .map(|mounted| { + ( + &mounted.projection.descriptor, + mounted.projection.interface_validator.as_deref(), + ) + }) + } + + fn projection(&self, path: &str) -> Option<&WipProjection> { + self.registry + .mounts + .get(path) + .map(|mounted| &mounted.projection) + } + + fn object_at(&self, path: &str) -> Option { + if path == WIP_ROOT || self.is_namespace(path) { + let name = if path == WIP_ROOT { + String::new() + } else { + path.rsplit('/').next().unwrap_or_default().to_string() + }; + return Some(Object { + name, + description: Some("WIP Worldspace namespace".into()), + interfaces: Vec::new(), + r#ref: Some(format!("namespace:{path}")), + validator: Some(self.generation.clone()), + }); + } + self.projection(path) + .map(|projection| projection.object.clone()) + } + + fn is_namespace(&self, path: &str) -> bool { + let prefix = if path == WIP_ROOT { + "/".to_string() + } else { + format!("{path}/") + }; + self.registry + .mounts + .keys() + .any(|route| route.starts_with(&prefix)) + } + + fn children(&self, path: &str) -> Vec { + let prefix = if path == WIP_ROOT { + "/".to_string() + } else { + format!("{path}/") + }; + let mut children = BTreeSet::new(); + for route in self.registry.mounts.keys() { + let Some(rest) = route.strip_prefix(&prefix) else { + continue; + }; + let Some(segment) = rest.split('/').next() else { + continue; + }; + if segment.is_empty() { + continue; + } + let child = if path == WIP_ROOT { + format!("/{segment}") + } else { + format!("{path}/{segment}") + }; + children.insert(child); + } + children.into_iter().collect() + } + + fn observe(&self, path: &str, depth: u32) -> Result { + let object = self.object_at(path).ok_or_else(|| { + protocol_error(ProtocolErrorCode::NotFound, "target path is not published") + })?; + let children = if depth == 0 { + None + } else { + let mut values = Vec::new(); + for child in self.children(path) { + values.push(self.observe(&child, depth - 1)?); + } + Some(values) + }; + Ok(ObjectObservation { object, children }) + } + + fn fetch_interface(&self, reference: &str) -> Result { + let (descriptor, validator) = self.descriptor(reference).ok_or_else(|| { + protocol_error( + ProtocolErrorCode::InterfaceNotFound, + "interface is not published", + ) + })?; + Ok(FetchInterfaceResponse { + interface: reference.to_string(), + descriptor: descriptor.clone(), + validator: validator.map(<[u8]>::to_vec), + }) + } + + async fn call( + &self, + request: CallOperationRequest, + context: WipCallContext, + ) -> Result { + let projection = self.projection(&request.target.path).ok_or_else(|| { + WipOperationError::Protocol(protocol_error( + ProtocolErrorCode::NotFound, + "target path is not published", + )) + })?; + let expected_object_validator = projection.object.validator.as_deref(); + match ( + expected_object_validator, + request.target.validator.as_deref(), + ) { + (Some(_), None) => { + return Err(WipOperationError::Protocol(protocol_error( + ProtocolErrorCode::ValidatorRequired, + "object validator is required", + ))); + } + (Some(expected), Some(actual)) if expected != actual => { + return Err(WipOperationError::Protocol(protocol_error( + ProtocolErrorCode::ValidatorMismatch, + "object validator is stale", + ))); + } + _ => {} + } + if request.interface.reference != projection.interface { + return Err(WipOperationError::Protocol(protocol_error( + ProtocolErrorCode::InterfaceMismatch, + "interface is not a member of the target object", + ))); + } + match ( + projection.interface_validator.as_deref(), + request.interface.validator.as_deref(), + ) { + (Some(_), None) => { + return Err(WipOperationError::Protocol(protocol_error( + ProtocolErrorCode::InterfaceValidatorRequired, + "interface validator is required", + ))); + } + (Some(expected), Some(actual)) if expected != actual => { + return Err(WipOperationError::Protocol(protocol_error( + ProtocolErrorCode::InterfaceValidatorMismatch, + "interface validator is stale", + ))); + } + _ => {} + } + if !projection + .descriptor + .operations + .iter() + .any(|operation| operation.name == request.operation) + { + return Err(WipOperationError::Protocol(protocol_error( + ProtocolErrorCode::OperationNotFound, + "operation is not published by the selected interface", + ))); + } + projection + .descriptor + .validate_call(&request) + .map_err(|error| { + WipOperationError::Protocol(protocol_error( + ProtocolErrorCode::InvalidArguments, + error.to_string(), + )) + })?; + projection + .handler + .call(&request.operation, &request.arguments, context) + .await + } + + async fn cancel(&self, path: &str, context: &ToolExecutionContext) -> Result<(), ToolError> { + let projection = self.projection(path).ok_or_else(|| { + ToolError::InvalidArgument("WIP cancellation target is no longer published".into()) + })?; + projection.handler.cancel(context).await + } +} + +fn protocol_error(code: ProtocolErrorCode, message: impl Into) -> ProtocolError { + ProtocolError { + code, + message: message.into(), + } +} + +struct CompatibilityHandler { + name: String, + schema: Json, + tool: Arc, + permissions: Option, +} + +#[async_trait] +impl WipOperationHandler for CompatibilityHandler { + async fn call( + &self, + operation: &str, + arguments: &BTreeMap, + context: WipCallContext, + ) -> Result { + if operation != "call" { + return Err(WipOperationError::Protocol(protocol_error( + ProtocolErrorCode::OperationNotFound, + "operation is not published", + ))); + } + let Some(input) = arguments.get("input") else { + return Err(WipOperationError::Protocol(protocol_error( + ProtocolErrorCode::InvalidArguments, + "compatibility call requires `input`", + ))); + }; + let input = wip_to_json(input).map_err(|message| { + WipOperationError::Protocol(protocol_error( + ProtocolErrorCode::InvalidArguments, + message, + )) + })?; + let validator = jsonschema::validator_for(&self.schema).map_err(|error| { + WipOperationError::Protocol(protocol_error( + ProtocolErrorCode::Internal, + format!("stored tool schema is invalid: {error}"), + )) + })?; + if let Err(error) = validator.validate(&input) { + return Err(WipOperationError::Protocol(protocol_error( + ProtocolErrorCode::InvalidArguments, + format!("tool input violates its original JSON Schema: {error}"), + ))); + } + if let Some(permissions) = &self.permissions { + match permission_action_for(permissions, &self.name, &input) { + ToolPermissionAction::Allow => {} + ToolPermissionAction::Deny => { + return Err(WipOperationError::Protocol(protocol_error( + ProtocolErrorCode::PermissionDenied, + format!("permission denied for projected tool `{}`", self.name), + ))); + } + ToolPermissionAction::Ask => { + return Err(WipOperationError::Protocol(protocol_error( + ProtocolErrorCode::PermissionDenied, + format!( + "permission approval is unavailable for projected tool `{}`; denied fail-closed", + self.name + ), + ))); + } + } + } + let input_json = serde_json::to_string(&input).map_err(|error| { + WipOperationError::Protocol(protocol_error( + ProtocolErrorCode::InvalidArguments, + error.to_string(), + )) + })?; + match self.tool.execute(&input_json, context.execution).await { + Ok(output) => { + let value = tool_output_value(&output); + Ok(WipOperationOutput::compatibility(value, output)) + } + Err(ToolError::InvalidArgument(message)) => Err(WipOperationError::Protocol( + protocol_error(ProtocolErrorCode::InvalidArguments, message), + )), + Err(ToolError::Cancelled(output)) => Err(WipOperationError::Cancelled(output)), + Err(ToolError::Interrupted(output)) => Err(WipOperationError::Interrupted(output)), + Err(ToolError::StructuredConflict { code, message }) => { + Err(WipOperationError::Protocol(protocol_error( + ProtocolErrorCode::InvalidArguments, + format!("{code}: {message}"), + ))) + } + Err(ToolError::ExecutionFailed(message) | ToolError::Internal(message)) => { + Err(WipOperationError::OutcomeUnknown(message)) + } + } + } + + async fn cancel(&self, context: &ToolExecutionContext) -> Result<(), ToolError> { + self.tool.cancel_execution(context).await + } +} + +fn compatibility_projection( + meta: ToolMeta, + tool: Arc, + permissions: Option, +) -> Result { + jsonschema::validator_for(&meta.input_schema).map_err(|error| { + WipMountError::InvalidProjection { + route: format!("{WIP_TOOLS_ROOT}/{}", meta.name), + message: format!("tool JSON Schema cannot be retained: {error}"), + } + })?; + let route = format!("{WIP_TOOLS_ROOT}/{}", meta.name); + let interface = format!("yoi.tool/{}/v1", meta.name); + let schema = serde_json::to_string(&meta.input_schema).map_err(|error| { + WipMountError::InvalidProjection { + route: route.clone(), + message: error.to_string(), + } + })?; + let mut digest = Sha256::new(); + digest.update(meta.name.as_bytes()); + digest.update([0]); + digest.update(schema.as_bytes()); + let validator = digest.finalize().to_vec(); + let descriptor = InterfaceDescriptor { + format: INTERFACE_FORMAT_V1.into(), + documentation: Some(Documentation { + summary: meta.description.clone(), + details: Some(format!( + "Compatibility projection of tool `{}`. The `input` parameter preserves and is validated against this exact JSON Schema before the original tool runs:\n{}", + meta.name, schema + )), + }), + types: Vec::new(), + operations: vec![OperationDeclaration { + name: "call".into(), + documentation: Some(Documentation { + summary: format!("Invoke {} through its existing authority boundary", meta.name), + details: Some( + "Pass the original tool argument object as `input`; IDs remain ordinary values and are never interpreted as Worldspace object paths." + .into(), + ), + }), + parameters: vec![ParameterDeclaration { + name: "input".into(), + required: true, + documentation: Some(Documentation { + summary: "Original JSON tool arguments".into(), + details: Some(schema), + }), + r#type: TypeExpr::Json, + }], + returns: ReturnDeclaration { + documentation: Some(Documentation { + summary: "Existing ToolOutput projected as JSON".into(), + details: None, + }), + r#type: TypeExpr::Json, + }, + }], + }; + let handler = Arc::new(CompatibilityHandler { + name: meta.name.clone(), + schema: meta.input_schema, + tool, + permissions, + }); + Ok(WipProjection { + route: route.clone(), + capability: format!("tool:{}", meta.name), + kind: WipProjectionKind::Compatibility, + object: Object { + name: meta.name.clone(), + description: Some(meta.description), + interfaces: vec![interface.clone()], + r#ref: Some(format!("tool:{}", meta.name)), + validator: Some(validator.clone()), + }, + interface, + descriptor, + interface_validator: Some(validator), + handler, + }) +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum WipAuditOutcome { + Success, + Rejected, + Cancelled, + Interrupted, + Disconnected, + OutcomeUnknown, +} + +#[derive(Debug, Clone)] +pub struct WipAuditRecord { + pub request_id: u64, + pub path: String, + pub operation: String, + pub outcome: WipAuditOutcome, +} + +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct WipMetricsSnapshot { + pub ordinary_schema_bytes: u64, + pub wip_schema_bytes: u64, + pub discover_round_trips: u64, + pub inspect_round_trips: u64, + pub operation_round_trips: u64, + pub operation_successes: u64, +} + +#[derive(Default)] +struct WipMetrics { + ordinary_schema_bytes: AtomicU64, + wip_schema_bytes: AtomicU64, + discover_round_trips: AtomicU64, + inspect_round_trips: AtomicU64, + operation_round_trips: AtomicU64, + operation_successes: AtomicU64, +} + +impl WipMetrics { + fn snapshot(&self) -> WipMetricsSnapshot { + WipMetricsSnapshot { + ordinary_schema_bytes: self.ordinary_schema_bytes.load(Ordering::Relaxed), + wip_schema_bytes: self.wip_schema_bytes.load(Ordering::Relaxed), + discover_round_trips: self.discover_round_trips.load(Ordering::Relaxed), + inspect_round_trips: self.inspect_round_trips.load(Ordering::Relaxed), + operation_round_trips: self.operation_round_trips.load(Ordering::Relaxed), + operation_successes: self.operation_successes.load(Ordering::Relaxed), + } + } +} + +struct ClientState { + client: Client, + session: SessionId, +} + +pub struct WipRuntime { + endpoint: Endpoint, + wire_limits: Limits, + client_limits: ClientLimits, + security_context: SecurityContext, + state: Arc>, + host: Arc, + active: Arc>>, + audit: Arc>>, + metrics: Arc, +} + +impl WipRuntime { + fn new( + host: WipHost, + security_context: SecurityContext, + ordinary_schema_bytes: u64, + ) -> Result { + let endpoint = Endpoint::parse(WIP_ENDPOINT).map_err(|error| error.to_string())?; + let client_limits = ClientLimits::new(4, 2048, 2048, 64, MAX_AUDIT_RECORDS) + .map_err(|error| error.to_string())?; + let wire_limits = Limits::new(2 * 1024 * 1024, 2 * 1024 * 1024, 128) + .map_err(|error| error.to_string())?; + let mut client = Client::new(client_limits, wire_limits); + let session = client + .open_session(WIP_ENDPOINT, security_context.clone()) + .map_err(|error| error.to_string())?; + let metrics = Arc::new(WipMetrics::default()); + metrics + .ordinary_schema_bytes + .store(ordinary_schema_bytes, Ordering::Relaxed); + Ok(Self { + endpoint, + wire_limits, + client_limits, + security_context, + state: Arc::new(Mutex::new(ClientState { client, session })), + host: Arc::new(host), + active: Arc::new(Mutex::new(HashMap::new())), + audit: Arc::new(Mutex::new(VecDeque::new())), + metrics, + }) + } + + pub fn metrics(&self) -> WipMetricsSnapshot { + self.metrics.snapshot() + } + + pub fn audit(&self) -> Vec { + self.audit + .lock() + .unwrap_or_else(|error| error.into_inner()) + .iter() + .cloned() + .collect() + } + + fn reset_observations(&self) -> Result<(), ToolError> { + let mut client = Client::new(self.client_limits, self.wire_limits); + let session = client + .open_session(WIP_ENDPOINT, self.security_context.clone()) + .map_err(|error| ToolError::ExecutionFailed(error.to_string()))?; + *self.state.lock().unwrap_or_else(|error| error.into_inner()) = + ClientState { client, session }; + Ok(()) + } + + async fn discover( + &self, + path: String, + depth: u32, + refresh: bool, + ) -> Result { + let prepared = { + let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner()); + let session = state.session.clone(); + if refresh { + state + .client + .refresh_observed(&session, path.clone(), depth) + .map(Some) + } else { + state.client.ensure_observed(&session, path.clone(), depth) + } + .map_err(client_tool_error)? + }; + if let Some(prepared) = prepared { + self.metrics + .discover_round_trips + .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 + .client + .complete(prepared.id, response) + .map_err(client_tool_error)?; + } + let state = self.state.lock().unwrap_or_else(|error| error.into_inner()); + let values = state + .client + .known_space(&state.session) + .ok_or_else(|| ToolError::ExecutionFailed("WIP session disappeared".into()))?; + let visible = values + .into_iter() + .filter(|observation| { + observation.path == path || is_descendant(&path, &observation.path) + }) + .map(object_observation_json) + .collect::>(); + Ok(json_output( + format!("Observed {} WIP object(s)", visible.len()), + json!({ + "known_space": visible, + "metrics": metrics_json(self.metrics.snapshot()), + "cache_policy": "fresh observations are isolated to this Worker endpoint/security context; use refresh or reset after external authority changes" + }), + )) + } + + async fn inspect(&self, interface: String, refresh: bool) -> Result { + let prepared = { + let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner()); + let session = state.session.clone(); + if refresh { + state + .client + .prepare_interface(&session, interface.clone()) + .map(Some) + } else { + state.client.ensure_interface(&session, interface.clone()) + } + .map_err(client_tool_error)? + }; + if let Some(prepared) = prepared { + self.metrics + .inspect_round_trips + .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 + .client + .complete(prepared.id, response) + .map_err(client_tool_error)?; + } + let state = self.state.lock().unwrap_or_else(|error| error.into_inner()); + let observed = state + .client + .interface(&state.session, &interface) + .ok_or_else(|| ToolError::InvalidArgument("interface has not been observed".into()))?; + Ok(json_output( + format!("Inspected WIP interface `{interface}`"), + json!({ + "interface": interface, + "state": observation_state_name(&observed.state), + "validator": observed.validator.as_ref().map(hex), + "descriptor": observed.descriptor.as_ref().map(descriptor_json), + "metrics": metrics_json(self.metrics.snapshot()) + }), + )) + } + + async fn call( + &self, + path: String, + interface: String, + operation: String, + arguments: Json, + execution: ToolExecutionContext, + ) -> Result { + let arguments = BTreeMap::from([( + "input".to_string(), + json_to_wip(&arguments).map_err(ToolError::InvalidArgument)?, + )]); + let prepared = { + let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner()); + let session = state.session.clone(); + state + .client + .prepare_call(&session, &path, &interface, operation.clone(), arguments) + .map_err(client_tool_error)? + }; + { + let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner()); + state + .client + .mark_dispatched(prepared.id) + .map_err(client_tool_error)?; + } + self.metrics + .operation_round_trips + .fetch_add(1, Ordering::Relaxed); + let execution_id = execution.execution_id(); + self.active + .lock() + .unwrap_or_else(|error| error.into_inner()) + .insert(execution_id.clone(), path.clone()); + let mut guard = DispatchedCallGuard::new( + Arc::clone(&self.state), + Arc::clone(&self.active), + prepared.id, + execution_id, + ); + let exchange = self + .dispatch_call(&prepared.request, execution.clone()) + .await; + let (response, passthrough, audit_outcome) = match exchange { + Ok(exchange) => exchange, + Err(error) => { + guard.fail_before_response(); + self.push_audit(prepared.id, path, operation, WipAuditOutcome::Disconnected); + return Err(error); + } + }; + let completion = { + let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner()); + state.client.complete(prepared.id, response) + }; + guard.disarm(); + let completion = completion.map_err(client_tool_error)?; + self.push_audit(prepared.id, path, operation, audit_outcome); + if let Some(result) = passthrough { + if result.is_ok() { + self.metrics + .operation_successes + .fetch_add(1, Ordering::Relaxed); + } + return result; + } + render_call_completion(completion).map(|output| { + self.metrics + .operation_successes + .fetch_add(1, Ordering::Relaxed); + output + }) + } + + async fn cancel(&self, execution: &ToolExecutionContext) -> Result<(), ToolError> { + let path = self + .active + .lock() + .unwrap_or_else(|error| error.into_inner()) + .get(&execution.execution_id()) + .cloned(); + match path { + Some(path) => self.host.cancel(&path, execution).await, + None => Ok(()), + } + } + + fn push_audit( + &self, + request_id: RequestId, + path: String, + operation: String, + outcome: WipAuditOutcome, + ) { + let mut audit = self.audit.lock().unwrap_or_else(|error| error.into_inner()); + audit.push_back(WipAuditRecord { + request_id: request_id.get(), + path, + operation, + outcome, + }); + while audit.len() > MAX_AUDIT_RECORDS { + audit.pop_front(); + } + } + + async fn dispatch_retrieval( + &self, + request: &Request>, + ) -> Result>, ToolError> { + let route = self + .endpoint + .recognize(&request.uri().to_string().parse().map_err(|error| { + ToolError::Internal(format!("invalid prepared WIP URI: {error}")) + })?) + .ok_or_else(|| ToolError::Internal("prepared WIP route is not recognized".into()))?; + match route { + Route::Observe => { + let request_value = decode_observe_request(request.body(), self.wire_limits) + .map_err(|error| ToolError::Internal(error.to_string()))?; + match self.host.observe(&request_value.path, request_value.depth) { + Ok(value) => encode_observe_response(&request_value, &value, self.wire_limits), + Err(error) => encode_protocol_error_response( + ProtocolInteraction::Observe, + &error, + self.wire_limits, + ), + } + .map_err(|error| ToolError::Internal(error.to_string())) + } + Route::FetchInterface => { + let request_value = + decode_fetch_interface_request(request.body(), self.wire_limits) + .map_err(|error| ToolError::Internal(error.to_string()))?; + match self.host.fetch_interface(&request_value.interface) { + Ok(value) => { + encode_fetch_interface_response(&request_value, &value, self.wire_limits) + } + Err(error) => encode_protocol_error_response( + ProtocolInteraction::FetchInterface, + &error, + self.wire_limits, + ), + } + .map_err(|error| ToolError::Internal(error.to_string())) + } + Route::CallOperation => Err(ToolError::Internal( + "call operation used retrieval transport".into(), + )), + } + } + + async fn dispatch_call( + &self, + request: &Request>, + execution: ToolExecutionContext, + ) -> Result< + ( + Response>, + Option>, + WipAuditOutcome, + ), + ToolError, + > { + let metadata = decode_call_operation_metadata(request.body(), self.wire_limits) + .map_err(|error| ToolError::InvalidArgument(error.to_string()))?; + let descriptor = self + .host + .descriptor(&metadata.interface.reference) + .map(|(descriptor, _)| descriptor.clone()) + .ok_or_else(|| ToolError::InvalidArgument("interface is not published".into()))?; + let request_value = metadata + .decode_request(request.body(), &descriptor, self.wire_limits) + .map_err(|error| ToolError::InvalidArgument(error.to_string()))?; + let context = WipCallContext { + execution, + security_context: self.security_context.as_str().to_string(), + }; + match self.host.call(request_value.clone(), context).await { + Ok(output) => { + let response = encode_call_operation_response( + &request_value, + &descriptor, + &CallOperationResponse { + result: output.value, + validator: self + .host + .projection(&request_value.target.path) + .and_then(|projection| projection.object.validator.clone()), + }, + self.wire_limits, + ) + .map_err(|error| ToolError::Internal(error.to_string()))?; + Ok(( + response, + output.tool_output.map(Ok), + WipAuditOutcome::Success, + )) + } + Err(WipOperationError::Protocol(error)) => { + let response = encode_protocol_error_response( + ProtocolInteraction::CallOperation, + &error, + self.wire_limits, + ) + .map_err(|encode| ToolError::Internal(encode.to_string()))?; + Ok((response, None, WipAuditOutcome::Rejected)) + } + Err(WipOperationError::Cancelled(output)) => { + let response = encode_protocol_error_response( + ProtocolInteraction::CallOperation, + &protocol_error( + ProtocolErrorCode::ResourceLimitExceeded, + "operation cancelled", + ), + self.wire_limits, + ) + .map_err(|encode| ToolError::Internal(encode.to_string()))?; + Ok(( + response, + Some(Err(ToolError::Cancelled(output))), + WipAuditOutcome::Cancelled, + )) + } + Err(WipOperationError::Interrupted(output)) => { + let response = encode_protocol_error_response( + ProtocolInteraction::CallOperation, + &protocol_error( + ProtocolErrorCode::OperationOutcomeUnknown, + "operation interrupted after dispatch", + ), + self.wire_limits, + ) + .map_err(|encode| ToolError::Internal(encode.to_string()))?; + Ok(( + response, + Some(Err(ToolError::Interrupted(output))), + WipAuditOutcome::Interrupted, + )) + } + Err(WipOperationError::OutcomeUnknown(message)) => { + let response = encode_protocol_error_response( + ProtocolInteraction::CallOperation, + &protocol_error(ProtocolErrorCode::OperationOutcomeUnknown, message.clone()), + self.wire_limits, + ) + .map_err(|encode| ToolError::Internal(encode.to_string()))?; + Ok(( + response, + Some(Err(ToolError::ExecutionFailed(format!( + "WIP operation outcome unknown; do not retry automatically: {message}" + )))), + WipAuditOutcome::OutcomeUnknown, + )) + } + } + } +} + +struct DispatchedCallGuard { + state: Arc>, + active: Arc>>, + request_id: RequestId, + execution_id: String, + armed: bool, +} + +impl DispatchedCallGuard { + fn new( + state: Arc>, + active: Arc>>, + request_id: RequestId, + execution_id: String, + ) -> Self { + Self { + state, + active, + request_id, + execution_id, + armed: true, + } + } + + fn fail_before_response(&mut self) { + if self.armed { + let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner()); + let _ = state.client.fail_dispatched_call( + self.request_id, + DispatchedTransportFailure::Disconnect, + "in-process WIP transport disconnected after dispatch", + ); + self.armed = false; + } + } + + fn disarm(&mut self) { + self.armed = false; + } +} + +impl Drop for DispatchedCallGuard { + fn drop(&mut self) { + if self.armed { + let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner()); + let _ = state.client.fail_dispatched_call( + self.request_id, + DispatchedTransportFailure::Disconnect, + "WIP call future ended without a terminal response", + ); + } + self.active + .lock() + .unwrap_or_else(|error| error.into_inner()) + .remove(&self.execution_id); + } +} + +#[derive(Debug, Deserialize, schemars::JsonSchema)] +struct WipDiscoverInput { + /// Canonical Worldspace path. Begin at `/`. + #[serde(default = "root_path")] + path: String, + /// Descendant depth to materialize. Use 1 to explore one level at a time. + #[serde(default = "one")] + depth: u32, + /// Explicitly supersede cached observations for this path. + #[serde(default)] + refresh: bool, + /// Drop all Known Space for this Worker/security context before observing. + #[serde(default)] + reset: bool, +} + +fn root_path() -> String { + WIP_ROOT.into() +} + +fn one() -> u32 { + 1 +} + +#[derive(Debug, Deserialize, schemars::JsonSchema)] +struct WipInspectInput { + /// Opaque interface reference returned by WipDiscover. + interface: String, + /// Explicitly supersede the cached descriptor observation. + #[serde(default)] + refresh: bool, +} + +#[derive(Debug, Deserialize, schemars::JsonSchema)] +struct WipCallInput { + /// Exact object path returned by WipDiscover. + path: String, + /// Exact interface reference selected from that object. + interface: String, + /// Exact operation name from WipInspect. + operation: String, + /// Original operation arguments. Compatibility operations accept the original tool object here. + arguments: Json, +} + +struct WipDiscoverTool { + runtime: Arc, +} + +#[async_trait] +impl Tool for WipDiscoverTool { + async fn execute( + &self, + input_json: &str, + _context: ToolExecutionContext, + ) -> Result { + let input: WipDiscoverInput = serde_json::from_str(input_json) + .map_err(|error| ToolError::InvalidArgument(error.to_string()))?; + if input.reset { + self.runtime.reset_observations()?; + } + self.runtime + .discover(input.path, input.depth.min(8), input.refresh) + .await + } +} + +struct WipInspectTool { + runtime: Arc, +} + +#[async_trait] +impl Tool for WipInspectTool { + async fn execute( + &self, + input_json: &str, + _context: ToolExecutionContext, + ) -> Result { + let input: WipInspectInput = serde_json::from_str(input_json) + .map_err(|error| ToolError::InvalidArgument(error.to_string()))?; + self.runtime.inspect(input.interface, input.refresh).await + } +} + +struct WipCallTool { + runtime: Arc, +} + +#[async_trait] +impl Tool for WipCallTool { + async fn execute( + &self, + input_json: &str, + context: ToolExecutionContext, + ) -> Result { + let input: WipCallInput = serde_json::from_str(input_json) + .map_err(|error| ToolError::InvalidArgument(error.to_string()))?; + self.runtime + .call( + input.path, + input.interface, + input.operation, + input.arguments, + context, + ) + .await + } + + async fn cancel_execution(&self, context: &ToolExecutionContext) -> Result<(), ToolError> { + self.runtime.cancel(context).await + } +} + +fn wip_tool_definitions(runtime: Arc) -> Vec { + let discover_runtime = Arc::clone(&runtime); + let inspect_runtime = Arc::clone(&runtime); + let call_runtime = Arc::clone(&runtime); + vec![ + Arc::new(move || { + let schema = schemars::schema_for!(WipDiscoverInput); + ( + ToolMeta::new("WipDiscover") + .description("Explore the authorized WIP Worldspace and stateful Known Space. Start at `/`; use reset after reconnect or authority changes.") + .input_schema(serde_json::to_value(schema).expect("WIP discover schema serializes")), + Arc::new(WipDiscoverTool { + runtime: Arc::clone(&discover_runtime), + }) as Arc, + ) + }), + Arc::new(move || { + let schema = schemars::schema_for!(WipInspectInput); + ( + ToolMeta::new("WipInspect") + .description("Fetch and validate one descriptor returned by WipDiscover. Inspect before calling an operation.") + .input_schema(serde_json::to_value(schema).expect("WIP inspect schema serializes")), + Arc::new(WipInspectTool { + runtime: Arc::clone(&inspect_runtime), + }) as Arc, + ) + }), + Arc::new(move || { + let schema = schemars::schema_for!(WipCallInput); + ( + ToolMeta::new("WipCall") + .description("Invoke an operation only from fresh object/interface observations. Stale validators and invalid original JSON Schema inputs fail closed; unknown write outcomes must not be retried automatically.") + .input_schema(serde_json::to_value(schema).expect("WIP call schema serializes")), + Arc::new(WipCallTool { + runtime: Arc::clone(&call_runtime), + }) as Arc, + ) + }), + ] +} + +/// Replace all currently registered ordinary tools with one WIP client surface. +/// Call this once after every enabled Feature has contributed its tools. +pub(crate) fn install_wip_mode( + engine: &mut Engine, + permissions: Option, + security_context: String, +) -> Result, String> { + install_wip_mode_with_mounts( + engine, + permissions, + security_context, + WipMountRegistry::new(), + ) +} + +/// Host extension point for aggregating Feature-owned native projections before +/// compatibility tools are finalized. Native mounts already present in `registry` +/// replace a matching `tool:` compatibility capability and all unrelated +/// route/interface collisions remain fatal. +pub fn install_wip_mode_with_mounts( + engine: &mut Engine, + permissions: Option, + security_context: String, + mut registry: WipMountRegistry, +) -> Result, String> { + let handle = engine.tool_server_handle(); + handle.flush_pending(); + let definitions = handle.tool_definitions_sorted(); + let ordinary_schema_bytes = definitions + .iter() + .map(|definition| { + definition.name.len() + + definition + .description + .as_deref() + .map(str::len) + .unwrap_or_default() + + serde_json::to_vec(&definition.input_schema) + .map(|bytes| bytes.len()) + .unwrap_or_default() + }) + .sum::() as u64; + let mut names = Vec::new(); + for definition in definitions { + let (meta, tool) = handle + .get_tool(&definition.name) + .ok_or_else(|| format!("registered tool `{}` disappeared", definition.name))?; + let projection = compatibility_projection(meta, tool, permissions.clone()) + .map_err(|error| error.to_string())?; + registry + .mount(projection) + .map_err(|error| error.to_string())?; + names.push(definition.name); + } + let runtime = Arc::new(WipRuntime::new( + WipHost::new(registry), + SecurityContext::new(security_context), + ordinary_schema_bytes, + )?); + for name in names { + handle + .unregister(&name) + .map_err(|error| error.to_string())?; + } + let definitions = wip_tool_definitions(Arc::clone(&runtime)); + let wip_schema_bytes = definitions + .iter() + .map(|definition| { + let (meta, _) = definition(); + meta.name.len() + + meta.description.len() + + serde_json::to_vec(&meta.input_schema) + .map(|bytes| bytes.len()) + .unwrap_or_default() + }) + .sum::() as u64; + runtime + .metrics + .wip_schema_bytes + .store(wip_schema_bytes, Ordering::Relaxed); + engine.register_tools(definitions); + Ok(runtime) +} + +fn client_tool_error(error: wip_client::ClientError) -> ToolError { + ToolError::ExecutionFailed(error.to_string()) +} + +fn is_descendant(parent: &str, candidate: &str) -> bool { + if parent == WIP_ROOT { + candidate.starts_with('/') + } else { + candidate.starts_with(&format!("{parent}/")) + } +} + +fn object_observation_json(observation: wip_client::ObjectObservation) -> Json { + json!({ + "path": observation.path, + "state": observation_state_name(&observation.state), + "validator": observation.validator.as_ref().map(hex), + "object": observation.object.map(|object| json!({ + "name": object.name, + "description": object.description, + "interfaces": object.interfaces, + "ref": object.r#ref, + "validator": object.validator.as_ref().map(hex), + })) + }) +} + +fn observation_state_name(state: &ObservationState) -> &'static str { + match state { + ObservationState::Loading(_) => "loading", + ObservationState::Fresh => "fresh", + ObservationState::Stale => "stale", + ObservationState::Error(_) => "error", + } +} + +fn descriptor_json(descriptor: &InterfaceDescriptor) -> Json { + json!({ + "format": descriptor.format, + "documentation": descriptor.documentation.as_ref().map(documentation_json), + "operations": descriptor.operations.iter().map(|operation| json!({ + "name": operation.name, + "documentation": operation.documentation.as_ref().map(documentation_json), + "parameters": operation.parameters.iter().map(|parameter| json!({ + "name": parameter.name, + "required": parameter.required, + "documentation": parameter.documentation.as_ref().map(documentation_json), + "type": type_json(¶meter.r#type), + })).collect::>(), + "returns": { + "documentation": operation.returns.documentation.as_ref().map(documentation_json), + "type": type_json(&operation.returns.r#type), + } + })).collect::>() + }) +} + +fn documentation_json(documentation: &Documentation) -> Json { + json!({ + "summary": documentation.summary, + "details": documentation.details, + }) +} + +fn type_json(value: &TypeExpr) -> Json { + match value { + TypeExpr::Unit => json!("unit"), + TypeExpr::Boolean => json!("boolean"), + TypeExpr::Integer => json!("integer"), + TypeExpr::Number => json!("number"), + TypeExpr::String => json!("string"), + TypeExpr::Bytes => json!("bytes"), + TypeExpr::Json => json!("json"), + TypeExpr::Entry => json!("entry"), + TypeExpr::Named { name } => json!({"named": name}), + TypeExpr::Record { .. } => json!("record"), + TypeExpr::List { .. } => json!("list"), + TypeExpr::Enum { .. } => json!("enum"), + TypeExpr::Union { .. } => json!("union"), + } +} + +fn metrics_json(metrics: WipMetricsSnapshot) -> Json { + json!({ + "ordinary_schema_bytes": metrics.ordinary_schema_bytes, + "wip_schema_bytes": metrics.wip_schema_bytes, + "discover_round_trips": metrics.discover_round_trips, + "inspect_round_trips": metrics.inspect_round_trips, + "operation_round_trips": metrics.operation_round_trips, + "operation_successes": metrics.operation_successes, + }) +} + +fn tool_output_value(output: &ToolOutput) -> Value { + let mut record = BTreeMap::new(); + record.insert("summary".into(), Value::String(output.summary.clone())); + record.insert( + "content".into(), + output + .content + .clone() + .map(Value::String) + .unwrap_or(Value::Unit), + ); + record.insert( + "attachment_count".into(), + Value::Integer(i64::try_from(output.attachments.len()).unwrap_or(i64::MAX)), + ); + Value::Record(record) +} + +fn render_call_completion(completion: Completion) -> Result { + let Completion::Call(record) = completion else { + return Err(ToolError::Internal( + "WIP operation completed with a retrieval result".into(), + )); + }; + match record.outcome { + CallOutcome::Success(response) => Ok(json_output( + "WIP operation completed".into(), + wip_to_json(&response.result).map_err(ToolError::Internal)?, + )), + CallOutcome::ProtocolFailure(error) => Err(ToolError::ExecutionFailed(error.to_string())), + CallOutcome::Unknown { reason, failure } => Err(ToolError::ExecutionFailed(format!( + "WIP operation outcome unknown ({reason:?}); do not retry automatically{}", + failure + .map(|failure| format!(": {failure}")) + .unwrap_or_default() + ))), + CallOutcome::NotDispatched(error) => Err(ToolError::ExecutionFailed(format!( + "WIP operation was not dispatched: {error}" + ))), + } +} + +fn json_output(summary: String, value: Json) -> ToolOutput { + ToolOutput { + summary, + content: Some(serde_json::to_string_pretty(&value).unwrap_or_else(|_| value.to_string())), + attachments: Vec::new(), + } +} + +fn json_to_wip(value: &Json) -> Result { + match value { + Json::Null => Ok(Value::Unit), + Json::Bool(value) => Ok(Value::Boolean(*value)), + Json::Number(value) => { + if let Some(value) = value.as_i64() { + if (wip_protocol::MIN_SAFE_INTEGER..=wip_protocol::MAX_SAFE_INTEGER) + .contains(&value) + { + Ok(Value::Integer(value)) + } else { + Err("integer is outside the WIP lossless range".into()) + } + } else { + let value = value + .as_f64() + .ok_or_else(|| "JSON number is not representable".to_string())?; + if value.is_finite() { + Ok(Value::Number(value)) + } else { + Err("WIP numbers must be finite".into()) + } + } + } + Json::String(value) => Ok(Value::String(value.clone())), + Json::Array(values) => values + .iter() + .map(json_to_wip) + .collect::, _>>() + .map(Value::List), + Json::Object(values) => values + .iter() + .map(|(name, value)| Ok((name.clone(), json_to_wip(value)?))) + .collect::, _>>() + .map(Value::Record), + } +} + +fn wip_to_json(value: &Value) -> Result { + match value { + Value::Unit => Ok(Json::Null), + Value::Boolean(value) => Ok(Json::Bool(*value)), + Value::Integer(value) => Ok(json!(value)), + Value::Number(value) if value.is_finite() => Ok(json!(value)), + Value::Number(_) => Err("WIP number is not finite".into()), + Value::String(value) => Ok(Json::String(value.clone())), + Value::Bytes(_) => Err("WIP bytes cannot be projected into JSON tool arguments".into()), + Value::Record(values) => values + .iter() + .map(|(name, value)| Ok((name.clone(), wip_to_json(value)?))) + .collect::, _>>() + .map(Json::Object), + Value::List(values) => values + .iter() + .map(wip_to_json) + .collect::, _>>() + .map(Json::Array), + } +} + +fn hex(bytes: &Vec) -> String { + bytes.iter().map(|byte| format!("{byte:02x}")).collect() +} + +#[cfg(test)] +mod tests { + use super::*; + use agen::llm_client::client::LlmClient; + use agen::llm_client::{ClientError, Request as LlmRequest, ResponseStream}; + use futures::stream; + use std::sync::atomic::{AtomicUsize, Ordering}; + + struct EchoTool { + calls: Arc, + } + + #[async_trait] + impl Tool for EchoTool { + async fn execute( + &self, + input: &str, + _context: ToolExecutionContext, + ) -> Result { + self.calls.fetch_add(1, Ordering::SeqCst); + Ok(input.to_string().into()) + } + } + + fn meta(name: &str) -> ToolMeta { + ToolMeta::new(name) + .description("test tool") + .input_schema(json!({ + "type": "object", + "additionalProperties": false, + "properties": {"message": {"type": "string", "minLength": 2}}, + "required": ["message"] + })) + } + + fn registry_with_tool( + name: &str, + calls: Arc, + permissions: Option, + ) -> WipMountRegistry { + let mut registry = WipMountRegistry::new(); + registry + .mount( + compatibility_projection(meta(name), Arc::new(EchoTool { calls }), permissions) + .unwrap(), + ) + .unwrap(); + registry + } + + #[derive(Clone)] + struct DummyClient; + + #[async_trait] + impl LlmClient for DummyClient { + async fn stream(&self, _request: LlmRequest) -> Result { + Ok(Box::pin(stream::empty())) + } + + fn clone_boxed(&self) -> Box { + Box::new(self.clone()) + } + } + + fn echo_definition(name: String, calls: Arc) -> ToolDefinition { + Arc::new(move || { + ( + meta(&name), + Arc::new(EchoTool { + calls: Arc::clone(&calls), + }) as Arc, + ) + }) + } + + #[test] + fn mode_installation_replaces_tool_surface_without_changing_normal_engines() { + let calls = Arc::new(AtomicUsize::new(0)); + let mut normal = Engine::<_, Mutable, ()>::new_annotated(DummyClient); + normal.register_tool(echo_definition("Echo".into(), Arc::clone(&calls))); + normal.tool_server_handle().flush_pending(); + assert_eq!( + normal + .tool_server_handle() + .tool_definitions_sorted() + .into_iter() + .map(|definition| definition.name) + .collect::>(), + ["Echo"] + ); + + let mut wip = Engine::<_, Mutable, ()>::new_annotated(DummyClient); + for index in 0..24 { + wip.register_tool(echo_definition(format!("Echo{index}"), Arc::clone(&calls))); + } + let runtime = install_wip_mode(&mut wip, None, "worker-a".into()).unwrap(); + wip.tool_server_handle().flush_pending(); + assert_eq!( + wip.tool_server_handle() + .tool_definitions_sorted() + .into_iter() + .map(|definition| definition.name) + .collect::>(), + ["WipCall", "WipDiscover", "WipInspect"] + ); + let metrics = runtime.metrics(); + assert!(metrics.ordinary_schema_bytes > metrics.wip_schema_bytes); + } + + #[test] + fn mount_collision_and_native_replacement_are_explicit() { + struct Native; + #[async_trait] + impl WipOperationHandler for Native { + async fn call( + &self, + _operation: &str, + _arguments: &BTreeMap, + _context: WipCallContext, + ) -> Result { + Ok(WipOperationOutput::native(Value::Unit)) + } + } + + let calls = Arc::new(AtomicUsize::new(0)); + let mut registry = registry_with_tool("Echo", calls, None); + let mut native = compatibility_projection( + meta("Echo"), + Arc::new(EchoTool { + calls: Arc::new(AtomicUsize::new(0)), + }), + None, + ) + .unwrap(); + native.kind = WipProjectionKind::Native; + native.handler = Arc::new(Native); + assert_eq!( + registry.mount(native).unwrap(), + WipMountDisposition::ReplacedCompatibility + ); + + let mut collision = compatibility_projection( + meta("Other"), + Arc::new(EchoTool { + calls: Arc::new(AtomicUsize::new(0)), + }), + None, + ) + .unwrap(); + collision.route = "/tools/Echo".into(); + collision.object.name = "Echo".into(); + assert!(matches!( + registry.mount(collision), + Err(WipMountError::RouteCollision { .. }) + )); + } + + #[tokio::test] + async fn stateful_discover_inspect_call_round_trip_validates_original_schema() { + let calls = Arc::new(AtomicUsize::new(0)); + let host = WipHost::new(registry_with_tool("Echo", Arc::clone(&calls), None)); + let runtime = WipRuntime::new(host, SecurityContext::new("worker-a"), 4096).unwrap(); + + runtime.discover("/".into(), 2, false).await.unwrap(); + runtime + .inspect("yoi.tool/Echo/v1".into(), false) + .await + .unwrap(); + let output = runtime + .call( + "/tools/Echo".into(), + "yoi.tool/Echo/v1".into(), + "call".into(), + json!({"message": "ok"}), + ToolExecutionContext::direct(), + ) + .await + .unwrap(); + assert_eq!(output.summary, "{\"message\":\"ok\"}"); + assert_eq!(calls.load(Ordering::SeqCst), 1); + + let invalid = runtime + .call( + "/tools/Echo".into(), + "yoi.tool/Echo/v1".into(), + "call".into(), + json!({"message": "x"}), + ToolExecutionContext::direct(), + ) + .await + .unwrap_err(); + assert!(invalid.to_string().contains("InvalidArguments")); + assert_eq!(calls.load(Ordering::SeqCst), 1); + let metrics = runtime.metrics(); + assert_eq!(metrics.discover_round_trips, 1); + assert_eq!(metrics.inspect_round_trips, 1); + assert_eq!(metrics.operation_successes, 1); + } + + #[tokio::test] + async fn sessions_are_isolated_and_stale_validators_fail_before_dispatch() { + let calls = Arc::new(AtomicUsize::new(0)); + let host = WipHost::new(registry_with_tool("Echo", Arc::clone(&calls), None)); + let runtime = WipRuntime::new(host, SecurityContext::new("worker-a"), 0).unwrap(); + runtime.discover("/".into(), 2, false).await.unwrap(); + runtime + .inspect("yoi.tool/Echo/v1".into(), false) + .await + .unwrap(); + + let projection = runtime.host.projection("/tools/Echo").unwrap(); + let mut request = CallOperationRequest { + target: wip_protocol::Target { + path: "/tools/Echo".into(), + validator: Some(vec![0]), + }, + interface: wip_protocol::InterfaceTarget { + reference: projection.interface.clone(), + validator: projection.interface_validator.clone(), + }, + operation: "call".into(), + arguments: BTreeMap::from([( + "input".into(), + json_to_wip(&json!({"message": "ok"})).unwrap(), + )]), + }; + let result = runtime + .host + .call( + request.clone(), + WipCallContext { + execution: ToolExecutionContext::direct(), + security_context: "worker-a".into(), + }, + ) + .await; + assert!(matches!( + result, + Err(WipOperationError::Protocol(ProtocolError { + code: ProtocolErrorCode::ValidatorMismatch, + .. + })) + )); + request.target.validator = projection.object.validator.clone(); + request.interface.validator = Some(vec![0]); + let result = runtime + .host + .call( + request.clone(), + WipCallContext { + execution: ToolExecutionContext::direct(), + security_context: "worker-a".into(), + }, + ) + .await; + assert!(matches!( + result, + Err(WipOperationError::Protocol(ProtocolError { + code: ProtocolErrorCode::InterfaceValidatorMismatch, + .. + })) + )); + + request.interface.validator = projection.interface_validator.clone(); + request.operation = "forged".into(); + let result = runtime + .host + .call( + request.clone(), + WipCallContext { + execution: ToolExecutionContext::direct(), + security_context: "worker-a".into(), + }, + ) + .await; + assert!(matches!( + result, + Err(WipOperationError::Protocol(ProtocolError { + code: ProtocolErrorCode::OperationNotFound, + .. + })) + )); + + request.target.path = "/tools/NotPublished".into(); + let result = runtime + .host + .call( + request, + WipCallContext { + execution: ToolExecutionContext::direct(), + security_context: "worker-a".into(), + }, + ) + .await; + assert!(matches!( + result, + Err(WipOperationError::Protocol(ProtocolError { + code: ProtocolErrorCode::NotFound, + .. + })) + )); + assert_eq!(calls.load(Ordering::SeqCst), 0); + + runtime.reset_observations().unwrap(); + let state = runtime.state.lock().unwrap(); + assert!(state.client.known_space(&state.session).unwrap().is_empty()); + assert_eq!(state.session.security_context().as_str(), "worker-a"); + } + + struct FailingTool { + cancelled: bool, + } + + #[async_trait] + impl Tool for FailingTool { + async fn execute( + &self, + _input: &str, + _context: ToolExecutionContext, + ) -> Result { + if self.cancelled { + Err(ToolError::Cancelled("cancelled safely".to_string().into())) + } else { + Err(ToolError::ExecutionFailed("provider reply was lost".into())) + } + } + } + + async fn prepare_runtime_with_tool(name: &str, tool: Arc) -> WipRuntime { + let mut registry = WipMountRegistry::new(); + registry + .mount(compatibility_projection(meta(name), tool, None).unwrap()) + .unwrap(); + let runtime = + WipRuntime::new(WipHost::new(registry), SecurityContext::new("worker-a"), 0).unwrap(); + runtime.discover("/".into(), 2, false).await.unwrap(); + runtime + .inspect(format!("yoi.tool/{name}/v1"), false) + .await + .unwrap(); + runtime + } + + #[tokio::test] + async fn cancelled_unknown_and_disconnected_calls_have_distinct_terminal_states() { + let cancelled = + prepare_runtime_with_tool("Cancelled", Arc::new(FailingTool { cancelled: true })).await; + let error = cancelled + .call( + "/tools/Cancelled".into(), + "yoi.tool/Cancelled/v1".into(), + "call".into(), + json!({"message": "ok"}), + ToolExecutionContext::direct(), + ) + .await + .unwrap_err(); + assert!(matches!(error, ToolError::Cancelled(_))); + assert_eq!( + cancelled.audit().last().unwrap().outcome, + WipAuditOutcome::Cancelled + ); + + let unknown = + prepare_runtime_with_tool("Unknown", Arc::new(FailingTool { cancelled: false })).await; + let error = unknown + .call( + "/tools/Unknown".into(), + "yoi.tool/Unknown/v1".into(), + "call".into(), + json!({"message": "ok"}), + ToolExecutionContext::direct(), + ) + .await + .unwrap_err(); + assert!(error.to_string().contains("outcome unknown")); + assert_eq!( + unknown.audit().last().unwrap().outcome, + WipAuditOutcome::OutcomeUnknown + ); + unknown + .discover("/tools/Unknown".into(), 0, true) + .await + .unwrap(); + + let prepared = { + let mut state = unknown.state.lock().unwrap(); + let session = state.session.clone(); + let prepared = state + .client + .prepare_call( + &session, + "/tools/Unknown", + "yoi.tool/Unknown/v1", + "call", + BTreeMap::from([( + "input".into(), + json_to_wip(&json!({"message": "ok"})).unwrap(), + )]), + ) + .unwrap(); + state.client.mark_dispatched(prepared.id).unwrap(); + prepared + }; + drop(DispatchedCallGuard::new( + Arc::clone(&unknown.state), + Arc::clone(&unknown.active), + prepared.id, + "dropped-execution".into(), + )); + let state = unknown.state.lock().unwrap(); + let record = state + .client + .call_history(&state.session) + .unwrap() + .back() + .unwrap(); + assert!(matches!( + record.outcome, + CallOutcome::Unknown { + reason: wip_client::OutcomeUnknownReason::Disconnect, + .. + } + )); + } + + #[tokio::test] + async fn projected_permission_policy_is_applied_to_original_tool_identity() { + let calls = Arc::new(AtomicUsize::new(0)); + let permissions = ToolPermissionConfig { + default_action: ToolPermissionAction::Deny, + rules: Vec::new(), + }; + let host = WipHost::new(registry_with_tool( + "Echo", + Arc::clone(&calls), + Some(permissions), + )); + let runtime = WipRuntime::new(host, SecurityContext::new("worker-a"), 0).unwrap(); + runtime.discover("/".into(), 2, false).await.unwrap(); + runtime + .inspect("yoi.tool/Echo/v1".into(), false) + .await + .unwrap(); + let error = runtime + .call( + "/tools/Echo".into(), + "yoi.tool/Echo/v1".into(), + "call".into(), + json!({"message": "ok"}), + ToolExecutionContext::direct(), + ) + .await + .unwrap_err(); + assert!(error.to_string().contains("PermissionDenied")); + assert_eq!(calls.load(Ordering::SeqCst), 0); + } +} diff --git a/docs/README.md b/docs/README.md index 00c0e123f..433d6b42e 100644 --- a/docs/README.md +++ b/docs/README.md @@ -33,6 +33,7 @@ It is not a dumping ground for external research, old plans, API inventories, or 25. [`design/subjektiv-store.md`](design/subjektiv-store.md) — subject-scoped Memoryのownership、authorization、Session attribution、revision、candidate consolidation境界。 26. [`design/subjektiv-surface.md`](design/subjektiv-surface.md) — confirmed Memoryから生成するresident surface、availability、append-only restore injection。 27. [`development/subjektiv-product-cutover.md`](development/subjektiv-product-cutover.md) — legacy Workspace Memoryからsubjektivへのoperator cutover、explicit reset、backup／rollback runbook。 +28. [`design/worker-wip-mode.md`](design/worker-wip-mode.md) — explicit-opt-in Worker WIP mode, Worldspace projection, authority, cache, and retry contracts. ## What belongs here diff --git a/docs/design/worker-wip-mode.md b/docs/design/worker-wip-mode.md new file mode 100644 index 000000000..b152a6171 --- /dev/null +++ b/docs/design/worker-wip-mode.md @@ -0,0 +1,109 @@ +# Worker WIP mode + +## Status and dependencies + +Workers support two explicit model-facing modes: + +```toml +[worker] +name = "example" +mode = "tools" # default +``` + +```toml +[worker] +name = "example" +mode = "wip" +``` + +Profiles select the same policy without owning the runtime Worker name: + +```toml +worker = { mode = "wip" } +``` + +Omission resolves to `tools`, so existing manifests, profiles, persisted snapshots, and restored Workers retain the ordinary individual-tool surface. A direct Internal SubWorker selected with `profile = "inherit"` inherits the parent mode as reusable behavior. `default` and named profile selections use that profile's own mode and do not implicitly inherit WIP. + +The implementation uses the crates.io `wip-client`, `wip-protocol`, and `wip-http` **0.1.0** releases. It does not use Git/path dependencies or the old `wip-https` name. Those releases require Rust 1.88; the workspace declares `rust-version = "1.88"`, and the changed crates inherit it. WIP Host/interface crates are not needed for the async compatibility adapter: the published client/protocol/HTTP state machines remain authoritative while Yoi supplies an in-process async Host transport. + +## LLM surface and route contract + +A WIP Worker exposes exactly these generic LLM tools instead of exposing every enabled ordinary tool twice: + +- `WipDiscover`: observe an object path and bounded descendants, beginning at `/`. +- `WipInspect`: fetch the descriptor for an opaque interface reference returned by discovery. +- `WipCall`: call an operation using fresh object and interface observations. + +The compatibility projection mounts enabled ordinary tools at: + +```text +/tools/ +``` + +Each object has one opaque interface reference: + +```text +yoi.tool//v1 +``` + +Its single operation is `call`. `WipCall.arguments` is the original ordinary tool argument object. Tool argument IDs remain JSON values; the adapter never infers a domain object or Worldspace route from them. + +The interface descriptor uses WIP `Json` for compatibility input and includes the exact original JSON Schema in descriptor documentation. Before execution, the Host validates the JSON value with that original schema and then delegates to the original async `Tool`. Constraints are therefore neither approximated nor silently dropped. A schema that cannot be compiled prevents WIP startup instead of creating a weaker projection. + +## Native projection extension and collisions + +`worker::wip::WipMountRegistry` is the Host-owned mount allocator. A Feature can independently construct a `WipProjection` with an async `WipOperationHandler`, then pass its aggregate registry through `install_wip_mode_with_mounts` before compatibility finalization. + +- Routes must be canonical absolute WIP paths. +- Object names must match the final route segment. +- A projected object exposes exactly its declared interface. +- Unrelated duplicate routes and conflicting interface descriptors fail startup. +- A native projection replaces a compatibility projection only when both declare the same semantic capability (`tool:` for compatibility tools). +- When native is already selected, the compatibility tool is hidden rather than exposed through a second entry. + +The unit sample `mount_collision_and_native_replacement_are_explicit` demonstrates independent native registration, deterministic replacement, and collision rejection. + +## Authority and execution + +Discovery is not authority. The Worldspace contains only the tools already registered from enabled Features and the Worker's current execution context. Calling a compatibility operation: + +1. resolves the exact mounted route; +2. verifies object and interface membership and validators; +3. validates the WIP descriptor call; +4. validates the original JSON Schema; +5. evaluates manifest tool permission rules against the **original tool name and original arguments**; +6. invokes the original async `Tool` without blocking the Tokio runtime. + +Filesystem scope, provider Workdir capability checks, Backend Workspace authority, Feature enablement, and input-specific restrictions stay in the original tool implementation and are checked again at invocation. `ask` permissions remain denied fail-closed because the runtime has no approval protocol. The gateway does not mint authority or expose authentication material. + +Pre/post Engine history still records one bounded `WipCall` result, while the compatibility result retains the original `ToolOutput`, including attachments and normal Engine output pruning. WIP call audit records retain the route, operation, request identity, and terminal classification. + +## Client lifetime, restoration, and compaction + +The stateful `wip-client::Client` is owned by one Worker WIP runtime. Sessions are keyed by the canonical endpoint and an opaque security-context identity derived by the Host from Workspace, Worker, and session identity. Authentication credentials are not included in that identity and are never emitted to descriptors, tool output, or history. + +Known Space and interface observations are therefore not shared across Workers, endpoints, or security contexts. A restored Worker creates a new runtime/client and must explore again. `WipDiscover(reset = true)` drops all observations before reconnect/authority-change exploration; `refresh = true` explicitly supersedes one cached observation. A normal compaction keeps the live runtime but the three gateway schemas are supplied again on every LLM request, including the instruction to begin discovery at `/`; the compacted transcript is not treated as cache authority. + +## Errors, cancellation, and retries + +The adapter keeps these outcomes distinct: + +- **rejected/not dispatched**: route, membership, validator, descriptor, schema, or permission failure; the original operation does not run; +- **stale**: object/interface validator mismatch; refresh before reconstructing a call; +- **cancelled**: the original provider confirms cooperative cancellation and its bounded cancelled output is retained; +- **interrupted**: provider interruption is retained separately from cancellation; +- **disconnected/timeout after dispatch**: `wip-client` records an unknown outcome; +- **operation outcome unknown**: an executing provider fails without a rollback/non-execution guarantee, a response cannot be completed after dispatch, or the call future is force-closed. + +An unknown mutating outcome must never be retried automatically. The model must inspect external state or ask for operator guidance. Retrieval failures may be explicitly refreshed. + +## Measurement + +The runtime tracks bounded counters for: + +- ordinary individual-tool schema bytes hidden by WIP mode; +- the three WIP gateway schema bytes; +- discover, descriptor-inspection, and operation round trips; +- successful operations. + +These values are returned with discovery/inspection output and are intended for comparing prompt input volume, exploration round trips, and representative operation success with ordinary Tool mode. No fixed reduction percentage is promised.