Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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: |
Expand Down Expand Up @@ -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
Expand Down
5 changes: 3 additions & 2 deletions baml_language/crates/baml_query_btel/src/catalog.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down Expand Up @@ -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,
Expand Down
5 changes: 3 additions & 2 deletions baml_language/crates/baml_query_btel/src/ingest.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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![
Expand All @@ -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));
Expand Down
12 changes: 9 additions & 3 deletions baml_language/crates/baml_query_btel/src/ingest/bulk.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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];
Expand All @@ -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))
},
)
}
Expand Down
5 changes: 4 additions & 1 deletion baml_language/crates/baml_query_btel/src/schema.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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);
Expand Down
71 changes: 71 additions & 0 deletions baml_language/crates/baml_query_btel/tests/fresh_equivalence.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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};
Expand All @@ -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];
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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),
Expand Down Expand Up @@ -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);
}
1 change: 1 addition & 0 deletions baml_language/crates/baml_query_btel/tests/network.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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,
}),
],
)])),
Expand Down
2 changes: 2 additions & 0 deletions baml_language/crates/baml_query_btel/tests/structure.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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,
})
}

Expand Down Expand Up @@ -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,
}),
],
)],
Expand Down
Loading
Loading