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

Create run tools from immutable workflow versions

Browse source
This commit is contained in:
Scott Werner 2026-08-31 17:58:55 -04:00
parent bfcf82e367
commit c67c60eeba
17 changed files with 1687 additions and 397 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

48
docs/public/agents/mcp.mdx
View file

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

42
docs/public/execution/child-runs.mdx
View file

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

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

@ -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<fabro_types::RunTarget>,
source_directory: Option<&str>,
run_dir: &Path,
) -> Option<FabroRunToolServices> {
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<RunManifest> {
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<fabro_types::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.
@ -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),

1
lib/apps/fabro-mcp-server/src/lib.rs
View file

@ -1,6 +1,5 @@
mod config;
mod executable_monitor;
mod manifest_builder;
mod server;
use std::future::Future;

19
lib/apps/fabro-mcp-server/src/manifest_builder.rs
View file

@ -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<types::RunManifest> {
run_tool_manifest::build_run_tool_manifest(spec, cwd, user_settings_path)
}
}

46
lib/apps/fabro-mcp-server/src/server.rs
View file

@ -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<dyn FabroToolBackend>
})
.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))
}));
}
}
}

2
lib/apps/fabro-server/src/lib.rs
View file

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

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

@ -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<PathBuf>,
},
Worker {
provider: EnvironmentProvider,
inherited_target: Option<RunTarget>,
user_workflows_root: Option<PathBuf>,
},
}
impl ServerRunCreateAdapter {
#[must_use]
pub fn standalone(user_workflows_root: Option<PathBuf>) -> Self {
Self {
mode: RunCreateMode::Standalone {
user_workflows_root,
},
}
}
#[must_use]
pub fn worker(
provider: EnvironmentProvider,
inherited_target: Option<RunTarget>,
user_workflows_root: Option<PathBuf>,
) -> 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<Option<String>> {
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<ResolvedTarget> {
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<LocalWorkflowSource> {
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<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 {
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::<Vec<_>>();
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<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() {
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::<HashMap<_, _>>();
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<RunLayer> {
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<Mutex<Vec<WorkflowVersion>>>,
) -> 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(&registered)).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(&registered)).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"))
);
}
}

218
lib/apps/fabro-server/src/run_tool_manifest.rs
View file

@ -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<types::RunManifest> {
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<types::ManifestArgs> {
let mut input = spec
.inputs
.iter()
.map(|(key, value)| format!("{key}={value}"))
.collect::<Vec<_>>();
input.sort();
let mut label = spec
.labels
.iter()
.map(|(key, value)| format!("{key}={value}"))
.collect::<Vec<_>>();
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<RunLayer> {
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");
}
}

7
lib/apps/fabro-server/src/server/handler/sessions.rs
View file

@ -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}");

74
lib/apps/fabro-server/src/server/handler/workflow_versions.rs
View file

@ -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();

84
lib/components/fabro-tool/src/common.rs
View file

@ -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<T> = Result<T, ToolError>;
#[derive(Debug)]
pub struct PreparedRunCreate {
pub workflow_version_id: WorkflowVersionId,
pub target: RunTarget,
pub goal: Option<String>,
pub warnings: Vec<String>,
}
#[derive(Debug)]
pub struct CreateRunSubmission {
pub run_id: RunId,
pub warnings: Vec<String>,
}
/// 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<PreparedRunCreate>;
}
#[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<RunId>,
) -> anyhow::Result<RunId>;
) -> anyhow::Result<CreateRunSubmission>;
async fn resolve_run(&self, selector: &str) -> anyhow::Result<Run>;
async fn retrieve_run(&self, run_id: &RunId) -> anyhow::Result<Run>;
@ -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<types::RunManifest>;
}
#[derive(Debug, Serialize, JsonSchema)]
pub struct RunSummaryResult {
pub run_id: String,
@ -201,7 +220,7 @@ static TOOL_DEFINITIONS: LazyLock<Vec<ToolDefinition>> = LazyLock::new(|| {
),
tool_definition::<crate::FabroRunCreateParams>(
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::<crate::FabroRunSearchParams>(
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()

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

@ -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<CreateRunSpecInput>,
@ -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<D>(deserializer: D) -> Result<Self, D::Error>
where
D: Deserializer<'de>,
{
#[derive(Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
enum TaggedSource {
Inline {
entrypoint: WorkflowPath,
files: BTreeMap<WorkflowPath, String>,
},
Stored {
workflow_version_id: WorkflowVersionId,
},
}
match Value::deserialize(deserializer)? {
Value::String(selector) => Ok(Self::Selector(selector)),
value @ Value::Object(_) => {
match serde_json::from_value::<TaggedSource>(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<WorkflowPath, String>,
}
#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct CreateRunSpec {
pub workflow: String,
pub workflow: CreateRunWorkflowSource,
pub cwd: Option<PathBuf>,
pub parent_id: Option<String>,
pub target: Option<RunTarget>,
pub goal: Option<String>,
pub goal_file: Option<PathBuf>,
#[serde(default)]
@ -252,12 +385,13 @@ pub struct ValidatedCreateRuns {
#[derive(Debug)]
pub struct ValidatedCreateRunSpec {
pub workflow: String,
pub workflow: ValidatedCreateRunWorkflowSource,
pub cwd: Option<PathBuf>,
pub parent_id: Option<String>,
pub target: Option<RunTarget>,
pub goal: Option<String>,
pub goal_file: Option<PathBuf>,
pub inputs: HashMap<String, toml::Value>,
pub inputs: HashMap<String, ValidatedRunInputValue>,
pub labels: HashMap<String, String>,
pub dry_run: Option<bool>,
pub auto_approve: Option<bool>,
@ -268,6 +402,46 @@ 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,
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<FabroRunCreateParams> for ValidatedCreateRuns {
type Error = ToolError;
@ -287,28 +461,23 @@ impl TryFrom<CreateRunSpecInput> for ValidatedCreateRunSpec {
fn try_from(spec: CreateRunSpecInput) -> Result<Self, Self::Error> {
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<CreateRunSpec> for ValidatedCreateRunSpec {
type Error = ToolError;
fn try_from(spec: CreateRunSpec) -> Result<Self, Self::Error> {
let workflow = validate_workflow_source(spec.workflow)?;
let parent_id = spec
.parent_id
.as_deref()
@ -343,14 +513,25 @@ impl TryFrom<CreateRunSpec> 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::<ToolResult<HashMap<_, _>>>()?;
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<CreateRunSpec> for ValidatedCreateRunSpec {
}
}
fn validate_workflow_source(
source: CreateRunWorkflowSource,
) -> ToolResult<ValidatedCreateRunWorkflowSource> {
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<CreatedRunResult>,
@ -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<String>,
}
#[derive(Debug, Clone, Copy, Default)]
@ -389,29 +625,21 @@ pub struct CreateRunOptions {
pub async fn create_runs(
backend: Arc<dyn FabroToolBackend>,
base_cwd: &Path,
user_settings_path: &Path,
params: ValidatedCreateRuns,
) -> ToolResult<CreateRunsResult> {
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<dyn FabroToolBackend>,
base_cwd: &Path,
user_settings_path: &Path,
params: ValidatedCreateRuns,
options: CreateRunOptions,
) -> ToolResult<CreateRunsResult> {
let mut created = Vec::with_capacity(params.runs.len());
let mut parent_id_cache = HashMap::<String, RunId>::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 = &params.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::<FabroRunCreateParams>(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::<serde_json::Map<_, _>>();
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 = &params.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::<FabroRunCreateParams>(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<CreateRunSpecInput> = (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<RunId>, 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<Vec<Option<RunId>>>,
resolved_selectors: Mutex<Vec<String>>,
started_run_ids: Mutex<Vec<RunId>>,
warnings: Vec<String>,
create_error: bool,
}
#[async_trait]
@ -954,11 +1418,16 @@ mod tests {
&self,
_spec: &ValidatedCreateRunSpec,
_cwd: &Path,
_user_settings_path: &Path,
parent_id: Option<RunId>,
) -> anyhow::Result<RunId> {
) -> anyhow::Result<crate::CreateRunSubmission> {
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<Run> {

69
lib/components/fabro-tool/src/fabro_client.rs
View file

@ -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<Arc<dyn RunManifestBuilder>>,
run_scope: Option<RunId>,
client: Arc<::fabro_client::Client>,
run_create_adapter: Option<Arc<dyn RunCreateAdapter>>,
run_scope: Option<RunId>,
workflow_version_packager: Option<Arc<dyn crate::WorkflowVersionPackager>>,
}
@ -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<dyn RunManifestBuilder>) -> Self {
self.manifest_builder = Some(builder);
pub fn with_run_create_adapter(mut self, adapter: Arc<dyn RunCreateAdapter>) -> 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<RunId>,
) -> (RunIntent, Vec<String>) {
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<RunId>,
) -> anyhow::Result<RunId> {
) -> anyhow::Result<CreateRunSubmission> {
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<Run> {

3
lib/components/fabro-tool/src/interact.rs
View file

@ -746,9 +746,8 @@ mod tests {
&self,
_spec: &crate::ValidatedCreateRunSpec,
_cwd: &Path,
_user_settings_path: &Path,
_parent_id: Option<RunId>,
) -> anyhow::Result<RunId> {
) -> anyhow::Result<crate::CreateRunSubmission> {
unreachable!()
}

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

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

7
lib/components/fabro-workflow/src/services.rs
View file

@ -77,10 +77,9 @@ impl RunLocations {
#[derive(Clone)]
pub struct FabroRunToolServices {
pub backend: Arc<dyn fabro_tool::FabroToolBackend>,
pub current_run_id: RunId,
pub base_cwd: PathBuf,
pub user_settings_path: PathBuf,
pub backend: Arc<dyn fabro_tool::FabroToolBackend>,
pub current_run_id: RunId,
pub base_cwd: PathBuf,
}
/// Services shared across workflow phases.