Improve hook routing handling

bt-daemon/docs/protocol.md

@@ -111,18 +111,22 @@ those are handled asynchronously and surfaced via `status.get`.
 
 ### `session.flush` (request)
 
-Block until the session's spans are delivered, or `timeout_ms` elapses.
+Block until every route's spans for the session are delivered, or `timeout_ms`
+elapses. A session_id may have opened more than one route (see "Multiple
+routes per session" below); this flushes all of them.
 
 Params:
 ```json
 { "session_id": "…", "timeout_ms": 10000 }
 ```
 Result:
 ```json
-{ "flushed": true, "pending": 0 }
+{ "flushed": true, "pending": 0, "accepted_sessions": 1 }
 ```
 `flushed: false` with `pending > 0` means the timeout was hit with work
-outstanding. Used by session-end hooks and flush-on-turn-end mode.
+outstanding across one or more routes. `accepted_sessions` counts how many
+independent routes this session_id has open. Used by session-end hooks and
+flush-on-turn-end mode.
 
 ### `managed_run.flush` (request)
 
@@ -147,6 +151,10 @@ Result:
     {
       "session_id": "…",
       "source": "codex",
+      "route": {
+        "auth": { "profile": "work", "org_name": "acme" },
+        "destination": { "type": "project_logs", "project_name": "codex" }
+      },
       "queued": 0,
       "spans_emitted": 42,
       "permalink": "https://www.braintrust.dev/app/…",
@@ -155,7 +163,10 @@ Result:
   ]
 }
 ```
-Powers a `status` CLI and pi's trace-link widget.
+`sessions` lists one entry **per route**, not per session_id: a session_id
+reporting to two destinations appears twice, each entry carrying its own
+`route`, counters, and permalink. Powers a `status` CLI and pi's trace-link
+widget.
 
 ### `daemon.shutdown` (request)
 
@@ -194,9 +205,11 @@ Field notes:
 
 - **`source`** selects the daemon-side translator. `debug` is a built-in
   pass-through translator used by the prototype and tests.
-- **`session_id`** is the per-session queue + state key. The shim extracts it
-  from the payload (default JSON field `session_id`, overridable with
-  `--session-id-field`); both Claude Code and Codex use `session_id`.
+- **`session_id`** identifies the source agent session. Combined with `route`
+  it forms the queue + state key (see "Multiple routes per session" below).
+  The shim extracts it from the payload (default JSON field `session_id`,
+  overridable with `--session-id-field`); both Claude Code and Codex use
+  `session_id`.
 - **`event`** is the agent-native hook name (not normalized). Extracted from
   the payload (default field `hook_event_name`, overridable with `--event`).
 - **`ts_ms`** is stamped by the shim **at capture time** (epoch millis),
@@ -211,13 +224,22 @@ Field notes:
   is optional and resolves through `bt`'s default profile when absent;
   `org_name` optionally constrains organization selection. The daemon resolves
   the live credential, pins the returned canonical profile for the lifetime
-  of the session, and refreshes an expiring lease without changing that route.
-  A route cannot change after a session's first accepted event. `destination`
-  is required so setup/run must make project or parent selection explicit.
-  `flush_mode` ∈ `fire_and_forget` | `flush_on_turn_end`.
-  New front-ends set the typed `destination`: `project_logs` accepts a project
-  id and/or name, `experiment` accepts an experiment id, and `parent_span`
-  carries the complete exported `SpanComponents` object.
+  of that route's pipeline, and refreshes an expiring lease without changing
+  the route. `destination` is required so setup/run must make project or
+  parent selection explicit. `flush_mode` ∈ `fire_and_forget` |
+  `flush_on_turn_end`. New front-ends set the typed `destination`:
+  `project_logs` accepts a project id and/or name, `experiment` accepts an
+  experiment id, and `parent_span` carries the complete exported
+  `SpanComponents` object.
+- **Multiple routes per session.** `session_id` plus the exact `route` forms
+  one independent delivery pipeline: its own auth resolution, translator,
+  sink, and queue. A session_id is not pinned to a single route — events for
+  the same session_id but a different route open a second, fully independent
+  pipeline rather than replacing or rejecting the first. This lets one source
+  session report concurrently to multiple destinations, including multiple
+  organizations (e.g. two `bt trace import` runs, or an active hook capture
+  alongside a concurrent import, targeting different projects or orgs for the
+  same underlying session).
 
 ### Redaction
 
@@ -277,15 +299,24 @@ continue using setup settings, and concurrent managed runs can select distinct
 profiles, organizations, and destinations while sharing one daemon.
 
 - **Journal (WAL).** Every accepted event is appended (auth-redacted) to
-  `<data_dir>/journal/<session_id>.ndjson` before/at enqueue. `data_dir`
-  defaults to `$XDG_STATE_HOME/braintrust/bt-daemon` or
+  `<data_dir>/journal/<session_id>.ndjson` before/at enqueue — one journal
+  file per session_id, shared across every route that session_id has opened.
+  `data_dir` defaults to `$XDG_STATE_HOME/braintrust/bt-daemon` or
   `$HOME/.braintrust/state/bt-daemon` on Unix, and
   `%LOCALAPPDATA%\Braintrust\bt-daemon` on Windows. On restart the daemon
-  rebuilds a session's unfinished correlation state by applying its journal to
-  a fresh translator. The resulting rows may be resubmitted to repair delivery
+  rebuilds each route's unfinished correlation state independently, replaying
+  only the journal entries whose `route` matches that pipeline into a fresh
+  translator. The resulting rows may be resubmitted to repair delivery
   interrupted by a crash, but their deterministic ids target the same backend
-  rows and must never create duplicate spans.
+  rows and must never create duplicate spans, and a route never receives
+  another route's rows.
   Journals are GC'd after 7 days.
+- **Managed-run acceptance records.** Alongside the journal, each accepted
+  event that carries a `managed_run_id` also appends `{session_id, route}` to
+  `<data_dir>/managed-runs/<managed_run_id>.ndjson`. `managed_run.flush` reads
+  this record when the daemon that accepted the events has since restarted or
+  idle-exited, so flush accounting for a child process tree survives a daemon
+  generation change. GC'd after 7 days like the journal.
 - **Deterministic span ids.** Translators derive span ids as UUIDv5 over stable
   keys (`session_id`, `turn_id`, `call_id`, …) so a replayed re-emit merges
   server-side (`_is_merge`) instead of duplicating.

bt-daemon/src/dispatch.rs

@@ -4,11 +4,10 @@
 //! concurrently.
 //!
 //! Ack semantics: `event.log` is acked once the event is journaled and handed
-//! to the session's queue (see [`Session::append_and_enqueue`]). Delivery to
+//! to the session's queue (see [`Session::enqueue`]). Delivery to
 //! Braintrust happens later in the actor; a downstream error never fails the
 //! caller's turn.
 
-use crate::journal::JournalWriter;
 use crate::sink::SinkFactory;
 use crate::translate::{Registry, SessionCtx};
 use crate::wire::Envelope;
@@ -33,7 +32,6 @@ enum SessionMsg {
 pub struct Session {
     pub source: String,
     tx: mpsc::UnboundedSender<SessionMsg>,
-    journal: tokio::sync::Mutex<JournalWriter>,
     pub counters: Arc<Counters>,
     pub last_error: Arc<Mutex<Option<String>>>,
     pub permalink: Arc<Mutex<Option<String>>>,
@@ -45,8 +43,8 @@ impl Session {
         session_id: String,
         source: String,
         plugin_version: Option<String>,
-        journal: JournalWriter,
         replay: Vec<Envelope>,
+        config: crate::wire::SessionConfig,
         translators: Arc<Registry>,
         sink_factory: Arc<dyn SinkFactory>,
     ) -> Arc<Session> {
@@ -65,26 +63,21 @@ impl Session {
             last_error: last_error.clone(),
             permalink: permalink.clone(),
             replay,
+            config,
         };
         tokio::spawn(actor.run(rx));
 
         Arc::new(Session {
             source,
             tx,
-            journal: tokio::sync::Mutex::new(journal),
             counters,
             last_error,
             permalink,
         })
     }
 
-    /// Journal (redacted) then enqueue. Both complete before the caller acks.
-    pub async fn append_and_enqueue(&self, mut env: Envelope) -> anyhow::Result<()> {
-        hydrate_transcript_snapshot(&mut env).await;
-        {
-            let mut j = self.journal.lock().await;
-            j.append(&env).await?;
-        }
+    /// Enqueue an event after the daemon has journaled it.
+    pub fn enqueue(&self, env: Envelope) -> anyhow::Result<()> {
         self.counters.queued.fetch_add(1, Ordering::Relaxed);
         self.tx
             .send(SessionMsg::Event(Box::new(env)))
@@ -130,7 +123,7 @@ impl Session {
 /// journal at lifecycle boundaries so recovery/replay does not depend on a
 /// path that Claude may later rewrite or delete. Fail open: a missing file is
 /// handled by the translator exactly as before.
-async fn hydrate_transcript_snapshot(env: &mut Envelope) {
+pub(crate) async fn hydrate_transcript_snapshot(env: &mut Envelope) {
     if env.source != "claude-code"
         || !matches!(
             env.event.as_str(),
@@ -173,6 +166,7 @@ struct SessionActor {
     last_error: Arc<Mutex<Option<String>>>,
     permalink: Arc<Mutex<Option<String>>>,
     replay: Vec<Envelope>,
+    config: crate::wire::SessionConfig,
 }
 
 impl SessionActor {
@@ -189,39 +183,36 @@ impl SessionActor {
                 // Still drain the queue so the daemon's counters settle and
                 // callers waiting on flush don't hang.
                 while let Some(msg) = rx.recv().await {
-                    if let SessionMsg::Event(_) = msg {
-                        self.counters.queued.fetch_sub(1, Ordering::Relaxed);
-                    } else if let SessionMsg::Configure(_, r) = msg {
-                        let _ = r.send(());
-                    } else if let SessionMsg::Flush(r) = msg {
-                        let _ = r.send(0);
-                    } else if let SessionMsg::Shutdown(r) = msg {
-                        let _ = r.send(());
-                        break;
+                    match msg {
+                        SessionMsg::Event(_) => {
+                            self.counters.queued.fetch_sub(1, Ordering::Relaxed);
+                        }
+                        SessionMsg::Configure(_, r) => {
+                            let _ = r.send(());
+                        }
+                        SessionMsg::Flush(r) => {
+                            let _ = r.send(0);
+                        }
+                        SessionMsg::Shutdown(r) => {
+                            let _ = r.send(());
+                            break;
+                        }
                     }
                 }
                 return;
             }
         };
         let mut ctx = SessionCtx {
             session_id: self.session_id.clone(),
-            config: None,
+            config: Some(self.config.clone()),
         };
-        // Rebuild translator state before accepting the first new event. Keep
-        // the deterministic replay ops buffered until live credentials arrive;
-        // then re-emitting them repairs any rows lost by a prior crash. The
-        // stable span ids ensure these target existing rows rather than create
-        // duplicate spans.
-        let mut replay_ops = Vec::new();
-        for env in &self.replay {
-            if let Some(cfg) = &env.config {
-                ctx.config = Some(cfg.clone());
-            }
-            match translator.handle(env, &ctx) {
-                Ok(mut ops) => replay_ops.append(&mut ops),
-                Err(e) => self.set_error(format!("journal replay failed: {e}")),
-            }
-        }
+        sink.configure(&self.config);
+        self.refresh_permalink(sink.as_ref());
+        // Rebuild translator state before accepting the first new event.
+        // Stable span ids make this both crash recovery and a complete copy
+        // when an existing source session is sent to another destination.
+        self.replay_into(&mut translator, &mut sink, &ctx, &self.replay)
+            .await;
 
         while let Some(msg) = rx.recv().await {
             match msg {
@@ -231,15 +222,6 @@ impl SessionActor {
                         ctx.config = Some(cfg.clone());
                         self.refresh_permalink(sink.as_ref());
                     }
-                    if !replay_ops.is_empty() {
-                        match sink.emit(&replay_ops).await {
-                            Ok(n) => {
-                                self.counters.spans_emitted.fetch_add(n, Ordering::Relaxed);
-                                replay_ops.clear();
-                            }
-                            Err(e) => self.set_error(format!("sink replay emit failed: {e}")),
-                        }
-                    }
                     match translator.handle(&env, &ctx) {
                         Ok(ops) => match sink.emit(&ops).await {
                             Ok(n) => {
@@ -270,6 +252,31 @@ impl SessionActor {
         }
     }
 
+    async fn replay_into(
+        &self,
+        translator: &mut Box<dyn crate::translate::AgentTranslator>,
+        sink: &mut Box<dyn crate::sink::Sink>,
+        ctx: &SessionCtx,
+        replay: &[Envelope],
+    ) {
+        let mut replay_ops = Vec::new();
+        for env in replay {
+            match translator.handle(env, ctx) {
+                Ok(mut ops) => replay_ops.append(&mut ops),
+                Err(e) => self.set_error(format!("journal replay failed: {e}")),
+            }
+        }
+        if replay_ops.is_empty() {
+            return;
+        }
+        match sink.emit(&replay_ops).await {
+            Ok(n) => {
+                self.counters.spans_emitted.fetch_add(n, Ordering::Relaxed);
+            }
+            Err(e) => self.set_error(format!("sink replay emit failed: {e}")),
+        }
+    }
+
     async fn drain_flush(
         &self,
         translator: &mut Box<dyn crate::translate::AgentTranslator>,