mirror of
https://github.com/fabro-sh/fabro.git
synced 2026-08-28 05:27:41 +00:00
Merge 077d020223 into dc55183468
This commit is contained in:
Scott Werner
GitHub
commit
1aa11da1c7
16 changed files with 1053 additions and 323 deletions
2
Cargo.lock
generated
2
Cargo.lock
generated
|
|
@ -2827,11 +2827,13 @@ dependencies = [
|
|||
"fabro-test",
|
||||
"fabro-types",
|
||||
"fabro-workflow",
|
||||
"fabro-workflow-version",
|
||||
"git2",
|
||||
"insta",
|
||||
"serde_json",
|
||||
"temp-env",
|
||||
"tempfile",
|
||||
"thiserror 2.0.18",
|
||||
"toml 0.8.23",
|
||||
]
|
||||
|
||||
|
|
|
|||
|
|
@ -2,11 +2,13 @@ use std::path::PathBuf;
|
|||
use std::sync::Arc;
|
||||
|
||||
use async_trait::async_trait;
|
||||
use fabro_api::types::RunManifest;
|
||||
use fabro_automation::AutomationId;
|
||||
use fabro_config::{EnvironmentLayer, MergeMap};
|
||||
use fabro_manifest::ManifestBuildInput;
|
||||
use fabro_types::{GitHubRepositorySlug, GitRunTarget, RunId, RunTarget, TargetValidationError};
|
||||
use fabro_manifest::WorkflowVersionCollectError;
|
||||
use fabro_types::{
|
||||
GitHubRepositorySlug, GitRunTarget, RunId, RunIntent, RunIntentArgs, RunTarget,
|
||||
TargetValidationError, WorkflowVersionId,
|
||||
};
|
||||
use fabro_workflow_version::{WorkflowVersionStore, WorkflowVersionStoreError};
|
||||
use tokio::{fs, task};
|
||||
|
||||
use crate::git_checkout::{
|
||||
|
|
@ -15,19 +17,33 @@ use crate::git_checkout::{
|
|||
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub(crate) struct AutomationRunMaterializeInput {
|
||||
pub automation_id: AutomationId,
|
||||
pub target: GitRunTarget,
|
||||
pub workflow: String,
|
||||
pub run_id: RunId,
|
||||
pub user_settings_path: PathBuf,
|
||||
pub temp_root: PathBuf,
|
||||
pub automation_id: AutomationId,
|
||||
pub target: GitRunTarget,
|
||||
pub workflow: String,
|
||||
pub run_id: RunId,
|
||||
pub temp_root: PathBuf,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub(crate) struct AutomationRunMaterialized {
|
||||
pub manifest: RunManifest,
|
||||
pub submitted_manifest_bytes: Vec<u8>,
|
||||
pub target: GitRunTarget,
|
||||
pub workflow_version_id: WorkflowVersionId,
|
||||
pub target: GitRunTarget,
|
||||
}
|
||||
|
||||
impl AutomationRunMaterialized {
|
||||
/// The admission request for an automation run: the packaged workflow
|
||||
/// version at the exact checked-out target, with no caller overrides.
|
||||
pub(crate) fn into_run_intent(self) -> RunIntent {
|
||||
RunIntent {
|
||||
workflow_version_id: self.workflow_version_id,
|
||||
target: RunTarget::Git(self.target),
|
||||
args: RunIntentArgs::default(),
|
||||
environment_id: None,
|
||||
parent_id: None,
|
||||
title: None,
|
||||
goal: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(thiserror::Error, Debug)]
|
||||
|
|
@ -48,25 +64,25 @@ pub(crate) enum RunMaterializeError {
|
|||
#[source]
|
||||
source: std::io::Error,
|
||||
},
|
||||
#[error("failed to resolve automation workflow")]
|
||||
#[error("automation workflow was not found")]
|
||||
WorkflowNotFound {
|
||||
#[source]
|
||||
source: anyhow::Error,
|
||||
source: WorkflowVersionCollectError,
|
||||
},
|
||||
#[error("failed to build run manifest")]
|
||||
Manifest {
|
||||
#[error("failed to package automation workflow versions")]
|
||||
Package {
|
||||
#[source]
|
||||
source: anyhow::Error,
|
||||
source: WorkflowVersionCollectError,
|
||||
},
|
||||
#[error("manifest build task failed")]
|
||||
ManifestTask {
|
||||
#[error("workflow-version packaging task failed")]
|
||||
PackageTask {
|
||||
#[source]
|
||||
source: task::JoinError,
|
||||
},
|
||||
#[error("failed to serialize materialized run manifest")]
|
||||
SerializeManifest {
|
||||
#[error("failed to store automation workflow versions")]
|
||||
VersionStore {
|
||||
#[source]
|
||||
source: serde_json::Error,
|
||||
source: WorkflowVersionStoreError,
|
||||
},
|
||||
#[error("failed to load GitHub credentials")]
|
||||
Credentials {
|
||||
|
|
@ -85,11 +101,11 @@ pub(crate) trait AutomationRunMaterializer: Send + Sync {
|
|||
|
||||
#[derive(Clone)]
|
||||
pub(crate) struct ProductionAutomationRunMaterializer {
|
||||
github_credentials: Option<fabro_github::GitHubCredentials>,
|
||||
github_api_base_url: String,
|
||||
http_client: Option<fabro_http::HttpClient>,
|
||||
environment_defaults: MergeMap<EnvironmentLayer>,
|
||||
repo_cache: Arc<GitRepoCache>,
|
||||
github_credentials: Option<fabro_github::GitHubCredentials>,
|
||||
github_api_base_url: String,
|
||||
http_client: Option<fabro_http::HttpClient>,
|
||||
repo_cache: Arc<GitRepoCache>,
|
||||
version_store: WorkflowVersionStore,
|
||||
}
|
||||
|
||||
impl ProductionAutomationRunMaterializer {
|
||||
|
|
@ -97,15 +113,15 @@ impl ProductionAutomationRunMaterializer {
|
|||
github_credentials: Option<fabro_github::GitHubCredentials>,
|
||||
github_api_base_url: String,
|
||||
http_client: Option<fabro_http::HttpClient>,
|
||||
environment_defaults: MergeMap<EnvironmentLayer>,
|
||||
repo_cache: Arc<GitRepoCache>,
|
||||
version_store: WorkflowVersionStore,
|
||||
) -> Self {
|
||||
Self {
|
||||
github_credentials,
|
||||
github_api_base_url,
|
||||
http_client,
|
||||
environment_defaults,
|
||||
repo_cache,
|
||||
version_store,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -161,101 +177,78 @@ impl AutomationRunMaterializer for ProductionAutomationRunMaterializer {
|
|||
let mut exact_target = input.target;
|
||||
exact_target.sha = Some(checked_out_sha);
|
||||
|
||||
let manifest_input = ManifestFromCheckoutInput {
|
||||
workflow: input.workflow,
|
||||
user_settings_path: input.user_settings_path,
|
||||
checkout_dir,
|
||||
target: exact_target,
|
||||
environment_defaults: self.environment_defaults.clone(),
|
||||
};
|
||||
task::spawn_blocking(move || build_manifest_from_checkout(manifest_input))
|
||||
.await
|
||||
.map_err(|source| RunMaterializeError::ManifestTask { source })?
|
||||
let workflow = PathBuf::from(input.workflow);
|
||||
let closure = task::spawn_blocking(move || {
|
||||
fabro_manifest::collect_workflow_versions(&workflow, &checkout_dir)
|
||||
.map_err(package_error)
|
||||
})
|
||||
.await
|
||||
.map_err(|source| RunMaterializeError::PackageTask { source })??;
|
||||
|
||||
// Versions arrive dependency-first, and the store derives the same
|
||||
// content hash the collector did, so the root ID is known up front.
|
||||
for (expected, version) in closure.versions() {
|
||||
let stored = self
|
||||
.version_store
|
||||
.put(version)
|
||||
.await
|
||||
.map_err(|source| RunMaterializeError::VersionStore { source })?;
|
||||
debug_assert_eq!(
|
||||
stored, expected,
|
||||
"store and collector disagree on version ID"
|
||||
);
|
||||
}
|
||||
Ok(AutomationRunMaterialized {
|
||||
workflow_version_id: closure.root_id(),
|
||||
target: exact_target,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub(crate) struct ManifestFromCheckoutInput {
|
||||
workflow: String,
|
||||
user_settings_path: PathBuf,
|
||||
checkout_dir: PathBuf,
|
||||
target: GitRunTarget,
|
||||
environment_defaults: MergeMap<EnvironmentLayer>,
|
||||
}
|
||||
|
||||
fn build_manifest_from_checkout(
|
||||
args: ManifestFromCheckoutInput,
|
||||
) -> Result<AutomationRunMaterialized, RunMaterializeError> {
|
||||
let ManifestFromCheckoutInput {
|
||||
workflow,
|
||||
user_settings_path,
|
||||
checkout_dir,
|
||||
target,
|
||||
environment_defaults,
|
||||
} = args;
|
||||
// Re-validating the exact target (now carrying the checked-out SHA) yields
|
||||
// the same `GitContext` projection the run-intent path uses.
|
||||
let validated = RunTarget::Git(target)
|
||||
.validate()
|
||||
.map_err(|source| RunMaterializeError::InvalidTarget { source })?;
|
||||
let RunTarget::Git(target) = validated.target else {
|
||||
unreachable!("validating a Git target yields a Git target");
|
||||
};
|
||||
let built = fabro_manifest::build_run_manifest(ManifestBuildInput {
|
||||
workflow: workflow.into(),
|
||||
cwd: checkout_dir,
|
||||
user_settings_path: Some(user_settings_path),
|
||||
environment_defaults,
|
||||
..ManifestBuildInput::default()
|
||||
})
|
||||
.map_err(manifest_build_error)?;
|
||||
|
||||
let mut manifest = built.manifest;
|
||||
manifest.git = validated.git;
|
||||
let submitted_manifest_bytes = serde_json::to_vec(&manifest)
|
||||
.map_err(|source| RunMaterializeError::SerializeManifest { source })?;
|
||||
Ok(AutomationRunMaterialized {
|
||||
manifest,
|
||||
submitted_manifest_bytes,
|
||||
target,
|
||||
})
|
||||
}
|
||||
|
||||
fn manifest_build_error(error: anyhow::Error) -> RunMaterializeError {
|
||||
if error.chain().any(|source| {
|
||||
source
|
||||
.downcast_ref::<fabro_config::Error>()
|
||||
.is_some_and(|err| matches!(err, fabro_config::Error::WorkflowNotFound(_)))
|
||||
}) {
|
||||
fn package_error(error: WorkflowVersionCollectError) -> RunMaterializeError {
|
||||
if matches!(&error, WorkflowVersionCollectError::WorkflowNotFound { .. }) {
|
||||
RunMaterializeError::WorkflowNotFound { source: error }
|
||||
} else {
|
||||
RunMaterializeError::Manifest { source: error }
|
||||
RunMaterializeError::Package { source: error }
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(any(test, feature = "test-support"))]
|
||||
#[derive(Clone)]
|
||||
pub struct TestAutomationRunMaterializer {
|
||||
inner: std::sync::Arc<std::sync::Mutex<TestAutomationRunMaterializerState>>,
|
||||
inner: std::sync::Arc<std::sync::Mutex<TestAutomationRunMaterializerState>>,
|
||||
version_store: Option<WorkflowVersionStore>,
|
||||
}
|
||||
|
||||
#[cfg(any(test, feature = "test-support"))]
|
||||
struct TestAutomationRunMaterializerState {
|
||||
captured_inputs: Vec<AutomationRunMaterializeInput>,
|
||||
response: Result<Box<AutomationRunMaterialized>, TargetValidationError>,
|
||||
response: Result<Box<TestMaterializedWorkflow>, TargetValidationError>,
|
||||
}
|
||||
|
||||
#[cfg(any(test, feature = "test-support"))]
|
||||
#[derive(Clone)]
|
||||
struct TestMaterializedWorkflow {
|
||||
version: fabro_workflow_version::ValidatedWorkflowVersion,
|
||||
target: GitRunTarget,
|
||||
store: bool,
|
||||
}
|
||||
|
||||
#[cfg(any(test, feature = "test-support"))]
|
||||
impl TestAutomationRunMaterializer {
|
||||
pub fn succeed(
|
||||
manifest: RunManifest,
|
||||
submitted_manifest_bytes: Vec<u8>,
|
||||
target: GitRunTarget,
|
||||
) -> Self {
|
||||
Self::new(Ok(Box::new(AutomationRunMaterialized {
|
||||
manifest,
|
||||
submitted_manifest_bytes,
|
||||
pub fn succeed(target: GitRunTarget) -> Self {
|
||||
Self::new(Ok(Box::new(TestMaterializedWorkflow {
|
||||
version: test_workflow_version(),
|
||||
target,
|
||||
store: true,
|
||||
})))
|
||||
}
|
||||
|
||||
pub fn return_unstored_version(target: GitRunTarget) -> Self {
|
||||
Self::new(Ok(Box::new(TestMaterializedWorkflow {
|
||||
version: test_workflow_version(),
|
||||
target,
|
||||
store: false,
|
||||
})))
|
||||
}
|
||||
|
||||
|
|
@ -263,12 +256,15 @@ impl TestAutomationRunMaterializer {
|
|||
Self::new(Err(TargetValidationError::Repository))
|
||||
}
|
||||
|
||||
fn new(response: Result<Box<AutomationRunMaterialized>, TargetValidationError>) -> Self {
|
||||
fn new(response: Result<Box<TestMaterializedWorkflow>, TargetValidationError>) -> Self {
|
||||
Self {
|
||||
inner: std::sync::Arc::new(std::sync::Mutex::new(TestAutomationRunMaterializerState {
|
||||
captured_inputs: Vec::new(),
|
||||
response,
|
||||
})),
|
||||
inner: std::sync::Arc::new(std::sync::Mutex::new(
|
||||
TestAutomationRunMaterializerState {
|
||||
captured_inputs: Vec::new(),
|
||||
response,
|
||||
},
|
||||
)),
|
||||
version_store: None,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -280,11 +276,35 @@ impl TestAutomationRunMaterializer {
|
|||
.clone()
|
||||
}
|
||||
|
||||
pub(crate) fn into_materializer(self) -> std::sync::Arc<dyn AutomationRunMaterializer> {
|
||||
pub(crate) fn into_materializer(
|
||||
mut self,
|
||||
version_store: WorkflowVersionStore,
|
||||
) -> std::sync::Arc<dyn AutomationRunMaterializer> {
|
||||
self.version_store = Some(version_store);
|
||||
std::sync::Arc::new(self)
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(any(test, feature = "test-support"))]
|
||||
fn test_workflow_version() -> fabro_workflow_version::ValidatedWorkflowVersion {
|
||||
use std::collections::BTreeMap;
|
||||
|
||||
let entrypoint = fabro_types::WorkflowPath::new("workflow.fabro")
|
||||
.expect("test workflow entrypoint should be valid");
|
||||
let version = fabro_types::WorkflowVersion::new(
|
||||
entrypoint.clone(),
|
||||
BTreeMap::from([(
|
||||
entrypoint,
|
||||
"digraph Test { graph [goal=\"Test\"] start [shape=Mdiamond] exit [shape=Msquare] start -> exit }"
|
||||
.to_string(),
|
||||
)]),
|
||||
BTreeMap::new(),
|
||||
)
|
||||
.expect("test workflow version should have a valid shape");
|
||||
fabro_workflow_version::ValidatedWorkflowVersion::new(version)
|
||||
.expect("test workflow version should validate")
|
||||
}
|
||||
|
||||
#[cfg(any(test, feature = "test-support"))]
|
||||
#[async_trait]
|
||||
impl AutomationRunMaterializer for TestAutomationRunMaterializer {
|
||||
|
|
@ -292,16 +312,36 @@ impl AutomationRunMaterializer for TestAutomationRunMaterializer {
|
|||
&self,
|
||||
input: AutomationRunMaterializeInput,
|
||||
) -> Result<AutomationRunMaterialized, RunMaterializeError> {
|
||||
let mut guard = self
|
||||
.inner
|
||||
.lock()
|
||||
.expect("test automation materializer lock poisoned");
|
||||
guard.captured_inputs.push(input);
|
||||
guard
|
||||
.response
|
||||
.clone()
|
||||
.map(|materialized| *materialized)
|
||||
.map_err(|source| RunMaterializeError::InvalidTarget { source })
|
||||
let response = {
|
||||
let mut guard = self
|
||||
.inner
|
||||
.lock()
|
||||
.expect("test automation materializer lock poisoned");
|
||||
guard.captured_inputs.push(input);
|
||||
guard.response.clone()
|
||||
};
|
||||
let materialized =
|
||||
*response.map_err(|source| RunMaterializeError::InvalidTarget { source })?;
|
||||
let store = self
|
||||
.version_store
|
||||
.as_ref()
|
||||
.expect("test materializer must be attached to a version store");
|
||||
let workflow_version_id = if materialized.store {
|
||||
store
|
||||
.put(&materialized.version)
|
||||
.await
|
||||
.map_err(|source| RunMaterializeError::VersionStore { source })?
|
||||
} else {
|
||||
materialized
|
||||
.version
|
||||
.version()
|
||||
.id()
|
||||
.expect("validated test workflow version should serialize canonically")
|
||||
};
|
||||
Ok(AutomationRunMaterialized {
|
||||
workflow_version_id,
|
||||
target: materialized.target,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -312,86 +352,57 @@ mod tests {
|
|||
reason = "Materializer unit tests write small temporary workflow fixtures synchronously."
|
||||
)]
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::fs;
|
||||
use std::path::Path;
|
||||
use std::time::Duration;
|
||||
|
||||
use fabro_types::DirtyStatus;
|
||||
use object_store::memory::InMemory;
|
||||
use tempfile::TempDir;
|
||||
|
||||
use super::*;
|
||||
|
||||
fn test_environment_defaults() -> MergeMap<EnvironmentLayer> {
|
||||
MergeMap::from(HashMap::from([("default".to_string(), EnvironmentLayer {
|
||||
provider: Some("local".to_string()),
|
||||
..EnvironmentLayer::default()
|
||||
})]))
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn manifest_builder_uses_checkout_for_workflow_and_separate_git_context() {
|
||||
#[tokio::test]
|
||||
async fn collected_closure_stores_dependency_first_and_idempotently() {
|
||||
let temp = TempDir::new().unwrap();
|
||||
let checkout = temp.path().join("checkout");
|
||||
let workflow_dir = checkout.join(".fabro/workflows/demo");
|
||||
let workflow_dir = checkout.join(".fabro/workflows/root");
|
||||
fs::create_dir_all(&workflow_dir).unwrap();
|
||||
fs::write(checkout.join(".fabro/project.toml"), "_version = 1\n").unwrap();
|
||||
fs::write(
|
||||
workflow_dir.join("workflow.fabro"),
|
||||
r#"digraph Demo { graph [goal="Ship automation"] start [shape=Mdiamond] exit [shape=Msquare] start -> exit }"#,
|
||||
)
|
||||
.unwrap();
|
||||
fs::write(
|
||||
workflow_dir.join("workflow.toml"),
|
||||
"_version = 1\n[workflow]\ngraph = \"workflow.fabro\"\n",
|
||||
)
|
||||
.unwrap();
|
||||
let user_settings_path = temp.path().join("settings.toml");
|
||||
fs::write(&user_settings_path, "_version = 1\n").unwrap();
|
||||
let sha = "0123456789abcdef0123456789abcdef01234567".to_string();
|
||||
fs::write(
|
||||
workflow_dir.join("workflow.fabro"),
|
||||
r#"digraph Root { child [stack.child_workflow="../child/workflow.fabro"] }"#,
|
||||
)
|
||||
.unwrap();
|
||||
let child_dir = checkout.join(".fabro/workflows/child");
|
||||
fs::create_dir_all(&child_dir).unwrap();
|
||||
fs::write(child_dir.join("workflow.fabro"), "digraph Child {}").unwrap();
|
||||
|
||||
let materialized = build_manifest_from_checkout(ManifestFromCheckoutInput {
|
||||
workflow: "demo".to_string(),
|
||||
user_settings_path: user_settings_path.clone(),
|
||||
checkout_dir: checkout.clone(),
|
||||
target: GitRunTarget {
|
||||
repo: "workspace-org/app".to_string(),
|
||||
branch: "release".to_string(),
|
||||
tag: Some("v1".to_string()),
|
||||
sha: Some(sha.clone()),
|
||||
},
|
||||
environment_defaults: test_environment_defaults(),
|
||||
})
|
||||
.expect("manifest should build from checkout");
|
||||
let closure =
|
||||
fabro_manifest::collect_workflow_versions(Path::new("root"), &checkout).unwrap();
|
||||
let database = fabro_store::test_support::test_database(
|
||||
Arc::new(InMemory::new()),
|
||||
"",
|
||||
Duration::from_millis(1),
|
||||
None,
|
||||
);
|
||||
let store = WorkflowVersionStore::new(database.blobs());
|
||||
|
||||
assert_eq!(materialized.manifest.cwd, checkout.display().to_string());
|
||||
assert_eq!(
|
||||
materialized.manifest.target.path,
|
||||
".fabro/workflows/demo/workflow.fabro"
|
||||
);
|
||||
assert!(
|
||||
materialized
|
||||
.manifest
|
||||
.configs
|
||||
.iter()
|
||||
.any(|config| config.path.as_deref() == Some(user_settings_path.to_str().unwrap()))
|
||||
);
|
||||
let git = materialized
|
||||
.manifest
|
||||
.git
|
||||
.as_ref()
|
||||
.expect("git context should be set");
|
||||
assert_eq!(git.origin_url, "https://github.com/workspace-org/app");
|
||||
assert_eq!(git.branch, "release");
|
||||
assert_eq!(git.sha.as_deref(), Some(sha.as_str()));
|
||||
assert_eq!(git.dirty, DirtyStatus::Clean);
|
||||
assert_eq!(materialized.target.tag.as_deref(), Some("v1"));
|
||||
assert_eq!(materialized.target.sha.as_deref(), Some(sha.as_str()));
|
||||
let submitted_manifest: serde_json::Value =
|
||||
serde_json::from_slice(&materialized.submitted_manifest_bytes)
|
||||
.expect("submitted bytes should be a manifest");
|
||||
assert!(submitted_manifest.get("run_id").is_none());
|
||||
assert_eq!(
|
||||
submitted_manifest,
|
||||
serde_json::to_value(&materialized.manifest).unwrap()
|
||||
);
|
||||
for _ in 0..2 {
|
||||
for (expected, version) in closure.versions() {
|
||||
assert_eq!(store.put(version).await.unwrap(), expected);
|
||||
}
|
||||
}
|
||||
let loaded = store
|
||||
.get_closure(&closure.root_id())
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
assert_eq!(loaded.root_id(), closure.root_id());
|
||||
assert_eq!(loaded.versions().count(), 2);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -303,8 +303,7 @@ fn mount_version(
|
|||
.get(version.entrypoint())
|
||||
.cloned()
|
||||
.expect("validated workflow versions contain their entrypoint file");
|
||||
let config_local = WorkflowPath::new("workflow.toml")
|
||||
.expect("the static workflow config path should be valid");
|
||||
let config_local = version.config_path();
|
||||
let config_path = version.files().get(&config_local).map(|source| {
|
||||
rebase_path(
|
||||
version.entrypoint(),
|
||||
|
|
@ -564,10 +563,10 @@ mod tests {
|
|||
"digraph Root { child [stack.child_workflow=\"../deps/run.fabro\"] }",
|
||||
),
|
||||
(
|
||||
"workflow.toml",
|
||||
"flows/workflow.toml",
|
||||
"_version = 1\n[run.goal]\nfile = \"goal.md\"\n",
|
||||
),
|
||||
("goal.md", "Ship {{ vars.owner }}"),
|
||||
("flows/goal.md", "Ship {{ vars.owner }}"),
|
||||
],
|
||||
[("deps/run.fabro", child_id)],
|
||||
);
|
||||
|
|
|
|||
|
|
@ -1210,8 +1210,8 @@ impl AppState {
|
|||
credentials,
|
||||
self.github_api_base_url.clone(),
|
||||
self.http_client.clone(),
|
||||
(*self.stores.environments.catalog_layer()).clone(),
|
||||
Arc::clone(&self.automation_repo_cache),
|
||||
fabro_workflow_version::WorkflowVersionStore::new(self.store_ref().blobs()),
|
||||
)
|
||||
.materialize(input)
|
||||
.await
|
||||
|
|
|
|||
|
|
@ -8,7 +8,7 @@ use croner::errors::CronError;
|
|||
use fabro_automation::{
|
||||
Automation, AutomationId, AutomationRevision, AutomationTriggerId, parse_schedule_expression,
|
||||
};
|
||||
use fabro_types::{AutomationRef, Principal, RunId, RunTarget, SystemActorKind};
|
||||
use fabro_types::{AutomationRef, Principal, RunId, SystemActorKind};
|
||||
use tokio::time::sleep;
|
||||
use tracing::{Instrument, error, info, info_span, warn};
|
||||
|
||||
|
|
@ -242,7 +242,6 @@ async fn fire_scheduled_automation_run(
|
|||
target,
|
||||
workflow: automation.workflow.clone(),
|
||||
run_id,
|
||||
user_settings_path: state.active_config_path().to_path_buf(),
|
||||
temp_root: state.automation_temp_root(),
|
||||
})
|
||||
.await
|
||||
|
|
@ -258,7 +257,6 @@ async fn fire_scheduled_automation_run(
|
|||
}
|
||||
};
|
||||
|
||||
let explicit_title_supplied = materialized.manifest.title.is_some();
|
||||
let actor = Principal::System {
|
||||
system_kind: SystemActorKind::Engine,
|
||||
};
|
||||
|
|
@ -267,19 +265,16 @@ async fn fire_scheduled_automation_run(
|
|||
name: Some(automation.name.clone()),
|
||||
trigger_id: Some(trigger_id.to_string()),
|
||||
};
|
||||
// `create_run_from_manifest` produces a large future; box it to keep our
|
||||
// RunIntent admission produces a large future; box it to keep our
|
||||
// stack frame small (matches handler/automations.rs).
|
||||
let response = Box::pin(handler::runs::create_run_from_manifest(
|
||||
let response = Box::pin(handler::runs::create_run_from_intent(
|
||||
Arc::clone(&state),
|
||||
handler::runs::CreateRunFromManifestRequest {
|
||||
manifest: materialized.manifest,
|
||||
submitted_manifest_bytes: materialized.submitted_manifest_bytes,
|
||||
handler::runs::CreateRunFromIntentRequest {
|
||||
intent: materialized.into_run_intent(),
|
||||
explicit_run_id: Some(run_id),
|
||||
explicit_title_supplied,
|
||||
actor: actor.clone(),
|
||||
headers: HeaderMap::new(),
|
||||
automation: Some(automation_ref),
|
||||
target: Some(RunTarget::Git(materialized.target)),
|
||||
actor: actor.clone(),
|
||||
headers: HeaderMap::new(),
|
||||
automation: Some(automation_ref),
|
||||
},
|
||||
))
|
||||
.await;
|
||||
|
|
@ -343,12 +338,10 @@ fn run_due_schedules_once<'a>(
|
|||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use fabro_api::types::RunManifest;
|
||||
use fabro_automation::{AutomationDraft, AutomationTrigger, ScheduleTrigger};
|
||||
use fabro_static::EnvVars;
|
||||
use fabro_store::ListRunsQuery;
|
||||
use fabro_types::{GitRunTarget, RunStatus};
|
||||
use serde_json::json;
|
||||
use fabro_types::{GitRunTarget, RunStatus, RunTarget};
|
||||
|
||||
use super::*;
|
||||
use crate::test_support::{TestAppStateBuilder, TestAutomationRunMaterializer};
|
||||
|
|
@ -412,35 +405,10 @@ mod tests {
|
|||
.expect("test automation should be created")
|
||||
}
|
||||
|
||||
fn minimal_manifest() -> RunManifest {
|
||||
serde_json::from_value(json!({
|
||||
"version": 1,
|
||||
"cwd": "/tmp",
|
||||
"target": {
|
||||
"path": "workflow.fabro",
|
||||
},
|
||||
"workflows": {
|
||||
"workflow.fabro": {
|
||||
"source": r#"digraph Test {
|
||||
graph [goal="Test"]
|
||||
start [shape=Mdiamond]
|
||||
exit [shape=Msquare]
|
||||
start -> exit
|
||||
}"#,
|
||||
"files": {},
|
||||
},
|
||||
},
|
||||
}))
|
||||
.expect("minimal manifest should deserialize")
|
||||
}
|
||||
|
||||
fn succeeding_materializer() -> TestAutomationRunMaterializer {
|
||||
let manifest = minimal_manifest();
|
||||
let submitted_manifest_bytes =
|
||||
serde_json::to_vec(&manifest).expect("manifest should serialize");
|
||||
let mut exact_target = git_target();
|
||||
exact_target.sha = Some("0123456789abcdef0123456789abcdef01234567".to_string());
|
||||
TestAutomationRunMaterializer::succeed(manifest, submitted_manifest_bytes, exact_target)
|
||||
TestAutomationRunMaterializer::succeed(exact_target)
|
||||
}
|
||||
|
||||
fn test_state_with_materializer(materializer: TestAutomationRunMaterializer) -> Arc<AppState> {
|
||||
|
|
@ -616,6 +584,28 @@ mod tests {
|
|||
assert_eq!(automation_ref.name.as_deref(), Some("Nightly"));
|
||||
assert_eq!(automation_ref.trigger_id.as_deref(), Some("schedule"));
|
||||
let run_id = runs[0].id;
|
||||
let run_store = state.stores.runs.open_run_reader(&run_id).await.unwrap();
|
||||
let projection = run_store.state().await.unwrap();
|
||||
assert!(projection.spec.workflow_version_id.is_some());
|
||||
assert_eq!(
|
||||
projection.spec.target,
|
||||
Some(RunTarget::Git(fabro_types::GitRunTarget {
|
||||
repo: "fabro-sh/fabro".to_string(),
|
||||
branch: "main".to_string(),
|
||||
tag: None,
|
||||
sha: Some("0123456789abcdef0123456789abcdef01234567".to_string()),
|
||||
}))
|
||||
);
|
||||
assert_eq!(
|
||||
run_store
|
||||
.list_events()
|
||||
.await
|
||||
.unwrap()
|
||||
.iter()
|
||||
.filter(|event| event.event.event_name() == "run.start_requested")
|
||||
.count(),
|
||||
1
|
||||
);
|
||||
assert!(matches!(
|
||||
state
|
||||
.runs
|
||||
|
|
|
|||
|
|
@ -6,7 +6,7 @@ use fabro_automation::{
|
|||
Automation, AutomationDraft, AutomationId, AutomationReplace, AutomationStoreError,
|
||||
};
|
||||
use fabro_store::{RunSummaryListQuery, RunSummaryVisibility};
|
||||
use fabro_types::{AutomationRef, RunId, RunTarget};
|
||||
use fabro_types::{AutomationRef, RunId};
|
||||
use fabro_util::error as error_util;
|
||||
use serde::Serialize;
|
||||
|
||||
|
|
@ -132,7 +132,6 @@ async fn create_automation_run(
|
|||
target,
|
||||
workflow: automation.workflow.clone(),
|
||||
run_id,
|
||||
user_settings_path: state.active_config_path().to_path_buf(),
|
||||
temp_root: state.automation_temp_root(),
|
||||
})
|
||||
.await
|
||||
|
|
@ -143,24 +142,20 @@ async fn create_automation_run(
|
|||
return ApiError::new(StatusCode::UNPROCESSABLE_ENTITY, message).into_response();
|
||||
}
|
||||
};
|
||||
let explicit_title_supplied = materialized.manifest.title.is_some();
|
||||
let automation_ref = AutomationRef {
|
||||
id: automation.id.to_string(),
|
||||
name: Some(automation.name.clone()),
|
||||
trigger_id: Some(api_trigger_id),
|
||||
};
|
||||
|
||||
let response = Box::pin(runs::create_run_from_manifest(
|
||||
let response = Box::pin(runs::create_run_from_intent(
|
||||
Arc::clone(&state),
|
||||
runs::CreateRunFromManifestRequest {
|
||||
manifest: materialized.manifest,
|
||||
submitted_manifest_bytes: materialized.submitted_manifest_bytes,
|
||||
runs::CreateRunFromIntentRequest {
|
||||
intent: materialized.into_run_intent(),
|
||||
explicit_run_id: Some(run_id),
|
||||
explicit_title_supplied,
|
||||
actor: actor.clone(),
|
||||
headers,
|
||||
automation: Some(automation_ref),
|
||||
target: Some(RunTarget::Git(materialized.target)),
|
||||
},
|
||||
))
|
||||
.await;
|
||||
|
|
|
|||
|
|
@ -541,7 +541,14 @@ async fn create_run(
|
|||
// round-trip would silently collapse duplicate keys to last-key-wins.
|
||||
let intent_error = match serde_json::from_slice::<RunIntent>(&body) {
|
||||
Ok(intent) => {
|
||||
return Box::pin(create_run_from_intent(state, intent, actor, headers)).await;
|
||||
return Box::pin(create_run_from_intent(state, CreateRunFromIntentRequest {
|
||||
intent,
|
||||
explicit_run_id: None,
|
||||
actor,
|
||||
headers,
|
||||
automation: None,
|
||||
}))
|
||||
.await;
|
||||
}
|
||||
Err(err) => err,
|
||||
};
|
||||
|
|
@ -604,12 +611,28 @@ fn create_run_parse_error(
|
|||
ApiError::bad_request(manifest_error.to_string()).into_response()
|
||||
}
|
||||
|
||||
async fn create_run_from_intent(
|
||||
pub(crate) struct CreateRunFromIntentRequest {
|
||||
pub(crate) intent: RunIntent,
|
||||
/// Run ID preallocated by server-side automation code, never supplied by
|
||||
/// an HTTP create body.
|
||||
pub(crate) explicit_run_id: Option<RunId>,
|
||||
pub(crate) actor: Principal,
|
||||
pub(crate) headers: HeaderMap,
|
||||
pub(crate) automation: Option<AutomationRef>,
|
||||
}
|
||||
|
||||
pub(crate) async fn create_run_from_intent(
|
||||
state: Arc<AppState>,
|
||||
intent: RunIntent,
|
||||
actor: Principal,
|
||||
headers: HeaderMap,
|
||||
request: CreateRunFromIntentRequest,
|
||||
) -> Response {
|
||||
let CreateRunFromIntentRequest {
|
||||
intent,
|
||||
explicit_run_id,
|
||||
actor,
|
||||
headers,
|
||||
automation,
|
||||
} = request;
|
||||
let explicit_title_supplied = intent.title.is_some();
|
||||
// Validate the pure, in-memory request facts before paying for
|
||||
// blob-store reads and closure lowering.
|
||||
let ValidatedRunTarget { target, git } = match intent.target.validate() {
|
||||
|
|
@ -713,7 +736,7 @@ async fn create_run_from_intent(
|
|||
cli_overrides: None,
|
||||
input_overrides,
|
||||
inline_goal_override: intent.goal,
|
||||
run_id: None,
|
||||
run_id: explicit_run_id,
|
||||
title,
|
||||
parent_id: intent.parent_id,
|
||||
// Target identity and its Git projection are attached after provider
|
||||
|
|
@ -726,7 +749,7 @@ async fn create_run_from_intent(
|
|||
provenance: run_provenance(&headers, &actor),
|
||||
web_url: None,
|
||||
submitted_manifest_bytes: None,
|
||||
automation: None,
|
||||
automation,
|
||||
};
|
||||
let normalized = match run_compiler::normalize_source(raw_compiler_input) {
|
||||
Ok(normalized) => normalized,
|
||||
|
|
@ -764,7 +787,7 @@ async fn create_run_from_intent(
|
|||
finalize_created_run(
|
||||
state,
|
||||
prepared,
|
||||
intent.title.is_some(),
|
||||
explicit_title_supplied,
|
||||
entrypoint,
|
||||
CreatedRunErrorStyle::Intent,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -4502,6 +4502,78 @@ async fn create_run_from_manifest_helper_persists_automation_metadata_and_exact_
|
|||
assert_eq!(run_store.state().await.unwrap().spec.target, Some(target));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn create_run_from_intent_helper_persists_automation_version_and_exact_target() {
|
||||
let state = TestAppStateBuilder::new()
|
||||
.env_lookup(|_| None)
|
||||
.vault_entries([(EnvVars::OPENAI_API_KEY, "test-openai-api-key")])
|
||||
.build();
|
||||
let workflow_version_id = store_workflow_version(&state, MINIMAL_DOT, None).await;
|
||||
let run_id = RunId::new();
|
||||
let automation = fabro_types::AutomationRef {
|
||||
id: "nightly".to_string(),
|
||||
name: Some("Nightly".to_string()),
|
||||
trigger_id: Some("schedule".to_string()),
|
||||
};
|
||||
let target = RunTarget::Git(GitRunTarget {
|
||||
repo: "fabro-sh/fabro".to_string(),
|
||||
branch: "main".to_string(),
|
||||
tag: Some("v1.2.3".to_string()),
|
||||
sha: Some("0123456789abcdef0123456789abcdef01234567".to_string()),
|
||||
});
|
||||
|
||||
let response = Box::pin(handler::runs::create_run_from_intent(
|
||||
Arc::clone(&state),
|
||||
handler::runs::CreateRunFromIntentRequest {
|
||||
intent: fabro_api::types::RunIntent {
|
||||
workflow_version_id,
|
||||
target: target.clone(),
|
||||
args: fabro_api::types::RunIntentArgs::default(),
|
||||
environment_id: None,
|
||||
parent_id: None,
|
||||
title: None,
|
||||
goal: None,
|
||||
},
|
||||
explicit_run_id: Some(run_id),
|
||||
actor: Principal::System {
|
||||
system_kind: SystemActorKind::Engine,
|
||||
},
|
||||
headers: HeaderMap::new(),
|
||||
automation: Some(automation.clone()),
|
||||
},
|
||||
))
|
||||
.await;
|
||||
|
||||
let body = response_json!(response, StatusCode::CREATED).await;
|
||||
assert_eq!(body["automation"]["id"], automation.id);
|
||||
let summary = state
|
||||
.stores
|
||||
.runs
|
||||
.get_cached_summary(&run_id, Utc::now())
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
assert_eq!(summary.automation, Some(automation.clone()));
|
||||
let run_store = state.stores.runs.open_run_reader(&run_id).await.unwrap();
|
||||
let projection = run_store.state().await.unwrap();
|
||||
assert_eq!(
|
||||
projection.spec.workflow_version_id,
|
||||
Some(workflow_version_id)
|
||||
);
|
||||
assert_eq!(projection.spec.target, Some(target));
|
||||
assert_eq!(projection.spec.automation, Some(automation));
|
||||
assert_eq!(
|
||||
run_store
|
||||
.list_events()
|
||||
.await
|
||||
.unwrap()
|
||||
.iter()
|
||||
.map(|event| event.event.event_name())
|
||||
.collect::<Vec<_>>(),
|
||||
["run.created", "run.submitted"]
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn create_run_from_manifest_pins_compiled_and_persisted_behavior() {
|
||||
let state = TestAppStateBuilder::new()
|
||||
|
|
@ -4854,24 +4926,17 @@ async fn create_run_from_manifest_resolves_generated_id_after_variable_snapshot(
|
|||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn fake_automation_materializer_injection_captures_input_and_returns_manifest() {
|
||||
let materialized_manifest: RunManifest =
|
||||
serde_json::from_value(minimal_manifest_json(MINIMAL_DOT)).unwrap();
|
||||
let fake = TestAutomationRunMaterializer::succeed(
|
||||
materialized_manifest.clone(),
|
||||
b"{\"fake\":true}".to_vec(),
|
||||
GitRunTarget {
|
||||
repo: "fabro-sh/fabro".to_string(),
|
||||
branch: "main".to_string(),
|
||||
tag: None,
|
||||
sha: Some("0123456789abcdef0123456789abcdef01234567".to_string()),
|
||||
},
|
||||
);
|
||||
async fn fake_automation_materializer_injection_captures_input_and_returns_version() {
|
||||
let fake = TestAutomationRunMaterializer::succeed(GitRunTarget {
|
||||
repo: "fabro-sh/fabro".to_string(),
|
||||
branch: "main".to_string(),
|
||||
tag: None,
|
||||
sha: Some("0123456789abcdef0123456789abcdef01234567".to_string()),
|
||||
});
|
||||
let state = TestAppStateBuilder::new()
|
||||
.automation_materializer(fake.clone())
|
||||
.build();
|
||||
let run_id = RunId::new();
|
||||
let user_settings_path = PathBuf::from("/tmp/fabro/settings.toml");
|
||||
let temp_root = PathBuf::from("/tmp/fabro/automation");
|
||||
let target = GitRunTarget {
|
||||
repo: "fabro-sh/fabro".to_string(),
|
||||
|
|
@ -4886,24 +4951,22 @@ async fn fake_automation_materializer_injection_captures_input_and_returns_manif
|
|||
target: target.clone(),
|
||||
workflow: "demo".to_string(),
|
||||
run_id,
|
||||
user_settings_path: user_settings_path.clone(),
|
||||
temp_root: temp_root.clone(),
|
||||
})
|
||||
.await
|
||||
.expect("fake materializer should succeed");
|
||||
|
||||
assert_eq!(
|
||||
serde_json::to_value(&output.manifest).unwrap(),
|
||||
serde_json::to_value(&materialized_manifest).unwrap()
|
||||
);
|
||||
assert_eq!(output.submitted_manifest_bytes, b"{\"fake\":true}".to_vec());
|
||||
let stored = fabro_workflow_version::WorkflowVersionStore::new(state.store_ref().blobs())
|
||||
.get(&output.workflow_version_id)
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(stored.is_some());
|
||||
let captured = fake.captured_inputs();
|
||||
assert_eq!(captured.len(), 1);
|
||||
assert_eq!(captured[0].automation_id.as_str(), "nightly");
|
||||
assert_eq!(captured[0].target, target);
|
||||
assert_eq!(captured[0].workflow, "demo");
|
||||
assert_eq!(captured[0].run_id, run_id);
|
||||
assert_eq!(captured[0].user_settings_path, user_settings_path);
|
||||
assert_eq!(captured[0].temp_root, temp_root);
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -33,7 +33,6 @@ use tokio::runtime::Builder as TokioRuntimeBuilder;
|
|||
use tokio_util::sync::CancellationToken;
|
||||
use ulid::Ulid;
|
||||
|
||||
use crate::automation_materializer::AutomationRunMaterializer;
|
||||
pub use crate::automation_materializer::TestAutomationRunMaterializer;
|
||||
use crate::interp::process_env_var;
|
||||
use crate::jwt_auth::{AuthMode, ConfiguredAuth};
|
||||
|
|
@ -102,7 +101,7 @@ pub struct TestAppStateBuilder {
|
|||
default_environment_provider: Option<EnvironmentProvider>,
|
||||
env_lookup: EnvLookup,
|
||||
llm_catalog_settings: LlmCatalogSettings,
|
||||
automation_materializer: Option<Arc<dyn AutomationRunMaterializer>>,
|
||||
automation_materializer: Option<TestAutomationRunMaterializer>,
|
||||
#[cfg(test)]
|
||||
worker_runtime: Option<Arc<dyn WorkerRuntime>>,
|
||||
}
|
||||
|
|
@ -184,7 +183,7 @@ impl TestAppStateBuilder {
|
|||
}
|
||||
|
||||
pub fn automation_materializer(mut self, materializer: TestAutomationRunMaterializer) -> Self {
|
||||
self.automation_materializer = Some(materializer.into_materializer());
|
||||
self.automation_materializer = Some(materializer);
|
||||
self
|
||||
}
|
||||
|
||||
|
|
@ -282,6 +281,11 @@ impl TestAppStateBuilder {
|
|||
server::automation_dir_for_active_config(&active_config_path),
|
||||
)?;
|
||||
let preloaded_vault = test_secret_snapshot(db_pool.clone())?;
|
||||
let automation_materializer_override = self.automation_materializer.map(|materializer| {
|
||||
materializer.into_materializer(fabro_workflow_version::WorkflowVersionStore::new(
|
||||
store.blobs(),
|
||||
))
|
||||
});
|
||||
build_app_state(AppStateConfig {
|
||||
resolved_settings: resolved_runtime_settings_for_tests(
|
||||
self.server_settings,
|
||||
|
|
@ -307,7 +311,7 @@ impl TestAppStateBuilder {
|
|||
worker_control_bus: None,
|
||||
#[cfg(test)]
|
||||
worker_runtime: self.worker_runtime,
|
||||
automation_materializer_override: self.automation_materializer,
|
||||
automation_materializer_override,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -13,10 +13,7 @@ use serde_json::{Value, json};
|
|||
use sqlx::Row as _;
|
||||
use tower::ServiceExt;
|
||||
|
||||
use crate::helpers::{
|
||||
MINIMAL_DOT, api, checked_response, minimal_manifest_json, response_json, response_status,
|
||||
run_json,
|
||||
};
|
||||
use crate::helpers::{api, checked_response, response_json, response_status, run_json};
|
||||
|
||||
fn automation_body(id: &str, name: &str) -> Value {
|
||||
json!({
|
||||
|
|
@ -78,29 +75,26 @@ fn automation_app() -> (axum::Router, tempfile::TempDir, PathBuf) {
|
|||
}
|
||||
|
||||
fn automation_app_with_fake_materializer() -> (axum::Router, tempfile::TempDir, PathBuf) {
|
||||
automation_app_with_materializer(TestAutomationRunMaterializer::succeed(GitRunTarget {
|
||||
repo: "fabro-sh/fabro".to_string(),
|
||||
branch: "main".to_string(),
|
||||
tag: None,
|
||||
sha: Some("0123456789abcdef0123456789abcdef01234567".to_string()),
|
||||
}))
|
||||
}
|
||||
|
||||
fn automation_app_with_materializer(
|
||||
materializer: TestAutomationRunMaterializer,
|
||||
) -> (axum::Router, tempfile::TempDir, PathBuf) {
|
||||
let temp_dir = tempfile::tempdir().expect("automation test tempdir should be created");
|
||||
let active_config_path = temp_dir.path().join("settings.toml");
|
||||
let vault_path = temp_dir.path().join("secrets.json");
|
||||
let sqlite_path = Storage::new(temp_dir.path()).sqlite_path();
|
||||
let materialized_manifest: fabro_api::types::RunManifest =
|
||||
serde_json::from_value(minimal_manifest_json(MINIMAL_DOT))
|
||||
.expect("minimal run manifest fixture should deserialize");
|
||||
let submitted_manifest_bytes =
|
||||
serde_json::to_vec(&materialized_manifest).expect("minimal run manifest should serialize");
|
||||
let state = TestAppStateBuilder::new()
|
||||
.active_config_path(active_config_path)
|
||||
.vault_path(vault_path)
|
||||
.vault_entries([(EnvVars::OPENAI_API_KEY, "test-openai-api-key")])
|
||||
.automation_materializer(TestAutomationRunMaterializer::succeed(
|
||||
materialized_manifest,
|
||||
submitted_manifest_bytes,
|
||||
GitRunTarget {
|
||||
repo: "fabro-sh/fabro".to_string(),
|
||||
branch: "main".to_string(),
|
||||
tag: None,
|
||||
sha: Some("0123456789abcdef0123456789abcdef01234567".to_string()),
|
||||
},
|
||||
))
|
||||
.automation_materializer(materializer)
|
||||
.build();
|
||||
(build_test_router(state), temp_dir, sqlite_path)
|
||||
}
|
||||
|
|
@ -920,6 +914,25 @@ async fn successful_api_triggered_automation_run_persists_automation_metadata()
|
|||
assert_eq!(retrieved["automation"], created["automation"]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn api_triggered_automation_with_missing_version_does_not_create_or_start_a_run() {
|
||||
let materializer = TestAutomationRunMaterializer::return_unstored_version(GitRunTarget {
|
||||
repo: "fabro-sh/fabro".to_string(),
|
||||
branch: "main".to_string(),
|
||||
tag: None,
|
||||
sha: Some("0123456789abcdef0123456789abcdef01234567".to_string()),
|
||||
});
|
||||
let (app, _temp_dir, _automation_dir) = automation_app_with_materializer(materializer);
|
||||
create_automation(&app, "nightly", "Nightly").await;
|
||||
|
||||
let error = create_automation_run(&app, "nightly", StatusCode::NOT_FOUND).await;
|
||||
|
||||
assert_eq!(error["errors"][0]["code"], "workflow_version_not_found");
|
||||
let runs = list_automation_runs(&app, "/automations/nightly/runs").await;
|
||||
assert_eq!(runs["meta"]["total"], 0);
|
||||
assert_eq!(runs["data"], json!([]));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn automation_run_listing_includes_only_runs_for_that_automation() {
|
||||
let (app, _temp_dir, _automation_dir) = automation_app_with_fake_materializer();
|
||||
|
|
|
|||
|
|
@ -21,7 +21,9 @@ fabro-graphviz = { path = "../fabro-graphviz" }
|
|||
fabro-template = { path = "../../foundation/fabro-template" }
|
||||
fabro-types = { path = "../../foundation/fabro-types" }
|
||||
fabro-workflow = { path = "../fabro-workflow" }
|
||||
fabro-workflow-version = { path = "../fabro-workflow-version" }
|
||||
git2.workspace = true
|
||||
thiserror.workspace = true
|
||||
toml.workspace = true
|
||||
|
||||
[dev-dependencies]
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@
|
|||
)]
|
||||
|
||||
mod workflow_bundler;
|
||||
mod workflow_version_collector;
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::path::{Component, Path, PathBuf};
|
||||
|
|
@ -29,6 +30,9 @@ use fabro_workflow::git::{
|
|||
};
|
||||
|
||||
use crate::workflow_bundler::WorkflowBundler;
|
||||
pub use crate::workflow_version_collector::{
|
||||
CollectedWorkflowClosure, WorkflowVersionCollectError, collect_workflow_versions,
|
||||
};
|
||||
|
||||
#[derive(Debug, Default)]
|
||||
pub struct ManifestBuildInput {
|
||||
|
|
@ -126,13 +130,7 @@ pub fn build_sparse_run_overrides(input: RunOverrideInput<'_>) -> Option<RunLaye
|
|||
}
|
||||
|
||||
pub fn build_run_manifest(input: ManifestBuildInput) -> Result<BuiltManifest> {
|
||||
let root_location = WorkflowLocation::resolve(&input.workflow, &input.cwd)?;
|
||||
if root_location.toml.is_none() && !root_location.graph.is_file() {
|
||||
return Err(fabro_config::Error::WorkflowNotFound(
|
||||
root_location.graph.display().to_string(),
|
||||
)
|
||||
.into());
|
||||
}
|
||||
let root_location = resolve_existing_workflow_location(&input.workflow, &input.cwd)?;
|
||||
let project_config = discover_project_config(&root_location.dir)?;
|
||||
let project_config_source = project_config
|
||||
.as_ref()
|
||||
|
|
@ -396,6 +394,19 @@ fn push_manifest_branch_best_effort(
|
|||
let _ = push_branch_noninteractive(repo_path, "origin", branch);
|
||||
}
|
||||
|
||||
/// Resolve a workflow reference and reject it when neither its config nor
|
||||
/// its graph exists on disk.
|
||||
/// A missing workflow surfaces as `fabro_config::Error::WorkflowNotFound`.
|
||||
fn resolve_existing_workflow_location(workflow: &Path, cwd: &Path) -> Result<WorkflowLocation> {
|
||||
let location = WorkflowLocation::resolve(workflow, cwd)?;
|
||||
if location.toml.is_none() && !location.graph.is_file() {
|
||||
return Err(
|
||||
fabro_config::Error::WorkflowNotFound(location.graph.display().to_string()).into(),
|
||||
);
|
||||
}
|
||||
Ok(location)
|
||||
}
|
||||
|
||||
fn normalize_absolute_path(base_dir: &Path, reference: &str) -> Option<PathBuf> {
|
||||
let path = Path::new(reference);
|
||||
if path.is_absolute() || reference.starts_with('~') {
|
||||
|
|
|
|||
|
|
@ -1,11 +1,13 @@
|
|||
use std::collections::{HashMap, HashSet};
|
||||
use std::collections::{BTreeSet, HashMap, HashSet};
|
||||
use std::path::{Component, Path, PathBuf};
|
||||
use std::sync::Arc;
|
||||
|
||||
use anyhow::{Context as _, Result, anyhow};
|
||||
use fabro_api::types;
|
||||
use fabro_config::project::WorkflowLocation;
|
||||
use fabro_config::{EnvironmentDockerfileLayer, EnvironmentImageLayer, SettingsLayer};
|
||||
use fabro_config::{
|
||||
EnvironmentDockerfileLayer, EnvironmentImageLayer, RunGoalLayer, SettingsLayer,
|
||||
};
|
||||
use fabro_graphviz::parser;
|
||||
use fabro_template::{
|
||||
BundleTemplateStore, FilesystemTemplateStore, GraphPosition, GraphReference,
|
||||
|
|
@ -18,11 +20,22 @@ use fabro_types::graph::ReferenceKind;
|
|||
use crate::{manifest_path_from_absolute, normalize_absolute_path};
|
||||
|
||||
pub(super) struct WorkflowBundler<'a> {
|
||||
cwd: &'a Path,
|
||||
inputs: &'a HashMap<String, toml::Value>,
|
||||
template_store: FilesystemTemplateStore,
|
||||
workflows: HashMap<String, types::ManifestWorkflow>,
|
||||
cwd: &'a Path,
|
||||
inputs: &'a HashMap<String, toml::Value>,
|
||||
template_store: FilesystemTemplateStore,
|
||||
workflows: HashMap<String, CollectedWorkflowSource>,
|
||||
visited_workflows: HashSet<String>,
|
||||
workflow_version_projection: bool,
|
||||
}
|
||||
|
||||
pub(super) struct CollectedWorkflowSources {
|
||||
pub(super) root_key: String,
|
||||
pub(super) workflows: HashMap<String, CollectedWorkflowSource>,
|
||||
}
|
||||
|
||||
pub(super) struct CollectedWorkflowSource {
|
||||
pub(super) workflow: types::ManifestWorkflow,
|
||||
pub(super) dependency_keys: BTreeSet<String>,
|
||||
}
|
||||
|
||||
impl<'a> WorkflowBundler<'a> {
|
||||
|
|
@ -33,6 +46,7 @@ impl<'a> WorkflowBundler<'a> {
|
|||
template_store: FilesystemTemplateStore::new(cwd),
|
||||
workflows: HashMap::new(),
|
||||
visited_workflows: HashSet::new(),
|
||||
workflow_version_projection: false,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -48,11 +62,29 @@ impl<'a> WorkflowBundler<'a> {
|
|||
.workflows
|
||||
.remove(&root_key)
|
||||
.ok_or_else(|| anyhow!("root workflow missing from manifest bundle"))?;
|
||||
self.collect_config_dockerfile(config_path, source, &mut root.files)?;
|
||||
let entrypoint = ManifestPath::from_wire(&root_key)
|
||||
.ok_or_else(|| anyhow!("invalid root workflow path: {root_key}"))?;
|
||||
self.collect_config_files(config_path, source, &entrypoint, &mut root.workflow.files)?;
|
||||
self.workflows.insert(root_key, root);
|
||||
}
|
||||
|
||||
Ok(self.workflows)
|
||||
Ok(self
|
||||
.workflows
|
||||
.into_iter()
|
||||
.map(|(key, source)| (key, source.workflow))
|
||||
.collect())
|
||||
}
|
||||
|
||||
pub(super) fn collect_versions(
|
||||
mut self,
|
||||
root: &WorkflowLocation,
|
||||
) -> Result<CollectedWorkflowSources> {
|
||||
self.workflow_version_projection = true;
|
||||
let root_key = self.collect_workflow_location(root)?;
|
||||
Ok(CollectedWorkflowSources {
|
||||
root_key,
|
||||
workflows: self.workflows,
|
||||
})
|
||||
}
|
||||
|
||||
/// Collects the workflow at `location` and returns its manifest key.
|
||||
|
|
@ -77,28 +109,33 @@ impl<'a> WorkflowBundler<'a> {
|
|||
|
||||
let scan = WorkflowScanInput {
|
||||
absolute_dot_path: location.graph.clone(),
|
||||
dot_path,
|
||||
source: source.clone(),
|
||||
dot_path: dot_path.clone(),
|
||||
source: source.clone(),
|
||||
};
|
||||
let mut files = HashMap::new();
|
||||
let mut visited_imports = HashSet::new();
|
||||
let mut dependency_keys = BTreeSet::new();
|
||||
if let Some(config) = config.as_ref() {
|
||||
let config_path = ManifestPath::from_wire(&config.path)
|
||||
.ok_or_else(|| anyhow!("invalid manifest workflow config path: {}", config.path))?;
|
||||
self.collect_config_dockerfile(&config_path, &config.source, &mut files)?;
|
||||
self.collect_config_files(&config_path, &config.source, &dot_path, &mut files)?;
|
||||
}
|
||||
self.collect_workflow_files(
|
||||
&scan,
|
||||
&mut files,
|
||||
&mut visited_imports,
|
||||
&mut dependency_keys,
|
||||
GraphPosition::Entrypoint,
|
||||
)?;
|
||||
|
||||
self.workflows
|
||||
.insert(dot_key.clone(), types::ManifestWorkflow {
|
||||
config,
|
||||
files,
|
||||
source,
|
||||
.insert(dot_key.clone(), CollectedWorkflowSource {
|
||||
workflow: types::ManifestWorkflow {
|
||||
config,
|
||||
files,
|
||||
source,
|
||||
},
|
||||
dependency_keys,
|
||||
});
|
||||
|
||||
Ok(dot_key)
|
||||
|
|
@ -128,6 +165,7 @@ impl<'a> WorkflowBundler<'a> {
|
|||
workflow: &WorkflowScanInput,
|
||||
files: &mut HashMap<String, types::ManifestFileEntry>,
|
||||
visited_imports: &mut HashSet<String>,
|
||||
dependency_keys: &mut BTreeSet<String>,
|
||||
position: GraphPosition,
|
||||
) -> Result<()> {
|
||||
let graph = parser::parse(&workflow.source)
|
||||
|
|
@ -136,7 +174,11 @@ impl<'a> WorkflowBundler<'a> {
|
|||
.absolute_dot_path
|
||||
.parent()
|
||||
.unwrap_or_else(|| Path::new("."));
|
||||
let workflow_template_root = manifest_parent_or_dot(&workflow.dot_path)?;
|
||||
let workflow_template_root = if self.workflow_version_projection {
|
||||
workflow_package_root()
|
||||
} else {
|
||||
manifest_parent_or_dot(&workflow.dot_path)?
|
||||
};
|
||||
|
||||
// Imports and child workflows require a mutable borrow of self, so
|
||||
// collect them during the walk and recurse after the visitor returns.
|
||||
|
|
@ -224,12 +266,15 @@ impl<'a> WorkflowBundler<'a> {
|
|||
&imported_scan,
|
||||
files,
|
||||
visited_imports,
|
||||
dependency_keys,
|
||||
GraphPosition::Imported,
|
||||
)?;
|
||||
}
|
||||
}
|
||||
for child in children {
|
||||
self.collect_workflow_entry(Path::new(child), workflow_base_dir)?;
|
||||
let dependency_key =
|
||||
self.collect_workflow_entry(Path::new(child), workflow_base_dir)?;
|
||||
dependency_keys.insert(dependency_key);
|
||||
}
|
||||
|
||||
Ok(())
|
||||
|
|
@ -320,10 +365,11 @@ impl<'a> WorkflowBundler<'a> {
|
|||
Ok(())
|
||||
}
|
||||
|
||||
fn collect_config_dockerfile(
|
||||
fn collect_config_files(
|
||||
&self,
|
||||
config_path: &ManifestPath,
|
||||
source: &str,
|
||||
entrypoint: &ManifestPath,
|
||||
files: &mut HashMap<String, types::ManifestFileEntry>,
|
||||
) -> Result<()> {
|
||||
let layer = source
|
||||
|
|
@ -337,9 +383,49 @@ impl<'a> WorkflowBundler<'a> {
|
|||
for image in layer.environment_images() {
|
||||
self.collect_environment_dockerfile(files, base_dir, config_path, image)?;
|
||||
}
|
||||
if self.workflow_version_projection {
|
||||
self.collect_config_goal_files(files, base_dir, config_path, entrypoint, &layer)?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn collect_config_goal_files(
|
||||
&self,
|
||||
files: &mut HashMap<String, types::ManifestFileEntry>,
|
||||
base_dir: &Path,
|
||||
config_path: &ManifestPath,
|
||||
entrypoint: &ManifestPath,
|
||||
layer: &SettingsLayer,
|
||||
) -> Result<()> {
|
||||
let Some(goal) = layer.run.as_ref().and_then(|run| run.goal.as_ref()) else {
|
||||
return Ok(());
|
||||
};
|
||||
let content = match goal {
|
||||
RunGoalLayer::Inline(goal) => goal.as_source(),
|
||||
RunGoalLayer::File { file } => {
|
||||
let reference = file.as_source();
|
||||
let bundled = self.collect_bundled_file(
|
||||
files,
|
||||
base_dir,
|
||||
&reference,
|
||||
types::ManifestFileRefType::FileInline,
|
||||
ReferenceKind::RunGoalFile,
|
||||
Some(config_path.clone()),
|
||||
)?;
|
||||
files
|
||||
.get(&bundled.path.to_string())
|
||||
.expect("collect_bundled_file inserts the goal file it returns")
|
||||
.content
|
||||
.clone()
|
||||
}
|
||||
};
|
||||
self.collect_template_include_files(
|
||||
files,
|
||||
TemplateSource::new(entrypoint.clone(), workflow_package_root(), content),
|
||||
Some(config_path),
|
||||
)
|
||||
}
|
||||
|
||||
fn collect_environment_dockerfile(
|
||||
&self,
|
||||
files: &mut HashMap<String, types::ManifestFileEntry>,
|
||||
|
|
@ -413,6 +499,10 @@ fn manifest_parent_or_dot(path: &ManifestPath) -> Result<ManifestPath> {
|
|||
.ok_or_else(|| anyhow!("invalid manifest parent path for {path}: {parent}"))
|
||||
}
|
||||
|
||||
fn workflow_package_root() -> ManifestPath {
|
||||
ManifestPath::from_wire(".").expect("the workflow package root must be a valid manifest path")
|
||||
}
|
||||
|
||||
fn template_root_for_bundled_file(
|
||||
path: &ManifestPath,
|
||||
workflow_template_root: &ManifestPath,
|
||||
|
|
|
|||
481
lib/components/fabro-manifest/src/workflow_version_collector.rs
Normal file
481
lib/components/fabro-manifest/src/workflow_version_collector.rs
Normal file
|
|
@ -0,0 +1,481 @@
|
|||
use std::collections::{BTreeMap, HashMap, HashSet};
|
||||
use std::path::{Path, PathBuf};
|
||||
|
||||
use fabro_api::types;
|
||||
use fabro_types::{
|
||||
WorkflowPath, WorkflowPathParseError, WorkflowVersion, WorkflowVersionId,
|
||||
WorkflowVersionShapeError,
|
||||
};
|
||||
use fabro_workflow_version::{ValidatedWorkflowVersion, WorkflowVersionError};
|
||||
use thiserror::Error;
|
||||
|
||||
use crate::workflow_bundler::{CollectedWorkflowSource, CollectedWorkflowSources, WorkflowBundler};
|
||||
|
||||
/// One locally packaged workflow-version closure in dependency-first order.
|
||||
#[derive(Debug)]
|
||||
pub struct CollectedWorkflowClosure {
|
||||
root_id: WorkflowVersionId,
|
||||
versions: Vec<(WorkflowVersionId, ValidatedWorkflowVersion)>,
|
||||
}
|
||||
|
||||
impl CollectedWorkflowClosure {
|
||||
#[must_use]
|
||||
pub fn root_id(&self) -> WorkflowVersionId {
|
||||
self.root_id
|
||||
}
|
||||
|
||||
/// Iterate over every unique version with dependencies before parents.
|
||||
pub fn versions(
|
||||
&self,
|
||||
) -> impl Iterator<Item = (WorkflowVersionId, &ValidatedWorkflowVersion)> + '_ {
|
||||
self.versions.iter().map(|(id, version)| (*id, version))
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Error)]
|
||||
pub enum WorkflowVersionCollectError {
|
||||
#[error("workflow `{path}` was not found")]
|
||||
WorkflowNotFound { path: PathBuf },
|
||||
#[error("failed to collect workflow `{path}`")]
|
||||
Collect {
|
||||
path: PathBuf,
|
||||
#[source]
|
||||
source: anyhow::Error,
|
||||
},
|
||||
#[error("collected workflow path `{path}` is invalid")]
|
||||
InvalidPath {
|
||||
path: String,
|
||||
#[source]
|
||||
source: WorkflowPathParseError,
|
||||
},
|
||||
#[error("collected workflow `{entrypoint}` has conflicting content at `{path}`")]
|
||||
PathCollision {
|
||||
entrypoint: WorkflowPath,
|
||||
path: WorkflowPath,
|
||||
},
|
||||
#[error("collected workflow `{entrypoint}` has an invalid shape")]
|
||||
InvalidShape {
|
||||
entrypoint: WorkflowPath,
|
||||
#[source]
|
||||
source: WorkflowVersionShapeError,
|
||||
},
|
||||
#[error("collected workflow `{entrypoint}` is invalid")]
|
||||
InvalidVersion {
|
||||
entrypoint: WorkflowPath,
|
||||
#[source]
|
||||
source: WorkflowVersionError,
|
||||
},
|
||||
#[error("workflow dependency cycle reaches `{path}`")]
|
||||
DependencyCycle { path: WorkflowPath },
|
||||
#[error("collected workflow dependency `{path}` is missing")]
|
||||
MissingWorkflow { path: String },
|
||||
}
|
||||
|
||||
/// Package one workflow and every separately runnable dependency from a local
|
||||
/// checkout. All paths are rooted at `checkout_root`, so moving the physical
|
||||
/// checkout does not change canonical version bytes or IDs.
|
||||
pub fn collect_workflow_versions(
|
||||
workflow: &Path,
|
||||
checkout_root: &Path,
|
||||
) -> Result<CollectedWorkflowClosure, WorkflowVersionCollectError> {
|
||||
let location =
|
||||
crate::resolve_existing_workflow_location(workflow, checkout_root).map_err(|source| {
|
||||
if matches!(
|
||||
source.downcast_ref::<fabro_config::Error>(),
|
||||
Some(fabro_config::Error::WorkflowNotFound(_))
|
||||
) {
|
||||
WorkflowVersionCollectError::WorkflowNotFound {
|
||||
path: workflow.to_path_buf(),
|
||||
}
|
||||
} else {
|
||||
WorkflowVersionCollectError::Collect {
|
||||
path: workflow.to_path_buf(),
|
||||
source,
|
||||
}
|
||||
}
|
||||
})?;
|
||||
|
||||
let inputs = HashMap::new();
|
||||
let collected = WorkflowBundler::new(checkout_root, &inputs)
|
||||
.collect_versions(&location)
|
||||
.map_err(|source| WorkflowVersionCollectError::Collect {
|
||||
path: workflow.to_path_buf(),
|
||||
source,
|
||||
})?;
|
||||
VersionAssembler::new(collected).assemble()
|
||||
}
|
||||
|
||||
struct VersionAssembler {
|
||||
root_key: String,
|
||||
/// Sources still waiting to be assembled; each is removed once visited.
|
||||
pending: HashMap<String, CollectedWorkflowSource>,
|
||||
visiting: HashSet<String>,
|
||||
ids: HashMap<String, WorkflowVersionId>,
|
||||
versions: Vec<(WorkflowVersionId, ValidatedWorkflowVersion)>,
|
||||
}
|
||||
|
||||
impl VersionAssembler {
|
||||
fn new(collected: CollectedWorkflowSources) -> Self {
|
||||
Self {
|
||||
root_key: collected.root_key,
|
||||
pending: collected.workflows,
|
||||
visiting: HashSet::new(),
|
||||
ids: HashMap::new(),
|
||||
versions: Vec::new(),
|
||||
}
|
||||
}
|
||||
|
||||
fn assemble(mut self) -> Result<CollectedWorkflowClosure, WorkflowVersionCollectError> {
|
||||
let root_key = std::mem::take(&mut self.root_key);
|
||||
let root_id = self.assemble_one(&root_key)?;
|
||||
Ok(CollectedWorkflowClosure {
|
||||
root_id,
|
||||
versions: self.versions,
|
||||
})
|
||||
}
|
||||
|
||||
fn assemble_one(
|
||||
&mut self,
|
||||
key: &str,
|
||||
) -> Result<WorkflowVersionId, WorkflowVersionCollectError> {
|
||||
if let Some(id) = self.ids.get(key) {
|
||||
return Ok(*id);
|
||||
}
|
||||
if !self.visiting.insert(key.to_owned()) {
|
||||
return Err(WorkflowVersionCollectError::DependencyCycle {
|
||||
path: workflow_path(key)?,
|
||||
});
|
||||
}
|
||||
let source = self.pending.remove(key).ok_or_else(|| {
|
||||
WorkflowVersionCollectError::MissingWorkflow {
|
||||
path: key.to_owned(),
|
||||
}
|
||||
})?;
|
||||
|
||||
let mut dependencies = BTreeMap::new();
|
||||
for dependency_key in &source.dependency_keys {
|
||||
let dependency_id = self.assemble_one(dependency_key)?;
|
||||
dependencies.insert(workflow_path(dependency_key)?, dependency_id);
|
||||
}
|
||||
|
||||
let entrypoint = workflow_path(key)?;
|
||||
let files = workflow_files(&entrypoint, source.workflow)?;
|
||||
let version =
|
||||
WorkflowVersion::new(entrypoint.clone(), files, dependencies).map_err(|source| {
|
||||
WorkflowVersionCollectError::InvalidShape {
|
||||
entrypoint: entrypoint.clone(),
|
||||
source,
|
||||
}
|
||||
})?;
|
||||
let validated = ValidatedWorkflowVersion::new(version).map_err(|source| {
|
||||
WorkflowVersionCollectError::InvalidVersion {
|
||||
entrypoint: entrypoint.clone(),
|
||||
source,
|
||||
}
|
||||
})?;
|
||||
let id = validated.version().id().map_err(|source| {
|
||||
WorkflowVersionCollectError::InvalidShape {
|
||||
entrypoint: entrypoint.clone(),
|
||||
source,
|
||||
}
|
||||
})?;
|
||||
|
||||
self.visiting.remove(key);
|
||||
// Keys are distinct entrypoints and the entrypoint is part of the
|
||||
// canonical bytes, so each key yields a distinct ID.
|
||||
self.ids.insert(key.to_owned(), id);
|
||||
self.versions.push((id, validated));
|
||||
Ok(id)
|
||||
}
|
||||
}
|
||||
|
||||
fn workflow_files(
|
||||
entrypoint: &WorkflowPath,
|
||||
workflow: types::ManifestWorkflow,
|
||||
) -> Result<BTreeMap<WorkflowPath, String>, WorkflowVersionCollectError> {
|
||||
let mut files = BTreeMap::new();
|
||||
insert_file(&mut files, entrypoint, entrypoint.clone(), workflow.source)?;
|
||||
if let Some(config) = workflow.config {
|
||||
insert_file(
|
||||
&mut files,
|
||||
entrypoint,
|
||||
workflow_path(&config.path)?,
|
||||
config.source,
|
||||
)?;
|
||||
}
|
||||
for (path, file) in workflow.files {
|
||||
insert_file(&mut files, entrypoint, workflow_path(&path)?, file.content)?;
|
||||
}
|
||||
Ok(files)
|
||||
}
|
||||
|
||||
fn insert_file(
|
||||
files: &mut BTreeMap<WorkflowPath, String>,
|
||||
entrypoint: &WorkflowPath,
|
||||
path: WorkflowPath,
|
||||
content: String,
|
||||
) -> Result<(), WorkflowVersionCollectError> {
|
||||
if let Some(existing) = files.get(&path) {
|
||||
if existing == &content {
|
||||
return Ok(());
|
||||
}
|
||||
return Err(WorkflowVersionCollectError::PathCollision {
|
||||
entrypoint: entrypoint.clone(),
|
||||
path,
|
||||
});
|
||||
}
|
||||
files.insert(path, content);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn workflow_path(value: &str) -> Result<WorkflowPath, WorkflowVersionCollectError> {
|
||||
WorkflowPath::new(value).map_err(|source| WorkflowVersionCollectError::InvalidPath {
|
||||
path: value.to_owned(),
|
||||
source,
|
||||
})
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
#![expect(
|
||||
clippy::disallowed_methods,
|
||||
reason = "collector tests write small temporary workflow fixtures synchronously"
|
||||
)]
|
||||
|
||||
use std::fs;
|
||||
|
||||
use super::*;
|
||||
|
||||
fn write(root: &Path, path: &str, content: &str) {
|
||||
let path = root.join(path);
|
||||
fs::create_dir_all(path.parent().expect("fixture path should have a parent")).unwrap();
|
||||
fs::write(path, content).unwrap();
|
||||
}
|
||||
|
||||
fn write_complete_fixture(root: &Path) {
|
||||
write(root, ".fabro/project.toml", "_version = 1\n");
|
||||
write(
|
||||
root,
|
||||
".fabro/workflows/root/workflow.toml",
|
||||
r#"_version = 1
|
||||
[workflow]
|
||||
graph = "workflow.fabro"
|
||||
[run.goal]
|
||||
file = "goal.md"
|
||||
[run.environment.image]
|
||||
dockerfile = { path = "Dockerfile" }
|
||||
"#,
|
||||
);
|
||||
write(
|
||||
root,
|
||||
".fabro/workflows/root/workflow.fabro",
|
||||
r#"digraph Root {
|
||||
graph [goal="@graph-goal.md"]
|
||||
prompt [prompt="@prompts/task.md"]
|
||||
child [stack.child_workflow="../child/workflow.fabro"]
|
||||
}"#,
|
||||
);
|
||||
write(
|
||||
root,
|
||||
".fabro/workflows/root/goal.md",
|
||||
r#"Ship it. {% include "shared.md" %}"#,
|
||||
);
|
||||
write(root, ".fabro/workflows/root/shared.md", "shared");
|
||||
write(root, ".fabro/workflows/root/Dockerfile", "FROM alpine\n");
|
||||
write(root, ".fabro/workflows/root/graph-goal.md", "graph goal");
|
||||
write(
|
||||
root,
|
||||
".fabro/workflows/root/prompts/task.md",
|
||||
r#"Task {% include "detail.md" %}"#,
|
||||
);
|
||||
write(root, ".fabro/workflows/root/prompts/detail.md", "detail");
|
||||
write(
|
||||
root,
|
||||
".fabro/workflows/child/workflow.fabro",
|
||||
"digraph Child {}",
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn packages_complete_checkout_relative_dependency_closure() {
|
||||
let temp = tempfile::tempdir().unwrap();
|
||||
write_complete_fixture(temp.path());
|
||||
|
||||
let closure = collect_workflow_versions(Path::new("root"), temp.path()).unwrap();
|
||||
let versions = closure.versions().collect::<Vec<_>>();
|
||||
|
||||
assert_eq!(versions.len(), 2);
|
||||
let (child_id, child) = versions[0];
|
||||
assert_eq!(
|
||||
child.version().entrypoint().as_str(),
|
||||
".fabro/workflows/child/workflow.fabro"
|
||||
);
|
||||
let (root_id, root) = versions[1];
|
||||
assert_eq!(root_id, closure.root_id());
|
||||
assert_eq!(
|
||||
root.version().workflow_dependencies(),
|
||||
&BTreeMap::from([(
|
||||
WorkflowPath::new(".fabro/workflows/child/workflow.fabro").unwrap(),
|
||||
child_id,
|
||||
)])
|
||||
);
|
||||
for path in [
|
||||
".fabro/workflows/root/workflow.fabro",
|
||||
".fabro/workflows/root/workflow.toml",
|
||||
".fabro/workflows/root/goal.md",
|
||||
".fabro/workflows/root/shared.md",
|
||||
".fabro/workflows/root/Dockerfile",
|
||||
".fabro/workflows/root/graph-goal.md",
|
||||
".fabro/workflows/root/prompts/task.md",
|
||||
".fabro/workflows/root/prompts/detail.md",
|
||||
] {
|
||||
assert!(
|
||||
root.version()
|
||||
.files()
|
||||
.contains_key(&WorkflowPath::new(path).unwrap()),
|
||||
"missing {path}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn package_identity_does_not_depend_on_checkout_location() {
|
||||
let first = tempfile::tempdir().unwrap();
|
||||
let second = tempfile::tempdir().unwrap();
|
||||
write_complete_fixture(first.path());
|
||||
write_complete_fixture(second.path());
|
||||
|
||||
let first = collect_workflow_versions(Path::new("root"), first.path()).unwrap();
|
||||
let second = collect_workflow_versions(Path::new("root"), second.path()).unwrap();
|
||||
|
||||
assert_eq!(first.root_id(), second.root_id());
|
||||
assert_eq!(
|
||||
first
|
||||
.versions()
|
||||
.map(|(id, version)| (id, version.version().canonical_bytes().unwrap()))
|
||||
.collect::<Vec<_>>(),
|
||||
second
|
||||
.versions()
|
||||
.map(|(id, version)| (id, version.version().canonical_bytes().unwrap()))
|
||||
.collect::<Vec<_>>()
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn packages_nested_imported_and_diamond_dependencies_deterministically() {
|
||||
let temp = tempfile::tempdir().unwrap();
|
||||
write(temp.path(), ".fabro/project.toml", "_version = 1\n");
|
||||
write(
|
||||
temp.path(),
|
||||
".fabro/workflows/root/workflow.toml",
|
||||
"_version = 1\n[workflow]\ngraph = \"workflow.fabro\"\n",
|
||||
);
|
||||
write(
|
||||
temp.path(),
|
||||
".fabro/workflows/root/workflow.fabro",
|
||||
r#"digraph Root {
|
||||
imported [import="imports/extension.fabro"]
|
||||
left [stack.child_workflow="../left/workflow.fabro"]
|
||||
right [stack.child_workflow="../right/workflow.fabro"]
|
||||
}"#,
|
||||
);
|
||||
write(
|
||||
temp.path(),
|
||||
".fabro/workflows/root/imports/extension.fabro",
|
||||
r#"digraph Extension {
|
||||
child [stack.child_workflow="../../imported-child/workflow.fabro"]
|
||||
}"#,
|
||||
);
|
||||
write(
|
||||
temp.path(),
|
||||
".fabro/workflows/imported-child/workflow.fabro",
|
||||
"digraph ImportedChild {}",
|
||||
);
|
||||
write(
|
||||
temp.path(),
|
||||
".fabro/workflows/left/workflow.fabro",
|
||||
r#"digraph Left { nested [stack.child_workflow="../nested/workflow.fabro"] }"#,
|
||||
);
|
||||
write(
|
||||
temp.path(),
|
||||
".fabro/workflows/nested/workflow.fabro",
|
||||
r#"digraph Nested { shared [stack.child_workflow="../shared/workflow.fabro"] }"#,
|
||||
);
|
||||
write(
|
||||
temp.path(),
|
||||
".fabro/workflows/right/workflow.fabro",
|
||||
r#"digraph Right { shared [stack.child_workflow="../shared/workflow.fabro"] }"#,
|
||||
);
|
||||
write(
|
||||
temp.path(),
|
||||
".fabro/workflows/shared/workflow.fabro",
|
||||
"digraph Shared {}",
|
||||
);
|
||||
|
||||
let first = collect_workflow_versions(Path::new("root"), temp.path()).unwrap();
|
||||
let second = collect_workflow_versions(Path::new("root"), temp.path()).unwrap();
|
||||
let entrypoint_order = |closure: &CollectedWorkflowClosure| {
|
||||
closure
|
||||
.versions()
|
||||
.map(|(_, version)| version.version().entrypoint().as_str().to_owned())
|
||||
.collect::<Vec<_>>()
|
||||
};
|
||||
|
||||
let expected = vec![
|
||||
".fabro/workflows/imported-child/workflow.fabro",
|
||||
".fabro/workflows/shared/workflow.fabro",
|
||||
".fabro/workflows/nested/workflow.fabro",
|
||||
".fabro/workflows/left/workflow.fabro",
|
||||
".fabro/workflows/right/workflow.fabro",
|
||||
".fabro/workflows/root/workflow.fabro",
|
||||
];
|
||||
assert_eq!(entrypoint_order(&first), expected);
|
||||
assert_eq!(entrypoint_order(&second), expected);
|
||||
assert_eq!(first.root_id(), second.root_id());
|
||||
assert_eq!(
|
||||
entrypoint_order(&first)
|
||||
.iter()
|
||||
.filter(|path| path.as_str() == ".fabro/workflows/shared/workflow.fabro")
|
||||
.count(),
|
||||
1
|
||||
);
|
||||
let root = first.versions().last().unwrap().1;
|
||||
assert!(root.version().files().contains_key(
|
||||
&WorkflowPath::new(".fabro/workflows/root/imports/extension.fabro").unwrap()
|
||||
));
|
||||
assert_eq!(root.version().workflow_dependencies().len(), 3);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_dependency_cycles() {
|
||||
let temp = tempfile::tempdir().unwrap();
|
||||
write(temp.path(), ".fabro/project.toml", "_version = 1\n");
|
||||
write(
|
||||
temp.path(),
|
||||
".fabro/workflows/root/workflow.toml",
|
||||
"_version = 1\n[workflow]\ngraph = \"workflow.fabro\"\n",
|
||||
);
|
||||
write(
|
||||
temp.path(),
|
||||
".fabro/workflows/child/workflow.toml",
|
||||
"_version = 1\n[workflow]\ngraph = \"workflow.fabro\"\n",
|
||||
);
|
||||
write(
|
||||
temp.path(),
|
||||
".fabro/workflows/root/workflow.fabro",
|
||||
r#"digraph Root { child [stack.child_workflow="../child/workflow.fabro"] }"#,
|
||||
);
|
||||
write(
|
||||
temp.path(),
|
||||
".fabro/workflows/child/workflow.fabro",
|
||||
r#"digraph Child { root [stack.child_workflow="../root/workflow.fabro"] }"#,
|
||||
);
|
||||
|
||||
let error = collect_workflow_versions(Path::new("root"), temp.path()).unwrap_err();
|
||||
|
||||
assert!(
|
||||
matches!(&error, WorkflowVersionCollectError::DependencyCycle { .. }),
|
||||
"unexpected error: {error:?}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
|
@ -116,8 +116,7 @@ impl ValidatedWorkflowVersion {
|
|||
/// run-admission time read the same bytes validation proved present.
|
||||
#[must_use]
|
||||
pub fn resolved_goal_file_content(&self) -> Option<&str> {
|
||||
let config_path = WorkflowPath::new("workflow.toml")
|
||||
.expect("the static workflow config path must be valid");
|
||||
let config_path = self.0.config_path();
|
||||
let source = self.0.files().get(&config_path)?;
|
||||
let layer: SettingsLayer = source.parse().expect("validated workflow.toml must parse");
|
||||
let RunGoalLayer::File { file } = layer.run.as_ref().and_then(|run| run.goal.as_ref())?
|
||||
|
|
@ -164,8 +163,7 @@ fn validate_config(
|
|||
version: &WorkflowVersion,
|
||||
template_roots: &mut TemplateRoots,
|
||||
) -> Result<(), WorkflowVersionError> {
|
||||
let config_path =
|
||||
WorkflowPath::new("workflow.toml").expect("the static workflow config path must be valid");
|
||||
let config_path = version.config_path();
|
||||
let Some(source) = version.files().get(&config_path) else {
|
||||
return Ok(());
|
||||
};
|
||||
|
|
@ -649,7 +647,7 @@ mod tests {
|
|||
BTreeMap::from([
|
||||
(path("graphs/main.fabro"), "digraph W {}".to_owned()),
|
||||
(
|
||||
path("workflow.toml"),
|
||||
path("graphs/workflow.toml"),
|
||||
"_version = 1\n[run]\ngoal = \"{% include \\\"shared.md\\\" %}\"\n"
|
||||
.to_owned(),
|
||||
),
|
||||
|
|
@ -677,6 +675,38 @@ mod tests {
|
|||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn discovers_workflow_config_beside_a_nested_entrypoint() {
|
||||
let version = ValidatedWorkflowVersion::new(
|
||||
WorkflowVersion::new(
|
||||
path(".fabro/workflows/demo/workflow.fabro"),
|
||||
BTreeMap::from([
|
||||
(
|
||||
path(".fabro/workflows/demo/workflow.fabro"),
|
||||
"digraph W {}".to_owned(),
|
||||
),
|
||||
(
|
||||
path(".fabro/workflows/demo/workflow.toml"),
|
||||
"_version = 1\n[workflow]\ngraph = \"workflow.fabro\"\n[run.goal]\nfile = \"goal.md\"\n"
|
||||
.to_owned(),
|
||||
),
|
||||
(
|
||||
path(".fabro/workflows/demo/goal.md"),
|
||||
"Ship the nested workflow".to_owned(),
|
||||
),
|
||||
]),
|
||||
BTreeMap::new(),
|
||||
)
|
||||
.expect("nested workflow version should be structurally valid"),
|
||||
)
|
||||
.expect("entrypoint-adjacent workflow config should validate");
|
||||
|
||||
assert_eq!(
|
||||
version.resolved_goal_file_content(),
|
||||
Some("Ship the nested workflow")
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_non_static_or_nonportable_workflow_goal_file_references() {
|
||||
for reference in ["{{ vars.NAME }}", "{% include \"goal.md\" %}"] {
|
||||
|
|
|
|||
|
|
@ -6,7 +6,7 @@ use serde::de::{Error as _, MapAccess, Visitor};
|
|||
use serde::{Deserialize, Deserializer, Serialize};
|
||||
use thiserror::Error;
|
||||
|
||||
use crate::{WorkflowPath, WorkflowVersionId};
|
||||
use crate::{BlobHash, WorkflowPath, WorkflowVersionId};
|
||||
|
||||
pub const MAX_WORKFLOW_VERSION_FILES: usize = 512;
|
||||
pub const MAX_WORKFLOW_VERSION_DEPENDENCIES: usize = 512;
|
||||
|
|
@ -86,6 +86,22 @@ impl WorkflowVersion {
|
|||
&self.workflow_dependencies
|
||||
}
|
||||
|
||||
/// Path of the optional `workflow.toml` that configures this version. It
|
||||
/// always sits beside the entrypoint graph.
|
||||
#[must_use]
|
||||
pub fn config_path(&self) -> WorkflowPath {
|
||||
self.entrypoint
|
||||
.resolve_reference("workflow.toml")
|
||||
.expect("the static workflow config path must resolve beside a valid entrypoint")
|
||||
}
|
||||
|
||||
/// Content-addressed identity: the hash of the canonical wire form.
|
||||
pub fn id(&self) -> Result<WorkflowVersionId, WorkflowVersionShapeError> {
|
||||
Ok(WorkflowVersionId::from(BlobHash::new(
|
||||
&self.canonical_bytes()?,
|
||||
)))
|
||||
}
|
||||
|
||||
/// Serialize to the canonical wire form.
|
||||
///
|
||||
/// Structural validity is guaranteed by construction, so this only
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue