From d5dec0fffb6e6bee8e94379e51fb9292120342e8 Mon Sep 17 00:00:00 2001 From: Scott Werner Date: Tue, 8 Sep 2026 12:39:21 -0400 Subject: [PATCH] Add content-based workflow version registration tools --- Cargo.lock | 2 + docs/public/agents/mcp.mdx | 42 +- docs/public/api-reference/fabro-api.yaml | 3 + docs/public/execution/run-configuration.mdx | 2 +- lib/apps/fabro-cli/src/commands/run/runner.rs | 4 +- lib/apps/fabro-cli/tests/it/cmd/mcp.rs | 19 + lib/apps/fabro-mcp-server/src/server.rs | 82 +++- lib/apps/fabro-server/src/lib.rs | 1 + .../src/server/handler/workflow_versions.rs | 5 +- lib/apps/fabro-server/src/server/tests.rs | 70 +++ .../fabro-server/src/workflow_version_tool.rs | 461 ++++++++++++++++++ lib/components/fabro-manifest/src/lib.rs | 1 + .../fabro-manifest/src/workflow_bundler.rs | 28 +- .../src/workflow_version_collector.rs | 10 +- lib/components/fabro-tool/src/common.rs | 28 ++ lib/components/fabro-tool/src/fabro_client.rs | 30 +- lib/components/fabro-tool/src/lib.rs | 9 +- .../fabro-tool/src/workflow_version.rs | 234 +++++++++ lib/foundation/fabro-config/src/project.rs | 15 + lib/foundation/fabro-types/Cargo.toml | 2 + lib/foundation/fabro-types/src/lib.rs | 1 + .../fabro-types/src/workflow_version.rs | 97 +++- .../src/api/workflow-versions-api.ts | 8 +- 23 files changed, 1109 insertions(+), 45 deletions(-) create mode 100644 lib/apps/fabro-server/src/workflow_version_tool.rs create mode 100644 lib/components/fabro-tool/src/workflow_version.rs diff --git a/Cargo.lock b/Cargo.lock index 58a7ae412..b7c953564 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3214,6 +3214,8 @@ dependencies = [ "thiserror 2.0.18", "toml 0.8.23", "ulid", + "unicase", + "unicode-normalization", "url", ] diff --git a/docs/public/agents/mcp.mdx b/docs/public/agents/mcp.mdx index b0920090f..9ee9bf5a1 100644 --- a/docs/public/agents/mcp.mdx +++ b/docs/public/agents/mcp.mdx @@ -5,7 +5,7 @@ description: "Connect MCP tools to agents and expose Fabro runs to MCP clients" MCP ([Model Context Protocol](https://modelcontextprotocol.io/)) lets you connect external tool servers to Fabro agents. An MCP server exposes tools over a standardized protocol — databases, APIs, file systems, custom services — and Fabro discovers and registers them automatically. Agents call MCP tools the same way they call built-in tools. -Fabro can also run as an MCP server. MCP clients can use Fabro's run-management tools to create, inspect, control, wait for, and read events from workflow runs through the authenticated `fabro` CLI. +Fabro can also run as an MCP server. MCP clients can register reusable workflow versions and use Fabro's run-management tools to create, inspect, control, wait for, and read events from workflow runs through the authenticated `fabro` CLI. Workflow agents can opt in to that same run-management tool catalog with `[run.agent] fabro_tools = true`. This is not the same as configuring external MCP servers for the agent. When a workflow agent calls `fabro_run_create`, created runs are always [child runs](/execution/child-runs) of the current run; an explicit `parent_id` must match the current run ID. @@ -39,6 +39,7 @@ fabro mcp init claude --name fabro-testing --server https://fabro-testing.exampl | Tool | Purpose | |---|---| +| `fabro_workflow_version_create` | Register supplied workflow contents and local dependencies as an immutable version ID, without creating a run. | | `fabro_run_create` | Create one or more workflow runs, optionally under a parent run, starting them by default. | | `fabro_run_search` | Search runs by ID, parent, workflow, labels, status, archive state, and creation time. | | `fabro_run_get` | Read-only inspection of a run: returns its summary, projection, and pending questions without mutating state. | @@ -47,6 +48,45 @@ fabro mcp init claude --name fabro-testing --server https://fabro-testing.exampl | `fabro_run_pair` | Inspect, start, message, end, or read transcript for a live run pairing session. | | `fabro_run_events` | List, inspect, or search stored events for a run. | +### Register workflow contents from a sandbox + +Use shell and read tools in your sandbox to acquire the workflow and all its local +config, graph, prompt, script, import, and child-workflow files. For example, clone +a repository with your sandbox's shell tool, then read `workflow.fabro` and its +referenced `prompt.md`. Submit the actual contents: + +```json +{ + "entrypoint": "workflow.fabro", + "files": { + "workflow.fabro": "digraph W { start [shape=Mdiamond] work [prompt=\"@prompt.md\"] exit [shape=Msquare] start -> work -> exit }", + "prompt.md": "Review the implementation." + } +} +``` + +Call `fabro_workflow_version_create` with this object and keep the returned +`workflow_version_id`. You can reuse it in a `RunIntent` submitted through the +[Create Run API](/api-reference/runs/create-run). Registration packages and uploads +child workflows before their parents; callers do not calculate dependency IDs. + +`entrypoint` is an exact supplied key, including when it has no extension. File +values are text, never host paths or URLs to fetch. Missing references and paths +that escape the supplied tree fail. The source tree is limited to 512 files, +512 KiB per file, and 2 MiB of text; each resulting serialized version must also +fit the existing 2 MiB API limit. Case-insensitive file and ancestor collisions +are rejected before staging. Collection follows declared file references; command +`script` values remain literal text, and paths embedded in shell commands are not +inspected or acquired. + +Registration resolves no runtime secrets, selects no environment, and starts no +execution. It requires a user credential or a worker token with `agent:run_tools`; +ordinary worker tokens and same-run Ask Fabro sessions cannot register versions. +If an upload fails, retry the same contents: previously registered immutable +versions remain reusable. The existing `fabro_run_create` input is unchanged. + +### Create runs + For a simple create call, `fabro_run_create` accepts a workflow selector string: ```json diff --git a/docs/public/api-reference/fabro-api.yaml b/docs/public/api-reference/fabro-api.yaml index 9ec0dbb2d..96ca10147 100644 --- a/docs/public/api-reference/fabro-api.yaml +++ b/docs/public/api-reference/fabro-api.yaml @@ -1086,6 +1086,9 @@ paths: description: >- Validates and stores an immutable workflow package in content-addressed storage. Repeating the same canonical content returns the same identifier. + Requires an authenticated user or a worker token with the `agent:run_tools` + capability. Ordinary worker tokens cannot register versions. Registration + creates no run and starts no execution. requestBody: required: true content: diff --git a/docs/public/execution/run-configuration.mdx b/docs/public/execution/run-configuration.mdx index 1a617e977..79637e882 100644 --- a/docs/public/execution/run-configuration.mdx +++ b/docs/public/execution/run-configuration.mdx @@ -509,7 +509,7 @@ Configure workflow agent behavior that is not tied to a single stage. fabro_tools = true ``` -`fabro_tools` defaults to `false`. Set it to `true` only for runs whose agents should be able to use the same Fabro run-management MCP tool catalog exposed to human MCP clients: create, search, get, interact, gather, events, and pair. +`fabro_tools` defaults to `false`. Set it to `true` only for runs whose agents should be able to use the same Fabro run-management MCP tool catalog exposed to human MCP clients: workflow version registration, create, search, get, interact, gather, events, and pair. One workflow-agent exception is intentional: `fabro_run_create` always creates [child runs](/execution/child-runs) parented to the current run. If an agent supplies `parent_id`, it must match the current run ID. diff --git a/lib/apps/fabro-cli/src/commands/run/runner.rs b/lib/apps/fabro-cli/src/commands/run/runner.rs index 16776bc88..df7b2f531 100644 --- a/lib/apps/fabro-cli/src/commands/run/runner.rs +++ b/lib/apps/fabro-cli/src/commands/run/runner.rs @@ -16,6 +16,7 @@ use fabro_interview::{ WorkerControlMessage, }; use fabro_server::run_tool_manifest; +use fabro_server::workflow_version_tool::ServerWorkflowVersionCreateAdapter; use fabro_store::{EventEnvelope, RunProjection, RunProjectionReducer}; use fabro_tool::fabro_client::ClientBackend; use fabro_types::settings::run::{RunMode, RunNamespace}; @@ -236,7 +237,8 @@ fn build_fabro_run_tool_services( return None; } let backend = ClientBackend::new(Arc::new(client)) - .with_manifest_builder(Arc::new(WorkerRunManifestBuilder)); + .with_manifest_builder(Arc::new(WorkerRunManifestBuilder)) + .with_workflow_version_create_adapter(Arc::new(ServerWorkflowVersionCreateAdapter)); Some(FabroRunToolServices { backend: Arc::new(backend), current_run_id, diff --git a/lib/apps/fabro-cli/tests/it/cmd/mcp.rs b/lib/apps/fabro-cli/tests/it/cmd/mcp.rs index 552134dcb..d791334b4 100644 --- a/lib/apps/fabro-cli/tests/it/cmd/mcp.rs +++ b/lib/apps/fabro-cli/tests/it/cmd/mcp.rs @@ -38,6 +38,7 @@ const MCP_RUN_TOOL_NAMES: &[&str] = &[ "fabro_run_interact", "fabro_run_pair", "fabro_run_search", + "fabro_workflow_version_create", ]; async fn assert_mcp_run_tool_count(client: &McpClient) { @@ -1847,6 +1848,24 @@ async fn mcp_get_rejects_blank_run_id_before_auth_or_network() { .expect("MCP client should shut down"); } +#[tokio::test(flavor = "multi_thread")] +async fn mcp_workflow_version_validation_happens_before_auth_or_network() { + let context = test_context!(); + let client = spawn_mcp_client(&context, &["--server", "http://127.0.0.1:9"]).await; + let error = call_tool_error_text( + &client, + "fabro_workflow_version_create", + serde_json::json!({"entrypoint":"workflow","files":{}}), + ) + .await; + assert_eq!(error, "entrypoint must be an exact supplied file key"); + assert_mcp_run_tool_count(&client).await; + client + .shutdown() + .await + .expect("MCP client should shut down"); +} + #[tokio::test(flavor = "multi_thread")] async fn mcp_create_validation_errors_happen_before_auth_or_network() { let context = test_context!(); diff --git a/lib/apps/fabro-mcp-server/src/server.rs b/lib/apps/fabro-mcp-server/src/server.rs index 6a9c9941d..794da973a 100644 --- a/lib/apps/fabro-mcp-server/src/server.rs +++ b/lib/apps/fabro-mcp-server/src/server.rs @@ -3,6 +3,7 @@ use std::sync::Arc; use std::time::Duration; use anyhow::Result; +use fabro_server::workflow_version_tool::ServerWorkflowVersionCreateAdapter; use fabro_tool::fabro_client::ClientBackend; use fabro_tool::{self as run_tools, FabroToolBackend}; use fabro_util::version::FABRO_VERSION; @@ -74,7 +75,7 @@ impl ServerHandler for FabroMcpServer { fn get_info(&self) -> ServerInfo { ServerInfo::new(ServerCapabilities::builder().enable_tools().build()) .with_server_info(Implementation::new(SERVER_NAME, FABRO_VERSION).with_title("Fabro")) - .with_instructions("Use these tools to create, inspect, control, wait for, and read events from Fabro workflow runs.") + .with_instructions("Use these tools to register workflow versions and create, inspect, control, wait for, and read events from Fabro workflow runs.") } } @@ -90,6 +91,27 @@ impl FabroMcpServer { } } + #[tool( + name = "fabro_workflow_version_create", + description = "Register supplied workflow file contents and all local dependencies as a reusable immutable workflow version ID. Obtain files with shell/read tools first; this does not create or start a run." + )] + async fn fabro_workflow_version_create( + &self, + params: Parameters, + ) -> Result { + if let Err(err) = params.0.validate() { + return Ok(error_result(&err)); + } + let backend = match self.backend().await { + Ok(backend) => backend, + Err(err) => return Ok(error_result(&err)), + }; + match run_tools::create_workflow_version(backend, params.0).await { + Ok(result) => success_result(&result, run_tools::workflow_version_create_text(&result)), + Err(err) => Ok(error_result(&err)), + } + } + #[tool( name = "fabro_run_create", description = "Create one or more Fabro workflow runs, optionally under a parent run, starting them by default." @@ -252,7 +274,10 @@ impl FabroMcpServer { .map(|client| { Arc::new( ClientBackend::new(Arc::new(client)) - .with_manifest_builder(Arc::new(McpRunManifestBuilder)), + .with_manifest_builder(Arc::new(McpRunManifestBuilder)) + .with_workflow_version_create_adapter(Arc::new( + ServerWorkflowVersionCreateAdapter, + )), ) as Arc }) .map_err(|err| run_tools::ToolError::from_anyhow(&err)) @@ -283,6 +308,7 @@ fn error_result(err: &run_tools::ToolError) -> CallToolResult { #[cfg(test)] mod tests { + use std::collections::BTreeMap; use std::path::PathBuf; use std::sync::Arc; @@ -291,6 +317,58 @@ mod tests { use super::*; use crate::FabroMcpServerSettings; + #[tokio::test] + async fn workflow_version_mcp_surface_matches_catalog_and_returns_minimal_result() { + let mock = httpmock::MockServer::start_async().await; + let version = fabro_types::WorkflowVersion::new( + "workflow".parse().unwrap(), + BTreeMap::from([("workflow".parse().unwrap(), "digraph W {}".to_string())]), + BTreeMap::new(), + ) + .unwrap(); + let id = version.id().unwrap(); + let upload = mock + .mock_async(|when, then| { + when.method(httpmock::Method::POST) + .path("/api/v1/workflow-versions") + .json_body_obj(&version); + then.status(201) + .json_body(serde_json::json!({"workflow_version_id":id})); + }) + .await; + let url = mock.url(""); + let server = FabroMcpServer::new(Arc::new(FabroMcpServerSettings { + cwd: PathBuf::from("/does-not-exist"), + config_path: PathBuf::from("/does-not-exist"), + client_factory: Arc::new(move || { + let url = url.clone(); + Box::pin(async move { fabro_client::Client::new_no_proxy(&url) }) + }), + })); + let tools = server.tool_router.list_all(); + let tool = tools + .iter() + .find(|tool| tool.name == "fabro_workflow_version_create") + .unwrap(); + let definition = run_tools::tool_definitions() + .iter() + .find(|tool| tool.name == "fabro_workflow_version_create") + .unwrap(); + let mut expected = definition.parameters.clone(); + expected.as_object_mut().unwrap().remove("$schema"); + let mut actual = Value::Object(tool.input_schema.as_ref().clone()); + actual.as_object_mut().unwrap().remove("$schema"); + assert_eq!(actual, expected); + assert_eq!(tool.description.as_deref(), Some(definition.description)); + let result = server.fabro_workflow_version_create(Parameters(serde_json::from_value(serde_json::json!({"entrypoint":"workflow","files":{"workflow":"digraph W {}"}})).unwrap())).await.unwrap(); + assert_eq!( + result.structured_content, + Some(serde_json::json!({"workflow_version_id":id})) + ); + assert_ne!(result.is_error, Some(true)); + upload.assert_calls_async(1).await; + } + #[test] fn server_info_reports_fabro_version() { let settings = FabroMcpServerSettings { diff --git a/lib/apps/fabro-server/src/lib.rs b/lib/apps/fabro-server/src/lib.rs index 98f7a2c77..227855389 100644 --- a/lib/apps/fabro-server/src/lib.rs +++ b/lib/apps/fabro-server/src/lib.rs @@ -55,6 +55,7 @@ pub mod web_auth; mod worker_control; mod worker_runtime; mod worker_token; +pub mod workflow_version_tool; pub use error::{ApiError, Error, Result}; pub use run_manifest::workflow_bundle_from_manifest; diff --git a/lib/apps/fabro-server/src/server/handler/workflow_versions.rs b/lib/apps/fabro-server/src/server/handler/workflow_versions.rs index 69845679b..cd8daafc7 100644 --- a/lib/apps/fabro-server/src/server/handler/workflow_versions.rs +++ b/lib/apps/fabro-server/src/server/handler/workflow_versions.rs @@ -10,8 +10,9 @@ use fabro_workflow_version::{ }; use super::super::{ - ApiError, AppState, IntoResponse, Json, RequiredUser, Response, Router, State, StatusCode, post, + ApiError, AppState, IntoResponse, Json, Response, Router, State, StatusCode, post, }; +use crate::principal_middleware::RequiredRunManagementActor; const INVALID_JSON_CODE: &str = "invalid_json"; const INVALID_VERSION_CODE: &str = "workflow_version_invalid"; @@ -26,7 +27,7 @@ pub(super) fn routes() -> Router> { } async fn create_workflow_version( - _auth: RequiredUser, + _auth: RequiredRunManagementActor, State(state): State>, payload: Result, JsonRejection>, ) -> Result { diff --git a/lib/apps/fabro-server/src/server/tests.rs b/lib/apps/fabro-server/src/server/tests.rs index 3be6e1a29..e81a3b576 100644 --- a/lib/apps/fabro-server/src/server/tests.rs +++ b/lib/apps/fabro-server/src/server/tests.rs @@ -19738,3 +19738,73 @@ fn validate_github_slug_rejects_overlong() { let long = "a".repeat(40); assert!(super::validate_github_slug("owner", &long, 39).is_err()); } + +#[tokio::test] +async fn workflow_version_registration_requires_user_or_run_tools_capability() { + let (state, app) = jwt_auth_app(); + let run_id = RunId::new(); + let body = json!({"entrypoint":"workflow.fabro","files":{"workflow.fabro":"digraph W {}"},"workflow_dependencies":{}}); + for (token, expected) in [ + (issue_test_user_jwt(), StatusCode::CREATED), + ( + issue_test_run_tools_worker_token(&run_id), + StatusCode::CREATED, + ), + (issue_test_worker_token(&run_id), StatusCode::FORBIDDEN), + ] { + let response = app + .clone() + .oneshot(json_bearer_request( + Method::POST, + "/workflow-versions", + &token, + &body, + )) + .await + .unwrap(); + fabro_test::expect_axum_status(response, expected, "POST /workflow-versions actor matrix") + .await; + } + for body in [serde_json::to_string(&body).unwrap(), "{".to_string()] { + let response = app + .clone() + .oneshot( + Request::builder() + .method(Method::POST) + .uri(api("/workflow-versions")) + .header(header::CONTENT_TYPE, "application/json") + .body(Body::from(body)) + .unwrap(), + ) + .await + .unwrap(); + fabro_test::expect_axum_status( + response, + StatusCode::UNAUTHORIZED, + "anonymous POST /workflow-versions", + ) + .await; + } + let response = app + .oneshot(bearer_request( + Method::GET, + "/runs", + &issue_test_user_jwt(), + Body::empty(), + )) + .await + .unwrap(); + let listed = + fabro_test::expect_axum_json(response, StatusCode::OK, "GET /runs after registration") + .await; + assert_eq!(listed["data"], json!([])); + assert!( + state + .stores + .runs + .load_run_projection(&run_id) + .await + .unwrap() + .is_none() + ); +} diff --git a/lib/apps/fabro-server/src/workflow_version_tool.rs b/lib/apps/fabro-server/src/workflow_version_tool.rs new file mode 100644 index 000000000..27081eea0 --- /dev/null +++ b/lib/apps/fabro-server/src/workflow_version_tool.rs @@ -0,0 +1,461 @@ +use std::path::Path; + +use anyhow::Context as _; +use async_trait::async_trait; +use fabro_client::Client; +use fabro_config::project::WorkflowLocation; +use fabro_manifest::CollectedWorkflowClosure; +use fabro_tool::{FabroWorkflowVersionCreateParams, ToolError, WorkflowVersionCreateAdapter}; +use fabro_types::WorkflowVersionId; +use tokio::task; + +/// Content-only registration shared by standalone MCP and capable run workers. +pub struct ServerWorkflowVersionCreateAdapter; + +#[async_trait] +impl WorkflowVersionCreateAdapter for ServerWorkflowVersionCreateAdapter { + async fn create_workflow_version( + &self, + params: FabroWorkflowVersionCreateParams, + client: &Client, + ) -> anyhow::Result { + params.validate()?; + let closure = task::spawn_blocking(move || { + let staging = tempfile::Builder::new().prefix("fabro-workflow-version-").tempdir()?; + collect_supplied_workflow(¶ms, staging) + }) + .await + .context("workflow packaging task failed")? + // Parser diagnostics may contain supplied source. Keep them off the + // tool result boundary, including nested error chains. + .map_err(|_| ToolError::message("workflow source could not be packaged; check configuration, syntax, local references, and package limits"))?; + let versions = closure + .versions() + .map(|(_, version)| version.version()) + .collect::>(); + client.register_workflow_versions(versions).await?; + Ok(closure.root_id()) + } +} + +#[expect( + clippy::disallowed_methods, + reason = "bounded file staging and collection run on spawn_blocking" +)] +fn collect_supplied_workflow( + params: &FabroWorkflowVersionCreateParams, + staging: tempfile::TempDir, +) -> anyhow::Result { + // TempDir owns cleanup on every return path, including collection errors. + let root = staging.path().canonicalize()?; + for (path, contents) in ¶ms.files { + let destination = root.join(path.as_str()); + if let Some(parent) = destination.parent() { + std::fs::create_dir_all(parent)?; + } + std::fs::write(destination, contents)?; + } + let location = WorkflowLocation::from_exact_path(Path::new(params.entrypoint.as_str()), &root)?; + let closure = fabro_manifest::collect_workflow_versions_at_location( + &location, + &root, + Path::new(params.entrypoint.as_str()), + )?; + // Collection validates the whole closure, including serialized request budgets. + // A case-insensitive host must not satisfy a missing exact source key. + for (_, version) in closure.versions() { + for (path, content) in version.version().files() { + anyhow::ensure!( + params.files.get(path) == Some(content), + "collected file does not match supplied source" + ); + } + } + drop(staging); + Ok(closure) +} + +#[cfg(test)] +mod tests { + #![expect( + clippy::disallowed_methods, + reason = "hermetic temporary source fixtures" + )] + use std::path::Path; + use std::sync::{Arc, Mutex}; + + use axum::http::StatusCode; + use axum::routing::post; + use axum::{Json, Router}; + use fabro_types::WorkflowVersion; + use serde_json::json; + + use super::*; + + fn params(entrypoint: &str, files: &[(&str, &str)]) -> FabroWorkflowVersionCreateParams { + FabroWorkflowVersionCreateParams { + entrypoint: entrypoint.parse().unwrap(), + files: files + .iter() + .map(|(path, content)| (path.parse().unwrap(), (*content).to_string())) + .collect(), + } + } + + fn fixture() -> FabroWorkflowVersionCreateParams { + params("workflow.toml", &[ + ( + "workflow.toml", + "_version = 1\n[workflow]\ngraph = \"workflow.fabro\"\n", + ), + ( + "workflow.fabro", + r#"digraph W { p [prompt="@prompt.md"] child [stack.child_workflow="child.fabro"] }"#, + ), + ( + "prompt.md", + "Keep {{ secrets.TEST }} and {{ env.TEST }} for runtime.", + ), + ("child.fabro", "digraph Child {}"), + ]) + } + + fn collect(params: &FabroWorkflowVersionCreateParams) -> CollectedWorkflowClosure { + params.validate().unwrap(); + collect_supplied_workflow(params, tempfile::tempdir().unwrap()).unwrap() + } + + #[test] + fn workflow_version_content_matches_existing_collector_and_cleans_staging() { + for input in [ + params("workflow.fabro", &[("workflow.fabro", "digraph W {}")]), + fixture(), + ] { + let source = tempfile::tempdir().unwrap(); + for (path, content) in &input.files { + std::fs::write(source.path().join(path.as_str()), content).unwrap(); + } + let expected = fabro_manifest::collect_workflow_versions( + Path::new(input.entrypoint.as_str()), + source.path(), + ) + .unwrap(); + let staging = tempfile::tempdir().unwrap(); + let path = staging.path().to_owned(); + let actual = collect_supplied_workflow(&input, staging).unwrap(); + assert!(!path.exists()); + assert_eq!(actual.root_id(), expected.root_id()); + assert_eq!( + actual + .versions() + .map(|(_, v)| v.version()) + .collect::>(), + expected + .versions() + .map(|(_, v)| v.version()) + .collect::>() + ); + } + } + + #[test] + fn workflow_version_exact_extensionless_entrypoint_and_child_ignore_selectors() { + let input = params("workflow", &[ + ( + "workflow", + r#"digraph W { child [stack.child_workflow="child"] }"#, + ), + ("child", "digraph Child {}"), + ( + ".fabro/project.toml", + "malformed project config must not be read", + ), + ( + ".fabro/workflows/workflow/workflow.toml", + "misleading named workflow", + ), + ]); + let closure = collect(&input); + let versions = closure.versions().collect::>(); + assert_eq!(versions.len(), 2); + assert_eq!(versions[1].1.version().entrypoint().as_str(), "workflow"); + assert_eq!(versions[0].1.version().entrypoint().as_str(), "child"); + } + + #[test] + fn workflow_version_rejects_missing_and_escaping_references_and_cleans_failure() { + let parent = tempfile::tempdir().unwrap(); + std::fs::write( + parent.path().join("outside.md"), + "host content must never satisfy a reference", + ) + .unwrap(); + std::fs::write(parent.path().join("child.fabro"), "digraph Host {}").unwrap(); + for (index, input) in [ + params("workflow.fabro", &[ + ("workflow.fabro", r#"digraph W { p [prompt="@prompt.md"] }"#), + ("Prompt.md", "wrong case"), + ]), + params("workflow.fabro", &[( + "workflow.fabro", + r#"digraph W { p [prompt="@../outside.md"] }"#, + )]), + params("workflow.fabro", &[( + "workflow.fabro", + r#"digraph W { p [prompt="@sub/../../outside.md"] }"#, + )]), + params("workflow.fabro", &[( + "workflow.fabro", + r#"digraph W { p [prompt="@outside.md"] }"#, + )]), + params("workflow.fabro", &[( + "workflow.fabro", + r#"digraph W { p [output_schema="@../outside.md"] }"#, + )]), + params("workflow.fabro", &[( + "workflow.fabro", + r#"digraph W { p [stack.child_workflow="../child.fabro"] }"#, + )]), + params("workflow.fabro", &[( + "workflow.fabro", + r#"digraph W { p [stack.child_workflow="sub/../../child.fabro"] }"#, + )]), + params("workflow.fabro", &[( + "workflow.fabro", + r#"digraph W { p [stack.child_workflow="missing"] }"#, + )]), + params("workflow.toml", &[( + "workflow.toml", + "_version = 1\n[workflow]\ngraph = \"../child.fabro\"\n", + )]), + params("workflow.fabro", &[( + "workflow.fabro", + "invalid source PRIVATE_CONTENT", + )]), + ] + .into_iter() + .enumerate() + { + let staging = tempfile::tempdir_in(parent.path()).unwrap(); + let path = staging.path().to_owned(); + assert!( + collect_supplied_workflow(&input, staging).is_err(), + "accepted invalid fixture {index}" + ); + assert!(!path.exists()); + } + } + + #[test] + fn workflow_version_registration_preserves_literal_scripts_without_executing() { + let directory = tempfile::tempdir().unwrap(); + let marker = directory.path().join("must-not-exist"); + // Script is literal command text in Fabro, not an @file import. + let graph = format!( + "digraph W {{ command [script=\"touch {}\"] }}", + marker.display() + ); + let input = params("workflow", &[("workflow", &graph)]); + let closure = collect(&input); + let root = closure.versions().last().unwrap().1.version(); + assert_eq!(root.files()[&"workflow".parse().unwrap()], graph); + assert!(!marker.exists()); + } + + #[test] + fn workflow_version_map_order_is_irrelevant_and_reachable_changes_change_ids() { + let input = fixture(); + let first = collect(&input); + let mut reordered = input.clone(); + reordered.files = input.files.into_iter().rev().collect(); + assert_eq!(first.root_id(), collect(&reordered).root_id()); + reordered + .files + .insert("prompt.md".parse().unwrap(), "changed".into()); + let changed = collect(&reordered); + assert_ne!(first.root_id(), changed.root_id()); + assert_eq!( + first.versions().next().unwrap().0, + changed.versions().next().unwrap().0 + ); + reordered + .files + .insert("child.fabro".parse().unwrap(), "digraph Changed {}".into()); + let changed_child = collect(&reordered); + assert_ne!(changed.root_id(), changed_child.root_id()); + assert_ne!( + changed.versions().next().unwrap().0, + changed_child.versions().next().unwrap().0 + ); + } + + #[tokio::test] + async fn workflow_version_uploads_dependencies_first_and_retries_immutable_content() { + let uploads = Arc::new(Mutex::new(Vec::::new())); + let seen = uploads.clone(); + let app = Router::new().route( + "/api/v1/workflow-versions", + post(move |Json(version): Json| { + let seen = seen.clone(); + async move { + let mut seen = seen.lock().unwrap(); + for id in version.workflow_dependencies().values() { + assert!( + seen.iter().any(|prior| prior.id().unwrap() == *id), + "dependency must be registered first" + ); + } + let id = version.id().unwrap(); + seen.push(version); + (StatusCode::CREATED, Json(json!({"workflow_version_id":id}))) + } + }), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let client = + Client::new_no_proxy(&format!("http://{}", listener.local_addr().unwrap())).unwrap(); + let server = tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + let expected = collect(&fixture()).root_id(); + for _ in 0..2 { + let id = ServerWorkflowVersionCreateAdapter + .create_workflow_version(fixture(), &client) + .await + .unwrap(); + assert_eq!(id, expected); + } + assert_eq!(uploads.lock().unwrap().len(), 4); + server.abort(); + assert!(server.await.unwrap_err().is_cancelled()); + } + + #[tokio::test] + async fn workflow_version_invalid_closure_has_no_uploads_or_source_in_errors() { + let server = httpmock::MockServer::start_async().await; + let upload = server + .mock_async(|when, then| { + when.method(httpmock::Method::POST); + then.status(500); + }) + .await; + let client = Client::new_no_proxy(&server.url("")).unwrap(); + let mut invalid = fixture(); + // Child is valid, but the root fails after its dependency is assembled. + invalid.files.insert("workflow.toml".parse().unwrap(), "_version = 1\n[workflow]\ngraph = \"workflow.fabro\"\n[run.goal]\nfile = \"missing.md\"".into()); + let mut oversized = fixture(); + let huge_prompt = "\u{1}".repeat(fabro_types::MAX_WORKFLOW_VERSION_FILE_BYTES - 1); + oversized + .files + .insert("prompt.md".parse().unwrap(), huge_prompt); + for input in [ + invalid, + oversized, + params("workflow", &[( + "workflow", + "PRIVATE_CONTENT invalid source", + )]), + ] { + let error = ServerWorkflowVersionCreateAdapter + .create_workflow_version(input, &client) + .await + .unwrap_err(); + assert!(!format!("{error:#}").contains("PRIVATE_CONTENT")); + } + upload.assert_calls_async(0).await; + } + + #[tokio::test] + async fn workflow_version_root_upload_failure_leaves_child_for_safe_retry() { + let server = httpmock::MockServer::start_async().await; + let closure = collect(&fixture()); + let child = closure.versions().next().unwrap().1.version(); + let root = closure.versions().last().unwrap().1.version(); + let child_upload = server + .mock_async(|when, then| { + when.method(httpmock::Method::POST) + .path("/api/v1/workflow-versions") + .json_body_obj(child); + then.status(201) + .json_body(json!({"workflow_version_id":child.id().unwrap()})); + }) + .await; + let failed_root = server + .mock_async(|when, then| { + when.method(httpmock::Method::POST) + .path("/api/v1/workflow-versions") + .json_body_obj(root); + then.status(400); + }) + .await; + let deletion = server + .mock_async(|when, then| { + when.method(httpmock::Method::DELETE); + then.status(500); + }) + .await; + let client = Client::new_no_proxy(&server.url("")).unwrap(); + assert!( + ServerWorkflowVersionCreateAdapter + .create_workflow_version(fixture(), &client) + .await + .is_err() + ); + child_upload.assert_calls_async(1).await; + failed_root.assert_calls_async(1).await; + failed_root.delete_async().await; + let root_upload = server + .mock_async(|when, then| { + when.method(httpmock::Method::POST) + .path("/api/v1/workflow-versions") + .json_body_obj(root); + then.status(201) + .json_body(json!({"workflow_version_id":root.id().unwrap()})); + }) + .await; + assert_eq!( + ServerWorkflowVersionCreateAdapter + .create_workflow_version(fixture(), &client) + .await + .unwrap(), + closure.root_id() + ); + child_upload.assert_calls_async(2).await; + root_upload.assert_calls_async(1).await; + deletion.assert_calls_async(0).await; + } + + #[tokio::test] + async fn workflow_version_failed_upload_and_wrong_server_id_never_return_success() { + for wrong_id in [false, true] { + let server = httpmock::MockServer::start_async().await; + let closure = collect(&fixture()); + let child = closure.versions().next().unwrap().1.version(); + let upload = server.mock_async(|when, then| { + when.method(httpmock::Method::POST).path("/api/v1/workflow-versions").json_body_obj(child); + if wrong_id { + then.status(201).json_body(json!({"workflow_version_id": WorkflowVersionId::from(fabro_types::BlobHash::new(b"wrong"))})); + } else { then.status(400); } + }).await; + let root = closure.versions().last().unwrap().1.version(); + let root_upload = server + .mock_async(|when, then| { + when.method(httpmock::Method::POST) + .path("/api/v1/workflow-versions") + .json_body_obj(root); + then.status(201) + .json_body(json!({"workflow_version_id":root.id().unwrap()})); + }) + .await; + let client = Client::new_no_proxy(&server.url("")).unwrap(); + assert!( + ServerWorkflowVersionCreateAdapter + .create_workflow_version(fixture(), &client) + .await + .is_err() + ); + upload.assert_calls_async(1).await; + root_upload.assert_calls_async(0).await; + } + } +} diff --git a/lib/components/fabro-manifest/src/lib.rs b/lib/components/fabro-manifest/src/lib.rs index db317abb1..a3f36c3e4 100644 --- a/lib/components/fabro-manifest/src/lib.rs +++ b/lib/components/fabro-manifest/src/lib.rs @@ -37,6 +37,7 @@ pub use crate::local_workflow_package::{ use crate::workflow_bundler::WorkflowBundler; pub use crate::workflow_version_collector::{ CollectedWorkflowClosure, WorkflowVersionCollectError, collect_workflow_versions, + collect_workflow_versions_at_location, }; #[derive(Debug, Default)] diff --git a/lib/components/fabro-manifest/src/workflow_bundler.rs b/lib/components/fabro-manifest/src/workflow_bundler.rs index e01eb629d..4aaa989f5 100644 --- a/lib/components/fabro-manifest/src/workflow_bundler.rs +++ b/lib/components/fabro-manifest/src/workflow_bundler.rs @@ -145,17 +145,27 @@ impl<'a> WorkflowBundler<'a> { /// `~` rejected) before resolution, so the file read matches the manifest /// key. Returns the collected workflow's manifest key. fn collect_workflow_entry(&mut self, workflow: &Path, resolve_from: &Path) -> Result { - let normalized_workflow = if workflow.extension().is_some() && workflow.is_relative() { - normalize_absolute_path(resolve_from, &workflow.to_string_lossy()).ok_or_else(|| { - anyhow!( - "unsupported manifest workflow reference: {}", - workflow.display() - ) - })? + let location = if self.workflow_version_projection { + let exact = normalize_absolute_path(resolve_from, &workflow.to_string_lossy()) + .ok_or_else(|| anyhow!("unsupported workflow reference"))?; + // Check containment before location resolution can read a config. + manifest_path_from_absolute(&exact, self.package_root)?; + WorkflowLocation::from_exact_path(&exact, self.package_root)? } else { - workflow.to_path_buf() + let normalized_workflow = if workflow.extension().is_some() && workflow.is_relative() { + normalize_absolute_path(resolve_from, &workflow.to_string_lossy()).ok_or_else( + || { + anyhow!( + "unsupported manifest workflow reference: {}", + workflow.display() + ) + }, + )? + } else { + workflow.to_path_buf() + }; + WorkflowLocation::resolve(&normalized_workflow, resolve_from)? }; - let location = WorkflowLocation::resolve(&normalized_workflow, resolve_from)?; self.collect_workflow_location(&location) } diff --git a/lib/components/fabro-manifest/src/workflow_version_collector.rs b/lib/components/fabro-manifest/src/workflow_version_collector.rs index a0fb815b6..8928cfd81 100644 --- a/lib/components/fabro-manifest/src/workflow_version_collector.rs +++ b/lib/components/fabro-manifest/src/workflow_version_collector.rs @@ -137,7 +137,9 @@ pub(super) fn canonicalize_location( }) } -pub(super) fn collect_workflow_versions_at_location( +/// Collect an already resolved exact location inside a canonical package root. +/// Every dependency is validated before this returns; no registration occurs. +pub fn collect_workflow_versions_at_location( location: &WorkflowLocation, package_root: &Path, workflow: &Path, @@ -263,6 +265,12 @@ fn workflow_files( for (path, file) in workflow.files { insert_file(&mut files, entrypoint, workflow_path(&path)?, file.content)?; } + fabro_types::validate_workflow_source_paths(files.keys()).map_err(|source| { + WorkflowVersionCollectError::InvalidShape { + entrypoint: entrypoint.clone(), + source, + } + })?; Ok(files) } diff --git a/lib/components/fabro-tool/src/common.rs b/lib/components/fabro-tool/src/common.rs index 159ec426d..f88249fa2 100644 --- a/lib/components/fabro-tool/src/common.rs +++ b/lib/components/fabro-tool/src/common.rs @@ -48,6 +48,13 @@ pub type ToolResult = Result; #[async_trait] pub trait FabroToolBackend: Send + Sync { + async fn create_workflow_version( + &self, + _params: crate::FabroWorkflowVersionCreateParams, + ) -> anyhow::Result { + anyhow::bail!("fabro_workflow_version_create is not available") + } + async fn create_run_from_spec( &self, spec: &crate::ValidatedCreateRunSpec, @@ -170,6 +177,7 @@ pub struct ToolDefinition { pub parameters: Value, } +pub const FABRO_WORKFLOW_VERSION_CREATE_TOOL_NAME: &str = "fabro_workflow_version_create"; pub const FABRO_RUN_CREATE_TOOL_NAME: &str = "fabro_run_create"; pub const FABRO_RUN_SEARCH_TOOL_NAME: &str = "fabro_run_search"; pub const FABRO_RUN_GET_TOOL_NAME: &str = "fabro_run_get"; @@ -180,6 +188,10 @@ pub const FABRO_RUN_PAIR_TOOL_NAME: &str = "fabro_run_pair"; static TOOL_DEFINITIONS: LazyLock> = LazyLock::new(|| { vec![ + tool_definition::( + FABRO_WORKFLOW_VERSION_CREATE_TOOL_NAME, + "Register supplied workflow file contents and all local dependencies as a reusable immutable workflow version ID. Obtain files with shell/read tools first; this does not create or start a run.", + ), tool_definition::( FABRO_RUN_CREATE_TOOL_NAME, "Create one or more Fabro workflow runs, optionally under a parent run, starting them by default.", @@ -323,6 +335,7 @@ mod tests { #[test] fn shared_tool_definitions_include_run_management_catalog() { assert_eq!(shared_tool_names(), vec![ + FABRO_WORKFLOW_VERSION_CREATE_TOOL_NAME, FABRO_RUN_CREATE_TOOL_NAME, FABRO_RUN_SEARCH_TOOL_NAME, FABRO_RUN_GET_TOOL_NAME, @@ -333,6 +346,21 @@ mod tests { ]); } + #[test] + fn workflow_version_create_has_strict_content_schema() { + let definition = tool_definitions() + .iter() + .find(|definition| definition.name == "fabro_workflow_version_create") + .expect("workflow version creation should be in the shared catalog"); + let schema = &definition.parameters; + assert_eq!(schema["additionalProperties"], false); + assert_eq!(schema["properties"].as_object().unwrap().len(), 2); + assert_eq!( + schema["required"], + serde_json::json!(["entrypoint", "files"]) + ); + } + #[test] fn pair_tool_definition_exposes_pair_schema() { let definition = tool_definitions() diff --git a/lib/components/fabro-tool/src/fabro_client.rs b/lib/components/fabro-tool/src/fabro_client.rs index 59b935a85..eb5599c5e 100644 --- a/lib/components/fabro-tool/src/fabro_client.rs +++ b/lib/components/fabro-tool/src/fabro_client.rs @@ -12,9 +12,10 @@ use crate::{FabroToolBackend, RunManifestBuilder, ToolError}; #[derive(Clone)] pub struct ClientBackend { - client: Arc<::fabro_client::Client>, + client: Arc<::fabro_client::Client>, manifest_builder: Option>, - run_scope: Option, + run_scope: Option, + workflow_version_create_adapter: Option>, } impl ClientBackend { @@ -24,6 +25,7 @@ impl ClientBackend { client, manifest_builder: None, run_scope: None, + workflow_version_create_adapter: None, } } @@ -33,6 +35,15 @@ impl ClientBackend { self } + #[must_use] + pub fn with_workflow_version_create_adapter( + mut self, + adapter: Arc, + ) -> Self { + self.workflow_version_create_adapter = Some(adapter); + self + } + /// Restrict this backend to a single run. /// /// Ask Fabro sessions use this with a same-run worker token so accidental @@ -55,6 +66,21 @@ impl ClientBackend { #[async_trait] impl FabroToolBackend for ClientBackend { + async fn create_workflow_version( + &self, + params: crate::FabroWorkflowVersionCreateParams, + ) -> anyhow::Result { + anyhow::ensure!( + self.run_scope.is_none(), + "workflow version creation is outside this tool session's run scope" + ); + let adapter = self + .workflow_version_create_adapter + .as_ref() + .ok_or_else(|| anyhow::anyhow!("fabro_workflow_version_create is not available"))?; + adapter.create_workflow_version(params, &self.client).await + } + async fn create_run_from_spec( &self, spec: &crate::ValidatedCreateRunSpec, diff --git a/lib/components/fabro-tool/src/lib.rs b/lib/components/fabro-tool/src/lib.rs index e854c26e3..64ac87f27 100644 --- a/lib/components/fabro-tool/src/lib.rs +++ b/lib/components/fabro-tool/src/lib.rs @@ -14,12 +14,13 @@ mod interact; mod manifest; mod pair; mod search; +mod workflow_version; pub use common::{ FABRO_RUN_CREATE_TOOL_NAME, FABRO_RUN_EVENTS_TOOL_NAME, FABRO_RUN_GATHER_TOOL_NAME, FABRO_RUN_GET_TOOL_NAME, FABRO_RUN_INTERACT_TOOL_NAME, FABRO_RUN_PAIR_TOOL_NAME, - FABRO_RUN_SEARCH_TOOL_NAME, FabroToolBackend, RunManifestBuilder, RunSummaryResult, - ToolDefinition, ToolError, ToolResult, tool_definitions, + FABRO_RUN_SEARCH_TOOL_NAME, FABRO_WORKFLOW_VERSION_CREATE_TOOL_NAME, FabroToolBackend, + RunManifestBuilder, RunSummaryResult, ToolDefinition, ToolError, ToolResult, tool_definitions, }; pub use create::{ CreateRunOptions, CreateRunSpec, CreateRunSpecInput, CreateRunsResult, CreatedRunResult, @@ -47,3 +48,7 @@ pub use search::{ FabroRunSearchParams, SearchRunSummaryResult, SearchRunsResult, ValidatedSearchRuns, search_runs, search_runs_text, }; +pub use workflow_version::{ + FabroWorkflowVersionCreateParams, WorkflowVersionCreateAdapter, create_workflow_version, + workflow_version_create_text, +}; diff --git a/lib/components/fabro-tool/src/workflow_version.rs b/lib/components/fabro-tool/src/workflow_version.rs new file mode 100644 index 000000000..11567df8a --- /dev/null +++ b/lib/components/fabro-tool/src/workflow_version.rs @@ -0,0 +1,234 @@ +use std::collections::BTreeMap; +use std::sync::Arc; + +use async_trait::async_trait; +use fabro_api::types::CreateWorkflowVersionResponse; +use fabro_types::{ + MAX_WORKFLOW_VERSION_BYTES, MAX_WORKFLOW_VERSION_FILE_BYTES, MAX_WORKFLOW_VERSION_FILES, + WorkflowPath, WorkflowVersionId, +}; +use schemars::JsonSchema; +use serde::de::{Error as _, MapAccess, Visitor}; +use serde::{Deserialize, Serialize}; + +use crate::{FabroToolBackend, ToolError, ToolResult}; + +/// Caller-supplied source content, before packaging resolves workflow +/// dependencies. +#[derive(Clone, Deserialize, Serialize, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct FabroWorkflowVersionCreateParams { + /// Exact package-relative key of the graph or workflow configuration file. + #[schemars(with = "String")] + pub entrypoint: WorkflowPath, + /// All local dependencies, keyed by package-relative path. Values are text + /// contents. + #[serde(deserialize_with = "deserialize_files")] + #[schemars(with = "BTreeMap")] + pub files: BTreeMap, +} + +fn deserialize_files<'de, D: serde::Deserializer<'de>>( + deserializer: D, +) -> Result, D::Error> { + struct FilesVisitor; + impl<'de> Visitor<'de> for FilesVisitor { + type Value = BTreeMap; + fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str("workflow files with unique path keys and text contents") + } + fn visit_map>(self, mut map: A) -> Result { + let mut files = BTreeMap::new(); + while let Some((path, content)) = map.next_entry()? { + if files.insert(path, content).is_some() { + return Err(A::Error::custom("duplicate workflow file key")); + } + } + Ok(files) + } + } + deserializer.deserialize_map(FilesVisitor) +} + +impl FabroWorkflowVersionCreateParams { + /// Validate the complete supplied tree before any filesystem writes. + pub fn validate(&self) -> ToolResult<()> { + if !self.files.contains_key(&self.entrypoint) { + return Err(ToolError::message( + "entrypoint must be an exact supplied file key", + )); + } + if self.files.len() > MAX_WORKFLOW_VERSION_FILES { + return Err(ToolError::message("workflow source exceeds 512 files")); + } + let mut total = 0; + for content in self.files.values() { + if content.len() > MAX_WORKFLOW_VERSION_FILE_BYTES { + return Err(ToolError::message("workflow source file exceeds 512 KiB")); + } + total += content.len(); + } + if total > MAX_WORKFLOW_VERSION_BYTES { + return Err(ToolError::message("workflow source exceeds 2 MiB")); + } + fabro_types::validate_workflow_source_paths(self.files.keys()) + .map_err(|_| ToolError::message("workflow source paths collide"))?; + Ok(()) + } +} + +/// Application seam for packaging and registering content without a dependency +/// cycle. Implementations must validate before staging, confine reads to +/// supplied files, validate the entire closure before uploading, and register +/// dependencies first. +#[async_trait] +pub trait WorkflowVersionCreateAdapter: Send + Sync { + async fn create_workflow_version( + &self, + params: FabroWorkflowVersionCreateParams, + client: &fabro_client::Client, + ) -> anyhow::Result; +} + +pub async fn create_workflow_version( + backend: Arc, + params: FabroWorkflowVersionCreateParams, +) -> ToolResult { + params.validate()?; + let workflow_version_id = backend + .create_workflow_version(params) + .await + .map_err(|err| ToolError::from_anyhow(&err))?; + Ok(CreateWorkflowVersionResponse { + workflow_version_id, + }) +} + +#[must_use] +pub fn workflow_version_create_text(result: &CreateWorkflowVersionResponse) -> String { + serde_json::to_string(result).expect("workflow version response should serialize") +} + +#[cfg(test)] +mod tests { + use serde_json::json; + + use super::*; + use crate::fabro_client::ClientBackend; + + #[test] + fn workflow_version_request_rejects_unknown_fields_and_invalid_paths() { + let valid = json!({"entrypoint": "workflow", "files": {"workflow": "digraph W {}"}}); + let params: FabroWorkflowVersionCreateParams = + serde_json::from_value(valid.clone()).unwrap(); + params.validate().unwrap(); + assert!( + serde_json::from_str::( + r#"{"entrypoint":"workflow","files":{"workflow":"a","workflow":"b"}}"# + ) + .is_err() + ); + for field in [ + "cwd", + "url", + "workflow", + "environment", + "parent_id", + "workflow_dependencies", + ] { + let mut value = valid.clone(); + value[field] = json!("unexpected"); + assert!(serde_json::from_value::(value).is_err()); + } + for path in [ + "../workflow", + "/workflow", + "a/../workflow", + "a//b", + "a\\b", + "~/workflow", + "", + ] { + for value in [ + json!({"entrypoint":path,"files":{"workflow":"x"}}), + json!({"entrypoint":"workflow","files":{path:"x"}}), + ] { + assert!(serde_json::from_value::(value).is_err()); + } + } + } + + #[test] + fn workflow_version_source_enforces_presence_collisions_and_budgets() { + for files in [ + json!({}), + json!({"A":"x","a":"y"}), + json!({"A":"x","a/b.md":"y"}), + json!({"a":"x","A/b.md":"y"}), + ] { + let mut files = files.as_object().unwrap().clone(); + files.insert("workflow".into(), json!("digraph W {}")); + let mut params: FabroWorkflowVersionCreateParams = + serde_json::from_value(json!({"entrypoint":"workflow","files":files})).unwrap(); + if params.files.len() == 1 { + params.entrypoint = "missing".parse().unwrap(); + } + assert!(params.validate().is_err()); + } + let mut params = FabroWorkflowVersionCreateParams { + entrypoint: "workflow".parse().unwrap(), + files: BTreeMap::from([( + "workflow".parse().unwrap(), + "x".repeat(MAX_WORKFLOW_VERSION_FILE_BYTES + 1), + )]), + }; + assert!(params.validate().is_err()); + params.files = (0..MAX_WORKFLOW_VERSION_FILES) + .map(|i| (format!("file{i}").parse().unwrap(), String::new())) + .collect(); + params + .files + .insert(params.entrypoint.clone(), String::new()); + assert!(params.validate().is_err()); + params.files = (0..5) + .map(|i| { + ( + format!("file{i}").parse().unwrap(), + "x".repeat(MAX_WORKFLOW_VERSION_FILE_BYTES), + ) + }) + .collect(); + params + .files + .insert(params.entrypoint.clone(), String::new()); + assert!(params.validate().is_err()); + } + + #[tokio::test] + async fn workflow_version_same_run_backend_denies_before_adapter() { + let client = fabro_client::Client::new_no_proxy("http://127.0.0.1:1").unwrap(); + let backend = ClientBackend::new(Arc::new(client)) + .with_workflow_version_create_adapter(Arc::new(UnreachableAdapter)) + .with_run_scope("01KRBZW4DW0000000000000002".parse().unwrap()); + let params = serde_json::from_value( + json!({"entrypoint":"workflow","files":{"workflow":"digraph W {}"}}), + ) + .unwrap(); + let error = create_workflow_version(Arc::new(backend), params) + .await + .unwrap_err(); + assert!(error.as_str().contains("run scope")); + } + + struct UnreachableAdapter; + #[async_trait] + impl WorkflowVersionCreateAdapter for UnreachableAdapter { + async fn create_workflow_version( + &self, + _: FabroWorkflowVersionCreateParams, + _: &fabro_client::Client, + ) -> anyhow::Result { + panic!("scoped backend must not invoke the adapter") + } + } +} diff --git a/lib/foundation/fabro-config/src/project.rs b/lib/foundation/fabro-config/src/project.rs index 4fc58e61e..c33f3fc79 100644 --- a/lib/foundation/fabro-config/src/project.rs +++ b/lib/foundation/fabro-config/src/project.rs @@ -54,6 +54,21 @@ impl WorkflowLocation { } } + /// Resolve an exact file path without workflow-name or ambient config + /// lookup. Relative paths are interpreted against the supplied + /// directory only. + pub fn from_exact_path(path: &Path, directory: &Path) -> Result { + let path = directory.join(path); + if path + .extension() + .is_some_and(|extension| extension == "toml") + { + Self::from_toml(path) + } else { + Ok(Self::from_graph(path)) + } + } + fn from_toml(toml_path: PathBuf) -> Result { let cfg = match run::load_run_config(&toml_path) { Ok(cfg) => cfg, diff --git a/lib/foundation/fabro-types/Cargo.toml b/lib/foundation/fabro-types/Cargo.toml index 2e777317b..005fe228f 100644 --- a/lib/foundation/fabro-types/Cargo.toml +++ b/lib/foundation/fabro-types/Cargo.toml @@ -33,6 +33,8 @@ strum.workspace = true thiserror.workspace = true toml.workspace = true ulid.workspace = true +unicase = "2" +unicode-normalization = "0.1" url.workspace = true [dev-dependencies] diff --git a/lib/foundation/fabro-types/src/lib.rs b/lib/foundation/fabro-types/src/lib.rs index b4b26fe74..e678a8a1f 100644 --- a/lib/foundation/fabro-types/src/lib.rs +++ b/lib/foundation/fabro-types/src/lib.rs @@ -198,5 +198,6 @@ pub use workflow_path::{ pub use workflow_version::{ MAX_WORKFLOW_VERSION_BYTES, MAX_WORKFLOW_VERSION_DEPENDENCIES, MAX_WORKFLOW_VERSION_FILE_BYTES, MAX_WORKFLOW_VERSION_FILES, WorkflowVersion, WorkflowVersionShapeError, + validate_workflow_source_paths, }; pub use workflow_version_id::{WorkflowVersionId, WorkflowVersionIdParseError}; diff --git a/lib/foundation/fabro-types/src/workflow_version.rs b/lib/foundation/fabro-types/src/workflow_version.rs index 125544aa3..996a26444 100644 --- a/lib/foundation/fabro-types/src/workflow_version.rs +++ b/lib/foundation/fabro-types/src/workflow_version.rs @@ -5,6 +5,8 @@ use std::marker::PhantomData; use serde::de::{Error as _, MapAccess, Visitor}; use serde::{Deserialize, Deserializer, Serialize}; use thiserror::Error; +use unicase::UniCase; +use unicode_normalization::UnicodeNormalization as _; use crate::{BlobHash, WorkflowPath, WorkflowVersionId}; @@ -149,31 +151,52 @@ impl WorkflowVersion { } fn validate_path_collisions(&self) -> Result<(), WorkflowVersionShapeError> { - // Keys are unique within each map, so equality can only collide - // across files and workflow dependencies. - let mut by_text = - HashMap::with_capacity(self.files.len() + self.workflow_dependencies.len()); - for path in self.files.keys().chain(self.workflow_dependencies.keys()) { - if let Some(existing) = by_text.insert(path.as_str(), path) { + validate_path_collisions( + self.files.keys().chain(self.workflow_dependencies.keys()), + false, + ) + } +} + +/// Reject file and directory aliases before materializing a portable source +/// tree, including Unicode case folding and normalization. Canonical versions +/// themselves retain their exact, case-sensitive +/// semantics. +pub fn validate_workflow_source_paths<'a>( + paths: impl IntoIterator, +) -> Result<(), WorkflowVersionShapeError> { + validate_path_collisions(paths, true) +} + +fn validate_path_collisions<'a>( + paths: impl IntoIterator, + case_insensitive: bool, +) -> Result<(), WorkflowVersionShapeError> { + let mut by_text = HashMap::new(); + for path in paths { + let text = if case_insensitive { + UniCase::new(path.as_str()).to_folded_case().nfc().collect() + } else { + path.as_str().to_owned() + }; + if let Some(existing) = by_text.insert(text, path) { + return Err(WorkflowVersionShapeError::PathCollision { + first: existing.clone(), + second: path.clone(), + }); + } + } + for (text, path) in &by_text { + for (index, _) in text.match_indices('/') { + if let Some(ancestor) = by_text.get(&text[..index]) { return Err(WorkflowVersionShapeError::PathCollision { - first: existing.clone(), - second: path.clone(), + first: (*ancestor).clone(), + second: (*path).clone(), }); } } - for path in self.files.keys().chain(self.workflow_dependencies.keys()) { - let text = path.as_str(); - for (index, _) in text.match_indices('/') { - if let Some(ancestor) = by_text.get(&text[..index]) { - return Err(WorkflowVersionShapeError::PathCollision { - first: (*ancestor).clone(), - second: path.clone(), - }); - } - } - } - Ok(()) } + Ok(()) } impl<'de> Deserialize<'de> for WorkflowVersion { @@ -509,3 +532,37 @@ mod tests { assert!(serde_json::from_str::(duplicate).is_err()); } } + +#[cfg(test)] +mod source_path_tests { + use super::*; + + #[test] + fn workflow_source_collisions_are_portable_in_both_orders() { + for pair in [ + ["A", "a"], + ["A", "a/b.md"], + ["a", "A/b.md"], + ["é", "É/b"], + ["ΟΣ", "οσ/b"], + ["é", "e\u{301}/b"], + ["Straße", "STRASSE/b"], + ] { + let paths = pair.map(|path| WorkflowPath::new(path).unwrap()); + assert!(validate_workflow_source_paths(paths.iter()).is_err()); + assert!(validate_workflow_source_paths(paths.iter().rev()).is_err()); + } + let version = WorkflowVersion::new( + WorkflowPath::new("A").unwrap(), + BTreeMap::from([ + (WorkflowPath::new("A").unwrap(), "x".into()), + (WorkflowPath::new("a").unwrap(), "y".into()), + ]), + BTreeMap::new(), + ); + assert!( + version.is_ok(), + "canonical versions keep exact path semantics" + ); + } +} diff --git a/lib/packages/fabro-api-client/src/api/workflow-versions-api.ts b/lib/packages/fabro-api-client/src/api/workflow-versions-api.ts index 4285c1849..14173fac5 100644 --- a/lib/packages/fabro-api-client/src/api/workflow-versions-api.ts +++ b/lib/packages/fabro-api-client/src/api/workflow-versions-api.ts @@ -33,7 +33,7 @@ import type { WorkflowVersion } from '../models'; export const WorkflowVersionsApiAxiosParamCreator = function (configuration?: Configuration) { return { /** - * Validates and stores an immutable workflow package in content-addressed storage. Repeating the same canonical content returns the same identifier. + * Validates and stores an immutable workflow package in content-addressed storage. Repeating the same canonical content returns the same identifier. Requires an authenticated user or a worker token with the `agent:run_tools` capability. Ordinary worker tokens cannot register versions. Registration creates no run and starts no execution. * @summary Create Workflow Version * @param {WorkflowVersion} workflowVersion * @param {*} [options] Override http request option. @@ -83,7 +83,7 @@ export const WorkflowVersionsApiFp = function(configuration?: Configuration) { const localVarAxiosParamCreator = WorkflowVersionsApiAxiosParamCreator(configuration) return { /** - * Validates and stores an immutable workflow package in content-addressed storage. Repeating the same canonical content returns the same identifier. + * Validates and stores an immutable workflow package in content-addressed storage. Repeating the same canonical content returns the same identifier. Requires an authenticated user or a worker token with the `agent:run_tools` capability. Ordinary worker tokens cannot register versions. Registration creates no run and starts no execution. * @summary Create Workflow Version * @param {WorkflowVersion} workflowVersion * @param {*} [options] Override http request option. @@ -105,7 +105,7 @@ export const WorkflowVersionsApiFactory = function (configuration?: Configuratio const localVarFp = WorkflowVersionsApiFp(configuration) return { /** - * Validates and stores an immutable workflow package in content-addressed storage. Repeating the same canonical content returns the same identifier. + * Validates and stores an immutable workflow package in content-addressed storage. Repeating the same canonical content returns the same identifier. Requires an authenticated user or a worker token with the `agent:run_tools` capability. Ordinary worker tokens cannot register versions. Registration creates no run and starts no execution. * @summary Create Workflow Version * @param {WorkflowVersion} workflowVersion * @param {*} [options] Override http request option. @@ -122,7 +122,7 @@ export const WorkflowVersionsApiFactory = function (configuration?: Configuratio */ export class WorkflowVersionsApi extends BaseAPI { /** - * Validates and stores an immutable workflow package in content-addressed storage. Repeating the same canonical content returns the same identifier. + * Validates and stores an immutable workflow package in content-addressed storage. Repeating the same canonical content returns the same identifier. Requires an authenticated user or a worker token with the `agent:run_tools` capability. Ordinary worker tokens cannot register versions. Registration creates no run and starts no execution. * @summary Create Workflow Version * @param {WorkflowVersion} workflowVersion * @param {*} [options] Override http request option.