diff --git a/Cargo.lock b/Cargo.lock index 1f04b08872..ddd5806b1e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6949,6 +6949,7 @@ dependencies = [ "dunce", "filetime", "fs2", + "futures", "git2", "globset", "ignore", diff --git a/docs/features/remote-workspaces.md b/docs/features/remote-workspaces.md index eb8ba9e759..4cc0f3f23c 100644 --- a/docs/features/remote-workspaces.md +++ b/docs/features/remote-workspaces.md @@ -234,6 +234,33 @@ working execution and text-query paths and receive an explicit upgrade instructi for a missing binary capability. Existing Remote Connect file commands and persisted session records remain compatible; the file-chunk revision is additive. +## User Skill resources + +User Skill availability is controlled by the runtime host's +`workspace.remote_user_skills` preference: `skill_md_only` (the upgrade-compatible +default), `disabled`, or `full`. It applies to remote discovery, prompt listings +and explicit invocation, without changing remote project Skills. See the +[Desktop guide](../../src/apps/desktop/README.md#user-skills-in-remote-workspaces) +for prerequisites and behavior. + +Full loading deploys the selected directory through the bound filesystem +provider, never through a local fallback. SSH workspaces use the existing SFTP +connection; synchronization does not execute a remote shell or require Python, +an archive extractor or an external synchronization program. Deterministic file +hashes and permissions identify package versions. Bounded parallel streamed +reads check the complete remote tree before reuse; otherwise files are uploaded +and verified in a staging directory before publication. Many small files can +still take time over high-latency connections, so progress remains visible. +Published directories are never moved or removed by synchronization. A changed +or damaged cached copy leads to a new generation; matching verified generations +can be reused. This preserves resource paths already returned to other turns +without a persistent target-side lock. +Returned Skill paths use target POSIX semantics, +and references rooted in the original package are mapped to that deployment. +The runtime reads instructions back from the deployed snapshot before reporting +success. This copies package resources; it does not install interpreters, +relocate executable hooks, or resolve external machine-specific dependencies. + ## Upgrade compatibility Existing SSH profiles remain plain SSH targets because the new `proxyJump` and diff --git a/scripts/check-core-boundaries.test.mjs b/scripts/check-core-boundaries.test.mjs index 96efc90eab..73f07fc9eb 100644 --- a/scripts/check-core-boundaries.test.mjs +++ b/scripts/check-core-boundaries.test.mjs @@ -706,6 +706,7 @@ test('Services Core feature-free dependencies stay behind exact text and async I 'workspace-runtime', 'workspace-text-runtime', 'workspace-transfer', + 'skill-package-sync', ]), ); }); @@ -4315,6 +4316,7 @@ test('Services Core accepts only the reviewed feature-owned Tokio runtime graph' ], 'session-git': ['local-storage'], 'workspace-transfer': ['dep:tokio', 'tokio/fs', 'tokio/io-util', 'tokio/rt', 'tokio/sync'], + 'skill-package-sync': ['dep:tokio', 'tokio/fs', 'tokio/io-util', 'tokio/rt'], }, }; @@ -5068,6 +5070,78 @@ test('workspace transfer admits IO without process or networking capabilities', v.message.includes('workspace-transfer has unexpected effective Tokio capabilities'))); }); +test('Skill package synchronization keeps streamed workspace IO without archive or process owners', async () => { + const manifestPath = 'src/crates/services/services-core/Cargo.toml'; + const manifest = await readFile(new URL(`../${manifestPath}`, import.meta.url), 'utf8'); + const features = parseManifestFeatures(manifest); + const expected = [ + 'dep:anyhow', 'dep:openbitfun-runtime-ports', + 'openbitfun-runtime-ports/workspace-ports', 'dep:sha2', 'dep:futures', 'dep:uuid', + 'dep:tokio', 'tokio/fs', 'tokio/io-util', 'tokio/rt', + ]; + const profile = coreClosedFeatureProfileRules.find(rule => + rule.manifestPath === manifestPath && rule.featureName === 'skill-package-sync'); + assert.equal(profile?.exact, true); + assert.deepEqual(profile.requiredFeatureRefs, expected); + assert.deepEqual(features['skill-package-sync'], expected); + const serviceOwners = optionalDependencyFeatureOwnerRules.find(rule => rule.crateName === 'services-core'); + assert.deepEqual(serviceOwners.dependencies.find(rule => rule.depName === 'zip').ownerFeatures, ['pet-packages']); + assert.deepEqual(serviceOwners.dependencies.find(rule => rule.depName === 'futures').ownerFeatures, ['skill-package-sync']); + + const pkg = { + name: 'openbitfun-services-core', + manifest_path: manifestPath, + dependencies: [{ name: 'tokio', kind: null, optional: true, features: [] }], + features, + }; + assert.deepEqual(findTokioDependencyFeatureViolations([pkg]), []); + for (const added of ['tokio/process', 'tokio/net', 'workspace-runtime']) { + const mutated = structuredClone(pkg); + mutated.features['skill-package-sync'].push(added); + assert.ok(findTokioDependencyFeatureViolations([mutated]).some(violation => + violation.message.includes('skill-package-sync has unexpected effective Tokio capabilities')), added); + } + + const { findCapabilityContractConsumerViolations } = await import( + './core-boundaries/cargo-dependency-boundaries.mjs' + ); + const runtimePorts = capabilityPackage( + 'openbitfun-runtime-ports', + 'src/crates/contracts/runtime-ports/Cargo.toml', + RUNTIME_PORT_FEATURE_PROFILES, + ); + const consumer = { + ...packageAt('openbitfun-services-core', manifestPath, [ + pathDependency('src/crates/contracts/runtime-ports', { + name: 'openbitfun-runtime-ports', optional: true, usesDefaultFeatures: false, + }), + ]), + features, + }; + const rules = capabilityContractDependencyRules.filter(rule => rule.packageName === 'openbitfun-runtime-ports'); + assert.deepEqual(findTestCapabilityViolations(findCapabilityContractConsumerViolations, [runtimePorts, consumer], rules), []); + for (const added of ['runtime-event-port', 'remote-exec-port', 'tool-runtime-handles']) { + const mutated = structuredClone(consumer); + mutated.features['skill-package-sync'].push(`openbitfun-runtime-ports/${added}`); + assert.ok(findTestCapabilityViolations(findCapabilityContractConsumerViolations, [runtimePorts, mutated], rules) + .some(violation => violation.message.includes('skill-package-sync') && violation.message.includes(added)), added); + } +}); + +test('remote workspace feature selects the Skill package service only at its owning surface', async () => { + const manifestPath = 'src/crates/assembly/core/Cargo.toml'; + const manifest = await readFile(new URL(`../${manifestPath}`, import.meta.url), 'utf8'); + const expected = [ + 'workspace-runtime', 'dep:openbitfun-services-integrations', + 'openbitfun-services-integrations/remote-ssh', 'openbitfun-services-core/skill-package-sync', + ]; + const profile = coreClosedFeatureProfileRules.find(rule => + rule.manifestPath === manifestPath && rule.featureName === 'remote-workspace'); + assert.equal(profile?.exact, true); + assert.deepEqual(profile.requiredFeatureRefs, expected); + assert.deepEqual(new Set(parseManifestFeatures(manifest)['remote-workspace']), new Set(expected)); +}); + test('core loopback WebSocket fixture cannot broaden the runtime dependency', () => { const runtime = { name: 'tokio-tungstenite', kind: null, optional: true, uses_default_features: true, features: [] }; diff --git a/scripts/core-boundaries/cargo-dependency-boundaries.mjs b/scripts/core-boundaries/cargo-dependency-boundaries.mjs index eb1a6863a8..4ff1aac8c6 100644 --- a/scripts/core-boundaries/cargo-dependency-boundaries.mjs +++ b/scripts/core-boundaries/cargo-dependency-boundaries.mjs @@ -188,6 +188,7 @@ const SERVICES_CORE_TOKIO_FEATURES = new Map([ ['workspace-text-runtime', ['rt']], ['workspace-runtime', ['fs', 'io-util', 'process', 'rt', 'sync', 'time']], ['workspace-transfer', ['fs', 'io-util', 'rt', 'sync']], + ['skill-package-sync', ['fs', 'io-util', 'rt']], ]); const SERVICES_CORE_BASE_TOKIO_FEATURES = []; const SERVICES_INTEGRATIONS_TOKIO_AGGREGATES = new Set(['product-full']); diff --git a/scripts/core-boundaries/rules/feature-rules.mjs b/scripts/core-boundaries/rules/feature-rules.mjs index f75ebe5c05..9f329ed9ab 100644 --- a/scripts/core-boundaries/rules/feature-rules.mjs +++ b/scripts/core-boundaries/rules/feature-rules.mjs @@ -46,9 +46,9 @@ export const optionalDependencyFeatureOwnerRules = [ dependencies: [ { depName: 'image', ownerFeatures: ['pet-packages'] }, { depName: 'zip', ownerFeatures: ['pet-packages'] }, - { depName: 'uuid', ownerFeatures: ['pet-packages', 'resource-capacity'] }, + { depName: 'uuid', ownerFeatures: ['pet-packages', 'resource-capacity', 'skill-package-sync'] }, { depName: 'aes-gcm', ownerFeatures: ['credential-vault'] }, - { depName: 'anyhow', ownerFeatures: ['credential-vault', 'dispatch-workspace', 'workspace-runtime', 'workspace-transfer', 'session-recovery'] }, + { depName: 'anyhow', ownerFeatures: ['credential-vault', 'dispatch-workspace', 'workspace-runtime', 'workspace-transfer', 'session-recovery', 'skill-package-sync'] }, { depName: 'async-trait', ownerFeatures: ['permission', 'workspace-runtime'] }, { depName: 'base64', ownerFeatures: ['pet-packages', 'credential-vault', 'filesystem'] }, { @@ -56,11 +56,12 @@ export const optionalDependencyFeatureOwnerRules = [ ownerFeatures: ['filesystem', 'local-storage', 'product-identity', 'workspace-persistence'], }, { depName: 'openbitfun-events', ownerFeatures: ['local-storage', 'session-event-format'] }, - { depName: 'openbitfun-runtime-ports', ownerFeatures: ['permission', 'workspace-runtime', 'workspace-persistence', 'workspace-transfer'] }, + { depName: 'openbitfun-runtime-ports', ownerFeatures: ['permission', 'workspace-runtime', 'workspace-persistence', 'workspace-transfer', 'skill-package-sync'] }, { depName: 'chrono', ownerFeatures: ['filesystem', 'local-storage', 'workspace-persistence'] }, { depName: 'chrono-tz', ownerFeatures: ['token-usage-statistics'] }, { depName: 'dunce', ownerFeatures: ['pet-packages', 'runtime-ownership', 'workspace-identity', 'workspace-runtime'] }, { depName: 'fs2', ownerFeatures: ['pet-packages', 'credential-vault', 'json-io', 'local-storage', 'runtime-ownership'] }, + { depName: 'futures', ownerFeatures: ['skill-package-sync'] }, { depName: 'git2', ownerFeatures: ['session-git'] }, { depName: 'globset', ownerFeatures: ['workspace-instructions'] }, { depName: 'ignore', ownerFeatures: ['filesystem'] }, @@ -92,6 +93,7 @@ export const optionalDependencyFeatureOwnerRules = [ 'resource-capacity', 'workspace-identity', 'workspace-transfer', + 'skill-package-sync', ], }, { depName: 'which', ownerFeatures: ['process-runtime'] }, @@ -112,6 +114,7 @@ export const optionalDependencyFeatureOwnerRules = [ 'workspace-runtime', 'workspace-text-runtime', 'workspace-transfer', + 'skill-package-sync', ], }, ], @@ -517,8 +520,9 @@ export const capabilityContractDependencyRules = [ capabilityForwarder('workspace-runtime', 'runtime-event-port'), capabilityForwarder('workspace-runtime', 'workspace-ports'), capabilityForwarder('workspace-transfer', 'workspace-ports'), + capabilityForwarder('skill-package-sync', 'workspace-ports'), ], - ['permission', 'workspace-runtime', 'workspace-persistence', 'workspace-transfer'], + ['permission', 'workspace-runtime', 'workspace-persistence', 'workspace-transfer', 'skill-package-sync'], ['session-recovery'], )], ['openbitfun-services-integrations', capabilityConsumer( @@ -1672,6 +1676,17 @@ export const coreClosedFeatureProfileRules = [ exact: true, reason: 'workspace-transfer owns bounded upload IO and optimistic commit through injected workspace ports, without process or transport implementations', }, + { + manifestPath: 'src/crates/services/services-core/Cargo.toml', + featureName: 'skill-package-sync', + requiredFeatureRefs: [ + 'dep:anyhow', 'dep:openbitfun-runtime-ports', + 'openbitfun-runtime-ports/workspace-ports', 'dep:sha2', 'dep:futures', 'dep:uuid', + 'dep:tokio', 'tokio/fs', 'tokio/io-util', 'tokio/rt', + ], + exact: true, + reason: 'skill-package-sync owns bounded concurrent file transfer and streamed verification through injected workspace filesystem ports, without archive, process or transport implementations', + }, { manifestPath: 'src/crates/services/services-core/Cargo.toml', featureName: 'workspace-runtime', @@ -1776,10 +1791,11 @@ export const coreClosedFeatureProfileRules = [ 'workspace-runtime', 'dep:openbitfun-services-integrations', 'openbitfun-services-integrations/remote-ssh', + 'openbitfun-services-core/skill-package-sync', ], allowedTransitiveFeatureRefs: ['filesystem', 'local-storage', 'process-runtime'], exact: true, - reason: 'openbitfun-core remote-workspace must add only the remote workspace service surface', + reason: 'openbitfun-core remote-workspace must add only SSH workspace services and on-demand user Skill package deployment', }, { manifestPath: 'src/crates/assembly/core/Cargo.toml', diff --git a/scripts/core-boundaries/self-test.mjs b/scripts/core-boundaries/self-test.mjs index 93804e6636..9c09b07d7b 100644 --- a/scripts/core-boundaries/self-test.mjs +++ b/scripts/core-boundaries/self-test.mjs @@ -513,6 +513,15 @@ export function runManifestParserSelfTest({ ], ], [servicesCoreManifest, 'session-git', ['local-storage', 'dep:git2']], + [ + servicesCoreManifest, + 'skill-package-sync', + [ + 'dep:anyhow', 'dep:openbitfun-runtime-ports', + 'openbitfun-runtime-ports/workspace-ports', 'dep:sha2', 'dep:futures', 'dep:uuid', + 'dep:tokio', 'tokio/fs', 'tokio/io-util', 'tokio/rt', + ], + ], [ servicesCoreManifest, 'token-usage-statistics', @@ -555,6 +564,7 @@ export function runManifestParserSelfTest({ 'workspace-runtime', 'dep:openbitfun-services-integrations', 'openbitfun-services-integrations/remote-ssh', + 'openbitfun-services-core/skill-package-sync', ], ], [ diff --git a/src/apps/desktop/README.md b/src/apps/desktop/README.md index 27aeaa9ac0..196abfba7e 100644 --- a/src/apps/desktop/README.md +++ b/src/apps/desktop/README.md @@ -642,6 +642,46 @@ setup. Detached Dispatch does not provision WSL connections. SSH port forwarding is unavailable for native WSL targets; use Windows WSL networking to reach a Linux service. +## User Skills in remote workspaces + +In **Settings → Workspace & Git → User Skills in remote workspaces**, choose: + +- **Only inject SKILL.md** (default): load local user instructions without copying + their resources. The loaded Skill and its completed tool card explicitly report + that supporting files have not been synced. +- **Do not run**: omit host-local user Skills from remote session listings and + reject their invocation. Remote project Skills remain available. +- **Full**: on invocation, check the complete package against its remote copy; + transfer files through the bound workspace filesystem and expose a verified + POSIX resource path. SSH workspaces use SFTP over the existing SSH connection, + with bounded parallel file operations. The Skill card shows preparation, + upload and verification progress. + Its completed state confirms package availability, not that runtime dependencies + are installed or that the Skill has executed successfully. + +Full mode needs a workspace file provider that supports streamed reads/writes, +metadata, permissions and staging publication. It does not require Python, an +archive extractor or another synchronization program on the target. Any runtime +dependencies the Skill itself needs must still be available there. Packages are cached +under `/.openbitfun/cache/user-skills`; exclude that cache from +version control if your project tracks `.openbitfun`. Many small files or large +packages can take time to verify and transfer, especially on high-latency SSH +connections. Reuse checks read and hash remote file contents before skipping +uploads. Nested symbolic links and special +files are rejected with an explicit error; files outside the Skill directory are +not copied. Synchronization reuses a matching verified directory. If a cached +copy was modified or damaged, it publishes a new generation and retains the old +directory, so concurrent loads never move another turn's resource path. +Project file operations and commands continue to run on the remote host. +Standalone Markdown Skills in a shared directory must first be converted into +an isolated `name/SKILL.md` package for Full mode; OpenBitFun never copies their +entire shared parent directory implicitly. + +This preference belongs to the OpenBitFun runtime host. A mobile or peer +controller uses that host's setting and Skills; it does not upload its own local +Skill library. Older hosts without this setting show an unsupported state in the +settings UI. + ## Remote SSH file handle errors If writing files and browsing directories both start failing with diff --git a/src/crates/assembly/core/AGENTS.md b/src/crates/assembly/core/AGENTS.md index 2441d2ae9c..58451fdde2 100644 --- a/src/crates/assembly/core/AGENTS.md +++ b/src/crates/assembly/core/AGENTS.md @@ -192,6 +192,17 @@ Narrower local guides already exist for some subtrees: ## Verification +For remote user Skill availability, Markdown-only warnings, package path mapping, +and on-demand deployment, use: + +```bash +cargo test --locked -p openbitfun-core --no-default-features --features agent-runtime,git,remote-workspace --lib remote_user_skills +cargo test --locked -p openbitfun-core --no-default-features --features agent-runtime,git,remote-workspace --lib skill_tool::tests +``` + +These owner fixtures use local providers or mocks. The live SSH package resource +fixture is documented in `services-integrations/AGENTS.md`. + For smart routing, unchanged subagent model policy and resumed subagent choices: ```bash diff --git a/src/crates/assembly/core/Cargo.toml b/src/crates/assembly/core/Cargo.toml index 0e2d10985c..56bd13174c 100644 --- a/src/crates/assembly/core/Cargo.toml +++ b/src/crates/assembly/core/Cargo.toml @@ -475,6 +475,7 @@ workspace-runtime = [ workspace-watch = ["workspace-runtime", "dep:notify"] remote-workspace = [ "workspace-runtime", + "openbitfun-services-core/skill-package-sync", "dep:openbitfun-services-integrations", "openbitfun-services-integrations/remote-ssh", ] diff --git a/src/crates/assembly/core/src/agentic/deep_review/capabilities.rs b/src/crates/assembly/core/src/agentic/deep_review/capabilities.rs index d62c0ce035..f5c0e74142 100644 --- a/src/crates/assembly/core/src/agentic/deep_review/capabilities.rs +++ b/src/crates/assembly/core/src/agentic/deep_review/capabilities.rs @@ -4,7 +4,7 @@ use crate::agentic::agents::{ use crate::agentic::tools::framework::ToolUseContext; use crate::agentic::tools::implementations::skills::get_skill_registry; use crate::util::errors::{OpenBitFunError, OpenBitFunResult}; -use openbitfun_agent_runtime::skills::{SkillData, SkillInfo}; +use openbitfun_agent_runtime::skills::{SkillData, SkillInfo, SkillLocation}; use sha2::{Digest, Sha256}; const CAPABILITY_CATALOG_LIMIT: usize = 24; @@ -202,15 +202,48 @@ pub async fn resolve_review_capability( }); } if let Some(skill_key) = key.strip_prefix("skill:") { + let remote_user_source = if context.is_remote() { + implicitly_invocable_skills(context) + .await + .into_iter() + .find(|skill| skill.key == skill_key && skill.level == SkillLocation::User) + } else { + None + }; + if let Some(info) = &remote_user_source { + let preview = load_discovered_review_skill(context, info).await?; + verify_review_skill_fingerprint(key, &preview, fingerprint)?; + // Remote command hooks are unsupported. Check the same contract as + // SkillTool before spending time deploying an unusable package. + crate::native_hooks::activate_skill_hooks(&preview, context).await?; + } let skill = load_review_skill(context, skill_key).await?; if !is_compatible_review_skill(&skill.dir_name) { return Err(capability_changed_error()); } let guidance = bounded_selected_guidance(&skill.content)?; - ensure_capability_fingerprint( - &ReviewCapabilityDescriptor::new(key, &skill.name, &skill.description, &guidance), + verify_selected_review_skill_fingerprint( + context, + key, + &skill, + remote_user_source.as_ref(), fingerprint, - )?; + ) + .await?; + let guidance = if let Some(info) = remote_user_source { + if info.path != skill.path { + format!( + "Skill resources are relative to '{}'. Use that package root for scripts, references and templates; project operations still use the remote workspace.\n\n{guidance}", + skill.path, + ) + } else if !skill.compatibility_warnings.is_empty() { + format!("{}\n\n{guidance}", skill.compatibility_warnings.join("\n")) + } else { + guidance + } + } else { + guidance + }; return Ok(ResolvedReviewCapability { guidance, preferred_model: None, @@ -266,6 +299,37 @@ pub async fn resolve_review_capability( )) } +fn verify_review_skill_fingerprint( + key: &str, + skill: &SkillData, + fingerprint: &str, +) -> OpenBitFunResult<()> { + let guidance = bounded_selected_guidance(&skill.content)?; + ensure_capability_fingerprint( + &ReviewCapabilityDescriptor::new(key, &skill.name, &skill.description, &guidance), + fingerprint, + ) +} + +async fn verify_selected_review_skill_fingerprint( + context: &ToolUseContext, + key: &str, + skill: &SkillData, + source: Option<&SkillInfo>, + fingerprint: &str, +) -> OpenBitFunResult<()> { + if let Some(info) = source.filter(|info| info.path != skill.path) { + // The admitted catalog fingerprints source guidance. Verify the raw + // deployed snapshot before comparing its path-mapped instructions; + // rereading the local source here would race the completed upload. + let markdown = read_remote_review_skill_markdown(context, &skill.path, info).await?; + let deployed_preview = parse_review_skill_preview(info, &markdown)?; + verify_review_skill_fingerprint(key, &deployed_preview, fingerprint) + } else { + verify_review_skill_fingerprint(key, skill, fingerprint) + } +} + fn explicit_preferred_model(agent: &AgentInfo) -> Option { agent .model @@ -303,11 +367,12 @@ async fn load_review_skill( .map(|workspace| workspace.root_path_string()) .unwrap_or_default(); registry - .find_and_load_skill_by_key_for_remote_workspace( + .find_and_load_skill_by_key_for_remote_workspace_with_progress( skill_key, fs, &root, context.agent_type.as_deref(), + Some(crate::agentic::tools::implementations::skills::remote_skill_progress::skill_load_progress(context, "AgentSpawn")), ) .await } else { @@ -325,28 +390,42 @@ async fn load_discovered_review_skill( context: &ToolUseContext, info: &SkillInfo, ) -> OpenBitFunResult { - let skill_file = format!("{}/SKILL.md", info.path.trim_end_matches(['/', '\\'])); - let markdown = if context.is_remote() { - context - .ws_fs() - .ok_or_else(|| { - OpenBitFunError::tool("Remote review skill loading is unavailable".to_string()) - })? - .read_file_text(&skill_file) - .await - .map_err(|error| { - OpenBitFunError::tool(format!("Failed to read review skill: {error}")) - })? + let entry_file = info.entry_file.as_deref().unwrap_or("SKILL.md"); + // Catalog construction previews guidance; deployment happens only when a + // capability is selected through the registry above. + let markdown = if context.is_remote() && info.level == SkillLocation::Project { + read_remote_review_skill_markdown(context, &info.path, info).await? } else { - tokio::fs::read_to_string(&skill_file) + tokio::fs::read_to_string(std::path::Path::new(&info.path).join(entry_file)) .await .map_err(|error| { OpenBitFunError::tool(format!("Failed to read review skill: {error}")) })? }; + parse_review_skill_preview(info, &markdown) +} + +async fn read_remote_review_skill_markdown( + context: &ToolUseContext, + path: &str, + info: &SkillInfo, +) -> OpenBitFunResult { + let entry_file = info.entry_file.as_deref().unwrap_or("SKILL.md"); + let skill_file = format!("{}/{}", path.trim_end_matches('/'), entry_file); + context + .ws_fs() + .ok_or_else(|| { + OpenBitFunError::tool("Remote review skill loading is unavailable".to_string()) + })? + .read_file_text(&skill_file) + .await + .map_err(|error| OpenBitFunError::tool(format!("Failed to read review skill: {error}"))) +} + +fn parse_review_skill_preview(info: &SkillInfo, markdown: &str) -> OpenBitFunResult { let mut data = SkillData::from_markdown_for_source_slot( - info.path.clone(), - &markdown, + info.parser_path(), + markdown, info.level, true, info.parser_source_slot(), @@ -556,6 +635,122 @@ mod tests { assert!(!rendered.contains("full guidance")); } + #[tokio::test] + async fn remote_catalog_previews_local_user_markdown_without_remote_io_or_deployment() { + let temp = tempfile::tempdir().unwrap(); + std::fs::write( + temp.path().join("custom-entry.md"), + "---\nname: review-local\ndescription: User guidance\n---\nCheck scripts/check.py.\n", + ) + .unwrap(); + let info: SkillInfo = serde_json::from_value(serde_json::json!({ + "key": "user::openbitfun-user::code-review-local", + "name": "review-local", + "description": "User guidance", + "path": temp.path().to_str().unwrap(), + "entryFile": "custom-entry.md", + "level": "user", + "sourceSlot": "openbitfun-user", + "dirName": "code-review-local" + })) + .unwrap(); + let identity = crate::service::remote_ssh::workspace_state::workspace_session_identity( + "/remote/project", + Some("catalog-test-connection"), + Some("catalog-test-host"), + ) + .unwrap(); + let mut context = local_tool_context(temp.path().to_path_buf()); + context.workspace = Some(WorkspaceBinding::new_remote( + None, + "/remote/project".into(), + "catalog-test-connection".into(), + "Remote".into(), + identity, + )); + assert!(context.ws_fs().is_none()); + let data = load_discovered_review_skill(&context, &info).await.unwrap(); + assert_eq!(data.content.trim(), "Check scripts/check.py."); + assert_eq!(data.key, info.key); + let mut project = info; + project.level = SkillLocation::Project; + assert!(load_discovered_review_skill(&context, &project) + .await + .is_err()); + } + + #[tokio::test] + async fn deployed_review_guidance_verifies_raw_snapshot_before_path_mapping() { + let temp = tempfile::tempdir().unwrap(); + let source_dir = temp.path().join("source"); + let remote_dir = temp.path().join("remote"); + std::fs::create_dir(&source_dir).unwrap(); + std::fs::create_dir(&remote_dir).unwrap(); + let source_path = source_dir.to_string_lossy().to_string(); + let remote_path = remote_dir.to_string_lossy().to_string(); + let info: SkillInfo = serde_json::from_value(serde_json::json!({ + "key": "user::openbitfun-user::code-review-local", + "name": "review-local", + "description": "User guidance", + "path": source_path, + "level": "user", + "sourceSlot": "openbitfun-user", + "dirName": "code-review-local" + })) + .unwrap(); + let markdown = format!( + "---\nname: review-local\ndescription: User guidance\n---\nRead {}/references/guide.md.\n", + info.path + ); + std::fs::write(source_dir.join("SKILL.md"), &markdown).unwrap(); + std::fs::write(remote_dir.join("SKILL.md"), &markdown).unwrap(); + let preview = parse_review_skill_preview(&info, &markdown).unwrap(); + let key = format!("skill:{}", info.key); + let descriptor = ReviewCapabilityDescriptor::new( + &key, + &preview.name, + &preview.description, + &bounded_selected_guidance(&preview.content).unwrap(), + ); + let mut loaded = preview; + loaded.content = loaded.content.replace(&info.path, &remote_path); + loaded.path = remote_path.clone(); + assert!(verify_review_skill_fingerprint(&key, &loaded, descriptor.fingerprint()).is_err()); + + let mut context = local_tool_context(temp.path().to_path_buf()); + context.runtime_handles = openbitfun_runtime_ports::ToolRuntimeHandles::new( + Some(crate::agentic::workspace::local_workspace_services( + remote_path, + )), + None, + ); + // A later local edit does not change the snapshot already deployed. + std::fs::write(source_dir.join("SKILL.md"), "changed local source").unwrap(); + verify_selected_review_skill_fingerprint( + &context, + &key, + &loaded, + Some(&info), + descriptor.fingerprint(), + ) + .await + .unwrap(); + std::fs::write( + remote_dir.join("SKILL.md"), + markdown.replace("Read ", "Execute "), + ) + .unwrap(); + assert!(verify_selected_review_skill_fingerprint( + &context, + &key, + &loaded, + Some(&info), + descriptor.fingerprint(), + ) + .await + .is_err()); + } + #[tokio::test] async fn catalog_uses_discovered_claude_skill_and_restores_it_after_import_undo() { let temp = tempfile::tempdir().expect("temporary workspace"); diff --git a/src/crates/assembly/core/src/agentic/tools/implementations/skill_tool.rs b/src/crates/assembly/core/src/agentic/tools/implementations/skill_tool.rs index 4b706fc6a1..2e0543fa46 100644 --- a/src/crates/assembly/core/src/agentic/tools/implementations/skill_tool.rs +++ b/src/crates/assembly/core/src/agentic/tools/implementations/skill_tool.rs @@ -12,6 +12,8 @@ use log::debug; use openbitfun_services_core::markdown::expand_prompt_template_arguments_with_names; use serde_json::{json, Value}; +use super::skills::remote_skill_progress::skill_load_progress; + // Use skills module use super::skills::{get_skill_registry, render_loaded_skill_for_assistant}; @@ -65,9 +67,7 @@ Important: ) .await } else { - registry - .get_resolved_skills_xml_for_workspace(None, ctx.agent_type.as_deref()) - .await + vec!["Remote workspace Skill loading is unavailable because its filesystem provider is missing. No local fallback is available.".to_string()] } } Some(ctx) => { @@ -161,7 +161,8 @@ impl Tool for SkillTool { "description": {"type": "string"}, "location": {"type": "string"}, "content": {"type": "string"}, - "compatibility_warnings": {"type": "array", "items": {"type": "string"}} + "compatibility_warnings": {"type": "array", "items": {"type": "string"}}, + "resource_availability": {"type": ["string", "null"], "enum": ["markdown_only", "remote_package", null]} }, "required": ["success", "skill_name", "skill_key", "content"], "additionalProperties": false @@ -236,7 +237,7 @@ impl Tool for SkillTool { fn render_tool_use_message(&self, input: &Value, _options: &ToolRenderOptions) -> String { if let Some(command) = input.get("command").and_then(|v| v.as_str()) { - format!("The \"{}\" skill is loaded.", command) + format!("Loading the \"{}\" skill...", command) } else { "Loading skill...".to_string() } @@ -258,6 +259,7 @@ impl Tool for SkillTool { let registry = get_skill_registry(); let use_stable_key = skill_name.starts_with("user::") || skill_name.starts_with("project::"); + let progress = skill_load_progress(context, "Skill"); let mut skill_data = if context.is_remote() { if let Some(ws_fs) = context.ws_fs() { let root = context @@ -267,41 +269,29 @@ impl Tool for SkillTool { .unwrap_or_default(); if use_stable_key { registry - .find_and_load_skill_by_key_for_remote_workspace( + .find_and_load_skill_by_key_for_remote_workspace_with_progress( skill_name, ws_fs, &root, context.agent_type.as_deref(), + Some(progress.clone()), ) .await? } else { registry - .find_and_load_skill_for_remote_workspace( + .find_and_load_skill_for_remote_workspace_with_progress( skill_name, ws_fs, &root, context.agent_type.as_deref(), + Some(progress.clone()), ) .await? } } else { - if use_stable_key { - registry - .find_and_load_skill_by_key_for_workspace( - skill_name, - None, - context.agent_type.as_deref(), - ) - .await? - } else { - registry - .find_and_load_skill_for_workspace( - skill_name, - None, - context.agent_type.as_deref(), - ) - .await? - } + return Err(OpenBitFunError::tool( + "Remote workspace Skill loading requires its bound filesystem provider; no local fallback is available.".to_string(), + )); } } else { if use_stable_key { @@ -346,6 +336,7 @@ impl Tool for SkillTool { "location": location_str, "content": skill_data.content, "compatibility_warnings": skill_data.compatibility_warnings, + "resource_availability": skill_data.resource_availability, "success": true }), result_for_assistant: Some(result_for_assistant), @@ -1131,8 +1122,15 @@ Use the remote project skill. ), }; - let results = SkillTool::new() - .call_impl(&json!({ "command": "create-agent" }), &context) + use super::super::skills::remote_user_skills::TEST_REMOTE_USER_SKILLS_MODE; + use crate::service::config::types::RemoteUserSkillsMode; + let tool = SkillTool::new(); + let input = json!({ "command": "create-agent" }); + let results = TEST_REMOTE_USER_SKILLS_MODE + .scope( + RemoteUserSkillsMode::SkillMdOnly, + tool.call_impl(&input, &context), + ) .await .expect("explicit create-agent invocation should load the local built-in skill"); @@ -1146,6 +1144,7 @@ Use the remote project skill. }; assert_eq!(data["skill_name"], "create-agent"); assert_eq!(data["location"], "user"); + assert_eq!(data["resource_availability"], "markdown_only"); assert!(data["content"] .as_str() .unwrap_or_default() @@ -1155,6 +1154,45 @@ Use the remote project skill. assert!(assistant.contains("\n")); assert!(assistant.contains("# Create an OpenBitFun Agent")); assert!(!assistant.contains("from stable key")); + assert!(assistant.contains("NOT available in this remote workspace")); + + TEST_REMOTE_USER_SKILLS_MODE + .scope(RemoteUserSkillsMode::Disabled, async { + let listing = SkillTool::build_available_skills_context_section(Some(&context)) + .await + .unwrap(); + assert!(listing.contains("remote-only-skill-for-test")); + assert!(!listing.contains("user::")); + assert!(tool.call_impl(&input, &context).await.is_err()); + let error = tool + .call_impl( + &json!({"command": "user::openbitfun-system::create-agent"}), + &context, + ) + .await + .unwrap_err(); + assert!(error.to_string().contains("disabled")); + }) + .await; + + let error = TEST_REMOTE_USER_SKILLS_MODE + .scope(RemoteUserSkillsMode::Full, tool.call_impl(&input, &context)) + .await + .unwrap_err(); + #[cfg(feature = "remote-workspace")] + assert!(error + .to_string() + .contains("Failed to deploy complete user Skill package")); + #[cfg(not(feature = "remote-workspace"))] + assert!(error + .to_string() + .contains("does not support complete user Skill deployment")); + + let mut missing_provider = context; + missing_provider.runtime_handles = + openbitfun_runtime_ports::ToolRuntimeHandles::new(None, None); + let error = tool.call_impl(&input, &missing_provider).await.unwrap_err(); + assert!(error.to_string().contains("bound filesystem provider")); } #[tokio::test] diff --git a/src/crates/assembly/core/src/agentic/tools/implementations/skills/mod.rs b/src/crates/assembly/core/src/agentic/tools/implementations/skills/mod.rs index 394bd6da63..4aca6dc42b 100644 --- a/src/crates/assembly/core/src/agentic/tools/implementations/skills/mod.rs +++ b/src/crates/assembly/core/src/agentic/tools/implementations/skills/mod.rs @@ -7,6 +7,9 @@ pub mod catalog; pub mod mode_overrides; pub mod policy; pub mod registry; +pub(crate) mod remote_skill_progress; +pub(crate) mod remote_user_skills; +pub use remote_user_skills::SkillLoadProgress; pub mod resolver; mod runtime_settings; #[cfg(feature = "file-watch")] diff --git a/src/crates/assembly/core/src/agentic/tools/implementations/skills/registry.rs b/src/crates/assembly/core/src/agentic/tools/implementations/skills/registry.rs index 1d22cd3a06..ab3fd929db 100644 --- a/src/crates/assembly/core/src/agentic/tools/implementations/skills/registry.rs +++ b/src/crates/assembly/core/src/agentic/tools/implementations/skills/registry.rs @@ -8,6 +8,10 @@ use super::mode_overrides::{ load_globally_disabled_project_skills, load_globally_disabled_user_skills, load_user_mode_skill_overrides, SkillPolicyWorkspace, UserModeSkillOverrides, }; +use super::remote_user_skills::{ + check_remote_user_skill_mode, prepare_remote_user_skill, remote_user_skills_mode, + SkillLoadProgress, +}; use super::runtime_settings::apply_builtin_runtime_settings; #[cfg(feature = "file-watch")] use super::source_cache::{LocalSkillWatchMonitor, LocalSkillWatchRoot, VersionedSnapshotCache}; @@ -21,6 +25,7 @@ use crate::external_sources::{ LocalConfiguredSkillRootContribution, }; use crate::infrastructure::get_path_manager_arc; +use crate::service::config::types::RemoteUserSkillsMode; use crate::util::errors::{OpenBitFunError, OpenBitFunResult}; use futures::{stream, StreamExt}; use log::{debug, error, warn}; @@ -1516,10 +1521,44 @@ impl SkillRegistry { fs: &dyn WorkspaceFileSystem, remote_root: &str, ) -> SkillCandidateScan { - let (user, project) = tokio::join!( - self.scan_skill_candidates_with_diagnostics_for_workspace(None), + let mode = remote_user_skills_mode().await; + self.scan_remote_skill_candidates_with_mode(fs, remote_root, mode) + .await + } + + async fn scan_remote_skill_candidates_with_mode( + &self, + fs: &dyn WorkspaceFileSystem, + remote_root: &str, + mode: OpenBitFunResult, + ) -> SkillCandidateScan { + let (user, mut project) = tokio::join!( + async { + match &mode { + Ok(RemoteUserSkillsMode::SkillMdOnly | RemoteUserSkillsMode::Full) => { + self.scan_skill_candidates_with_diagnostics_for_workspace(None) + .await + } + _ => SkillCandidateScan::default(), + } + }, Self::scan_remote_project_skills(fs, remote_root), ); + let notice = match &mode { + Err(error) => Some(format!("Local user Skills are unavailable: {error}")), + Ok(RemoteUserSkillsMode::Unsupported(value)) => Some(format!( + "Local user Skills are unavailable: unsupported remote user Skills mode '{value}'" + )), + _ => None, + }; + if let Some(message) = notice { + project.diagnostics.push(SkillScanDiagnostic { + path: "workspace.remote_user_skills".into(), + source_id: "openbitfun".into(), + message, + unsupported_field: None, + }); + } Self::merge_remote_skill_scans(user, project) } @@ -2052,6 +2091,24 @@ impl SkillRegistry { fs: &dyn WorkspaceFileSystem, remote_root: &str, agent_type: Option<&str>, + ) -> OpenBitFunResult { + self.find_and_load_skill_for_remote_workspace_with_progress( + skill_name, + fs, + remote_root, + agent_type, + None, + ) + .await + } + + pub async fn find_and_load_skill_for_remote_workspace_with_progress( + &self, + skill_name: &str, + fs: &dyn WorkspaceFileSystem, + remote_root: &str, + agent_type: Option<&str>, + progress: Option, ) -> OpenBitFunResult { let info = self .find_skill_info_for_explicit_invocation_remote_workspace( @@ -2071,6 +2128,7 @@ impl SkillRegistry { info.parser_source_slot(), ) .map_err(|error| OpenBitFunError::tool(error.to_string()))?; + let parser_source_slot = info.parser_source_slot().to_string(); data.path = info.path; data.entry_file = info.entry_file; data.key = info.key; @@ -2078,6 +2136,8 @@ impl SkillRegistry { data.source_id = info.source_id; data.source_label = info.source_label; data.dir_name = info.dir_name; + prepare_remote_user_skill(&mut data, fs, remote_root, progress, &parser_source_slot) + .await?; apply_builtin_runtime_settings(&mut data).await?; Ok(data) } @@ -2089,6 +2149,27 @@ impl SkillRegistry { remote_root: &str, agent_type: Option<&str>, ) -> OpenBitFunResult { + self.find_and_load_skill_by_key_for_remote_workspace_with_progress( + skill_key, + fs, + remote_root, + agent_type, + None, + ) + .await + } + + pub async fn find_and_load_skill_by_key_for_remote_workspace_with_progress( + &self, + skill_key: &str, + fs: &dyn WorkspaceFileSystem, + remote_root: &str, + agent_type: Option<&str>, + progress: Option, + ) -> OpenBitFunResult { + if skill_key.starts_with("user::") { + check_remote_user_skill_mode(&remote_user_skills_mode().await?)?; + } let candidates = self .scan_skill_candidates_for_remote_workspace(fs, remote_root) .await; @@ -2115,6 +2196,7 @@ impl SkillRegistry { info.parser_source_slot(), ) .map_err(|error| OpenBitFunError::tool(error.to_string()))?; + let parser_source_slot = info.parser_source_slot().to_string(); data.path = info.path; data.entry_file = info.entry_file; data.key = info.key; @@ -2122,6 +2204,8 @@ impl SkillRegistry { data.source_id = info.source_id; data.source_label = info.source_label; data.dir_name = info.dir_name; + prepare_remote_user_skill(&mut data, fs, remote_root, progress, &parser_source_slot) + .await?; apply_builtin_runtime_settings(&mut data).await?; Ok(data) } @@ -2195,6 +2279,58 @@ impl SkillRegistry { } } +#[cfg(test)] +mod remote_user_skills_policy_tests { + use super::*; + use crate::agentic::workspace::LocalWorkspaceFs; + + #[tokio::test] + async fn remote_user_skills_disabled_or_unavailable_keeps_only_project_candidates() { + let temp = tempfile::tempdir().unwrap(); + let project_skill = temp.path().join(".openbitfun/skills/project-guide"); + std::fs::create_dir_all(&project_skill).unwrap(); + std::fs::write( + project_skill.join("SKILL.md"), + "---\nname: project-guide\ndescription: Remote project guide\n---\nRead the project.", + ) + .unwrap(); + let registry = SkillRegistry::new(); + for mode in [ + Ok(RemoteUserSkillsMode::Disabled), + Ok(RemoteUserSkillsMode::Unsupported("future".into())), + Err(OpenBitFunError::config("unreadable preference")), + ] { + let should_warn = !matches!(mode, Ok(RemoteUserSkillsMode::Disabled)); + let scan = registry + .scan_remote_skill_candidates_with_mode( + &LocalWorkspaceFs, + &temp.path().to_string_lossy(), + mode, + ) + .await; + assert_eq!(scan.candidates.len(), 1); + assert!(scan + .candidates + .iter() + .all(|candidate| candidate.info.level == SkillLocation::Project)); + assert_eq!(scan.candidates[0].info.name, "project-guide"); + let prompt: String = scan + .candidates + .iter() + .map(|candidate| candidate.info.to_xml_desc()) + .collect(); + assert!(prompt.contains("project-guide")); + assert!(!prompt.contains("user::")); + assert_eq!( + scan.diagnostics + .iter() + .any(|notice| notice.path == "workspace.remote_user_skills"), + should_warn + ); + } + } +} + #[cfg(all(test, feature = "external-sources"))] mod opencode_configured_skill_tests { use super::{SkillRegistry, SkillRootEntry}; diff --git a/src/crates/assembly/core/src/agentic/tools/implementations/skills/remote_skill_progress.rs b/src/crates/assembly/core/src/agentic/tools/implementations/skills/remote_skill_progress.rs new file mode 100644 index 0000000000..ca0bbb1f24 --- /dev/null +++ b/src/crates/assembly/core/src/agentic/tools/implementations/skills/remote_skill_progress.rs @@ -0,0 +1,36 @@ +//! Forward package preparation status through the invoking tool's existing stream. + +use super::SkillLoadProgress; +use crate::agentic::tools::framework::ToolUseContext; +use crate::infrastructure::events::event_system::{get_global_event_system, BackendEvent}; +use crate::util::types::event::ToolExecutionProgressInfo; +use std::sync::Arc; + +pub(crate) fn skill_load_progress( + context: &ToolUseContext, + tool_name: &'static str, +) -> SkillLoadProgress { + let tool_call_id = context.tool_call_id.clone(); + // Package scanning invokes this callback on a blocking worker. Capture the + // runtime here so that those updates can use the same event transport. + let runtime = tokio::runtime::Handle::current(); + Arc::new(move |progress_message, percentage| { + let Some(tool_use_id) = tool_call_id.clone() else { + return; + }; + let event_system = get_global_event_system(); + let event = BackendEvent::ToolExecutionProgress(ToolExecutionProgressInfo { + tool_use_id, + tool_name: tool_name.to_string(), + progress_message, + percentage, + timestamp: std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs(), + }); + runtime.spawn(async move { + let _ = event_system.emit(event).await; + }); + }) +} diff --git a/src/crates/assembly/core/src/agentic/tools/implementations/skills/remote_user_skills.rs b/src/crates/assembly/core/src/agentic/tools/implementations/skills/remote_user_skills.rs new file mode 100644 index 0000000000..69ce6dd71c --- /dev/null +++ b/src/crates/assembly/core/src/agentic/tools/implementations/skills/remote_user_skills.rs @@ -0,0 +1,505 @@ +//! Execution-host preference and on-demand deployment of local user Skills. + +use super::types::{SkillData, SkillLocation}; +use crate::agentic::workspace::WorkspaceFileSystem; +use crate::service::config::global::GlobalConfigManager; +use crate::service::config::types::{RemoteUserSkillsMode, WorkspaceConfig}; +use crate::util::errors::{OpenBitFunError, OpenBitFunResult}; +use openbitfun_agent_runtime::skills::SkillResourceAvailability; +use std::sync::Arc; + +pub type SkillLoadProgress = Arc) + Send + Sync>; + +#[cfg(test)] +tokio::task_local! { + // Keep per-call policy fixtures isolated from the process-global settings. + pub(crate) static TEST_REMOTE_USER_SKILLS_MODE: RemoteUserSkillsMode; +} + +pub(super) async fn remote_user_skills_mode() -> OpenBitFunResult { + #[cfg(test)] + if let Ok(mode) = TEST_REMOTE_USER_SKILLS_MODE.try_with(Clone::clone) { + return Ok(mode); + } + // Read the runtime host's preference, including when a peer/mobile drives it. + let service = GlobalConfigManager::get_service().await?; + let config: WorkspaceConfig = service.get_config(Some("workspace")).await?; + Ok(config.remote_user_skills) +} + +pub(super) fn check_remote_user_skill_mode(mode: &RemoteUserSkillsMode) -> OpenBitFunResult<()> { + match mode { + RemoteUserSkillsMode::Disabled => Err(OpenBitFunError::tool( + "Local user Skills are disabled in remote workspaces. Change Settings > Workspace & Git to enable them.".to_string(), + )), + RemoteUserSkillsMode::Unsupported(value) => Err(OpenBitFunError::tool(format!( + "Unsupported remote user Skills mode '{value}'. Select a supported mode in Settings > Workspace & Git." + ))), + _ => Ok(()), + } +} + +pub(super) async fn prepare_remote_user_skill( + skill: &mut SkillData, + fs: &dyn WorkspaceFileSystem, + remote_root: &str, + progress: Option, + parser_source_slot: &str, +) -> OpenBitFunResult<()> { + if skill.location == SkillLocation::Project { + return Ok(()); + } + let mode = remote_user_skills_mode().await?; + prepare_remote_user_skill_with_mode(skill, fs, remote_root, progress, &mode, parser_source_slot) + .await +} + +async fn prepare_remote_user_skill_with_mode( + skill: &mut SkillData, + fs: &dyn WorkspaceFileSystem, + remote_root: &str, + progress: Option, + mode: &RemoteUserSkillsMode, + parser_source_slot: &str, +) -> OpenBitFunResult<()> { + check_remote_user_skill_mode(mode)?; + if *mode == RemoteUserSkillsMode::SkillMdOnly { + skill.resource_availability = Some(SkillResourceAvailability::MarkdownOnly); + skill.compatibility_warnings.push( + "Only SKILL.md was loaded from the runtime host. Its local scripts, references and templates are NOT available in this remote workspace. Do not read or execute those local paths remotely. Select Full in Settings > Workspace & Git to deploy the package before using its resources.".to_string(), + ); + return Ok(()); + } + if skill + .entry_file + .as_deref() + .is_some_and(|entry| entry != "SKILL.md") + { + return Err(OpenBitFunError::tool( + "Full remote loading is unsupported for standalone Markdown Skills without an isolated package directory. Move this Skill and its resources into a dedicated directory with SKILL.md, or select Only inject SKILL.md. The shared parent directory was not copied.", + )); + } + #[cfg(feature = "remote-workspace")] + { + use openbitfun_services_core::skill_package_sync::{ + sync_local_skill_package, SkillPackageSyncPhase, + }; + let notify = progress.map(|callback| { + Arc::new(move |state: openbitfun_services_core::skill_package_sync::SkillPackageSyncProgress| { + let stage = match state.phase { + SkillPackageSyncPhase::Scanning => "Preparing local Skill package", + SkillPackageSyncPhase::Checking => "Checking remote Skill package", + SkillPackageSyncPhase::Uploading => "Uploading Skill package", + SkillPackageSyncPhase::Verifying => "Verifying and publishing remote Skill package", + SkillPackageSyncPhase::Complete => "Remote Skill package ready", + }; + let percentage = (state.phase == SkillPackageSyncPhase::Uploading && state.total_bytes > 0) + .then(|| (state.transferred_bytes as f32 / state.total_bytes as f32 * 100.0).min(100.0)); + let transfer = if state.phase == SkillPackageSyncPhase::Uploading { + format!(", {} bytes uploaded", state.transferred_bytes) + } else { String::new() }; + callback(format!("{stage} ({} files, {} bytes{transfer}); large packages may take a while", state.file_count, state.total_bytes), percentage); + }) as Arc + }); + let cache_root = format!( + "{}/.openbitfun/cache/user-skills", + remote_root.trim_end_matches('/') + ); + let deployment = sync_local_skill_package( + std::path::Path::new(&skill.path), + &cache_root, + remote_root, + &skill.key, + fs, + notify, + ) + .await + .map_err(|error| { + OpenBitFunError::tool(format!( + "Failed to deploy complete user Skill package: {error:#}" + )) + })?; + + // Read the deployed snapshot so the instructions and resources agree even + // if the source directory changed during preparation. Paths stay POSIX. + let entry = skill.entry_file.as_deref().unwrap_or("SKILL.md"); + let markdown = fs + .read_file_text(&format!("{}/{entry}", deployment.path)) + .await + .map_err(|error| { + OpenBitFunError::tool(format!("Failed to read deployed Skill: {error}")) + })?; + // Dialects can derive an omitted name from the original directory/file. + // The hash-named deployment is only an IO location, never Skill identity. + let parser_path = if entry == "SKILL.md" { + skill.path.clone() + } else { + format!("{}/{entry}", skill.path.trim_end_matches(['/', '\\'])) + }; + let mut deployed = SkillData::from_markdown_for_source_slot( + parser_path, + &markdown, + skill.location, + true, + parser_source_slot, + ) + .map_err(|error| OpenBitFunError::tool(error.to_string()))?; + deployed.content = map_package_paths(&deployed.content, &skill.path, &deployment.path); + skill.content = deployed.content; + skill.name = deployed.name; + skill.description = deployed.description; + skill.hooks = deployed.hooks; + skill.argument_names = deployed.argument_names; + skill.argument_hint = deployed.argument_hint; + skill.allow_implicit_invocation = deployed.allow_implicit_invocation; + skill.allow_user_invocation = deployed.allow_user_invocation; + skill.compatibility_warnings = deployed.compatibility_warnings; + skill.path = deployment.path; + skill.resource_availability = Some(SkillResourceAvailability::RemotePackage); + skill.compatibility_warnings.push( + "The complete Skill package is available at the remote path above. Use POSIX paths there; commands and project operations still run on the remote host. Runtime dependencies and files outside this Skill package are not installed or copied.".to_string(), + ); + Ok(()) + } + #[cfg(not(feature = "remote-workspace"))] + { + let _ = (skill, fs, remote_root, progress, parser_source_slot); + Err(OpenBitFunError::tool( + "This runtime build does not support complete user Skill deployment to remote workspaces.".to_string(), + )) + } +} + +/// Rewrite references rooted in the source package, including Windows paths. +/// Other absolute paths remain unchanged: they are external dependencies. +#[cfg(any(test, feature = "remote-workspace"))] +fn map_package_paths(content: &str, source: &str, target: &str) -> String { + let windows_source = (source.as_bytes().get(1) == Some(&b':') + && source + .as_bytes() + .first() + .is_some_and(u8::is_ascii_alphabetic)) + || source.starts_with("\\\\"); + let mut prefixes = vec![source.to_string(), source.replace('\\', "/")]; + prefixes.retain(|prefix| !prefix.is_empty()); + prefixes.sort(); + prefixes.dedup(); + let mut mapped = String::new(); + let mut offset = 0; + while let Some((index, prefix)) = prefixes + .iter() + .filter_map(|prefix| { + content[offset..] + .find(prefix) + .map(|index| (offset + index, prefix)) + }) + .min_by_key(|(index, _)| *index) + { + mapped.push_str(&content[offset..index]); + let after_start = index + prefix.len(); + let after = &content[after_start..]; + let before = &content[..index]; + if before.chars().next_back().is_some_and(|c| { + !c.is_whitespace() && !matches!(c, '\'' | '"' | '`' | '(' | '<' | '[' | '=' | ':') + }) { + mapped.push_str(prefix); + offset = after_start; + continue; + } + let boundary = |c: char| { + c.is_whitespace() || matches!(c, '\'' | '"' | '`' | ')' | ']' | '>' | ';' | ',') + }; + if after.starts_with('/') || (windows_source && after.starts_with('\\')) { + mapped.push_str(target); + // Quoted resource names may contain spaces in any path component. + let delimiter = enclosing_path_quote(before).or_else(|| { + before + .chars() + .next_back() + .filter(|c| matches!(c, '\'' | '"' | '`' | '(' | '<')) + }); + let end = match delimiter { + Some('(') => after.find(')'), + Some('<') => after.find('>'), + Some(quote) => after.find(quote), + None => after.find(boundary), + } + .unwrap_or(after.len()); + if windows_source { + mapped.push_str(&after[..end].replace('\\', "/")); + } else { + // POSIX command text may escape spaces with backslashes. They + // are shell syntax, not Windows directory separators. + mapped.push_str(&after[..end]); + } + offset = after_start + end; + } else if after.is_empty() || after.starts_with(boundary) { + mapped.push_str(target); + offset = after_start; + } else { + mapped.push_str(prefix); + offset = after_start; + } + } + mapped.push_str(&content[offset..]); + mapped +} + +#[cfg(any(test, feature = "remote-workspace"))] +fn enclosing_path_quote(before: &str) -> Option { + let mut quote = None; + let mut inline_code = false; + let mut escaped = false; + let mut previous = None::; + let mut chars = before + .rsplit('\n') + .next() + .unwrap_or_default() + .chars() + .peekable(); + while let Some(current) = chars.next() { + if escaped { + escaped = false; + } else if quote == Some('\'') { + if current == '\'' { + quote = None; + } + } else if current == '\\' { + escaped = true; + } else if let Some(active) = quote { + if current == active { + quote = None; + } + } else if matches!(current, '\'' | '"') { + // A possessive apostrophe in prose does not open a command token. + if current != '\'' + || !previous.is_some_and(char::is_alphanumeric) + || !chars.peek().is_some_and(|next| next.is_alphanumeric()) + { + quote = Some(current); + } + } else if current == '`' { + inline_code = !inline_code; + } + previous = Some(current); + } + quote.or(inline_code.then_some('`')) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn user_skill(path: &str, markdown: &str, source: &str) -> SkillData { + SkillData::from_markdown_for_source_slot( + path.into(), + markdown, + SkillLocation::User, + true, + source, + ) + .unwrap() + } + + #[tokio::test] + async fn markdown_only_warns_without_accessing_or_deploying_resources() { + let mut skill = user_skill( + "/nonexistent/local-skill", + "---\nname: demo\ndescription: demo\n---\nRun scripts/hello.py", + "openbitfun", + ); + let original = skill.content.clone(); + prepare_remote_user_skill_with_mode( + &mut skill, + &crate::agentic::workspace::LocalWorkspaceFs, + "/nonexistent/remote", + None, + &RemoteUserSkillsMode::SkillMdOnly, + "openbitfun", + ) + .await + .unwrap(); + assert_eq!(skill.content, original); + assert_eq!(skill.path, "/nonexistent/local-skill"); + assert_eq!( + skill.resource_availability, + Some(SkillResourceAvailability::MarkdownOnly) + ); + assert!(skill + .compatibility_warnings + .iter() + .any(|warning| warning.contains("NOT available"))); + let error = prepare_remote_user_skill_with_mode( + &mut skill, + &crate::agentic::workspace::LocalWorkspaceFs, + "/nonexistent/remote", + None, + &RemoteUserSkillsMode::Disabled, + "openbitfun", + ) + .await + .unwrap_err(); + assert!(error.to_string().contains("disabled")); + } + + #[tokio::test] + async fn full_mode_rejects_flat_markdown_without_copying_shared_parent() { + let mut skill = user_skill( + "/user/shared-skills", + "---\nname: standalone\ndescription: standalone\n---\nRead references/shared.md", + "openbitfun", + ); + skill.entry_file = Some("standalone.md".into()); + let error = prepare_remote_user_skill_with_mode( + &mut skill, + &crate::agentic::workspace::LocalWorkspaceFs, + "/unavailable/remote", + None, + &RemoteUserSkillsMode::Full, + "openbitfun", + ) + .await + .unwrap_err(); + assert!(error + .to_string() + .contains("shared parent directory was not copied")); + assert_eq!(skill.path, "/user/shared-skills"); + } + + #[cfg(all(unix, feature = "remote-workspace"))] + #[tokio::test] + async fn full_user_skill_loads_deployed_snapshot_and_preserves_dialect_identity() { + use crate::agentic::workspace::{LocalWorkspaceFs, LocalWorkspaceShell, WorkspaceShell}; + let temp = tempfile::tempdir().unwrap(); + let source = temp.path().join("local user/review-package"); + let target = temp.path().join("remote project"); + std::fs::create_dir_all(source.join("scripts")).unwrap(); + std::fs::create_dir_all(source.join("references")).unwrap(); + std::fs::create_dir_all(&target).unwrap(); + let original = format!("---\ndescription: Review package\narguments: target\n---\nRead {}/references/guide.txt and run scripts/hello.sh for $target.", source.display()); + std::fs::write(source.join("SKILL.md"), &original).unwrap(); + std::fs::write(source.join("references/guide.txt"), "deployed reference").unwrap(); + std::fs::write(source.join("scripts/hello.sh"), "pwd -P\n").unwrap(); + let mut skill = user_skill(&source.to_string_lossy(), &original, "claude"); + let key = skill.key.clone(); + prepare_remote_user_skill_with_mode( + &mut skill, + &LocalWorkspaceFs, + &target.to_string_lossy(), + None, + &RemoteUserSkillsMode::Full, + "claude", + ) + .await + .unwrap(); + assert_eq!(skill.name, "review-package"); + assert_eq!( + skill.resource_availability, + Some(SkillResourceAvailability::RemotePackage) + ); + assert_eq!(skill.key, key); + assert_eq!(skill.argument_names, ["target"]); + assert!(skill + .path + .starts_with(&target.to_string_lossy().to_string())); + assert!(!skill.path.contains('\\')); + assert!(skill + .content + .contains(&format!("{}/references/guide.txt", skill.path))); + assert!(!skill + .content + .contains(&source.to_string_lossy().to_string())); + assert_eq!( + LocalWorkspaceFs + .read_file_text(&format!("{}/references/guide.txt", skill.path)) + .await + .unwrap(), + "deployed reference" + ); + let shell = LocalWorkspaceShell::new(target.to_string_lossy().into_owned()); + let (stdout, stderr, exit) = shell + .exec(&format!("sh '{}/scripts/hello.sh'", skill.path), None) + .await + .unwrap(); + assert_eq!(exit, 0, "{stderr}"); + assert_eq!( + stdout.trim(), + std::fs::canonicalize(&target).unwrap().to_string_lossy() + ); + } + + #[test] + fn remote_user_skill_modes_fail_closed_for_disabled_and_future_values() { + assert!(check_remote_user_skill_mode(&RemoteUserSkillsMode::Disabled).is_err()); + assert!( + check_remote_user_skill_mode(&RemoteUserSkillsMode::Unsupported("future".to_string())) + .is_err() + ); + assert!(check_remote_user_skill_mode(&RemoteUserSkillsMode::SkillMdOnly).is_ok()); + assert!(check_remote_user_skill_mode(&RemoteUserSkillsMode::Full).is_ok()); + } + + #[test] + fn source_package_paths_become_posix_without_rewriting_external_paths() { + assert_eq!( + map_package_paths( + "python \"--config=C:\\skills\\demo\\my config\\settings.json\"", + "C:\\skills\\demo", + "/remote/demo", + ), + "python \"--config=/remote/demo/my config/settings.json\"" + ); + assert_eq!( + map_package_paths( + "`python \"--config=C:/skills/demo/my config\\settings.json\"`", + "C:/skills/demo", + "/remote/demo", + ), + "`python \"--config=/remote/demo/my config/settings.json\"`" + ); + assert_eq!( + map_package_paths( + "python /local/skill/my\\ scripts/hello.py", + "/local/skill", + "/remote/skill", + ), + "python /remote/skill/my\\ scripts/hello.py" + ); + assert_eq!( + map_package_paths( + "Read /backup/home/me/skill/references/x and /home/me/skill/references/x", + "/home/me/skill", + "/remote/cache", + ), + "Read /backup/home/me/skill/references/x and /remote/cache/references/x" + ); + assert_eq!( + map_package_paths( + "python \"C:\\skills\\demo\\my scripts\\hello.py\"", + "C:\\skills\\demo", + "/remote/demo", + ), + "python \"/remote/demo/my scripts/hello.py\"" + ); + assert_eq!( + map_package_paths( + "Read /local/skill/references/a.md", + "/local/skill", + "/local/skill/cache", + ), + "Read /local/skill/cache/references/a.md" + ); + assert_eq!(map_package_paths( + "python \"C:\\Users\\Alice\\my skill\\scripts\\hello.py\"; see `C:/Users/Alice/my skill/references/doc.md`", + "C:\\Users\\Alice\\my skill", "/remote/my-skill", + ), "python \"/remote/my-skill/scripts/hello.py\"; see `/remote/my-skill/references/doc.md`"); + assert_eq!( + map_package_paths( + "Read /local/skill/references/a.md and /local/skill-extra/x and /other/file", + "/local/skill", + "/remote/skill", + ), + "Read /remote/skill/references/a.md and /local/skill-extra/x and /other/file" + ); + } +} diff --git a/src/crates/contracts/config-contracts/src/types.rs b/src/crates/contracts/config-contracts/src/types.rs index 025a54b777..624091f342 100644 --- a/src/crates/contracts/config-contracts/src/types.rs +++ b/src/crates/contracts/config-contracts/src/types.rs @@ -812,6 +812,8 @@ pub struct TerminalConfig { #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(default)] pub struct WorkspaceConfig { + /// Availability and resource deployment of host-local user Skills in remote workspaces. + pub remote_user_skills: RemoteUserSkillsMode, pub exclude_patterns: Vec, pub include_patterns: Vec, pub watch_ignore: Vec, @@ -823,6 +825,39 @@ pub struct WorkspaceConfig { pub insert_final_newline: bool, } +/// Keep unknown future values intact on disk; runtime callers must reject them. +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(from = "String", into = "String")] +pub enum RemoteUserSkillsMode { + Disabled, + #[default] + SkillMdOnly, + Full, + Unsupported(String), +} + +impl From for RemoteUserSkillsMode { + fn from(value: String) -> Self { + match value.as_str() { + "disabled" => Self::Disabled, + "skill_md_only" => Self::SkillMdOnly, + "full" => Self::Full, + _ => Self::Unsupported(value), + } + } +} + +impl From for String { + fn from(value: RemoteUserSkillsMode) -> Self { + match value { + RemoteUserSkillsMode::Disabled => "disabled".into(), + RemoteUserSkillsMode::SkillMdOnly => "skill_md_only".into(), + RemoteUserSkillsMode::Full => "full".into(), + RemoteUserSkillsMode::Unsupported(value) => value, + } + } +} + /// Model capability type (a model can have multiple capabilities). #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Hash)] #[serde(rename_all = "snake_case")] @@ -2260,6 +2295,7 @@ impl Default for TerminalConfig { impl Default for WorkspaceConfig { fn default() -> Self { Self { + remote_user_skills: RemoteUserSkillsMode::default(), exclude_patterns: vec![ "**/node_modules/**".to_string(), "**/target/**".to_string(), @@ -2492,6 +2528,29 @@ impl AIModelConfig { #[cfg(test)] mod tests { + #[test] + fn remote_user_skills_preserves_legacy_workspace_settings_and_future_modes() { + use super::{RemoteUserSkillsMode, WorkspaceConfig}; + let legacy = serde_json::json!({ + "exclude_patterns": ["legacy/**"], "encoding": "utf8", "max_file_size": 1234 + }); + let config: WorkspaceConfig = serde_json::from_value(legacy.clone()).unwrap(); + assert_eq!(config.remote_user_skills, RemoteUserSkillsMode::SkillMdOnly); + let saved = serde_json::to_value(config).unwrap(); + for (key, value) in legacy.as_object().unwrap() { + assert_eq!(&saved[key], value); + } + for mode in ["skill_md_only", "disabled", "full", "future_mode"] { + let mut payload = legacy.clone(); + payload["remote_user_skills"] = mode.into(); + let config: WorkspaceConfig = serde_json::from_value(payload).unwrap(); + let saved = serde_json::to_value(&config).unwrap(); + assert_eq!(saved["remote_user_skills"], mode); + let restored: WorkspaceConfig = serde_json::from_value(saved).unwrap(); + assert_eq!(restored.remote_user_skills, config.remote_user_skills); + } + } + #[test] fn model_routing_summary_defaults_to_automatic_and_preserves_old_switch_values() { let legacy: super::ModelRoutingConfig = diff --git a/src/crates/contracts/runtime-ports/src/workspace_ports.rs b/src/crates/contracts/runtime-ports/src/workspace_ports.rs index b4452c55a5..3261c88e00 100644 --- a/src/crates/contracts/runtime-ports/src/workspace_ports.rs +++ b/src/crates/contracts/runtime-ports/src/workspace_ports.rs @@ -254,6 +254,11 @@ pub trait WorkspaceFileSystem: Send + Sync { async fn create_dir_all(&self, _path: &str) -> anyhow::Result<()> { anyhow::bail!("directory creation is not supported by this workspace filesystem") } + /// Create one new directory exclusively. An existing file, directory or + /// symbolic link must fail; parents are not created implicitly. + async fn create_dir_new(&self, _path: &str) -> anyhow::Result<()> { + anyhow::bail!("exclusive directory creation is not supported by this workspace filesystem") + } async fn set_permissions(&self, _path: &str, _permissions: u32) -> anyhow::Result<()> { anyhow::bail!("POSIX permissions are not supported by this workspace filesystem") } @@ -472,6 +477,17 @@ pub(crate) mod tests { } } + #[tokio::test] + async fn exclusive_directory_creation_is_explicitly_unsupported_by_legacy_providers() { + let error = FakeWorkspaceFileSystem + .create_dir_new("/workspace/new") + .await + .unwrap_err(); + assert!(error + .to_string() + .contains("exclusive directory creation is not supported")); + } + #[test] fn workspace_services_contract_is_runtime_port_owned() { let services = fake_workspace_services(); diff --git a/src/crates/execution/agent-runtime/src/skills/mod.rs b/src/crates/execution/agent-runtime/src/skills/mod.rs index 49406af68f..2d64d50ea8 100644 --- a/src/crates/execution/agent-runtime/src/skills/mod.rs +++ b/src/crates/execution/agent-runtime/src/skills/mod.rs @@ -46,6 +46,6 @@ pub use selection::{ }; pub use types::{ render_loaded_skill_for_assistant, ModeSkillInfo, ModeSkillStateReason, SkillData, - SkillImportOrigin, SkillInfo, SkillLocation, SkillParseError, SkillScanDiagnostic, - SkillScanReport, + SkillImportOrigin, SkillInfo, SkillLocation, SkillParseError, SkillResourceAvailability, + SkillScanDiagnostic, SkillScanReport, }; diff --git a/src/crates/execution/agent-runtime/src/skills/types.rs b/src/crates/execution/agent-runtime/src/skills/types.rs index 686ac6675d..0a1712c330 100644 --- a/src/crates/execution/agent-runtime/src/skills/types.rs +++ b/src/crates/execution/agent-runtime/src/skills/types.rs @@ -231,6 +231,8 @@ pub struct SkillData { pub location: SkillLocation, pub path: String, pub entry_file: Option, + /// Resource availability for this load; absent for local/project/legacy loads. + pub resource_availability: Option, pub source_slot: String, pub source_id: String, pub source_label: String, @@ -245,6 +247,13 @@ pub struct SkillData { pub hooks: Option, } +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum SkillResourceAvailability { + MarkdownOnly, + RemotePackage, +} + fn default_allow_implicit_invocation() -> bool { true } @@ -619,6 +628,7 @@ impl SkillData { location, path, entry_file: None, + resource_availability: None, source_slot: String::new(), source_id: String::new(), source_label: String::new(), diff --git a/src/crates/execution/agent-runtime/tests/agent_definition_contracts/skill_contracts.rs b/src/crates/execution/agent-runtime/tests/agent_definition_contracts/skill_contracts.rs index 9d0ed05ef6..2996413e20 100644 --- a/src/crates/execution/agent-runtime/tests/agent_definition_contracts/skill_contracts.rs +++ b/src/crates/execution/agent-runtime/tests/agent_definition_contracts/skill_contracts.rs @@ -835,6 +835,7 @@ Use the pdf workflow. assert_eq!(data.description, "Work with PDF files."); assert_eq!(data.dir_name, "pdf"); assert_eq!(data.content, "Use the pdf workflow.\n"); + assert_eq!(data.resource_availability, None); let assistant = render_loaded_skill_for_assistant(&data, false); assert!(assistant.contains("Skill 'pdf' loaded successfully.")); diff --git a/src/crates/services/services-core/AGENTS.md b/src/crates/services/services-core/AGENTS.md index 9c3f91c5ce..62f543ccbb 100644 --- a/src/crates/services/services-core/AGENTS.md +++ b/src/crates/services/services-core/AGENTS.md @@ -219,6 +219,25 @@ For bounded workspace upload ownership, offset recovery, conflict detection, and cargo test -p openbitfun-services-core --no-default-features --features workspace-transfer,workspace-runtime --lib workspace_transfer::tests ``` +For complete local Skill package deployment, cache integrity, source changes, +executable modes and unsupported targets: + +```bash +cargo test --locked -p openbitfun-services-core --no-default-features --features skill-package-sync,workspace-runtime --lib skill_package_sync::tests +``` + +These tests use local POSIX workspace providers. The real loopback SSH/SFTP +exercise is documented in the services-integrations guide. Deployment uses only +injected workspace filesystem operations with bounded concurrency and streamed +content checks; it never launches a target shell or requires target-side Python, +archive tools or a synchronization executable. Nested symlinks or reparse points +are rejected; a symlinked package root is canonicalized before scanning. +The selected remote workspace may have symlink ancestors; only that trusted +root is resolved. Managed cache components below it must remain plain directories. +Synchronization never moves or removes a published generation. Corrupt or +modified copies are retained, and repairs publish a new verified generation so +concurrent invocations cannot invalidate each other's returned resource paths. + Workspace catalog identity and upgrade deserialization (without live filesystem/SSH activation): ```bash diff --git a/src/crates/services/services-core/Cargo.toml b/src/crates/services/services-core/Cargo.toml index e11052106a..f2d21ef96c 100644 --- a/src/crates/services/services-core/Cargo.toml +++ b/src/crates/services/services-core/Cargo.toml @@ -34,6 +34,7 @@ thiserror = { workspace = true } log = { workspace = true } fs2 = { workspace = true, optional = true } filetime = { workspace = true, optional = true } +futures = { workspace = true, optional = true } rand = { workspace = true, optional = true } ignore = { workspace = true, optional = true } globset = { workspace = true, optional = true } @@ -158,6 +159,18 @@ workspace-transfer = [ "tokio/rt", "tokio/sync", ] +skill-package-sync = [ + "dep:anyhow", + "dep:openbitfun-runtime-ports", + "openbitfun-runtime-ports/workspace-ports", + "dep:sha2", + "dep:futures", + "dep:uuid", + "dep:tokio", + "tokio/fs", + "tokio/io-util", + "tokio/rt", +] workspace-runtime = [ "dep:anyhow", "dep:async-trait", diff --git a/src/crates/services/services-core/src/lib.rs b/src/crates/services/services-core/src/lib.rs index d17bdcd365..9bd7bf691d 100644 --- a/src/crates/services/services-core/src/lib.rs +++ b/src/crates/services/services-core/src/lib.rs @@ -61,6 +61,8 @@ pub mod resource_capacity; pub mod routing_state; #[cfg(feature = "session-event-format")] pub mod session_projection_format; +#[cfg(feature = "skill-package-sync")] +pub mod skill_package_sync; pub mod storage_error; #[cfg(feature = "workspace-persistence")] pub mod workspace_persistence; diff --git a/src/crates/services/services-core/src/skill_package_sync.rs b/src/crates/services/services-core/src/skill_package_sync.rs new file mode 100644 index 0000000000..a693719e02 --- /dev/null +++ b/src/crates/services/services-core/src/skill_package_sync.rs @@ -0,0 +1,766 @@ +//! Verified, staged directory deployment through workspace filesystem ports. +//! +//! Hashing and synchronization run on the host. The target needs only its bound +//! filesystem provider (SFTP for SSH); no interpreter or remote command runs. + +use anyhow::{bail, Context, Result}; +use futures::{stream, FutureExt, StreamExt, TryStreamExt}; +use openbitfun_runtime_ports::{WorkspaceFileSystem, WorkspaceMetadata, WorkspacePathKind}; +use serde::Serialize; +use sha2::{Digest, Sha256}; +use std::{ + collections::{HashMap, HashSet}, + fs::{self, File}, + io::Read, + path::{Path, PathBuf}, + sync::{ + atomic::{AtomicBool, AtomicU64, Ordering}, + Arc, Mutex, + }, + time::{Duration, Instant}, +}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; + +const MANIFEST_NAME: &str = ".openbitfun-package-manifest.json"; +// A bounded number of independent streams hides small-file round-trip latency +// without exhausting a connection's SFTP handle budget. +const MAX_FILE_TRANSFERS: usize = 4; +const STREAM_BUFFER_BYTES: usize = 64 * 1024; + +#[cfg(test)] +#[path = "skill_package_sync_tests.rs"] +mod tests; + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum SkillPackageSyncPhase { + Scanning, + Checking, + Uploading, + Verifying, + Complete, +} + +#[derive(Clone, Debug)] +pub struct SkillPackageSyncProgress { + pub phase: SkillPackageSyncPhase, + pub file_count: usize, + pub total_bytes: u64, + pub transferred_bytes: u64, +} + +pub type SkillPackageProgressCallback = Arc; + +#[derive(Clone, Debug)] +pub struct SkillPackageDeployment { + /// Absolute POSIX path on the target, independent of the controller OS. + pub path: String, + pub content_hash: String, + pub file_count: usize, + pub total_bytes: u64, + pub reused: bool, +} + +#[derive(Serialize)] +struct Manifest { + version: u8, + entries: Vec, +} + +#[derive(Serialize)] +struct ManifestEntry { + path: String, + kind: &'static str, + mode: u32, + size: u64, + sha256: String, +} + +struct LocalPackage { + root: PathBuf, + manifest: Manifest, + manifest_bytes: Vec, + package_id: String, + content_hash: String, + file_count: usize, + total_bytes: u64, + cancelled: Arc, +} + +struct CancelLocalWork(Arc); +impl Drop for CancelLocalWork { + fn drop(&mut self) { + self.0.store(true, Ordering::Relaxed); + } +} + +fn check_cancelled(cancelled: &AtomicBool) -> Result<()> { + if cancelled.load(Ordering::Relaxed) { + bail!("Local Skill package preparation was cancelled"); + } + Ok(()) +} + +fn report( + callback: &Option, + phase: SkillPackageSyncPhase, + file_count: usize, + total_bytes: u64, + transferred_bytes: u64, +) { + if let Some(callback) = callback { + callback(SkillPackageSyncProgress { + phase, + file_count, + total_bytes, + transferred_bytes, + }); + } +} + +/// Deploy a complete local directory through filesystem operations only. Cache +/// hits still verify all bytes, modes and directory entries. Misses stream files +/// concurrently into an exclusive staging directory and publish only after full +/// target verification. The target needs no Python, shell, tar, unzip or rsync. +/// +/// The selected workspace is trusted and may have symlink ancestors; every +/// managed cache component below that boundary must be a plain directory. Old +/// version-1 manifest/digest directories remain readable and reusable. +pub async fn sync_local_skill_package( + local_root: &Path, + remote_cache_root: &str, + remote_workspace_root: &str, + package_key: &str, + fs: &dyn WorkspaceFileSystem, + progress: Option, +) -> Result { + validate_remote_cache_location(remote_cache_root, remote_workspace_root)?; + let cancelled = Arc::new(AtomicBool::new(false)); + let _cancel_local_work = CancelLocalWork(cancelled.clone()); + let progress: Option = progress.map(|callback| { + let last = Mutex::new(None::<(SkillPackageSyncPhase, Instant)>); + Arc::new(move |event: SkillPackageSyncProgress| { + let now = Instant::now(); + let mut last = last.lock().unwrap_or_else(|error| error.into_inner()); + if last.is_some_and(|(phase, time)| { + phase == event.phase && now.duration_since(time) < Duration::from_millis(250) + }) { + return; + } + *last = Some((event.phase, now)); + drop(last); + callback(event); + }) as SkillPackageProgressCallback + }); + report(&progress, SkillPackageSyncPhase::Scanning, 0, 0, 0); + let root = local_root.to_owned(); + let key = package_key.to_owned(); + let callback = progress.clone(); + let package = + tokio::task::spawn_blocking(move || scan_package(&root, &key, &callback, cancelled)) + .await + .context("Local Skill package scan failed")??; + report( + &progress, + SkillPackageSyncPhase::Checking, + package.file_count, + package.total_bytes, + 0, + ); + ensure_cache_root(fs, remote_cache_root, remote_workspace_root).await?; + let parent = remote_join(remote_cache_root, &package.package_id); + ensure_plain_directory(fs, &parent).await?; + if let Some(target) = find_verified_deployment(fs, &parent, &package).await? { + report( + &progress, + SkillPackageSyncPhase::Complete, + package.file_count, + package.total_bytes, + 0, + ); + return Ok(deployment(&package, target, true)); + } + + let staging = remote_join(&parent, &format!(".staging-{}", uuid::Uuid::new_v4())); + fs.create_dir_new(&staging) + .await + .context("Target does not support exclusive Skill staging directories")?; + let result = async { + fs.set_permissions(&staging, 0o700).await?; + upload_package(fs, &staging, &package, &progress).await?; + report( + &progress, + SkillPackageSyncPhase::Verifying, + package.file_count, + package.total_bytes, + package.total_bytes, + ); + if !verify_remote_package(fs, &staging, &package).await? { + bail!("Uploaded Skill package failed full integrity verification"); + } + let (target, reused) = publish_package(fs, &staging, &parent, &package).await?; + Ok::<_, anyhow::Error>(deployment(&package, target, reused)) + } + .await; + // Never remove a published directory on failure. Interrupted transfers may + // leave an unreferenced staging sibling if the target is disconnected. + let _ = fs.remove_dir(&staging, true).await; + let deployed = result?; + report( + &progress, + SkillPackageSyncPhase::Complete, + package.file_count, + package.total_bytes, + package.total_bytes, + ); + Ok(deployed) +} + +fn deployment(package: &LocalPackage, path: String, reused: bool) -> SkillPackageDeployment { + SkillPackageDeployment { + path, + content_hash: package.content_hash.clone(), + file_count: package.file_count, + total_bytes: package.total_bytes, + reused, + } +} + +fn remote_join(root: &str, relative: &str) -> String { + format!("{}/{}", root.trim_end_matches('/'), relative) +} + +fn validate_remote_root(root: &str) -> Result<()> { + if !root.starts_with('/') + || root.contains('\0') + || root.split('/').any(|part| matches!(part, "." | "..")) + { + bail!("Skill package cache requires an absolute POSIX target path without dot segments"); + } + Ok(()) +} + +fn validate_remote_cache_location(cache: &str, workspace: &str) -> Result<()> { + validate_remote_root(cache)?; + validate_remote_root(workspace)?; + let workspace: Vec<_> = workspace + .split('/') + .filter(|part| !part.is_empty()) + .collect(); + let cache: Vec<_> = cache.split('/').filter(|part| !part.is_empty()).collect(); + if cache.len() <= workspace.len() || !cache.starts_with(&workspace) { + bail!("Skill package cache must be inside the selected remote workspace"); + } + Ok(()) +} + +async fn ensure_plain_directory(fs: &dyn WorkspaceFileSystem, path: &str) -> Result<()> { + match fs.metadata(path, false).await? { + Some(metadata) if metadata.kind == WorkspacePathKind::Directory => Ok(()), + Some(_) => bail!("Skill cache path must be a directory without symbolic links: {path}"), + None => { + if let Err(error) = fs.create_dir_new(path).await { + // Another invocation may have created this shared component. + if fs + .metadata(path, false) + .await? + .is_some_and(|m| m.kind == WorkspacePathKind::Directory) + { + return Ok(()); + } + return Err(error).context("Failed to create a plain Skill cache directory"); + } + fs.set_permissions(path, 0o700).await?; + Ok(()) + } + } +} + +async fn ensure_cache_root( + fs: &dyn WorkspaceFileSystem, + cache: &str, + workspace: &str, +) -> Result<()> { + let root = fs + .metadata(workspace, true) + .await? + .ok_or_else(|| anyhow::anyhow!("Selected remote workspace is unavailable"))?; + if root.kind != WorkspacePathKind::Directory { + bail!("Selected remote workspace is not a directory"); + } + let root_parts = workspace.split('/').filter(|part| !part.is_empty()).count(); + let mut path = workspace.trim_end_matches('/').to_string(); + for component in cache + .split('/') + .filter(|part| !part.is_empty()) + .skip(root_parts) + { + path = remote_join(&path, component); + ensure_plain_directory(fs, &path).await?; + } + Ok(()) +} + +fn permissions(metadata: &WorkspaceMetadata) -> Result { + metadata + .permissions + .map(|mode| mode & 0o777) + .ok_or_else(|| { + anyhow::anyhow!("Full user Skill sync requires filesystem POSIX permission metadata") + }) +} + +async fn verify_remote_package( + fs: &dyn WorkspaceFileSystem, + root: &str, + package: &LocalPackage, +) -> Result { + let Some(metadata) = fs.metadata(root, false).await? else { + return Ok(false); + }; + if metadata.kind != WorkspacePathKind::Directory || permissions(&metadata)? != 0o700 { + return Ok(false); + } + let marker = remote_join(root, MANIFEST_NAME); + let Some(metadata) = fs.metadata(&marker, false).await? else { + return Ok(false); + }; + if metadata.kind != WorkspacePathKind::File { + return Ok(false); + } + let mut reader = fs + .open_read(&marker) + .await + .context("Target does not support streamed Skill verification")?; + let mut buffer = vec![0; STREAM_BUFFER_BYTES]; + let mut offset = 0; + loop { + let read = reader.read(&mut buffer).await?; + if read == 0 { + break; + } + if offset + read > package.manifest_bytes.len() + || buffer[..read] != package.manifest_bytes[offset..offset + read] + { + return Ok(false); + } + offset += read; + } + if offset != package.manifest_bytes.len() { + return Ok(false); + } + drop(reader); + + let expected: HashMap<_, _> = package + .manifest + .entries + .iter() + .map(|entry| (entry.path.as_str(), entry)) + .collect(); + let mut seen = HashSet::new(); + let mut directories = vec![String::new()]; + while let Some(relative) = directories.pop() { + let directory = if relative.is_empty() { + root.to_string() + } else { + remote_join(root, &relative) + }; + let Some(metadata) = fs.metadata(&directory, false).await? else { + return Ok(false); + }; + if metadata.kind != WorkspacePathKind::Directory { + return Ok(false); + } + if !relative.is_empty() && permissions(&metadata)? != expected[relative.as_str()].mode { + return Ok(false); + } + for child in fs.read_dir(&directory).await? { + if child.name.is_empty() + || matches!(child.name.as_str(), "." | "..") + || child.name.contains(['/', '\\', '\0']) + { + bail!("Target returned an invalid Skill package entry name"); + } + let name = if relative.is_empty() { + child.name + } else { + format!("{relative}/{}", child.name) + }; + if name == MANIFEST_NAME { + continue; + } + let Some(entry) = expected.get(name.as_str()) else { + return Ok(false); + }; + if child.is_symlink || !seen.insert(name.clone()) { + return Ok(false); + } + if entry.kind == "directory" { + directories.push(name); + } + } + } + if seen.len() != expected.len() { + return Ok(false); + } + let checks: Vec<_> = package + .manifest + .entries + .iter() + .map(|entry| verify_remote_entry(fs, root, entry).boxed()) + .collect(); + let mut checks = stream::iter(checks).buffer_unordered(MAX_FILE_TRANSFERS); + while let Some(valid) = checks.next().await { + if !valid? { + return Ok(false); + } + } + Ok(true) +} + +async fn verify_remote_entry( + fs: &dyn WorkspaceFileSystem, + root: &str, + entry: &ManifestEntry, +) -> Result { + let path = remote_join(root, &entry.path); + let Some(metadata) = fs.metadata(&path, false).await? else { + return Ok(false); + }; + let kind = if entry.kind == "directory" { + WorkspacePathKind::Directory + } else { + WorkspacePathKind::File + }; + if metadata.kind != kind || permissions(&metadata)? != entry.mode { + return Ok(false); + } + if entry.kind == "directory" { + return Ok(true); + } + if metadata.size.is_some_and(|size| size != entry.size) { + return Ok(false); + } + let mut reader = fs + .open_read(&path) + .await + .context("Target does not support streamed Skill verification")?; + let mut buffer = vec![0; STREAM_BUFFER_BYTES]; + let mut digest = Sha256::new(); + let mut size = 0; + loop { + let read = reader.read(&mut buffer).await?; + if read == 0 { + break; + } + size += read as u64; + if size > entry.size { + return Ok(false); + } + digest.update(&buffer[..read]); + } + Ok(size == entry.size && format!("{:x}", digest.finalize()) == entry.sha256) +} + +async fn upload_package( + fs: &dyn WorkspaceFileSystem, + staging: &str, + package: &LocalPackage, + progress: &Option, +) -> Result<()> { + let mut directories: Vec<_> = package + .manifest + .entries + .iter() + .filter(|entry| entry.kind == "directory") + .collect(); + directories.sort_by_key(|entry| entry.path.split('/').count()); + for entry in &directories { + let path = remote_join(staging, &entry.path); + fs.create_dir_new(&path).await?; + fs.set_permissions(&path, 0o700).await?; + } + report( + progress, + SkillPackageSyncPhase::Uploading, + package.file_count, + package.total_bytes, + 0, + ); + let transferred = AtomicU64::new(0); + let uploads: Vec<_> = package + .manifest + .entries + .iter() + .filter(|entry| entry.kind == "file") + .map(|entry| upload_file(fs, staging, package, entry, progress, &transferred).boxed()) + .collect(); + stream::iter(uploads) + .buffer_unordered(MAX_FILE_TRANSFERS) + .try_collect::>() + .await?; + let marker = remote_join(staging, MANIFEST_NAME); + let mut writer = fs + .open_write_new(&marker) + .await + .context("Target does not support streaming Skill package uploads")?; + writer.write_all(&package.manifest_bytes).await?; + writer.shutdown().await?; + drop(writer); + fs.set_permissions(&marker, 0o600).await?; + for entry in directories.into_iter().rev() { + fs.set_permissions(&remote_join(staging, &entry.path), entry.mode) + .await?; + } + Ok(()) +} + +async fn upload_file( + fs: &dyn WorkspaceFileSystem, + staging: &str, + package: &LocalPackage, + entry: &ManifestEntry, + progress: &Option, + transferred: &AtomicU64, +) -> Result<()> { + check_cancelled(&package.cancelled)?; + let path = package.root.join(&entry.path); + let metadata = tokio::fs::symlink_metadata(&path).await?; + if crate::bounded_fs::is_symlink_or_reparse(&metadata) + || !metadata.is_file() + || local_mode(&metadata) != entry.mode + || !tokio::fs::canonicalize(&path) + .await? + .starts_with(&package.root) + { + bail!("Local Skill package changed while preparing its transfer: {}. Try loading the Skill again", entry.path); + } + let mut input = tokio::fs::File::open(&path).await?; + let target = remote_join(staging, &entry.path); + let mut output = fs + .open_write_new(&target) + .await + .context("Target does not support streaming Skill package uploads")?; + let mut buffer = vec![0; STREAM_BUFFER_BYTES]; + let mut digest = Sha256::new(); + let mut size = 0; + loop { + check_cancelled(&package.cancelled)?; + let read = input.read(&mut buffer).await?; + if read == 0 { + break; + } + digest.update(&buffer[..read]); + size += read as u64; + if size > entry.size { + bail!("Local Skill file changed while preparing its transfer: {}. Try loading the Skill again", entry.path); + } + output.write_all(&buffer[..read]).await?; + let transferred = transferred.fetch_add(read as u64, Ordering::Relaxed) + read as u64; + report( + progress, + SkillPackageSyncPhase::Uploading, + package.file_count, + package.total_bytes, + transferred, + ); + } + if size != entry.size || format!("{:x}", digest.finalize()) != entry.sha256 { + bail!("Local Skill file changed while preparing its transfer: {}. Try loading the Skill again", entry.path); + } + output.shutdown().await?; + drop(output); + fs.set_permissions(&target, entry.mode).await?; + Ok(()) +} + +async fn find_verified_deployment( + fs: &dyn WorkspaceFileSystem, + parent: &str, + package: &LocalPackage, +) -> Result> { + let mut candidates: Vec<_> = fs + .read_dir(parent) + .await? + .into_iter() + .filter_map(|entry| { + let is_legacy = entry.name == package.content_hash; + let is_generation = entry + .name + .strip_prefix(&format!("{}-", package.content_hash)) + .is_some_and(|suffix| { + uuid::Uuid::parse_str(suffix) + .is_ok_and(|id| id.hyphenated().to_string() == suffix) + }); + (is_legacy || is_generation).then_some(entry.name) + }) + .collect(); + // Prefer newer-format generations over legacy fixed-digest directories. + // UUID order is not chronology; all verified generations contain the same + // immutable bytes. This merely makes the selection deterministic. + candidates.sort_by(|left, right| right.cmp(left)); + for name in candidates { + let path = remote_join(parent, &name); + if verify_remote_package(fs, &path, package).await? { + return Ok(Some(path)); + } + } + Ok(None) +} + +async fn publish_package( + fs: &dyn WorkspaceFileSystem, + staging: &str, + parent: &str, + package: &LocalPackage, +) -> Result<(String, bool)> { + if let Some(winner) = find_verified_deployment(fs, parent, package).await? { + return Ok((winner, true)); + } + let target = remote_join( + parent, + &format!("{}-{}", package.content_hash, uuid::Uuid::new_v4()), + ); + fs.rename(staging, &target) + .await + .context("Skill package publication failed; existing deployments were preserved")?; + // Each publisher owns its UUID destination. A concurrent repair can never + // move a path another invocation has already returned. Corrupt generations + // remain intact, and readers discover any verified generation next time. + Ok((target, false)) +} + +fn scan_package( + local_root: &Path, + key: &str, + progress: &Option, + cancelled: Arc, +) -> Result { + check_cancelled(&cancelled)?; + let root = fs::canonicalize(local_root) + .with_context(|| format!("Cannot read local Skill directory {}", local_root.display()))?; + if !root.is_dir() { + bail!("Local Skill package is not a directory: {}", root.display()); + } + let package_id = format!( + "{:x}", + Sha256::digest(serde_json::to_vec(&(key, root.to_string_lossy()))?) + ); + let mut package = LocalPackage { + root: root.clone(), + manifest: Manifest { + version: 1, + entries: Vec::new(), + }, + manifest_bytes: Vec::new(), + package_id, + content_hash: String::new(), + file_count: 0, + total_bytes: 0, + cancelled, + }; + let mut manifest = Manifest { + version: 1, + entries: Vec::new(), + }; + scan_directory(&root, "", &mut manifest.entries, &mut package, progress)?; + manifest.entries.sort_by(|a, b| a.path.cmp(&b.path)); + package.manifest_bytes = serde_json::to_vec(&manifest)?; + package.content_hash = format!("{:x}", Sha256::digest(&package.manifest_bytes)); + package.manifest = manifest; + Ok(package) +} + +fn scan_directory( + directory: &Path, + prefix: &str, + manifest: &mut Vec, + package: &mut LocalPackage, + progress: &Option, +) -> Result<()> { + let mut children = fs::read_dir(directory)?.collect::>>()?; + children.sort_by_key(|entry| entry.file_name()); + for child in children { + check_cancelled(&package.cancelled)?; + let name = child + .file_name() + .into_string() + .map_err(|_| anyhow::anyhow!("Full user Skill sync requires UTF-8 file names"))?; + if name.contains('\\') { + bail!("Full user Skill sync cannot map a file name containing a backslash: {name}"); + } + let relative = if prefix.is_empty() { + name + } else { + format!("{prefix}/{name}") + }; + if relative == MANIFEST_NAME { + bail!("Skill package contains the reserved file {MANIFEST_NAME}"); + } + let path = child.path(); + let metadata = fs::symlink_metadata(&path)?; + if crate::bounded_fs::is_symlink_or_reparse(&metadata) { + bail!( + "Full user Skill sync does not support nested symbolic links: {relative}. Replace the link with package-local files" + ); + } + let mode = local_mode(&metadata); + let mut entry = ManifestEntry { + path: relative.clone(), + kind: "directory", + mode, + size: 0, + sha256: String::new(), + }; + if metadata.is_dir() { + scan_directory(&path, &relative, manifest, package, progress)?; + } else if metadata.is_file() { + entry.kind = "file"; + let mut input = File::open(&path)?; + let mut digest = Sha256::new(); + let mut buffer = [0; 64 * 1024]; + loop { + check_cancelled(&package.cancelled)?; + let read = input.read(&mut buffer)?; + if read == 0 { + break; + } + digest.update(&buffer[..read]); + entry.size += read as u64; + } + if input.metadata()?.modified().ok() != metadata.modified().ok() + || entry.size != metadata.len() + { + bail!("Local Skill file changed while preparing its package: {relative}. Try loading the Skill again"); + } + entry.sha256 = format!("{:x}", digest.finalize()); + package.file_count += 1; + package.total_bytes += entry.size; + report( + progress, + SkillPackageSyncPhase::Scanning, + package.file_count, + package.total_bytes, + 0, + ); + } else { + bail!("Full user Skill sync only supports regular files and directories: {relative}"); + } + manifest.push(entry); + } + Ok(()) +} + +#[cfg(unix)] +fn local_mode(metadata: &fs::Metadata) -> u32 { + use std::os::unix::fs::PermissionsExt; + metadata.permissions().mode() & 0o777 +} + +#[cfg(not(unix))] +fn local_mode(metadata: &fs::Metadata) -> u32 { + if metadata.is_dir() { + 0o755 + } else { + 0o644 + } +} diff --git a/src/crates/services/services-core/src/skill_package_sync_tests.rs b/src/crates/services/services-core/src/skill_package_sync_tests.rs new file mode 100644 index 0000000000..e613f25fd6 --- /dev/null +++ b/src/crates/services/services-core/src/skill_package_sync_tests.rs @@ -0,0 +1,748 @@ +use super::*; + +#[test] +fn remote_cache_paths_use_posix_syntax_and_stay_inside_selected_workspace() { + assert!( + validate_remote_cache_location("/home/user/project/cache", "/home/user/project").is_ok() + ); + assert!(validate_remote_cache_location("/home/user/a\\b/cache", "/home/user/a\\b").is_ok()); + for path in [ + "C:\\Users\\me\\cache", + "relative/cache", + "/tmp/../escape", + "/tmp/./cache", + ] { + assert!(validate_remote_root(path).is_err(), "{path}"); + } + assert!(validate_remote_cache_location("/workspace-other/cache", "/workspace").is_err()); + assert!(validate_remote_cache_location("/workspace", "/workspace").is_err()); +} + +#[cfg(all(unix, feature = "workspace-runtime"))] +mod posix { + use super::*; + use crate::workspace::{LocalWorkspaceFs, LocalWorkspaceShell}; + use openbitfun_runtime_ports::{ + WorkspaceDirEntry, WorkspaceReader, WorkspaceShell, WorkspaceWriter, + }; + use std::{ + os::unix::fs::{symlink, PermissionsExt}, + pin::Pin, + sync::atomic::AtomicUsize, + task::{Context as TaskContext, Poll}, + }; + use tokio::io::AsyncWrite; + + fn write(root: &Path, name: &str, bytes: &[u8]) { + let path = root.join(name); + fs::create_dir_all(path.parent().unwrap()).unwrap(); + fs::write(path, bytes).unwrap(); + } + + fn mode(path: &Path, value: u32) { + fs::set_permissions(path, fs::Permissions::from_mode(value)).unwrap(); + } + + #[derive(Default)] + struct TransferCounts { + active: AtomicUsize, + peak: AtomicUsize, + writes: AtomicUsize, + } + + #[derive(Default)] + struct PublicationGate { + pause: AtomicBool, + started: tokio::sync::Notify, + release: tokio::sync::Notify, + } + + #[derive(Default)] + struct UploadBlock { + blocked: AtomicBool, + started: tokio::sync::Notify, + } + + struct CountedWriter { + inner: WorkspaceWriter, + counts: Arc, + upload_block: Option>, + } + impl Drop for CountedWriter { + fn drop(&mut self) { + self.counts.active.fetch_sub(1, Ordering::SeqCst); + } + } + impl AsyncWrite for CountedWriter { + fn poll_write( + mut self: Pin<&mut Self>, + cx: &mut TaskContext<'_>, + data: &[u8], + ) -> Poll> { + if let Some(block) = &self.upload_block { + if block.blocked.load(Ordering::SeqCst) { + block.started.notify_one(); + return Poll::Pending; + } + } + Pin::new(&mut self.inner).poll_write(cx, data) + } + fn poll_flush( + mut self: Pin<&mut Self>, + cx: &mut TaskContext<'_>, + ) -> Poll> { + Pin::new(&mut self.inner).poll_flush(cx) + } + fn poll_shutdown( + mut self: Pin<&mut Self>, + cx: &mut TaskContext<'_>, + ) -> Poll> { + Pin::new(&mut self.inner).poll_shutdown(cx) + } + } + + #[derive(Default)] + struct InstrumentedFs { + counts: Arc, + fail_upload: AtomicBool, + fail_publish: AtomicBool, + mutate_source: Mutex>, + publication_gate: Mutex>>, + upload_block: Mutex>>, + fail_metadata: AtomicBool, + bogus_listing_paths: AtomicBool, + } + #[async_trait::async_trait] + impl WorkspaceFileSystem for InstrumentedFs { + async fn open_read(&self, path: &str) -> Result { + LocalWorkspaceFs.open_read(path).await + } + async fn open_write_new(&self, path: &str) -> Result { + if self.fail_upload.load(Ordering::SeqCst) { + bail!("fixture rejects streamed upload"); + } + if let Some(source) = self.mutate_source.lock().unwrap().take() { + fs::write(source, b"after!")?; + } + let inner = LocalWorkspaceFs.open_write_new(path).await?; + self.counts.writes.fetch_add(1, Ordering::SeqCst); + let active = self.counts.active.fetch_add(1, Ordering::SeqCst) + 1; + self.counts.peak.fetch_max(active, Ordering::SeqCst); + Ok(Box::new(CountedWriter { + inner, + counts: self.counts.clone(), + upload_block: self.upload_block.lock().unwrap().clone(), + })) + } + async fn metadata(&self, path: &str, follow: bool) -> Result> { + if self.fail_metadata.load(Ordering::SeqCst) { + bail!("fixture transport unavailable"); + } + LocalWorkspaceFs.metadata(path, follow).await + } + async fn create_dir_new(&self, path: &str) -> Result<()> { + LocalWorkspaceFs.create_dir_new(path).await + } + async fn set_permissions(&self, path: &str, permissions: u32) -> Result<()> { + LocalWorkspaceFs.set_permissions(path, permissions).await + } + async fn rename(&self, from: &str, to: &str) -> Result<()> { + if self.fail_publish.load(Ordering::SeqCst) + && from.rsplit('/').next().unwrap().starts_with(".staging-") + { + bail!("fixture rejects publication"); + } + let gate = self.publication_gate.lock().unwrap().clone(); + if let Some(gate) = gate { + if gate.pause.swap(false, Ordering::SeqCst) { + gate.started.notify_one(); + gate.release.notified().await; + } + } + LocalWorkspaceFs.rename(from, to).await + } + async fn remove_dir(&self, path: &str, recursive: bool) -> Result<()> { + LocalWorkspaceFs.remove_dir(path, recursive).await + } + async fn read_file(&self, path: &str) -> Result> { + LocalWorkspaceFs.read_file(path).await + } + async fn read_file_text(&self, path: &str) -> Result { + LocalWorkspaceFs.read_file_text(path).await + } + async fn write_file(&self, path: &str, content: &[u8]) -> Result<()> { + LocalWorkspaceFs.write_file(path, content).await + } + async fn exists(&self, path: &str) -> Result { + LocalWorkspaceFs.exists(path).await + } + async fn is_file(&self, path: &str) -> Result { + LocalWorkspaceFs.is_file(path).await + } + async fn is_dir(&self, path: &str) -> Result { + LocalWorkspaceFs.is_dir(path).await + } + async fn read_dir(&self, path: &str) -> Result> { + LocalWorkspaceFs.read_dir(path).await + } + } + + #[tokio::test] + async fn streams_complete_package_with_bounded_parallelism_and_preserves_executables() { + let temp = tempfile::tempdir().unwrap(); + let local = temp.path().join("local user's skill"); + let remote = temp.path().join("remote workspace"); + fs::create_dir(&remote).unwrap(); + write( + &local, + "SKILL.md", + b"Run scripts/hello.sh and read references/guide.md", + ); + write( + &local, + "references/guide 'one' \u{4e2d}.md", + b"full reference\n", + ); + write(&local, "templates/example.bin", &[0, 255, 1, 2]); + write( + &local, + "scripts/hello.sh", + b"#!/bin/sh\nprintf remote-script-ok", + ); + mode(&local.join("scripts/hello.sh"), 0o751); + fs::create_dir(local.join("empty-directory")).unwrap(); + for index in 0..200 { + write(&local, &format!("tiny/{index}.txt"), b"small"); + } + let target_fs = InstrumentedFs::default(); + target_fs.bogus_listing_paths.store(true, Ordering::SeqCst); + let phases = Arc::new(Mutex::new(Vec::new())); + let observed = phases.clone(); + let cache = remote.join("cache"); + let result = sync_local_skill_package( + &local, + cache.to_str().unwrap(), + remote.to_str().unwrap(), + "user::fixture", + &target_fs, + Some(Arc::new(move |event| { + observed.lock().unwrap().push(event.phase); + })), + ) + .await + .unwrap(); + assert!(!result.reused); + assert_eq!(result.file_count, 204); + assert!( + target_fs.counts.peak.load(Ordering::SeqCst) > 1, + "small files should overlap transfers" + ); + assert!(target_fs.counts.peak.load(Ordering::SeqCst) <= MAX_FILE_TRANSFERS); + assert_eq!(target_fs.counts.active.load(Ordering::SeqCst), 0); + assert_eq!( + fs::read(Path::new(&result.path).join("references/guide 'one' \u{4e2d}.md")).unwrap(), + b"full reference\n" + ); + assert_eq!( + fs::read(Path::new(&result.path).join("templates/example.bin")).unwrap(), + [0, 255, 1, 2] + ); + assert!(Path::new(&result.path).join("empty-directory").is_dir()); + let shell = LocalWorkspaceShell::new(remote.to_string_lossy().into_owned()); + let (stdout, stderr, exit) = shell + .exec(&format!("'{}/scripts/hello.sh'", result.path), None) + .await + .unwrap(); + assert_eq!(exit, 0, "{stderr}"); + assert_eq!(stdout, "remote-script-ok"); + assert_eq!( + fs::metadata(Path::new(&result.path).join("scripts/hello.sh")) + .unwrap() + .permissions() + .mode() + & 0o777, + 0o751 + ); + for phase in [ + SkillPackageSyncPhase::Scanning, + SkillPackageSyncPhase::Checking, + SkillPackageSyncPhase::Uploading, + SkillPackageSyncPhase::Verifying, + SkillPackageSyncPhase::Complete, + ] { + assert!(phases.lock().unwrap().contains(&phase)); + } + let before = target_fs.counts.writes.load(Ordering::SeqCst); + let reused = sync_local_skill_package( + &local, + cache.to_str().unwrap(), + remote.to_str().unwrap(), + "user::fixture", + &target_fs, + None, + ) + .await + .unwrap(); + assert!(reused.reused); + assert_eq!( + target_fs.counts.writes.load(Ordering::SeqCst), + before, + "verified cache reuse does not upload files" + ); + } + + #[tokio::test] + async fn cache_integrity_detects_bytes_modes_missing_extra_files_and_source_changes() { + let temp = tempfile::tempdir().unwrap(); + let local = temp.path().join("local"); + write(&local, "SKILL.md", b"skill body"); + write(&local, "references/guide.md", b"original"); + write(&local, "templates/deleted.txt", b"delete locally later"); + let cache = temp.path().join("cache"); + let deploy = || { + sync_local_skill_package( + &local, + cache.to_str().unwrap(), + temp.path().to_str().unwrap(), + "user::fixture", + &LocalWorkspaceFs, + None, + ) + }; + let first = deploy().await.unwrap(); + assert!(!first.reused); + assert!(deploy().await.unwrap().reused); + let mut current = first.clone(); + for mutation in 0..4 { + let guide = Path::new(¤t.path).join("references/guide.md"); + match mutation { + 0 => fs::write(&guide, b"tampered").unwrap(), + 1 => fs::remove_file(&guide).unwrap(), + 2 => fs::write(Path::new(¤t.path).join("extra.txt"), b"extra").unwrap(), + _ => mode(&guide, 0o700), + } + let corrupted_path = current.path.clone(); + current = deploy().await.unwrap(); + assert!(!current.reused); + assert_ne!( + current.path, corrupted_path, + "repair publishes a new immutable generation" + ); + assert!( + Path::new(&corrupted_path).is_dir(), + "target-side data is preserved" + ); + assert_eq!( + fs::read(Path::new(¤t.path).join("references/guide.md")).unwrap(), + b"original" + ); + assert!(!Path::new(¤t.path).join("extra.txt").exists()); + let again = deploy().await.unwrap(); + assert!(again.reused); + assert_eq!(again.path, current.path); + } + write(&local, "references/guide.md", b"updated locally"); + fs::remove_file(local.join("templates/deleted.txt")).unwrap(); + let updated = deploy().await.unwrap(); + assert_ne!(updated.content_hash, first.content_hash); + assert_eq!( + fs::read(Path::new(&updated.path).join("references/guide.md")).unwrap(), + b"updated locally" + ); + assert!(!Path::new(&updated.path) + .join("templates/deleted.txt") + .exists()); + assert!( + Path::new(¤t.path) + .join("templates/deleted.txt") + .is_file(), + "old source versions remain available" + ); + } + + #[tokio::test] + async fn reuses_legacy_version_one_manifest_and_fixed_digest_directory() { + let temp = tempfile::tempdir().unwrap(); + let local = temp.path().join("local"); + write(&local, "SKILL.md", b"legacy body"); + mode(&local.join("SKILL.md"), 0o640); + let package = scan_package( + &local, + "user::fixture", + &None, + Arc::new(AtomicBool::new(false)), + ) + .unwrap(); + let legacy = format!("{{\"version\":1,\"entries\":[{{\"path\":\"SKILL.md\",\"kind\":\"file\",\"mode\":416,\"size\":11,\"sha256\":\"{:x}\"}}]}}", Sha256::digest(b"legacy body")); + assert_eq!(package.manifest_bytes, legacy.as_bytes()); + let cache = temp.path().join("cache"); + let target = cache.join(&package.package_id).join(&package.content_hash); + fs::create_dir_all(&target).unwrap(); + mode(&target, 0o700); + write(&target, "SKILL.md", b"legacy body"); + mode(&target.join("SKILL.md"), 0o640); + write(&target, MANIFEST_NAME, legacy.as_bytes()); + mode(&target.join(MANIFEST_NAME), 0o644); + let fs = InstrumentedFs::default(); + let loaded = sync_local_skill_package( + &local, + cache.to_str().unwrap(), + temp.path().to_str().unwrap(), + "user::fixture", + &fs, + None, + ) + .await + .unwrap(); + assert!(loaded.reused); + assert_eq!(loaded.path, target.to_str().unwrap()); + assert_eq!(fs.counts.writes.load(Ordering::SeqCst), 0); + } + + #[tokio::test] + async fn supports_trusted_symlink_roots_and_rejects_managed_cache_escape() { + let temp = tempfile::tempdir().unwrap(); + let local = temp.path().join("local"); + write(&local, "SKILL.md", b"skill body"); + let source_link = temp.path().join("source-link"); + symlink(&local, &source_link).unwrap(); + let real = temp.path().join("real"); + fs::create_dir_all(real.join("workspace")).unwrap(); + let alias = temp.path().join("alias"); + symlink(&real, &alias).unwrap(); + let workspace = alias.join("workspace"); + let cache = workspace.join(".openbitfun/cache/user-skills"); + let first = sync_local_skill_package( + &source_link, + cache.to_str().unwrap(), + workspace.to_str().unwrap(), + "user::fixture", + &LocalWorkspaceFs, + None, + ) + .await + .unwrap(); + assert!(first.path.starts_with(cache.to_str().unwrap())); + assert!( + sync_local_skill_package( + &local, + cache.to_str().unwrap(), + workspace.to_str().unwrap(), + "user::fixture", + &LocalWorkspaceFs, + None + ) + .await + .unwrap() + .reused + ); + symlink(temp.path(), local.join("escaping-link")).unwrap(); + let error = sync_local_skill_package( + &local, + cache.to_str().unwrap(), + workspace.to_str().unwrap(), + "user::fixture", + &LocalWorkspaceFs, + None, + ) + .await + .unwrap_err(); + assert!(error.to_string().contains("nested symbolic links")); + fs::remove_file(local.join("escaping-link")).unwrap(); + fs::remove_dir_all(workspace.join(".openbitfun")).unwrap(); + fs::create_dir(workspace.join(".openbitfun")).unwrap(); + let outside = temp.path().join("outside"); + fs::create_dir(&outside).unwrap(); + write(&outside, "sentinel", b"untouched"); + symlink(&outside, workspace.join(".openbitfun/cache")).unwrap(); + let error = sync_local_skill_package( + &local, + cache.to_str().unwrap(), + workspace.to_str().unwrap(), + "user::fixture", + &LocalWorkspaceFs, + None, + ) + .await + .unwrap_err(); + assert!(error.to_string().contains("without symbolic links")); + assert_eq!(fs::read(outside.join("sentinel")).unwrap(), b"untouched"); + assert!(!outside.join("user-skills").exists()); + } + + #[tokio::test] + async fn source_mutation_upload_failure_and_publication_failure_never_report_success() { + let temp = tempfile::tempdir().unwrap(); + let local = temp.path().join("local"); + write(&local, "SKILL.md", b"before"); + let cache = temp.path().join("cache"); + let fault = InstrumentedFs::default(); + fault + .mutate_source + .lock() + .unwrap() + .replace(local.join("SKILL.md")); + let error = sync_local_skill_package( + &local, + cache.to_str().unwrap(), + temp.path().to_str().unwrap(), + "user::fixture", + &fault, + None, + ) + .await + .unwrap_err(); + assert!( + error.to_string().contains("changed while preparing"), + "{error:#}" + ); + write(&local, "SKILL.md", b"before"); + fault.fail_upload.store(true, Ordering::SeqCst); + let error = sync_local_skill_package( + &local, + cache.to_str().unwrap(), + temp.path().to_str().unwrap(), + "user::fixture", + &fault, + None, + ) + .await + .unwrap_err(); + assert!(error + .to_string() + .contains("streaming Skill package uploads")); + fault.fail_upload.store(false, Ordering::SeqCst); + let first = sync_local_skill_package( + &local, + cache.to_str().unwrap(), + temp.path().to_str().unwrap(), + "user::fixture", + &fault, + None, + ) + .await + .unwrap(); + fs::write(Path::new(&first.path).join("SKILL.md"), b"target change").unwrap(); + fault.fail_publish.store(true, Ordering::SeqCst); + let error = sync_local_skill_package( + &local, + cache.to_str().unwrap(), + temp.path().to_str().unwrap(), + "user::fixture", + &fault, + None, + ) + .await + .unwrap_err(); + assert!( + error.to_string().contains("publication failed"), + "{error:#}" + ); + assert_eq!( + fs::read(Path::new(&first.path).join("SKILL.md")).unwrap(), + b"target change", + "failed repair preserves target-side data" + ); + } + + #[tokio::test] + async fn cancellation_during_streaming_preserves_prior_version_and_allows_retry() { + let temp = tempfile::tempdir().unwrap(); + let local = temp.path().join("local"); + write(&local, "SKILL.md", b"original body"); + let cache = temp.path().join("cache"); + let original = sync_local_skill_package( + &local, + cache.to_str().unwrap(), + temp.path().to_str().unwrap(), + "user::fixture", + &LocalWorkspaceFs, + None, + ) + .await + .unwrap(); + write(&local, "SKILL.md", b"new body"); + let blocked = Arc::new(UploadBlock::default()); + blocked.blocked.store(true, Ordering::SeqCst); + let target_fs = Arc::new(InstrumentedFs::default()); + target_fs + .upload_block + .lock() + .unwrap() + .replace(blocked.clone()); + let source = local.clone(); + let cache_string = cache.to_string_lossy().into_owned(); + let workspace = temp.path().to_string_lossy().into_owned(); + let fs_clone = target_fs.clone(); + let task = tokio::spawn(async move { + sync_local_skill_package( + &source, + &cache_string, + &workspace, + "user::fixture", + fs_clone.as_ref(), + None, + ) + .await + }); + tokio::time::timeout(Duration::from_secs(5), blocked.started.notified()) + .await + .unwrap(); + task.abort(); + assert!(task.await.unwrap_err().is_cancelled()); + assert_eq!( + target_fs.counts.active.load(Ordering::SeqCst), + 0, + "cancel releases every opened writer" + ); + assert_eq!( + fs::read(Path::new(&original.path).join("SKILL.md")).unwrap(), + b"original body" + ); + target_fs.upload_block.lock().unwrap().take(); + let retried = sync_local_skill_package( + &local, + cache.to_str().unwrap(), + temp.path().to_str().unwrap(), + "user::fixture", + target_fs.as_ref(), + None, + ) + .await + .unwrap(); + assert_eq!( + fs::read(Path::new(&retried.path).join("SKILL.md")).unwrap(), + b"new body" + ); + target_fs.fail_metadata.store(true, Ordering::SeqCst); + let error = sync_local_skill_package( + &local, + cache.to_str().unwrap(), + temp.path().to_str().unwrap(), + "user::fixture", + target_fs.as_ref(), + None, + ) + .await + .unwrap_err(); + assert!( + error.to_string().contains("transport unavailable"), + "offline metadata cannot fabricate a cache hit" + ); + } + + #[tokio::test] + async fn paused_repair_cannot_move_a_generation_returned_by_another_publisher() { + let temp = tempfile::tempdir().unwrap(); + let local = temp.path().join("local"); + write(&local, "SKILL.md", b"expected body"); + let cache = temp.path().join("cache"); + let original = sync_local_skill_package( + &local, + cache.to_str().unwrap(), + temp.path().to_str().unwrap(), + "user::fixture", + &LocalWorkspaceFs, + None, + ) + .await + .unwrap(); + fs::write(Path::new(&original.path).join("SKILL.md"), b"target edits").unwrap(); + let gate = Arc::new(PublicationGate::default()); + gate.pause.store(true, Ordering::SeqCst); + let paused_fs = Arc::new(InstrumentedFs::default()); + paused_fs + .publication_gate + .lock() + .unwrap() + .replace(gate.clone()); + let source = local.clone(); + let cache_string = cache.to_string_lossy().into_owned(); + let workspace = temp.path().to_string_lossy().into_owned(); + let fs_clone = paused_fs.clone(); + let first = tokio::spawn(async move { + sync_local_skill_package( + &source, + &cache_string, + &workspace, + "user::fixture", + fs_clone.as_ref(), + None, + ) + .await + }); + tokio::time::timeout(Duration::from_secs(5), gate.started.notified()) + .await + .unwrap(); + let winner = sync_local_skill_package( + &local, + cache.to_str().unwrap(), + temp.path().to_str().unwrap(), + "user::fixture", + &LocalWorkspaceFs, + None, + ) + .await + .unwrap(); + assert_eq!( + fs::read(Path::new(&winner.path).join("SKILL.md")).unwrap(), + b"expected body" + ); + gate.release.notify_one(); + let other = first.await.unwrap().unwrap(); + assert_ne!(other.path, winner.path); + assert_eq!( + fs::read(Path::new(&winner.path).join("SKILL.md")).unwrap(), + b"expected body", + "a previously returned resource root stays reachable" + ); + assert_eq!( + fs::read(Path::new(&original.path).join("SKILL.md")).unwrap(), + b"target edits", + "corrupt target-side state is preserved" + ); + assert!( + sync_local_skill_package( + &local, + cache.to_str().unwrap(), + temp.path().to_str().unwrap(), + "user::fixture", + &LocalWorkspaceFs, + None + ) + .await + .unwrap() + .reused + ); + } + + #[tokio::test] + async fn concurrent_publishers_leave_verified_deployments_and_remain_reusable() { + let temp = tempfile::tempdir().unwrap(); + let local = temp.path().join("local"); + write(&local, "SKILL.md", b"concurrent body"); + for index in 0..20 { + write(&local, &format!("references/{index}.txt"), b"content"); + } + let cache = temp.path().join("cache"); + let deploy = || { + sync_local_skill_package( + &local, + cache.to_str().unwrap(), + temp.path().to_str().unwrap(), + "user::fixture", + &LocalWorkspaceFs, + None, + ) + }; + let (left, right) = tokio::join!(deploy(), deploy()); + assert!(left.is_ok() || right.is_ok(), "{left:?}; {right:?}"); + for result in [left, right].into_iter().flatten() { + assert_eq!( + fs::read(Path::new(&result.path).join("SKILL.md")).unwrap(), + b"concurrent body" + ); + } + assert!(deploy().await.unwrap().reused); + } +} diff --git a/src/crates/services/services-core/src/workspace.rs b/src/crates/services/services-core/src/workspace.rs index 61a04d9101..b613720fcc 100644 --- a/src/crates/services/services-core/src/workspace.rs +++ b/src/crates/services/services-core/src/workspace.rs @@ -176,6 +176,10 @@ impl WorkspaceFileSystem for LocalWorkspaceFs { Ok(tokio::fs::create_dir_all(path).await?) } + async fn create_dir_new(&self, path: &str) -> anyhow::Result<()> { + Ok(tokio::fs::create_dir(path).await?) + } + async fn set_permissions(&self, path: &str, permissions: u32) -> anyhow::Result<()> { #[cfg(unix)] { @@ -375,6 +379,38 @@ mod tests { use super::*; use openbitfun_runtime_ports::WorkspaceFileSystem; + #[tokio::test] + async fn local_workspace_exclusive_directory_creation_preserves_existing_paths() { + let temp = tempfile::tempdir().unwrap(); + let directory = temp.path().join("new directory"); + let fs = LocalWorkspaceFs; + fs.create_dir_new(directory.to_str().unwrap()) + .await + .unwrap(); + std::fs::write(directory.join("keep.txt"), b"keep").unwrap(); + assert!(fs + .create_dir_new(directory.to_str().unwrap()) + .await + .is_err()); + assert_eq!(std::fs::read(directory.join("keep.txt")).unwrap(), b"keep"); + let existing_file = temp.path().join("file"); + std::fs::write(&existing_file, b"file").unwrap(); + assert!(fs + .create_dir_new(existing_file.to_str().unwrap()) + .await + .is_err()); + assert!(fs + .create_dir_new(temp.path().join("missing/child").to_str().unwrap()) + .await + .is_err()); + #[cfg(unix)] + { + let link = temp.path().join("directory-link"); + std::os::unix::fs::symlink(&directory, &link).unwrap(); + assert!(fs.create_dir_new(link.to_str().unwrap()).await.is_err()); + } + } + #[cfg(windows)] #[tokio::test] async fn local_workspace_unlinks_directory_junction_without_removing_target() { diff --git a/src/crates/services/services-integrations/AGENTS.md b/src/crates/services/services-integrations/AGENTS.md index ee2fe4fb37..7b9c2abe81 100644 --- a/src/crates/services/services-integrations/AGENTS.md +++ b/src/crates/services/services-integrations/AGENTS.md @@ -179,6 +179,15 @@ advertise a small handle limit and are included in the existing CI the same command with the filter ending in `workspace_sftp::openssh_real_files_over_loopback_ssh -- --ignored`. +For full user Skill deployment, resource reads, executable scripts, content reuse +and stale-copy repair over the same real loopback SSH / OpenSSH SFTP fixture, +run that command with the filter ending in +`workspace_sftp::openssh_skill_package_sync_reads_executes_and_repairs_remote_resources -- --ignored`. +The fixture asserts that synchronization, cache validation and repair perform +zero remote command executions. Only the selected Skill's POSIX shell script is +executed through SSH; neither synchronization nor that script requires Python. +It does not establish external-host, ProxyJump, Docker or Windows-client coverage. + For directory symlink browsing and workspace-path validation, use the same package/features with `--lib directory_symlink_browsing -- --include-ignored` and `OPENBITFUN_TEST_SFTP_SERVER` set. This runs real OpenSSH SFTP over loopback diff --git a/src/crates/services/services-integrations/Cargo.toml b/src/crates/services/services-integrations/Cargo.toml index b3e73a73cf..9230783df5 100644 --- a/src/crates/services/services-integrations/Cargo.toml +++ b/src/crates/services/services-integrations/Cargo.toml @@ -514,6 +514,7 @@ ssh_config = ["dep:ssh_config"] [dev-dependencies] tempfile = { workspace = true } +openbitfun-services-core = { path = "../services-core", features = ["skill-package-sync"] } tokio = { workspace = true, features = ["io-util", "macros", "net", "rt", "test-util"] } # Used by contract tests that implement the `EventEmitter` port to observe # emitted product events. diff --git a/src/crates/services/services-integrations/src/remote_ssh/manager.rs b/src/crates/services/services-integrations/src/remote_ssh/manager.rs index 898f5a0761..90abaea8db 100644 --- a/src/crates/services/services-integrations/src/remote_ssh/manager.rs +++ b/src/crates/services/services-integrations/src/remote_ssh/manager.rs @@ -5261,9 +5261,18 @@ impl SSHConnectionManager { .await; } let sftp = self.get_sftp(connection_id).await?; + // russh-sftp's default attributes contain size/owner/timestamps. A + // chmod request must leave those fields absent rather than truncate or + // chown the path while updating its permissions. let attrs = russh_sftp::protocol::FileAttributes { + size: None, + uid: None, + user: None, + gid: None, + group: None, permissions: Some(permissions), - ..Default::default() + atime: None, + mtime: None, }; sftp.set_metadata(&path, attrs) .await @@ -5300,9 +5309,14 @@ impl SSHConnectionManager { // SFTP v3 sends access and modification time together. Preserve access // time while avoiding unrelated size, owner or permission updates. let attrs = russh_sftp::protocol::FileAttributes { + size: None, + uid: None, + user: None, + gid: None, + group: None, + permissions: None, atime: Some(atime), mtime: Some(seconds), - ..Default::default() }; sftp.set_metadata(&path, attrs) .await diff --git a/src/crates/services/services-integrations/src/remote_ssh/manager/tests/workspace_sftp.rs b/src/crates/services/services-integrations/src/remote_ssh/manager/tests/workspace_sftp.rs index 90c0ff6570..006b8373c6 100644 --- a/src/crates/services/services-integrations/src/remote_ssh/manager/tests/workspace_sftp.rs +++ b/src/crates/services/services-integrations/src/remote_ssh/manager/tests/workspace_sftp.rs @@ -15,6 +15,7 @@ struct State { closed: AtomicUsize, live: AtomicUsize, channels: AtomicUsize, + exec_calls: AtomicUsize, pause_open: AtomicBool, pause_read: AtomicBool, pause_dir: AtomicBool, @@ -227,6 +228,32 @@ impl russh::server::Handler for SshServer { self.channels.insert(channel.id(), channel); Ok(true) } + #[cfg(unix)] + async fn exec_request( + &mut self, + channel: russh::ChannelId, + command: &[u8], + session: &mut russh::server::Session, + ) -> Result<(), Self::Error> { + // This loopback fixture runs commands only against test-owned paths. + // Keep the SSH execution path real while using the local POSIX shell. + self.state.exec_calls.fetch_add(1, Ordering::SeqCst); + session.channel_success(channel); + let output = process_manager::create_tokio_command("sh") + .arg("-c") + .arg(std::str::from_utf8(command).unwrap()) + .stdin(Stdio::null()) + .kill_on_drop(true) + .output() + .await + .unwrap(); + session.data(channel, russh::CryptoVec::from_slice(&output.stdout)); + session.extended_data(channel, 1, russh::CryptoVec::from_slice(&output.stderr)); + session.exit_status_request(channel, output.status.code().unwrap_or(1) as u32); + session.eof(channel); + session.close(channel); + Ok(()) + } async fn subsystem_request( &mut self, id: russh::ChannelId, @@ -1028,3 +1055,247 @@ async fn openssh_streaming_upload_create_is_exclusive_and_closes_handle() { assert_eq!(data.len(), chunk.len() * 32); assert!(data.iter().all(|byte| *byte == 29)); } + +#[cfg(unix)] +#[tokio::test] +#[ignore = "Set OPENBITFUN_TEST_SFTP_SERVER to the OpenSSH sftp-server executable"] +async fn openssh_skill_package_sync_reads_executes_and_repairs_remote_resources() { + use crate::remote_ssh::workspace_services::{RemoteWorkspaceFs, RemoteWorkspaceShell}; + use openbitfun_runtime_ports::{WorkspaceFileSystem, WorkspaceShell}; + use openbitfun_services_core::skill_package_sync::sync_local_skill_package; + use std::os::unix::fs::PermissionsExt; + + let executable = std::env::var("OPENBITFUN_TEST_SFTP_SERVER") + .expect("OpenSSH sftp-server executable required"); + let fixture = Fixture::with_openssh(Some(executable)).await; + let local_root = fixture._dir.path().join("local user skill"); + let remote_root = fixture._dir.path().join("remote workspace"); + let cache_root = remote_root.join(".skill-cache"); + for directory in ["scripts", "references", "templates"] { + std::fs::create_dir_all(local_root.join(directory)).unwrap(); + } + std::fs::create_dir(&remote_root).unwrap(); + std::fs::write( + local_root.join("SKILL.md"), + "Read references/guide.md and run scripts/hello.sh.", + ) + .unwrap(); + std::fs::write(local_root.join("references/guide.md"), "remote reference").unwrap(); + std::fs::write(local_root.join("templates/binary.dat"), [0, 255, 1, 13, 10]).unwrap(); + std::fs::write( + local_root.join("references/quoted ' Unicode 你好.txt"), + "quoted path", + ) + .unwrap(); + let script = local_root.join("scripts/hello.sh"); + std::fs::write(&script, "#!/bin/sh\nscript_dir=${0%/*}\ncat \"$script_dir/../references/guide.md\"\nprintf '\\n'\npwd -P\n").unwrap(); + std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).unwrap(); + + let file_service = crate::remote_ssh::RemoteFileService::new(Arc::new( + tokio::sync::RwLock::new(Some(fixture.manager.as_ref().clone())), + )); + let fs = RemoteWorkspaceFs::new("sftp-test".into(), file_service); + let exclusive_directory = remote_root.join("exclusive-directory"); + fs.create_dir_new(exclusive_directory.to_str().unwrap()) + .await + .unwrap(); + assert!(fs + .create_dir_new(exclusive_directory.to_str().unwrap()) + .await + .is_err()); + fs.remove_dir(exclusive_directory.to_str().unwrap(), false) + .await + .unwrap(); + let shell = RemoteWorkspaceShell::new( + "sftp-test".into(), + fixture.manager.as_ref().clone(), + remote_root.to_str().unwrap().into(), + ); + let cache = cache_root.to_str().unwrap(); + let first = sync_local_skill_package( + &local_root, + cache, + remote_root.to_str().unwrap(), + "user::test::hello", + &fs, + None, + ) + .await + .unwrap(); + assert_eq!( + fixture.state.exec_calls.load(Ordering::SeqCst), + 0, + "Synchronization must use SSH/SFTP file IO without Python or shell execution" + ); + assert!(!first.reused); + assert!(first.path.starts_with(cache)); + assert!(!first.path.contains("local user skill")); + let remote_reference = format!("{}/references/guide.md", first.path); + let remote_template = format!("{}/templates/binary.dat", first.path); + assert_eq!( + fs.read_file_text(&remote_reference).await.unwrap(), + "remote reference" + ); + assert_eq!( + fs.read_file(&remote_template).await.unwrap(), + [0, 255, 1, 13, 10] + ); + let reference_before = fs + .metadata(&remote_reference, false) + .await + .unwrap() + .unwrap(); + let attributes_before = fixture + .manager + .sftp_stat("sftp-test", &remote_reference) + .await + .unwrap(); + let modified = std::time::UNIX_EPOCH + std::time::Duration::from_secs(1_700_000_000); + fs.set_modified(&remote_reference, modified).await.unwrap(); + let reference_after = fs + .metadata(&remote_reference, false) + .await + .unwrap() + .unwrap(); + assert_eq!(reference_after.modified, Some(modified)); + let attributes_after = fixture + .manager + .sftp_stat("sftp-test", &remote_reference) + .await + .unwrap(); + assert_eq!( + attributes_after.atime, attributes_before.atime, + "SFTP v3's paired access time must be preserved when changing modification time" + ); + assert_eq!( + (attributes_after.uid, attributes_after.gid), + (attributes_before.uid, attributes_before.gid), + "Metadata updates must not change ownership" + ); + assert_eq!( + reference_after.size, reference_before.size, + "SFTP mtime updates must not truncate the synchronized file" + ); + assert_eq!( + reference_after.permissions, reference_before.permissions, + "SFTP mtime updates must not change executable or data permissions" + ); + assert_eq!( + fs.read_file_text(&remote_reference).await.unwrap(), + "remote reference" + ); + assert_eq!( + fs.read_file_text(&format!( + "{}/references/quoted ' Unicode 你好.txt", + first.path + )) + .await + .unwrap(), + "quoted path" + ); + let script_command = + crate::remote_ssh::shell::quote_arg(&format!("{}/scripts/hello.sh", first.path)); + let (stdout, stderr, status) = shell.exec(&script_command, Some(30_000)).await.unwrap(); + assert_eq!(status, 0, "{stderr}"); + assert_eq!( + fixture.state.exec_calls.load(Ordering::SeqCst), + 1, + "The selected Skill script is the only remote command" + ); + assert_eq!( + stdout, + format!( + "remote reference\n{}\n", + remote_root.canonicalize().unwrap().display() + ) + ); + + let unchanged = sync_local_skill_package( + &local_root, + cache, + remote_root.to_str().unwrap(), + "user::test::hello", + &fs, + None, + ) + .await + .unwrap(); + assert!(unchanged.reused); + assert_eq!(unchanged.path, first.path); + fs.write_file(&remote_reference, b"remote corruption") + .await + .unwrap(); + let repaired = sync_local_skill_package( + &local_root, + cache, + remote_root.to_str().unwrap(), + "user::test::hello", + &fs, + None, + ) + .await + .unwrap(); + assert!(!repaired.reused); + assert_ne!( + repaired.path, first.path, + "Repair must publish a new immutable generation instead of moving an active path" + ); + assert_eq!( + fs.read_file_text(&remote_reference).await.unwrap(), + "remote corruption", + "Previously published generations must remain in place during repair" + ); + assert_eq!( + fs.read_file_text(&format!("{}/references/guide.md", repaired.path)) + .await + .unwrap(), + "remote reference" + ); + + let repaired_reuse = sync_local_skill_package( + &local_root, + cache, + remote_root.to_str().unwrap(), + "user::test::hello", + &fs, + None, + ) + .await + .unwrap(); + assert!(repaired_reuse.reused); + assert_eq!(repaired_reuse.path, repaired.path); + + std::fs::write( + local_root.join("references/guide.md"), + "updated local reference", + ) + .unwrap(); + std::fs::remove_file(local_root.join("templates/binary.dat")).unwrap(); + let updated = sync_local_skill_package( + &local_root, + cache, + remote_root.to_str().unwrap(), + "user::test::hello", + &fs, + None, + ) + .await + .unwrap(); + assert!(!updated.reused); + assert_ne!(updated.content_hash, first.content_hash); + assert_eq!( + fs.read_file_text(&format!("{}/references/guide.md", updated.path)) + .await + .unwrap(), + "updated local reference" + ); + assert!(!fs + .exists(&format!("{}/templates/binary.dat", updated.path)) + .await + .unwrap()); + assert_eq!( + fixture.state.exec_calls.load(Ordering::SeqCst), + 1, + "Cache checking, corruption repair and updates must not execute target commands" + ); +} diff --git a/src/crates/services/services-integrations/src/remote_ssh/workspace_services.rs b/src/crates/services/services-integrations/src/remote_ssh/workspace_services.rs index 66f9b07e4b..f749613262 100644 --- a/src/crates/services/services-integrations/src/remote_ssh/workspace_services.rs +++ b/src/crates/services/services-integrations/src/remote_ssh/workspace_services.rs @@ -231,6 +231,12 @@ impl WorkspaceFileSystem for RemoteWorkspaceFs { .await } + async fn create_dir_new(&self, path: &str) -> anyhow::Result<()> { + self.file_service + .create_dir(&self.connection_id, path) + .await + } + #[cfg(feature = "remote-ssh-concrete")] async fn set_permissions(&self, path: &str, permissions: u32) -> anyhow::Result<()> { self.file_service diff --git a/src/web-ui/src/app/scenes/settings/pages/development/RemoteUserSkillsSection.test.tsx b/src/web-ui/src/app/scenes/settings/pages/development/RemoteUserSkillsSection.test.tsx new file mode 100644 index 0000000000..7f410ff569 --- /dev/null +++ b/src/web-ui/src/app/scenes/settings/pages/development/RemoteUserSkillsSection.test.tsx @@ -0,0 +1,108 @@ +// @vitest-environment jsdom +import { act } from 'react'; +import { createRoot, type Root } from 'react-dom/client'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { activateSurface } from '@/infrastructure/peer-device/deviceSurface'; +import RemoteUserSkillsSection from './RemoteUserSkillsSection'; + +const mocks = vi.hoisted(() => ({ get: vi.fn(), set: vi.fn(), error: vi.fn() })); +vi.mock('@/infrastructure/config/services/ConfigManager', () => ({ configManager: { + hasCachedConfig: () => false, getCachedConfig: () => undefined, + getConfig: mocks.get, setConfig: mocks.set, +} })); +vi.mock('@openbitfun/ui', async original => ({ + ...await original(), + Select: ({ value, disabled, options, onValueChange }: { + value: string; disabled: boolean; options: Array<{ value: string; label: string }>; + onValueChange: (value: string) => void; + }) => , +})); +vi.mock('@/infrastructure/i18n', () => ({ useI18n: () => ({ t: (key: string) => key }) })); +vi.mock('@/shared/notification-system', () => ({ notificationService: { error: mocks.error } })); + +describe('remote user Skill settings', () => { + let root: Root; + let container: HTMLDivElement; + const select = () => container.querySelector('select')!; + const choose = async (value: string) => act(async () => { + select().value = value; + select().dispatchEvent(new Event('change', { bubbles: true })); + }); + + beforeEach(() => { + globalThis.IS_REACT_ACT_ENVIRONMENT = true; + activateSurface('local'); + vi.clearAllMocks(); + mocks.get.mockReset().mockResolvedValue({ remote_user_skills: 'skill_md_only' }); + mocks.set.mockReset().mockResolvedValue(undefined); + container = document.createElement('div'); + document.body.append(container); + root = createRoot(container); + }); + afterEach(() => { act(() => root.unmount()); container.remove(); activateSurface('local'); }); + + it('preserves the markdown-only default and saves each choice without replacing other workspace settings', async () => { + await act(async () => root.render()); + expect(select().value).toBe('skill_md_only'); + expect(container.textContent).toContain('remoteUserSkills.skillMdOnlyDescription'); + await choose('full'); + expect(mocks.set).toHaveBeenLastCalledWith('workspace.remote_user_skills', 'full'); + expect(container.textContent).toContain('remoteUserSkills.syncHint'); + await choose('disabled'); + expect(mocks.set).toHaveBeenLastCalledWith('workspace.remote_user_skills', 'disabled'); + expect(container.textContent).not.toContain('remoteUserSkills.syncHint'); + await choose('skill_md_only'); + expect(mocks.set).toHaveBeenLastCalledWith('workspace.remote_user_skills', 'skill_md_only'); + expect(select().value).toBe('skill_md_only'); + }); + + it('locks an older host or unknown future mode instead of writing unsupported configuration', async () => { + mocks.get.mockResolvedValue({ remote_user_skills: 'future_mode' }); + await act(async () => root.render()); + expect(select().disabled).toBe(true); + expect(container.textContent).toContain('remoteUserSkills.unsupported'); + expect(mocks.set).not.toHaveBeenCalled(); + }); + + it('keeps older hosts without the field explicitly unsupported', async () => { + mocks.get.mockResolvedValue({}); + await act(async () => root.render()); + expect(select().disabled).toBe(true); + expect(container.textContent).toContain('remoteUserSkills.unsupported'); + expect(mocks.set).not.toHaveBeenCalled(); + }); + + it('preserves the persisted mode after a failed write', async () => { + mocks.set.mockRejectedValue(new Error('Host unavailable')); + await act(async () => root.render()); + await choose('full'); + expect(select().value).toBe('skill_md_only'); + expect(mocks.error).toHaveBeenCalledWith('remoteUserSkills.saveFailed'); + expect(select().disabled).toBe(false); + }); + + it('locks a failed read and retries without writing a default', async () => { + mocks.get.mockRejectedValueOnce(new Error('Host unavailable')); + await act(async () => root.render()); + expect(select()).toBeNull(); + expect(container.textContent).toContain('remoteUserSkills.loadFailed'); + expect(mocks.set).not.toHaveBeenCalled(); + await act(async () => container.querySelector('button')!.click()); + expect(select().value).toBe('skill_md_only'); + expect(mocks.set).not.toHaveBeenCalled(); + }); + + it('ignores a completed write from the previously selected device', async () => { + let finishSave!: () => void; + mocks.set.mockImplementation(() => new Promise(resolve => { finishSave = resolve; })); + await act(async () => root.render()); + await choose('full'); + expect(select().disabled).toBe(true); + activateSurface('peer-device'); + await act(async () => finishSave()); + expect(select().value).toBe('skill_md_only'); + expect(mocks.error).not.toHaveBeenCalled(); + }); +}); diff --git a/src/web-ui/src/app/scenes/settings/pages/development/RemoteUserSkillsSection.tsx b/src/web-ui/src/app/scenes/settings/pages/development/RemoteUserSkillsSection.tsx new file mode 100644 index 0000000000..c0884d68c9 --- /dev/null +++ b/src/web-ui/src/app/scenes/settings/pages/development/RemoteUserSkillsSection.tsx @@ -0,0 +1,98 @@ +import { useCallback, useEffect, useRef, useState } from 'react'; +import { Select } from '@openbitfun/ui'; +import { + ConfigLoadingState, ConfigMessage, ConfigPageRow, ConfigPageSection, ConfigRetryState, +} from '@/infrastructure/config/components/common'; +import { useConfigSeed } from '@/infrastructure/config/hooks/useConfigSeed'; +import { configManager } from '@/infrastructure/config/services/ConfigManager'; +import type { RemoteUserSkillsMode, WorkspaceConfig } from '@/infrastructure/config/types'; +import { useI18n } from '@/infrastructure/i18n'; +import { getActiveSurfaceScope } from '@/infrastructure/peer-device/deviceSurface'; +import { notificationService } from '@/shared/notification-system'; +import { createLogger } from '@/shared/utils/logger'; +import { useSettingsSectionAnchor } from '../shared/useSettingsSectionAnchor'; + +const log = createLogger('RemoteUserSkillsSection'); + +function supportedMode(value: unknown): RemoteUserSkillsMode | undefined { + return value === 'skill_md_only' || value === 'disabled' || value === 'full' ? value : undefined; +} + +export default function RemoteUserSkillsSection() { + const { t } = useI18n('settings/runtime'); + const seed = useConfigSeed(['workspace']); + const [mode, setMode] = useState(() => supportedMode(seed.get>('workspace', {}).remote_user_skills)); + const [loading, setLoading] = useState(!seed.loaded); + const [loadFailed, setLoadFailed] = useState(false); + const [saving, setSaving] = useState(false); + const saveInFlight = useRef(false); + const sectionAnchor = useSettingsSectionAnchor('remote-user-skills', !loading); + + const load = useCallback(async () => { + const scope = getActiveSurfaceScope(); + setLoading(true); + setLoadFailed(false); + try { + const workspace = await configManager.getConfig('workspace'); + if (scope.isCurrent()) setMode(supportedMode(workspace.remote_user_skills)); + } catch (reason) { + if (scope.isCurrent()) { + log.error('Failed to load remote user Skill settings', reason); + setLoadFailed(true); + } + } finally { + if (scope.isCurrent()) setLoading(false); + } + }, []); + + useEffect(() => { if (!seed.loaded) void load(); }, [load, seed.loaded]); + + const save = async (next: RemoteUserSkillsMode) => { + if (saveInFlight.current || mode === undefined || next === mode) return; + const scope = getActiveSurfaceScope(); + saveInFlight.current = true; + setSaving(true); + try { + await configManager.setConfig('workspace.remote_user_skills', next); + if (scope.isCurrent()) setMode(next); + } catch (reason) { + if (scope.isCurrent()) { + log.error('Failed to save remote user Skill settings', reason); + notificationService.error(t('remoteUserSkills.saveFailed')); + } + } finally { + saveInFlight.current = false; + if (scope.isCurrent()) setSaving(false); + } + }; + + const description = mode === 'disabled' ? t('remoteUserSkills.disabledDescription') + : mode === 'full' ? t('remoteUserSkills.fullDescription') + : mode === 'skill_md_only' ? t('remoteUserSkills.skillMdOnlyDescription') + : t('remoteUserSkills.unsupported'); + + return ( + + {loading ? + : loadFailed ? void load()} /> + : <> + +