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

cortexkit_usage_steps/
row.rs
