diff --git a/baml_language/crates/baml_builtins2/keyword_docs/baml_keywords.yaml b/baml_language/crates/baml_builtins2/keyword_docs/baml_keywords.yaml
index e7dc86f2d3f..c76d65878da 100644
--- a/baml_language/crates/baml_builtins2/keyword_docs/baml_keywords.yaml
+++ b/baml_language/crates/baml_builtins2/keyword_docs/baml_keywords.yaml
@@ -209,6 +209,35 @@ trace.redact:
   details: |
     A recording keeps a marker saying the value was redacted in its place, wherever the value is captured: a function's inputs or output, a thrown error, or `log.*` data. `@trace.redact` applies to a class field, and `@@trace.redact` to a class, whose every instance is withheld whole, fields and all. Only recordings change: the program reads, prints and serializes the value as before. A value copied out of a redacted field into a plain string or another field is recorded as that copy. Takes no arguments, at most once per declaration.
 
+$trace:
+  summary: "The tracing options of one call: a hidden parameter every function takes, and a directive on a function declaration."
+  syntax: |
+    // At a call site, by name.
+    charge(account, 500, $trace = trace.span())
+    charge(account, 500, $trace = trace.context(
+        distinct_id = "user-42",
+        metadata = { "phase": "billing" },
+    ))
+
+    // On a declaration, for every call of the function.
+    /// baml:$trace=trace.span
+    function charge(account: Account, cents: int) -> int {
+        cents
+    }
+  details: |
+    A recording keeps a call as a span or as timing. A span is one row of `baml query`'s `spans` table, with its parent, times, status and captured values. A timing call adds to its function's totals in `profiler` and keeps no row and no values. At the default level (`BAML_TELEMETRY=medium`), AI functions are spans and other functions are timing. A root call and every `spawn` are spans of type `future`.
+
+    `trace.span()` records the call as a span and captures its inputs, its output and its error. `inputs = false`, `output = false` or `error = false` leaves one out, and `trace.empty_span()` leaves out all three. Any other span, an AI function's or one made by `BAML_TELEMETRY=high` included, records the error it ends with when it throws or panics, and a failed future carries the error that escaped it. `trace.timing()` keeps timing only and `trace.hidden()` leaves the call out. `trace.context(distinct_id = ..., metadata = {...})` sets the distinct id and metadata of the call and of everything it runs; `baml query` serves them as `context_distinct_id` and `context_metadata`. Builders chain: `trace.context(...).span()`. The mode and the capture requests apply to that one call, not to the calls it makes.
+
+    The directive `/// baml:$trace=<name>` takes `trace.span`, `trace.empty_span`, `trace.timing` or `trace.hidden`, or the name of a `throws never` function that returns the `trace.Options` to use. A value that must not be recorded is withheld with `@trace.redact`. See `baml describe trace` and `baml describe trace.redact`.
+
+tracing:
+  summary: "Alias topic for `$trace`."
+  syntax: |
+    charge(account, 500, $trace = trace.span())
+  details: |
+    See `baml describe '$trace'` for spans, timing, capture defaults and the declaration directive.
+
 function:
   summary: "Declares a named callable with typed parameters and a return type."
   syntax: |
@@ -1225,7 +1254,8 @@ query:
     Relations: `processes` (one row per process, with its status),
     `spans` (completed function calls that were retained, and futures: a
     root call is a future without a parent; `input_args`, `output_value`,
-    `error_value` (a `baml.errors.Context`), `self_usage` (the model usage
+    `error_value` (a `baml.errors.Context`: the error a failed call threw or
+    the error that escaped a failed future), `self_usage` (the model usage
     of the calls an `ai.Client.invoke` span made) and `total_usage` (every
     call under a span, one entry per model with its token counts) are BAML values), `span_announcements` (every
     started span), and `profiler` (one row per calling-context node of an
diff --git a/baml_language/crates/baml_query_btel/src/catalog.rs b/baml_language/crates/baml_query_btel/src/catalog.rs
index 4e69e102c6c..441303fd886 100644
--- a/baml_language/crates/baml_query_btel/src/catalog.rs
+++ b/baml_language/crates/baml_query_btel/src/catalog.rs
@@ -282,7 +282,7 @@ pub const SPANS: Relation = Relation {
         value("network_event_values", NETWORK_EVENTS_DOC),
         value(
             "error_value",
-            "the baml.errors.Context of a failed invocation; for a manual operation, the value passed to fail()",
+            "the baml.errors.Context of a failed invocation: of a function, the error it threw; of a future, the error that escaped it (recorded since format minor 14); for a manual operation, the value passed to fail(). A span records it by default; a cancellation only on request",
         ),
         col(
             "context_distinct_id",
@@ -381,7 +381,8 @@ SELECT __btel_pubid(r.recording_id, t.thread_id) AS span_id,
   NULL AS input_args,
   NULL AS output_value,
   NULL AS network_event_values,
-  NULL AS error_value,
+  -- The error that escaped the future's outermost frame (format minor 14).
+  IIF(t.outcome = 2, __btel_ref(3, t.error_cas, NULL, 0), NULL) AS error_value,
   cx.distinct_id AS context_distinct_id,
   __btel_context_metadata(ec.state, ec.cas, cx.state) AS context_metadata,
   __btel_usage_value(NULL) AS self_usage,
diff --git a/baml_language/crates/baml_query_btel/src/ingest.rs b/baml_language/crates/baml_query_btel/src/ingest.rs
index 07b9dcb427f..9f898201270 100644
--- a/baml_language/crates/baml_query_btel/src/ingest.rs
+++ b/baml_language/crates/baml_query_btel/src/ingest.rs
@@ -1188,7 +1188,7 @@ impl<'t> Applier<'t> {
                                 .tx
                                 .prepare_cached(
                                     "UPDATE thread SET completed_ticks = ?3, outcome = ?4, panicked = ?5,
-                                       usage = ?6
+                                       usage = ?6, error_cas = ?7
                                      WHERE rec = ?1 AND thread_id = ?2 AND outcome IS NULL",
                                 )?
                                 .execute(params![
@@ -1197,7 +1197,8 @@ impl<'t> Applier<'t> {
                                     completed,
                                     done.outcome,
                                     done.panicked,
-                                    crate::usage::to_json(&usage)
+                                    crate::usage::to_json(&usage),
+                                    done.error_cas_id.as_ref().map(cas_id)
                                 ])?;
                             if changed > 0 && !usage.is_empty() {
                                 self.charged.push((id(section.thread_id), usage));
diff --git a/baml_language/crates/baml_query_btel/src/ingest/bulk.rs b/baml_language/crates/baml_query_btel/src/ingest/bulk.rs
index 894ec7f505e..f8539945548 100644
--- a/baml_language/crates/baml_query_btel/src/ingest/bulk.rs
+++ b/baml_language/crates/baml_query_btel/src/ingest/bulk.rs
@@ -402,6 +402,8 @@ pub(super) struct Bulk {
     calls: Table<u64, CallRow>,
     /// Futures' names, first one wins; rare next to threads.
     thread_names: FxHashMap<u64, String>,
+    /// The error a thread's first completion carried; rare next to threads.
+    thread_errors: FxHashMap<u64, [u8; 16]>,
     /// Sysop count and ticks per call path.
     sysops: BTreeMap<u32, (u128, u128)>,
     /// The model usage a call's or thread's first completion carried, stored
@@ -662,6 +664,9 @@ impl Bulk {
                             match row.completion() {
                                 None => {
                                     row.complete(completed, outcome, done.panicked);
+                                    if let Some(error) = &done.error_cas_id {
+                                        self.thread_errors.insert(section.thread_id, cas_id(error));
+                                    }
                                     let usage = crate::usage::from_proto(&done.usage);
                                     if !usage.is_empty() {
                                         self.usage.insert(section.thread_id, usage);
@@ -1503,8 +1508,8 @@ impl Bulk {
             tx,
             "thread (rec, thread_id, defined, parent_id, spawn_call_path_id, started_ticks,
                epoch_id, announced, ran_ticks, completed_ticks, outcome, panicked, name,
-               completion_conflict, conflict, entry_call_path_id, usage)",
-            17,
+               completion_conflict, conflict, entry_call_path_id, usage, error_cas)",
+            18,
             &ids,
             |b, thread| {
                 let row = &self.threads[&thread];
@@ -1530,7 +1535,8 @@ impl Bulk {
                     self.usage
                         .get(&thread)
                         .and_then(|u| crate::usage::to_json(u)),
-                )
+                )?;
+                b.bind(self.thread_errors.get(&thread).map(<[u8; 16]>::as_slice))
             },
         )
     }
diff --git a/baml_language/crates/baml_query_btel/src/schema.rs b/baml_language/crates/baml_query_btel/src/schema.rs
index 3e1a7692952..807ccd99123 100644
--- a/baml_language/crates/baml_query_btel/src/schema.rs
+++ b/baml_language/crates/baml_query_btel/src/schema.rs
@@ -15,7 +15,7 @@
 //!   visible; they never make a thread a root or a path a top-level path.
 
 /// Physical layout of these tables. Change on any DDL change.
-pub const SCHEMA_VERSION: i64 = 22;
+pub const SCHEMA_VERSION: i64 = 23;
 /// Interpretation of evidence into rows. Change when reconciliation changes
 /// meaning without a DDL change; either mismatch rebuilds the index.
 pub const NORMALIZATION_VERSION: i64 = 1;
@@ -217,6 +217,9 @@ CREATE TABLE thread (
   -- as on `call`.
   usage TEXT,
   total_usage TEXT,
+  -- The baml.errors.Context of the error it ended with, from its first
+  -- completion (format minor 14); NULL when none was recorded.
+  error_cas BLOB,
   PRIMARY KEY (rec, thread_id)
 ) STRICT, WITHOUT ROWID;
 CREATE INDEX thread_by_parent ON thread (rec, parent_id);
diff --git a/baml_language/crates/baml_query_btel/tests/fresh_equivalence.rs b/baml_language/crates/baml_query_btel/tests/fresh_equivalence.rs
index 63d3c8e5a1f..f74caa07b94 100644
--- a/baml_language/crates/baml_query_btel/tests/fresh_equivalence.rs
+++ b/baml_language/crates/baml_query_btel/tests/fresh_equivalence.rs
@@ -213,9 +213,37 @@ async fn fresh_and_incremental_indexes_are_identical() {
         24 * 4,
         "type_args_cas"
     );
+    // Futures that failed carry the error they ended with: `crash` (a root
+    // future) in both recordings, and the future each `main` of the second
+    // one spawns. No other future names an error.
+    let threads = &fresh.iter().find(|(table, _)| table == "thread").unwrap().1;
+    let failed = |spawned: bool| {
+        threads
+            .iter()
+            .filter(|row| {
+                row[THREAD_OUTCOME] == Value::Integer(2)
+                    && (row[THREAD_PARENT] != Value::Null) == spawned
+            })
+            .collect::<Vec<_>>()
+    };
+    assert_eq!(failed(false).len(), 2, "one crash per recording");
+    assert_eq!(failed(true).len(), 24, "one failing spawn per main");
+    for row in threads {
+        assert_eq!(
+            row[THREAD_ERROR_CAS] != Value::Null,
+            row[THREAD_OUTCOME] == Value::Integer(2),
+            "{row:?}"
+        );
+    }
     assert_same(&fresh, &dump(&incremental));
 }
 
+/// Columns of `thread`, in table order.
+const THREAD_ID: usize = 1;
+const THREAD_PARENT: usize = 3;
+const THREAD_OUTCOME: usize = 10;
+const THREAD_ERROR_CAS: usize = 18;
+
 mod crafted {
     //! Evidence out of order and in conflict, written as recording files.
     use btel_recorder::proto::{self, function_definition::Resolution, span_event::Event};
@@ -227,6 +255,9 @@ mod crafted {
     const EPOCH: u64 = 5;
     const CALLER: u64 = 21;
     const CALLEE: u64 = 22;
+    /// The thread that fails with an error, and that error's snapshot.
+    pub(super) const FAILED_THREAD: u64 = T3;
+    pub(super) const T3_ERROR: u64 = 66;
 
     pub(super) fn write(project: &std::path::Path, files: Vec<proto::RecordingFile>) {
         let id = [7_u8; 16];
@@ -524,6 +555,18 @@ mod crafted {
                             network_done(202, 50, proto::InvocationOutcome::Unspecified),
                         ],
                     ),
+                    section(
+                        T3,
+                        vec![
+                            // Fails with an error, before its definition.
+                            Event::ThreadCompletion(proto::ThreadCompletion {
+                                completed_at_ticks: 70,
+                                outcome: 2,
+                                error_cas_id: Some(snapshot(T3_ERROR)),
+                                ..Default::default()
+                            }),
+                        ],
+                    ),
                 ],
             }),
             aggregates: Some(proto::AggregateBatch {
@@ -568,6 +611,8 @@ mod crafted {
                             outcome: 2,
                             panicked: true,
                             usage: Vec::new(),
+                            // The first completion is kept, without this error.
+                            error_cas_id: Some(snapshot(91)),
                         }),
                         announce(200, 4, "POST", 22),
                         network_event(200, "await", 41, None),
@@ -682,5 +727,31 @@ fn late_and_conflicting_evidence_indexes_identically() {
         })
         .collect();
     assert_eq!(type_args, vec![100]);
+    // A thread keeps the error of its first completion, whichever file
+    // defines the thread: the conflicting second completion's is dropped.
+    let errors: Vec<(i64, Value)> = fresh
+        .iter()
+        .find(|(table, _)| table == "thread")
+        .unwrap()
+        .1
+        .iter()
+        .map(|row| match &row[THREAD_ID] {
+            Value::Blob(id) => (
+                i64::from_be_bytes(id[..].try_into().unwrap()),
+                row[THREAD_ERROR_CAS].clone(),
+            ),
+            other => panic!("{other:?}"),
+        })
+        .filter(|(_, error)| *error != Value::Null)
+        .collect();
+    let mut expected = crafted::T3_ERROR.to_le_bytes().to_vec();
+    expected.extend([0; 8]);
+    assert_eq!(
+        errors,
+        vec![(
+            i64::try_from(crafted::FAILED_THREAD).unwrap(),
+            Value::Blob(expected)
+        )]
+    );
     assert_same(&fresh, &incremental);
 }
diff --git a/baml_language/crates/baml_query_btel/tests/network.rs b/baml_language/crates/baml_query_btel/tests/network.rs
index 42fcad4883b..e005dfb9661 100644
--- a/baml_language/crates/baml_query_btel/tests/network.rs
+++ b/baml_language/crates/baml_query_btel/tests/network.rs
@@ -333,6 +333,7 @@ fn network_spans_are_spans_with_their_request_and_events() {
                         outcome: proto::InvocationOutcome::Ok as i32,
                         panicked: false,
                         usage: Vec::new(),
+                        error_cas_id: None,
                     }),
                 ],
             )])),
diff --git a/baml_language/crates/baml_query_btel/tests/structure.rs b/baml_language/crates/baml_query_btel/tests/structure.rs
index cf50c0d62d2..8a334e0cd81 100644
--- a/baml_language/crates/baml_query_btel/tests/structure.rs
+++ b/baml_language/crates/baml_query_btel/tests/structure.rs
@@ -177,6 +177,7 @@ fn thread_done(ticks: u64, outcome: proto::InvocationOutcome, panicked: bool) ->
         outcome: outcome as i32,
         panicked,
         usage: Vec::new(),
+        error_cas_id: None,
     })
 }
 
@@ -901,6 +902,7 @@ fn self_usage_is_an_invokes_own_and_total_usage_sums_the_tree() {
                     outcome: proto::InvocationOutcome::Ok as i32,
                     panicked: false,
                     usage: vec![usage(Some("some-new-model"), 10, None, None)],
+                    error_cas_id: None,
                 }),
             ],
         )],
diff --git a/baml_language/crates/baml_query_btel/tests/thread_errors.rs b/baml_language/crates/baml_query_btel/tests/thread_errors.rs
new file mode 100644
index 00000000000..021fa14300a
--- /dev/null
+++ b/baml_language/crates/baml_query_btel/tests/thread_errors.rs
@@ -0,0 +1,242 @@
+//! A future's span carries the error the future ended with: the recording's
+//! `ThreadCompletion.error_cas_id`, served as `spans.error_value`. A future
+//! that returned or was cancelled carries none.
+mod support;
+
+use std::path::Path;
+
+use bex_engine::BexExternalValue;
+use btel_recorder::proto::{self, span_event::Event};
+use prost::Message as _;
+use serde_json::{Value, json};
+use support::*;
+
+const SOURCE: &str = r#"
+class Failure {
+    code: int,
+    reason: string,
+}
+
+function Thrower(n: int) -> int throws Failure {
+    throw Failure { code: n, reason: "refused" }
+}
+
+function fine(n: int) -> int {
+    n + 1
+}
+
+// A root future that fails with a thrown value.
+function crash(n: int) -> int {
+    Thrower(n)
+}
+
+// A root future that fails with a panic.
+function panics(n: int) -> int {
+    baml.sys.panic("out of options")
+}
+
+// A spawned future fails; the root future handles its error and returns.
+function caught_child(n: int) -> int {
+    let child = spawn { Thrower(n) };
+    (await child) catch (e) {
+        Failure => -1,
+    }
+}
+
+class Flag {
+    set: bool,
+}
+
+function until_cancelled(started: Flag) -> int {
+    started.set = true;
+    baml.sys.sleep(baml.time.Duration.from_milliseconds(60000));
+    1
+}
+
+// A spawned future is cancelled while it runs.
+function cancels(n: int) -> int {
+    let stop = baml.spawn.CancelToken.new();
+    let started = Flag { set: false };
+    let running = spawn with stop { until_cancelled(started) };
+    while (!started.set) {
+        baml.sys.sleep(baml.time.Duration.from_milliseconds(1));
+    }
+    let _ = stop.cancel();
+    (await baml.future.all_settled([running])).length()
+}
+"#;
+
+async fn record(project: &Path) {
+    let results = record_program(
+        project,
+        SOURCE,
+        &[],
+        &[
+            ("fine", 1),
+            ("crash", 7),
+            ("panics", 0),
+            ("caught_child", 9),
+            ("cancels", 0),
+        ],
+    )
+    .await;
+    assert_eq!(results[0], Ok(BexExternalValue::Int(2)));
+    assert!(results[1].is_err() && results[2].is_err(), "{results:?}");
+    assert_eq!(results[3], Ok(BexExternalValue::Int(-1)));
+    assert_eq!(results[4], Ok(BexExternalValue::Int(1)));
+}
+
+/// Every thread completion of the project's recordings, with the format
+/// minor of the file that carries it.
+fn completions(project: &Path) -> Vec<(u32, proto::ThreadCompletion)> {
+    let mut found = Vec::new();
+    for recording in std::fs::read_dir(project.join(".baml/btel/recordings")).unwrap() {
+        let mut files: Vec<_> = std::fs::read_dir(recording.unwrap().path())
+            .unwrap()
+            .map(|entry| entry.unwrap().path())
+            .filter(|path| path.extension().is_some_and(|ext| ext == "btel"))
+            .collect();
+        files.sort();
+        for path in files {
+            let file = proto::RecordingFile::decode(std::fs::read(path).unwrap().as_slice())
+                .expect("a recording file");
+            let minor = file.header.unwrap().format_minor;
+            for section in file.spans.unwrap_or_default().sections {
+                for event in section.events {
+                    if let Some(Event::ThreadCompletion(done)) = event.event {
+                        found.push((minor, done));
+                    }
+                }
+            }
+        }
+    }
+    found
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn a_failed_future_carries_its_error_in_the_recording() {
+    let project = tempfile::tempdir().unwrap();
+    record(project.path()).await;
+    let completions = completions(project.path());
+    let errored = proto::InvocationOutcome::Errored as i32;
+    let failed: Vec<_> = completions
+        .iter()
+        .filter(|(_, done)| done.outcome == errored)
+        .collect();
+    // `crash`, `panics` and the future `caught_child` spawned.
+    assert_eq!(failed.len(), 3, "{completions:?}");
+    for (minor, done) in &failed {
+        assert!(done.error_cas_id.is_some(), "{done:?}");
+        assert!(
+            *minor >= btel_settings::encoding::THREAD_ERROR_FORMAT_MINOR,
+            "a file with a thread error is minor {minor}"
+        );
+    }
+    assert_eq!(failed.iter().filter(|(_, done)| done.panicked).count(), 1);
+    // The futures that returned and the cancelled one name no error.
+    let others: Vec<_> = completions
+        .iter()
+        .filter(|(_, done)| done.outcome != errored)
+        .collect();
+    assert!(
+        others
+            .iter()
+            .any(|(_, done)| done.outcome == proto::InvocationOutcome::Cancelled as i32),
+        "{completions:?}"
+    );
+    assert!(
+        others
+            .iter()
+            .filter(|(_, done)| done.outcome == proto::InvocationOutcome::Ok as i32)
+            .count()
+            >= 3,
+        "{completions:?}"
+    );
+    for (_, done) in others {
+        assert_eq!(done.error_cas_id, None, "{done:?}");
+    }
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn a_failed_future_span_serves_the_error_it_ended_with() {
+    let project = tempfile::tempdir().unwrap();
+    record(project.path()).await;
+    let mut index = index(project.path());
+    let rows = sql(
+        &mut index,
+        "SELECT span_name, parent_span_id IS NULL AS root, status, error_value IS NULL AS no_error,
+           error_value['error'] AS error,
+           error_value['stack_trace']['frames'][0]['function_name'] AS outermost
+         FROM spans WHERE span_type = 'future' AND span_name != 'baml.gc'
+         ORDER BY start_time",
+    );
+    let by_name = |name: &str| -> &Vec<Value> {
+        let mut matching = rows
+            .rows
+            .iter()
+            .filter(|row| row[0].as_str().is_some_and(|n| n.contains(name)));
+        let row = matching
+            .next()
+            .unwrap_or_else(|| panic!("{name}: {:?}", rows.rows));
+        assert!(matching.next().is_none(), "{name}: {:?}", rows.rows);
+        row
+    };
+
+    // A root future that threw: the thrown value, and the stack it crossed,
+    // outermost frame first.
+    let crash = by_name("user.crash");
+    assert_eq!(
+        crash[1..4],
+        [json!(1), json!("user_error"), json!(0)],
+        "{crash:?}"
+    );
+    assert_eq!(crash[4]["$class"], json!("user.Failure"));
+    assert_eq!(crash[4]["code"], json!(7));
+    assert_eq!(crash[4]["reason"], json!("refused"));
+    assert_eq!(crash[5], json!("user.crash"));
+
+    // A root future that panicked.
+    let panics = by_name("user.panics");
+    assert_eq!(
+        panics[1..4],
+        [json!(1), json!("panic_error"), json!(0)],
+        "{panics:?}"
+    );
+    assert_eq!(panics[4]["$class"], json!("baml.panics.UserPanic"));
+    assert_eq!(panics[4]["message"], json!("out of options"));
+
+    // A spawned future that threw, under a root future that returned.
+    let child = by_name("lambda(caught_child");
+    assert_eq!(
+        child[1..4],
+        [json!(0), json!("user_error"), json!(0)],
+        "{child:?}"
+    );
+    assert_eq!(child[4]["code"], json!(9));
+    assert_eq!(child[5], child[0]);
+
+    // Futures that returned or were cancelled have no error.
+    for (name, status) in [
+        ("user.fine", "return"),
+        ("user.caught_child", "return"),
+        ("user.cancels", "return"),
+        ("lambda(cancels", "cancel_error"),
+    ] {
+        let row = by_name(name);
+        assert_eq!(row[2..5], [json!(status), json!(1), Value::Null], "{row:?}");
+    }
+
+    // The value is addressable in a predicate, like a function span's.
+    let found = sql(
+        &mut index,
+        "SELECT span_name FROM spans
+         WHERE span_type = 'future' AND error_value['error']['code'] = 7",
+    );
+    assert_eq!(found.rows, vec![vec![json!("user.crash")]]);
+    // The whole value is a `baml.errors.Context`.
+    let context = sql(
+        &mut index,
+        "SELECT error_value FROM spans WHERE span_type = 'future' AND span_name = 'user.crash'",
+    );
+    assert_eq!(context.rows[0][0]["$class"], json!("baml.errors.Context"));
+}
diff --git a/baml_language/crates/bcs_api/README.md b/baml_language/crates/bcs_api/README.md
index 5ead176b379..5ea0c316c52 100644
--- a/baml_language/crates/bcs_api/README.md
+++ b/baml_language/crates/bcs_api/README.md
@@ -59,7 +59,7 @@ The recording-level binding is independently explicit: omission or `false` freez
 
 Newly packed binaries and generated bridges default to `warn` for initial telemetry failures; publishers can explicitly choose `abort`, `warn`, or `ignore`. Authorization starts immediately in the background. Initial failure cancels BAML execution for `abort`; `warn` prints the configured message and continues; `ignore` continues silently. Bridge `abort` returns an operation error while the host process stays running. Ordinary caller-configured CLI cloud recording also defaults to `warn` and continues without cloud telemetry after an initial failure. After initial success, later telemetry failures remain nonfatal under every policy.
 
-At `medium` and `high`, network spans record request and response bodies, streamed events included. `http_bodies = false` keeps every body out of them at every level, for an API whose responses carry secrets such as session tokens. They still record the method, the sanitized URL and headers, the status, timing and whether the request failed, but not its error, which can quote the body. Model calls are network spans too, so this also keeps rendered prompts and raw model responses out; token usage is still recorded. Function spans record no error value either, whatever their trace options or function policy ask, because an error can carry a body, such as a model error's `raw_body`; a failed span still records that it failed. Their arguments and results are recorded as their trace policy says. Omission or `true` records bodies.
+At `medium` and `high`, network spans record request and response bodies, streamed events included. `http_bodies = false` keeps every body out of them at every level, for an API whose responses carry secrets such as session tokens. They still record the method, the sanitized URL and headers, the status, timing and whether the request failed, but not its error, which can quote the body. Model calls are network spans too, so this also keeps rendered prompts and raw model responses out; token usage is still recorded. Function spans record no error value either, whatever their trace options or function policy ask, because an error can carry a body, such as a model error's `raw_body`; a failed span still records that it failed, and a failed future does not carry the error that escaped it. Their arguments and results are recorded as their trace policy says. Omission or `true` records bodies.
 
 `--embed-telemetry` is a boolean opt-in and accepts no value. `[pack.build]` and `[bridge.build]` have identical settings: `embed_telemetry_api_key_from_env_var` selects the credential variable, and `embed_telemetry_env` names the expected environment. Pack reads `[pack.build]`; generate reads `[bridge.build]`. Both require `boundary.project = "org_handle/project_name"`. Without the flag the builder does not read the embedding credential or verify it.
 
diff --git a/baml_language/crates/bex_engine/tests/log_recording.rs b/baml_language/crates/bex_engine/tests/log_recording.rs
index 2a31019f3c4..9374ae6aab3 100644
--- a/baml_language/crates/bex_engine/tests/log_recording.rs
+++ b/baml_language/crates/bex_engine/tests/log_recording.rs
@@ -139,7 +139,7 @@ fn assert_logs(
                 };
                 assert!(
                     file.header.as_ref().unwrap().format_minor
-                        >= btel_settings::encoding::LOG_FORMAT_MINOR
+                        >= btel_settings::encoding::LOG_LINE_FORMAT_MINOR
                 );
                 let ContextReference::Snapshot(context) = reference(section) else {
                     panic!("missing log context")
@@ -175,6 +175,26 @@ fn assert_logs(
                         .as_ref()
                         .is_some_and(|map| log.pc < map.code_bytes)
                 );
+                // The event carries the line a reader resolves from the definition's line
+                // table, so a file without the definition still places it.
+                let own_file = function.source_span.as_ref().map(|span| span.file_id);
+                let resolved = btel_reader::source_map::SourceMap::from_wire(
+                    function.source_map.as_ref().unwrap(),
+                    own_file,
+                )
+                .unwrap()
+                .resolve(log.pc)
+                .unwrap();
+                assert_eq!(log.line, Some(resolved.line));
+                assert!(
+                    SOURCE
+                        .lines()
+                        .nth(resolved.line as usize - 1)
+                        .unwrap()
+                        .contains("log."),
+                    "line {} is a log call",
+                    resolved.line
+                );
                 if log.event_name.as_deref() == Some("span") {
                     assert!(spans.contains(&log.parent_id));
                 } else {
diff --git a/baml_language/crates/bex_vm/src/telemetry.rs b/baml_language/crates/bex_vm/src/telemetry.rs
index 6e88ae27929..cde782736c9 100644
--- a/baml_language/crates/bex_vm/src/telemetry.rs
+++ b/baml_language/crates/bex_vm/src/telemetry.rs
@@ -195,6 +195,9 @@ pub struct ThreadTelemetry {
     running: bool,
     completed: bool,
     name: Option<btel_records::ThreadName>,
+    /// The error that escaped the thread's outermost frame, until the thread's
+    /// completion, which is written with it.
+    error: Option<btel_snapshot::Snapshot>,
 }
 
 pub struct TelemetryState {
@@ -381,6 +384,7 @@ impl TelemetryState {
                 running: false,
                 completed: false,
                 name: None,
+                error: None,
             },
             call_paths: FxHashMap::default(),
             call_path_keys: FxHashMap::default(),
@@ -1379,6 +1383,22 @@ impl TelemetryState {
         }
     }
 
+    /// Capture `error`, the value that escaped the thread's outermost frame,
+    /// for the thread's completion. The last one wins: a host can resume a
+    /// thread after an unhandled throw, and only the throw that ends it counts.
+    ///
+    /// # Safety
+    /// `error` and its reachable objects must remain live under the heap permit.
+    pub unsafe fn record_thread_error(&mut self, error: Value) {
+        // Without HTTP bodies no error is recorded: it can quote a response
+        // body, as a failed `send_sse`'s does.
+        self.thread.error = if self.records_errors() {
+            self.capture(snapshot::Input::Value(error))
+        } else {
+            None
+        };
+    }
+
     pub fn complete_thread(&mut self, outcome: InvocationOutcome) {
         if self.thread.completed {
             return;
@@ -1399,6 +1419,18 @@ impl TelemetryState {
             self.write_span(SpanRecord::ThreadSpanRunning { at: completed_at });
         }
         self.flush_model_usage(self.thread.id);
+        // A thread that ended otherwise than with the error captured for it,
+        // returning after a host resumed it or cancelled, says nothing of it.
+        let error = self.thread.error.take();
+        if matches!(
+            outcome,
+            InvocationOutcome::Errored | InvocationOutcome::Panicked
+        ) && error.is_some()
+        {
+            self.write_span(SpanRecord::ThreadSpanError {
+                captured_error: error,
+            });
+        }
         self.write_span(SpanRecord::ThreadSpanCompletion {
             id: self.thread.id,
             parent_id: self.thread.parent_id,
@@ -3755,6 +3787,76 @@ mod tests {
         }
     }
 
+    #[test]
+    fn thread_completion_follows_the_error_the_thread_ended_with() {
+        for (outcome, recorded, carried) in [
+            (InvocationOutcome::Errored, true, true),
+            (InvocationOutcome::Panicked, true, true),
+            // No error was captured for the thread: nothing precedes its completion.
+            (InvocationOutcome::Errored, false, false),
+            // A host resumed the thread after the throw, or cancelled it: it
+            // did not end with the error.
+            (InvocationOutcome::Ok, true, false),
+            (InvocationOutcome::Cancelled, true, false),
+        ] {
+            let mut state = test_state(Arc::new(TelemetryPolicies::new()), test_clock());
+            state.start_thread();
+            if recorded {
+                // An earlier unhandled throw is superseded by the one that ends the thread.
+                unsafe {
+                    state.record_thread_error(Value::int(1));
+                    state.record_thread_error(Value::int(9));
+                }
+            }
+            state.complete_thread(outcome);
+            let records = state.span_records();
+            let completion = records
+                .iter()
+                .position(|record| matches!(record, SpanRecord::ThreadSpanCompletion { .. }))
+                .unwrap();
+            let errors: Vec<usize> = records
+                .iter()
+                .enumerate()
+                .filter(|(_, record)| matches!(record, SpanRecord::ThreadSpanError { .. }))
+                .map(|(index, _)| index)
+                .collect();
+            if carried {
+                assert_eq!(errors, vec![completion - 1], "{outcome:?}");
+                let SpanRecord::ThreadSpanError {
+                    captured_error: Some(error),
+                } = &records[completion - 1]
+                else {
+                    panic!("the thread error record owns its capture");
+                };
+                assert!(matches!(
+                    error.roots()[0],
+                    btel_snapshot::SnapshotValue::Int(9)
+                ));
+            } else {
+                assert!(errors.is_empty(), "{outcome:?}/{recorded}");
+            }
+            // A completed thread writes nothing more.
+            let written = state.span_records().len();
+            state.complete_thread(InvocationOutcome::Errored);
+            assert_eq!(state.span_records().len(), written);
+        }
+        // An engine that keeps HTTP bodies out of its recordings carries no
+        // thread error: it can quote a response body.
+        let mut state = test_state(
+            Arc::new(TelemetryPolicies::new().with_http_bodies(false)),
+            test_clock(),
+        );
+        state.start_thread();
+        unsafe { state.record_thread_error(Value::int(9)) };
+        state.complete_thread(InvocationOutcome::Errored);
+        assert!(
+            !state
+                .span_records()
+                .iter()
+                .any(|record| matches!(record, SpanRecord::ThreadSpanError { .. }))
+        );
+    }
+
     #[test]
     fn spawned_thread_has_explicit_parent_and_structural_call_path() {
         let mut parent = test_state(Arc::new(TelemetryPolicies::new()), test_clock());
diff --git a/baml_language/crates/bex_vm/src/vm.rs b/baml_language/crates/bex_vm/src/vm.rs
index 8b893e10057..ed95a06bbf5 100644
--- a/baml_language/crates/bex_vm/src/vm.rs
+++ b/baml_language/crates/bex_vm/src/vm.rs
@@ -6992,6 +6992,17 @@ impl BexVm {
                 if let Some(evidence) = evidence.take() {
                     self.error_not_caught(evidence, btel_records::UnwindResult::Unhandled);
                 }
+                // No frame is left to carry the error: the thread's own
+                // completion does, as its `baml.errors.Context`.
+                if self.telemetry.is_some() {
+                    let recorded = self.recorded_context(&mut context, exception_value);
+                    if let Some(state) = self.telemetry.as_mut() {
+                        // SAFETY: unwinding runs under the heap permit, and
+                        // `recorded` is the in-flight context or the thrown
+                        // value, both rooted by the unwinder.
+                        unsafe { state.record_thread_error(recorded) };
+                    }
+                }
                 let trace = match &context {
                     InFlightContext::Carried(context) => self.context_trace(*context)?,
                     InFlightContext::Fresh { trace, .. } => trace.to_vec(),
diff --git a/baml_language/crates/btel_recorder/proto/recording.proto b/baml_language/crates/btel_recorder/proto/recording.proto
index 09dc704a18b..73f35c936ec 100644
--- a/baml_language/crates/btel_recorder/proto/recording.proto
+++ b/baml_language/crates/btel_recorder/proto/recording.proto
@@ -42,6 +42,8 @@ message RecordingHeader {
                                  // 12 adds general owned-function closure and named input roots.
                                  // 13 adds model usage on completions and model
                                  // invoke functions.
+                                 // 14 adds the source line of a log event and the
+                                 // error a thread ended with.
   bytes recording_id = 3;       // Exactly 16 random bytes, not all zero.
   optional bytes source_snapshot_id = 4; // Exactly 32 bytes when supplied; not identity.
   // Minor 3. The OS process that ran this engine. Every engine of one
@@ -338,6 +340,12 @@ message LogEvent {
   LogLevel level = 5;
   optional string event_name = 6; // Absent is unnamed; an empty string remains present.
   CasId data_cas_id = 7;         // May be unavailable; null data is a captured value.
+  // Minor 14. One-based line of the log call in the emitting function's own
+  // source file, from the line table of that function's definition. A function
+  // is defined once per recording, so a reader of a later file alone cannot
+  // place `pc`; the line travels with the event. Absent when the function has
+  // no line table or the table does not place `pc` in the function's own file.
+  optional uint32 line = 8;
 }
 
 enum LogLevel {
@@ -363,6 +371,11 @@ message ThreadCompletion {
   InvocationOutcome outcome = 2;
   bool panicked = 3;            // Minor 3. Only with ERRORED: a panic ended it.
   repeated ModelUsage usage = 4; // Minor 13. Usage charged to the thread node.
+  // Minor 14. Only with ERRORED: the baml.errors.Context of the error that
+  // escaped the thread's outermost frame, as a function completion's value is
+  // for the function it escaped. Absent from older producers, for a thread
+  // that was cancelled, and when the error could not be captured.
+  CasId error_cas_id = 5;
 }
 
 message FunctionAnnouncement {
diff --git a/baml_language/crates/btel_recorder/src/encoding.rs b/baml_language/crates/btel_recorder/src/encoding.rs
index 980ac8b044b..19bc9b1d933 100644
--- a/baml_language/crates/btel_recorder/src/encoding.rs
+++ b/baml_language/crates/btel_recorder/src/encoding.rs
@@ -97,7 +97,11 @@ impl EncodedSpans {
             Event::FunctionCompletion(m) => self.completion(4, &m),
             Event::LateFunctionCompletion(m) => self.completion(5, &m),
             Event::Log(message) => {
-                self.require(btel_settings::encoding::LOG_FORMAT_MINOR);
+                self.require(if message.line.is_some() {
+                    btel_settings::encoding::LOG_LINE_FORMAT_MINOR
+                } else {
+                    btel_settings::encoding::LOG_FORMAT_MINOR
+                });
                 self.variable(6, &message);
             }
             Event::NetworkAnnouncement(message) => {
@@ -112,7 +116,12 @@ impl EncodedSpans {
                 let length = begin(&mut self.bytes, 2, 1);
                 match other {
                     Event::ThreadAnnouncement(m) => leaf(&mut self.bytes, 1, &m),
-                    Event::ThreadCompletion(m) => leaf(&mut self.bytes, 2, &m),
+                    Event::ThreadCompletion(m) => {
+                        if m.error_cas_id.is_some() {
+                            self.require(btel_settings::encoding::THREAD_ERROR_FORMAT_MINOR);
+                        }
+                        leaf(&mut self.bytes, 2, &m);
+                    }
                     Event::ThreadRunning(m) => {
                         self.require(btel_settings::encoding::THREAD_RUNNING_FORMAT_MINOR);
                         leaf(&mut self.bytes, 7, &m);
@@ -356,6 +365,10 @@ mod tests {
                 outcome: proto::InvocationOutcome::Errored as i32,
                 panicked: true,
                 usage: Vec::new(),
+                error_cas_id: Some(proto::CasId {
+                    low: u64::MAX,
+                    high: u64::MAX,
+                }),
             }),
             Event::FunctionAnnouncement(proto::FunctionAnnouncement {
                 manual: false,
diff --git a/baml_language/crates/btel_recorder/src/functions.rs b/baml_language/crates/btel_recorder/src/functions.rs
index edddf54f875..ccaa18f22d9 100644
--- a/baml_language/crates/btel_recorder/src/functions.rs
+++ b/baml_language/crates/btel_recorder/src/functions.rs
@@ -26,6 +26,26 @@ impl FunctionDefinitions {
             .or_insert_with(|| metadata.clone());
     }
 
+    /// The one-based line of `pc` in `function`'s own source file, as the
+    /// function's line table places it: what a reader resolves from the
+    /// published definition (`btel_reader::source_map`). `None` without a
+    /// table, for a PC the table does not cover, and for code inlined from
+    /// another file, whose line is not a line of this function's file.
+    pub(crate) fn line(&self, function: FunctionId, pc: u32) -> Option<u32> {
+        let metadata = self
+            .host
+            .get(&function)
+            .or_else(|| self.table.as_ref().and_then(|table| table.get(function)))?;
+        let map = metadata.source_map.as_ref()?;
+        if pc == u32::MAX || pc >= map.code_bytes {
+            return None;
+        }
+        let covering = map.entries.partition_point(|entry| entry.pc <= pc);
+        let entry = map.entries.get(covering.checked_sub(1)?)?;
+        let own_file = metadata.source_span.as_ref()?.file_id;
+        (entry.file_id == own_file).then_some(entry.line)
+    }
+
     /// The definition to publish for a newly referenced function, if any.
     /// Known metadata is published once per recording: later files resolve
     /// the reference through the earlier definition. Unknown functions keep
@@ -117,3 +137,129 @@ fn source_map(m: &FunctionMetadata, map: &btel_types::SourceMap) -> proto::Sourc
         line: map.entries.iter().map(|e| e.line).collect(),
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use btel_types::{SourceMap, SourceMapEntry, SourceSpan};
+
+    use super::*;
+
+    const OWN_FILE: u32 = 3;
+    const OTHER_FILE: u32 = 4;
+
+    fn entry(pc: u32, file_id: u32, line: u32) -> SourceMapEntry {
+        SourceMapEntry {
+            pc,
+            file_id,
+            start: line * 10,
+            end: line * 10 + 5,
+            line,
+        }
+    }
+
+    fn metadata(
+        function_id: FunctionId,
+        source_span: Option<SourceSpan>,
+        source_map: Option<SourceMap>,
+    ) -> FunctionMetadata {
+        FunctionMetadata {
+            function_id,
+            fqn: "user.Logger".into(),
+            display_name: "Logger".into(),
+            source_file: Some("main.baml".into()),
+            source_span,
+            kind: btel_types::RuntimeFunctionKind::Bytecode,
+            origin: btel_types::RuntimeFunctionOrigin::UserDefined,
+            owner_type: None,
+            parent_function: None,
+            lambda_path: None,
+            definition_key: None,
+            package_name: Some("user".into()),
+            namespace: Vec::new(),
+            argument_layout: None,
+            source_map,
+            model_invoke: false,
+        }
+    }
+
+    fn own_span() -> SourceSpan {
+        SourceSpan {
+            file_id: OWN_FILE,
+            start: 0,
+            end: 500,
+        }
+    }
+
+    /// Code of 30 bytes: nothing maps PCs 0..4, lines 10 and 11 of the
+    /// function's own file follow, and PCs 20..30 are code from another file.
+    fn map() -> SourceMap {
+        SourceMap {
+            code_bytes: 30,
+            entries: vec![
+                entry(4, OWN_FILE, 10),
+                entry(12, OWN_FILE, 11),
+                entry(20, OTHER_FILE, 99),
+            ],
+        }
+    }
+
+    #[test]
+    fn line_is_the_covering_entry_of_the_functions_own_file() {
+        let allocator = btel_types::FunctionIdAllocator::default();
+        let ids: Vec<FunctionId> = (0..5).map(|_| allocator.allocate().unwrap()).collect();
+        let [mapped, unmapped, no_span, host, unknown] = ids[..] else {
+            unreachable!()
+        };
+        let mut definitions = FunctionDefinitions::default();
+        // Without a table, no function has a line.
+        assert_eq!(definitions.line(mapped, 4), None);
+        definitions.set_table(Arc::new(FunctionMetadataTable {
+            functions: vec![
+                metadata(mapped, Some(own_span()), Some(map())),
+                metadata(unmapped, Some(own_span()), None),
+                metadata(no_span, None, Some(map())),
+            ],
+        }));
+        definitions.register_host(&metadata(host, Some(own_span()), Some(map())));
+
+        for (pc, line) in [
+            // Before the first entry: no instruction there is mapped.
+            (0, None),
+            (3, None),
+            // An entry applies from its PC until the next entry's.
+            (4, Some(10)),
+            (11, Some(10)),
+            (12, Some(11)),
+            (19, Some(11)),
+            // Code inlined from another file has no line in this one.
+            (20, None),
+            (29, None),
+            // At or past the end of the code, and the saturated sentinel.
+            (30, None),
+            (31, None),
+            (u32::MAX - 1, None),
+            (u32::MAX, None),
+        ] {
+            assert_eq!(definitions.line(mapped, pc), line, "pc {pc}");
+            // A host-registered definition resolves like a table one.
+            assert_eq!(definitions.line(host, pc), line, "host pc {pc}");
+            assert_eq!(definitions.line(unmapped, pc), None, "no map, pc {pc}");
+            // Without a definition span there is no own file to be a line of.
+            assert_eq!(definitions.line(no_span, pc), None, "no span, pc {pc}");
+            assert_eq!(definitions.line(unknown, pc), None, "unknown, pc {pc}");
+        }
+
+        // An empty table covers no PC.
+        let empty = allocator.allocate().unwrap();
+        definitions.register_host(&metadata(
+            empty,
+            Some(own_span()),
+            Some(SourceMap {
+                code_bytes: 30,
+                entries: Vec::new(),
+            }),
+        ));
+        assert_eq!(definitions.line(empty, 0), None);
+        assert_eq!(definitions.line(empty, 29), None);
+    }
+}
diff --git a/baml_language/crates/btel_recorder/src/lib.rs b/baml_language/crates/btel_recorder/src/lib.rs
index baa3fa7f759..f1539721df3 100644
--- a/baml_language/crates/btel_recorder/src/lib.rs
+++ b/baml_language/crates/btel_recorder/src/lib.rs
@@ -72,6 +72,8 @@ struct ConversionBuffer {
     // Model usage awaiting its node's completion, which the VM writes right
     // after it; kept across files like `unwinds`. Empty without model calls.
     usage: BTreeMap<u64, Vec<proto::ModelUsage>>,
+    /// The error each thread ended with, until the thread's completion, which carries it.
+    thread_errors: BTreeMap<u64, proto::CasId>,
 }
 // Count each newly constructed metadata message once; never rescan the file.
 fn push_message<M: prost::Message>(output: &mut Vec<M>, bytes: &mut usize, tag: u32, value: M) {
@@ -81,6 +83,11 @@ fn push_message<M: prost::Message>(output: &mut Vec<M>, bytes: &mut usize, tag:
 impl ConversionBuffer {
     fn event(&mut self, thread: TelemetryId, mut event: proto::span_event::Event) {
         use proto::span_event::Event;
+        if let Event::ThreadCompletion(completion) = &mut event
+            && let Some(error) = self.thread_errors.remove(&thread.get())
+        {
+            completion.error_cas_id = Some(error);
+        }
         if !self.usage.is_empty() {
             match &mut event {
                 Event::FunctionCompletion(completion)
@@ -244,6 +251,7 @@ impl ConversionBuffer {
                         level: level as i32,
                         event_name: event_name.as_deref().map(str::to_owned),
                         data_cas_id: captured_data.as_ref().map(cas_id),
+                        line: self.functions.line(*function, *pc),
                     }),
                 );
             }
@@ -269,6 +277,11 @@ impl ConversionBuffer {
                     ))),
                 );
             }
+            SpanRecord::ThreadSpanError { captured_error } => {
+                if let Some(error) = captured_error.as_ref().map(cas_id) {
+                    self.thread_errors.insert(thread.get(), error);
+                }
+            }
             SpanRecord::SpanUsage(usage) => {
                 self.usage.insert(
                     usage.node.get(),
@@ -336,6 +349,7 @@ impl ConversionBuffer {
                         outcome: flags::outcome(*outcome) as i32,
                         panicked: *outcome == btel_types::InvocationOutcome::Panicked,
                         usage: Vec::new(),
+                        error_cas_id: None,
                     }),
                 );
                 // A raise that never ended cannot claim anything after its thread.
diff --git a/baml_language/crates/btel_recorder/src/recording/tests.rs b/baml_language/crates/btel_recorder/src/recording/tests.rs
index 468879ca101..eed4855281a 100644
--- a/baml_language/crates/btel_recorder/src/recording/tests.rs
+++ b/baml_language/crates/btel_recorder/src/recording/tests.rs
@@ -1820,3 +1820,204 @@ fn owned_span_uses_function_wire_shape_and_general_neutral_close() {
     assert_eq!(done.self_await_ticks, 4);
     assert_eq!(done.exited_at_ticks, 9);
 }
+
+#[test]
+fn a_log_carries_its_line_and_raises_the_file_minor_only_with_one() {
+    use proto::span_event::Event;
+
+    let files = RefCell::new(Vec::new());
+    let allocator = FunctionIdAllocator::default();
+    let (mapped, unmapped) = (allocator.allocate().unwrap(), allocator.allocate().unwrap());
+    let metadata_for = |function_id, source_map| btel_types::FunctionMetadata {
+        function_id,
+        fqn: "user.Logger".into(),
+        display_name: "Logger".into(),
+        source_file: Some("main.baml".into()),
+        source_span: Some(btel_types::SourceSpan {
+            file_id: 3,
+            start: 0,
+            end: 500,
+        }),
+        kind: btel_types::RuntimeFunctionKind::Bytecode,
+        origin: btel_types::RuntimeFunctionOrigin::UserDefined,
+        owner_type: None,
+        parent_function: None,
+        lambda_path: None,
+        definition_key: None,
+        package_name: Some("user".into()),
+        namespace: Vec::new(),
+        argument_layout: None,
+        source_map,
+        model_invoke: false,
+    };
+    let table = Arc::new(btel_types::FunctionMetadataTable {
+        functions: vec![
+            metadata_for(
+                mapped,
+                Some(btel_types::SourceMap {
+                    code_bytes: 30,
+                    entries: vec![btel_types::SourceMapEntry {
+                        pc: 4,
+                        file_id: 3,
+                        start: 100,
+                        end: 110,
+                        line: 12,
+                    }],
+                }),
+            ),
+            metadata_for(unmapped, None),
+        ],
+    });
+    let mut p = RecordingPublisher::new(RecordingId::generate(), config(), |f| {
+        files.borrow_mut().push(f);
+    })
+    .unwrap()
+    .with_function_metadata(table);
+    let thread = allocate_telemetry_id();
+    // One log per file: (function, pc) and the line the event must carry.
+    let cases = [
+        (mapped, 4, Some(12)),
+        (mapped, 29, Some(12)),
+        // Before the first mapped instruction, past the code, the sentinel.
+        (mapped, 0, None),
+        (mapped, 30, None),
+        (mapped, u32::MAX, None),
+        (unmapped, 4, None),
+    ];
+    for (function, pc, _) in cases {
+        p.span(
+            thread,
+            &mut SpanRecord::Log {
+                parent_id: thread,
+                function,
+                pc,
+                at: ClockInstant::from_ticks(1),
+                level: btel_records::LogLevel::Warn,
+                event_name: None,
+                captured_data: None,
+            },
+        );
+        p.flush_recording().unwrap();
+    }
+    let files = files.borrow();
+    assert_eq!(files.len(), cases.len());
+    for (file, (function, pc, line)) in files.iter().zip(cases) {
+        let file = decode(file);
+        // A file without a line stays readable at the minor logs came with.
+        assert_eq!(
+            file.header.unwrap().format_minor,
+            if line.is_some() {
+                btel_settings::encoding::LOG_LINE_FORMAT_MINOR
+            } else {
+                btel_settings::encoding::LOG_FORMAT_MINOR
+            },
+            "pc {pc}"
+        );
+        let Some(Event::Log(log)) = file
+            .spans
+            .unwrap()
+            .sections
+            .remove(0)
+            .events
+            .remove(0)
+            .event
+        else {
+            panic!("a log event");
+        };
+        assert_eq!((log.function_id, log.pc), (function.get(), pc));
+        assert_eq!(log.line, line, "pc {pc}");
+    }
+}
+
+#[test]
+fn a_thread_error_travels_on_its_threads_completion_and_raises_the_file_minor() {
+    use proto::span_event::Event;
+
+    let files = RefCell::new(Vec::new());
+    let received = RefCell::new(Vec::new());
+    let mut p = RecordingPublisher::with_snapshot_receiver(
+        RecordingId::generate(),
+        config(),
+        |f| files.borrow_mut().push(f),
+        |snapshot| received.borrow_mut().push(crate::cas_id(&snapshot)),
+    )
+    .unwrap();
+    let clock = ClockRuntime::new(ClockMode::Monotonic).start_run();
+    let complete = |p: &mut RecordingPublisher<_, _>, thread, outcome| {
+        p.span(
+            thread,
+            &mut SpanRecord::ThreadSpanCompletion {
+                id: thread,
+                parent_id: None,
+                spawn_call_path: CallPathId::ROOT,
+                started_at: ClockInstant::from_ticks(1),
+                completed_at: ClockInstant::from_ticks(2),
+                outcome,
+                clock: Arc::clone(&clock),
+                name: None,
+            },
+        );
+    };
+    let error = snapshot();
+    let error_id = crate::cas_id(&error);
+    let (failed, plain, uncaptured) = (
+        allocate_telemetry_id(),
+        allocate_telemetry_id(),
+        allocate_telemetry_id(),
+    );
+    p.span(
+        failed,
+        &mut SpanRecord::ThreadSpanError {
+            captured_error: Some(error),
+        },
+    );
+    // The error waits for its own thread's completion, across a file and
+    // another thread's completion.
+    p.flush_recording().unwrap();
+    complete(&mut p, plain, InvocationOutcome::Errored);
+    p.flush_recording().unwrap();
+    complete(&mut p, failed, InvocationOutcome::Panicked);
+    p.flush_recording().unwrap();
+    // A capture the pool refused leaves the completion without an error.
+    p.span(
+        uncaptured,
+        &mut SpanRecord::ThreadSpanError {
+            captured_error: None,
+        },
+    );
+    complete(&mut p, uncaptured, InvocationOutcome::Errored);
+    p.flush_recording().unwrap();
+
+    // The error's snapshot is retained for the CAS like any capture.
+    assert_eq!(*received.borrow(), [error_id]);
+    let completions: Vec<(u32, u64, proto::ThreadCompletion)> = files
+        .borrow()
+        .iter()
+        .map(decode)
+        .filter_map(|file| {
+            let minor = file.header.unwrap().format_minor;
+            let mut section = file.spans?.sections.remove(0);
+            match section.events.remove(0).event {
+                Some(Event::ThreadCompletion(done)) => Some((minor, section.thread_id, done)),
+                other => panic!("only completions are events here: {other:?}"),
+            }
+        })
+        .collect();
+    let base = btel_settings::encoding::FORMAT_MINOR;
+    assert_eq!(
+        completions
+            .iter()
+            .map(|(minor, thread, done)| (*minor, *thread, done.error_cas_id, done.panicked))
+            .collect::<Vec<_>>(),
+        vec![
+            (base, plain.get(), None, false),
+            (
+                btel_settings::encoding::THREAD_ERROR_FORMAT_MINOR,
+                failed.get(),
+                Some(error_id),
+                true
+            ),
+            (base, uncaptured.get(), None, false),
+        ]
+    );
+}
diff --git a/baml_language/crates/btel_records/src/lib.rs b/baml_language/crates/btel_records/src/lib.rs
index 0b59723801c..d40d90f711f 100644
--- a/baml_language/crates/btel_records/src/lib.rs
+++ b/baml_language/crates/btel_records/src/lib.rs
@@ -168,6 +168,13 @@ pub enum SpanRecord<InputCapture, ValueCapture> {
     /// before that node's completion, which carries it. Boxed so the model
     /// names never widen the slot.
     SpanUsage(Box<SpanUsage>),
+    /// Rare: the error that escaped the thread's outermost frame, written
+    /// just before the thread's completion, which carries it. A record of its
+    /// own keeps the capture out of `ThreadSpanCompletion`, which already
+    /// fills the slot.
+    ThreadSpanError {
+        captured_error: Option<ValueCapture>,
+    },
     /// An HTTP request started. Boxed, like the event and completion below,
     /// so its method and URL never widen the slot.
     NetworkSpanAnnouncement(Box<NetworkAnnouncement<InputCapture>>),
@@ -1344,6 +1351,7 @@ macro_rules! capture_slots {
             Self::ManualSpanAnnouncement(entry) => [Some($input($($borrow)+ entry.captured_inputs)), Some($input($($borrow)+ entry.captured_type_args))],
             Self::ManualSpanCompletion(done) => [Some($value($($borrow)+ done.captured_value)), None],
             Self::Log { captured_data, .. } => [Some($value(captured_data)), None],
+            Self::ThreadSpanError { captured_error } => [Some($value(captured_error)), None],
             Self::ContextSelected { captured_context } => [Some($input(captured_context)), None],
             Self::FunctionSpanAnnouncement {
                 captured_inputs,
diff --git a/baml_language/crates/btel_settings/src/encoding.rs b/baml_language/crates/btel_settings/src/encoding.rs
index b744179c484..33b63ab6408 100644
--- a/baml_language/crates/btel_settings/src/encoding.rs
+++ b/baml_language/crates/btel_settings/src/encoding.rs
@@ -55,3 +55,13 @@ pub const OWNED_SPAN_FORMAT_MINOR: u32 = 12;
 /// (`FunctionCompletion.usage`, `ThreadCompletion.usage`), and function
 /// metadata marks model invoke functions (`FunctionMetadata.model_invoke`).
 pub const MODEL_USAGE_FORMAT_MINOR: u32 = 13;
+
+/// Log events carry the source line of the log call (`LogEvent.line`), so a
+/// reader that has not seen the emitting function's definition, which a
+/// recording publishes once, can still place the event in its file.
+pub const LOG_LINE_FORMAT_MINOR: u32 = 14;
+
+/// A thread's completion carries the error the thread ended with
+/// (`ThreadCompletion.error_cas_id`), so a failed future, a failed root call
+/// among them, says what it failed with as a failed function span does.
+pub const THREAD_ERROR_FORMAT_MINOR: u32 = 14;
diff --git a/skills/baml-core/SKILL.md b/skills/baml-core/SKILL.md
index e2dd238bbf9..eda2afcbad2 100644
--- a/skills/baml-core/SKILL.md
+++ b/skills/baml-core/SKILL.md
@@ -86,6 +86,45 @@ The rest of the language:
 - **Concurrency = green threads.** `spawn { … }` returns a `Future`; `await` collects it. Combine many with `baml.future.all` / `all_settled` / `race` / `any`. `all_settled` returns `Success<T>`, `Failure<E>`, or `Panicked` for each input; input cancellation is a `Panicked` outcome. Configure a spawn with a `with` clause: `spawn with limit, token { … }`. Each value listed is a `baml.spawn.Modifier` that transforms the plan the spawn builds, left to right. `baml.spawn.Limit.new(n)` caps concurrency (excess spawns queue), a `baml.spawn.CancelToken` cancels cooperatively, `baml.spawn.Root.new()` detaches the task from its spawner's cancellation. Use it to parallelize LLM and HTTP calls. `baml describe spawn` / `baml describe baml.future`.
 - **Resource safety.** `defer { … }` runs a block at scope exit, LIFO, on *every* path (return / throw / fall-through), like Go. A class method named `function cleanup(self) -> void` is a **finalizer**: it runs at most once per instance whether you call it, `defer` it, or the GC reclaims it.
 
+## Tracing
+
+Every `baml run` and `baml test` records what ran, and `baml query` reads it back (see [Debugging a recorded call](#debugging-a-recorded-call)).
+
+- **Spans and timing.** A recording keeps a call as a *span* or as *timing*. A span is one row of the `spans` table, with its parent, times, status and captured values. A timing call adds to its function's totals in `profiler` and keeps no row and no values. AI functions are spans by default, and other functions are timing. A root call and every `spawn` are spans of type `future`.
+- **Make a function a span** with a directive above its declaration: `/// baml:$trace=trace.span`. `trace.empty_span` (a span without values), `trace.timing` and `trace.hidden` are the other choices.
+- **Or choose per call.** Every function takes a hidden `$trace` parameter, passed by name: `fee(paid, $trace = trace.span())`. `trace.context(distinct_id = …, metadata = {…})` labels the call and everything it runs; `spans` serves the labels as `context_distinct_id` and `context_metadata['phase']`. Builders chain: `trace.context(…).span()`.
+- **What a span captures.** `trace.span()` captures the call's inputs, output and error (`input_args`, `output_value`, `error_value`). `inputs = false`, `output = false` or `error = false` leaves one out, and `trace.empty_span()` leaves out all three. Any other span records the error it ends with when it throws or panics, whether or not it captures anything else. A failed future, a failed root call included, carries the error that escaped it. A timing call captures nothing.
+- **Keep a value out** with `@trace.redact` on a class field or `@@trace.redact` on a class. A recording holds a marker in its place wherever the value would be captured.
+
+```baml
+class Account {
+    name: string,
+    api_key: string @trace.redact,
+}
+
+/// baml:$trace=trace.span
+function charge(account: Account, cents: int) -> int {
+    cents
+}
+
+function fee(cents: int) -> int {
+    cents / 100
+}
+
+function checkout(account: Account) -> int {
+    let paid = charge(account, 500);
+    let billing = trace.context(distinct_id = account.name, metadata = { "phase": "billing" });
+    paid + fee(paid, $trace = billing.span())
+}
+
+test "checkout charges and adds the fee" {
+    let account = Account { name: "ann", api_key: "sk-1" };
+    assert.equal(checkout(account), 505);
+}
+```
+
+`baml describe '$trace'` / `baml describe trace` / `baml describe trace.redact`.
+
 ## LLM functions
 
 - **LLM function = typed return.** The RETURN TYPE *is* the schema the model must produce (`class`, `enum`, literal union, `string[]`, `T?`). Structured output is just a typed value; hand it to ordinary code.
@@ -116,7 +155,7 @@ To add cross-cutting behavior (metering, caching, replay, budgets), write a clas
 
 ### Debugging a recorded call
 
-Runs record their calls, and AI functions record their inputs, output, error and type arguments by default. `baml query` answers SQL over them (`baml query --schema` lists the tables). To turn a failing call into a test:
+Runs record their calls. AI functions record their inputs, output, error and type arguments by default, and every failed span and future records its error (see [Tracing](#tracing)). `baml query` answers SQL over them (`baml query --schema` lists the tables). To turn a failing call into a test:
 
 ```bash
 # 1. Find the call.