fix(bt-daemon): report trace permalinks in session status

bt-daemon/docs/protocol.md

@@ -180,15 +180,18 @@ Result:
       },
       "queued": 0,
       "spans_emitted": 42,
-      "permalink": "https://www.braintrust.dev/app/…",
+      "permalink": "https://www.braintrust.dev/app/acme/p/codex/logs?r=…",
       "last_error": null
     }
   ]
 }
 ```
 `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
+`route`, counters, and permalink. `permalink` opens the route's most recently
+delivered trace once a flush succeeds. It stays `null` for experiment and
+parent-span destinations, for project logs addressed only by `project_id`, and
+when the route has no `org_name`. Powers a `status` CLI and pi's trace-link
 widget.
 
 ### `auth.diagnose` (request)

bt-daemon/src/sink/braintrust.rs

@@ -161,6 +161,7 @@ impl SinkFactory for BraintrustSinkFactory {
             client: None,
             open: HashMap::new(),
             pending_roots: HashMap::new(),
+            delivered_root: None,
         }))
     }
 }
@@ -199,6 +200,9 @@ struct BraintrustSink {
     /// Roots marked by translators as provisional, keyed by their local span
     /// id. Each is exported only once direct descendant work arrives.
     pending_roots: HashMap<String, Vec<SpanOp>>,
+    /// Root span id of the trace that most recently received a row, which
+    /// is also that root's deterministic row id in Braintrust.
+    delivered_root: Option<String>,
 }
 
 impl BraintrustSink {
@@ -438,6 +442,10 @@ impl Sink for BraintrustSink {
             }
             self.emit_op(op).await?;
             n += 1;
+            let root = op_root_span_id(op);
+            if self.delivered_root.as_deref() != Some(root) {
+                self.delivered_root = Some(root.to_string());
+            }
         }
         Ok(n)
     }
@@ -452,6 +460,33 @@ impl Sink for BraintrustSink {
     fn has_pending_delivery(&self) -> bool {
         !self.pending_roots.is_empty()
     }
+
+    fn permalink(&self) -> Option<String> {
+        let creds = self.creds.as_ref()?;
+        // Logs URLs address projects by name. Experiment and attached-parent
+        // destinations, and routes that only know a project id, have none.
+        match &creds.destination {
+            None
+            | Some(TraceDestination::ProjectLogs {
+                project_name: Some(_),
+                ..
+            })
+            | Some(TraceDestination::ProjectLogs {
+                project_id: None, ..
+            }) => {}
+            Some(_) => return None,
+        }
+        let org_name = creds.org_name.as_deref()?;
+        let root = self.delivered_root.as_deref()?;
+        let (_, app_url) = self.urls.as_ref()?;
+        Some(format!(
+            "{}/app/{}/p/{}/logs?r={}",
+            app_url.trim_end_matches('/'),
+            encode_url_component(org_name),
+            encode_url_component(&self.project(creds)),
+            encode_url_component(root),
+        ))
+    }
 }
 
 fn op_span_id(op: &SpanOp) -> &str {
@@ -460,6 +495,28 @@ fn op_span_id(op: &SpanOp) -> &str {
     }
 }
 
+fn op_root_span_id(op: &SpanOp) -> &str {
+    match op {
+        SpanOp::Insert(row) | SpanOp::Merge(row) => &row.root_span_id,
+    }
+}
+
+/// Percent-encodes everything outside the RFC 3986 unreserved set.
+fn encode_url_component(value: &str) -> String {
+    const HEX: &[u8; 16] = b"0123456789ABCDEF";
+    let mut encoded = String::with_capacity(value.len());
+    for byte in value.bytes() {
+        if byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.' | b'~') {
+            encoded.push(char::from(byte));
+        } else {
+            encoded.push('%');
+            encoded.push(char::from(HEX[usize::from(byte >> 4)]));
+            encoded.push(char::from(HEX[usize::from(byte & 0x0F)]));
+        }
+    }
+    encoded
+}
+
 fn direct_pending_parent(
     op: &SpanOp,
     pending_roots: &HashMap<String, Vec<SpanOp>>,

bt-daemon/tests/braintrust_sink.rs

@@ -909,3 +909,88 @@ async fn rejected_route_does_not_poison_a_sibling_flush_on_the_same_backend() {
     assert!(bad_result.is_err());
     good_result.unwrap();
 }
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn permalink_opens_the_most_recently_delivered_trace() {
+    let server = mock_backend().await;
+    let base = server.uri();
+    let factory = BraintrustSinkFactory::new(BraintrustSinkConfig::default());
+    let mut sink = factory.create("linked", "pi").unwrap();
+    let mut config = session_config(&base);
+    config.auth.app_url = Some(format!("{base}/"));
+    config.auth.org_name = Some("Acme Labs".into());
+    config.destination = Some(TraceDestination::ProjectLogs {
+        project_id: None,
+        project_name: Some("coding agents".into()),
+    });
+    sink.configure(&config);
+
+    let mut root = row("root-a", "root-a", &[], "pi", SpanType::Task, 1, None);
+    root.metadata = Some(json!({"_bt_defer_root": true}));
+    sink.emit(&[SpanOp::Insert(root)]).await.unwrap();
+    sink.flush().await.unwrap();
+    assert_eq!(sink.permalink(), None, "a held root is not yet a trace");
+
+    let turn = row(
+        "turn-a",
+        "root-a",
+        &["root-a"],
+        "Turn 1",
+        SpanType::Task,
+        2,
+        Some(3),
+    );
+    sink.emit(&[SpanOp::Insert(turn)]).await.unwrap();
+    sink.flush().await.unwrap();
+    assert_eq!(
+        sink.permalink().as_deref(),
+        Some(format!("{base}/app/Acme%20Labs/p/coding%20agents/logs?r=root-a").as_str())
+    );
+
+    let next = row("root-b", "root-b", &[], "pi", SpanType::Task, 4, Some(5));
+    sink.emit(&[SpanOp::Insert(next)]).await.unwrap();
+    assert_eq!(
+        sink.permalink().as_deref(),
+        Some(format!("{base}/app/Acme%20Labs/p/coding%20agents/logs?r=root-b").as_str())
+    );
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn permalink_is_absent_when_no_logs_url_can_be_built() {
+    let server = mock_backend().await;
+    let base = server.uri();
+    let factory = BraintrustSinkFactory::new(BraintrustSinkConfig::default());
+    let mut without_org = session_config(&base);
+    without_org.auth.org_name = None;
+    let mut project_id_only = session_config(&base);
+    project_id_only.destination = Some(TraceDestination::ProjectLogs {
+        project_id: Some("proj-1".into()),
+        project_name: None,
+    });
+    let mut experiment = session_config(&base);
+    experiment.destination = Some(TraceDestination::Experiment {
+        experiment_id: "exp-1".into(),
+    });
+
+    for (name, config) in [
+        ("without_org", without_org),
+        ("project_id_only", project_id_only),
+        ("experiment", experiment),
+    ] {
+        let mut sink = factory.create(name, "pi").unwrap();
+        sink.configure(&config);
+        sink.emit(&[SpanOp::Insert(row(
+            "root",
+            "root",
+            &[],
+            "pi",
+            SpanType::Task,
+            1,
+            Some(2),
+        ))])
+        .await
+        .unwrap();
+        sink.flush().await.unwrap();
+        assert_eq!(sink.permalink(), None, "{name} has no logs URL");
+    }
+}