use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
mod tensor_wire;
use crate::{
ObservationCatalog, ObservationPoint, ObservationSupportReport, ObservationSupportStatus,
SymbolicDimension, TensorObservation,
};
pub const CAPTURE_SCHEMA_VERSION: u32 = 1;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CapturePhase {
Prefill,
Decode,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct CaptureSchedule {
pub prefill: bool,
pub decode: bool,
pub first_prediction: u64,
pub end_prediction: Option<u64>,
pub every: u64,
}
impl Default for CaptureSchedule {
fn default() -> Self {
Self {
prefill: true,
decode: true,
first_prediction: 0,
end_prediction: None,
every: 1,
}
}
}
impl CaptureSchedule {
pub fn count_and_last(
&self,
phase: CapturePhase,
maximum: u64,
) -> Result<Option<(u64, u64)>, CaptureError> {
self.count_and_last_from(phase, 0, maximum)
}
pub fn count_and_last_from(
&self,
phase: CapturePhase,
next_prediction: u64,
maximum: u64,
) -> Result<Option<(u64, u64)>, CaptureError> {
if self.every == 0 {
return Err(CaptureError::Invalid("zero capture frequency".into()));
}
if phase == CapturePhase::Prefill {
return Ok(
(next_prediction == 0 && maximum > 0 && self.includes(phase, 0)).then_some((1, 0)),
);
}
if !self.decode {
return Ok(None);
}
let end = self.end_prediction.unwrap_or(maximum).min(maximum);
let lower = self.first_prediction.max(1).max(next_prediction);
if lower >= end {
return Ok(None);
}
let offset = (lower - self.first_prediction).div_ceil(self.every);
let first = add(self.first_prediction, mul(offset, self.every)?)?;
if first >= end {
return Ok(None);
}
let steps = (end - 1 - first) / self.every;
Ok(Some((add(steps, 1)?, add(first, mul(steps, self.every)?)?)))
}
pub fn includes(&self, phase: CapturePhase, prediction: u64) -> bool {
(match phase {
CapturePhase::Prefill => self.prefill,
CapturePhase::Decode => self.decode,
}) && prediction >= self.first_prediction
&& self.end_prediction.is_none_or(|end| prediction < end)
&& self.every != 0
&& (prediction - self.first_prediction).is_multiple_of(self.every)
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct CaptureSlice {
pub axis: String,
pub start: u64,
pub end: u64,
pub stride: u64,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum CaptureTransform {
Preview {
max_elements: u64,
},
Slice,
FullTensor,
Summary,
Histogram {
edges: Vec<f32>,
},
TopCandidates {
count: u64,
},
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CaptureTransformKind {
Preview,
Slice,
FullTensor,
Summary,
Histogram,
TopCandidates,
}
impl CaptureTransform {
pub fn kind(&self) -> CaptureTransformKind {
match self {
Self::Preview { .. } => CaptureTransformKind::Preview,
Self::Slice => CaptureTransformKind::Slice,
Self::FullTensor => CaptureTransformKind::FullTensor,
Self::Summary => CaptureTransformKind::Summary,
Self::Histogram { .. } => CaptureTransformKind::Histogram,
Self::TopCandidates { .. } => CaptureTransformKind::TopCandidates,
}
}
}
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct CaptureCapabilities {
pub transformations: Vec<CaptureTransformKind>,
pub max_histogram_bins: u64,
pub physical_native_limit: bool,
pub conditions: Vec<String>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct CaptureSelection {
pub id: String,
pub path: String,
pub schedule: CaptureSchedule,
pub slices: Vec<CaptureSlice>,
pub transform: CaptureTransform,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CaptureLimitPolicy {
Fail,
Skip,
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct CaptureUsage {
pub captures: u64,
pub retained_bytes: u64,
pub host_bytes: u64,
pub encoded_bytes: u64,
}
impl CaptureUsage {
pub fn checked_mul(self, count: u64) -> Result<Self, CaptureError> {
Ok(Self {
captures: mul(self.captures, count)?,
retained_bytes: mul(self.retained_bytes, count)?,
host_bytes: mul(self.host_bytes, count)?,
encoded_bytes: mul(self.encoded_bytes, count)?,
})
}
pub fn checked_add(self, other: Self) -> Result<Self, CaptureError> {
Ok(Self {
captures: add(self.captures, other.captures)?,
retained_bytes: add(self.retained_bytes, other.retained_bytes)?,
host_bytes: add(self.host_bytes, other.host_bytes)?,
encoded_bytes: add(self.encoded_bytes, other.encoded_bytes)?,
})
}
pub fn exceeded(self, limit: Self) -> Option<CaptureBudget> {
if self.captures > limit.captures {
Some(CaptureBudget::Captures)
} else if self.retained_bytes > limit.retained_bytes {
Some(CaptureBudget::Retention)
} else if self.host_bytes > limit.host_bytes {
Some(CaptureBudget::Host)
} else if self.encoded_bytes > limit.encoded_bytes {
Some(CaptureBudget::Encoded)
} else {
None
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct CaptureLimits {
pub per_step: CaptureUsage,
pub cumulative: CaptureUsage,
pub physical_native_bytes: Option<u64>,
pub on_limit: CaptureLimitPolicy,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct CapturePlan {
pub schema_version: u32,
pub selections: Vec<CaptureSelection>,
pub limits: CaptureLimits,
}
impl CapturePlan {
pub fn none() -> Self {
Self {
schema_version: CAPTURE_SCHEMA_VERSION,
selections: Vec::new(),
limits: CaptureLimits {
per_step: CaptureUsage::default(),
cumulative: CaptureUsage::default(),
physical_native_bytes: None,
on_limit: CaptureLimitPolicy::Fail,
},
}
}
pub fn admit(
self,
catalog: &ObservationCatalog,
support: &ObservationSupportReport,
capabilities: &CaptureCapabilities,
request: CaptureRequestShape,
) -> Result<AdmittedCapturePlan, CaptureError> {
if self.schema_version != CAPTURE_SCHEMA_VERSION
|| catalog.schema_version != crate::DISCOVERY_SCHEMA_VERSION
|| support.schema_version != crate::DISCOVERY_SCHEMA_VERSION
{
return Err(CaptureError::Invalid("unsupported schema version".into()));
}
if self.limits.physical_native_bytes.is_some() && !capabilities.physical_native_limit {
return Err(CaptureError::Unsupported(
"physical native allocator/workspace bound".into(),
));
}
if request.batch == 0 || request.prompt_tokens == 0 || request.max_predictions == 0 {
return Err(CaptureError::Invalid(
"batch, prompt, and prediction limits must be positive".into(),
));
}
add(request.prompt_tokens, request.max_predictions)?;
mul(request.batch, request.prompt_tokens)?;
let mut ids = std::collections::BTreeSet::new();
let mut points = Vec::new();
for selection in &self.selections {
if selection.id.is_empty() || !ids.insert(selection.id.as_str()) {
return Err(CaptureError::Invalid(
"capture IDs must be nonempty and unique".into(),
));
}
let point = catalog
.get(&selection.path)
.ok_or_else(|| CaptureError::MissingPath(selection.path.clone()))?;
if !capabilities
.transformations
.contains(&selection.transform.kind())
{
return Err(CaptureError::Unsupported(format!(
"{:?}",
selection.transform.kind()
)));
}
if selection.schedule.every == 0
|| selection
.schedule
.end_prediction
.is_some_and(|end| end <= selection.schedule.first_prediction)
{
return Err(CaptureError::Invalid("invalid capture schedule".into()));
}
let phase_support = support
.points
.iter()
.find(|p| p.path == selection.path)
.ok_or_else(|| {
CaptureError::Unsupported(format!("no selected support for {}", selection.path))
})?;
for (enabled, status) in [
(selection.schedule.prefill, &phase_support.prefill),
(selection.schedule.decode, &phase_support.decode),
] {
if enabled && !matches!(status, ObservationSupportStatus::Supported) {
return Err(CaptureError::Unsupported(format!(
"{}: {status:?}",
selection.path
)));
}
}
if matches!(selection.transform, CaptureTransform::Slice) && selection.slices.is_empty()
{
return Err(CaptureError::Invalid(
"slice capture requires an explicit axis slice; use FullTensor to opt in"
.into(),
));
}
if let CaptureTransform::Histogram { edges } = &selection.transform {
if edges.len() < 2
|| (edges.len() - 1) as u64 > capabilities.max_histogram_bins
|| edges.iter().any(|edge| !edge.is_finite())
|| edges.windows(2).any(|w| w[0] >= w[1])
{
return Err(CaptureError::Invalid(
"histogram edges must be finite, increasing, and within the bin limit"
.into(),
));
}
}
if let CaptureTransform::TopCandidates { count } = selection.transform {
if count == 0
|| selection.path != crate::MODEL_LOGITS_OBSERVATION_PATH
|| !selection.slices.is_empty()
{
return Err(CaptureError::Invalid("candidate capture requires positive count, unsliced model.logits, and single-sequence execution".into()));
}
if request.batch != 1 {
return Err(CaptureError::Unsupported(
"candidate capture requires batch one".into(),
));
}
if let Some(SymbolicDimension::Known(vocabulary)) = point
.axes
.as_ref()
.and_then(|axes| axes.last())
.map(|a| &a.dimension)
{
if count > *vocabulary as u64 {
return Err(CaptureError::Invalid(
"candidate count exceeds vocabulary".into(),
));
}
}
}
let mut axes = std::collections::BTreeSet::new();
for slice in &selection.slices {
if slice.stride == 0 || slice.start > slice.end || !axes.insert(&slice.axis) {
return Err(CaptureError::Invalid(
"invalid or duplicate axis slice".into(),
));
}
if !point
.axes
.as_ref()
.is_some_and(|axes| axes.iter().any(|axis| axis.name == slice.axis))
{
return Err(CaptureError::Invalid(format!(
"unknown axis {}",
slice.axis
)));
}
}
for phase in [CapturePhase::Prefill, CapturePhase::Decode] {
if let Some((count, last)) = selection
.schedule
.count_and_last(phase, request.max_predictions)?
{
let first = last - mul(count - 1, selection.schedule.every)?;
for prediction in [first, last] {
for slice in &selection.slices {
let axis = point
.axes
.as_ref()
.and_then(|axes| axes.iter().find(|axis| axis.name == slice.axis))
.expect("axis was validated");
if request
.extent(&axis.dimension, phase, prediction)?
.is_some_and(|extent| slice.end > extent)
{
return Err(CaptureError::Invalid(format!(
"slice {} exceeds known request extent",
slice.axis
)));
}
}
if let Some(shape) = request.resolve(point, phase, prediction)? {
resolve_slice(point, selection, &shape)?;
}
}
}
}
points.push(point.clone());
}
let bytes = serde_json::to_vec(&(&self, &points, request))
.map_err(|e| CaptureError::Invalid(e.to_string()))?;
let identity = Sha256::digest(bytes)
.iter()
.map(|byte| format!("{byte:02x}"))
.collect();
Ok(AdmittedCapturePlan {
plan: self,
points,
request,
identity,
})
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub struct CaptureRequestShape {
pub batch: u64,
pub prompt_tokens: u64,
pub max_predictions: u64,
}
impl CaptureRequestShape {
fn extent(
self,
dimension: &SymbolicDimension,
phase: CapturePhase,
prediction: u64,
) -> Result<Option<u64>, CaptureError> {
let sequence = if phase == CapturePhase::Prefill {
self.prompt_tokens
} else {
1
};
Ok(match dimension {
SymbolicDimension::Known(n) => {
Some(u64::try_from(*n).map_err(|_| CaptureError::Overflow)?)
}
SymbolicDimension::Batch => Some(self.batch),
SymbolicDimension::Sequence => Some(sequence),
SymbolicDimension::TokenRows => Some(mul(self.batch, sequence)?),
SymbolicDimension::Context => Some(add(self.prompt_tokens, prediction)?),
SymbolicDimension::MediaPositions | SymbolicDimension::Unknown => None,
})
}
pub fn validate_actual(
self,
point: &ObservationPoint,
phase: CapturePhase,
prediction: u64,
shape: &[u64],
) -> Result<(), CaptureError> {
let Some(axes) = &point.axes else {
if shape.len() > 32 {
return Err(CaptureError::Unsupported(
"capture rank exceeds the 32-axis metadata bound".into(),
));
}
return elements(shape).map(|_| ());
};
if axes.len() != shape.len() {
return Err(CaptureError::Invalid(
"runtime rank differs from the catalog".into(),
));
}
for (axis, actual) in axes.iter().zip(shape) {
let expected = self.extent(&axis.dimension, phase, prediction)?;
if expected.is_some_and(|expected| expected != *actual) {
return Err(CaptureError::Invalid(format!(
"runtime extent for {} differs from catalog/request",
axis.name
)));
}
}
elements(shape).map(|_| ())
}
pub fn resolve(
self,
point: &ObservationPoint,
phase: CapturePhase,
prediction: u64,
) -> Result<Option<Vec<u64>>, CaptureError> {
let Some(axes) = &point.axes else {
return Ok(None);
};
let mut shape = Vec::with_capacity(axes.len());
for axis in axes {
let Some(extent) = self.extent(&axis.dimension, phase, prediction)? else {
return Ok(None);
};
shape.push(extent);
}
elements(&shape)?;
Ok(Some(shape))
}
}
#[derive(Debug, Clone)]
pub struct AdmittedCapturePlan {
plan: CapturePlan,
points: Vec<ObservationPoint>,
request: CaptureRequestShape,
identity: String,
}
impl AdmittedCapturePlan {
pub fn identity(&self) -> &str {
&self.identity
}
pub fn plan(&self) -> &CapturePlan {
&self.plan
}
pub fn points(&self) -> &[ObservationPoint] {
&self.points
}
pub fn request(&self) -> CaptureRequestShape {
self.request
}
pub fn is_empty(&self) -> bool {
self.plan.selections.is_empty()
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ResolvedCaptureSlice {
pub starts: Vec<u64>,
pub ends: Vec<u64>,
pub strides: Vec<u64>,
pub shape: Vec<u64>,
}
pub fn resolve_slice(
point: &ObservationPoint,
selection: &CaptureSelection,
shape: &[u64],
) -> Result<ResolvedCaptureSlice, CaptureError> {
if point
.axes
.as_ref()
.is_some_and(|axes| axes.len() != shape.len())
{
return Err(CaptureError::Invalid(
"runtime tensor rank differs from catalog".into(),
));
}
let mut output = ResolvedCaptureSlice {
starts: vec![0; shape.len()],
ends: shape.to_vec(),
strides: vec![1; shape.len()],
shape: shape.to_vec(),
};
for slice in &selection.slices {
let axis = point
.axes
.as_ref()
.and_then(|axes| axes.iter().position(|axis| axis.name == slice.axis))
.ok_or_else(|| CaptureError::Invalid(format!("unknown axis {}", slice.axis)))?;
if slice.stride == 0 || slice.start > slice.end || slice.end > shape[axis] {
return Err(CaptureError::Invalid(format!(
"slice {} exceeds runtime extent",
slice.axis
)));
}
output.starts[axis] = slice.start;
output.ends[axis] = slice.end;
output.strides[axis] = slice.stride;
output.shape[axis] = (slice.end - slice.start).div_ceil(slice.stride);
}
elements(&output.shape)?;
Ok(output)
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct CaptureSummary {
pub elements: u64,
pub finite: u64,
pub non_finite: u64,
pub nan: u64,
pub positive_infinity: u64,
pub negative_infinity: u64,
pub min: Option<f64>,
pub max: Option<f64>,
pub mean: Option<f64>,
pub rms: Option<f64>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct CaptureHistogram {
pub edges: Vec<f32>,
pub counts: Vec<u64>,
pub below: u64,
pub above: u64,
pub non_finite: u64,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "kind", content = "value", rename_all = "snake_case")]
pub enum CapturePayload {
Tensor(#[serde(with = "tensor_wire")] TensorObservation),
Summary(CaptureSummary),
Histogram(CaptureHistogram),
Candidates(CaptureCandidates),
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CandidateScoreStage {
RawLogitsBeforeSampling,
}
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CandidateLogitsSource {
#[default]
Original,
Effective,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct CaptureCandidate {
pub token_id: u32,
pub score: f32,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct CaptureCandidates {
pub stage: CandidateScoreStage,
#[serde(default)]
pub source: CandidateLogitsSource,
pub candidates: Vec<CaptureCandidate>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CaptureBudget {
Captures,
Retention,
Host,
Encoded,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum CaptureOutcome {
Captured,
Truncated {
available_elements: u64,
emitted_elements: u64,
},
Skipped {
reason: CaptureSkipReason,
},
Missing,
Failed {
reason: CaptureFailureReason,
message: String,
},
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum CaptureFailureReason {
Limit {
budget: CaptureBudget,
cumulative: bool,
},
Unsupported,
Invalid,
Native,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum CaptureSkipReason {
Schedule,
Limit {
budget: CaptureBudget,
cumulative: bool,
},
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct CaptureRecord {
pub schema_version: u32,
pub selection_id: String,
pub path: String,
pub node_id: String,
pub position: crate::ObservationPosition,
pub source_shape: Option<Vec<u64>>,
pub selected_shape: Option<Vec<u64>>,
pub outcome: CaptureOutcome,
pub payload: Option<CapturePayload>,
pub charged: CaptureUsage,
}
#[derive(Debug)]
pub struct CaptureLedger {
limits: CaptureLimits,
step: CaptureUsage,
total: CaptureUsage,
}
impl CaptureLedger {
pub fn new(plan: &AdmittedCapturePlan) -> Self {
Self {
limits: plan.plan.limits.clone(),
step: CaptureUsage::default(),
total: CaptureUsage::default(),
}
}
pub fn with_inherited_usage(
plan: &AdmittedCapturePlan,
inherited: CaptureUsage,
) -> Result<Self, CaptureError> {
if let Some(budget) = inherited.exceeded(plan.plan.limits.cumulative) {
return Err(CaptureError::Limit {
budget,
cumulative: true,
});
}
Ok(Self {
limits: plan.plan.limits.clone(),
step: CaptureUsage::default(),
total: inherited,
})
}
pub fn begin_step(&mut self) {
self.step = CaptureUsage::default();
}
pub fn step(&self) -> CaptureUsage {
self.step
}
pub fn total(&self) -> CaptureUsage {
self.total
}
pub fn reserve(
&mut self,
usage: CaptureUsage,
) -> Result<Option<CaptureSkipReason>, CaptureError> {
let step = self.step.checked_add(usage)?;
let total = self.total.checked_add(usage)?;
let exceeded = step
.exceeded(self.limits.per_step)
.map(|b| (b, false))
.or_else(|| total.exceeded(self.limits.cumulative).map(|b| (b, true)));
if let Some((budget, cumulative)) = exceeded {
return match self.limits.on_limit {
CaptureLimitPolicy::Fail => Err(CaptureError::Limit { budget, cumulative }),
CaptureLimitPolicy::Skip => {
Ok(Some(CaptureSkipReason::Limit { budget, cumulative }))
}
};
}
self.step = step;
self.total = total;
Ok(None)
}
}
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum CaptureError {
#[error("invalid capture plan or tensor: {0}")]
Invalid(String),
#[error("capture operation unsupported: {0}")]
Unsupported(String),
#[error("capture path absent from catalog: {0}")]
MissingPath(String),
#[error("capture size arithmetic overflow")]
Overflow,
#[error("capture {budget:?} limit exceeded (cumulative: {cumulative})")]
Limit {
budget: CaptureBudget,
cumulative: bool,
},
}
pub fn elements(shape: &[u64]) -> Result<u64, CaptureError> {
let nonzero = shape
.iter()
.filter(|dimension| **dimension != 0)
.try_fold(1, |count, dimension| mul(count, *dimension))?;
Ok(if shape.contains(&0) { 0 } else { nonzero })
}
pub fn add(a: u64, b: u64) -> Result<u64, CaptureError> {
a.checked_add(b).ok_or(CaptureError::Overflow)
}
pub fn mul(a: u64, b: u64) -> Result<u64, CaptureError> {
a.checked_mul(b).ok_or(CaptureError::Overflow)
}
pub trait CaptureBackend {
type Tensor;
type Error: std::error::Error + 'static;
fn shape(&self, tensor: &Self::Tensor) -> Result<Vec<u64>, Self::Error>;
fn estimate(
&self,
tensor: &Self::Tensor,
selection: &CaptureSelection,
slice: &ResolvedCaptureSlice,
) -> Result<CaptureUsage, CaptureError>;
fn transform(
&mut self,
tensor: &Self::Tensor,
selection: &CaptureSelection,
slice: &ResolvedCaptureSlice,
) -> Result<CapturePayload, Self::Error>;
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct CapturedStep {
pub phase: CapturePhase,
pub prediction_index: u64,
pub records: Vec<CaptureRecord>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub interventions: Vec<crate::intervention::InterventionRecord>,
pub step_usage: CaptureUsage,
pub cumulative_usage: CaptureUsage,
pub capture_seconds: f64,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct CaptureDiscovery {
pub artifact_identity: String,
pub catalog: ObservationCatalog,
pub support: ObservationSupportReport,
}