From c67c60eebab3c54b63fc9d6912c8223dc000cb04 Mon Sep 17 00:00:00 2001 From: Scott Werner Date: Mon, 31 Aug 2026 17:58:55 -0400 Subject: [PATCH] Create run tools from immutable workflow versions --- docs/public/agents/mcp.mdx | 48 +- docs/public/execution/child-runs.mdx | 42 + lib/apps/fabro-cli/src/commands/run/runner.rs | 76 +- lib/apps/fabro-mcp-server/src/lib.rs | 1 - .../fabro-mcp-server/src/manifest_builder.rs | 19 - lib/apps/fabro-mcp-server/src/server.rs | 46 +- lib/apps/fabro-server/src/lib.rs | 2 +- lib/apps/fabro-server/src/run_tool_create.rs | 737 ++++++++++++++++++ .../fabro-server/src/run_tool_manifest.rs | 218 ------ .../src/server/handler/sessions.rs | 7 +- .../src/server/handler/workflow_versions.rs | 74 ++ lib/components/fabro-tool/src/common.rs | 84 +- lib/components/fabro-tool/src/create.rs | 635 +++++++++++++-- lib/components/fabro-tool/src/fabro_client.rs | 69 +- lib/components/fabro-tool/src/interact.rs | 3 +- lib/components/fabro-tool/src/lib.rs | 16 +- lib/components/fabro-workflow/src/services.rs | 7 +- 17 files changed, 1687 insertions(+), 397 deletions(-) delete mode 100644 lib/apps/fabro-mcp-server/src/manifest_builder.rs create mode 100644 lib/apps/fabro-server/src/run_tool_create.rs delete mode 100644 lib/apps/fabro-server/src/run_tool_manifest.rs diff --git a/docs/public/agents/mcp.mdx b/docs/public/agents/mcp.mdx index e769a4bf3..47e592f39 100644 --- a/docs/public/agents/mcp.mdx +++ b/docs/public/agents/mcp.mdx @@ -90,7 +90,7 @@ 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: +For a simple create call, `fabro_run_create` accepts a workflow selector string. Selectors are compatible only when the MCP process and Fabro operation share the native filesystem: ```json { "runs": ["sleeper"] } @@ -113,7 +113,51 @@ Use the object form when you need create options: } ``` -Use `goal` for inline goal text or `goal_file` to read the run goal from a file. They are mutually exclusive. Relative `goal_file` paths resolve from the run's `cwd`, or from the MCP server working directory when `cwd` is omitted. +For agent-authored content or callers without a shared filesystem, send an inline package by value. `entrypoint` and every `files` key are portable paths inside that package: + +```json +{ + "runs": [ + { + "workflow": { + "kind": "inline", + "entrypoint": "workflows/review/workflow.fabro", + "files": { + "workflows/review/workflow.fabro": "digraph Review { start [shape=Mdiamond] exit [shape=Msquare] start -> exit }" + } + }, + "target": { + "kind": "git", + "repo": "acme/widgets", + "branch": "feature/checkout" + } + } + ] +} +``` + +You can also reuse an exact immutable workflow version without uploading content again: + +```json +{ + "runs": [ + { + "workflow": { + "kind": "stored", + "workflow_version_id": "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef" + }, + "target": { "kind": "none" }, + "start": false + } + ] +} +``` + +Workflow source and run target are independent. Object-form calls can provide an explicit Git, `none`, or folder target. A workflow agent inherits its parent run's canonical target when `target` is omitted; standalone MCP instead derives an attached GitHub checkout when it can do so truthfully and otherwise requires an explicit target. Docker and Daytona agents must send inline files or a stored version ID—never send a sandbox path as a selector, because the run worker cannot use that path to read the sandbox's files. + +Use `goal` for inline goal text or `goal_file` to read the run goal from a file. They are mutually exclusive. Relative `goal_file` paths resolve from the run's `cwd`, or from the MCP server working directory when `cwd` is omitted. Like selectors, `goal_file` requires a native/shared filesystem; Docker and Daytona agents must send `goal` text by value. + +Creation and execution remain separate operations. `fabro_run_create` first creates a durable run, then requests one start by default. Set `start: false` to leave the new run submitted without requesting execution. Run summaries returned by the MCP server include parent metadata. Use `parent_id` on `fabro_run_create` to create a child run, `parent_id` on `fabro_run_search` to list direct children, and the `link_parent` or `unlink_parent` actions on `fabro_run_interact` to change an existing run's parent. See [Child Runs](/execution/child-runs) for the orchestration model. diff --git a/docs/public/execution/child-runs.mdx b/docs/public/execution/child-runs.mdx index 1b85039c1..a01110e31 100644 --- a/docs/public/execution/child-runs.mdx +++ b/docs/public/execution/child-runs.mdx @@ -96,6 +96,48 @@ The parent can create several children in one call: } ``` +Selector strings and selector paths are available only when the parent agent runs in a genuinely shared Local filesystem context. Docker and Daytona agents must send the workflow bytes inline or reuse an exact stored workflow version; a path inside their sandbox is not a usable selector on the worker host. + +Send agent-authored workflow content with a strict inline source: + +```json +{ + "runs": [ + { + "workflow": { + "kind": "inline", + "entrypoint": "child/workflow.fabro", + "files": { + "child/workflow.fabro": "digraph Child { start [shape=Mdiamond] exit [shape=Msquare] start -> exit }" + } + }, + "goal": "Run the workflow content supplied by this parent." + } + ] +} +``` + +Reuse content already registered with Fabro by supplying its exact immutable ID: + +```json +{ + "runs": [ + { + "workflow": { + "kind": "stored", + "workflow_version_id": "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef" + }, + "target": { "kind": "none" }, + "start": false + } + ] +} +``` + +Workflow content and workspace target are separate choices. If `target` is omitted, a child inherits the parent's full canonical Git, `none`, or folder target. An explicit Git, `none`, or folder target overrides that inheritance while the current run remains the forced parent. + +`goal_file` is also a shared-filesystem feature. Local agents can read it relative to the operation `cwd`; Docker and Daytona agents must send the resolved `goal` text by value. + By default, `fabro_run_create` requests start for each created run. Set `"start": false` when the parent should create the child now and start it later. ## Start and approval diff --git a/lib/apps/fabro-cli/src/commands/run/runner.rs b/lib/apps/fabro-cli/src/commands/run/runner.rs index 506f2d83a..12ec98cf0 100644 --- a/lib/apps/fabro-cli/src/commands/run/runner.rs +++ b/lib/apps/fabro-cli/src/commands/run/runner.rs @@ -5,7 +5,6 @@ use std::time::Duration; use anyhow::{Context, Result, anyhow}; use async_trait::async_trait; -use fabro_api::types::RunManifest; use fabro_client::ServerTarget; use fabro_config::user::active_settings_path; use fabro_config::{ServerSettingsBuilder, Storage}; @@ -16,7 +15,7 @@ use fabro_interview::{ WorkerControlMessage, }; use fabro_manifest::SuppliedWorkflowVersionPackager; -use fabro_server::run_tool_manifest; +use fabro_server::run_tool_create::ServerRunCreateAdapter; use fabro_store::{EventEnvelope, RunProjection, RunProjectionReducer}; use fabro_tool::fabro_client::ClientBackend; use fabro_types::settings::run::{RunMode, RunNamespace}; @@ -102,6 +101,8 @@ pub(crate) async fn execute( worker_token, client.clone_for_reuse(), run_id, + run_spec.settings.run.environment.provider, + run_spec.target.clone(), run_spec.source_directory.as_deref(), &run_dir, ) @@ -230,34 +231,34 @@ fn build_fabro_run_tool_services( worker_token: &str, client: fabro_client::Client, current_run_id: RunId, + provider: fabro_types::settings::run::EnvironmentProvider, + inherited_target: Option, source_directory: Option<&str>, run_dir: &Path, ) -> Option { if worker_token.trim().is_empty() { return None; } - let backend = ClientBackend::new(Arc::new(client)) - .with_manifest_builder(Arc::new(WorkerRunManifestBuilder)) - .with_workflow_version_packager(Arc::new(SuppliedWorkflowVersionPackager)); + let settings_path = active_settings_path(None); + let user_workflows_root = settings_path + .parent() + .map(|parent| parent.join("workflows")); + let backend = ClientBackend::new(Arc::new(client)).with_run_create_adapter(Arc::new( + worker_run_create_adapter(provider, inherited_target, user_workflows_root), + )).with_workflow_version_packager(Arc::new(SuppliedWorkflowVersionPackager)); Some(FabroRunToolServices { backend: Arc::new(backend), current_run_id, base_cwd: source_directory.map_or_else(|| run_dir.to_path_buf(), PathBuf::from), - user_settings_path: active_settings_path(None), }) } -struct WorkerRunManifestBuilder; - -impl fabro_tool::RunManifestBuilder for WorkerRunManifestBuilder { - fn build_run_manifest( - &self, - spec: &fabro_tool::ValidatedCreateRunSpec, - cwd: &Path, - user_settings_path: &Path, - ) -> fabro_tool::ToolResult { - run_tool_manifest::build_run_tool_manifest(spec, cwd, user_settings_path) - } +fn worker_run_create_adapter( + provider: fabro_types::settings::run::EnvironmentProvider, + inherited_target: Option, + user_workflows_root: Option, +) -> ServerRunCreateAdapter { + ServerRunCreateAdapter::worker(provider, inherited_target, user_workflows_root) } /// Load the worker's secret vault from the run's storage root. @@ -1196,6 +1197,7 @@ fn install_signal_handlers( reason = "This test module prefers explicit type paths over extra imports." )] mod tests { + use std::path::Path; use std::sync::Arc; use std::time::Duration; @@ -1211,7 +1213,7 @@ mod tests { }; use fabro_types::{ AuthMethod, EventBody, FailureCategory, FailureDetail, FailureReason, IdpIdentity, - Principal, QuestionType, RunFailure, SuccessReason, fixtures, + Principal, QuestionType, RunFailure, RunTarget, SuccessReason, WorkflowVersionId, fixtures, }; use fabro_vault::{SecretType, Vault}; use fabro_workflow::event::RunEventSink; @@ -1269,6 +1271,44 @@ mod tests { )); } + #[tokio::test] + async fn fabro_run_create_worker_adapter_uses_runtime_provider_and_canonical_target() { + use fabro_tool::{FabroRunCreateParams, RunCreateAdapter, ValidatedCreateRuns}; + use fabro_types::settings::run::EnvironmentProvider; + + let temp = tempfile::tempdir().unwrap(); + let workflow_version_id: WorkflowVersionId = + fabro_types::BlobHash::new(b"stored workflow").into(); + let inherited = RunTarget::None {}; + let params: FabroRunCreateParams = serde_json::from_value(serde_json::json!({ + "runs": [{ + "workflow": { + "kind": "stored", + "workflow_version_id": workflow_version_id + } + }] + })) + .unwrap(); + let spec = ValidatedCreateRuns::try_from(params) + .unwrap() + .runs + .remove(0); + let adapter = super::worker_run_create_adapter( + EnvironmentProvider::Docker, + Some(inherited.clone()), + Some(temp.path().join("workflows")), + ); + let client = fabro_client::Client::new_no_proxy("http://127.0.0.1:9").unwrap(); + + let prepared = adapter + .prepare(&client, &spec, Path::new("/host/that-must-not-be-read")) + .await + .unwrap(); + + assert_eq!(prepared.workflow_version_id, workflow_version_id); + assert_eq!(prepared.target, inherited); + } + fn worker_token_with_claims(claims: &serde_json::Value) -> String { jsonwebtoken::encode( &jsonwebtoken::Header::new(jsonwebtoken::Algorithm::HS256), diff --git a/lib/apps/fabro-mcp-server/src/lib.rs b/lib/apps/fabro-mcp-server/src/lib.rs index d345ec5c1..ff555322c 100644 --- a/lib/apps/fabro-mcp-server/src/lib.rs +++ b/lib/apps/fabro-mcp-server/src/lib.rs @@ -1,6 +1,5 @@ mod config; mod executable_monitor; -mod manifest_builder; mod server; use std::future::Future; diff --git a/lib/apps/fabro-mcp-server/src/manifest_builder.rs b/lib/apps/fabro-mcp-server/src/manifest_builder.rs deleted file mode 100644 index b09be76de..000000000 --- a/lib/apps/fabro-mcp-server/src/manifest_builder.rs +++ /dev/null @@ -1,19 +0,0 @@ -use std::path::Path; - -use fabro_api::types; -use fabro_server::run_tool_manifest; -use fabro_tool::{RunManifestBuilder, ToolResult, ValidatedCreateRunSpec}; - -#[derive(Default)] -pub(crate) struct McpRunManifestBuilder; - -impl RunManifestBuilder for McpRunManifestBuilder { - fn build_run_manifest( - &self, - spec: &ValidatedCreateRunSpec, - cwd: &Path, - user_settings_path: &Path, - ) -> ToolResult { - run_tool_manifest::build_run_tool_manifest(spec, cwd, user_settings_path) - } -} diff --git a/lib/apps/fabro-mcp-server/src/server.rs b/lib/apps/fabro-mcp-server/src/server.rs index fba1fc5ac..670fda34b 100644 --- a/lib/apps/fabro-mcp-server/src/server.rs +++ b/lib/apps/fabro-mcp-server/src/server.rs @@ -18,7 +18,6 @@ use tokio::time; use tracing::warn; use crate::executable_monitor::ExecutableMonitor; -use crate::manifest_builder::McpRunManifestBuilder; use crate::{FabroMcpServerSettings, SERVER_NAME}; #[derive(Clone)] @@ -115,7 +114,7 @@ impl FabroMcpServer { #[tool( name = "fabro_run_create", - description = "Create one or more Fabro workflow runs, optionally under a parent run, starting them by default." + description = "Create one or more Fabro workflow runs from a native selector, inline files, or an exact stored workflow version, optionally under a parent run and starting them by default." )] async fn fabro_run_create( &self, @@ -129,7 +128,7 @@ impl FabroMcpServer { Ok(backend) => backend, Err(err) => return Ok(error_result(&err)), }; - match run_tools::create_runs(backend, &self.cwd, &self.settings.config_path, params).await { + match run_tools::create_runs(backend, &self.cwd, params).await { Ok(result) => success_result(&result, run_tools::create_runs_text(&result)), Err(err) => Ok(error_result(&err)), } @@ -273,12 +272,17 @@ impl FabroMcpServer { (self.settings.client_factory)() .await .map(|client| { + let user_workflows_root = self + .settings + .config_path + .parent() + .map(|parent| parent.join("workflows")); Arc::new( - ClientBackend::new(Arc::new(client)) - .with_manifest_builder(Arc::new(McpRunManifestBuilder)) - .with_workflow_version_packager(Arc::new( - SuppliedWorkflowVersionPackager, - )), + ClientBackend::new(Arc::new(client)).with_run_create_adapter(Arc::new( + fabro_server::run_tool_create::ServerRunCreateAdapter::standalone( + user_workflows_root, + ), + )).with_workflow_version_packager(Arc::new(SuppliedWorkflowVersionPackager)), ) as Arc }) .map_err(|err| run_tools::ToolError::from_anyhow(&err)) @@ -421,7 +425,7 @@ mod tests { } #[test] - fn fabro_run_create_tool_advertises_string_and_object_run_specs() { + fn fabro_run_create_tool_advertises_complete_workflow_source_and_target_grammar() { let settings = FabroMcpServerSettings { cwd: PathBuf::from("."), config_path: PathBuf::from("fabro.toml"), @@ -451,6 +455,7 @@ mod tests { .unwrap_or_else(|| { panic!("runs items should include object create spec variant: {schema}") }); + assert_eq!(object_variant["additionalProperties"], false); assert!( object_variant.pointer("/properties/workflow").is_some(), "object create spec should expose workflow property: {schema}" @@ -466,5 +471,28 @@ mod tests { .is_some_and(|required| required.iter().any(|name| name == "workflow")), "object create spec should require workflow: {schema}" ); + let workflow_variants = object_variant + .pointer("/properties/workflow/anyOf") + .and_then(Value::as_array) + .expect("workflow should advertise selector, inline, and stored sources"); + assert!( + workflow_variants + .iter() + .any(|variant| variant["type"] == "string") + ); + for kind in ["inline", "stored"] { + assert!(workflow_variants.iter().any(|variant| { + variant.pointer("/properties/kind/const") == Some(&Value::from(kind)) + })); + } + let target_variants = object_variant + .pointer("/properties/target/anyOf") + .and_then(Value::as_array) + .expect("target should advertise the canonical target union"); + for kind in ["git", "none", "folder"] { + assert!(target_variants.iter().any(|variant| { + variant.pointer("/properties/kind/const") == Some(&Value::from(kind)) + })); + } } } diff --git a/lib/apps/fabro-server/src/lib.rs b/lib/apps/fabro-server/src/lib.rs index 98f7a2c77..14d9addc1 100644 --- a/lib/apps/fabro-server/src/lib.rs +++ b/lib/apps/fabro-server/src/lib.rs @@ -41,7 +41,7 @@ mod run_intent; mod run_manifest; mod run_selector; mod run_title_generation; -pub mod run_tool_manifest; +pub mod run_tool_create; pub mod security_headers; pub mod serve; pub mod server; diff --git a/lib/apps/fabro-server/src/run_tool_create.rs b/lib/apps/fabro-server/src/run_tool_create.rs new file mode 100644 index 000000000..d2146f6b4 --- /dev/null +++ b/lib/apps/fabro-server/src/run_tool_create.rs @@ -0,0 +1,737 @@ +use std::collections::HashMap; +use std::path::{Path, PathBuf}; + +use anyhow::{Context, Result, bail}; +use async_trait::async_trait; +use fabro_config::RunLayer; +use fabro_manifest::{ + CollectedWorkflowClosure, ResolvedLocalWorkflowPackage, RunOverrideInput, + collect_workflow_versions, observe_git_run_target, resolve_local_workflow_package, +}; +use fabro_tool::{ + PreparedRunCreate, RunCreateAdapter, ValidatedCreateRunSpec, ValidatedCreateRunWorkflowSource, +}; +use fabro_types::settings::run::EnvironmentProvider; +use fabro_types::{DirtyStatus, RunTarget}; +use tokio::io::AsyncWriteExt; + +use crate::manifest_validation; + +#[derive(Clone, Debug)] +pub struct ServerRunCreateAdapter { + mode: RunCreateMode, +} + +#[derive(Clone, Debug)] +enum RunCreateMode { + Standalone { + user_workflows_root: Option, + }, + Worker { + provider: EnvironmentProvider, + inherited_target: Option, + user_workflows_root: Option, + }, +} + +impl ServerRunCreateAdapter { + #[must_use] + pub fn standalone(user_workflows_root: Option) -> Self { + Self { + mode: RunCreateMode::Standalone { + user_workflows_root, + }, + } + } + + #[must_use] + pub fn worker( + provider: EnvironmentProvider, + inherited_target: Option, + user_workflows_root: Option, + ) -> Self { + Self { + mode: RunCreateMode::Worker { + provider, + inherited_target, + user_workflows_root, + }, + } + } + + fn has_shared_filesystem(&self) -> bool { + match self.mode { + RunCreateMode::Standalone { .. } => true, + RunCreateMode::Worker { provider, .. } => provider.is_local(), + } + } + + fn user_workflows_root(&self) -> Option<&Path> { + match &self.mode { + RunCreateMode::Standalone { + user_workflows_root, + } + | RunCreateMode::Worker { + user_workflows_root, + .. + } => user_workflows_root.as_deref(), + } + } + + async fn resolve_goal( + &self, + spec: &ValidatedCreateRunSpec, + cwd: &Path, + ) -> Result> { + if let Some(goal) = &spec.goal { + return Ok(Some(goal.clone())); + } + let Some(goal_file) = &spec.goal_file else { + return Ok(None); + }; + if !self.has_shared_filesystem() { + bail!( + "goal_file requires a shared Local filesystem; Docker and Daytona callers must send goal text by value" + ); + } + let path = cwd.join(goal_file); + tokio::fs::read_to_string(&path) + .await + .with_context(|| format!("failed to read goal file {}", path.display())) + .map(Some) + } + + fn resolve_target(&self, spec: &ValidatedCreateRunSpec, cwd: &Path) -> Result { + if let Some(target) = &spec.target { + return Ok(ResolvedTarget { + target: target.clone(), + warnings: Vec::new(), + }); + } + + match &self.mode { + RunCreateMode::Worker { + inherited_target: Some(target), + .. + } => Ok(ResolvedTarget { + target: target.clone(), + warnings: Vec::new(), + }), + RunCreateMode::Worker { + inherited_target: None, + .. + } => bail!( + "the parent run has no canonical target; send an explicit target for this child run" + ), + RunCreateMode::Standalone { .. } => { + let observation = observe_git_run_target(cwd, None).ok_or_else(|| { + anyhow::anyhow!( + "target is required outside an attached local GitHub checkout with a branch" + ) + })?; + let target = observation.run_target.ok_or_else(|| { + anyhow::anyhow!( + "target is required because the local checkout cannot be represented as a GitHub run target" + ) + })?; + let mut warnings = Vec::new(); + if observation.legacy_git_context.dirty == DirtyStatus::Dirty { + warnings.push( + "the local checkout has uncommitted changes; those changes are excluded from the run target" + .to_string(), + ); + } + if observation + .legacy_git_context + .sha + .as_deref() + .is_some_and(|sha| !sha.is_empty()) + && target.sha.is_none() + { + warnings.push( + "the local HEAD commit is not fetchable and is not pinned; the remote branch will be selected" + .to_string(), + ); + } + Ok(ResolvedTarget { + target: RunTarget::Git(target), + warnings, + }) + } + } + } + + fn collect_selector(&self, selector: &str, cwd: &Path) -> Result { + if !self.has_shared_filesystem() { + bail!( + "workflow selectors require a shared Local filesystem; send inline files or an exact stored workflow version ID from Docker or Daytona" + ); + } + resolve_local_workflow_package(Path::new(selector), cwd, self.user_workflows_root()) + .map(LocalWorkflowSource::Selector) + .map_err(anyhow::Error::new) + } +} + +#[async_trait] +impl RunCreateAdapter for ServerRunCreateAdapter { + async fn prepare( + &self, + client: &fabro_client::Client, + spec: &ValidatedCreateRunSpec, + cwd: &Path, + ) -> Result { + if !self.has_shared_filesystem() + && matches!(spec.workflow, ValidatedCreateRunWorkflowSource::Selector(_)) + { + bail!( + "workflow selectors require a shared Local filesystem; send inline files or an exact stored workflow version ID from Docker or Daytona" + ); + } + if !self.has_shared_filesystem() && spec.goal_file.is_some() { + bail!( + "goal_file requires a shared Local filesystem; Docker and Daytona callers must send goal text by value" + ); + } + + let goal = self.resolve_goal(spec, cwd).await?; + if let ValidatedCreateRunWorkflowSource::Stored { + workflow_version_id, + } = spec.workflow + { + let resolved_target = self.resolve_target(spec, cwd)?; + return Ok(PreparedRunCreate { + workflow_version_id, + target: resolved_target.target, + goal, + warnings: resolved_target.warnings, + }); + } + + let local_source = match &spec.workflow { + ValidatedCreateRunWorkflowSource::Selector(selector) => { + self.collect_selector(selector, cwd)? + } + ValidatedCreateRunWorkflowSource::Inline(source) => { + LocalWorkflowSource::inline(source).await? + } + ValidatedCreateRunWorkflowSource::Stored { .. } => unreachable!(), + }; + validate_local_source(local_source.closure(), spec, goal.as_deref())?; + let resolved_target = self.resolve_target(spec, cwd)?; + let closure = local_source.closure(); + let versions = closure + .versions() + .map(|(_, version)| version.version()) + .collect::>(); + client.register_workflow_versions(versions).await?; + + Ok(PreparedRunCreate { + workflow_version_id: closure.root_id(), + target: resolved_target.target, + goal, + warnings: resolved_target.warnings, + }) + } +} + +struct ResolvedTarget { + target: RunTarget, + warnings: Vec, +} + +enum LocalWorkflowSource { + Selector(ResolvedLocalWorkflowPackage), + Inline { + closure: CollectedWorkflowClosure, + _root: tempfile::TempDir, + }, +} + +impl LocalWorkflowSource { + async fn inline(source: &fabro_tool::InlineWorkflowSource) -> Result { + let root = tempfile::tempdir().context("failed to create private inline workflow root")?; + for (path, content) in &source.files { + let destination = root.path().join(path.as_str()); + if let Some(parent) = destination.parent() { + tokio::fs::create_dir_all(parent).await.with_context(|| { + format!( + "failed to create inline workflow directory {}", + parent.display() + ) + })?; + } + let mut file = tokio::fs::OpenOptions::new() + .write(true) + .create_new(true) + .open(&destination) + .await + .with_context(|| { + format!( + "failed to create inline workflow file {}", + destination.display() + ) + })?; + file.write_all(content.as_bytes()).await.with_context(|| { + format!( + "failed to write inline workflow file {}", + destination.display() + ) + })?; + } + let closure = collect_workflow_versions(Path::new(source.entrypoint.as_str()), root.path()) + .map_err(anyhow::Error::new)?; + Ok(Self::Inline { + closure, + _root: root, + }) + } + + fn closure(&self) -> &CollectedWorkflowClosure { + match self { + Self::Selector(package) => package.closure(), + Self::Inline { closure, .. } => closure, + } + } +} + +fn validate_local_source( + closure: &CollectedWorkflowClosure, + spec: &ValidatedCreateRunSpec, + goal: Option<&str>, +) -> Result<()> { + let run_overrides = run_tool_run_overrides(spec, goal); + let inputs = spec + .inputs + .iter() + .map(|(key, value)| (key.clone(), value.toml().clone())) + .collect::>(); + let response = + manifest_validation::validate_collected_workflow(closure, run_overrides.as_ref(), &inputs)?; + if !response.ok { + bail!("workflow validation failed"); + } + Ok(()) +} + +fn run_tool_run_overrides(spec: &ValidatedCreateRunSpec, goal: Option<&str>) -> Option { + fabro_manifest::build_sparse_run_overrides(RunOverrideInput { + goal, + model: spec.model.as_deref(), + provider: spec.provider.as_deref(), + environment: spec.environment.as_deref(), + preserve_sandbox: spec.preserve_sandbox, + dry_run: spec.dry_run, + auto_approve: spec.auto_approve, + labels: spec.labels.clone(), + }) +} + +#[cfg(test)] +mod tests { + use std::collections::BTreeMap; + use std::process::Command; + use std::sync::{Arc, Mutex}; + + use fabro_tool::{FabroRunCreateParams, ValidatedCreateRuns}; + use fabro_types::{GitRunTarget, WorkflowVersion, WorkflowVersionId}; + use httpmock::Method::POST; + use httpmock::{HttpMockRequest, HttpMockResponse, MockServer}; + use serde_json::json; + + use super::*; + + fn validated_spec(value: serde_json::Value) -> ValidatedCreateRunSpec { + let params: FabroRunCreateParams = serde_json::from_value(json!({ "runs": [value] })) + .expect("create input should deserialize"); + ValidatedCreateRuns::try_from(params) + .expect("create input should validate") + .runs + .remove(0) + } + + fn no_proxy_client(base_url: &str) -> fabro_client::Client { + fabro_client::Client::new_no_proxy(base_url).expect("test client should build") + } + + fn run_git(cwd: &Path, args: &[&str]) { + let output = Command::new("git") + .args(args) + .current_dir(cwd) + .output() + .expect("git command should run"); + assert!( + output.status.success(), + "git {args:?} failed: {}", + String::from_utf8_lossy(&output.stderr) + ); + } + + async fn dynamic_version_registration_mock<'a>( + server: &'a MockServer, + registered: Arc>>, + ) -> httpmock::Mock<'a> { + server + .mock_async(move |when, then| { + when.method(POST).path("/api/v1/workflow-versions"); + then.respond_with(move |request: &HttpMockRequest| { + let version: WorkflowVersion = serde_json::from_str(&request.body_string()) + .expect("registration request should contain a workflow version"); + let id = version.id().expect("registered version should have an ID"); + registered.lock().unwrap().push(version); + HttpMockResponse::builder() + .status(201) + .header("content-type", "application/json") + .body(json!({ "workflow_version_id": id }).to_string()) + .build() + }); + }) + .await + } + + #[tokio::test] + async fn workflow_version_inline_create_registers_exact_dependency_first_bytes() { + let server = MockServer::start_async().await; + let registered = Arc::new(Mutex::new(Vec::new())); + let registration = + dynamic_version_registration_mock(&server, Arc::clone(®istered)).await; + let client = no_proxy_client(&server.url("")); + let spec = validated_spec(json!({ + "workflow": { + "kind": "inline", + "entrypoint": "root/workflow.fabro", + "files": { + "root/workflow.fabro": r#"digraph Root { + start [shape=Mdiamond] + prompt [prompt="@prompt.md"] + child [stack.child_workflow="../child/workflow.fabro"] + exit [shape=Msquare] + start -> prompt -> child -> exit + }"#, + "root/prompt.md": "runtime-authored root bytes", + "child/workflow.fabro": r#"digraph Child { + start [shape=Mdiamond] + task [prompt="@support.md"] + exit [shape=Msquare] + start -> task -> exit + }"#, + "child/support.md": "runtime-authored child bytes" + } + }, + "target": { "kind": "none" }, + "start": false + })); + let adapter = ServerRunCreateAdapter::worker(EnvironmentProvider::Docker, None, None); + + let prepared = adapter + .prepare(&client, &spec, Path::new("/host/that-must-not-be-read")) + .await + .expect("inline workflow should prepare"); + + registration.assert_calls_async(2).await; + let registered = registered.lock().unwrap(); + assert_eq!(registered.len(), 2); + let child_id = registered[0].id().unwrap(); + let root_id = registered[1].id().unwrap(); + assert_eq!(prepared.workflow_version_id, root_id); + assert_eq!(prepared.target, RunTarget::None {}); + assert_eq!( + registered[1].workflow_dependencies(), + &BTreeMap::from([( + fabro_types::WorkflowPath::new("child/workflow.fabro").unwrap(), + child_id, + )]) + ); + assert_eq!( + registered[0] + .files() + .get(&fabro_types::WorkflowPath::new("child/support.md").unwrap()) + .map(String::as_str), + Some("runtime-authored child bytes") + ); + assert_eq!( + registered[1] + .files() + .get(&fabro_types::WorkflowPath::new("root/prompt.md").unwrap()) + .map(String::as_str), + Some("runtime-authored root bytes") + ); + } + + #[tokio::test] + async fn workflow_version_stored_create_skips_registration_and_inherits_exact_worker_target() { + let client = no_proxy_client("http://127.0.0.1:9"); + let workflow_version_id: WorkflowVersionId = fabro_types::BlobHash::new(b"stored").into(); + let inherited = RunTarget::Git(GitRunTarget { + repo: "fabro-sh/fabro".to_string(), + branch: "main".to_string(), + tag: Some("v1.0.0".to_string()), + sha: Some("0123456789abcdef0123456789abcdef01234567".to_string()), + }); + let spec = validated_spec(json!({ + "workflow": { + "kind": "stored", + "workflow_version_id": workflow_version_id + } + })); + let adapter = ServerRunCreateAdapter::worker( + EnvironmentProvider::Docker, + Some(inherited.clone()), + None, + ); + + let prepared = adapter + .prepare(&client, &spec, Path::new("/ignored")) + .await + .unwrap(); + + assert_eq!(prepared.workflow_version_id, workflow_version_id); + assert_eq!(prepared.target, inherited); + } + + #[tokio::test] + async fn workflow_version_selector_uses_resolved_package_root_not_operation_cwd() { + let temp = tempfile::tempdir().unwrap(); + let operation_cwd = temp.path().join("nested/operation"); + let workflow_dir = temp.path().join(".fabro/workflows/demo"); + std::fs::create_dir_all(&operation_cwd).unwrap(); + std::fs::create_dir_all(&workflow_dir).unwrap(); + std::fs::write(temp.path().join(".fabro/project.toml"), "_version = 1\n").unwrap(); + std::fs::write( + workflow_dir.join("workflow.toml"), + "_version = 1\n[workflow]\ngraph = \"workflow.fabro\"\n", + ) + .unwrap(); + std::fs::write( + workflow_dir.join("workflow.fabro"), + "digraph Demo { start [shape=Mdiamond] exit [shape=Msquare] start -> exit }", + ) + .unwrap(); + let server = MockServer::start_async().await; + let registered = Arc::new(Mutex::new(Vec::new())); + let registration = + dynamic_version_registration_mock(&server, Arc::clone(®istered)).await; + let client = no_proxy_client(&server.url("")); + let spec = validated_spec(json!({ + "workflow": "demo", + "target": { "kind": "none" } + })); + let adapter = ServerRunCreateAdapter::worker(EnvironmentProvider::Local, None, None); + + let prepared = adapter + .prepare(&client, &spec, &operation_cwd) + .await + .unwrap(); + + registration.assert_calls_async(1).await; + let registered = registered.lock().unwrap(); + assert_eq!(prepared.workflow_version_id, registered[0].id().unwrap()); + assert_eq!( + registered[0].entrypoint().as_str(), + ".fabro/workflows/demo/workflow.fabro" + ); + } + + #[tokio::test] + async fn workflow_version_worker_capabilities_gate_selector_and_goal_file_before_reads() { + let temp = tempfile::tempdir().unwrap(); + let workflow = temp.path().join("same-name.fabro"); + std::fs::write( + &workflow, + "digraph HostCopy { start [shape=Mdiamond] exit [shape=Msquare] start -> exit }", + ) + .unwrap(); + std::fs::write( + temp.path().join("goal.md"), + "host goal that must not be read", + ) + .unwrap(); + let client = no_proxy_client("http://127.0.0.1:9"); + let adapter = ServerRunCreateAdapter::worker(EnvironmentProvider::Daytona, None, None); + + let selector = validated_spec(json!({ + "workflow": "same-name.fabro", + "target": { "kind": "none" } + })); + let selector_error = adapter + .prepare(&client, &selector, temp.path()) + .await + .expect_err("Daytona worker must reject host selectors"); + assert!( + selector_error + .to_string() + .contains("inline files or an exact stored") + ); + + let workflow_version_id: WorkflowVersionId = fabro_types::BlobHash::new(b"stored").into(); + let goal_file = validated_spec(json!({ + "workflow": { + "kind": "stored", + "workflow_version_id": workflow_version_id + }, + "target": { "kind": "none" }, + "goal_file": "goal.md" + })); + let goal_error = adapter + .prepare(&client, &goal_file, temp.path()) + .await + .expect_err("Daytona worker must reject host goal files"); + assert!(goal_error.to_string().contains("send goal text by value")); + } + + #[tokio::test] + async fn workflow_version_local_content_is_validated_before_registration() { + let client = no_proxy_client("http://127.0.0.1:9"); + let adapter = ServerRunCreateAdapter::worker(EnvironmentProvider::Docker, None, None); + + let invalid_graph = validated_spec(json!({ + "workflow": { + "kind": "inline", + "entrypoint": "workflow.fabro", + "files": { "workflow.fabro": "this is not a graph" } + }, + "target": { "kind": "none" } + })); + adapter + .prepare(&client, &invalid_graph, Path::new("/ignored")) + .await + .expect_err("invalid graph should fail before registration"); + + let undefined_input = validated_spec(json!({ + "workflow": { + "kind": "inline", + "entrypoint": "workflow.fabro", + "files": { + "workflow.fabro": r#"digraph W { + start [shape=Mdiamond] + task [prompt="@prompt.md"] + exit [shape=Msquare] + start -> task -> exit + }"#, + "prompt.md": "Hello {{ inputs.owner }}" + } + }, + "target": { "kind": "none" } + })); + let error = adapter + .prepare(&client, &undefined_input, Path::new("/ignored")) + .await + .expect_err("undefined input should fail before registration"); + assert!(error.to_string().contains("workflow validation failed")); + } + + #[tokio::test] + async fn workflow_version_target_failure_precedes_registration() { + let client = no_proxy_client("http://127.0.0.1:9"); + let adapter = ServerRunCreateAdapter::worker(EnvironmentProvider::Docker, None, None); + let spec = validated_spec(json!({ + "workflow": { + "kind": "inline", + "entrypoint": "workflow.fabro", + "files": { + "workflow.fabro": "digraph W { start [shape=Mdiamond] exit [shape=Msquare] start -> exit }" + } + } + })); + + let error = adapter + .prepare(&client, &spec, Path::new("/ignored")) + .await + .expect_err("missing inherited target should fail before registration"); + + assert!( + error + .to_string() + .contains("parent run has no canonical target") + ); + } + + #[tokio::test] + async fn workflow_version_shared_goal_file_and_explicit_target_are_preserved() { + let temp = tempfile::tempdir().unwrap(); + std::fs::write(temp.path().join("goal.md"), "goal from shared filesystem").unwrap(); + let client = no_proxy_client("http://127.0.0.1:9"); + let workflow_version_id: WorkflowVersionId = fabro_types::BlobHash::new(b"stored").into(); + let spec = validated_spec(json!({ + "workflow": { + "kind": "stored", + "workflow_version_id": workflow_version_id + }, + "target": { "kind": "none" }, + "goal_file": "goal.md" + })); + let inherited = RunTarget::Folder { + path: "/parent/workspace".to_string(), + }; + let adapter = + ServerRunCreateAdapter::worker(EnvironmentProvider::Local, Some(inherited), None); + + let prepared = adapter.prepare(&client, &spec, temp.path()).await.unwrap(); + + assert_eq!(prepared.target, RunTarget::None {}); + assert_eq!( + prepared.goal.as_deref(), + Some("goal from shared filesystem") + ); + } + + #[tokio::test] + async fn workflow_version_standalone_git_fallback_reports_excluded_local_bytes() { + let temp = tempfile::tempdir().unwrap(); + let workspace = temp.path().join("workspace"); + std::fs::create_dir(&workspace).unwrap(); + run_git(&workspace, &[ + "init", + "--quiet", + "--initial-branch", + "feature", + ]); + run_git(&workspace, &["config", "user.name", "Fabro Test"]); + run_git(&workspace, &["config", "user.email", "fabro@example.com"]); + std::fs::write(workspace.join("tracked.txt"), "committed").unwrap(); + run_git(&workspace, &["add", "tracked.txt"]); + run_git(&workspace, &["commit", "--quiet", "-m", "initial"]); + run_git(&workspace, &[ + "remote", + "add", + "origin", + "https://github.com/acme/widgets.git", + ]); + let missing = format!("file://{}/missing.git", temp.path().display()); + run_git(&workspace, &[ + "remote", "set-url", "--push", "origin", &missing, + ]); + std::fs::write(workspace.join("dirty.txt"), "uncommitted").unwrap(); + + let workflow_version_id: WorkflowVersionId = fabro_types::BlobHash::new(b"stored").into(); + let spec = validated_spec(json!({ + "workflow": { + "kind": "stored", + "workflow_version_id": workflow_version_id + } + })); + let client = no_proxy_client("http://127.0.0.1:9"); + let adapter = ServerRunCreateAdapter::standalone(None); + + let prepared = adapter.prepare(&client, &spec, &workspace).await.unwrap(); + + let RunTarget::Git(target) = prepared.target else { + panic!("standalone attached Git checkout should derive a Git target"); + }; + assert_eq!(target.repo, "acme/widgets"); + assert_eq!(target.branch, "feature"); + assert_eq!(target.sha, None); + assert!( + prepared + .warnings + .iter() + .any(|warning| warning.contains("uncommitted changes")) + ); + assert!( + prepared + .warnings + .iter() + .any(|warning| warning.contains("not fetchable") && warning.contains("not pinned")) + ); + } +} diff --git a/lib/apps/fabro-server/src/run_tool_manifest.rs b/lib/apps/fabro-server/src/run_tool_manifest.rs deleted file mode 100644 index a27f92e05..000000000 --- a/lib/apps/fabro-server/src/run_tool_manifest.rs +++ /dev/null @@ -1,218 +0,0 @@ -use std::path::{Path, PathBuf}; - -use fabro_api::types; -use fabro_config::{CliLayer, RunGoalLayer, RunLayer}; -use fabro_manifest::{ManifestBuildInput, RunOverrideInput}; -use fabro_tool::{ToolError, ToolResult, ValidatedCreateRunSpec}; -use fabro_types::settings::interp::InterpString; - -use crate::manifest_validation; - -/// Build and validate a run manifest for the `fabro_run_create` tool. -/// -/// Validation is structural. The caller is a client — an MCP server or a run -/// worker — whose catalog is its own, not the server's, so judging model and -/// provider availability here would reject workflows the server can run. -pub fn build_run_tool_manifest( - spec: &ValidatedCreateRunSpec, - cwd: &Path, - user_settings_path: &Path, -) -> ToolResult { - let built = fabro_manifest::build_run_manifest(ManifestBuildInput { - workflow: PathBuf::from(&spec.workflow), - cwd: cwd.to_path_buf(), - run_overrides: run_tool_run_overrides(spec), - cli_overrides: Some(CliLayer::default()), - input_overrides: spec.inputs.clone(), - args: run_tool_manifest_args(spec), - environment_defaults: fabro_environment::seeded_catalog_layer(), - user_settings_path: Some(user_settings_path.to_path_buf()), - }) - .map_err(|err| ToolError::from_anyhow(&err))?; - - let mut validation = - manifest_validation::validate_manifest(&RunLayer::default(), &built.manifest) - .map_err(|err| ToolError::from_anyhow(&err))?; - manifest_validation::promote_template_undefined_variables_to_errors(&mut validation); - if !validation.ok { - return Err(ToolError::message("workflow manifest validation failed")); - } - - Ok(built.manifest) -} - -pub fn run_tool_manifest_args(spec: &ValidatedCreateRunSpec) -> Option { - let mut input = spec - .inputs - .iter() - .map(|(key, value)| format!("{key}={value}")) - .collect::>(); - input.sort(); - let mut label = spec - .labels - .iter() - .map(|(key, value)| format!("{key}={value}")) - .collect::>(); - label.sort(); - - let payload = types::ManifestArgs { - auto_approve: spec.auto_approve.filter(|value| *value), - dry_run: spec.dry_run.filter(|value| *value), - input, - label, - model: spec.model.clone(), - preserve_sandbox: spec.preserve_sandbox.filter(|value| *value), - provider: spec.provider.clone(), - environment: spec.environment.clone(), - verbose: None, - }; - (!fabro_manifest::manifest_args_is_empty(&payload)).then_some(payload) -} - -pub fn run_tool_run_overrides(spec: &ValidatedCreateRunSpec) -> Option { - let mut run = fabro_manifest::build_run_overrides(RunOverrideInput { - goal: spec.goal.as_deref(), - model: spec.model.as_deref(), - provider: spec.provider.as_deref(), - environment: spec.environment.as_deref(), - preserve_sandbox: spec.preserve_sandbox, - dry_run: spec.dry_run, - auto_approve: spec.auto_approve, - labels: spec.labels.clone(), - }); - if let Some(goal_file) = spec.goal_file.as_ref() { - run.goal = Some(RunGoalLayer::File { - file: InterpString::parse(&goal_file.to_string_lossy()), - }); - } - (run.goal.is_some() - || !run.metadata.is_empty() - || run.model.is_some() - || run.environment.is_some() - || run.execution.is_some()) - .then_some(run) -} - -#[cfg(test)] -mod tests { - use std::collections::HashMap; - - use fabro_tool::CreateRunSpec; - use serde_json::json; - - use super::*; - - fn create_run_spec(workflow: &str) -> ValidatedCreateRunSpec { - ValidatedCreateRunSpec::try_from(CreateRunSpec { - workflow: workflow.to_string(), - parent_id: None, - cwd: None, - goal: None, - goal_file: None, - inputs: HashMap::new(), - labels: HashMap::new(), - model: None, - provider: None, - environment: None, - dry_run: None, - auto_approve: None, - preserve_sandbox: None, - start: None, - }) - .expect("create spec should validate") - } - - /// The tool runs on a client, whose catalog is not the server's, so a - /// server-owned model must reach the server rather than fail here. - #[expect( - clippy::disallowed_methods, - reason = "sync test writes one workflow fixture before building the manifest" - )] - #[test] - fn server_owned_provider_is_not_rejected_by_tool_manifest_validation() { - let dir = tempfile::tempdir().expect("temp dir should be created"); - let workflow = dir.path().join("server-model.fabro"); - std::fs::write( - &workflow, - r#"digraph ServerModel { - graph [goal="Use a server-owned model"] - start [shape=Mdiamond] - work [prompt="Do work", model="private-model", provider="server-only"] - exit [shape=Msquare] - start -> work -> exit - }"#, - ) - .expect("workflow fixture should be written"); - - let manifest = build_run_tool_manifest( - &create_run_spec(&workflow.to_string_lossy()), - dir.path(), - &dir.path().join("settings.toml"), - ) - .expect("tool validation should leave provider availability to the server"); - - let encoded = serde_json::to_string(&manifest).expect("manifest should serialize"); - assert!( - encoded.contains("server-only"), - "the authored provider should survive into the manifest: {encoded}" - ); - } - - #[test] - fn manifest_args_preserve_input_provenance() { - let spec = ValidatedCreateRunSpec::try_from(CreateRunSpec { - workflow: "simple".to_string(), - parent_id: None, - cwd: None, - goal: None, - goal_file: None, - inputs: HashMap::from([ - ("count".to_string(), json!(3).into()), - ("decision".to_string(), json!("approve").into()), - ]), - labels: HashMap::new(), - model: None, - provider: None, - environment: None, - dry_run: None, - auto_approve: None, - preserve_sandbox: None, - start: None, - }) - .expect("create spec should validate"); - let args = run_tool_manifest_args(&spec).expect("input args should be present"); - - assert_eq!(args.input, vec![r"count=3", r#"decision="approve""#]); - } - - #[expect( - clippy::disallowed_methods, - reason = "test asserts the raw template source" - )] - #[test] - fn run_overrides_preserve_goal_file_as_file_goal() { - let spec = ValidatedCreateRunSpec::try_from(CreateRunSpec { - workflow: "implement-plan".to_string(), - parent_id: None, - cwd: None, - goal: None, - goal_file: Some(PathBuf::from("plans/ship-it.md")), - inputs: HashMap::new(), - labels: HashMap::new(), - model: None, - provider: None, - environment: None, - dry_run: None, - auto_approve: None, - preserve_sandbox: None, - start: None, - }) - .expect("create spec with goal_file should validate"); - - let run = run_tool_run_overrides(&spec).expect("goal_file should produce run overrides"); - let Some(fabro_config::RunGoalLayer::File { file }) = run.goal else { - panic!("goal_file should become a file goal override"); - }; - assert_eq!(file.as_source(), "plans/ship-it.md"); - } -} diff --git a/lib/apps/fabro-server/src/server/handler/sessions.rs b/lib/apps/fabro-server/src/server/handler/sessions.rs index 3f63337b5..1e7c1f67c 100644 --- a/lib/apps/fabro-server/src/server/handler/sessions.rs +++ b/lib/apps/fabro-server/src/server/handler/sessions.rs @@ -749,10 +749,9 @@ async fn build_agent( .map_err(AskFabroBuildError::Agent)?; let backend = ClientBackend::new(Arc::new(api_client)).with_run_scope(run_id); let services = FabroRunToolServices { - backend: Arc::new(backend), - current_run_id: run_id, - base_cwd: PathBuf::new(), - user_settings_path: PathBuf::new(), + backend: Arc::new(backend), + current_run_id: run_id, + base_cwd: PathBuf::new(), }; let run_tools = register_named_fabro_run_tools(&services, ASK_FABRO_RUN_TOOL_NAMES); let selector = format!("{provider_id}/{model}"); 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 cd8daafc7..e8f8b85a4 100644 --- a/lib/apps/fabro-server/src/server/handler/workflow_versions.rs +++ b/lib/apps/fabro-server/src/server/handler/workflow_versions.rs @@ -120,6 +120,7 @@ mod tests { }; use crate::server; use crate::test_support::{self, TestAppStateBuilder}; + use crate::worker_token::{WorkerScopeSet, issue_worker_token, issue_worker_token_with_scopes}; const GRAPH: &str = "digraph W { start [shape=Mdiamond] exit [shape=Msquare] start -> exit }"; @@ -161,6 +162,79 @@ mod tests { assert_eq!(response.status(), StatusCode::UNAUTHORIZED); } + #[tokio::test] + async fn workflow_version_registration_authorization_is_narrow() { + let state = TestAppStateBuilder::new().build(); + let app = server::build_router(Arc::clone(&state), test_support::test_auth_mode()); + let run_id = fabro_types::RunId::new(); + let scoped = issue_worker_token_with_scopes( + state.worker_token_keys(), + &run_id, + WorkerScopeSet::run_worker_with_agent_run_tools(), + ) + .unwrap(); + let ordinary = issue_worker_token(state.worker_token_keys(), &run_id).unwrap(); + + let authorized = app + .clone() + .oneshot( + Request::builder() + .method(Method::POST) + .uri("/api/v1/workflow-versions") + .header(header::CONTENT_TYPE, "application/json") + .header(header::AUTHORIZATION, format!("Bearer {scoped}")) + .body(Body::from(serde_json::to_vec(&version(GRAPH)).unwrap())) + .unwrap(), + ) + .await + .unwrap(); + fabro_test::expect_axum_status( + authorized, + StatusCode::CREATED, + "scoped worker POST /api/v1/workflow-versions", + ) + .await; + + let forbidden = app + .clone() + .oneshot( + Request::builder() + .method(Method::POST) + .uri("/api/v1/workflow-versions") + .header(header::CONTENT_TYPE, "application/json") + .header(header::AUTHORIZATION, format!("Bearer {ordinary}")) + .body(Body::from(serde_json::to_vec(&version(GRAPH)).unwrap())) + .unwrap(), + ) + .await + .unwrap(); + fabro_test::expect_axum_status( + forbidden, + StatusCode::FORBIDDEN, + "ordinary worker POST /api/v1/workflow-versions", + ) + .await; + + let malformed = app + .oneshot( + Request::builder() + .method(Method::POST) + .uri("/api/v1/workflow-versions") + .header(header::CONTENT_TYPE, "application/json") + .header(header::AUTHORIZATION, "Bearer not-a-worker-token") + .body(Body::from(serde_json::to_vec(&version(GRAPH)).unwrap())) + .unwrap(), + ) + .await + .unwrap(); + fabro_test::expect_axum_status( + malformed, + StatusCode::UNAUTHORIZED, + "malformed worker POST /api/v1/workflow-versions", + ) + .await; + } + #[tokio::test] async fn valid_and_equivalent_requests_return_the_same_id() { let state = TestAppStateBuilder::new().build(); diff --git a/lib/components/fabro-tool/src/common.rs b/lib/components/fabro-tool/src/common.rs index f154759b8..399606dbe 100644 --- a/lib/components/fabro-tool/src/common.rs +++ b/lib/components/fabro-tool/src/common.rs @@ -7,7 +7,7 @@ use chrono::{DateTime, NaiveDate, Utc}; use fabro_api::types; use fabro_types::{ PairId, PairMessageRecord, PairMessageRequest, PairRecord, PairTranscriptResponse, Run, RunId, - RunPairStatusResponse, StageId, + RunPairStatusResponse, RunTarget, StageId, WorkflowVersionId, }; use fabro_util::exit::{self, ExitClass}; use schemars::JsonSchema; @@ -46,6 +46,35 @@ impl std::error::Error for ToolError {} pub type ToolResult = Result; +#[derive(Debug)] +pub struct PreparedRunCreate { + pub workflow_version_id: WorkflowVersionId, + pub target: RunTarget, + pub goal: Option, + pub warnings: Vec, +} + +#[derive(Debug)] +pub struct CreateRunSubmission { + pub run_id: RunId, + pub warnings: Vec, +} + +/// Trusted producer-local preparation for one tool-created run. +/// +/// Implementations acquire permitted workflow bytes and goal files, validate +/// local content, resolve the independent target, and register immutable +/// workflow versions before returning intent-ready fields. +#[async_trait] +pub trait RunCreateAdapter: Send + Sync { + async fn prepare( + &self, + client: &fabro_client::Client, + spec: &crate::ValidatedCreateRunSpec, + cwd: &Path, + ) -> anyhow::Result; +} + #[async_trait] pub trait FabroToolBackend: Send + Sync { async fn create_workflow_version( @@ -59,9 +88,8 @@ pub trait FabroToolBackend: Send + Sync { &self, spec: &crate::ValidatedCreateRunSpec, cwd: &Path, - user_settings_path: &Path, parent_id: Option, - ) -> anyhow::Result; + ) -> anyhow::Result; async fn resolve_run(&self, selector: &str) -> anyhow::Result; async fn retrieve_run(&self, run_id: &RunId) -> anyhow::Result; @@ -149,15 +177,6 @@ pub(crate) fn workflow_version_tool_unavailable_error() -> anyhow::Error { .into() } -pub trait RunManifestBuilder: Send + Sync { - fn build_run_manifest( - &self, - spec: &crate::ValidatedCreateRunSpec, - cwd: &Path, - user_settings_path: &Path, - ) -> ToolResult; -} - #[derive(Debug, Serialize, JsonSchema)] pub struct RunSummaryResult { pub run_id: String, @@ -201,7 +220,7 @@ static TOOL_DEFINITIONS: LazyLock> = LazyLock::new(|| { ), tool_definition::( FABRO_RUN_CREATE_TOOL_NAME, - "Create one or more Fabro workflow runs, optionally under a parent run, starting them by default.", + "Create one or more Fabro workflow runs from a native selector, inline files, or an exact stored workflow version, optionally under a parent run and starting them by default.", ), tool_definition::( FABRO_RUN_SEARCH_TOOL_NAME, @@ -329,6 +348,7 @@ mod tests { use fabro_types::{ RunLifecycle, RunLinks, RunOrigin, RunStatus, RunTimestamps, WorkflowRef, test_support, }; + use serde_json::Value; use super::*; @@ -368,6 +388,44 @@ mod tests { ); } + #[test] + fn create_tool_definition_advertises_complete_workflow_source_and_target_grammar() { + let definition = tool_definitions() + .iter() + .find(|definition| definition.name == FABRO_RUN_CREATE_TOOL_NAME) + .expect("create tool should be in the shared catalog"); + let variants = definition + .parameters + .pointer("/properties/runs/items/anyOf") + .and_then(Value::as_array) + .expect("run item union should be present"); + assert!(variants.iter().any(|variant| variant["type"] == "string")); + let object = variants + .iter() + .find(|variant| variant["type"] == "object") + .expect("object create specification should be present"); + assert_eq!(object["additionalProperties"], false); + let workflow = object + .pointer("/properties/workflow/anyOf") + .and_then(Value::as_array) + .expect("workflow source union should be present"); + assert!(workflow.iter().any(|variant| variant["type"] == "string")); + for kind in ["inline", "stored"] { + assert!(workflow.iter().any(|variant| { + variant.pointer("/properties/kind/const") == Some(&Value::from(kind)) + })); + } + let target = object + .pointer("/properties/target/anyOf") + .and_then(Value::as_array) + .expect("target union should be present"); + for kind in ["git", "none", "folder"] { + assert!(target.iter().any(|variant| { + variant.pointer("/properties/kind/const") == Some(&Value::from(kind)) + })); + } + } + #[test] fn pair_tool_definition_exposes_pair_schema() { let definition = tool_definitions() diff --git a/lib/components/fabro-tool/src/create.rs b/lib/components/fabro-tool/src/create.rs index 448e5e84f..9de538550 100644 --- a/lib/components/fabro-tool/src/create.rs +++ b/lib/components/fabro-tool/src/create.rs @@ -1,9 +1,9 @@ use std::borrow::Cow; -use std::collections::HashMap; +use std::collections::{BTreeMap, HashMap}; use std::path::{Path, PathBuf}; use std::sync::Arc; -use fabro_types::RunId; +use fabro_types::{RunId, RunTarget, WorkflowPath, WorkflowVersionId}; use schemars::{JsonSchema, Schema, SchemaGenerator, json_schema}; use serde::{Deserialize, Deserializer, Serialize, de}; use serde_json::Value; @@ -11,6 +11,10 @@ use serde_json::Value; use super::common::{self, FabroToolBackend, ToolError, ToolResult}; use super::manifest; +const MAX_INLINE_WORKFLOW_FILES: usize = fabro_types::MAX_WORKFLOW_VERSION_FILES; +const MAX_INLINE_WORKFLOW_FILE_BYTES: usize = fabro_types::MAX_WORKFLOW_VERSION_FILE_BYTES; +const MAX_INLINE_WORKFLOW_TOTAL_BYTES: usize = fabro_types::MAX_WORKFLOW_VERSION_BYTES; + #[derive(Debug, Deserialize, JsonSchema)] pub struct FabroRunCreateParams { pub runs: Vec, @@ -80,10 +84,38 @@ impl JsonSchema for CreateRunSpecInput { "type": "object", "description": "Full create-run specification.", "required": ["workflow"], + "additionalProperties": false, "properties": { "workflow": { - "type": "string", - "description": "Workflow selector, such as a workflow name or workflow file path." + "description": "Workflow content source. Selector strings require a proven shared filesystem; inline files and exact stored IDs are portable.", + "anyOf": [ + { + "type": "string", + "description": "Workflow selector, such as a workflow name or workflow file path." + }, + { + "type": "object", + "required": ["kind", "entrypoint", "files"], + "additionalProperties": false, + "properties": { + "kind": { "const": "inline" }, + "entrypoint": { "type": "string" }, + "files": { + "type": "object", + "additionalProperties": { "type": "string" } + } + } + }, + { + "type": "object", + "required": ["kind", "workflow_version_id"], + "additionalProperties": false, + "properties": { + "kind": { "const": "stored" }, + "workflow_version_id": { "type": "string" } + } + } + ] }, "cwd": { "anyOf": [ @@ -99,6 +131,51 @@ impl JsonSchema for CreateRunSpecInput { ], "description": "Optional parent run id or selector." }, + "target": { + "description": "Canonical run workspace target. Worker calls inherit the parent target when omitted; standalone calls require an observable Git checkout when omitted.", + "anyOf": [ + { "type": "null" }, + { + "type": "object", + "required": ["kind", "repo", "branch"], + "additionalProperties": false, + "properties": { + "kind": { "const": "git" }, + "repo": { "type": "string" }, + "branch": { "type": "string" }, + "tag": { + "anyOf": [ + { "type": "string" }, + { "type": "null" } + ] + }, + "sha": { + "anyOf": [ + { "type": "string" }, + { "type": "null" } + ] + } + } + }, + { + "type": "object", + "required": ["kind"], + "additionalProperties": false, + "properties": { + "kind": { "const": "none" } + } + }, + { + "type": "object", + "required": ["kind", "path"], + "additionalProperties": false, + "properties": { + "kind": { "const": "folder" }, + "path": { "type": "string" } + } + } + ] + }, "goal": { "anyOf": [ { "type": "string" }, @@ -187,11 +264,67 @@ impl JsonSchema for CreateRunSpecInput { } } -#[derive(Debug, Deserialize, JsonSchema)] +#[derive(Debug, Clone)] +pub enum CreateRunWorkflowSource { + Selector(String), + Inline(InlineWorkflowSource), + Stored { + workflow_version_id: WorkflowVersionId, + }, +} + +impl<'de> Deserialize<'de> for CreateRunWorkflowSource { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + #[derive(Deserialize)] + #[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] + enum TaggedSource { + Inline { + entrypoint: WorkflowPath, + files: BTreeMap, + }, + Stored { + workflow_version_id: WorkflowVersionId, + }, + } + + match Value::deserialize(deserializer)? { + Value::String(selector) => Ok(Self::Selector(selector)), + value @ Value::Object(_) => { + match serde_json::from_value::(value).map_err(de::Error::custom)? { + TaggedSource::Inline { entrypoint, files } => { + Ok(Self::Inline(InlineWorkflowSource { entrypoint, files })) + } + TaggedSource::Stored { + workflow_version_id, + } => Ok(Self::Stored { + workflow_version_id, + }), + } + } + other => Err(de::Error::custom(format!( + "expected workflow selector string or tagged source object, got {}", + json_value_kind(&other) + ))), + } + } +} + +#[derive(Debug, Clone)] +pub struct InlineWorkflowSource { + pub entrypoint: WorkflowPath, + pub files: BTreeMap, +} + +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] pub struct CreateRunSpec { - pub workflow: String, + pub workflow: CreateRunWorkflowSource, pub cwd: Option, pub parent_id: Option, + pub target: Option, pub goal: Option, pub goal_file: Option, #[serde(default)] @@ -252,12 +385,13 @@ pub struct ValidatedCreateRuns { #[derive(Debug)] pub struct ValidatedCreateRunSpec { - pub workflow: String, + pub workflow: ValidatedCreateRunWorkflowSource, pub cwd: Option, pub parent_id: Option, + pub target: Option, pub goal: Option, pub goal_file: Option, - pub inputs: HashMap, + pub inputs: HashMap, pub labels: HashMap, pub dry_run: Option, pub auto_approve: Option, @@ -268,6 +402,46 @@ pub struct ValidatedCreateRunSpec { pub start: Option, } +#[derive(Debug, Clone)] +pub enum ValidatedCreateRunWorkflowSource { + Selector(String), + Inline(InlineWorkflowSource), + Stored { + workflow_version_id: WorkflowVersionId, + }, +} + +impl ValidatedCreateRunWorkflowSource { + #[must_use] + pub fn display(&self) -> String { + match self { + Self::Selector(selector) => selector.clone(), + Self::Inline(source) => source.entrypoint.to_string(), + Self::Stored { + workflow_version_id, + } => workflow_version_id.to_string(), + } + } +} + +#[derive(Debug)] +pub struct ValidatedRunInputValue { + json: Value, + toml: toml::Value, +} + +impl ValidatedRunInputValue { + #[must_use] + pub fn json(&self) -> &Value { + &self.json + } + + #[must_use] + pub fn toml(&self) -> &toml::Value { + &self.toml + } +} + impl TryFrom for ValidatedCreateRuns { type Error = ToolError; @@ -287,28 +461,23 @@ impl TryFrom for ValidatedCreateRunSpec { fn try_from(spec: CreateRunSpecInput) -> Result { match spec { - CreateRunSpecInput::Workflow(workflow) => { - let workflow = workflow.trim(); - if workflow.is_empty() { - return Err(ToolError::message("workflow must not be blank")); - } - Self::try_from(CreateRunSpec { - workflow: workflow.to_string(), - cwd: None, - parent_id: None, - goal: None, - goal_file: None, - inputs: HashMap::new(), - labels: HashMap::new(), - dry_run: None, - auto_approve: None, - model: None, - provider: None, - environment: None, - preserve_sandbox: None, - start: None, - }) - } + CreateRunSpecInput::Workflow(workflow) => Self::try_from(CreateRunSpec { + workflow: CreateRunWorkflowSource::Selector(workflow), + cwd: None, + parent_id: None, + target: None, + goal: None, + goal_file: None, + inputs: HashMap::new(), + labels: HashMap::new(), + dry_run: None, + auto_approve: None, + model: None, + provider: None, + environment: None, + preserve_sandbox: None, + start: None, + }), CreateRunSpecInput::Spec(spec) => Self::try_from(*spec), } } @@ -318,6 +487,7 @@ impl TryFrom for ValidatedCreateRunSpec { type Error = ToolError; fn try_from(spec: CreateRunSpec) -> Result { + let workflow = validate_workflow_source(spec.workflow)?; let parent_id = spec .parent_id .as_deref() @@ -343,14 +513,25 @@ impl TryFrom for ValidatedCreateRunSpec { .inputs .into_iter() .map(|(key, value)| { - let value = value.into_inner(); - manifest::json_to_toml_value(&key, &value).map(|value| (key, value)) + let json = value.into_inner(); + manifest::json_to_toml_value(&key, &json) + .map(|toml| (key, ValidatedRunInputValue { json, toml })) }) .collect::>>()?; + let target = spec + .target + .map(|target| { + target + .validate() + .map(|validated| validated.target) + .map_err(|err| ToolError::message(format!("invalid run target: {err}"))) + }) + .transpose()?; Ok(Self { - workflow: spec.workflow, + workflow, cwd: spec.cwd, parent_id, + target, goal: spec.goal, goal_file: spec.goal_file, inputs, @@ -366,6 +547,58 @@ impl TryFrom for ValidatedCreateRunSpec { } } +fn validate_workflow_source( + source: CreateRunWorkflowSource, +) -> ToolResult { + match source { + CreateRunWorkflowSource::Selector(selector) => { + let selector = selector.trim(); + if selector.is_empty() { + return Err(ToolError::message("workflow selector must not be blank")); + } + Ok(ValidatedCreateRunWorkflowSource::Selector( + selector.to_string(), + )) + } + CreateRunWorkflowSource::Inline(source) => { + if source.files.len() > MAX_INLINE_WORKFLOW_FILES { + return Err(ToolError::message(format!( + "inline workflow contains more than {MAX_INLINE_WORKFLOW_FILES} files" + ))); + } + if !source.files.contains_key(&source.entrypoint) { + return Err(ToolError::message(format!( + "inline workflow entrypoint `{}` is missing from files", + source.entrypoint + ))); + } + let mut total_bytes = 0usize; + for (path, content) in &source.files { + let bytes = content.len(); + if bytes > MAX_INLINE_WORKFLOW_FILE_BYTES { + return Err(ToolError::message(format!( + "inline workflow file `{path}` exceeds 512 KiB" + ))); + } + total_bytes = total_bytes + .checked_add(bytes) + .ok_or_else(|| ToolError::message("inline workflow content size overflowed"))?; + if total_bytes > MAX_INLINE_WORKFLOW_TOTAL_BYTES { + return Err(ToolError::message( + "inline workflow content exceeds 2 MiB in aggregate", + )); + } + } + Ok(ValidatedCreateRunWorkflowSource::Inline(source)) + } + CreateRunWorkflowSource::Stored { + workflow_version_id, + } => Ok(ValidatedCreateRunWorkflowSource::Stored { + workflow_version_id, + }), + } +} + #[derive(Debug, Serialize, JsonSchema)] pub struct CreateRunsResult { pub runs: Vec, @@ -379,6 +612,9 @@ pub struct CreatedRunResult { pub workflow: String, pub start_requested: bool, pub status: String, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + #[schemars(default, skip_serializing_if = "Vec::is_empty")] + pub warnings: Vec, } #[derive(Debug, Clone, Copy, Default)] @@ -389,29 +625,21 @@ pub struct CreateRunOptions { pub async fn create_runs( backend: Arc, base_cwd: &Path, - user_settings_path: &Path, params: ValidatedCreateRuns, ) -> ToolResult { - create_runs_with_options( - backend, - base_cwd, - user_settings_path, - params, - CreateRunOptions::default(), - ) - .await + create_runs_with_options(backend, base_cwd, params, CreateRunOptions::default()).await } pub async fn create_runs_with_options( backend: Arc, base_cwd: &Path, - user_settings_path: &Path, params: ValidatedCreateRuns, options: CreateRunOptions, ) -> ToolResult { let mut created = Vec::with_capacity(params.runs.len()); let mut parent_id_cache = HashMap::::new(); for spec in params.runs { + let workflow = spec.workflow.display(); let cwd = spec.cwd.clone().unwrap_or_else(|| base_cwd.to_path_buf()); let parent_id = if let Some(forced_parent_id) = options.forced_parent_id { Some(forced_parent_id) @@ -423,10 +651,11 @@ pub async fn create_runs_with_options( } else { None }; - let run_id = backend - .create_run_from_spec(&spec, &cwd, user_settings_path, parent_id) + let submission = backend + .create_run_from_spec(&spec, &cwd, parent_id) .await .map_err(|err| ToolError::from_anyhow(&err))?; + let run_id = submission.run_id; let start_requested = spec.start.unwrap_or(true); let summary = if start_requested { backend @@ -443,9 +672,10 @@ pub async fn create_runs_with_options( run_id: summary.id.to_string(), parent_id: summary.parent_id.map(|parent_id| parent_id.to_string()), children_count: summary.children_count, - workflow: spec.workflow, + workflow, start_requested, status: summary.lifecycle.status.kind().to_string(), + warnings: submission.warnings, }); } Ok(CreateRunsResult { runs: created }) @@ -474,10 +704,15 @@ async fn resolve_parent_run_id( pub fn create_runs_text(result: &CreateRunsResult) -> String { let start_requested = result.runs.iter().filter(|run| run.start_requested).count(); - format!( + let mut text = format!( "created {} Fabro run(s), start requested for {start_requested}", result.runs.len() - ) + ); + for warning in result.runs.iter().flat_map(|run| &run.warnings) { + text.push_str("\nwarning: "); + text.push_str(warning); + } + text } #[cfg(test)] @@ -514,7 +749,7 @@ mod tests { } #[test] - fn create_spec_schema_omits_run_id() { + fn create_spec_schema_advertises_the_accepted_workflow_and_target_grammar() { let mut generator = SchemaGenerator::default(); let schema = CreateRunSpecInput::json_schema(&mut generator); let schema = serde_json::to_value(schema).expect("schema should serialize"); @@ -523,14 +758,37 @@ mod tests { .expect("object form should have properties"); assert!(!properties.contains_key("run_id")); + assert_eq!(schema["anyOf"][1]["additionalProperties"], false); + let workflow_variants = properties["workflow"]["anyOf"] + .as_array() + .expect("workflow should advertise all source variants"); + assert!( + workflow_variants + .iter() + .any(|variant| variant["type"] == "string") + ); + for kind in ["inline", "stored"] { + assert!(workflow_variants.iter().any(|variant| { + variant.pointer("/properties/kind/const") == Some(&json!(kind)) + })); + } + let target_variants = properties["target"]["anyOf"] + .as_array() + .expect("target should advertise the canonical target variants"); + for kind in ["git", "none", "folder"] { + assert!(target_variants.iter().any(|variant| { + variant.pointer("/properties/kind/const") == Some(&json!(kind)) + })); + } } #[test] fn create_spec_accepts_parent_selector() { let spec = ValidatedCreateRunSpec::try_from(CreateRunSpec { - workflow: "simple.fabro".to_string(), + workflow: selector("simple.fabro"), cwd: None, parent_id: Some(" nightly-parent ".to_string()), + target: None, goal: None, goal_file: None, inputs: HashMap::new(), @@ -558,7 +816,7 @@ mod tests { let params = ValidatedCreateRuns::try_from(params) .expect("string shorthand should validate as workflow selector"); let spec = ¶ms.runs[0]; - assert_eq!(spec.workflow, "simple.fabro"); + assert_eq!(spec.workflow.display(), "simple.fabro"); assert_eq!(spec.cwd, None); assert_eq!(spec.parent_id, None); assert!(spec.inputs.is_empty()); @@ -566,6 +824,148 @@ mod tests { assert_eq!(spec.start, None); } + #[test] + fn create_params_accept_inline_and_stored_workflow_sources() { + let stored_id = fabro_types::BlobHash::new(b"stored workflow").to_string(); + let inline: FabroRunCreateParams = serde_json::from_value(json!({ + "runs": [{ + "workflow": { + "kind": "inline", + "entrypoint": "flows/main.fabro", + "files": { + "flows/main.fabro": "digraph Main {}", + "prompts/goal.md": "Ship it" + } + }, + "target": { "kind": "none" }, + "start": false + }] + })) + .expect("inline workflow source should deserialize"); + let inline = + ValidatedCreateRuns::try_from(inline).expect("inline workflow source should validate"); + assert_eq!(inline.runs[0].workflow.display(), "flows/main.fabro"); + + let stored: FabroRunCreateParams = serde_json::from_value(json!({ + "runs": [{ + "workflow": { + "kind": "stored", + "workflow_version_id": stored_id + }, + "target": { "kind": "git", "repo": "fabro-sh/fabro", "branch": "main" } + }] + })) + .expect("stored workflow source should deserialize"); + let stored = + ValidatedCreateRuns::try_from(stored).expect("stored workflow source should validate"); + assert_eq!(stored.runs[0].workflow.display(), stored_id); + } + + #[test] + fn create_params_reject_invalid_workflow_sources_and_inputs_before_backend_work() { + for workflow in [ + json!({ + "kind": "inline", + "entrypoint": "../escape.fabro", + "files": { "../escape.fabro": "digraph W {}" } + }), + json!({ + "kind": "stored", + "workflow_version_id": "not-an-id" + }), + json!({ + "kind": "stored", + "workflow_version_id": fabro_types::BlobHash::new(b"stored").to_string(), + "extra": true + }), + ] { + serde_json::from_value::(json!({ + "runs": [{ "workflow": workflow }] + })) + .expect_err("invalid workflow source should fail deserialization"); + } + + let missing_entrypoint: FabroRunCreateParams = serde_json::from_value(json!({ + "runs": [{ + "workflow": { + "kind": "inline", + "entrypoint": "main.fabro", + "files": { "other.fabro": "digraph Other {}" } + } + }] + })) + .unwrap(); + assert!( + ValidatedCreateRuns::try_from(missing_entrypoint) + .unwrap_err() + .to_string() + .contains("entrypoint") + ); + + let too_many = (0..=MAX_INLINE_WORKFLOW_FILES) + .map(|index| (format!("files/{index}.md"), json!("x"))) + .collect::>(); + let too_many: FabroRunCreateParams = serde_json::from_value(json!({ + "runs": [{ + "workflow": { + "kind": "inline", + "entrypoint": "files/0.md", + "files": too_many + } + }] + })) + .unwrap(); + assert!(ValidatedCreateRuns::try_from(too_many).is_err()); + + let oversized: FabroRunCreateParams = serde_json::from_value(json!({ + "runs": [{ + "workflow": { + "kind": "inline", + "entrypoint": "main.fabro", + "files": { "main.fabro": "x".repeat(MAX_INLINE_WORKFLOW_FILE_BYTES + 1) } + } + }] + })) + .unwrap(); + assert!(ValidatedCreateRuns::try_from(oversized).is_err()); + + let aggregate: FabroRunCreateParams = serde_json::from_value(json!({ + "runs": [{ + "workflow": { + "kind": "inline", + "entrypoint": "0.fabro", + "files": { + "0.fabro": "x".repeat(MAX_INLINE_WORKFLOW_FILE_BYTES), + "1.md": "x".repeat(MAX_INLINE_WORKFLOW_FILE_BYTES), + "2.md": "x".repeat(MAX_INLINE_WORKFLOW_FILE_BYTES), + "3.md": "x".repeat(MAX_INLINE_WORKFLOW_FILE_BYTES), + "4.md": "x" + } + } + }] + })) + .unwrap(); + assert!(ValidatedCreateRuns::try_from(aggregate).is_err()); + + let invalid_target: FabroRunCreateParams = serde_json::from_value(json!({ + "runs": [{ + "workflow": "main.fabro", + "target": { "kind": "git", "repo": "not-a-slug", "branch": "HEAD" } + }] + })) + .unwrap(); + assert!(ValidatedCreateRuns::try_from(invalid_target).is_err()); + + let nonscalar: FabroRunCreateParams = serde_json::from_value(json!({ + "runs": [{ + "workflow": "main.fabro", + "inputs": { "nested": { "no": "objects" } } + }] + })) + .unwrap(); + assert!(ValidatedCreateRuns::try_from(nonscalar).is_err()); + } + #[test] fn create_params_preserve_object_form_options() { let params: FabroRunCreateParams = serde_json::from_value(json!({ @@ -582,7 +982,7 @@ mod tests { let params = ValidatedCreateRuns::try_from(params).expect("object form should still validate"); let spec = ¶ms.runs[0]; - assert_eq!(spec.workflow, "simple.fabro"); + assert_eq!(spec.workflow.display(), "simple.fabro"); assert_eq!(spec.dry_run, Some(true)); assert_eq!(spec.auto_approve, Some(true)); assert_eq!( @@ -593,20 +993,17 @@ mod tests { } #[test] - fn create_params_ignore_removed_run_id() { - let params: FabroRunCreateParams = serde_json::from_value(json!({ + fn create_params_reject_unknown_object_fields() { + let err = serde_json::from_value::(json!({ "runs": [{ "workflow": "simple.fabro", "run_id": "not-a-valid-run-id", "start": false }] })) - .expect("old object form should deserialize with run_id ignored"); + .expect_err("unknown create fields should be rejected"); - let params = - ValidatedCreateRuns::try_from(params).expect("remaining create fields should validate"); - assert_eq!(params.runs[0].workflow, "simple.fabro"); - assert_eq!(params.runs[0].start, Some(false)); + assert!(err.to_string().contains("unknown field `run_id`"), "{err}"); } #[test] @@ -676,7 +1073,6 @@ mod tests { #[tokio::test] async fn create_runs_resolves_parent_selector_and_sends_parent_id_to_backend() { let temp = tempfile::tempdir().expect("tempdir should be created"); - let settings = temp.path().join("settings.toml"); let child_id = run_id("01KRBZW5C00000000000000001"); let parent_id = run_id("01KRBZW4DW0000000000000002"); let backend = Arc::new(MockCreateBackend { @@ -685,13 +1081,16 @@ mod tests { created_parent_ids: Mutex::new(Vec::new()), resolved_selectors: Mutex::new(Vec::new()), started_run_ids: Mutex::new(Vec::new()), + warnings: Vec::new(), + create_error: false, }); let params = ValidatedCreateRuns::try_from(FabroRunCreateParams { runs: vec![ CreateRunSpec { - workflow: "simple.fabro".to_string(), + workflow: selector("simple.fabro"), cwd: None, parent_id: Some("nightly-parent".to_string()), + target: None, goal: None, goal_file: None, inputs: HashMap::new(), @@ -709,7 +1108,7 @@ mod tests { }) .expect("create params should validate"); - let result = create_runs(backend.clone(), temp.path(), &settings, params) + let result = create_runs(backend.clone(), temp.path(), params) .await .expect("run should be created"); @@ -726,7 +1125,6 @@ mod tests { #[tokio::test] async fn create_runs_reuses_parent_selector_resolution_within_batch() { let temp = tempfile::tempdir().expect("tempdir should be created"); - let settings = temp.path().join("settings.toml"); let child_id = run_id("01KRBZW5C00000000000000001"); let parent_id = run_id("01KRBZW4DW0000000000000002"); let backend = Arc::new(MockCreateBackend { @@ -735,13 +1133,16 @@ mod tests { created_parent_ids: Mutex::new(Vec::new()), resolved_selectors: Mutex::new(Vec::new()), started_run_ids: Mutex::new(Vec::new()), + warnings: Vec::new(), + create_error: false, }); let runs: Vec = (0..2) .map(|_| { CreateRunSpecInput::from(CreateRunSpec { - workflow: "simple.fabro".to_string(), + workflow: selector("simple.fabro"), cwd: None, parent_id: Some("nightly-parent".to_string()), + target: None, goal: None, goal_file: None, inputs: HashMap::new(), @@ -759,7 +1160,7 @@ mod tests { let params = ValidatedCreateRuns::try_from(FabroRunCreateParams { runs }) .expect("create params should validate"); - create_runs(backend.clone(), temp.path(), &settings, params) + create_runs(backend.clone(), temp.path(), params) .await .expect("runs should be created"); @@ -775,7 +1176,6 @@ mod tests { #[tokio::test] async fn create_runs_forced_parent_id_skips_selector_resolution() { let temp = tempfile::tempdir().expect("tempdir should be created"); - let settings = temp.path().join("settings.toml"); let child_id = run_id("01KRBZW5C00000000000000001"); let parent_id = run_id("01KRBZW4DW0000000000000002"); let backend = Arc::new(MockCreateBackend { @@ -784,13 +1184,16 @@ mod tests { created_parent_ids: Mutex::new(Vec::new()), resolved_selectors: Mutex::new(Vec::new()), started_run_ids: Mutex::new(Vec::new()), + warnings: Vec::new(), + create_error: false, }); let params = ValidatedCreateRuns::try_from(FabroRunCreateParams { runs: vec![ CreateRunSpec { - workflow: "simple.fabro".to_string(), + workflow: selector("simple.fabro"), cwd: None, parent_id: Some(parent_id.to_string()), + target: None, goal: None, goal_file: None, inputs: HashMap::new(), @@ -808,15 +1211,9 @@ mod tests { }) .expect("create params should validate"); - create_runs_with_options( - backend.clone(), - temp.path(), - &settings, - params, - CreateRunOptions { - forced_parent_id: Some(parent_id), - }, - ) + create_runs_with_options(backend.clone(), temp.path(), params, CreateRunOptions { + forced_parent_id: Some(parent_id), + }) .await .expect("run should be created"); @@ -829,7 +1226,6 @@ mod tests { #[tokio::test] async fn create_runs_defaults_to_start_request_and_reports_pending_child_status() { let temp = tempfile::tempdir().expect("tempdir should be created"); - let settings = temp.path().join("settings.toml"); let child_id = run_id("01KRBZW5C00000000000000001"); let parent_id = run_id("01KRBZW4DW0000000000000002"); let backend = Arc::new(MockCreateBackend { @@ -838,13 +1234,16 @@ mod tests { created_parent_ids: Mutex::new(Vec::new()), resolved_selectors: Mutex::new(Vec::new()), started_run_ids: Mutex::new(Vec::new()), + warnings: vec!["uncommitted changes are excluded".to_string()], + create_error: false, }); let params = ValidatedCreateRuns::try_from(FabroRunCreateParams { runs: vec![ CreateRunSpec { - workflow: "simple.fabro".to_string(), + workflow: selector("simple.fabro"), cwd: None, parent_id: Some(parent_id.to_string()), + target: None, goal: None, goal_file: None, inputs: HashMap::new(), @@ -862,7 +1261,7 @@ mod tests { }) .expect("create params should validate"); - let result = create_runs(backend.clone(), temp.path(), &settings, params) + let result = create_runs(backend.clone(), temp.path(), params) .await .expect("run should be created and start requested"); @@ -873,7 +1272,66 @@ mod tests { ]); assert_eq!( create_runs_text(&result), - "created 1 Fabro run(s), start requested for 1" + "created 1 Fabro run(s), start requested for 1\nwarning: uncommitted changes are excluded" + ); + assert_eq!(result.runs[0].warnings, [ + "uncommitted changes are excluded" + ]); + let wire = serde_json::to_value(&result).unwrap(); + assert_eq!( + wire["runs"][0]["warnings"], + json!(["uncommitted changes are excluded"]) + ); + } + + #[tokio::test] + async fn create_runs_create_failure_does_not_request_start() { + let temp = tempfile::tempdir().expect("tempdir should be created"); + let child_id = run_id("01KRBZW5C00000000000000001"); + let parent_id = run_id("01KRBZW4DW0000000000000002"); + let backend = Arc::new(MockCreateBackend { + child_id, + parent_id, + created_parent_ids: Mutex::new(Vec::new()), + resolved_selectors: Mutex::new(Vec::new()), + started_run_ids: Mutex::new(Vec::new()), + warnings: Vec::new(), + create_error: true, + }); + let params: FabroRunCreateParams = serde_json::from_value(json!({ + "runs": ["simple.fabro"] + })) + .unwrap(); + + create_runs( + backend.clone(), + temp.path(), + ValidatedCreateRuns::try_from(params).unwrap(), + ) + .await + .expect_err("create failure should be returned"); + + assert!(backend.started_run_ids.lock().unwrap().is_empty()); + } + + #[test] + fn create_run_result_omits_empty_warnings() { + let result = CreateRunsResult { + runs: vec![CreatedRunResult { + run_id: RunId::new().to_string(), + parent_id: None, + children_count: 0, + workflow: "simple".to_string(), + start_requested: false, + status: "submitted".to_string(), + warnings: Vec::new(), + }], + }; + + assert!( + serde_json::to_value(result).unwrap()["runs"][0] + .get("warnings") + .is_none() ); } @@ -881,6 +1339,10 @@ mod tests { raw.parse().expect("test run id should parse") } + fn selector(value: &str) -> CreateRunWorkflowSource { + CreateRunWorkflowSource::Selector(value.to_string()) + } + fn run(run_id: RunId, parent_id: Option, children_count: u64) -> Run { run_with_status(run_id, parent_id, children_count, RunStatus::Submitted) } @@ -946,6 +1408,8 @@ mod tests { created_parent_ids: Mutex>>, resolved_selectors: Mutex>, started_run_ids: Mutex>, + warnings: Vec, + create_error: bool, } #[async_trait] @@ -954,11 +1418,16 @@ mod tests { &self, _spec: &ValidatedCreateRunSpec, _cwd: &Path, - _user_settings_path: &Path, parent_id: Option, - ) -> anyhow::Result { + ) -> anyhow::Result { self.created_parent_ids.lock().unwrap().push(parent_id); - Ok(self.child_id) + if self.create_error { + anyhow::bail!("create failed"); + } + Ok(crate::CreateRunSubmission { + run_id: self.child_id, + warnings: self.warnings.clone(), + }) } async fn resolve_run(&self, selector: &str) -> anyhow::Result { diff --git a/lib/components/fabro-tool/src/fabro_client.rs b/lib/components/fabro-tool/src/fabro_client.rs index 120176183..62cff2625 100644 --- a/lib/components/fabro-tool/src/fabro_client.rs +++ b/lib/components/fabro-tool/src/fabro_client.rs @@ -5,16 +5,20 @@ use async_trait::async_trait; use fabro_api::types; use fabro_types::{ EventEnvelope, PairId, PairMessageRecord, PairMessageRequest, PairRecord, - PairTranscriptResponse, Run, RunId, RunPairStatusResponse, RunProjection, StageId, + PairTranscriptResponse, Run, RunId, RunIntent, RunIntentArgs, RunPairStatusResponse, + RunProjection, StageId, }; -use crate::{FabroToolBackend, RunManifestBuilder, ToolError, common}; +use crate::{ + CreateRunSubmission, FabroToolBackend, PreparedRunCreate, RunCreateAdapter, ToolError, + ValidatedCreateRunSpec, common, +}; #[derive(Clone)] pub struct ClientBackend { - client: Arc<::fabro_client::Client>, - manifest_builder: Option>, - run_scope: Option, + client: Arc<::fabro_client::Client>, + run_create_adapter: Option>, + run_scope: Option, workflow_version_packager: Option>, } @@ -23,15 +27,15 @@ impl ClientBackend { pub fn new(client: Arc<::fabro_client::Client>) -> Self { Self { client, - manifest_builder: None, + run_create_adapter: None, run_scope: None, workflow_version_packager: None, } } #[must_use] - pub fn with_manifest_builder(mut self, builder: Arc) -> Self { - self.manifest_builder = Some(builder); + pub fn with_run_create_adapter(mut self, adapter: Arc) -> Self { + self.run_create_adapter = Some(adapter); self } @@ -64,6 +68,41 @@ impl ClientBackend { } } +fn run_intent_from_spec( + spec: &ValidatedCreateRunSpec, + prepared: PreparedRunCreate, + parent_id: Option, +) -> (RunIntent, Vec) { + let PreparedRunCreate { + workflow_version_id, + target, + goal, + warnings, + } = prepared; + let intent = RunIntent { + workflow_version_id, + target, + args: RunIntentArgs { + model: spec.model.clone(), + provider: spec.provider.clone(), + inputs: spec + .inputs + .iter() + .map(|(key, value)| (key.clone(), value.json().clone())) + .collect(), + labels: spec.labels.clone(), + dry_run: spec.dry_run, + auto_approve: spec.auto_approve, + preserve_sandbox: spec.preserve_sandbox, + }, + environment_id: spec.environment.clone(), + parent_id, + title: None, + goal, + }; + (intent, warnings) +} + #[async_trait] impl FabroToolBackend for ClientBackend { /// Package the supplied tree, then register dependencies before parents. @@ -94,24 +133,22 @@ impl FabroToolBackend for ClientBackend { &self, spec: &crate::ValidatedCreateRunSpec, cwd: &Path, - user_settings_path: &Path, parent_id: Option, - ) -> anyhow::Result { + ) -> anyhow::Result { if let Some(parent_id) = parent_id.as_ref() { self.ensure_run_scope(parent_id)?; } - let Some(builder) = self.manifest_builder.as_ref() else { + let Some(adapter) = self.run_create_adapter.as_ref() else { return Err(ToolError::message(format!( "{} is not available", crate::FABRO_RUN_CREATE_TOOL_NAME )) .into()); }; - let mut manifest = builder - .build_run_manifest(spec, cwd, user_settings_path) - .map_err(anyhow::Error::new)?; - manifest.parent_id = parent_id.map(|run_id| run_id.to_string()); - self.client.create_run_from_manifest(manifest).await + let prepared = adapter.prepare(&self.client, spec, cwd).await?; + let (intent, warnings) = run_intent_from_spec(spec, prepared, parent_id); + let run_id = self.client.create_run_from_intent(intent).await?; + Ok(CreateRunSubmission { run_id, warnings }) } async fn resolve_run(&self, selector: &str) -> anyhow::Result { diff --git a/lib/components/fabro-tool/src/interact.rs b/lib/components/fabro-tool/src/interact.rs index 503147ece..16ba9edd4 100644 --- a/lib/components/fabro-tool/src/interact.rs +++ b/lib/components/fabro-tool/src/interact.rs @@ -746,9 +746,8 @@ mod tests { &self, _spec: &crate::ValidatedCreateRunSpec, _cwd: &Path, - _user_settings_path: &Path, _parent_id: Option, - ) -> anyhow::Result { + ) -> anyhow::Result { unreachable!() } diff --git a/lib/components/fabro-tool/src/lib.rs b/lib/components/fabro-tool/src/lib.rs index 5145e7608..ce33c90dc 100644 --- a/lib/components/fabro-tool/src/lib.rs +++ b/lib/components/fabro-tool/src/lib.rs @@ -17,15 +17,17 @@ 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, FABRO_WORKFLOW_VERSION_CREATE_TOOL_NAME, FabroToolBackend, - RunManifestBuilder, RunSummaryResult, ToolDefinition, ToolError, ToolResult, tool_definitions, + CreateRunSubmission, 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, PreparedRunCreate, + RunCreateAdapter, RunSummaryResult, ToolDefinition, ToolError, ToolResult, tool_definitions, + FABRO_WORKFLOW_VERSION_CREATE_TOOL_NAME, }; pub use create::{ - CreateRunOptions, CreateRunSpec, CreateRunSpecInput, CreateRunsResult, CreatedRunResult, - FabroRunCreateParams, RunInputValue, ValidatedCreateRunSpec, ValidatedCreateRuns, create_runs, - create_runs_text, create_runs_with_options, + CreateRunOptions, CreateRunSpec, CreateRunSpecInput, CreateRunWorkflowSource, CreateRunsResult, + CreatedRunResult, FabroRunCreateParams, InlineWorkflowSource, RunInputValue, + ValidatedCreateRunSpec, ValidatedCreateRunWorkflowSource, ValidatedCreateRuns, + ValidatedRunInputValue, create_runs, create_runs_text, create_runs_with_options, }; pub use events::{ FabroRunEventsParams, RunEventResult, RunEventsAction, RunEventsResult, ValidatedRunEvents, diff --git a/lib/components/fabro-workflow/src/services.rs b/lib/components/fabro-workflow/src/services.rs index d5f48a5c7..949bbf7d4 100644 --- a/lib/components/fabro-workflow/src/services.rs +++ b/lib/components/fabro-workflow/src/services.rs @@ -77,10 +77,9 @@ impl RunLocations { #[derive(Clone)] pub struct FabroRunToolServices { - pub backend: Arc, - pub current_run_id: RunId, - pub base_cwd: PathBuf, - pub user_settings_path: PathBuf, + pub backend: Arc, + pub current_run_id: RunId, + pub base_cwd: PathBuf, } /// Services shared across workflow phases.