diff --git a/Cargo.lock b/Cargo.lock index df85b11db..ff0091029 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3176,6 +3176,7 @@ dependencies = [ "fabro-workflow-version", "futures", "httpmock", + "jsonschema", "schemars 1.2.1", "serde", "serde_json", diff --git a/lib/apps/fabro-cli/src/commands/run/runner.rs b/lib/apps/fabro-cli/src/commands/run/runner.rs index a3d6ab858..76755495d 100644 --- a/lib/apps/fabro-cli/src/commands/run/runner.rs +++ b/lib/apps/fabro-cli/src/commands/run/runner.rs @@ -246,7 +246,7 @@ fn build_fabro_run_tool_services( .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), + ServerRunCreateAdapter::worker(provider, inherited_target, user_workflows_root), )).with_workflow_version_packager(Arc::new(SuppliedWorkflowVersionPackager)); Some(FabroRunToolServices { backend: Arc::new(backend), @@ -255,14 +255,6 @@ fn build_fabro_run_tool_services( }) } -fn worker_run_create_adapter( - provider: 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. /// /// A worker always receives the server storage root so it can load the same @@ -1209,6 +1201,7 @@ mod tests { use fabro_interview::{ AnswerValue, ControlInterviewer, Interviewer, Question, WorkerControlEnvelope, }; + use fabro_server::run_tool_create::ServerRunCreateAdapter; use fabro_types::run_event::{ InterviewCompletedProps, InterviewStartedProps, RunCompletedProps, RunControlEffectProps, RunFailedProps, RunStatusTransitionProps, @@ -1295,7 +1288,7 @@ mod tests { .unwrap() .runs .remove(0); - let adapter = super::worker_run_create_adapter( + let adapter = ServerRunCreateAdapter::worker( EnvironmentProvider::Docker, Some(inherited.clone()), Some(temp.path().join("workflows")), diff --git a/lib/apps/fabro-server/src/run_tool_create.rs b/lib/apps/fabro-server/src/run_tool_create.rs index 415b28b3f..f4f36e8da 100644 --- a/lib/apps/fabro-server/src/run_tool_create.rs +++ b/lib/apps/fabro-server/src/run_tool_create.rs @@ -9,12 +9,12 @@ use fabro_manifest::{ collect_workflow_versions, observe_git_run_target, resolve_local_workflow_package, }; use fabro_tool::{ - PreparedRunCreate, RunCreateAdapter, ValidatedCreateRunSpec, ValidatedCreateRunWorkflowSource, + CreateRunWorkflowSource, PreparedRunCreate, RunCreateAdapter, ValidatedCreateRunSpec, }; use fabro_types::settings::run::EnvironmentProvider; use fabro_types::{DirtyStatus, RunTarget}; -use tokio::fs; use tokio::io::AsyncWriteExt; +use tokio::{fs, task}; use crate::manifest_validation; @@ -102,7 +102,11 @@ impl ServerRunCreateAdapter { .map(Some) } - fn resolve_target(&self, spec: &ValidatedCreateRunSpec, cwd: &Path) -> Result { + async fn resolve_target( + &self, + spec: &ValidatedCreateRunSpec, + cwd: &Path, + ) -> Result { if let Some(target) = &spec.target { return Ok(ResolvedTarget { target: target.clone(), @@ -125,7 +129,13 @@ impl ServerRunCreateAdapter { "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(|| { + let observation_cwd = cwd.to_path_buf(); + let observation = task::spawn_blocking(move || { + observe_git_run_target(&observation_cwd, None) + }) + .await + .context("git target observation task failed")? + .ok_or_else(|| { anyhow::anyhow!( "target is required outside an attached local GitHub checkout with a branch" ) @@ -162,15 +172,26 @@ impl ServerRunCreateAdapter { } } - fn collect_selector(&self, selector: &str, cwd: &Path) -> Result { + async 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) + let selector = PathBuf::from(selector); + let cwd = cwd.to_path_buf(); + let user_workflows_root = self.user_workflows_root().map(Path::to_path_buf); + task::spawn_blocking(move || { + resolve_local_workflow_package(&selector, &cwd, user_workflows_root.as_deref()) + .map(ResolvedLocalWorkflowPackage::into_closure) + .map_err(anyhow::Error::new) + }) + .await + .context("workflow package collection task failed")? } } @@ -182,45 +203,26 @@ impl RunCreateAdapter for ServerRunCreateAdapter { 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 { + let closure = match &spec.workflow { + CreateRunWorkflowSource::Stored { 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)? + } => { + let resolved_target = self.resolve_target(spec, cwd).await?; + return Ok(PreparedRunCreate { + workflow_version_id: *workflow_version_id, + target: resolved_target.target, + goal, + warnings: resolved_target.warnings, + }); } - ValidatedCreateRunWorkflowSource::Inline(source) => { - LocalWorkflowSource::inline(source).await? + CreateRunWorkflowSource::Selector(selector) => { + self.collect_selector(selector, cwd).await? } - ValidatedCreateRunWorkflowSource::Stored { .. } => unreachable!(), + CreateRunWorkflowSource::Inline(source) => collect_inline_workflow(source).await?, }; - validate_local_source(local_source.closure(), spec, goal.as_deref())?; - let resolved_target = self.resolve_target(spec, cwd)?; - let closure = local_source.closure(); + validate_local_source(&closure, spec, goal.as_deref())?; + let resolved_target = self.resolve_target(spec, cwd).await?; let versions = closure .versions() .map(|(_, version)| version.version()) @@ -241,59 +243,51 @@ struct ResolvedTarget { 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() { - fs::create_dir_all(parent).await.with_context(|| { - format!( - "failed to create inline workflow directory {}", - parent.display() - ) - })?; - } - let mut file = 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(|| { +/// Collect an inline workflow by staging its bytes in a private temporary +/// root; the collected closure owns every file, so the root is discarded on +/// return. +async fn collect_inline_workflow( + 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() { + fs::create_dir_all(parent).await.with_context(|| { format!( - "failed to write inline workflow file {}", - destination.display() + "failed to create inline workflow directory {}", + parent.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, - } + let mut file = 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() + ) + })?; + // Dropping a tokio File does not wait for queued writes; the + // collector below reads these files synchronously, so flush first. + file.flush().await.with_context(|| { + format!( + "failed to flush inline workflow file {}", + destination.display() + ) + })?; } + collect_workflow_versions(Path::new(source.entrypoint.as_str()), root.path()) + .map_err(anyhow::Error::new) } fn validate_local_source( diff --git a/lib/components/fabro-manifest/src/lib.rs b/lib/components/fabro-manifest/src/lib.rs index f020b5828..1441d36c9 100644 --- a/lib/components/fabro-manifest/src/lib.rs +++ b/lib/components/fabro-manifest/src/lib.rs @@ -341,11 +341,8 @@ pub fn observe_git_run_target( ) -> Option { let local = inspect_local_git(repo_path, configured_repo_origin_url)?; let legacy_git_context = local.legacy_git_context; - let mut run_target = github_run_target( - &legacy_git_context.origin_url, - &legacy_git_context.branch, - None, - ); + let mut run_target = + github_run_target(&legacy_git_context.origin_url, &legacy_git_context.branch); if let Some(target) = run_target.as_mut() { let publish_status = publish_manifest_branch_best_effort( repo_path, @@ -422,14 +419,14 @@ fn build_legacy_git_context( Some(local.legacy_git_context) } -fn github_run_target(origin_url: &str, branch: &str, sha: Option) -> Option { +fn github_run_target(origin_url: &str, branch: &str) -> Option { let (owner, repository) = fabro_github::parse_github_owner_repo(origin_url).ok()?; let slug = GitHubRepositorySlug::try_new(&format!("{owner}/{repository}"))?; let validated = RunTarget::Git(GitRunTarget { - repo: slug.to_string(), + repo: slug.to_string(), branch: branch.to_owned(), - tag: None, - sha, + tag: None, + sha: None, }) .validate() .ok()?; diff --git a/lib/components/fabro-manifest/src/local_workflow_package.rs b/lib/components/fabro-manifest/src/local_workflow_package.rs index 5c32a4412..0e9270676 100644 --- a/lib/components/fabro-manifest/src/local_workflow_package.rs +++ b/lib/components/fabro-manifest/src/local_workflow_package.rs @@ -66,6 +66,11 @@ impl ResolvedLocalWorkflowPackage { pub fn closure(&self) -> &CollectedWorkflowClosure { &self.closure } + + #[must_use] + pub fn into_closure(self) -> CollectedWorkflowClosure { + self.closure + } } /// Resolve producer-readable workflow bytes under one stable local source diff --git a/lib/components/fabro-tool/Cargo.toml b/lib/components/fabro-tool/Cargo.toml index 1e5395b9e..dc2936a3f 100644 --- a/lib/components/fabro-tool/Cargo.toml +++ b/lib/components/fabro-tool/Cargo.toml @@ -32,4 +32,5 @@ toml.workspace = true [dev-dependencies] fabro-types = { path = "../../foundation/fabro-types", features = ["test-support"] } httpmock = "0.8" +jsonschema.workspace = true tempfile = "3" diff --git a/lib/components/fabro-tool/src/create.rs b/lib/components/fabro-tool/src/create.rs index 9de538550..53cb1800b 100644 --- a/lib/components/fabro-tool/src/create.rs +++ b/lib/components/fabro-tool/src/create.rs @@ -11,10 +11,6 @@ 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, @@ -273,6 +269,19 @@ pub enum CreateRunWorkflowSource { }, } +impl CreateRunWorkflowSource { + #[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(), + } + } +} + impl<'de> Deserialize<'de> for CreateRunWorkflowSource { fn deserialize(deserializer: D) -> Result where @@ -385,7 +394,7 @@ pub struct ValidatedCreateRuns { #[derive(Debug)] pub struct ValidatedCreateRunSpec { - pub workflow: ValidatedCreateRunWorkflowSource, + pub workflow: CreateRunWorkflowSource, pub cwd: Option, pub parent_id: Option, pub target: Option, @@ -402,28 +411,6 @@ 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, @@ -549,21 +536,20 @@ impl TryFrom for ValidatedCreateRunSpec { fn validate_workflow_source( source: CreateRunWorkflowSource, -) -> ToolResult { +) -> 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(), - )) + Ok(CreateRunWorkflowSource::Selector(selector.to_string())) } CreateRunWorkflowSource::Inline(source) => { - if source.files.len() > MAX_INLINE_WORKFLOW_FILES { + if source.files.len() > fabro_types::MAX_WORKFLOW_VERSION_FILES { return Err(ToolError::message(format!( - "inline workflow contains more than {MAX_INLINE_WORKFLOW_FILES} files" + "inline workflow contains more than {} files", + fabro_types::MAX_WORKFLOW_VERSION_FILES ))); } if !source.files.contains_key(&source.entrypoint) { @@ -575,27 +561,25 @@ fn validate_workflow_source( let mut total_bytes = 0usize; for (path, content) in &source.files { let bytes = content.len(); - if bytes > MAX_INLINE_WORKFLOW_FILE_BYTES { + if bytes > fabro_types::MAX_WORKFLOW_VERSION_FILE_BYTES { return Err(ToolError::message(format!( - "inline workflow file `{path}` exceeds 512 KiB" + "inline workflow file `{path}` exceeds {} KiB", + fabro_types::MAX_WORKFLOW_VERSION_FILE_BYTES / 1024 ))); } 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", - )); + if total_bytes > fabro_types::MAX_WORKFLOW_VERSION_BYTES { + return Err(ToolError::message(format!( + "inline workflow content exceeds {} MiB in aggregate", + fabro_types::MAX_WORKFLOW_VERSION_BYTES / (1024 * 1024) + ))); } } - Ok(ValidatedCreateRunWorkflowSource::Inline(source)) + Ok(CreateRunWorkflowSource::Inline(source)) } - CreateRunWorkflowSource::Stored { - workflow_version_id, - } => Ok(ValidatedCreateRunWorkflowSource::Stored { - workflow_version_id, - }), + stored @ CreateRunWorkflowSource::Stored { .. } => Ok(stored), } } @@ -782,6 +766,51 @@ mod tests { } } + #[test] + fn create_spec_schema_stays_in_parity_with_run_target_serde() { + let mut generator = SchemaGenerator::default(); + let schema = CreateRunSpecInput::json_schema(&mut generator); + let schema = serde_json::to_value(schema).expect("schema should serialize"); + let validator = jsonschema::validator_for(&schema).expect("advertised schema must compile"); + + // Every serde-produced target shape must satisfy the hand-written + // schema literal; a field added to a target variant without updating + // the literal fails here because the schema denies unknown fields. + let targets = [ + RunTarget::Git(fabro_types::GitRunTarget { + repo: "fabro-sh/fabro".to_string(), + branch: "main".to_string(), + tag: Some("v1.0.0".to_string()), + sha: Some("0123456789abcdef0123456789abcdef01234567".to_string()), + }), + RunTarget::None {}, + RunTarget::Folder { + path: "/srv/workspace".to_string(), + }, + ]; + for target in targets { + let target = serde_json::to_value(&target).expect("target should serialize"); + let spec = json!({ "workflow": "demo", "target": target }); + assert!( + validator.is_valid(&spec), + "advertised schema rejects serde-produced target {target}" + ); + } + + // The schema must actually enforce the field lists, so the parity + // assertions above have teeth. + let unknown_field = json!({ + "workflow": "demo", + "target": { + "kind": "git", + "repo": "fabro-sh/fabro", + "branch": "main", + "unknown_field": true + } + }); + assert!(!validator.is_valid(&unknown_field)); + } + #[test] fn create_spec_accepts_parent_selector() { let spec = ValidatedCreateRunSpec::try_from(CreateRunSpec { @@ -902,7 +931,7 @@ mod tests { .contains("entrypoint") ); - let too_many = (0..=MAX_INLINE_WORKFLOW_FILES) + let too_many = (0..=fabro_types::MAX_WORKFLOW_VERSION_FILES) .map(|index| (format!("files/{index}.md"), json!("x"))) .collect::>(); let too_many: FabroRunCreateParams = serde_json::from_value(json!({ @@ -922,7 +951,7 @@ mod tests { "workflow": { "kind": "inline", "entrypoint": "main.fabro", - "files": { "main.fabro": "x".repeat(MAX_INLINE_WORKFLOW_FILE_BYTES + 1) } + "files": { "main.fabro": "x".repeat(fabro_types::MAX_WORKFLOW_VERSION_FILE_BYTES + 1) } } }] })) @@ -935,10 +964,10 @@ mod tests { "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), + "0.fabro": "x".repeat(fabro_types::MAX_WORKFLOW_VERSION_FILE_BYTES), + "1.md": "x".repeat(fabro_types::MAX_WORKFLOW_VERSION_FILE_BYTES), + "2.md": "x".repeat(fabro_types::MAX_WORKFLOW_VERSION_FILE_BYTES), + "3.md": "x".repeat(fabro_types::MAX_WORKFLOW_VERSION_FILE_BYTES), "4.md": "x" } } diff --git a/lib/components/fabro-tool/src/lib.rs b/lib/components/fabro-tool/src/lib.rs index ce33c90dc..7ce805cc1 100644 --- a/lib/components/fabro-tool/src/lib.rs +++ b/lib/components/fabro-tool/src/lib.rs @@ -26,8 +26,8 @@ pub use common::{ pub use create::{ CreateRunOptions, CreateRunSpec, CreateRunSpecInput, CreateRunWorkflowSource, CreateRunsResult, CreatedRunResult, FabroRunCreateParams, InlineWorkflowSource, RunInputValue, - ValidatedCreateRunSpec, ValidatedCreateRunWorkflowSource, ValidatedCreateRuns, - ValidatedRunInputValue, create_runs, create_runs_text, create_runs_with_options, + ValidatedCreateRunSpec, ValidatedCreateRuns, ValidatedRunInputValue, create_runs, + create_runs_text, create_runs_with_options, }; pub use events::{ FabroRunEventsParams, RunEventResult, RunEventsAction, RunEventsResult, ValidatedRunEvents,