fix(daemon): make version handover forward-only

bt-daemon/Cargo.toml

@@ -27,6 +27,7 @@ fs2 = "0.4"
 regex = "1"
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
+semver = "1"
 sha2 = "0.10"
 sysinfo = { version = "0.33.1", default-features = false, features = ["system"] }
 thiserror = "2"

bt-daemon/docs/protocol.md

@@ -82,7 +82,12 @@ Params:
 ```json
 {
   "protocol_version": 1,
-  "client": { "source": "codex", "plugin_version": "1.2.3", "pid": 12345 }
+  "client": {
+    "source": "codex",
+    "daemon_version": "0.20.0",
+    "plugin_version": "1.2.3",
+    "pid": 12345
+  }
 }
 ```
 Result:
@@ -93,11 +98,11 @@ Result:
   "capabilities": { "sources": ["codex", "claude-code", "opencode", "pi", "debug"] }
 }
 ```
-If `protocol_version` is incompatible the daemon returns an application error;
-the client decides whether to drop events or (if the client is newer) trigger a
-version handover (`daemon.shutdown` → respawn).
+If `protocol_version` is incompatible the daemon returns an application error.
+Daemon versions move forward: an older client uses a newer compatible daemon,
+while a newer client drains and replaces an older daemon once.
 
-### `event.log` (request or notification)
+### `event.log` (request)
 
 The hot path. Params are the **Envelope** (see below). Request result:
 ```json
@@ -112,6 +117,11 @@ 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.
 
+During version handover, the draining daemon returns `{ "accepted": false }`.
+The client waits for the replacement daemon, initializes a new connection, and
+retries the same envelope. Capture adapters must use requests because a JSON-RPC
+notification has no acknowledgement and therefore cannot make this retry safe.
+
 ### `session.flush` (request)
 
 Block until every route's spans for the session are delivered, or `timeout_ms`
@@ -173,9 +183,12 @@ widget.
 
 ### `daemon.shutdown` (request)
 
-Graceful: stop accepting new events, drain all session queues, flush sinks,
-release the local endpoint, exit. Result `{ "ok": true }` is sent before exit.
-Used for version handover and by tests.
+Graceful: reject new events, wait for captures already being journaled, drain all
+session queues, and flush sinks. Result `{ "ok": true }` is sent only after the
+drain completes; the daemon then releases the local endpoint and exits. Used
+for version handover and explicit stop commands. A client whose declared daemon
+version is older than the running daemon receives `{ "ok": false }` and cannot
+downgrade it.
 
 ## Envelope (`event.log` params)
 
@@ -208,6 +221,10 @@ Field notes:
 
 - **`source`** selects the daemon-side translator. `debug` is a built-in
   pass-through translator used by the prototype and tests.
+
+The Braintrust sink keeps the capture package version in
+`context.span_origin.version` and adds `metadata.bt_daemon_version` to every
+span so the daemon build that performed translation can be queried separately.
 - **`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`,
@@ -288,10 +305,12 @@ journal, logs, status, or RPC response. Envelopes journal only their non-secret
   event rebuilds it from the journal, and deterministic span ids merge the
   re-emitted rows. This matters because the idle exit above requires *every*
   session to be quiet, which for a continuously active user never happens.
-- **Version handover.** `initialize` compares versions. A newer client sends
-  `daemon.shutdown`, waits until the endpoint no longer accepts connections,
-  and spawns its own daemon. In-flight session state is rebuilt from the
-  journal.
+- **Version handover.** `initialize` compares versions. Older clients use a
+  newer compatible daemon. A newer client sends `daemon.shutdown`; the old
+  daemon quiesces and drains before acknowledging, then the client spawns its
+  daemon. In-flight session state is rebuilt from the journal. This makes
+  upgrades monotonic instead of allowing mixed-version clients to repeatedly
+  replace each other.
 
 ## Durability & idempotence
 

bt-daemon/src/lib.rs

@@ -460,11 +460,12 @@ pub(crate) fn apply_tags(route: &mut SessionRoute, tags: &[String]) -> anyhow::R
     Ok(())
 }
 
-fn initialize_params(env: &Envelope) -> serde_json::Value {
+fn initialize_params(env: &Envelope, daemon_version: &str) -> serde_json::Value {
     serde_json::json!({
         "protocol_version": PROTOCOL_VERSION,
         "client": {
             "source": env.source,
+            "daemon_version": daemon_version,
             "plugin_version": env.plugin_version,
             "pid": std::process::id()
         }
@@ -480,36 +481,103 @@ pub async fn forward_envelope(
     host: &HostInfo,
     no_spawn: bool,
 ) -> anyhow::Result<()> {
-    let stream = client::ensure_daemon(socket, host, no_spawn).await?;
-    let mut conn = client::Conn::new(stream);
-    let initialized = conn
-        .request(method::INITIALIZE, initialize_params(env))
-        .await?;
-    let initialized: wire::InitializeResult = serde_json::from_value(initialized)?;
-    if initialized.daemon_version != host.version {
-        if no_spawn {
+    const MAX_HANDOVER_ATTEMPTS: usize = 3;
+    let mut handover_attempts = 0;
+    let mut capture_retry_deadline = None;
+
+    loop {
+        if capture_retry_deadline.is_some_and(|deadline| tokio::time::Instant::now() >= deadline) {
             anyhow::bail!(
-                "daemon version {} does not match client {} and --no-spawn is set",
-                initialized.daemon_version,
-                host.version
+                "replacement daemon did not accept the event at {}",
+                socket.display()
             );
         }
-        conn.request(method::DAEMON_SHUTDOWN, serde_json::json!({}))
-            .await?;
-        drop(conn);
-        for _ in 0..100 {
-            if client::connect(socket).await.is_err() {
-                break;
+
+        let stream = match client::connect(socket).await {
+            Ok(stream) => stream,
+            Err(_) if capture_retry_deadline.is_some() && no_spawn => {
+                tokio::time::sleep(std::time::Duration::from_millis(20)).await;
+                continue;
+            }
+            Err(_) => client::ensure_daemon(socket, host, no_spawn).await?,
+        };
+        let mut conn = client::Conn::new(stream);
+        let initialized = match conn
+            .request(method::INITIALIZE, initialize_params(env, &host.version))
+            .await
+        {
+            Ok(initialized) => initialized,
+            Err(_) if capture_retry_deadline.is_some() => {
+                tokio::time::sleep(std::time::Duration::from_millis(20)).await;
+                continue;
+            }
+            Err(error) => return Err(error),
+        };
+        let initialized: wire::InitializeResult = serde_json::from_value(initialized)?;
+        if daemon_needs_upgrade(&initialized.daemon_version, &host.version) {
+            if no_spawn {
+                anyhow::bail!(
+                    "daemon version {} is older than client {} and --no-spawn is set",
+                    initialized.daemon_version,
+                    host.version
+                );
+            }
+            if handover_attempts == MAX_HANDOVER_ATTEMPTS {
+                anyhow::bail!(
+                    "daemon version {} is still older than client {} after {MAX_HANDOVER_ATTEMPTS} handover attempts",
+                    initialized.daemon_version,
+                    host.version
+                );
+            }
+            conn.request(method::DAEMON_SHUTDOWN, serde_json::json!({}))
+                .await?;
+            handover_attempts += 1;
+            drop(conn);
+            wait_for_daemon_exit(socket).await;
+            continue;
+        }
+        let result = match conn.request(method::EVENT_LOG, env).await {
+            Ok(result) => result,
+            Err(_) if capture_retry_deadline.is_some() => {
+                tokio::time::sleep(std::time::Duration::from_millis(20)).await;
+                continue;
             }
-            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
+            Err(error) => return Err(error),
+        };
+        let result: wire::EventLogResult = serde_json::from_value(result)?;
+        if result.accepted {
+            return Ok(());
         }
-        let stream = client::ensure_daemon(socket, host, false).await?;
-        conn = client::Conn::new(stream);
-        conn.request(method::INITIALIZE, initialize_params(env))
-            .await?;
+        capture_retry_deadline
+            .get_or_insert_with(|| tokio::time::Instant::now() + std::time::Duration::from_secs(5));
+        tokio::time::sleep(std::time::Duration::from_millis(20)).await;
     }
-    conn.request(method::EVENT_LOG, env).await?;
-    Ok(())
+}
+
+async fn wait_for_daemon_exit(socket: &std::path::Path) {
+    for _ in 0..100 {
+        if client::connect(socket).await.is_err() {
+            return;
+        }
+        tokio::time::sleep(std::time::Duration::from_millis(20)).await;
+    }
+}
+
+/// A shared daemon only moves forward. Released versions use semver; opaque
+/// development versions retain the old exact-match handover behavior.
+fn daemon_needs_upgrade(daemon_version: &str, client_version: &str) -> bool {
+    if daemon_version == client_version {
+        return false;
+    }
+    compare_daemon_versions(daemon_version, client_version).is_none_or(|ordering| ordering.is_lt())
+}
+
+pub(crate) fn compare_daemon_versions(left: &str, right: &str) -> Option<std::cmp::Ordering> {
+    Some(
+        semver::Version::parse(left)
+            .ok()?
+            .cmp(&semver::Version::parse(right).ok()?),
+    )
 }
 
 /// Ask the daemon to flush a session, bounded by `timeout_ms`. A reliable
@@ -1451,8 +1519,9 @@ mod tests {
 
         // Both the initial connection and the post-restart retry use this
         // shared parameter builder.
-        let initialize = initialize_params(&env);
+        let initialize = initialize_params(&env, "1.0.13");
         assert_eq!(initialize["client"]["source"], "grok");
+        assert_eq!(initialize["client"]["daemon_version"], "1.0.13");
         assert_eq!(initialize["client"]["plugin_version"], "0.1.0");
         assert_ne!(initialize["client"]["plugin_version"], "1.0.13");
     }