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 the event's `attempt_id`, the id the runner gave that
132///   provider attempt and wrote in its own durable step record. Each attempt
133///   actually sent is its own row, because a billed retry is spend and its
134///   cache behaviour differs. The event's `run_id`, `step_id` and `attempt`
135///   are not used: that triple can repeat after a session is restored. For a
136///   runner that is its own durable source (`history_key: "step_ref"`) this
137///   is the identity the provider gives the attempt when it reads the
138///   runner's step log. For a host-keyed runner with no host record, the row
139///   stays live-only.
140///
141/// Returns `None` for an event with neither `harness_step_ref` nor
142/// `attempt_id`. The runner contract refuses to publish such an event, so a
143/// provider that receives one drops it rather than inventing an identity.
144pub fn live_row_identity(
145    runner: &str,
146    harness: &str,
147    harness_step_ref: Option<&str>,
148    attempt_id: Option<&str>,
149) -> Option<(String, String)> {
150    match (harness_step_ref, attempt_id) {
151        (Some(host_record), _) => Some((harness.to_owned(), host_record.to_owned())),
152        (None, Some(attempt_id)) => Some((runner.to_owned(), attempt_id.to_owned())),
153        (None, None) => None,
154    }
155}
156
157/// A JSON value that is not within a published v1 member list.
158#[derive(Clone, Debug, PartialEq, Eq)]
159#[non_exhaustive]
160pub enum NotClosed {
161    /// The value checked is not a JSON object.
162    NotAnObject { object: &'static str },
163    /// The object carries a member outside its published list.
164    UnknownMember {
165        object: &'static str,
166        member: String,
167    },
168}
169
170impl std::fmt::Display for NotClosed {
171    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
172        match self {
173            Self::NotAnObject { object } => write!(f, "{object} is not a JSON object"),
174            Self::UnknownMember { object, member } => {
175                write!(f, "{object} carries unknown member `{member}`")
176            }
177        }
178    }
179}
180
181impl std::error::Error for NotClosed {}
182
183pub(crate) fn check_members(
184    object: &'static str,
185    value: &serde_json::Value,
186    allowed: &[&str],
187) -> Result<(), NotClosed> {
188    let members = value.as_object().ok_or(NotClosed::NotAnObject { object })?;
189    match members.keys().find(|key| !allowed.contains(&key.as_str())) {
190        Some(member) => Err(NotClosed::UnknownMember {
191            object,
192            member: member.clone(),
193        }),
194        None => Ok(()),
195    }
196}
197
198impl StepRow {
199    /// Members every row carries.
200    pub const REQUIRED_MEMBERS: &'static [&'static str] = &[
201        "cursor",
202        "harness",
203        "source",
204        "step_ref",
205        "session",
206        "occurred_at_ms",
207        "prompt",
208        "verdict",
209        "prompt_dropped",
210    ];
211
212    /// Members omitted when unknown.
213    pub const OPTIONAL_MEMBERS: &'static [&'static str] = &[
214        "run_id",
215        "provider",
216        "model",
217        "input",
218        "cache_read",
219        "cache_write",
220        "cache_write_5m",
221        "cache_write_1h",
222        "output",
223        "reasoning",
224        "finish",
225        "context_limit",
226        "request_body_len",
227        "retention",
228        "request_size",
229    ];
230
231    /// Builds a row from its required members; optional members start absent.
232    #[allow(clippy::too_many_arguments)]
233    pub fn new(
234        cursor: impl Into<Cursor>,
235        harness: impl Into<String>,
236        source: StepSource,
237        step_ref: impl Into<String>,
238        session: impl Into<String>,
239        occurred_at_ms: u64,
240        prompt: u64,
241        verdict: Verdict,
242        prompt_dropped: bool,
243    ) -> Self {
244        Self {
245            cursor: cursor.into(),
246            harness: harness.into(),
247            source,
248            step_ref: step_ref.into(),
249            session: session.into(),
250            run_id: None,
251            occurred_at_ms,
252            provider: None,
253            model: None,
254            input: None,
255            cache_read: None,
256            cache_write: None,
257            cache_write_5m: None,
258            cache_write_1h: None,
259            output: None,
260            reasoning: None,
261            finish: None,
262            context_limit: None,
263            request_body_len: None,
264            prompt,
265            retention: None,
266            verdict,
267            prompt_dropped,
268            request_size: None,
269        }
270    }
271
272    /// The row's identity: `(harness, step_ref)`.
273    pub fn identity(&self) -> (&str, &str) {
274        (&self.harness, &self.step_ref)
275    }
276
277    /// Reports a member outside the published v1 row list, by name. For the
278    /// provider's own tests: whatever it emits must stay within the published
279    /// list until this crate updates it. Readers ignore unknown members
280    /// instead, so decoding never calls this.
281    pub fn check_closed(value: &serde_json::Value) -> Result<(), NotClosed> {
282        let allowed: Vec<&str> = Self::REQUIRED_MEMBERS
283            .iter()
284            .chain(Self::OPTIONAL_MEMBERS)
285            .copied()
286            .collect();
287        check_members("row", value, &allowed)
288    }
289
290    /// Reports a row whose `reasoning` exceeds its `output`, naming it by
291    /// identity. `reasoning` is a breakdown inside `output`, so a row with
292    /// both must have `reasoning <= output`. For the provider's own tests,
293    /// like [`StepRow::check_closed`]; readers still decode such a row.
294    pub fn check_counts(&self) -> Result<(), BadCounts> {
295        match (self.reasoning, self.output) {
296            (Some(reasoning), Some(output)) if reasoning > output => {
297                Err(BadCounts::ReasoningExceedsOutput {
298                    harness: self.harness.clone(),
299                    step_ref: self.step_ref.clone(),
300                    reasoning,
301                    output,
302                })
303            }
304            _ => Ok(()),
305        }
306    }
307
308    pub fn with_run_id(mut self, run_id: impl Into<String>) -> Self {
309        self.run_id = Some(run_id.into());
310        self
311    }
312    pub fn with_provider(mut self, provider: impl Into<String>) -> Self {
313        self.provider = Some(provider.into());
314        self
315    }
316    pub fn with_model(mut self, model: impl Into<String>) -> Self {
317        self.model = Some(model.into());
318        self
319    }
320    pub fn with_input(mut self, input: u64) -> Self {
321        self.input = Some(input);
322        self
323    }
324    pub fn with_cache_read(mut self, cache_read: u64) -> Self {
325        self.cache_read = Some(cache_read);
326        self
327    }
328    pub fn with_cache_write(mut self, cache_write: u64) -> Self {
329        self.cache_write = Some(cache_write);
330        self
331    }
332    pub fn with_cache_write_5m(mut self, cache_write_5m: u64) -> Self {
333        self.cache_write_5m = Some(cache_write_5m);
334        self
335    }
336    pub fn with_cache_write_1h(mut self, cache_write_1h: u64) -> Self {
337        self.cache_write_1h = Some(cache_write_1h);
338        self
339    }
340    pub fn with_output(mut self, output: u64) -> Self {
341        self.output = Some(output);
342        self
343    }
344    pub fn with_reasoning(mut self, reasoning: u64) -> Self {
345        self.reasoning = Some(reasoning);
346        self
347    }
348    pub fn with_finish(mut self, finish: impl Into<String>) -> Self {
349        self.finish = Some(finish.into());
350        self
351    }
352    pub fn with_context_limit(mut self, context_limit: u64) -> Self {
353        self.context_limit = Some(context_limit);
354        self
355    }
356    pub fn with_request_body_len(mut self, request_body_len: u64) -> Self {
357        self.request_body_len = Some(request_body_len);
358        self
359    }
360    pub fn with_retention(mut self, retention: f64) -> Self {
361        self.retention = Some(retention);
362        self
363    }
364    pub fn with_request_size(mut self, request_size: RequestSize) -> Self {
365        self.request_size = Some(request_size);
366        self
367    }
368}
cortexkit_usage_steps/row.rs

cortexkit_usage_steps/
row.rs
