cortexkit_usage_steps/
row.rs

1use serde::{Deserialize, Serialize};
2
3use crate::{Cursor, RequestSize, StepSource, Verdict};
4
5/// One step in the feed. Rows carry no content: never prompt text, tool names
6/// or error text.
7///
8/// A row's identity is `harness` plus `step_ref`. When any field of a step
9/// changes, the provider sends the whole row again at a new cursor; the last
10/// row for an identity wins, and a reader replaces, never appends (see
11/// [`crate::LatestRows`]).
12///
13/// Optional members are omitted when unknown and are never zero for unknown.
14/// `input`, `cache_read` and `cache_write` are disjoint. When all are known,
15/// `prompt` equals their sum; summing only present components can give a partial
16/// total, and absence never means zero. The usage-feed provider subtracts cache
17/// counts still included in a source's prompt or input count; already-normalized
18/// runner counts pass through unchanged. It counts aliases for the same writes
19/// once.
20/// Price the three fields separately; never subtract one from another.
21/// `output` includes reasoning: `reasoning` is a breakdown of `output`, never
22/// an addition to it, so `reasoning <= output` whenever both are present (see
23/// [`StepRow::check_counts`]). `cache_write` is the sum of every cache
24/// lifetime. Unknown members are
25/// ignored when decoding; [`StepRow::check_closed`] names them for the
26/// provider's own tests.
27#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
28#[non_exhaustive]
29pub struct StepRow {
30    /// This row's feed position.
31    pub cursor: Cursor,
32    /// The producer: a coding harness or a runner's module id. For a step a
33    /// runner announced, see [`live_row_identity`].
34    pub harness: String,
35    pub source: StepSource,
36    /// The producer's own id for the step. For a step a runner announced, see
37    /// [`live_row_identity`].
38    pub step_ref: String,
39    pub session: String,
40    /// Runner steps only.
41    #[serde(default, skip_serializing_if = "Option::is_none")]
42    pub run_id: Option<String>,
43    /// Unix milliseconds. Steps without a time are not in the feed.
44    pub occurred_at_ms: u64,
45    #[serde(default, skip_serializing_if = "Option::is_none")]
46    pub provider: Option<String>,
47    #[serde(default, skip_serializing_if = "Option::is_none")]
48    pub model: Option<String>,
49    /// Input tokens neither read from nor written to the prompt cache.
50    #[serde(default, skip_serializing_if = "Option::is_none")]
51    pub input: Option<u64>,
52    #[serde(default, skip_serializing_if = "Option::is_none")]
53    pub cache_read: Option<u64>,
54    /// The sum of every cache lifetime.
55    #[serde(default, skip_serializing_if = "Option::is_none")]
56    pub cache_write: Option<u64>,
57    /// Present only when the source split cache writes by lifetime.
58    #[serde(default, skip_serializing_if = "Option::is_none")]
59    pub cache_write_5m: Option<u64>,
60    /// Present only when the source split cache writes by lifetime.
61    #[serde(default, skip_serializing_if = "Option::is_none")]
62    pub cache_write_1h: Option<u64>,
63    /// Output tokens, reasoning included. Charge it once; never add
64    /// `reasoning` on top.
65    #[serde(default, skip_serializing_if = "Option::is_none")]
66    pub output: Option<u64>,
67    /// The part of `output` the provider reports as reasoning: a breakdown,
68    /// never an addition. Absent means not reported, not zero.
69    #[serde(default, skip_serializing_if = "Option::is_none")]
70    pub reasoning: Option<u64>,
71    /// The producer's finish or stop reason, as a classification string.
72    #[serde(default, skip_serializing_if = "Option::is_none")]
73    pub finish: Option<String>,
74    #[serde(default, skip_serializing_if = "Option::is_none")]
75    pub context_limit: Option<u64>,
76    /// Request body length in bytes.
77    #[serde(default, skip_serializing_if = "Option::is_none")]
78    pub request_body_len: Option<u64>,
79    /// Input plus cache read plus cache write.
80    pub prompt: u64,
81    /// This step's cache read divided by the expected cache read: a
82    /// non-negative number, never rounded or clamped, so it exceeds 1 when a
83    /// step reads more than expected. Absent when there was nothing to
84    /// compare with.
85    #[serde(default, skip_serializing_if = "Option::is_none")]
86    pub retention: Option<f64>,
87    pub verdict: Verdict,
88    pub prompt_dropped: bool,
89    /// On a prompt drop, present when both compared steps report body lengths.
90    #[serde(default, skip_serializing_if = "Option::is_none")]
91    pub request_size: Option<RequestSize>,
92}
93
94/// A row whose counts break a contract rule. Readers still decode such a row;
95/// this is for the provider's own tests.
96#[derive(Clone, Debug, PartialEq, Eq)]
97#[non_exhaustive]
98pub enum BadCounts {
99    /// `reasoning` is a breakdown of `output`, so it can never exceed it.
100    ReasoningExceedsOutput {
101        harness: String,
102        step_ref: String,
103        reasoning: u64,
104        output: u64,
105    },
106}
107
108impl std::fmt::Display for BadCounts {
109    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
110        match self {
111            Self::ReasoningExceedsOutput {
112                harness,
113                step_ref,
114                reasoning,
115                output,
116            } => write!(
117                f,
118                "row {harness} {step_ref}: reasoning {reasoning} exceeds output {output}"
119            ),
120        }
121    }
122}
123
124impl std::error::Error for BadCounts {}
125
126/// The identity `(harness, step_ref)` of the row the provider serves for a step
127/// a runner announced on the bus (`step.finished.v1`), from that event's
128/// members. Plain strings, so it needs nothing from the runner contract crate.
129///
130/// The live row must carry the identity of the row the provider later records
131/// from a durable read, or last-row-wins could not replace it:
132///
133/// - With a `harness_step_ref` (the runner declares `history_key: "harness"`
134///   and the host's own store is the durable source): `harness` is the
135///   event's `harness`, the hosting program, and `step_ref` is the host's
136///   record id.
137/// - Without one: `harness` is the event's `runner`, the runner's module id,
138///   and `step_ref` is the event's `attempt_id`, the id the runner gave that
139///   provider attempt and wrote in its own durable step record. Each attempt
140///   actually sent is its own row, because a billed retry is spend and its
141///   cache behaviour differs. The event's `run_id`, `step_id` and `attempt`
142///   are not used: that triple can repeat after a session is restored. For a
143///   runner that is its own durable source (`history_key: "step_ref"`) this
144///   is the identity the provider gives the attempt when it reads the
145///   runner's step log. For a host-keyed runner with no host record, the row
146///   stays live-only.
147///
148/// Returns `None` for an event with neither `harness_step_ref` nor
149/// `attempt_id`. The runner contract refuses to publish such an event, so a
150/// provider that receives one drops it rather than inventing an identity.
151pub fn live_row_identity(
152    runner: &str,
153    harness: &str,
154    harness_step_ref: Option<&str>,
155    attempt_id: Option<&str>,
156) -> Option<(String, String)> {
157    match (harness_step_ref, attempt_id) {
158        (Some(host_record), _) => Some((harness.to_owned(), host_record.to_owned())),
159        (None, Some(attempt_id)) => Some((runner.to_owned(), attempt_id.to_owned())),
160        (None, None) => None,
161    }
162}
163
164/// A JSON value that is not within a published v1 member list.
165#[derive(Clone, Debug, PartialEq, Eq)]
166#[non_exhaustive]
167pub enum NotClosed {
168    /// The value checked is not a JSON object.
169    NotAnObject { object: &'static str },
170    /// The object carries a member outside its published list.
171    UnknownMember {
172        object: &'static str,
173        member: String,
174    },
175}
176
177impl std::fmt::Display for NotClosed {
178    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
179        match self {
180            Self::NotAnObject { object } => write!(f, "{object} is not a JSON object"),
181            Self::UnknownMember { object, member } => {
182                write!(f, "{object} carries unknown member `{member}`")
183            }
184        }
185    }
186}
187
188impl std::error::Error for NotClosed {}
189
190pub(crate) fn check_members(
191    object: &'static str,
192    value: &serde_json::Value,
193    allowed: &[&str],
194) -> Result<(), NotClosed> {
195    let members = value.as_object().ok_or(NotClosed::NotAnObject { object })?;
196    match members.keys().find(|key| !allowed.contains(&key.as_str())) {
197        Some(member) => Err(NotClosed::UnknownMember {
198            object,
199            member: member.clone(),
200        }),
201        None => Ok(()),
202    }
203}
204
205impl StepRow {
206    /// Members every row carries.
207    pub const REQUIRED_MEMBERS: &'static [&'static str] = &[
208        "cursor",
209        "harness",
210        "source",
211        "step_ref",
212        "session",
213        "occurred_at_ms",
214        "prompt",
215        "verdict",
216        "prompt_dropped",
217    ];
218
219    /// Members omitted when unknown.
220    pub const OPTIONAL_MEMBERS: &'static [&'static str] = &[
221        "run_id",
222        "provider",
223        "model",
224        "input",
225        "cache_read",
226        "cache_write",
227        "cache_write_5m",
228        "cache_write_1h",
229        "output",
230        "reasoning",
231        "finish",
232        "context_limit",
233        "request_body_len",
234        "retention",
235        "request_size",
236    ];
237
238    /// Builds a row from its required members; optional members start absent.
239    #[allow(clippy::too_many_arguments)]
240    pub fn new(
241        cursor: impl Into<Cursor>,
242        harness: impl Into<String>,
243        source: StepSource,
244        step_ref: impl Into<String>,
245        session: impl Into<String>,
246        occurred_at_ms: u64,
247        prompt: u64,
248        verdict: Verdict,
249        prompt_dropped: bool,
250    ) -> Self {
251        Self {
252            cursor: cursor.into(),
253            harness: harness.into(),
254            source,
255            step_ref: step_ref.into(),
256            session: session.into(),
257            run_id: None,
258            occurred_at_ms,
259            provider: None,
260            model: None,
261            input: None,
262            cache_read: None,
263            cache_write: None,
264            cache_write_5m: None,
265            cache_write_1h: None,
266            output: None,
267            reasoning: None,
268            finish: None,
269            context_limit: None,
270            request_body_len: None,
271            prompt,
272            retention: None,
273            verdict,
274            prompt_dropped,
275            request_size: None,
276        }
277    }
278
279    /// The row's identity: `(harness, step_ref)`.
280    pub fn identity(&self) -> (&str, &str) {
281        (&self.harness, &self.step_ref)
282    }
283
284    /// Reports a member outside the published v1 row list, by name. For the
285    /// provider's own tests: whatever it emits must stay within the published
286    /// list until this crate updates it. Readers ignore unknown members
287    /// instead, so decoding never calls this.
288    pub fn check_closed(value: &serde_json::Value) -> Result<(), NotClosed> {
289        let allowed: Vec<&str> = Self::REQUIRED_MEMBERS
290            .iter()
291            .chain(Self::OPTIONAL_MEMBERS)
292            .copied()
293            .collect();
294        check_members("row", value, &allowed)
295    }
296
297    /// Reports a row whose `reasoning` exceeds its `output`, naming it by
298    /// identity. `reasoning` is a breakdown inside `output`, so a row with
299    /// both must have `reasoning <= output`. For the provider's own tests,
300    /// like [`StepRow::check_closed`]; readers still decode such a row.
301    pub fn check_counts(&self) -> Result<(), BadCounts> {
302        match (self.reasoning, self.output) {
303            (Some(reasoning), Some(output)) if reasoning > output => {
304                Err(BadCounts::ReasoningExceedsOutput {
305                    harness: self.harness.clone(),
306                    step_ref: self.step_ref.clone(),
307                    reasoning,
308                    output,
309                })
310            }
311            _ => Ok(()),
312        }
313    }
314
315    pub fn with_run_id(mut self, run_id: impl Into<String>) -> Self {
316        self.run_id = Some(run_id.into());
317        self
318    }
319    pub fn with_provider(mut self, provider: impl Into<String>) -> Self {
320        self.provider = Some(provider.into());
321        self
322    }
323    pub fn with_model(mut self, model: impl Into<String>) -> Self {
324        self.model = Some(model.into());
325        self
326    }
327    pub fn with_input(mut self, input: u64) -> Self {
328        self.input = Some(input);
329        self
330    }
331    pub fn with_cache_read(mut self, cache_read: u64) -> Self {
332        self.cache_read = Some(cache_read);
333        self
334    }
335    pub fn with_cache_write(mut self, cache_write: u64) -> Self {
336        self.cache_write = Some(cache_write);
337        self
338    }
339    pub fn with_cache_write_5m(mut self, cache_write_5m: u64) -> Self {
340        self.cache_write_5m = Some(cache_write_5m);
341        self
342    }
343    pub fn with_cache_write_1h(mut self, cache_write_1h: u64) -> Self {
344        self.cache_write_1h = Some(cache_write_1h);
345        self
346    }
347    pub fn with_output(mut self, output: u64) -> Self {
348        self.output = Some(output);
349        self
350    }
351    pub fn with_reasoning(mut self, reasoning: u64) -> Self {
352        self.reasoning = Some(reasoning);
353        self
354    }
355    pub fn with_finish(mut self, finish: impl Into<String>) -> Self {
356        self.finish = Some(finish.into());
357        self
358    }
359    pub fn with_context_limit(mut self, context_limit: u64) -> Self {
360        self.context_limit = Some(context_limit);
361        self
362    }
363    pub fn with_request_body_len(mut self, request_body_len: u64) -> Self {
364        self.request_body_len = Some(request_body_len);
365        self
366    }
367    pub fn with_retention(mut self, retention: f64) -> Self {
368        self.retention = Some(retention);
369        self
370    }
371    pub fn with_request_size(mut self, request_size: RequestSize) -> Self {
372        self.request_size = Some(request_size);
373        self
374    }
375}
cortexkit_usage_steps/row.rs

cortexkit_usage_steps/
row.rs
