From 9a8b8fb0da1336d4b79b1e209fceeb690fb4d03f Mon Sep 17 00:00:00 2001 From: Scott Werner Date: Thu, 29 Oct 2026 08:18:20 -0400 Subject: [PATCH 1/2] test: stabilize attach JSON across event log interleavings --- lib/apps/fabro-cli/tests/it/cmd/attach.rs | 280 +++++++++++++--------- 1 file changed, 167 insertions(+), 113 deletions(-) diff --git a/lib/apps/fabro-cli/tests/it/cmd/attach.rs b/lib/apps/fabro-cli/tests/it/cmd/attach.rs index 1f6125c4f..06b8e73b2 100644 --- a/lib/apps/fabro-cli/tests/it/cmd/attach.rs +++ b/lib/apps/fabro-cli/tests/it/cmd/attach.rs @@ -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 = 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]", From 9081d0fcc54f1a6d96e7051df1fc05bf9efa6e22 Mon Sep 17 00:00:00 2001 From: Scott Werner Date: Fri, 30 Oct 2026 10:53:24 -0400 Subject: [PATCH 2/2] test: tolerate valid startup interleavings in event snapshots --- lib/apps/fabro-cli/tests/it/cmd/events.rs | 215 ++++++++++++------ lib/apps/fabro-cli/tests/it/scenario/petri.rs | 24 +- lib/apps/fabro-cli/tests/it/support/mod.rs | 24 ++ 3 files changed, 177 insertions(+), 86 deletions(-) diff --git a/lib/apps/fabro-cli/tests/it/cmd/events.rs b/lib/apps/fabro-cli/tests/it/cmd/events.rs index ada175646..f67205912 100644 --- a/lib/apps/fabro-cli/tests/it/cmd/events.rs +++ b/lib/apps/fabro-cli/tests/it/cmd/events.rs @@ -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 { 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 = 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 = 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 = 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() { diff --git a/lib/apps/fabro-cli/tests/it/scenario/petri.rs b/lib/apps/fabro-cli/tests/it/scenario/petri.rs index 8e5fc4337..27c93b8cb 100644 --- a/lib/apps/fabro-cli/tests/it/scenario/petri.rs +++ b/lib/apps/fabro-cli/tests/it/scenario/petri.rs @@ -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 `.` 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 ----- diff --git a/lib/apps/fabro-cli/tests/it/support/mod.rs b/lib/apps/fabro-cli/tests/it/support/mod.rs index 25364a67c..1545bd762 100644 --- a/lib/apps/fabro-cli/tests/it/support/mod.rs +++ b/lib/apps/fabro-cli/tests/it/support/mod.rs @@ -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()));