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

Merge remote-tracking branch 'origin/main'

Browse source
This commit is contained in:
Bryan Helmkamp 2026-04-07 23:13:56 -04:00
commit 8bf619a095
No known key found for this signature in database
29 changed files with 559 additions and 135 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
lib/crates/fabro-cli/src/commands/run/rewind.rs
View file

@ -3,7 +3,7 @@ use anyhow::Result;
use cli_table::format::{Border, Separator};
use cli_table::{Cell, CellStruct, Color, Style, Table};
use fabro_checkpoint::git::Store;
use fabro_types::run_event::{CheckpointCompletedProps, RunRewoundProps, RunStatusTransitionProps};
use fabro_types::run_event::{CheckpointCompletedProps, RunRewoundProps, RunSubmittedProps};
use fabro_types::{EventBody, RunEvent};
use fabro_util::terminal::Styles;
use fabro_workflow::git::MetadataStore;
@ -104,6 +104,7 @@ async fn reset_rewound_run_state(
anyhow::anyhow!("failed to load durable store state before rewind: {err}")
})?;
let definition_blob = state.run.as_ref().and_then(|run| run.definition_blob);
let _run_record = state
.run
.context("failed to restore run record after rewind: missing run metadata")?;
@ -138,7 +139,10 @@ async fn reset_rewound_run_state(
&run_event(
*run_id,
None,
EventBody::RunSubmitted(RunStatusTransitionProps { reason: None }),
EventBody::RunSubmitted(RunSubmittedProps {
reason: None,
definition_blob,
}),
),
)
.await

3
lib/crates/fabro-cli/src/commands/store/dump.rs
View file

@ -344,6 +344,8 @@ mod tests {
labels: HashMap::from([("team".to_string(), "infra".to_string())]),
artifact_storage: None,
provenance: None,
manifest_blob: None,
definition_blob: None,
}
}
@ -490,6 +492,7 @@ mod tests {
db_prefix: None,
artifact_storage: run_record.artifact_storage,
provenance: run_record.provenance.clone(),
manifest_blob: None,
},
)
.await

22
lib/crates/fabro-cli/tests/it/cmd/attach.rs
View file

@ -352,6 +352,23 @@ fn attach_json_errors_without_prompting_for_human_input() {
.lines()
.filter(|line| !line.trim().is_empty())
.map(|line| serde_json::from_str(line).expect("attach JSON output should be JSONL"))
.map(|mut event: Value| {
if let Some(properties) = event.get_mut("properties").and_then(Value::as_object_mut) {
if properties.contains_key("manifest_blob") {
properties.insert(
"manifest_blob".to_string(),
Value::String("[BLOB_ID]".to_string()),
);
}
if properties.contains_key("definition_blob") {
properties.insert(
"definition_blob".to_string(),
Value::String("[BLOB_ID]".to_string()),
);
}
}
event
})
.collect();
fabro_json_snapshot!(context, &progress, @r#"
[
@ -461,6 +478,7 @@ fn attach_json_errors_without_prompting_for_human_input() {
}
},
"host_repo_path": "[TEMP_DIR]",
"manifest_blob": "[BLOB_ID]",
"provenance": {
"client": {
"name": "fabro-cli",
@ -503,7 +521,9 @@ fn attach_json_errors_without_prompting_for_human_input() {
{
"event": "run.submitted",
"id": "[EVENT_ID]",
"properties": {},
"properties": {
"definition_blob": "[BLOB_ID]"
},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},

5
lib/crates/fabro-cli/tests/it/cmd/rewind.rs
View file

@ -174,6 +174,11 @@ fn rewind_preserves_event_history_and_clears_terminal_snapshot_state() {
after_events[before_events.len() + 2].payload.as_value()["event"],
"run.submitted"
);
assert!(
after_events[before_events.len() + 2].payload.as_value()["properties"]["definition_blob"]
.is_string(),
"rewind should re-emit run.submitted with the definition_blob"
);
let state = run_state(&setup.run.run_dir);
assert_eq!(

19
lib/crates/fabro-cli/tests/it/cmd/run.rs
View file

@ -625,6 +625,20 @@ fn json_run_implies_auto_approve_for_human_gates() {
.map(|line| serde_json::from_str(line).expect("run JSON output should be JSONL"))
.collect();
for event in &mut progress {
if let Some(properties) = event.get_mut("properties").and_then(Value::as_object_mut) {
if properties.contains_key("manifest_blob") {
properties.insert(
"manifest_blob".to_string(),
Value::String("[BLOB_ID]".to_string()),
);
}
if properties.contains_key("definition_blob") {
properties.insert(
"definition_blob".to_string(),
Value::String("[BLOB_ID]".to_string()),
);
}
}
let Some(llm) = event.pointer_mut("/properties/settings/llm") else {
continue;
};
@ -748,6 +762,7 @@ fn json_run_implies_auto_approve_for_human_gates() {
}
},
"host_repo_path": "[TEMP_DIR]",
"manifest_blob": "[BLOB_ID]",
"provenance": {
"client": {
"name": "fabro-cli",
@ -791,7 +806,9 @@ fn json_run_implies_auto_approve_for_human_gates() {
{
"event": "run.submitted",
"id": "[EVENT_ID]",
"properties": {},
"properties": {
"definition_blob": "[BLOB_ID]"
},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},

8
lib/crates/fabro-cli/tests/it/support/mod.rs
View file

@ -18,6 +18,14 @@ macro_rules! fabro_json_snapshot {
r#""duration_ms":\s*\d+"#.to_string(),
r#""duration_ms": "[DURATION_MS]""#.to_string(),
));
filters.push((
r#""manifest_blob":\s*"[0-9a-f]{64}""#.to_string(),
r#""manifest_blob": "[BLOB_ID]""#.to_string(),
));
filters.push((
r#""definition_blob":\s*"[0-9a-f]{64}""#.to_string(),
r#""definition_blob": "[BLOB_ID]""#.to_string(),
));
filters.push((
r#""run_dir":\s*"\[STORAGE_DIR\]/scratch/\d{8}-\[ULID\]""#.to_string(),
r#""run_dir": "[RUN_DIR]""#.to_string(),

1
lib/crates/fabro-server/src/run_manifest.rs
View file

@ -124,6 +124,7 @@ pub(crate) fn create_run_input(prepared: PreparedManifest) -> CreateRunInput {
workflow_slug: None,
workflow_path: Some(prepared.target_path),
workflow_bundle: Some(prepared.workflow_bundle),
submitted_manifest_bytes: None,
run_id: prepared.run_id,
host_repo_path: Some(prepared.working_directory.display().to_string()),
repo_origin_url: prepared

98
lib/crates/fabro-server/src/server.rs
View file

@ -3103,8 +3103,12 @@ async fn create_run(
subject: AuthenticatedSubject,
State(state): State<Arc<AppState>>,
headers: HeaderMap,
Json(req): Json<RunManifest>,
body: Bytes,
) -> Response {
let req = match serde_json::from_slice::<RunManifest>(&body) {
Ok(req) => req,
Err(err) => return ApiError::bad_request(err.to_string()).into_response(),
};
let prepared = match run_manifest::prepare_manifest_with_mode(
&state.settings.read().unwrap(),
&req,
@ -3120,6 +3124,7 @@ async fn create_run(
create_input.run_id = Some(run_id);
create_input.artifact_storage = Some(RunArtifactStorage::ObjectStoreV1);
create_input.provenance = Some(run_provenance(&headers, &subject));
create_input.submitted_manifest_bytes = Some(body.to_vec());
let created = match Box::pin(operations::create(state.store.as_ref(), create_input)).await {
Ok(created) => created,
@ -5810,13 +5815,15 @@ async fn get_graph(
#[cfg(test)]
mod tests {
use std::path::PathBuf;
use super::*;
use axum::body::Body;
use axum::http::Request;
use fabro_config::server::{
AuthProvider, AuthSettings, GitAuthorSettings, GitProvider, GitSettings, WebSettings,
};
use fabro_types::{InterviewQuestionRecord, InterviewQuestionType, fixtures};
use fabro_types::{InterviewQuestionRecord, InterviewQuestionType, RunBlobId, RunId, fixtures};
#[cfg(unix)]
use std::process::Stdio;
use tower::ServiceExt;
@ -5944,6 +5951,7 @@ mod tests {
workflow_slug: None,
workflow_path: None,
workflow_bundle: None,
submitted_manifest_bytes: None,
run_id: None,
host_repo_path: None,
repo_origin_url: None,
@ -6477,6 +6485,72 @@ mod tests {
assert!(body["run"]["provenance"]["subject"]["login"].is_null());
}
#[tokio::test]
async fn create_run_persists_manifest_and_definition_blobs_without_bundle_file() {
let state = create_app_state();
let app = build_router(Arc::clone(&state), AuthMode::Disabled);
let raw_manifest =
serde_json::to_string_pretty(&minimal_manifest_json(MINIMAL_DOT)).unwrap();
let req = Request::builder()
.method("POST")
.uri(api("/runs"))
.header("content-type", "application/json")
.body(Body::from(raw_manifest.clone()))
.unwrap();
let response = app.clone().oneshot(req).await.unwrap();
assert_eq!(response.status(), StatusCode::CREATED);
let body = body_json(response.into_body()).await;
let run_id = body["id"].as_str().unwrap().parse::<RunId>().unwrap();
let run_store = state.store.open_run_reader(&run_id).await.unwrap();
let events = run_store.list_events().await.unwrap();
let created = events[0].payload.as_value();
let submitted = events[1].payload.as_value();
let manifest_blob = created["properties"]["manifest_blob"]
.as_str()
.expect("run.created should carry manifest_blob")
.parse::<RunBlobId>()
.unwrap();
let definition_blob = submitted["properties"]["definition_blob"]
.as_str()
.expect("run.submitted should carry definition_blob")
.parse::<RunBlobId>()
.unwrap();
let submitted_manifest_bytes = run_store
.read_blob(&manifest_blob)
.await
.unwrap()
.expect("submitted manifest blob should exist");
assert_eq!(submitted_manifest_bytes.as_ref(), raw_manifest.as_bytes());
let accepted_definition_bytes = run_store
.read_blob(&definition_blob)
.await
.unwrap()
.expect("accepted definition blob should exist");
let accepted_definition: serde_json::Value =
serde_json::from_slice(&accepted_definition_bytes).unwrap();
assert!(
accepted_definition.get("version").is_none(),
"accepted run definition should not carry compatibility versioning"
);
assert_eq!(accepted_definition["workflow_path"], "workflow.fabro");
assert!(accepted_definition["workflows"]["workflow.fabro"].is_object());
let run_dir = PathBuf::from(
created["properties"]["run_dir"]
.as_str()
.expect("run.created should include run_dir"),
);
assert!(
!run_dir.join("workflow_bundle.json").exists(),
"run scratch should no longer persist workflow_bundle.json"
);
}
#[tokio::test]
async fn list_run_events_returns_paginated_json() {
let state = create_app_state();
@ -7612,14 +7686,20 @@ mod tests {
create_durable_run_with_events(
&state,
fixtures::RUN_1,
&[workflow_event::Event::RunSubmitted { reason: None }],
&[workflow_event::Event::RunSubmitted {
reason: None,
definition_blob: None,
}],
)
.await;
create_durable_run_with_events(
&state,
fixtures::RUN_2,
&[
workflow_event::Event::RunSubmitted { reason: None },
workflow_event::Event::RunSubmitted {
reason: None,
definition_blob: None,
},
workflow_event::Event::RunStarting { reason: None },
workflow_event::Event::RunRunning { reason: None },
],
@ -7629,7 +7709,10 @@ mod tests {
&state,
fixtures::RUN_3,
&[
workflow_event::Event::RunSubmitted { reason: None },
workflow_event::Event::RunSubmitted {
reason: None,
definition_blob: None,
},
workflow_event::Event::RunStarting { reason: None },
workflow_event::Event::RunRunning { reason: None },
workflow_event::Event::RunPaused,
@ -7687,7 +7770,10 @@ mod tests {
&state,
run_id,
&[
workflow_event::Event::RunSubmitted { reason: None },
workflow_event::Event::RunSubmitted {
reason: None,
definition_blob: None,
},
workflow_event::Event::RunStarting { reason: None },
workflow_event::Event::RunRunning { reason: None },
],

39
lib/crates/fabro-store/src/keys.rs
View file

@ -2,8 +2,7 @@ use fabro_types::{RunBlobId, RunId};
const RUNS_PREFIX: &str = "runs#";
const RUNS_INDEX_BY_START_PREFIX: &str = "runs#_index#by-start#";
const BLOBS_PREFIX: &str = "blobs#";
const GLOBAL_BLOBS_PREFIX: &str = "blobs#sha256#";
const BLOBS_PREFIX: &str = "blobs#sha256#";
pub(crate) fn runs_index_by_start_prefix() -> &'static str {
RUNS_INDEX_BY_START_PREFIX
@ -28,18 +27,12 @@ pub(crate) fn run_event_key(run_id: &RunId, seq: u32, epoch_ms: i64) -> String {
format!("{}{seq:06}-{epoch_ms}", run_events_prefix(run_id))
}
pub(crate) fn blobs_prefix(run_id: &RunId) -> String {
let _ = run_id;
GLOBAL_BLOBS_PREFIX.to_string()
pub(crate) fn blobs_prefix() -> &'static str {
BLOBS_PREFIX
}
pub(crate) fn blob_key(run_id: &RunId, id: &RunBlobId) -> String {
let _ = run_id;
format!("{GLOBAL_BLOBS_PREFIX}{id}")
}
pub(crate) fn legacy_blob_key(run_id: &RunId, id: &RunBlobId) -> String {
format!("{BLOBS_PREFIX}{run_id}#{id}")
pub(crate) fn blob_key(id: &RunBlobId) -> String {
format!("{BLOBS_PREFIX}{id}")
}
pub(crate) fn parse_event_seq(key: &str) -> Option<u32> {
@ -52,13 +45,7 @@ pub(crate) fn parse_event_seq(key: &str) -> Option<u32> {
}
pub(crate) fn parse_blob_id(key: &str) -> Option<RunBlobId> {
if let Some(blob_id) = key.strip_prefix(GLOBAL_BLOBS_PREFIX) {
return blob_id.parse().ok();
}
let rest = key.strip_prefix(BLOBS_PREFIX)?;
let (_, blob_id) = rest.split_once('#')?;
blob_id.parse().ok()
key.strip_prefix(BLOBS_PREFIX)?.parse().ok()
}
pub(crate) fn parse_run_id_from_index_key(key: &str) -> Option<RunId> {
@ -99,12 +86,8 @@ mod tests {
#[test]
fn blob_keys_match_spec() {
let run_id: RunId = "01JT56VE4Z5NZ814GZN2JZD65A".parse().unwrap();
let blob_id = RunBlobId::new(b"summary");
assert_eq!(
blob_key(&run_id, &blob_id),
format!("blobs#sha256#{blob_id}")
);
assert_eq!(blob_key(&blob_id), format!("blobs#sha256#{blob_id}"));
}
#[test]
@ -118,10 +101,6 @@ mod tests {
parse_blob_id(&format!("blobs#sha256#{blob_id}")),
Some(blob_id)
);
assert_eq!(
parse_blob_id(&format!("blobs#01JT56VE4Z5NZ814GZN2JZD65A#{blob_id}")),
Some(blob_id)
);
assert_eq!(
parse_run_id_from_index_key(
"runs#_index#by-start#2026-03-27#01JT56VE4Z5NZ814GZN2JZD65A"
@ -134,5 +113,9 @@ mod tests {
fn parse_helpers_reject_invalid_keys() {
assert_eq!(parse_event_seq("runs#not-a-run#events#not-a-seq"), None);
assert_eq!(parse_blob_id("blobs#not-a-uuid"), None);
assert_eq!(
parse_blob_id("blobs#01JT56VE4Z5NZ814GZN2JZD65A#not-a-blob"),
None
);
}
}

72
lib/crates/fabro-store/src/run_state.rs
View file

@ -95,6 +95,8 @@ impl RunProjection {
labels,
artifact_storage: props.artifact_storage,
provenance: props.provenance.clone(),
manifest_blob: props.manifest_blob,
definition_blob: None,
});
self.graph_source.clone_from(&props.workflow_source);
}
@ -107,6 +109,9 @@ impl RunProjection {
});
}
EventBody::RunSubmitted(props) => {
if let Some(run) = self.run.as_mut() {
run.definition_blob = props.definition_blob;
}
self.status = Some(run_status_record(RunStatus::Submitted, props.reason, ts));
}
EventBody::RunStarting(props) => {
@ -590,12 +595,14 @@ mod tests {
use std::collections::HashMap;
use chrono::Utc;
use serde_json::json;
use super::{NodeState, RunProjection};
use crate::{EventEnvelope, EventPayload, StageId};
use fabro_types::run_event::{InterviewCompletedProps, InterviewOption, InterviewStartedProps};
use fabro_types::{
Checkpoint, EventBody, InterviewQuestionType, RunControlAction, RunEvent, fixtures,
Checkpoint, EventBody, InterviewQuestionType, RunBlobId, RunControlAction, RunEvent,
Settings, fixtures,
};
fn test_event(seq: u32, body: EventBody, node_id: Option<&str>) -> EventEnvelope {
@ -786,4 +793,67 @@ mod tests {
"completed interview should clear pending state"
);
}
#[test]
fn projection_serialization_includes_manifest_and_definition_blob_refs() {
let manifest_blob = RunBlobId::new(br#"{"version":1}"#).to_string();
let definition_blob =
RunBlobId::new(br#"{"version":1,"workflow_path":"workflow.fabro"}"#).to_string();
let events = vec![
EventEnvelope {
seq: 1,
payload: EventPayload::new(
json!({
"id": "evt-run-created",
"ts": "2026-04-07T12:00:00Z",
"run_id": fixtures::RUN_1,
"event": "run.created",
"properties": {
"settings": Settings::default(),
"graph": {
"name": "test",
"nodes": {},
"edges": [],
"attrs": {}
},
"labels": {},
"run_dir": "/tmp/run",
"working_directory": "/tmp/run",
"manifest_blob": manifest_blob
}
}),
&fixtures::RUN_1,
)
.unwrap(),
},
EventEnvelope {
seq: 2,
payload: EventPayload::new(
json!({
"id": "evt-run-submitted",
"ts": "2026-04-07T12:00:01Z",
"run_id": fixtures::RUN_1,
"event": "run.submitted",
"properties": {
"definition_blob": definition_blob
}
}),
&fixtures::RUN_1,
)
.unwrap(),
},
];
let state = RunProjection::apply_events(&events).unwrap();
let value = serde_json::to_value(&state).unwrap();
assert_eq!(
value["run"]["manifest_blob"],
events[0].payload.as_value()["properties"]["manifest_blob"]
);
assert_eq!(
value["run"]["definition_blob"],
events[1].payload.as_value()["properties"]["definition_blob"]
);
}
}

22
lib/crates/fabro-store/src/slate/mod.rs
View file

@ -173,7 +173,7 @@ impl Database {
let db = self.open_db().await?;
let mut keys_to_delete = Vec::new();
for prefix in [keys::run_data_prefix(run_id), keys::blobs_prefix(run_id)] {
for prefix in [keys::run_data_prefix(run_id)] {
let mut iter = db.scan_prefix(prefix.as_bytes()).await?;
while let Some(entry) = iter.next().await? {
keys_to_delete.push(String::from_utf8(entry.key.to_vec()).map_err(|err| {
@ -290,6 +290,8 @@ mod tests {
labels: std::collections::HashMap::from([("team".to_string(), "infra".to_string())]),
artifact_storage: None,
provenance: None,
manifest_blob: None,
definition_blob: None,
}
}
@ -403,6 +405,24 @@ mod tests {
assert!(!list_paths(object_store, "runs/db").await.is_empty());
}
#[tokio::test]
async fn delete_run_keeps_global_cas_blobs() {
let (_object_store, store) = make_store();
let run_1 = store.create_run(&test_run_id("run-1")).await.unwrap();
let run_2 = store.create_run(&test_run_id("run-2")).await.unwrap();
append_created(&run_1, "run-1", dt("2026-03-27T12:00:00Z")).await;
append_created(&run_2, "run-2", dt("2026-03-27T12:00:10Z")).await;
let shared_blob = br#"{"summary":"shared"}"#;
let shared_blob_id = run_1.write_blob(shared_blob).await.unwrap();
store.delete_run(&test_run_id("run-1")).await.unwrap();
let reopened = store.open_run(&test_run_id("run-2")).await.unwrap();
let read = reopened.read_blob(&shared_blob_id).await.unwrap();
assert_eq!(read.as_deref(), Some(shared_blob.as_slice()));
}
#[tokio::test]
async fn open_run_reader_is_read_only() {
let (_object_store, store) = make_store();

49
lib/crates/fabro-store/src/slate/run_store.rs
View file

@ -15,7 +15,6 @@ use crate::{EventEnvelope, EventPayload, Result, RunProjection, RunSummary, Stor
use fabro_types::{RunBlobId, RunId};
const DEFAULT_EVENT_TAIL_LIMIT: usize = 1024;
#[derive(Clone)]
pub struct RunDatabase {
inner: Arc<RunDatabaseInner>,
@ -297,32 +296,20 @@ impl RunDatabase {
return Err(StoreError::ReadOnly);
}
let id = RunBlobId::new(data);
self.inner
.db
.put(keys::blob_key(&self.inner.run_id, &id), data)
.await?;
self.inner.db.put(keys::blob_key(&id), data).await?;
Ok(id)
}
pub async fn read_blob(&self, id: &RunBlobId) -> Result<Option<Bytes>> {
let global = self
.inner
.db
.get(keys::blob_key(&self.inner.run_id, id))
.await?;
let global = self.inner.db.get(keys::blob_key(id)).await?;
if global.is_some() {
return Ok(global);
}
Ok(self
.inner
.db
.get(keys::legacy_blob_key(&self.inner.run_id, id))
.await?)
Ok(None)
}
pub async fn list_blobs(&self) -> Result<Vec<RunBlobId>> {
list_blobs(&self.inner.db, &self.inner.run_id).await
list_blobs(&self.inner.db).await
}
pub async fn state(&self) -> Result<RunProjection> {
@ -384,13 +371,11 @@ where
Ok(events)
}
async fn list_blobs<R>(db: &R, run_id: &RunId) -> Result<Vec<RunBlobId>>
async fn list_blobs<R>(db: &R) -> Result<Vec<RunBlobId>>
where
R: DbRead + Sync,
{
let mut iter = db
.scan_prefix(keys::blobs_prefix(run_id).as_bytes())
.await?;
let mut iter = db.scan_prefix(keys::blobs_prefix().as_bytes()).await?;
let mut blob_ids = Vec::new();
while let Some(entry) = iter.next().await? {
let key = key_to_string(&entry.key)?;
@ -416,28 +401,6 @@ mod tests {
use object_store::memory::InMemory;
use crate::Database;
use crate::keys;
#[tokio::test]
async fn read_blob_falls_back_to_legacy_run_scoped_key() {
let object_store = Arc::new(InMemory::new());
let store = Database::new(object_store, "", Duration::from_millis(1));
let run_id = "01JT56VE4Z5NZ814GZN2JZD65A".parse().unwrap();
let run = store.create_run(&run_id).await.unwrap();
let blob = br#"{"legacy":true}"#;
let blob_id = fabro_types::RunBlobId::new(blob);
run.inner
.db
.put(keys::legacy_blob_key(&run_id, &blob_id), blob.as_slice())
.await
.unwrap();
let read = run.read_blob(&blob_id).await.unwrap().unwrap();
assert_eq!(read.as_ref(), blob);
}
#[tokio::test]
async fn list_blobs_reads_global_cas_namespace() {
let object_store = Arc::new(InMemory::new());

5
lib/crates/fabro-types/src/run.rs
View file

@ -4,6 +4,7 @@ use std::path::PathBuf;
use serde::{Deserialize, Serialize};
use crate::graph::Graph;
use crate::run_blob_id::RunBlobId;
use crate::run_id::RunId;
use crate::settings::Settings;
@ -74,6 +75,10 @@ pub struct RunRecord {
pub artifact_storage: Option<RunArtifactStorage>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub provenance: Option<RunProvenance>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub manifest_blob: Option<RunBlobId>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub definition_blob: Option<RunBlobId>,
}
impl RunRecord {

51
lib/crates/fabro-types/src/run_event/mod.rs
View file

@ -48,7 +48,7 @@ pub enum EventBody {
#[serde(rename = "run.started")]
RunStarted(RunStartedProps),
#[serde(rename = "run.submitted")]
RunSubmitted(RunStatusTransitionProps),
RunSubmitted(RunSubmittedProps),
#[serde(rename = "run.starting")]
RunStarting(RunStatusTransitionProps),
#[serde(rename = "run.running")]
@ -682,7 +682,7 @@ mod tests {
use serde_json::json;
use crate::{Edge, Graph, Node, Settings, fixtures};
use crate::{Edge, Graph, Node, RunBlobId, Settings, fixtures};
use super::*;
@ -765,6 +765,53 @@ mod tests {
assert!(matches!(parsed.body, EventBody::RunCreated(_)));
}
#[test]
fn run_created_round_trip_preserves_manifest_blob() {
let line = json!({
"id": "evt_created_blob",
"ts": "2026-04-04T12:00:00.000Z",
"run_id": fixtures::RUN_1,
"event": "run.created",
"properties": {
"settings": Settings::default(),
"graph": Graph::new("test"),
"labels": {},
"run_dir": "/tmp/run",
"working_directory": "/tmp/run",
"manifest_blob": RunBlobId::new(br#"{"version":1}"#).to_string()
}
});
let parsed = RunEvent::from_value(line.clone()).unwrap();
let serialized = parsed.to_value().unwrap();
assert_eq!(
serialized["properties"]["manifest_blob"],
line["properties"]["manifest_blob"]
);
}
#[test]
fn run_submitted_round_trip_preserves_definition_blob() {
let line = json!({
"id": "evt_submitted_blob",
"ts": "2026-04-04T12:00:00.000Z",
"run_id": fixtures::RUN_1,
"event": "run.submitted",
"properties": {
"definition_blob": RunBlobId::new(br#"{"workflow_path":"workflow.fabro"}"#).to_string()
}
});
let parsed = RunEvent::from_value(line.clone()).unwrap();
let serialized = parsed.to_value().unwrap();
assert_eq!(
serialized["properties"]["definition_blob"],
line["properties"]["definition_blob"]
);
}
#[test]
fn event_body_event_name_matches_wire_name() {
let body = EventBody::StageCompleted(StageCompletedProps {

14
lib/crates/fabro-types/src/run_event/run.rs
View file

@ -2,7 +2,9 @@ use std::collections::BTreeMap;
use serde::{Deserialize, Serialize};
use crate::{Graph, RunArtifactStorage, RunControlAction, RunProvenance, Settings, StatusReason};
use crate::{
Graph, RunArtifactStorage, RunBlobId, RunControlAction, RunProvenance, Settings, StatusReason,
};
use super::{BilledTokenCounts, RunNoticeLevel};
@ -32,6 +34,8 @@ pub struct RunCreatedProps {
pub artifact_storage: Option<RunArtifactStorage>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub provenance: Option<RunProvenance>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub manifest_blob: Option<RunBlobId>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
@ -55,6 +59,14 @@ pub struct RunStatusTransitionProps {
pub reason: Option<StatusReason>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct RunSubmittedProps {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub reason: Option<StatusReason>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub definition_blob: Option<RunBlobId>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct RunControlRequestedProps {
pub action: RunControlAction,

25
lib/crates/fabro-workflow/src/event.rs
View file

@ -5,7 +5,7 @@ use std::sync::atomic::{AtomicI64, Ordering};
use ::fabro_types::run_event as fabro_types;
use ::fabro_types::{
BilledTokenCounts, RunControlAction, RunEvent, RunId, StageStatus, StatusReason,
BilledTokenCounts, RunBlobId, RunControlAction, RunEvent, RunId, StageStatus, StatusReason,
};
use anyhow::{Context, Result};
use chrono::Utc;
@ -57,6 +57,8 @@ pub enum Event {
artifact_storage: Option<::fabro_types::RunArtifactStorage>,
#[serde(default, skip_serializing_if = "Option::is_none")]
provenance: Option<::fabro_types::RunProvenance>,
#[serde(default, skip_serializing_if = "Option::is_none")]
manifest_blob: Option<RunBlobId>,
},
WorkflowRunStarted {
name: String,
@ -75,6 +77,8 @@ pub enum Event {
RunSubmitted {
#[serde(default, skip_serializing_if = "Option::is_none")]
reason: Option<StatusReason>,
#[serde(default, skip_serializing_if = "Option::is_none")]
definition_blob: Option<RunBlobId>,
},
RunStarting {
#[serde(default, skip_serializing_if = "Option::is_none")]
@ -551,8 +555,11 @@ impl Event {
Self::WorkflowRunStarted { name, run_id, .. } => {
info!(workflow = name.as_str(), run_id = %run_id, "Workflow run started");
}
Self::RunSubmitted { reason } => {
info!(?reason, "Run submitted");
Self::RunSubmitted {
reason,
definition_blob,
} => {
info!(?reason, ?definition_blob, "Run submitted");
}
Self::RunStarting { reason } => {
info!(?reason, "Run starting");
@ -1392,6 +1399,7 @@ fn event_body_from_event(event: &Event) -> EventBody {
db_prefix,
artifact_storage,
provenance,
manifest_blob,
..
} => EventBody::RunCreated(fabro_types::RunCreatedProps {
settings: serde_json::from_value(settings.clone()).expect("run.created settings"),
@ -1408,6 +1416,7 @@ fn event_body_from_event(event: &Event) -> EventBody {
db_prefix: db_prefix.clone(),
artifact_storage: *artifact_storage,
provenance: provenance.clone(),
manifest_blob: *manifest_blob,
}),
Event::WorkflowRunStarted {
name,
@ -1425,9 +1434,13 @@ fn event_body_from_event(event: &Event) -> EventBody {
worktree_dir: worktree_dir.clone(),
goal: goal.clone(),
}),
Event::RunSubmitted { reason } => {
EventBody::RunSubmitted(fabro_types::RunStatusTransitionProps { reason: *reason })
}
Event::RunSubmitted {
reason,
definition_blob,
} => EventBody::RunSubmitted(fabro_types::RunSubmittedProps {
reason: *reason,
definition_blob: *definition_blob,
}),
Event::RunStarting { reason } => {
EventBody::RunStarting(fabro_types::RunStatusTransitionProps { reason: *reason })
}

64
lib/crates/fabro-workflow/src/operations/create.rs
View file

@ -16,7 +16,7 @@ use crate::pipeline::{self, Persisted, TransformOptions, Validated};
use crate::records::RunRecord;
use crate::run_lookup::default_scratch_base;
use crate::transforms::{Transform, expand_vars};
use crate::workflow_bundle::{StoredWorkflowBundle, WorkflowBundle};
use crate::workflow_bundle::{RunDefinition, WorkflowBundle};
use fabro_sandbox::daytona::detect_repo_info;
use fabro_util::json::normalize_json_value;
@ -31,6 +31,7 @@ pub struct CreateRunInput {
pub workflow_slug: Option<String>,
pub workflow_path: Option<PathBuf>,
pub workflow_bundle: Option<WorkflowBundle>,
pub submitted_manifest_bytes: Option<Vec<u8>>,
pub run_id: Option<RunId>,
pub host_repo_path: Option<String>,
pub repo_origin_url: Option<String>,
@ -81,6 +82,7 @@ pub async fn create(store: &Database, request: CreateRunInput) -> Result<Created
workflow_slug,
workflow_path,
workflow_bundle,
submitted_manifest_bytes,
run_id,
host_repo_path,
repo_origin_url,
@ -111,6 +113,13 @@ pub async fn create(store: &Database, request: CreateRunInput) -> Result<Created
let goal_override = resolved.goal_override.clone();
let current_dir = resolved.current_dir.clone();
let file_resolver = resolved.file_resolver.clone();
let accepted_definition = match (&workflow_path, &workflow_bundle) {
(Some(workflow_path), Some(workflow_bundle)) => Some(RunDefinition::new(
workflow_path.clone(),
workflow_bundle.clone(),
)),
_ => None,
};
let persisted = create_from_source(
&resolved.raw_source,
@ -136,10 +145,15 @@ pub async fn create(store: &Database, request: CreateRunInput) -> Result<Created
.workflow_toml_path
.as_deref()
.and_then(|path| std::fs::read_to_string(path).ok());
persist_created_run(store, &persisted, &resolved.raw_source, workflow_config).await?;
if let (Some(workflow_path), Some(workflow_bundle)) = (workflow_path, workflow_bundle) {
persist_workflow_bundle(persisted.run_dir(), workflow_path, workflow_bundle)?;
}
persist_created_run(
store,
&persisted,
&resolved.raw_source,
workflow_config,
submitted_manifest_bytes.as_deref(),
accepted_definition.as_ref(),
)
.await?;
Ok(CreatedRun {
persisted,
@ -154,6 +168,8 @@ async fn persist_created_run(
persisted: &Persisted,
workflow_source: &str,
workflow_config: Option<String>,
submitted_manifest_bytes: Option<&[u8]>,
accepted_definition: Option<&RunDefinition>,
) -> Result<(), FabroError> {
let record = persisted.run_record();
let run_store = match store.create_run(&record.run_id).await {
@ -164,6 +180,18 @@ async fn persist_created_run(
.map_err(|open_err| FabroError::engine(open_err.to_string()))
.map_err(|_| FabroError::engine(err.to_string()))?,
};
let manifest_blob = match submitted_manifest_bytes {
Some(bytes) => Some(run_store.write_blob(bytes).await.map_err(store_error)?),
None => None,
};
let definition_blob = match accepted_definition {
Some(definition) => {
let bytes = serde_json::to_vec(definition)
.map_err(|err| FabroError::engine(err.to_string()))?;
Some(run_store.write_blob(&bytes).await.map_err(store_error)?)
}
None => None,
};
let stored = to_run_event_at(
&record.run_id,
@ -193,6 +221,7 @@ async fn persist_created_run(
db_prefix: None,
artifact_storage: record.artifact_storage,
provenance: record.provenance.clone(),
manifest_blob,
},
record.run_id.created_at(),
);
@ -209,7 +238,10 @@ async fn persist_created_run(
append_event(
&run_store,
&record.run_id,
&Event::RunSubmitted { reason: None },
&Event::RunSubmitted {
reason: None,
definition_blob,
},
)
.await
.map_err(store_error)
@ -219,18 +251,6 @@ fn store_error(err: impl std::fmt::Display) -> FabroError {
FabroError::engine(err.to_string())
}
fn persist_workflow_bundle(
run_dir: &Path,
workflow_path: PathBuf,
workflow_bundle: WorkflowBundle,
) -> Result<(), FabroError> {
let path = run_dir.join("workflow_bundle.json");
let payload =
serde_json::to_string_pretty(&StoredWorkflowBundle::new(workflow_path, workflow_bundle))
.map_err(|err| FabroError::engine(err.to_string()))?;
std::fs::write(&path, payload).map_err(|err| FabroError::Io(err.to_string()))
}
fn validate_sandbox_provider(settings: &Settings) -> Result<(), FabroError> {
if let Some(provider) = settings
.sandbox_settings()
@ -345,6 +365,8 @@ fn persist_validated(
labels,
artifact_storage,
provenance,
manifest_blob: None,
definition_blob: None,
};
pipeline::persist(
@ -699,6 +721,7 @@ mod tests {
workflow_slug: None,
workflow_path: None,
workflow_bundle: None,
submitted_manifest_bytes: None,
run_id: None,
host_repo_path: None,
repo_origin_url: None,
@ -748,6 +771,7 @@ mod tests {
workflow_slug: Some("slug".to_string()),
workflow_path: None,
workflow_bundle: None,
submitted_manifest_bytes: None,
run_id: Some(fixtures::RUN_1),
host_repo_path: Some(dir.path().display().to_string()),
repo_origin_url: None,
@ -829,6 +853,7 @@ mod tests {
workflow_slug: None,
workflow_path: None,
workflow_bundle: None,
submitted_manifest_bytes: None,
run_id: Some(fixtures::RUN_2),
host_repo_path: None,
repo_origin_url: None,
@ -873,6 +898,7 @@ mod tests {
workflow_slug: None,
workflow_path: None,
workflow_bundle: None,
submitted_manifest_bytes: None,
run_id: Some(fixtures::RUN_2),
host_repo_path: None,
repo_origin_url: Some("https://github.com/acme/widgets".to_string()),
@ -914,6 +940,7 @@ mod tests {
workflow_slug: Some("slug".to_string()),
workflow_path: None,
workflow_bundle: None,
submitted_manifest_bytes: None,
run_id: Some(fixtures::RUN_3),
host_repo_path: None,
repo_origin_url: None,
@ -957,6 +984,7 @@ mod tests {
workflow_slug: Some("slug".to_string()),
workflow_path: None,
workflow_bundle: None,
submitted_manifest_bytes: None,
run_id: Some(fixtures::RUN_64),
host_repo_path: None,
repo_origin_url: None,

3
lib/crates/fabro-workflow/src/operations/rebuild_meta.rs
View file

@ -380,6 +380,8 @@ mod tests {
labels: HashMap::new(),
artifact_storage: None,
provenance: None,
manifest_blob: None,
definition_blob: None,
}
}
@ -455,6 +457,7 @@ mod tests {
db_prefix: None,
artifact_storage: run_record.artifact_storage,
provenance: run_record.provenance.clone(),
manifest_blob: None,
},
)
.await

6
lib/crates/fabro-workflow/src/operations/resume.rs
View file

@ -36,12 +36,16 @@ pub async fn resume(run_dir: &Path, services: StartServices) -> Result<Started,
let checkpoint = state
.checkpoint
.ok_or_else(|| FabroError::Precondition("no checkpoint to resume from".to_string()))?;
let definition_blob = state.run.as_ref().and_then(|run| run.definition_blob);
cleanup_resume_artifacts(run_dir);
append_event_to_sink(
&services.event_sink,
&services.run_id,
&Event::RunSubmitted { reason: None },
&Event::RunSubmitted {
reason: None,
definition_blob,
},
)
.await
.map_err(|err| FabroError::engine(err.to_string()))?;

124
lib/crates/fabro-workflow/src/operations/start.rs
View file

@ -30,7 +30,7 @@ use crate::run_control::RunControlState;
use crate::run_options::{GitCheckpointOptions, LifecycleOptions, RunOptions};
use crate::run_status::{RunStatus, StatusReason};
use crate::runtime_store::RunStoreHandle;
use crate::workflow_bundle::{StoredWorkflowBundle, WorkflowBundle};
use crate::workflow_bundle::{RunDefinition, WorkflowBundle};
use fabro_config::run::PullRequestSettings;
use fabro_retro::retro::Retro;
use fabro_sandbox::daytona::DaytonaConfig;
@ -274,12 +274,18 @@ impl RunSession {
meta_branch: Some(MetadataStore::branch_name(&record.run_id.to_string())),
})
});
let stored_workflow_bundle = StoredWorkflowBundle::load_from_run_dir(persisted.run_dir())?;
let workflow_path = stored_workflow_bundle
let definition_blob = state.run.as_ref().and_then(|run| run.definition_blob);
let accepted_definition = match definition_blob {
Some(blob_id) => {
Some(load_accepted_run_definition(&services.run_store, blob_id).await?)
}
None => None,
};
let workflow_path = accepted_definition
.as_ref()
.map(|bundle| bundle.workflow_path.clone());
.map(|definition| definition.workflow_path.clone());
let workflow_bundle =
stored_workflow_bundle.map(|bundle| Arc::new(bundle.workflow_bundle()));
accepted_definition.map(|definition| Arc::new(definition.workflow_bundle()));
if let Some(env) = settings
.sandbox
@ -413,6 +419,22 @@ impl RunSession {
}
}
async fn load_accepted_run_definition(
run_store: &RunStoreHandle,
blob_id: fabro_types::RunBlobId,
) -> Result<RunDefinition, FabroError> {
let bytes = run_store
.read_blob(&blob_id)
.await
.map_err(|err| FabroError::engine(err.to_string()))?
.ok_or_else(|| {
FabroError::engine(format!(
"run definition blob is missing from the run store: {blob_id}"
))
})?;
serde_json::from_slice(&bytes).map_err(|err| FabroError::Parse(err.to_string()))
}
fn resolve_sandbox_provider(settings: &Settings) -> Result<SandboxProvider, FabroError> {
settings
.sandbox_settings()
@ -803,9 +825,11 @@ mod tests {
use crate::event::Emitter;
use crate::handler::HandlerRegistry;
use crate::handler::exit::ExitHandler;
use crate::handler::manager_loop::SubWorkflowHandler;
use crate::handler::start::StartHandler;
use crate::operations::resume;
use crate::records::CheckpointExt;
use crate::workflow_bundle::{BundledWorkflow, WorkflowBundle};
const MINIMAL_DOT: &str = r#"digraph Test {
graph [goal="Build feature"]
@ -842,6 +866,7 @@ mod tests {
workflow_slug: Some("test".to_string()),
workflow_path: None,
workflow_bundle: None,
submitted_manifest_bytes: None,
run_id: Some(fixtures::RUN_1),
host_repo_path: None,
repo_origin_url: None,
@ -859,6 +884,7 @@ mod tests {
let mut registry = HandlerRegistry::new(Box::new(StartHandler));
registry.register("start", Box::new(StartHandler));
registry.register("exit", Box::new(ExitHandler));
registry.register("stack.manager_loop", Box::new(SubWorkflowHandler));
registry
}
@ -958,6 +984,94 @@ mod tests {
assert!(run_store.state().await.unwrap().conclusion.is_some());
}
#[tokio::test]
async fn start_can_run_bundle_backed_child_workflow_without_workflow_bundle_json() {
let temp = tempfile::tempdir().unwrap();
let run_dir = temp.path().join("run");
let emitter = Arc::new(Emitter::new(fixtures::RUN_1));
let registry = Arc::new(test_registry());
let store = memory_store();
let workflow_bundle = WorkflowBundle::new(HashMap::from([
(
PathBuf::from("workflow.fabro"),
BundledWorkflow {
logical_path: PathBuf::from("workflow.fabro"),
source: r#"digraph Root {
graph [goal="Bundle child"]
start [shape=Mdiamond]
manager [
type="stack.manager_loop",
stack.child_workflow="./children/review.fabro",
manager.max_cycles=100,
manager.poll_interval="10ms"
]
exit [shape=Msquare]
start -> manager -> exit
}"#
.to_string(),
files: HashMap::new(),
},
),
(
PathBuf::from("children/review.fabro"),
BundledWorkflow {
logical_path: PathBuf::from("children/review.fabro"),
source: r#"digraph Review {
start [shape=Mdiamond]
exit [shape=Msquare]
start -> exit
}"#
.to_string(),
files: HashMap::new(),
},
),
]));
let created = crate::operations::create(
&store,
crate::operations::CreateRunInput {
workflow: crate::operations::WorkflowInput::Bundled(
workflow_bundle
.workflow(Path::new("workflow.fabro"))
.unwrap()
.clone(),
),
settings: Settings {
dry_run: Some(true),
..Default::default()
},
cwd: temp.path().to_path_buf(),
workflow_slug: Some("bundle-child".to_string()),
workflow_path: Some(PathBuf::from("workflow.fabro")),
workflow_bundle: Some(workflow_bundle),
submitted_manifest_bytes: None,
run_id: Some(fixtures::RUN_1),
host_repo_path: None,
repo_origin_url: None,
base_branch: None,
artifact_storage: None,
provenance: None,
},
)
.await
.unwrap();
let bundle_file = created.run_dir.join("workflow_bundle.json");
assert!(
!bundle_file.exists(),
"run scratch should not persist workflow_bundle.json"
);
let started = start(
&run_dir,
test_start_services(&store, &run_dir, emitter, registry).await,
)
.await
.unwrap();
assert_eq!(started.finalized.conclusion.status, StageStatus::Success);
}
#[tokio::test]
async fn start_invokes_on_node_callback_before_execution() {
let temp = tempfile::tempdir().unwrap();

2
lib/crates/fabro-workflow/src/pipeline/execute/tests.rs
View file

@ -146,6 +146,8 @@ fn persisted_workflow(graph: Graph, source: String, run_dir: &Path, run_id: RunI
labels: HashMap::new(),
artifact_storage: None,
provenance: None,
manifest_blob: None,
definition_blob: None,
},
)
}

2
lib/crates/fabro-workflow/src/pipeline/initialize.rs
View file

@ -751,6 +751,8 @@ mod tests {
labels: HashMap::new(),
artifact_storage: None,
provenance: None,
manifest_blob: None,
definition_blob: None,
},
)
}

3
lib/crates/fabro-workflow/src/pipeline/persist.rs
View file

@ -135,6 +135,8 @@ mod tests {
]),
artifact_storage: None,
provenance: None,
manifest_blob: None,
definition_blob: None,
}
}
@ -160,6 +162,7 @@ mod tests {
db_prefix: None,
artifact_storage: record.artifact_storage,
provenance: record.provenance.clone(),
manifest_blob: None,
},
)
.await

9
lib/crates/fabro-workflow/src/pipeline/pull_request.rs
View file

@ -1092,6 +1092,8 @@ mod tests {
labels: HashMap::new(),
artifact_storage: None,
provenance: None,
manifest_blob: None,
definition_blob: None,
};
append_event(
&run_store,
@ -1112,6 +1114,7 @@ mod tests {
db_prefix: None,
artifact_storage: run_record.artifact_storage,
provenance: run_record.provenance.clone(),
manifest_blob: None,
},
)
.await
@ -1164,6 +1167,8 @@ mod tests {
labels: HashMap::new(),
artifact_storage: None,
provenance: None,
manifest_blob: None,
definition_blob: None,
};
append_event(
&run_store,
@ -1184,6 +1189,7 @@ mod tests {
db_prefix: None,
artifact_storage: run_record.artifact_storage,
provenance: run_record.provenance.clone(),
manifest_blob: None,
},
)
.await
@ -1389,6 +1395,8 @@ mod tests {
labels: std::collections::HashMap::new(),
artifact_storage: None,
provenance: None,
manifest_blob: None,
definition_blob: None,
};
append_event(
&run_store,
@ -1409,6 +1417,7 @@ mod tests {
db_prefix: None,
artifact_storage: run_record.artifact_storage,
provenance: run_record.provenance.clone(),
manifest_blob: None,
},
)
.await

3
lib/crates/fabro-workflow/src/pipeline/retro.rs
View file

@ -243,6 +243,8 @@ mod tests {
labels: std::collections::HashMap::new(),
artifact_storage: None,
provenance: None,
manifest_blob: None,
definition_blob: None,
};
append_event(
&run_store,
@ -263,6 +265,7 @@ mod tests {
db_prefix: None,
artifact_storage: run_record.artifact_storage,
provenance: run_record.provenance.clone(),
manifest_blob: None,
},
)
.await

8
lib/crates/fabro-workflow/src/run_lookup.rs
View file

@ -425,6 +425,8 @@ mod tests {
labels: HashMap::new(),
artifact_storage: None,
provenance: None,
manifest_blob: None,
definition_blob: None,
}
}
@ -456,6 +458,7 @@ mod tests {
db_prefix: None,
artifact_storage: run_record.artifact_storage,
provenance: run_record.provenance.clone(),
manifest_blob: None,
},
)
.await
@ -463,7 +466,10 @@ mod tests {
append_event(
&run_store,
&fixtures::RUN_1,
&Event::RunSubmitted { reason: None },
&Event::RunSubmitted {
reason: None,
definition_blob: None,
},
)
.await
.unwrap();

10
lib/crates/fabro-workflow/src/runtime_store.rs
View file

@ -112,7 +112,7 @@ mod tests {
use fabro_graphviz::graph::Graph;
use fabro_store::Database;
use fabro_types::fixtures;
use fabro_types::run_event::RunStatusTransitionProps;
use fabro_types::run_event::RunSubmittedProps;
use fabro_types::{EventBody, RunEvent, Settings};
use object_store::memory::InMemory;
@ -142,6 +142,8 @@ mod tests {
labels: HashMap::new(),
artifact_storage: None,
provenance: None,
manifest_blob: None,
definition_blob: None,
}
}
@ -168,6 +170,7 @@ mod tests {
db_prefix: None,
artifact_storage: None,
provenance: None,
manifest_blob: None,
},
)
.await
@ -194,7 +197,10 @@ mod tests {
node_label: None,
session_id: None,
parent_session_id: None,
body: EventBody::RunSubmitted(RunStatusTransitionProps { reason: None }),
body: EventBody::RunSubmitted(RunSubmittedProps {
reason: None,
definition_blob: None,
}),
};
handle.append_run_event(&event).await.unwrap();

1
lib/crates/fabro-workflow/src/test_support.rs
View file

@ -99,6 +99,7 @@ async fn initialized(
db_prefix: None,
artifact_storage: None,
provenance: None,
manifest_blob: None,
},
)
.await

18
lib/crates/fabro-workflow/src/workflow_bundle.rs
View file

@ -4,7 +4,6 @@ use std::sync::Arc;
use serde::{Deserialize, Serialize};
use crate::error::FabroError;
use crate::file_resolver::{BundleFileResolver, FileResolver, normalize_logical_path};
#[derive(Clone, Debug, Default, Serialize, Deserialize)]
@ -62,12 +61,12 @@ impl WorkflowBundle {
}
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct StoredWorkflowBundle {
pub struct RunDefinition {
pub workflow_path: PathBuf,
pub workflows: HashMap<PathBuf, BundledWorkflow>,
}
impl StoredWorkflowBundle {
impl RunDefinition {
#[must_use]
pub fn new(workflow_path: PathBuf, bundle: WorkflowBundle) -> Self {
Self {
@ -80,17 +79,4 @@ impl StoredWorkflowBundle {
pub fn workflow_bundle(&self) -> WorkflowBundle {
WorkflowBundle::new(self.workflows.clone())
}
pub fn load_from_run_dir(run_dir: &Path) -> Result<Option<Self>, FabroError> {
let path = run_dir.join("workflow_bundle.json");
if !path.exists() {
return Ok(None);
}
let payload =
std::fs::read_to_string(&path).map_err(|err| FabroError::Io(err.to_string()))?;
serde_json::from_str(&payload)
.map(Some)
.map_err(|err| FabroError::Parse(err.to_string()))
}
}