merge: retain lifecycle receipts through CLI and TUI shutdown

crates/hooks/src/lifecycle_outbox.rs

@@ -36,11 +36,13 @@
 //! If no tokio runtime is available the event is dropped with a warning.
 //! Webhook POSTs (`{"at": …, "event": …}`) are attempted after the local
 //! append; failures are logged and dropped, never retried into the agent
-//! loop.
+//! loop. Process owners call [`LifecycleOutbox::flush`] before runtime teardown
+//! to give queued terminal events a bounded opportunity to finish delivery.
 
 use std::path::{Path, PathBuf};
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex};
+use std::time::Duration;
 
 use anyhow::{Context, Result};
 use chrono::Utc;
@@ -48,6 +50,7 @@ use codewhale_protocol::runtime::{RUNTIME_EVENT_ENVELOPE_SCHEMA_VERSION, Runtime
 use serde_json::{Value, json};
 use tokio::io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt};
 use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
+use tokio::sync::oneshot;
 
 use crate::WebhookHookSink;
 
@@ -157,18 +160,41 @@ impl LifecycleOutbox {
         let Some(inner) = self.inner.clone() else {
             return;
         };
-        if let Err(error) = inner.enqueue(event) {
+        if let Err(error) = inner.enqueue(OutboxCommand::Event(event)) {
             tracing::warn!(target: "lifecycle_outbox", %error, "lifecycle event dropped");
         }
     }
+
+    /// Wait for events queued before this call to finish their delivery attempt.
+    ///
+    /// Local writes are flushed before the writer acknowledges this barrier.
+    /// Delivery errors remain logged and best effort; a stalled writer/webhook
+    /// or a stopped runtime returns an error within `timeout` instead of holding
+    /// process teardown indefinitely. This does not close other cloned handles.
+    pub async fn flush(&self, timeout: Duration) -> Result<()> {
+        let Some(inner) = self.inner.as_ref() else {
+            return Ok(());
+        };
+        let (reply, completed) = oneshot::channel();
+        inner.enqueue(OutboxCommand::Flush(reply))?;
+        tokio::time::timeout(timeout, completed)
+            .await
+            .context("lifecycle outbox flush timed out; queued events may be incomplete")?
+            .context("lifecycle outbox writer stopped before flush completed")
+    }
+}
+
+enum OutboxCommand {
+    Event(LifecycleEvent),
+    Flush(oneshot::Sender<()>),
 }
 
 struct OutboxInner {
     path: PathBuf,
     webhook: Option<WebhookHookSink>,
-    sender: UnboundedSender<LifecycleEvent>,
+    sender: UnboundedSender<OutboxCommand>,
     /// The writer task's receive half. Taken exactly once by the writer task.
-    receiver: Mutex<Option<UnboundedReceiver<LifecycleEvent>>>,
+    receiver: Mutex<Option<UnboundedReceiver<OutboxCommand>>>,
     writer_spawned: AtomicBool,
     /// Serializes the lazy writer-task spawn so two racing first emits cannot
     /// start two writers.
@@ -180,9 +206,9 @@ impl OutboxInner {
     ///
     /// Ordering: `send` happens before the spawn so events queued before the
     /// writer starts are drained first, preserving enqueue order.
-    fn enqueue(self: &Arc<Self>, event: LifecycleEvent) -> Result<()> {
+    fn enqueue(self: &Arc<Self>, command: OutboxCommand) -> Result<()> {
         self.sender
-            .send(event)
+            .send(command)
             .map_err(|_| anyhow::anyhow!("lifecycle outbox writer task is gone"))?;
         self.ensure_writer_spawned();
         Ok(())
@@ -235,13 +261,20 @@ struct WriterState {
     /// Next seq to assign; filled in by [`Self::recover_seq`] on first use.
     next_seq: u64,
     recovered: bool,
-    receiver: UnboundedReceiver<LifecycleEvent>,
+    receiver: UnboundedReceiver<OutboxCommand>,
 }
 
 impl WriterState {
     /// Drain the queue until every sender is dropped, then exit.
     async fn run(&mut self) {
-        while let Some(event) = self.receiver.recv().await {
+        while let Some(command) = self.receiver.recv().await {
+            let event = match command {
+                OutboxCommand::Event(event) => event,
+                OutboxCommand::Flush(reply) => {
+                    let _ = reply.send(());
+                    continue;
+                }
+            };
             if let Err(error) = self.deliver(event).await {
                 tracing::warn!(
                     target: "lifecycle_outbox",
@@ -453,6 +486,94 @@ mod tests {
             .collect()
     }
 
+    #[test]
+    fn flush_persists_queued_turn_boundaries_before_runtime_teardown() {
+        let (_dir, path) = temp_outbox_path("shutdown.jsonl");
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("runtime");
+        runtime.block_on(async {
+            let outbox = LifecycleOutbox::new(Some(path.clone()), None, None);
+            outbox.emit(event("turn_start", "turn.started"));
+            let clone = outbox.clone();
+            clone.emit(event("turn_end", "turn.completed"));
+            outbox.flush(Duration::from_secs(2)).await.expect("flush");
+        });
+        drop(runtime);
+
+        let contents = std::fs::read_to_string(path).expect("outbox persisted before exit");
+        let lines: Vec<Value> = contents
+            .lines()
+            .map(|line| serde_json::from_str(line).expect("complete JSONL record"))
+            .collect();
+        assert_eq!(lines.len(), 2);
+        assert_eq!(lines[0]["event"], "turn_start");
+        assert_eq!(lines[0]["seq"], 1);
+        assert_eq!(lines[1]["event"], "turn_end");
+        assert_eq!(lines[1]["seq"], 2);
+    }
+
+    #[tokio::test]
+    async fn flush_disabled_or_unused_outbox_creates_no_file() {
+        let (_dir, path) = temp_outbox_path("unused.jsonl");
+        LifecycleOutbox::disabled()
+            .flush(Duration::ZERO)
+            .await
+            .expect("disabled flush");
+        LifecycleOutbox::new(Some(path.clone()), None, None)
+            .flush(Duration::from_secs(2))
+            .await
+            .expect("unused flush");
+        assert!(!path.exists());
+    }
+
+    #[tokio::test]
+    async fn flush_is_bounded_when_webhook_delivery_stalls() {
+        let (_dir, path) = temp_outbox_path("stalled.jsonl");
+        let server = wiremock::MockServer::start().await;
+        wiremock::Mock::given(wiremock::matchers::method("POST"))
+            .respond_with(wiremock::ResponseTemplate::new(200).set_delay(Duration::from_secs(5)))
+            .mount(&server)
+            .await;
+        let outbox = LifecycleOutbox::new(Some(path), Some(server.uri()), None);
+        outbox.emit(event("turn_start", "turn.started"));
+        let error = tokio::time::timeout(
+            Duration::from_secs(1),
+            outbox.flush(Duration::from_millis(20)),
+        )
+        .await
+        .expect("the flush deadline must bound a blocked writer")
+        .expect_err("stalled delivery must report incomplete drain");
+        assert!(error.to_string().contains("flush timed out"));
+    }
+
+    #[test]
+    fn flush_reports_a_writer_lost_with_its_runtime() {
+        let (_dir, path) = temp_outbox_path("stopped.jsonl");
+        let outbox = LifecycleOutbox::new(Some(path), None, None);
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("first runtime");
+        runtime.block_on(async {
+            outbox.emit(event("turn_start", "turn.started"));
+            outbox
+                .flush(Duration::from_secs(2))
+                .await
+                .expect("first flush");
+        });
+        drop(runtime);
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("second runtime");
+        let error = runtime
+            .block_on(outbox.flush(Duration::from_secs(2)))
+            .expect_err("a closed writer cannot acknowledge delivery");
+        assert!(error.to_string().contains("writer task is gone"));
+    }
+
     #[tokio::test]
     async fn appends_one_jsonl_line_per_event_with_envelope_schema() {
         let (_dir, path) = temp_outbox_path("schema.jsonl");

crates/tui/src/exec_agent.rs

@@ -1077,6 +1077,13 @@ pub(crate) async fn run_exec_agent(
         }
     }
 
+    // Drain the terminal receipt before either returning or taking the explicit
+    // retryable-failure process exit below. Outbox failures cannot change the
+    // authoritative turn outcome.
+    if let Err(error) = lifecycle_outbox.flush(Duration::from_secs(2)).await {
+        tracing::warn!(target: "lifecycle_outbox", %error, "exec lifecycle outbox did not drain before exit");
+    }
+
     if json_output {
         println!("{}", serde_json::to_string_pretty(&summary)?);
     }

crates/tui/src/tui/ui/event_loop.rs

@@ -883,6 +883,12 @@ pub async fn run_tui(
         });
     }
 
+    // Keep the final session/turn receipts ahead of runtime teardown. A failed
+    // observability sink must not prevent the user's session from shutting down.
+    if let Err(error) = app.lifecycle_outbox.flush(Duration::from_secs(2)).await {
+        tracing::warn!(target: "lifecycle_outbox", %error, "TUI lifecycle outbox did not drain before exit");
+    }
+
     // Flush the persistence actor, collect the durability report (write
     // failures are surfaced, not discarded), then shut down gracefully.
     //

crates/tui/tests/cucumber/plugin_e2e_acceptance.rs

@@ -844,6 +844,38 @@ fn wait_for_log(tui: &mut Harness, path: &std::path::Path, needle: &str) {
     }
 }
 
+/// A focused composer and a streamed answer can both appear before the turn
+/// settles. Use the runtime's terminal receipt, not either paint, to admit the
+/// next prompt; otherwise this fixture exercises the busy-turn queue by accident.
+#[cfg(all(unix, feature = "long-running-tests"))]
+fn wait_for_turn_receipt(tui: &mut Harness, outbox: &std::path::Path, count: usize, kind: &str) {
+    let receipt = || {
+        std::fs::read_to_string(outbox)
+            .unwrap_or_default()
+            .lines()
+            .filter_map(|line| serde_json::from_str::<serde_json::Value>(line).ok())
+            .filter(|event| event["event"] == "turn_end")
+            .collect::<Vec<_>>()
+    };
+    if tui
+        .wait_for(|_| receipt().len() >= count, BINARY_ACCEPTANCE_TIMEOUT)
+        .is_err()
+    {
+        panic!(
+            "terminal turn receipt {count} not observed within {:?}\n{}",
+            qa_harness::harness::ci_scaled(BINARY_ACCEPTANCE_TIMEOUT),
+            short_diagnostics(tui, Some(outbox)),
+        );
+    }
+    let events = receipt();
+    assert_eq!(
+        events.len(),
+        count,
+        "one terminal receipt per submitted turn"
+    );
+    assert_eq!(events[count - 1]["kind"], kind, "terminal turn outcome");
+}
+
 /// Exercise the distributed binary through a real PTY and a sealed home. The
 /// only socket is a test-owned loopback model endpoint; plugin execution is
 /// stdio-only and receives no real credentials or ambient secret environment.
@@ -857,6 +889,14 @@ async fn plugin_toml_binary_lifecycle_skill_and_stdio_mcp_acceptance() {
     let workspace = make_sealed_workspace().expect("sealed workspace");
     let bundle = write_reviewed_bundle_fixture(workspace.workspace());
     let mcp_log = workspace.home().join(".codewhale/plugin-acceptance.log");
+    let outbox = workspace.home().join(".codewhale/lifecycle-outbox.jsonl");
+    let config_path = workspace.home().join(".codewhale/config.toml");
+    let mut config = std::fs::read_to_string(&config_path).expect("sealed config");
+    config.push_str(&format!(
+        "\n[lifecycle_outbox]\npath = {}\n",
+        serde_json::json!(outbox.to_string_lossy()),
+    ));
+    std::fs::write(config_path, config).expect("configure sealed lifecycle receipt");
     let (base_url, shutdown_tx, model_thread) = spawn_hermetic_model_server();
     let mut tui = Harness::builder(Harness::cargo_bin("codewhale-tui"))
         .cwd(workspace.workspace())
@@ -884,6 +924,7 @@ async fn plugin_toml_binary_lifecycle_skill_and_stdio_mcp_acceptance() {
     // The binary begins at Tideline Startup, so choose its real New Session
     // action before exercising the existing-session plugin contract.
     begin_new_session_from_startup(&mut tui);
+    wait_for_turn_receipt(&mut tui, &outbox, 1, "turn.completed");
     wait_for_composer_ready(&mut tui);
     submit_tui_command(&mut tui, "/plugin show demo");
     expect_visible(
@@ -944,12 +985,12 @@ async fn plugin_toml_binary_lifecycle_skill_and_stdio_mcp_acceptance() {
         "binary plugin call complete",
         "plugin tool result returned to model",
     );
+    wait_for_turn_receipt(&mut tui, &outbox, 2, "turn.completed");
 
     submit_tui_command(&mut tui, "hang plugin call");
     wait_for_log(&mut tui, &mcp_log, "call:hang");
     tui.send([0x03]).expect("interrupt hanging plugin turn");
-    std::thread::sleep(std::time::Duration::from_millis(300));
-    tui.pump();
+    wait_for_turn_receipt(&mut tui, &outbox, 3, "turn.interrupted");
     tui.send([0x15])
         .expect("clear the interrupted prompt restored into the composer");
     submit_tui_command(&mut tui, "/plugin revoke demo");
@@ -966,6 +1007,19 @@ async fn plugin_toml_binary_lifecycle_skill_and_stdio_mcp_acceptance() {
     assert!(state.contains("\"trust\": null"));
     assert!(bundle.join("server.py").exists(), "source bundle preserved");
 
+    submit_tui_command(&mut tui, "/exit");
+    assert_eq!(
+        tui.wait_for_exit(BINARY_ACCEPTANCE_TIMEOUT),
+        Some(0),
+        "the TUI must exit gracefully after revoking the plugin",
+    );
+    let receipts = std::fs::read_to_string(&outbox).expect("outbox after process exit");
+    let final_event: serde_json::Value =
+        serde_json::from_str(receipts.lines().last().expect("final receipt"))
+            .expect("complete final JSONL event");
+    assert_eq!(final_event["event"], "session_end");
+    assert_eq!(final_event["kind"], "session.ended");
+
     let _ = tui.shutdown();
     let _ = shutdown_tx.send(());
     let _ = model_thread.join();

crates/tui/tests/integration/lifecycle_outbox_exec.rs

@@ -112,7 +112,11 @@ fn preserve_host_env(command: &mut Command) {
 /// Any `__OUTBOX_PATH__` token in it is replaced with the isolated home's
 /// absolute outbox path. Returns the isolated home and workspace dirs (the
 /// latter so callers can assert the outbox `payload.workspace` exactly).
-fn run_exec_with_outbox_config(server: &MockServer, outbox_toml: &str) -> (TempDir, TempDir) {
+fn run_exec_with_outbox_config(
+    server: &MockServer,
+    outbox_toml: &str,
+    expected_exit_code: i32,
+) -> (TempDir, TempDir) {
     let workspace = TempDir::new().expect("workspace tempdir");
     let home = TempDir::new().expect("home tempdir");
     let outbox_path = home_outbox_path(&home);
@@ -184,9 +188,10 @@ fn run_exec_with_outbox_config(server: &MockServer, outbox_toml: &str) -> (TempD
 
     let stdout = join_pipe_reader(stdout_reader, "stdout");
     let stderr = join_pipe_reader(stderr_reader, "stderr");
-    assert!(
-        status.success(),
-        "codewhale-tui exec failed\nstdout:\n{}\nstderr:\n{}",
+    assert_eq!(
+        status.code(),
+        Some(expected_exit_code),
+        "codewhale-tui exec returned the wrong exit status\nstdout:\n{}\nstderr:\n{}",
         String::from_utf8_lossy(&stdout),
         String::from_utf8_lossy(&stderr)
     );
@@ -254,6 +259,7 @@ async fn exec_emits_turn_start_and_turn_end_to_the_configured_outbox() {
     let (home, workspace) = run_exec_with_outbox_config(
         &server,
         &format!("[lifecycle_outbox]\npath = {}\n", json!(OUTBOX_PATH_TOKEN)),
+        0,
     );
 
     let outbox_path = home_outbox_path(&home);
@@ -301,7 +307,7 @@ async fn exec_emits_turn_start_and_turn_end_to_the_configured_outbox() {
 #[tokio::test(flavor = "multi_thread")]
 async fn exec_without_outbox_config_writes_no_file() {
     let server = start_mock_llm().await;
-    let (home, _workspace) = run_exec_with_outbox_config(&server, "");
+    let (home, _workspace) = run_exec_with_outbox_config(&server, "", 0);
 
     assert!(
         !home_outbox_path(&home).exists(),
@@ -317,6 +323,7 @@ async fn outbox_seq_recovers_across_processes() {
     let (home, _workspace) = run_exec_with_outbox_config(
         &server,
         &format!("[lifecycle_outbox]\npath = {}\n", json!(OUTBOX_PATH_TOKEN)),
+        0,
     );
     let shared_outbox = home_outbox_path(&home);
 
@@ -327,6 +334,7 @@ async fn outbox_seq_recovers_across_processes() {
             "[lifecycle_outbox]\npath = {}\n",
             json!(shared_outbox.display().to_string())
         ),
+        0,
     );
 
     let lines = read_outbox_lines(&shared_outbox);
@@ -341,3 +349,28 @@ async fn outbox_seq_recovers_across_processes() {
         "seq must be monotonic across processes"
     );
 }
+
+#[tokio::test(flavor = "multi_thread")]
+async fn failed_exec_persists_the_terminal_receipt_without_changing_its_exit() {
+    let server = start_mock_llm().await;
+    Mock::given(method("POST"))
+        .and(path("/v1/chat/completions"))
+        .respond_with(ResponseTemplate::new(500).set_body_string("upstream model failure"))
+        .with_priority(1)
+        .mount(&server)
+        .await;
+    let (home, _workspace) = run_exec_with_outbox_config(
+        &server,
+        &format!("[lifecycle_outbox]\npath = {}\n", json!(OUTBOX_PATH_TOKEN)),
+        1,
+    );
+    let lines = read_outbox_lines(&home_outbox_path(&home));
+    assert_eq!(lines.len(), 2, "failed turn retains both boundaries");
+    assert_eq!(lines[0]["event"], "turn_start");
+    assert_eq!(lines[0]["seq"], 1);
+    assert_eq!(lines[1]["event"], "turn_end");
+    assert_eq!(lines[1]["kind"], "turn.failed");
+    assert_eq!(lines[1]["seq"], 2);
+    assert_eq!(lines[1]["payload"]["status"], "failed");
+    assert!(!lines[1]["payload"]["error"].as_str().unwrap().is_empty());
+}