Make coding-agent hook capture journal-only

bt-daemon/README.md

@@ -141,12 +141,15 @@ different profiles, organizations, projects, experiments, or parent spans.
 
 Phases 0–5 are implemented: protocol, daemon lifecycle, Braintrust sink,
 Codex and Claude translators, `bt daemon` integration, and thin hook shims for
-both shipped plugins. Restart recovery replays the redacted journal with
-deterministic span ids, so resubmitted rows merge into the same spans instead
-of creating duplicates. Claude lifecycle entries reference a daemon-owned
-transcript mirror, so recovery does not depend on mutable external paths
-without re-recording the transcript on every turn. Explicit turn/session-end
-flushes are bounded, and sessions can target project logs or an experiment.
+both shipped plugins. Every coding-agent capture request returns after the raw
+event is flushed to its journal; authentication, correlation, translation, and
+reporting run on daemon-owned workers. Restart recovery replays the redacted
+journal with deterministic span ids, so resubmitted rows merge into the same
+spans instead of creating duplicates. Claude and Codex lifecycle entries
+reference a daemon-owned transcript mirror, so recovery does not depend on
+mutable external paths without re-recording the transcript on every turn.
+Explicit turn/session-end flushes are bounded, and sessions can target project
+logs or an experiment.
 
 Memory is bounded end to end, while on-disk records stay complete: the daemon
 never holds a transcript or a whole journal in memory, mirroring and replay

bt-daemon/docs/protocol.md

@@ -64,15 +64,13 @@ not found, `-32602` invalid params, `-32603` internal); application errors use
 
 ### Ordering & delivery
 
-A subprocess-style shim opens a connection, does one `event.log`, and exits.
-Per-session ordering is guaranteed because (a) the agent runs hooks in blocking
-mode, so it does not fire the next hook until the current one returns, and (b)
-`event.log` is a **request** whose success response means *the event has been
-appended to that session's ordered queue* (not that it has been delivered to
-Braintrust). The shim must await that response before exiting. Long-lived
-in-process clients (opencode/pi, later) hold one connection and may send
-`event.log` as a **notification** for the hot path, relying on the single
-connection for ordering.
+A capture adapter sends `event.log` as a **request** and waits only for the raw
+event and any transcript high-water reference to be flushed to the daemon's
+journal. The daemon then queues correlation, translation, and reporting on its
+own workers. Per-session journal appends are serialized, and the daemon keeps
+derived processing ordered without making the hook wait for it. Subprocess
+shims may exit as soon as they receive the response; long-lived in-process
+clients use the same request boundary.
 
 ## Methods
 
@@ -105,18 +103,14 @@ The hot path. Params are the **Envelope** (see below). Request result:
 ```json
 { "accepted": true }
 ```
-`accepted: true` means durably recorded: normally journaled and enqueued to the
-session's ordered queue, or held in the daemon's private correlation journal
-while multiple parent calls remain indistinguishable.
-For an event that opens a tool call, it also means the daemon has made that
-active-tool marker visible to local child-session correlation. A child hook
-that runs immediately after its parent's blocking pre-tool hook can therefore
-attach without an intervening flush.
-The daemon never fails the caller's turn for a downstream (Braintrust) error;
-those are handled asynchronously and surfaced via `status.get`. The queue is
-bounded, so a session whose sink has stalled applies backpressure here instead
-of accumulating events without limit; the event is already journaled by then,
-so this costs latency, never data.
+`accepted: true` means the raw capture is durably recorded in the source
+journal. It does not mean authentication, correlation, translation, session
+queueing, or Braintrust delivery has completed. Those steps run out-of-band;
+errors are surfaced via `status.get`. A child that arrives before its parent's
+tool marker is translated is held in durable pending-correlation state and
+reconciled by the daemon. On restart, uncheckpointed journal entries are queued
+again automatically. Explicit status and flush requests act as daemon-worker
+barriers, but hook capture never does.
 
 ### `session.flush` (request)
 

bt-daemon/src/dispatch.rs

@@ -3,11 +3,9 @@
 //! are processed strictly in arrival order. Different sessions run
 //! concurrently.
 //!
-//! Ack semantics: `event.log` is acked once the event is journaled and its
-//! first translation batch has updated local correlation state. Tool-start
-//! events drain bounded translator continuations before ack so the spawning
-//! marker is guaranteed visible. A downstream error never fails the caller's
-//! turn.
+//! Hook acknowledgement happens before this layer, immediately after the raw
+//! event is durably journaled. Translation, correlation, and delivery here are
+//! entirely out-of-band from hook execution.
 
 use crate::delivery_ledger::LedgerSink;
 use crate::journal::JournalWriter;
@@ -33,8 +31,9 @@ pub struct Counters {
 }
 
 enum SessionMsg {
-    Event(Box<Envelope>, u64, oneshot::Sender<()>),
+    Event(Box<Envelope>, u64),
     Configure(Box<crate::wire::SessionConfig>, oneshot::Sender<()>),
+    Barrier(oneshot::Sender<()>),
     Flush(oneshot::Sender<u64>),
     Finalize(oneshot::Sender<u64>),
     Shutdown(oneshot::Sender<()>),
@@ -65,6 +64,7 @@ pub(crate) struct SessionOptions {
     pub correlation: Arc<crate::correlation::CorrelationRegistry>,
     pub data_dir: PathBuf,
     pub journal: Arc<tokio::sync::Mutex<JournalWriter>>,
+    pub correlation_changed: Arc<tokio::sync::Notify>,
 }
 
 /// Handle to one live session: its queue plus observable counters/state.
@@ -96,6 +96,7 @@ impl Session {
             correlation,
             data_dir,
             journal,
+            correlation_changed,
         } = options;
         let (tx, rx) = mpsc::channel(QUEUE_CAPACITY);
         let counters = Arc::new(Counters::default());
@@ -119,6 +120,7 @@ impl Session {
             correlation,
             data_dir,
             journal,
+            correlation_changed,
         };
         tokio::spawn(actor.run(rx));
 
@@ -132,24 +134,29 @@ impl Session {
         })
     }
 
-    /// Enqueue an event after the daemon has journaled it.
+    /// Queue a journaled event without waiting for translation or delivery.
     pub async fn enqueue(&self, env: Envelope, journal_through: u64) -> anyhow::Result<()> {
         self.touch();
         self.counters.queued.fetch_add(1, Ordering::Relaxed);
-        let (reply_tx, reply_rx) = oneshot::channel();
         self.tx
-            .send(SessionMsg::Event(Box::new(env), journal_through, reply_tx))
+            .send(SessionMsg::Event(Box::new(env), journal_through))
             .await
-            .map_err(|_| anyhow::anyhow!("session actor is gone"))?;
-        reply_rx
-            .await
-            .map_err(|_| anyhow::anyhow!("session actor dropped event acknowledgement"))
+            .map_err(|_| anyhow::anyhow!("session actor is gone"))
     }
 
     fn touch(&self) {
         *self.last_activity.lock().unwrap() = Instant::now();
     }
 
+    /// Wait until events already accepted by this daemon worker have updated
+    /// translator and correlation state. Hook capture never calls this.
+    pub async fn barrier(&self) {
+        let (reply_tx, reply_rx) = oneshot::channel();
+        if self.tx.send(SessionMsg::Barrier(reply_tx)).await.is_ok() {
+            let _ = reply_rx.await;
+        }
+    }
+
     /// How long since this session last saw traffic. Drives idle retirement.
     pub fn idle_for(&self) -> std::time::Duration {
         self.last_activity.lock().unwrap().elapsed()
@@ -206,19 +213,22 @@ impl Session {
     }
 }
 
-/// Claude transcript files are external mutable state. Mirror them into
+/// Agent transcript files are external mutable state. Mirror them into
 /// daemon-owned storage at lifecycle boundaries and journal only a reference,
 /// so recovery/replay does not depend on a path that Claude may later rewrite
 /// or delete — and so the transcript is stored once rather than re-copied into
 /// every event. Fail open: without a reference the translator reads the live
 /// path exactly as before.
 pub(crate) async fn hydrate_transcript_reference(data_dir: &std::path::Path, env: &mut Envelope) {
-    if env.source != "claude-code"
-        || !matches!(
+    let should_capture = match env.source.as_str() {
+        "codex" => true,
+        "claude-code" => matches!(
             env.event.as_str(),
             "UserPromptSubmit" | "Stop" | "StopFailure" | "SubagentStop" | "SessionEnd"
-        )
-    {
+        ),
+        _ => false,
+    };
+    if !should_capture {
         return;
     }
     let field = if env.event == "SubagentStop" {
@@ -272,6 +282,7 @@ struct SessionActor {
     correlation: Arc<crate::correlation::CorrelationRegistry>,
     data_dir: PathBuf,
     journal: Arc<tokio::sync::Mutex<JournalWriter>>,
+    correlation_changed: Arc<tokio::sync::Notify>,
 }
 
 #[derive(Clone, Copy)]
@@ -328,13 +339,15 @@ impl SessionActor {
                 // callers waiting on flush don't hang.
                 while let Some(msg) = rx.recv().await {
                     match msg {
-                        SessionMsg::Event(_, _, reply) => {
+                        SessionMsg::Event(_, _) => {
                             self.counters.queued.fetch_sub(1, Ordering::Relaxed);
-                            let _ = reply.send(());
                         }
                         SessionMsg::Configure(_, r) => {
                             let _ = r.send(());
                         }
+                        SessionMsg::Barrier(r) => {
+                            let _ = r.send(());
+                        }
                         SessionMsg::Flush(r) => {
                             let _ = r.send(0);
                         }
@@ -381,18 +394,14 @@ impl SessionActor {
 
         while let Some(msg) = rx.recv().await {
             match msg {
-                SessionMsg::Event(env, journal_through, reply) => {
+                SessionMsg::Event(env, journal_through) => {
                     let correlation_barrier = is_tool_lifecycle_event(&env.event);
-                    let mut reply = Some(reply);
                     if let Some(cfg) = &env.config {
                         sink.configure(cfg);
                         ctx.config = Some(cfg.clone());
                         self.refresh_permalink(sink.as_ref());
                     }
                     let translated = translator.handle(&env, &ctx);
-                    if !correlation_barrier {
-                        let _ = reply.take().expect("event reply").send(());
-                    }
                     let (correlation_changed, delivered) = self
                         .emit_translator_batches(
                             &mut translator,
@@ -412,9 +421,7 @@ impl SessionActor {
                         {
                             self.set_error(error);
                         }
-                    }
-                    if let Some(reply) = reply {
-                        let _ = reply.send(());
+                        self.correlation_changed.notify_one();
                     }
                     self.counters.queued.fetch_sub(1, Ordering::Relaxed);
                     if delivered && checkpointable {
@@ -429,6 +436,9 @@ impl SessionActor {
                     self.refresh_permalink(sink.as_ref());
                     let _ = reply.send(());
                 }
+                SessionMsg::Barrier(reply) => {
+                    let _ = reply.send(());
+                }
                 SessionMsg::Flush(reply) => {
                     checkpointable &= self
                         .checkpoint_and_flush(&mut translator, &mut sink, &ctx)
@@ -510,12 +520,16 @@ impl SessionActor {
         while let Some(ops) = next {
             if !ops.is_empty() {
                 if mode.observes_correlation() {
-                    correlation_changed |= self.correlation.observe_ops(
+                    let changed = self.correlation.observe_ops(
                         &self.correlation_key,
                         &self.route,
                         ctx.config.as_ref().expect("session config"),
                         &ops,
                     );
+                    correlation_changed |= changed;
+                    if changed {
+                        self.persist_correlation_if_changed(true).await;
+                    }
                 }
                 match sink.emit(&ops).await {
                     Ok(n) => {

bt-daemon/src/journal.rs

@@ -227,6 +227,10 @@ impl JournalWriter {
         Ok(Self { file, position })
     }
 
+    pub(crate) fn position(&self) -> u64 {
+        self.position
+    }
+
     /// Append one event in redacted form and flush to the OS. Not fsync'd per
     /// event (that would dominate hook latency); an OS crash can lose the last
     /// few lines, which replay tolerates.

bt-daemon/src/lib.rs

@@ -115,12 +115,14 @@ pub struct HookArgs {
     /// Fail instead of spawning a daemon if none is running.
     #[arg(long)]
     pub no_spawn: bool,
-    /// Flush the session after a turn-ending event. Intended for short-lived
-    /// CI hosts; SessionEnd is always flushed.
+    /// Ask the daemon to flush the session after a turn-ending event. The
+    /// flush is scheduled out-of-band; hook capture still returns immediately
+    /// after the durable journal write.
     #[arg(long)]
     pub flush_on_turn_end: bool,
-    /// Bound an explicit turn/session-end flush.
-    #[arg(long, default_value_t = 10_000)]
+    /// Deprecated compatibility option. Hook capture never waits for daemon
+    /// translation, reporting, or flushing.
+    #[arg(long, default_value_t = 10_000, hide = true)]
     pub flush_timeout_ms: u64,
     /// JSON object merged into root-span metadata. Deliberately not read from
     /// the environment: a hook fires automatically on every event, so its
@@ -293,7 +295,7 @@ pub async fn run_serve(args: ServeArgs, opts: ServeOptions) -> anyhow::Result<()
 /// Capture one hook event from `stdin` and forward it to the daemon.
 ///
 /// `route` contains only non-secret profile and destination selection.
-/// Returns `Ok` once the daemon has acked (journaled + enqueued). Callers that
+/// Returns `Ok` once the daemon has durably journaled the event. Callers that
 /// must never fail the agent's turn should treat any `Err` as non-fatal and
 /// exit 0.
 pub async fn run_hook(
@@ -350,14 +352,6 @@ pub async fn run_hook(
 
     let socket = paths::socket_path(args.socket.as_deref());
     forward_envelope(&env, &socket, &host, args.no_spawn).await?;
-    let should_flush = env.event == "SessionEnd"
-        || (matches!(
-            env.route.as_ref().map(|r| r.flush_mode),
-            Some(wire::FlushMode::FlushOnTurnEnd)
-        ) && matches!(env.event.as_str(), "Stop" | "SubagentStop"));
-    if should_flush {
-        flush_session(&env.session_id, &socket, args.flush_timeout_ms).await?;
-    }
     Ok(())
 }