scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-11-01 07:02:30 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

Merge pull request #1088 from fabro-sh/codex/attach-json-event-order

Browse source 
test: stabilize run stream snapshots across valid interleavings
This commit is contained in:
Scott Werner 2026-10-30 11:10:59 -04:00 • committed by GitHub
commit 166c89e402
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 344 additions and 199 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

280
lib/apps/fabro-cli/tests/it/cmd/attach.rs
View file

@ -73,6 +73,9 @@ fn format_output_snapshot(output: &Output, filters: &[(String, String)]) -> Stri
}
fn normalize_attach_json_progress_event(mut event: Value) -> Value {
// Delivery positions depend on how the projector interleaves the logs.
// The caller checks the untouched envelopes against the recorded stream.
event["stream_seq"] = Value::String("[STREAM_SEQ]".to_string());
// The `run.created` record carries the whole run spec, whose graph
// and settings vary with the fixture's socket path and node order;
// the test does not check it.
@ -338,11 +341,35 @@ fn attach_json_emits_pending_question_once_across_replay_and_live_boundaries() {
items.truncate(question_index + 1);
assert!(question_index > 4);
for (boundary, replay_len, initially_pending) in [
("between replay and pending check", 4, true),
("in replay", items.len(), true),
("in live stream", 4, false),
for (boundary, replay_len, initially_pending, swap_startup) in [
("between replay and pending check", 4, true, false),
("in replay", items.len(), true, false),
("in live stream", 4, false, false),
(
"with the other startup interleaving",
items.len(),
true,
true,
),
] {
let mut items = items.clone();
if swap_startup {
let running = items
.iter()
.position(|item| {
item.pointer("/item/record/transition") == Some(&Value::from("running"))
})
.expect("the platform records the running lifecycle");
let coordinator = items
.iter()
.position(|item| item["id"] == "coordinator/0/0")
.expect("Petri records the coordinator start");
let running_seq = items[running]["stream_seq"].clone();
let coordinator_seq = items[coordinator]["stream_seq"].clone();
items.swap(running, coordinator);
items[running]["stream_seq"] = running_seq;
items[coordinator]["stream_seq"] = coordinator_seq;
}
let server = MockServer::start();
let resolve = server.mock(|when, then| {
when.method("GET").path("/api/v1/runs/resolve");
@ -970,13 +997,40 @@ 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"))
.collect();
assert_eq!(
log_events.get(..progress.len()),
Some(progress.as_slice()),
"attach must preserve every recorded envelope in delivery order"
);
assert!(
progress.iter().any(|item| {
item.pointer("/item/derived/parsed/kind") == Some(&Value::from("question"))
}),
"attach must include the pending question before exiting"
);
for (index, item) in progress.iter().enumerate() {
assert_eq!(
item["stream_seq"],
Value::from(index + 1),
"attach must emit contiguous delivery positions from the start"
);
}
// Platform and Petri records can share a millisecond timestamp, so their
// interleaving varies with projection pass boundaries. Snapshot each
// source's order and full payload; the raw comparison above checks their
// actual interleaving, cursors, and absence of skipped or duplicate items.
let mut progress: Vec<Value> = progress
.into_iter()
.map(normalize_attach_json_progress_event)
.collect();
progress.sort_by_key(|item| item["kind"] != "platform");
fabro_json_snapshot!(context, &progress, @r#"
[
{
"run_id": "[ULID]",
"stream_seq": 1,
"stream_seq": "[STREAM_SEQ]",
"kind": "platform",
"id": "[EVENT_ID]",
"recorded_at": "[EPOCH_MS]",
@ -993,7 +1047,7 @@ fn attach_json_errors_without_prompting_for_human_input() {
},
{
"run_id": "[ULID]",
"stream_seq": 2,
"stream_seq": "[STREAM_SEQ]",
"kind": "platform",
"id": "[EVENT_ID]",
"recorded_at": "[EPOCH_MS]",
@ -1011,7 +1065,7 @@ fn attach_json_errors_without_prompting_for_human_input() {
},
{
"run_id": "[ULID]",
"stream_seq": 3,
"stream_seq": "[STREAM_SEQ]",
"kind": "platform",
"id": "[EVENT_ID]",
"recorded_at": "[EPOCH_MS]",
@ -1027,7 +1081,7 @@ fn attach_json_errors_without_prompting_for_human_input() {
},
{
"run_id": "[ULID]",
"stream_seq": 4,
"stream_seq": "[STREAM_SEQ]",
"kind": "platform",
"id": "[EVENT_ID]",
"recorded_at": "[EPOCH_MS]",
@ -1046,7 +1100,7 @@ fn attach_json_errors_without_prompting_for_human_input() {
},
{
"run_id": "[ULID]",
"stream_seq": 5,
"stream_seq": "[STREAM_SEQ]",
"kind": "platform",
"id": "[EVENT_ID]",
"recorded_at": "[EPOCH_MS]",
@ -1064,7 +1118,7 @@ fn attach_json_errors_without_prompting_for_human_input() {
},
{
"run_id": "[ULID]",
"stream_seq": 6,
"stream_seq": "[STREAM_SEQ]",
"kind": "platform",
"id": "[EVENT_ID]",
"recorded_at": "[EPOCH_MS]",
@ -1082,7 +1136,82 @@ fn attach_json_errors_without_prompting_for_human_input() {
},
{
"run_id": "[ULID]",
"stream_seq": 7,
"stream_seq": "[STREAM_SEQ]",
"kind": "platform",
"id": "[EVENT_ID]",
"recorded_at": "[EPOCH_MS]",
"item": {
"seq": 7,
"recorded_at": "[EPOCH_MS]",
"record": {
"kind": "run.branch",
"run_branch": "fabro/run/[ULID]",
"base_sha": "[DIGEST]",
"workspace": "invocation-0-scope-0"
},
"position": {
"execution": 0,
"firing": 1
}
}
},
{
"run_id": "[ULID]",
"stream_seq": "[STREAM_SEQ]",
"kind": "platform",
"id": "[EVENT_ID]",
"recorded_at": "[EPOCH_MS]",
"item": {
"seq": 8,
"recorded_at": "[EPOCH_MS]",
"record": {
"kind": "git.identity",
"name": "Fabro",
"email": "noreply@fabro.sh",
"source": "default"
},
"position": {
"execution": 0,
"firing": 1
}
}
},
{
"run_id": "[ULID]",
"stream_seq": "[STREAM_SEQ]",
"kind": "platform",
"id": "[EVENT_ID]",
"recorded_at": "[EPOCH_MS]",
"item": {
"seq": 9,
"recorded_at": "[EPOCH_MS]",
"record": {
"kind": "checkpoint",
"execution": 0,
"firing": 1,
"attempt": 1,
"workspace": "invocation-0-scope-0",
"git_commit_sha": "[DIGEST]",
"operation": {
"execution": 0,
"decision": {
"attempt_start": {
"firing": 1,
"attempt": 1
}
},
"effect": "checkpoint"
}
},
"position": {
"execution": 0,
"firing": 1
}
}
},
{
"run_id": "[ULID]",
"stream_seq": "[STREAM_SEQ]",
"kind": "petri",
"id": "coordinator/0/0",
"recorded_at": "[EPOCH_MS]",
@ -1116,7 +1245,7 @@ fn attach_json_errors_without_prompting_for_human_input() {
},
{
"run_id": "[ULID]",
"stream_seq": 8,
"stream_seq": "[STREAM_SEQ]",
"kind": "petri",
"id": "coordinator/1/0",
"recorded_at": "[EPOCH_MS]",
@ -1142,7 +1271,7 @@ fn attach_json_errors_without_prompting_for_human_input() {
},
{
"run_id": "[ULID]",
"stream_seq": 9,
"stream_seq": "[STREAM_SEQ]",
"kind": "petri",
"id": "coordinator/2/0",
"recorded_at": "[EPOCH_MS]",
@ -1175,7 +1304,7 @@ fn attach_json_errors_without_prompting_for_human_input() {
},
{
"run_id": "[ULID]",
"stream_seq": 10,
"stream_seq": "[STREAM_SEQ]",
"kind": "petri",
"id": "coordinator/3/0",
"recorded_at": "[EPOCH_MS]",
@ -1223,7 +1352,7 @@ fn attach_json_errors_without_prompting_for_human_input() {
},
{
"run_id": "[ULID]",
"stream_seq": 11,
"stream_seq": "[STREAM_SEQ]",
"kind": "petri",
"id": "execution 0/0/0",
"recorded_at": "[EPOCH_MS]",
@ -1257,7 +1386,7 @@ fn attach_json_errors_without_prompting_for_human_input() {
},
{
"run_id": "[ULID]",
"stream_seq": 12,
"stream_seq": "[STREAM_SEQ]",
"kind": "petri",
"id": "execution 0/1/0",
"recorded_at": "[EPOCH_MS]",
@ -1289,7 +1418,7 @@ fn attach_json_errors_without_prompting_for_human_input() {
},
{
"run_id": "[ULID]",
"stream_seq": 13,
"stream_seq": "[STREAM_SEQ]",
"kind": "petri",
"id": "execution 0/1/1",
"recorded_at": "[EPOCH_MS]",
@ -1352,7 +1481,7 @@ fn attach_json_errors_without_prompting_for_human_input() {
},
{
"run_id": "[ULID]",
"stream_seq": 14,
"stream_seq": "[STREAM_SEQ]",
"kind": "petri",
"id": "execution 0/1/2",
"recorded_at": "[EPOCH_MS]",
@ -1408,7 +1537,7 @@ fn attach_json_errors_without_prompting_for_human_input() {
},
{
"run_id": "[ULID]",
"stream_seq": 15,
"stream_seq": "[STREAM_SEQ]",
"kind": "petri",
"id": "execution 0/2/0",
"recorded_at": "[EPOCH_MS]",
@ -1441,7 +1570,7 @@ fn attach_json_errors_without_prompting_for_human_input() {
},
{
"run_id": "[ULID]",
"stream_seq": 16,
"stream_seq": "[STREAM_SEQ]",
"kind": "petri",
"id": "execution 0/3/0",
"recorded_at": "[EPOCH_MS]",
@ -1509,7 +1638,7 @@ fn attach_json_errors_without_prompting_for_human_input() {
},
{
"run_id": "[ULID]",
"stream_seq": 17,
"stream_seq": "[STREAM_SEQ]",
"kind": "petri",
"id": "execution 0/4/0",
"recorded_at": "[EPOCH_MS]",
@ -1545,7 +1674,7 @@ fn attach_json_errors_without_prompting_for_human_input() {
},
{
"run_id": "[ULID]",
"stream_seq": 18,
"stream_seq": "[STREAM_SEQ]",
"kind": "petri",
"id": "execution 0/5/0",
"recorded_at": "[EPOCH_MS]",
@ -1607,7 +1736,7 @@ fn attach_json_errors_without_prompting_for_human_input() {
},
{
"run_id": "[ULID]",
"stream_seq": 19,
"stream_seq": "[STREAM_SEQ]",
"kind": "petri",
"id": "execution 0/5/1",
"recorded_at": "[EPOCH_MS]",
@ -1663,7 +1792,7 @@ fn attach_json_errors_without_prompting_for_human_input() {
},
{
"run_id": "[ULID]",
"stream_seq": 20,
"stream_seq": "[STREAM_SEQ]",
"kind": "petri",
"id": "execution 0/6/0",
"recorded_at": "[EPOCH_MS]",
@ -1730,7 +1859,7 @@ fn attach_json_errors_without_prompting_for_human_input() {
},
{
"run_id": "[ULID]",
"stream_seq": 21,
"stream_seq": "[STREAM_SEQ]",
"kind": "petri",
"id": "execution 0/7/0",
"recorded_at": "[EPOCH_MS]",
@ -1822,7 +1951,7 @@ fn attach_json_errors_without_prompting_for_human_input() {
},
{
"run_id": "[ULID]",
"stream_seq": 22,
"stream_seq": "[STREAM_SEQ]",
"kind": "petri",
"id": "execution 0/8/0",
"recorded_at": "[EPOCH_MS]",
@ -1902,7 +2031,7 @@ fn attach_json_errors_without_prompting_for_human_input() {
},
{
"run_id": "[ULID]",
"stream_seq": 23,
"stream_seq": "[STREAM_SEQ]",
"kind": "petri",
"id": "execution 0/8/1",
"recorded_at": "[EPOCH_MS]",
@ -1973,82 +2102,7 @@ fn attach_json_errors_without_prompting_for_human_input() {
},
{
"run_id": "[ULID]",
"stream_seq": 24,
"kind": "platform",
"id": "[EVENT_ID]",
"recorded_at": "[EPOCH_MS]",
"item": {
"seq": 7,
"recorded_at": "[EPOCH_MS]",
"record": {
"kind": "run.branch",
"run_branch": "fabro/run/[ULID]",
"base_sha": "[DIGEST]",
"workspace": "invocation-0-scope-0"
},
"position": {
"execution": 0,
"firing": 1
}
}
},
{
"run_id": "[ULID]",
"stream_seq": 25,
"kind": "platform",
"id": "[EVENT_ID]",
"recorded_at": "[EPOCH_MS]",
"item": {
"seq": 8,
"recorded_at": "[EPOCH_MS]",
"record": {
"kind": "git.identity",
"name": "Fabro",
"email": "noreply@fabro.sh",
"source": "default"
},
"position": {
"execution": 0,
"firing": 1
}
}
},
{
"run_id": "[ULID]",
"stream_seq": 26,
"kind": "platform",
"id": "[EVENT_ID]",
"recorded_at": "[EPOCH_MS]",
"item": {
"seq": 9,
"recorded_at": "[EPOCH_MS]",
"record": {
"kind": "checkpoint",
"execution": 0,
"firing": 1,
"attempt": 1,
"workspace": "invocation-0-scope-0",
"git_commit_sha": "[DIGEST]",
"operation": {
"execution": 0,
"decision": {
"attempt_start": {
"firing": 1,
"attempt": 1
}
},
"effect": "checkpoint"
}
},
"position": {
"execution": 0,
"firing": 1
}
}
},
{
"run_id": "[ULID]",
"stream_seq": 27,
"stream_seq": "[STREAM_SEQ]",
"kind": "petri",
"id": "execution 0/9/0",
"recorded_at": "[EPOCH_MS]",
@ -2156,7 +2210,7 @@ fn attach_json_errors_without_prompting_for_human_input() {
},
{
"run_id": "[ULID]",
"stream_seq": 28,
"stream_seq": "[STREAM_SEQ]",
"kind": "petri",
"id": "execution 0/9/1",
"recorded_at": "[EPOCH_MS]",
@ -2225,7 +2279,7 @@ fn attach_json_errors_without_prompting_for_human_input() {
},
{
"run_id": "[ULID]",
"stream_seq": 29,
"stream_seq": "[STREAM_SEQ]",
"kind": "petri",
"id": "execution 0/9/2",
"recorded_at": "[EPOCH_MS]",
@ -2284,7 +2338,7 @@ fn attach_json_errors_without_prompting_for_human_input() {
},
{
"run_id": "[ULID]",
"stream_seq": 30,
"stream_seq": "[STREAM_SEQ]",
"kind": "petri",
"id": "execution 0/10/0",
"recorded_at": "[EPOCH_MS]",
@ -2377,7 +2431,7 @@ fn attach_json_errors_without_prompting_for_human_input() {
},
{
"run_id": "[ULID]",
"stream_seq": 31,
"stream_seq": "[STREAM_SEQ]",
"kind": "petri",
"id": "execution 0/11/0",
"recorded_at": "[EPOCH_MS]",
@ -2444,7 +2498,7 @@ fn attach_json_errors_without_prompting_for_human_input() {
},
{
"run_id": "[ULID]",
"stream_seq": 32,
"stream_seq": "[STREAM_SEQ]",
"kind": "petri",
"id": "execution 0/12/0",
"recorded_at": "[EPOCH_MS]",
@ -2515,7 +2569,7 @@ fn attach_json_errors_without_prompting_for_human_input() {
},
{
"run_id": "[ULID]",
"stream_seq": 33,
"stream_seq": "[STREAM_SEQ]",
"kind": "petri",
"id": "execution 0/13/0",
"recorded_at": "[EPOCH_MS]",
@ -2580,7 +2634,7 @@ fn attach_json_errors_without_prompting_for_human_input() {
},
{
"run_id": "[ULID]",
"stream_seq": 34,
"stream_seq": "[STREAM_SEQ]",
"kind": "petri",
"id": "execution 0/13/1",
"recorded_at": "[EPOCH_MS]",
@ -2639,7 +2693,7 @@ fn attach_json_errors_without_prompting_for_human_input() {
},
{
"run_id": "[ULID]",
"stream_seq": 35,
"stream_seq": "[STREAM_SEQ]",
"kind": "petri",
"id": "execution 0/14/0",
"recorded_at": "[EPOCH_MS]",
@ -2746,7 +2800,7 @@ fn attach_json_errors_without_prompting_for_human_input() {
},
{
"run_id": "[ULID]",
"stream_seq": 36,
"stream_seq": "[STREAM_SEQ]",
"kind": "petri",
"id": "execution 0/14/1",
"recorded_at": "[EPOCH_MS]",
@ -2805,7 +2859,7 @@ fn attach_json_errors_without_prompting_for_human_input() {
},
{
"run_id": "[ULID]",
"stream_seq": 37,
"stream_seq": "[STREAM_SEQ]",
"kind": "petri",
"id": "execution 0/15/0",
"recorded_at": "[EPOCH_MS]",

215
lib/apps/fabro-cli/tests/it/cmd/events.rs
View file

@ -1,7 +1,12 @@
use fabro_test::{fabro_snapshot, json_elapsed_ms_snapshot_filters, test_context};
use fabro_test::{apply_filters, fabro_snapshot, test_context};
use httpmock::MockServer;
use serde_json::Value;
use super::support::{setup_detached_dry_run, setup_seeded_completed_dry_run};
use super::support::{
remote_run_summary_json, run_stream_items, setup_detached_dry_run,
setup_seeded_completed_dry_run,
};
use crate::support::{pretty_stream_snapshot_filters, run_projection_json, unique_run_id};
fn parse_ndjson(stdout: &[u8]) -> Vec<Value> {
String::from_utf8(stdout.to_vec())
@ -99,6 +104,12 @@ fn events_completed_run_outputs_raw_ndjson() {
let events = parse_ndjson(&output.stdout);
assert_events_belong_to_run(&events, &run.run_id);
let recorded: Vec<Value> = run_stream_items(&run.run_dir)
.into_iter()
.map(|item| serde_json::to_value(item).expect("stream item serializes"))
.collect();
assert_eq!(events, recorded, "events must preserve the recorded stream");
assert!(output.stderr.is_empty(), "{output:?}");
assert_event_sequence_contains(&events, &[
"run.created",
"run.lifecycle",
@ -111,64 +122,29 @@ fn events_completed_run_outputs_raw_ndjson() {
}
#[test]
fn events_completed_run_reads_store_without_progress_jsonl() {
fn events_tail_matches_recorded_stream_suffix() {
let context = test_context!();
let run = setup_seeded_completed_dry_run(&context);
let expected: Vec<Value> = run_stream_items(&run.run_dir)
.into_iter()
.map(|item| serde_json::to_value(item).expect("stream item serializes"))
.collect();
assert!(expected.len() > 2, "the fixture must exercise truncation");
let mut filters = json_elapsed_ms_snapshot_filters(context.filters());
filters.push((
r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?Z".to_string(),
"[TIMESTAMP]".to_string(),
));
filters.push((
r#""id":"[0-9a-f-]+""#.to_string(),
r#""id":"[EVENT_ID]""#.to_string(),
));
filters.push((
r#""recorded_at":\d{13}"#.to_string(),
r#""recorded_at":[EPOCH_MS]"#.to_string(),
));
let mut cmd = context.command();
cmd.args(["events", "--tail", "2", &run.run_id]);
fabro_snapshot!(filters, cmd, @r#"
success: true
exit_code: 0
----- stdout -----
{"run_id":"[ULID]","stream_seq":67,"kind":"petri","id":"coordinator/7/0","recorded_at":[EPOCH_MS],"item":{"id":{"log":"coordinator","seq":7,"index":0},"origin":"external","context":{},"recorded_at":[EPOCH_MS],"record":{"seq":7,"origin":"external","recorded_at":[EPOCH_MS],"body":{"event":"run.finished","status":"success"}}}}
{"run_id":"[ULID]","stream_seq":68,"kind":"platform","id":"[EVENT_ID]","recorded_at":[EPOCH_MS],"item":{"seq":14,"recorded_at":[EPOCH_MS],"record":{"kind":"run.lifecycle","transition":"succeeded","status":{"kind":"succeeded","reason":"completed"}}}}
----- stderr -----
"#);
}
#[test]
fn events_tail_limits_output() {
let context = test_context!();
let run = setup_seeded_completed_dry_run(&context);
let mut filters = json_elapsed_ms_snapshot_filters(context.filters());
filters.push((
r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?Z".to_string(),
"[TIMESTAMP]".to_string(),
));
filters.push((
r#""id":"[0-9a-f-]+""#.to_string(),
r#""id":"[EVENT_ID]""#.to_string(),
));
filters.push((
r#""recorded_at":\d{13}"#.to_string(),
r#""recorded_at":[EPOCH_MS]"#.to_string(),
));
let mut cmd = context.command();
cmd.args(["events", "--tail", "2", &run.run_id]);
fabro_snapshot!(filters, cmd, @r#"
success: true
exit_code: 0
----- stdout -----
{"run_id":"[ULID]","stream_seq":67,"kind":"petri","id":"coordinator/7/0","recorded_at":[EPOCH_MS],"item":{"id":{"log":"coordinator","seq":7,"index":0},"origin":"external","context":{},"recorded_at":[EPOCH_MS],"record":{"seq":7,"origin":"external","recorded_at":[EPOCH_MS],"body":{"event":"run.finished","status":"success"}}}}
{"run_id":"[ULID]","stream_seq":68,"kind":"platform","id":"[EVENT_ID]","recorded_at":[EPOCH_MS],"item":{"seq":14,"recorded_at":[EPOCH_MS],"record":{"kind":"run.lifecycle","transition":"succeeded","status":{"kind":"succeeded","reason":"completed"}}}}
----- stderr -----
"#);
for tail in [0, 1, 2, expected.len() + 1] {
let output = context
.command()
.args(["events", "--tail", &tail.to_string(), &run.run_id])
.output()
.expect("events should execute");
assert!(output.status.success(), "{output:?}");
assert!(output.stderr.is_empty(), "{output:?}");
assert_eq!(
parse_ndjson(&output.stdout),
expected[expected.len().saturating_sub(tail)..],
"--tail {tail} must preserve the recorded suffix, including cursors and payloads"
);
}
}
#[test]
@ -190,16 +166,7 @@ fn events_since_filters_output() {
fn events_pretty_formats_small_run() {
let context = test_context!();
let run = setup_seeded_completed_dry_run(&context);
let mut filters = context.filters();
filters.push((r"\b\d{2}:\d{2}:\d{2}\b".to_string(), "[CLOCK]".to_string()));
filters.push((
r"\b\d+(\.\d+)?(ms|s)\b".to_string(),
"[DURATION]".to_string(),
));
filters.push((
r"Checkpoint [0-9a-f]{7}\b".to_string(),
"Checkpoint [SHA]".to_string(),
));
let filters = pretty_stream_snapshot_filters(&context);
let mut cmd = context.command();
cmd.args(["events", "--pretty", &run.run_id]);
@ -238,6 +205,120 @@ fn events_pretty_formats_small_run() {
");
}
#[test]
fn events_preserve_each_valid_startup_interleaving() {
let context = test_context!();
let run_id = unique_run_id();
let status = serde_json::json!({"kind": "running"});
let records = [
(
"platform",
"6",
serde_json::json!({
"seq": 6, "recorded_at": 0,
"record": {"kind": "run.lifecycle", "transition": "running", "status": status}
}),
),
(
"petri",
"coordinator/0/0",
serde_json::json!({
"record": {"body": {"event": "run.started"}}
}),
),
(
"petri",
"execution 0/0/1",
serde_json::json!({
"subject": {"node": {"name": "Start", "meta": {"kind": "start"}}},
"derived": {"event": "visit.started"}
}),
),
];
let rendered = [
"00:00:00 · running\n",
"00:00:00 Engine: petri run started\n",
"00:00:00 ▶ Start\n",
];
for order in [[0, 1, 2], [1, 0, 2], [1, 2, 0]] {
let items: Vec<Value> = order
.iter()
.enumerate()
.map(|(index, source)| {
let (kind, id, item) = &records[*source];
serde_json::json!({
"run_id": run_id, "stream_seq": index + 6, "kind": kind,
"id": id, "recorded_at": 0, "item": item
})
})
.collect();
let server = MockServer::start();
server.mock(|when, then| {
when.method("GET").path("/api/v1/runs/resolve");
then.status(200).json_body(remote_run_summary_json(
&run_id,
"Startup",
"startup",
"Startup",
&status,
"2026-10-30T12:00:00Z",
));
});
server.mock(|when, then| {
when.method("GET")
.path(format!("/api/v1/runs/{run_id}/state"));
then.status(200)
.json_body(run_projection_json(&run_id, &status));
});
server.mock(|when, then| {
when.method("GET")
.path(format!("/api/v1/runs/{run_id}/events"));
then.status(200).json_body(serde_json::json!({
"data": items, "meta": {"has_more": false}, "event_contract_version": 3
}));
});
let raw = context
.command()
.args(["events", "--server", &server.base_url(), &run_id])
.output()
.expect("events should execute");
assert!(raw.status.success(), "{raw:?}");
assert!(raw.stderr.is_empty(), "{raw:?}");
assert_eq!(parse_ndjson(&raw.stdout), items, "order {order:?}");
let pretty = context
.command()
.args([
"events",
"--pretty",
"--server",
&server.base_url(),
&run_id,
])
.output()
.expect("pretty events should execute");
assert!(pretty.status.success(), "{pretty:?}");
assert!(pretty.stderr.is_empty(), "{pretty:?}");
let stdout = String::from_utf8(pretty.stdout).expect("pretty output is UTF-8");
let expected: String = order.iter().map(|index| rendered[*index]).collect();
assert_eq!(
stdout, expected,
"pretty output must retain delivery order {order:?}"
);
// Canonicalize only the snapshot, after checking both real output modes.
insta::allow_duplicates! {
insta::assert_snapshot!(
apply_filters(&stdout, &pretty_stream_snapshot_filters(&context)),
@"
[CLOCK] · running
[CLOCK] Engine: petri run started
[CLOCK] ▶ Start
"
);
}
}
}
#[test]
#[ignore = "pre-existing flake: events --follow hangs against detached dry-run on this branch and on origin/main; tracked separately"]
fn events_follow_detached_run_streams_until_completion() {

24
lib/apps/fabro-cli/tests/it/scenario/petri.rs
View file

@ -42,7 +42,9 @@ use fabro_types::RunId;
use fabro_vault::{SecretType, Vault};
use crate::cmd::support::created_run_id;
use crate::support::{TEST_DEV_TOKEN, TEST_SESSION_SECRET, seed_dev_token_auth};
use crate::support::{
TEST_DEV_TOKEN, TEST_SESSION_SECRET, pretty_stream_snapshot_filters, seed_dev_token_auth,
};
pub(super) const RUN_TIMEOUT: Duration = Duration::from_mins(1);
pub(super) const POLL: Duration = Duration::from_millis(50);
@ -1198,22 +1200,6 @@ fn stream_line_name(line: &serde_json::Value) -> String {
.to_string()
}
/// The snapshot filters for `events --pretty` over a Petri run: clocks,
/// durations and the run id vary per run.
fn pretty_filters(context: &fabro_test::TestContext) -> Vec<(String, String)> {
let mut filters = context.filters();
filters.push((r"\b\d{2}:\d{2}:\d{2}\b".to_string(), "[CLOCK]".to_string()));
filters.push((
r"\b\d+(\.\d+)?(ms|s)\b".to_string(),
"[DURATION]".to_string(),
));
filters.push((
r"Checkpoint [0-9a-f]{7}".to_string(),
"Checkpoint [SHA]".to_string(),
));
filters
}
/// A finished Petri run reads back through the CLI: `events` prints the
/// stream envelope raw, dense in `stream_seq`; `events --pretty` renders
/// the stages by `<subject>.<verb>` with their labels and the platform
@ -1273,12 +1259,12 @@ async fn a_finished_petri_run_reads_back_through_the_cli() {
// Tail: the last two items only.
let tail = cli(&context, &server, &["events", "--tail", "2", &run_id]);
assert_eq!(ndjson(&tail).len(), 2);
assert_eq!(ndjson(&tail), lines[lines.len() - 2..]);
// Pretty: stages and platform records.
let mut cmd = context.command();
cmd.args(["events", "--pretty", "--server", &target, &run_id]);
fabro_snapshot!(pretty_filters(&context), cmd, @"
fabro_snapshot!(pretty_stream_snapshot_filters(&context), cmd, @"
success: true
exit_code: 0
----- stdout -----

24
lib/apps/fabro-cli/tests/it/support/mod.rs
View file

@ -13,6 +13,30 @@ use fabro_test::{EnvVars, TestContext, preserve_coverage_env};
use fabro_types::{RunGraph, RunId, RunSpec, RunStreamItem, WorkflowSettings};
pub(crate) use mcp_client::McpStdioTestClient;
/// Snapshot a real run's pretty stream without depending on wall-clock values
/// or the relative delivery of the platform and Petri startup records.
pub(crate) fn pretty_stream_snapshot_filters(context: &TestContext) -> Vec<(String, String)> {
let mut filters = context.filters();
filters.push((r"\b\d{2}:\d{2}:\d{2}\b".to_string(), "[CLOCK]".to_string()));
filters.push((
r"\b\d+(\.\d+)?(ms|s)\b".to_string(),
"[DURATION]".to_string(),
));
filters.push((
r"Checkpoint [0-9a-f]{7}\b".to_string(),
"Checkpoint [SHA]".to_string(),
));
// Only move the independent `running` line ahead of Petri's start. Keep
// every line (including any intervening output) and all other ordering.
// Missing or repeated records must still fail the full transcript snapshot.
filters.push((
r"(?m)^(\[CLOCK\] Engine: petri run started\n)((?:[^\n]*\n)*?)(\[CLOCK\] · running\n)"
.to_string(),
"${3}${1}${2}".to_string(),
));
filters
}
pub(crate) fn run_output_filters(context: &TestContext) -> Vec<(String, String)> {
let mut filters = context.filters();
filters.push((r"\b\d+(\.\d+)?(ms|s)\b".to_string(), "[TIME]".to_string()));