Synchronize finalization regressions with persisted execution evidence

lib/apps/fabro-cli/src/commands/run/attach.rs

@@ -1060,6 +1060,14 @@ mod tests {
                     .json_body(serde_json::to_value(&state).unwrap());
             })
             .await;
+        server
+            .mock_async(|when, then| {
+                when.method("GET")
+                    .path(format!("/api/v1/runs/{run_id}/questions"));
+                then.status(200)
+                    .json_body(serde_json::json!({ "data": [] }));
+            })
+            .await;
         let waiting = server
             .mock_async(|when, then| {
                 when.method("GET")

lib/apps/fabro-cli/src/commands/run/petri_stream.rs

@@ -1307,10 +1307,10 @@ mod tests {
         assert_eq!(exit_code_of(&finished), Some(1));
         let mut state = PrettyState::default();
         let line = format_pretty(&finished, &Styles::new(false), &mut state).unwrap();
-        insta::assert_snapshot!(line, @r###"
-        12:43:08 ✗ FAILED 0s
-        12:43:08    the push was rejected
-        "###);
+        insta::assert_snapshot!(line, @"
+        04:43:08 ✗ FAILED 0ms
+        04:43:08    the push was rejected
+        ");
     }
 
     #[test]

lib/apps/fabro-server/src/petri_runs.rs

@@ -1044,6 +1044,7 @@ mod tests {
         expected: RunStatus,
         message: Option<&str>,
     ) {
+        state.petri_projector.settle(run_id).await;
         let mut values = Vec::new();
         for suffix in ["", "/state"] {
             let response = app
@@ -1067,7 +1068,9 @@ mod tests {
         }
         assert_eq!(
             values[0]["lifecycle"]["status"],
-            serde_json::to_value(expected).unwrap()
+            serde_json::to_value(expected).unwrap(),
+            "public state: {:#?}",
+            values[1]
         );
         assert_eq!(values[1]["status"], serde_json::to_value(expected).unwrap());
         assert_eq!(state.test_managed_run_status(&run_id), Some(expected));

lib/apps/fabro-server/src/server.rs

@@ -3531,7 +3531,22 @@ pub(crate) async fn persist_run_failure(
             )
             .await
             {
-                Ok(_) => fail_managed_run(state, run_id, reason, message),
+                Ok(_) => match run_records::projection(state, run_id).await {
+                    Ok(Some(committed)) if committed.status.is_terminal() => {
+                        let failure = committed
+                            .conclusion
+                            .as_ref()
+                            .and_then(|conclusion| conclusion.failure.as_ref())
+                            .map(|failure| failure.detail.message.clone());
+                        settle_managed_run_at_finish(state, run_id, committed.status, failure);
+                    }
+                    Ok(_) => {
+                        error!(run_id = %run_id, "Stored host failure has no terminal projection");
+                    }
+                    Err(err) => {
+                        error!(run_id = %run_id, error = %err, "Failed to read the committed host failure");
+                    }
+                },
                 Err(err) => {
                     error!(run_id = %run_id, error = %err, "Failed to persist run failure status");
                 }

lib/components/fabro-petri/tests/hooks.rs

@@ -44,8 +44,9 @@ use fabro_types::{GitIdentitySource, RunId, SandboxProviderKind};
 use object_store::local::LocalFileSystem;
 use petri_execution::inspect::{self, RunInspection};
 use petri_store::{Access, MemoryRunStore, RunKey, RunStore as _};
-use tokio::fs;
 use tokio::process::Command;
+use tokio::sync::{Notify, Semaphore};
+use tokio::{fs, time};
 use tokio_util::sync::CancellationToken;
 
 mod support;
@@ -1609,8 +1610,8 @@ async fn a_run_diff_failure_cannot_silently_skip_publication() {
 
 struct GatedPublisher {
     inner:   Arc<RecordingPublisher>,
-    entered: tokio::sync::Notify,
-    release: tokio::sync::Semaphore,
+    entered: Notify,
+    release: Semaphore,
 }
 
 #[async_trait::async_trait]
@@ -1621,7 +1622,11 @@ impl RunPublisher for GatedPublisher {
 
     async fn publish(&self, publication: &Publication) -> Result<(), String> {
         self.entered.notify_one();
-        self.release.acquire().await.unwrap().forget();
+        self.release
+            .acquire()
+            .await
+            .expect("publication gate stays open")
+            .forget();
         self.inner.publish(publication).await
     }
 }
@@ -1631,8 +1636,8 @@ async fn required_publication_blocks_the_terminal_result_and_cleanup() {
     for rejection in [None, Some("the push was rejected")] {
         let publisher = Arc::new(GatedPublisher {
             inner:   RecordingPublisher::new(rejection),
-            entered: tokio::sync::Notify::new(),
-            release: tokio::sync::Semaphore::new(0),
+            entered: Notify::new(),
+            release: Semaphore::new(0),
         });
         let mut harness = Harness::new().await;
         harness.publisher = Some(publisher.clone());
@@ -1649,7 +1654,7 @@ async fn required_publication_blocks_the_terminal_result_and_cleanup() {
                 )
                 .await
         });
-        tokio::time::timeout(
+        time::timeout(
             std::time::Duration::from_secs(15),
             publisher.entered.notified(),
         )
@@ -1674,10 +1679,6 @@ async fn required_publication_blocks_the_terminal_result_and_cleanup() {
                 .all(|record| record.record["body"]["event"] != "scope.released"),
             "scope cleanup waits for publication"
         );
-        assert!(
-            pending.invocations[0].result.is_some(),
-            "execution already ended"
-        );
         assert!(
             harness.workspace_path(&harness.workspace().await).exists(),
             "workspace is available to publication"

lib/components/fabro-petri/tests/projection.rs

@@ -27,7 +27,8 @@ use fabro_petri::providers::SandboxProviderConfig;
 use fabro_petri::runtime::RuntimeSpec;
 use fabro_petri::{SqliteRunStore, providers, test_support as petri_support};
 use fabro_store::platform_records::{
-    PlatformRecord, PlatformRecordStore, RunCreatedRecord, RunLifecycleKind, RunLifecycleRecord,
+    CheckpointRecord, PlatformRecord, PlatformRecordStore, RunCreatedRecord, RunDiffRecord,
+    RunLifecycleKind, RunLifecycleRecord,
 };
 use fabro_store::{BlobStore, test_support};
 use fabro_types::{
@@ -42,7 +43,7 @@ use petri_runtime::frontend::CompileInputs;
 use petri_runtime::ir::RunStatus as PetriRunStatus;
 use petri_store::{RunKey, RunStore};
 use tokio::fs;
-use tokio::time::sleep;
+use tokio::time::{self, sleep};
 
 const COMMAND_WORKFLOW: &str = r#"digraph Command {
     graph [goal="Run one command"]
@@ -1387,14 +1388,30 @@ async fn required_finalization_projects_only_the_committed_overall_result() {
             .await
             .unwrap()
         });
-        tokio::time::timeout(Duration::from_secs(15), finalizer.entered.notified())
+        time::timeout(Duration::from_secs(15), finalizer.entered.notified())
             .await
             .unwrap();
-        projector.settle(scenario.run_id).await;
-        let pending = petri_support::stored_projection(&scenario.pool, scenario.run_id)
-            .await
-            .unwrap()
-            .unwrap();
+        // The driver can still flush its execution journal after entering
+        // finalization. Wait for the exit-stage evidence, rather than racing
+        // that writer while comparing the live view with a full rebuild.
+        let pending = time::timeout(Duration::from_secs(5), async {
+            loop {
+                projector.signal(scenario.run_id);
+                projector.settle(scenario.run_id).await;
+                let pending = petri_support::stored_projection(&scenario.pool, scenario.run_id)
+                    .await
+                    .unwrap()
+                    .unwrap();
+                if pending.iter_stages().any(|(id, stage)| {
+                    id.node_id() == "exit" && stage.state == StageState::Succeeded
+                }) {
+                    break pending;
+                }
+                time::sleep(Duration::from_millis(5)).await;
+            }
+        })
+        .await
+        .expect("execution evidence flushes while publication is held");
         assert_eq!(pending.status, RunStatus::Running);
         assert!(pending.conclusion.is_none());
         assert!(!task.is_finished());
@@ -1408,7 +1425,7 @@ async fn required_finalization_projects_only_the_committed_overall_result() {
         let patch_blob = BlobHash::new(b"final patch");
         let platform = PlatformRecordStore::new(scenario.pool.clone());
         for record in [
-            PlatformRecord::Checkpoint(fabro_store::platform_records::CheckpointRecord {
+            PlatformRecord::Checkpoint(CheckpointRecord {
                 execution:      0,
                 firing:         0,
                 attempt:        Some(1),
@@ -1418,7 +1435,7 @@ async fn required_finalization_projects_only_the_committed_overall_result() {
                 patch_blob:     Some(patch_blob),
                 operation:      None,
             }),
-            PlatformRecord::RunDiff(fabro_store::platform_records::RunDiffRecord {
+            PlatformRecord::RunDiff(RunDiffRecord {
                 base_sha:     None,
                 head_sha:     Some(head_sha.to_string()),
                 diff_summary: Some(summary),