scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-15 23:32:46 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

Apply cleanup review fixes to the run-create path

Browse source 
Deduplicate shared-filesystem capability checks and simplify workflow-source dispatch and types. Move Git observation and local package collection onto spawn_blocking, and flush inline workflow files before collection.

Simplify validated source and input types, derive inline size-limit messages from shared constants, add target schema-parity coverage, and remove dead producer pass-through parameters.
This commit is contained in:
Scott Werner 2026-09-01 17:47:06 -04:00
parent 45741a3e6e
commit b38279fc44
8 changed files with 182 additions and 162 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
Cargo.lock generated
View file

@ -3176,6 +3176,7 @@ dependencies = [
"fabro-workflow-version",
"futures",
"httpmock",
"jsonschema",
"schemars 1.2.1",
"serde",
"serde_json",

13
lib/apps/fabro-cli/src/commands/run/runner.rs
View file

@ -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<RunTarget>,
user_workflows_root: Option<PathBuf>,
) -> 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")),

174
lib/apps/fabro-server/src/run_tool_create.rs
View file

@ -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<ResolvedTarget> {
async fn resolve_target(
&self,
spec: &ValidatedCreateRunSpec,
cwd: &Path,
) -> Result<ResolvedTarget> {
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<LocalWorkflowSource> {
async fn collect_selector(
&self,
selector: &str,
cwd: &Path,
) -> Result<CollectedWorkflowClosure> {
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<PreparedRunCreate> {
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<String>,
}
enum LocalWorkflowSource {
Selector(ResolvedLocalWorkflowPackage),
Inline {
closure: CollectedWorkflowClosure,
_root: tempfile::TempDir,
},
}
impl LocalWorkflowSource {
async fn inline(source: &fabro_tool::InlineWorkflowSource) -> Result<Self> {
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<CollectedWorkflowClosure> {
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(

15
lib/components/fabro-manifest/src/lib.rs
View file

@ -341,11 +341,8 @@ pub fn observe_git_run_target(
) -> Option<GitRunTargetObservation> {
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<String>) -> Option<GitRunTarget> {
fn github_run_target(origin_url: &str, branch: &str) -> Option<GitRunTarget> {
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()?;

5
lib/components/fabro-manifest/src/local_workflow_package.rs
View file

@ -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

1
lib/components/fabro-tool/Cargo.toml
View file

@ -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"

131
lib/components/fabro-tool/src/create.rs
View file

@ -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<CreateRunSpecInput>,
@ -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<D>(deserializer: D) -> Result<Self, D::Error>
where
@ -385,7 +394,7 @@ pub struct ValidatedCreateRuns {
#[derive(Debug)]
pub struct ValidatedCreateRunSpec {
pub workflow: ValidatedCreateRunWorkflowSource,
pub workflow: CreateRunWorkflowSource,
pub cwd: Option<PathBuf>,
pub parent_id: Option<String>,
pub target: Option<RunTarget>,
@ -402,28 +411,6 @@ pub struct ValidatedCreateRunSpec {
pub start: Option<bool>,
}
#[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<CreateRunSpec> for ValidatedCreateRunSpec {
fn validate_workflow_source(
source: CreateRunWorkflowSource,
) -> ToolResult<ValidatedCreateRunWorkflowSource> {
) -> ToolResult<CreateRunWorkflowSource> {
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::<serde_json::Map<_, _>>();
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"
}
}

4
lib/components/fabro-tool/src/lib.rs
View file

@ -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,