diff --git a/Cargo.lock b/Cargo.lock index 49169f0789..694dbb80b6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6873,6 +6873,16 @@ dependencies = [ "uuid", ] +[[package]] +name = "maka-eval" +version = "0.2.0" +dependencies = [ + "hex", + "serde", + "serde_json", + "sha2 0.11.1", +] + [[package]] name = "maka-event-log" version = "0.2.0" diff --git a/Cargo.toml b/Cargo.toml index 42d150ce5a..8023006b8e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -34,6 +34,7 @@ members = [ "plugins/workspace/workspace-files/native", "plugins/workspace/support", "products/maka/acp", + "products/eval", "plugins/agent/engine", "plugins/agent/native", "plugins/workspace/workspace-files/patch", @@ -149,6 +150,7 @@ imagesize = { version = "=0.15.0", default-features = false, features = ["png", jsonschema = { version = "0.60.1", default-features = false } libc = "0.2.191" maka-acp = { path = "products/maka/acp" } +maka-eval = { path = "products/eval" } maka-agent-engine = { path = "plugins/agent/engine" } maka-agent-native = { path = "plugins/agent/native" } maka-assistant = { path = "plugins/agent/assistant" } diff --git a/products/eval/Cargo.toml b/products/eval/Cargo.toml new file mode 100644 index 0000000000..ca6c29a94a --- /dev/null +++ b/products/eval/Cargo.toml @@ -0,0 +1,29 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +[package] +name = "maka-eval" +version.workspace = true +edition.workspace = true +license.workspace = true +publish.workspace = true + +[dependencies] +hex.workspace = true +serde.workspace = true +serde_json.workspace = true +sha2.workspace = true diff --git a/products/eval/README.md b/products/eval/README.md new file mode 100644 index 0000000000..523965d18a --- /dev/null +++ b/products/eval/README.md @@ -0,0 +1,41 @@ + + +# maka-eval + +[简体中文](README.zh-CN.md) + +Measurement and control plane contracts for multi-arm benchmark experiments (`maka.eval.v1` and `maka.eval.v2`). + +## Architecture + +- [`contract`](src/contract.rs): Strict `maka.eval.v1` spec validation and cell expansion, immutable `V2AttemptRecord` with 3-gate causal adjudication (`BlockedIndeterminate` -> `RetryableInfra` -> `Selected`), leaf `v2 -> v1` projection, and spec-only `SpecFingerprint`. +- [`measurement`](src/measurement.rs): Exact 6-counter token accounting (`inputTokens` normalized inclusive of cache reads/writes; `totalTokens` preserved without summation) and `PricingRates` (`DEEPSEEK_V4_FLASH_COST`) as the sole USD pricing authority. + +## Adjudicated `v2 -> v1` Rules + +| Scenario | `v2` Disposition | `v1` Projection & Selection | +| --- | --- | --- | +| Timeout + reward + `LowerBound` | `Selected(SubjectFailed)` | `subject_failed`, raw `score`, `costUsd: null`; selected & never resampled | +| Audit truncated / policy error | `Selected(Completed)` | `completed` with `egress-audit` metadata (`bytes`, `sha256`, `malformedLineCount`) | +| Missing audit / critical artifacts | `RetryableInfra` (unless Gate 1 indeterminate) | `infra_failed`, `score: null` (stricter than legacy TS when score existed) | +| Late cancellation | Before verify: `BlockedIndeterminate`; After: `Selected` | `indeterminate` (`score: null`) vs preserved score + marker | +| Post-hoc evidence recovery | `PostHocRecovery` appended monotonically | Derived view updates status/usage/cost without mutating the base attempt | + +[Source](src/lib.rs) · [Cargo manifest](Cargo.toml) diff --git a/products/eval/README.zh-CN.md b/products/eval/README.zh-CN.md new file mode 100644 index 0000000000..4d05620c44 --- /dev/null +++ b/products/eval/README.zh-CN.md @@ -0,0 +1,41 @@ + + +# maka-eval + +[English](README.md) + +多臂基准评测实验的测量与控制平面契约底座（支持 `maka.eval.v1` 与 `maka.eval.v2`）。 + +## 模块结构 + +- [`contract`](src/contract.rs)：严格的 `maka.eval.v1` 实验规格校验与单元展开、不可变 `V2AttemptRecord` 及三关因果裁定（`BlockedIndeterminate` -> `RetryableInfra` -> `Selected`）、末端只读 `v2 -> v1` 投影，以及仅覆盖声明规格的 `SpecFingerprint`。 +- [`measurement`](src/measurement.rs)：精确的 6 计数器 Token 核算（`inputTokens` 归一化后包含读写缓存；`totalTokens` 原样保留不重算求和），并以 `PricingRates`（`DEEPSEEK_V4_FLASH_COST`）作为唯一 USD 计价权威。 + +## 已裁定 `v2 -> v1` 语义规则 + +| 场景 | `v2` 裁定 | `v1` 投影与选样 | +| --- | --- | --- | +| 超时 + 有 reward + `LowerBound` | `Selected(SubjectFailed)` | `subject_failed`、保留 `score`、`costUsd: null`；直接入选且不重跑 | +| 审计日志截断 / 策略错误 | `Selected(Completed)` | `completed` 并输出完整 `egress-audit` 产物（含 `bytes`、`sha256`、`malformedLineCount`） | +| 缺失审计日志 / 关键产物 | `RetryableInfra`（若已被第一关判未决则保持未决） | `infra_failed`、`score: null`（比旧版 TS 保留分数更严格） | +| 迟到取消 | 验证前 `BlockedIndeterminate`；验证后 `Selected` | `indeterminate`（`score: null`）vs 保留分数并附加标记 | +| 事后证据恢复 | 通过 `PostHocRecovery` 单调叠加 | 派生视图更新状态与计量，不原地改写原始 attempt | + +[源码](src/lib.rs) · [Cargo 配置](Cargo.toml) diff --git a/products/eval/src/contract.rs b/products/eval/src/contract.rs new file mode 100644 index 0000000000..d6a434e23e --- /dev/null +++ b/products/eval/src/contract.rs @@ -0,0 +1,1611 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +//! Experiment specifications, immutable `v2` attempt records with 3-gate +//! causal adjudication, strict `v1` leaf projections, and spec fingerprinting. + +use crate::measurement::{NormalizedUsage, PricingRates, decode_usage}; +use serde::{Deserialize, Deserializer, Serialize, Serializer}; +use serde_json::{Map, Value}; +use sha2::{Digest, Sha256}; +use std::collections::BTreeSet; +use std::fmt; +use std::num::NonZeroU32; + +pub type JsonObject = Map; + +pub const EGRESS_AUDIT_ARTIFACT_PATH: &str = "artifacts/egress-hits.jsonl"; + +const MAX_SAFE_INTEGER_F64: f64 = 9_007_199_254_740_991.0; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ContractError(String); + +impl ContractError { + pub fn new(message: impl Into) -> Self { + Self(message.into()) + } + + pub fn as_str(&self) -> &str { + &self.0 + } +} + +impl fmt::Display for ContractError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(&self.0) + } +} + +impl std::error::Error for ContractError {} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +pub enum SchemaVersion { + #[serde(rename = "maka.eval.v1")] + V1, + #[serde(rename = "maka.eval.v2")] + V2, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum SubjectKind { + Maka, + External, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct BenchmarkSpec { + pub id: String, + pub version: String, + pub config: JsonObject, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct ExecutorSpec { + pub kind: String, + pub config: JsonObject, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct ExecutionSpec { + pub max_concurrent_task_groups: NonZeroU32, +} + +impl Default for ExecutionSpec { + fn default() -> Self { + Self { + max_concurrent_task_groups: NonZeroU32::MIN, + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct SubjectSpec { + pub id: String, + pub kind: SubjectKind, + pub credentials: Vec, + pub config: JsonObject, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct TaskSpec { + pub id: String, + pub input: String, + pub config: JsonObject, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct ExperimentSpec { + pub schema_version: SchemaVersion, + pub id: String, + pub benchmark: BenchmarkSpec, + pub executor: ExecutorSpec, + #[serde(default)] + pub execution: ExecutionSpec, + pub subjects: Vec, + pub tasks: Vec, + pub repetitions: NonZeroU32, + pub budget: JsonObject, + pub verifier: JsonObject, +} + +impl<'de> Deserialize<'de> for ExperimentSpec { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + let value = Value::deserialize(deserializer)?; + parse_experiment_spec(&value).map_err(serde::de::Error::custom) + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct ExperimentCell { + pub id: String, + pub experiment_id: String, + pub benchmark: BenchmarkSpec, + pub executor: ExecutorSpec, + pub subject: SubjectSpec, + pub task: TaskSpec, + pub repetition: NonZeroU32, + pub budget: JsonObject, + pub verifier: JsonObject, +} + +pub fn expand_experiment(spec: &ExperimentSpec) -> Vec { + let mut cells = Vec::with_capacity( + spec.tasks.len() * (spec.repetitions.get() as usize) * spec.subjects.len(), + ); + for task in &spec.tasks { + for r in 1..=spec.repetitions.get() { + let repetition = NonZeroU32::new(r).expect("range starts at 1"); + for subject in &spec.subjects { + cells.push(ExperimentCell { + id: format!("{}::{r}::{}", task.id, subject.id), + experiment_id: spec.id.clone(), + benchmark: spec.benchmark.clone(), + executor: spec.executor.clone(), + subject: subject.clone(), + task: task.clone(), + repetition, + budget: spec.budget.clone(), + verifier: spec.verifier.clone(), + }); + } + } + } + cells +} + +fn parse_experiment_spec(value: &Value) -> Result { + let root = exact_object( + value, + "experiment", + &[ + "schemaVersion", + "id", + "benchmark", + "executor", + "subjects", + "tasks", + "repetitions", + "budget", + "verifier", + ], + &["execution"], + )?; + + let schema_version = match root.get("schemaVersion").and_then(Value::as_str) { + Some("maka.eval.v1") => SchemaVersion::V1, + _ => { + return Err(ContractError::new("unsupported experiment schema")); + } + }; + + let benchmark_obj = exact_object( + root.get("benchmark").unwrap(), + "benchmark", + &["id", "version", "config"], + &[], + )?; + let executor_obj = exact_object( + root.get("executor").unwrap(), + "executor", + &["kind", "config"], + &[], + )?; + + let max_concurrent_task_groups = match root.get("execution") { + None => NonZeroU32::MIN, + Some(exec_val) => { + let exec_obj = exact_object(exec_val, "execution", &["maxConcurrentTaskGroups"], &[])?; + positive_nonzero_u32( + exec_obj.get("maxConcurrentTaskGroups").unwrap(), + "execution.maxConcurrentTaskGroups", + )? + } + }; + + let subjects_arr = root + .get("subjects") + .and_then(Value::as_array) + .ok_or_else(|| ContractError::new("subjects must be an array"))?; + let mut subjects = Vec::with_capacity(subjects_arr.len()); + for (idx, item) in subjects_arr.iter().enumerate() { + let where_label = format!("subjects[{idx}]"); + let sub_obj = exact_object( + item, + &where_label, + &["id", "kind", "credentials", "config"], + &[], + )?; + let kind = match sub_obj.get("kind").and_then(Value::as_str) { + Some("maka") => SubjectKind::Maka, + Some("external") => SubjectKind::External, + _ => { + return Err(ContractError::new(format!("{where_label}.kind is invalid"))); + } + }; + subjects.push(SubjectSpec { + id: identifier(sub_obj.get("id").unwrap(), &format!("{where_label}.id"))?, + kind, + credentials: unique_strings( + sub_obj.get("credentials").unwrap(), + &format!("{where_label}.credentials"), + )?, + config: json_object( + sub_obj.get("config").unwrap(), + &format!("{where_label}.config"), + )?, + }); + } + + let tasks_arr = root + .get("tasks") + .and_then(Value::as_array) + .ok_or_else(|| ContractError::new("tasks must be an array"))?; + let mut tasks = Vec::with_capacity(tasks_arr.len()); + for (idx, item) in tasks_arr.iter().enumerate() { + let where_label = format!("tasks[{idx}]"); + let task_obj = exact_object(item, &where_label, &["id", "input", "config"], &[])?; + tasks.push(TaskSpec { + id: identifier(task_obj.get("id").unwrap(), &format!("{where_label}.id"))?, + input: nonempty_string( + task_obj.get("input").unwrap(), + &format!("{where_label}.input"), + )?, + config: json_object( + task_obj.get("config").unwrap(), + &format!("{where_label}.config"), + )?, + }); + } + + if subjects.is_empty() || tasks.is_empty() { + return Err(ContractError::new("experiment arms are empty")); + } + ensure_unique_ids(subjects.iter().map(|s| s.id.as_str()), "subject")?; + ensure_unique_ids(tasks.iter().map(|t| t.id.as_str()), "task")?; + + let repetitions = positive_nonzero_u32(root.get("repetitions").unwrap(), "repetitions")?; + + Ok(ExperimentSpec { + schema_version, + id: identifier(root.get("id").unwrap(), "experiment.id")?, + benchmark: BenchmarkSpec { + id: identifier(benchmark_obj.get("id").unwrap(), "benchmark.id")?, + version: nonempty_string(benchmark_obj.get("version").unwrap(), "benchmark.version")?, + config: json_object(benchmark_obj.get("config").unwrap(), "benchmark.config")?, + }, + executor: ExecutorSpec { + kind: identifier(executor_obj.get("kind").unwrap(), "executor.kind")?, + config: json_object(executor_obj.get("config").unwrap(), "executor.config")?, + }, + execution: ExecutionSpec { + max_concurrent_task_groups, + }, + subjects, + tasks, + repetitions, + budget: json_object(root.get("budget").unwrap(), "budget")?, + verifier: json_object(root.get("verifier").unwrap(), "verifier")?, + }) +} + +fn canonical_sha256_hex(value: &Value) -> String { + let mut buf = Vec::with_capacity(256); + write_canonical_json(value, &mut buf); + let mut hasher = Sha256::new(); + hasher.update(&buf); + hex::encode(hasher.finalize()) +} + +fn write_canonical_json(value: &Value, out: &mut Vec) { + match value { + Value::Null | Value::Bool(_) | Value::String(_) => { + serde_json::to_writer(&mut *out, value) + .expect("canonical JSON value serialization cannot fail"); + } + Value::Number(n) => { + if n.is_u64() || n.is_i64() { + serde_json::to_writer(&mut *out, n) + .expect("canonical integer serialization cannot fail"); + } else if let Some(f) = n.as_f64() { + if f == 0.0 { + out.extend_from_slice(b"0"); + } else if f.is_finite() + && f.fract() == 0.0 + && (-MAX_SAFE_INTEGER_F64..=MAX_SAFE_INTEGER_F64).contains(&f) + { + if f >= 0.0 { + serde_json::to_writer(&mut *out, &(f as u64)) + .expect("canonical float->u64 serialization cannot fail"); + } else { + serde_json::to_writer(&mut *out, &(f as i64)) + .expect("canonical float->i64 serialization cannot fail"); + } + } else { + serde_json::to_writer(&mut *out, n) + .expect("canonical float serialization cannot fail"); + } + } else { + serde_json::to_writer(&mut *out, n) + .expect("canonical number serialization cannot fail"); + } + } + Value::Array(items) => { + out.push(b'['); + for (idx, item) in items.iter().enumerate() { + if idx > 0 { + out.push(b','); + } + write_canonical_json(item, out); + } + out.push(b']'); + } + Value::Object(map) => { + // Explicit key sorting is required because workspace feature unification + // enables serde_json/preserve_order. + let mut sorted: Vec<(&String, &Value)> = map.iter().collect(); + sorted.sort_by_key(|(k, _)| *k); + out.push(b'{'); + for (idx, (k, v)) in sorted.into_iter().enumerate() { + if idx > 0 { + out.push(b','); + } + serde_json::to_writer(&mut *out, k) + .expect("canonical object key serialization cannot fail"); + out.push(b':'); + write_canonical_json(v, out); + } + out.push(b'}'); + } + } +} + +/// Canonical digests of declared `ExperimentSpec` semantic and execution parameters. +/// +/// Fingerprints cover only the declared experiment specification; they do not capture +/// host/runtime binaries, prompt or tool schemas, container image digests, or product parity. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct SpecFingerprint { + /// Digest over schema version, benchmark, executor kind, subjects, tasks, repetitions, budget, and verifier. + pub semantic_sha256: String, + /// Digest over execution-only spec parameters (e.g. `maxConcurrentTaskGroups`, `executor.config`). + pub execution_sha256: String, +} + +impl SpecFingerprint { + pub fn compute(spec: &ExperimentSpec) -> Self { + let semantic_payload = serde_json::json!({ + "schemaVersion": spec.schema_version, + "benchmark": spec.benchmark, + "executorKind": spec.executor.kind, + "subjects": spec.subjects, + "tasks": spec.tasks, + "repetitions": spec.repetitions, + "budget": spec.budget, + "verifier": spec.verifier, + }); + let execution_payload = serde_json::json!({ + "maxConcurrentTaskGroups": spec.execution.max_concurrent_task_groups, + "executorConfig": spec.executor.config, + }); + Self { + semantic_sha256: canonical_sha256_hex(&semantic_payload), + execution_sha256: canonical_sha256_hex(&execution_payload), + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum EvalResultStatus { + Completed, + SubjectFailed, + InfraFailed, + Indeterminate, +} + +impl EvalResultStatus { + pub fn is_selectable(self) -> bool { + matches!(self, Self::Completed | Self::SubjectFailed) + } +} + +#[derive(Debug, Clone, PartialEq, Serialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct EvalResult { + pub score: Option, + pub usage: Option, + pub cost_usd: Option, + pub duration_ms: u64, + pub status: EvalResultStatus, + pub failure_reason: Option, + pub artifacts: Vec, +} + +impl<'de> Deserialize<'de> for EvalResult { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + let value = Value::deserialize(deserializer)?; + decode_eval_result(&value, "result").map_err(serde::de::Error::custom) + } +} + +#[derive(Debug, Clone, PartialEq, Serialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct V1CellAttempt { + pub cell_id: String, + pub sequence: NonZeroU32, + pub started_at: u64, + pub completed_at: u64, + pub result: EvalResult, +} + +impl<'de> Deserialize<'de> for V1CellAttempt { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + let value = Value::deserialize(deserializer)?; + decode_v1_cell_attempt(&value).map_err(serde::de::Error::custom) + } +} + +/// Legacy `v1` cell result selector for frozen oracle compatibility: picks the earliest +/// attempt by sequence whose status is `completed` or `subject_failed`. +pub fn select_cell_result(attempts: &[V1CellAttempt]) -> Option<&V1CellAttempt> { + attempts + .iter() + .filter(|a| a.result.status.is_selectable()) + .min_by_key(|a| a.sequence) +} + +fn decode_eval_result(value: &Value, where_label: &str) -> Result { + let obj = exact_object( + value, + where_label, + &[ + "score", + "usage", + "costUsd", + "durationMs", + "status", + "failureReason", + "artifacts", + ], + &[], + )?; + let status = match obj.get("status").and_then(Value::as_str) { + Some("completed") => EvalResultStatus::Completed, + Some("subject_failed") => EvalResultStatus::SubjectFailed, + Some("infra_failed") => EvalResultStatus::InfraFailed, + Some("indeterminate") => EvalResultStatus::Indeterminate, + _ => { + return Err(ContractError::new(format!( + "{where_label}.status is invalid" + ))); + } + }; + let failure_reason = match obj.get("failureReason").unwrap() { + Value::Null => None, + Value::String(s) => Some(s.clone()), + _ => { + return Err(ContractError::new(format!( + "{where_label}.failureReason is invalid" + ))); + } + }; + let score = nullable_f64(obj.get("score").unwrap(), &format!("{where_label}.score"))?; + let usage = match obj.get("usage").unwrap() { + Value::Null => None, + u => Some(decode_usage(u, &format!("{where_label}.usage"))?), + }; + let cost_usd = nullable_nonnegative_f64( + obj.get("costUsd").unwrap(), + &format!("{where_label}.costUsd"), + )?; + let duration_ms = safe_u64( + obj.get("durationMs").unwrap(), + &format!("{where_label}.durationMs"), + )?; + let artifacts_arr = obj + .get("artifacts") + .and_then(Value::as_array) + .ok_or_else(|| ContractError::new(format!("{where_label}.artifacts is invalid")))?; + let mut artifacts = Vec::with_capacity(artifacts_arr.len()); + for (idx, item) in artifacts_arr.iter().enumerate() { + artifacts.push(json_object( + item, + &format!("{where_label}.artifacts[{idx}]"), + )?); + } + Ok(EvalResult { + score, + usage, + cost_usd, + duration_ms, + status, + failure_reason, + artifacts, + }) +} + +fn decode_v1_cell_attempt(value: &Value) -> Result { + let obj = exact_object( + value, + "attempt", + &["cellId", "sequence", "startedAt", "completedAt", "result"], + &[], + )?; + Ok(V1CellAttempt { + cell_id: nonempty_string(obj.get("cellId").unwrap(), "attempt.cellId")?, + sequence: positive_nonzero_u32(obj.get("sequence").unwrap(), "attempt.sequence")?, + started_at: safe_u64(obj.get("startedAt").unwrap(), "attempt.startedAt")?, + completed_at: safe_u64(obj.get("completedAt").unwrap(), "attempt.completedAt")?, + result: decode_eval_result(obj.get("result").unwrap(), "attempt.result")?, + }) +} + +/// Monotonic phase durations (in milliseconds) recorded on native `v2` attempts. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct PhaseDurationsMs { + pub queue_wait_ms: u64, + pub prepare_ms: u64, + pub subject_ms: u64, + pub verify_ms: u64, + pub cleanup_ms: u64, + pub total_ms: u64, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] +pub enum SubjectOutcome { + Completed {}, + Failed { + #[serde(default, skip_serializing_if = "Option::is_none")] + reason: Option, + }, + Timeout { + #[serde(default, skip_serializing_if = "Option::is_none")] + reason: Option, + }, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] +pub enum VerifierOutcome { + Verified { + score: f64, + }, + SubjectFailed { + #[serde(default, skip_serializing_if = "Option::is_none")] + score: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + reason: Option, + }, + /// Missing reward file or empty verifier payload: never defaults to `0.0`. + MissingReward { + reason: String, + }, + InfraFailed { + reason: String, + }, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde( + tag = "stage", + rename_all = "snake_case", + rename_all_fields = "camelCase", + deny_unknown_fields +)] +pub enum TrialOutcome { + InfraFailed { + reason: String, + }, + Indeterminate { + reason: String, + }, + CancelledBeforeVerification { + #[serde(default, skip_serializing_if = "Option::is_none")] + reason: Option, + }, + Evaluated { + subject: SubjectOutcome, + verifier: VerifierOutcome, + #[serde(default)] + cancelled_after_verification: bool, + }, +} + +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)] +#[serde( + tag = "state", + rename_all = "snake_case", + rename_all_fields = "camelCase" +)] +pub enum AttemptMetering { + #[default] + Unknown, + LowerBound { + usage: NormalizedUsage, + }, + Complete { + usage: NormalizedUsage, + #[serde(default, skip_serializing_if = "Option::is_none")] + model: Option, + }, +} + +impl<'de> Deserialize<'de> for AttemptMetering { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + let value = Value::deserialize(deserializer)?; + let state = value + .as_object() + .and_then(|obj| obj.get("state")) + .and_then(Value::as_str) + .ok_or_else(|| serde::de::Error::custom("metering.state is required"))?; + let metering = match state { + "unknown" => { + exact_object(&value, "metering", &["state"], &[]) + .map_err(serde::de::Error::custom)?; + Self::Unknown + } + "lower_bound" => { + let obj = exact_object(&value, "metering", &["state", "usage"], &[]) + .map_err(serde::de::Error::custom)?; + Self::LowerBound { + usage: decode_usage(obj.get("usage").unwrap(), "metering.usage") + .map_err(serde::de::Error::custom)?, + } + } + "complete" => { + let obj = exact_object(&value, "metering", &["state", "usage"], &["model"]) + .map_err(serde::de::Error::custom)?; + let model = match obj.get("model") { + None | Some(Value::Null) => None, + Some(m) => Some( + nonempty_string(m, "metering.model").map_err(serde::de::Error::custom)?, + ), + }; + Self::Complete { + usage: decode_usage(obj.get("usage").unwrap(), "metering.usage") + .map_err(serde::de::Error::custom)?, + model, + } + } + _ => return Err(serde::de::Error::custom("metering.state is invalid")), + }; + metering.validate().map_err(serde::de::Error::custom)?; + Ok(metering) + } +} + +impl AttemptMetering { + pub fn usage(&self) -> Option<&NormalizedUsage> { + match self { + Self::Unknown => None, + Self::LowerBound { usage } | Self::Complete { usage, .. } => Some(usage), + } + } + + /// Standardized USD cost derived strictly from `PricingRates::for_model` on `Complete` metering. + pub fn cost_usd(&self) -> Option { + match self { + Self::Complete { + usage, + model: Some(model), + } => PricingRates::for_model(model).map(|rates| rates.cost_usd(usage)), + Self::Unknown | Self::LowerBound { .. } | Self::Complete { model: None, .. } => None, + } + } + + pub fn validate(&self) -> Result<(), ContractError> { + if let Self::Complete { + model: Some(model), .. + } = self + && model.is_empty() + { + return Err(ContractError::new("metering.model must not be empty")); + } + Ok(()) + } + + fn rank(&self) -> u8 { + match self { + Self::Unknown => 0, + Self::LowerBound { .. } => 1, + Self::Complete { .. } => 2, + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] +pub enum CleanupState { + Confirmed {}, + Unconfirmed { reason: String }, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde( + tag = "kind", + rename_all = "snake_case", + rename_all_fields = "camelCase", + deny_unknown_fields +)] +pub enum EgressAuditState { + NotRequired {}, + Recorded { + bytes: u64, + sha256: String, + #[serde(default)] + truncated: bool, + #[serde(default)] + policy_error_count: u64, + #[serde(default)] + malformed_line_count: u64, + }, + Missing {}, + Unreadable { + reason: String, + }, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] +pub enum CriticalArtifactsState { + Present {}, + Missing { paths: Vec }, +} + +#[derive(Debug, Clone, PartialEq)] +pub struct V2AttemptParts { + pub cell_id: String, + pub sequence: NonZeroU32, + pub started_at: u64, + pub completed_at: u64, + pub durations: PhaseDurationsMs, + pub trial_outcome: TrialOutcome, + pub metering: AttemptMetering, + pub cleanup_state: CleanupState, + pub audit_state: EgressAuditState, + pub critical_artifacts_state: CriticalArtifactsState, + pub artifacts: Vec, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +struct V2AttemptWire { + schema_version: SchemaVersion, + cell_id: String, + sequence: NonZeroU32, + started_at: u64, + completed_at: u64, + durations: PhaseDurationsMs, + trial_outcome: TrialOutcome, + metering: AttemptMetering, + cleanup_state: CleanupState, + audit_state: EgressAuditState, + critical_artifacts_state: CriticalArtifactsState, + artifacts: Vec, +} + +/// Immutable, validated `maka.eval.v2` attempt record. +#[derive(Debug, Clone, PartialEq)] +pub struct V2AttemptRecord(V2AttemptParts); + +impl V2AttemptRecord { + pub fn try_new(parts: V2AttemptParts) -> Result { + if parts.cell_id.is_empty() { + return Err(ContractError::new("v2 attempt record cellId is required")); + } + if parts.completed_at < parts.started_at { + return Err(ContractError::new( + "v2 attempt record completedAt must not precede startedAt", + )); + } + validate_trial_outcome(&parts.trial_outcome)?; + parts.metering.validate()?; + validate_cleanup_state(&parts.cleanup_state)?; + validate_audit_state(&parts.audit_state)?; + validate_critical_artifacts_state(&parts.critical_artifacts_state)?; + Ok(Self(parts)) + } + + pub fn cell_id(&self) -> &str { + &self.0.cell_id + } + + pub fn sequence(&self) -> NonZeroU32 { + self.0.sequence + } + + pub fn started_at(&self) -> u64 { + self.0.started_at + } + + pub fn completed_at(&self) -> u64 { + self.0.completed_at + } + + pub fn durations(&self) -> &PhaseDurationsMs { + &self.0.durations + } + + pub fn trial_outcome(&self) -> &TrialOutcome { + &self.0.trial_outcome + } + + pub fn metering(&self) -> &AttemptMetering { + &self.0.metering + } + + pub fn cleanup_state(&self) -> &CleanupState { + &self.0.cleanup_state + } + + pub fn audit_state(&self) -> &EgressAuditState { + &self.0.audit_state + } + + pub fn critical_artifacts_state(&self) -> &CriticalArtifactsState { + &self.0.critical_artifacts_state + } + + pub fn artifacts(&self) -> &[JsonObject] { + &self.0.artifacts + } + + pub fn into_parts(self) -> V2AttemptParts { + self.0 + } + + pub fn disposition(&self) -> V2Disposition { + self.disposition_with_recovery(&[]) + } + + pub fn disposition_with_recovery(&self, recoveries: &[PostHocRecovery]) -> V2Disposition { + let (effective_cleanup, effective_metering) = self.effective_evidence(recoveries); + self.adjudicate(effective_cleanup, effective_metering) + } + + /// Projects this `maka.eval.v2` record into a strict `maka.eval.v1` `V1CellAttempt`. + pub fn project_v1(&self) -> V1CellAttempt { + self.project_v1_with_recovery(&[]) + } + + /// Projects this `maka.eval.v2` record plus appended post-hoc recovery entries + /// into a strict `maka.eval.v1` `V1CellAttempt` without mutating the record. + pub fn project_v1_with_recovery(&self, recoveries: &[PostHocRecovery]) -> V1CellAttempt { + let (effective_cleanup, effective_metering) = self.effective_evidence(recoveries); + let disposition = self.adjudicate(effective_cleanup, effective_metering); + + let mut artifacts = self.0.artifacts.clone(); + if matches!( + self.0.trial_outcome, + TrialOutcome::Evaluated { + cancelled_after_verification: true, + .. + } + ) { + push_artifact_marker(&mut artifacts, "cancelled-after-verification"); + } + if !recoveries.is_empty() { + push_artifact_marker(&mut artifacts, "post-hoc-evidence-recovery"); + } + append_egress_audit_artifact(&mut artifacts, &self.0.audit_state); + + let (status, score, failure_reason) = match disposition { + V2Disposition::Selected { + status: SelectedStatus::Completed, + score, + failure_reason, + } => (EvalResultStatus::Completed, score, failure_reason), + V2Disposition::Selected { + status: SelectedStatus::SubjectFailed, + score, + failure_reason, + } => (EvalResultStatus::SubjectFailed, score, failure_reason), + V2Disposition::RetryableInfra { failure_reason } => { + (EvalResultStatus::InfraFailed, None, Some(failure_reason)) + } + V2Disposition::BlockedIndeterminate { failure_reason } => { + (EvalResultStatus::Indeterminate, None, Some(failure_reason)) + } + }; + + V1CellAttempt { + cell_id: self.0.cell_id.clone(), + sequence: self.0.sequence, + started_at: self.0.started_at, + completed_at: self.0.completed_at, + result: EvalResult { + score, + usage: effective_metering.usage().cloned(), + cost_usd: effective_metering.cost_usd(), + duration_ms: self.0.durations.subject_ms, + status, + failure_reason, + artifacts, + }, + } + } + + fn effective_evidence<'a>( + &'a self, + recoveries: &'a [PostHocRecovery], + ) -> (&'a CleanupState, &'a AttemptMetering) { + let mut cleanup = &self.0.cleanup_state; + let mut metering = &self.0.metering; + for recovery in recoveries { + if let Some(next_cleanup) = recovery.cleanup_state() + && matches!(cleanup, CleanupState::Unconfirmed { .. }) + && matches!(next_cleanup, CleanupState::Confirmed {}) + { + cleanup = next_cleanup; + } + if let Some(next_metering) = recovery.metering() + && next_metering.rank() > metering.rank() + { + metering = next_metering; + } + } + (cleanup, metering) + } + + fn adjudicate( + &self, + effective_cleanup: &CleanupState, + effective_metering: &AttemptMetering, + ) -> V2Disposition { + // Gate 1: Epistemic certainty — unresolved cleanup, pre-verification cancellation, + // indeterminate execution, or unknown metering on an evaluated trial block retry. + if let CleanupState::Unconfirmed { reason } = effective_cleanup { + return V2Disposition::BlockedIndeterminate { + failure_reason: reason.clone(), + }; + } + match &self.0.trial_outcome { + TrialOutcome::CancelledBeforeVerification { reason } => { + return V2Disposition::BlockedIndeterminate { + failure_reason: reason.clone().unwrap_or_else(|| { + "executor cancelled before verification completed".into() + }), + }; + } + TrialOutcome::Indeterminate { reason } => { + return V2Disposition::BlockedIndeterminate { + failure_reason: reason.clone(), + }; + } + TrialOutcome::Evaluated { .. } + if matches!(effective_metering, AttemptMetering::Unknown) => + { + return V2Disposition::BlockedIndeterminate { + failure_reason: "metering state unknown".into(), + }; + } + TrialOutcome::InfraFailed { .. } | TrialOutcome::Evaluated { .. } => {} + } + + // Gate 2: Infrastructure & evidence integrity — preserve root infra failures before + // checking downstream artifacts/audit, and reject LowerBound metering on Completed runs. + match &self.0.trial_outcome { + TrialOutcome::InfraFailed { reason } => { + return V2Disposition::RetryableInfra { + failure_reason: reason.clone(), + }; + } + TrialOutcome::Evaluated { + verifier: + VerifierOutcome::InfraFailed { reason } | VerifierOutcome::MissingReward { reason }, + .. + } => { + return V2Disposition::RetryableInfra { + failure_reason: reason.clone(), + }; + } + TrialOutcome::Evaluated { .. } => {} + TrialOutcome::Indeterminate { .. } + | TrialOutcome::CancelledBeforeVerification { .. } => { + unreachable!("handled in Gate 1"); + } + } + if let CriticalArtifactsState::Missing { paths } = &self.0.critical_artifacts_state { + return V2Disposition::RetryableInfra { + failure_reason: format!("critical artifacts missing: {}", paths.join(", ")), + }; + } + match &self.0.audit_state { + EgressAuditState::Missing {} => { + return V2Disposition::RetryableInfra { + failure_reason: "egress audit log missing".into(), + }; + } + EgressAuditState::Unreadable { reason } => { + return V2Disposition::RetryableInfra { + failure_reason: reason.clone(), + }; + } + EgressAuditState::NotRequired {} | EgressAuditState::Recorded { .. } => {} + } + if matches!( + self.0.trial_outcome, + TrialOutcome::Evaluated { + subject: SubjectOutcome::Completed {}, + .. + } + ) && matches!(effective_metering, AttemptMetering::LowerBound { .. }) + { + return V2Disposition::RetryableInfra { + failure_reason: "incomplete lower-bound metering on completed subject".into(), + }; + } + + // Gate 3: Capability settlement — permanent selection, never resampled. + let TrialOutcome::Evaluated { + subject, verifier, .. + } = &self.0.trial_outcome + else { + unreachable!("non-evaluated outcomes handled in Gates 1 and 2"); + }; + match (subject, verifier) { + (SubjectOutcome::Completed {}, VerifierOutcome::Verified { score }) => { + V2Disposition::Selected { + status: SelectedStatus::Completed, + score: Some(*score), + failure_reason: None, + } + } + (SubjectOutcome::Completed {}, VerifierOutcome::SubjectFailed { score, reason }) => { + V2Disposition::Selected { + status: SelectedStatus::SubjectFailed, + score: *score, + failure_reason: reason.clone(), + } + } + (SubjectOutcome::Timeout { reason }, VerifierOutcome::Verified { score }) => { + V2Disposition::Selected { + status: SelectedStatus::SubjectFailed, + score: Some(*score), + failure_reason: Some( + reason + .clone() + .unwrap_or_else(|| "subject exceeded the framework timeout".into()), + ), + } + } + ( + SubjectOutcome::Timeout { reason }, + VerifierOutcome::SubjectFailed { + score, + reason: v_reason, + }, + ) => V2Disposition::Selected { + status: SelectedStatus::SubjectFailed, + score: *score, + failure_reason: Some( + reason + .clone() + .or_else(|| v_reason.clone()) + .unwrap_or_else(|| "subject exceeded the framework timeout".into()), + ), + }, + (SubjectOutcome::Failed { reason }, VerifierOutcome::Verified { score }) => { + V2Disposition::Selected { + status: SelectedStatus::SubjectFailed, + score: Some(*score), + failure_reason: reason.clone(), + } + } + ( + SubjectOutcome::Failed { reason }, + VerifierOutcome::SubjectFailed { + score, + reason: v_reason, + }, + ) => V2Disposition::Selected { + status: SelectedStatus::SubjectFailed, + score: *score, + failure_reason: reason.clone().or_else(|| v_reason.clone()), + }, + (_, VerifierOutcome::MissingReward { .. } | VerifierOutcome::InfraFailed { .. }) => { + unreachable!("handled in Gate 2") + } + } + } +} + +impl Serialize for V2AttemptRecord { + fn serialize(&self, serializer: S) -> Result { + let wire = V2AttemptWire { + schema_version: SchemaVersion::V2, + cell_id: self.0.cell_id.clone(), + sequence: self.0.sequence, + started_at: self.0.started_at, + completed_at: self.0.completed_at, + durations: self.0.durations, + trial_outcome: self.0.trial_outcome.clone(), + metering: self.0.metering.clone(), + cleanup_state: self.0.cleanup_state.clone(), + audit_state: self.0.audit_state.clone(), + critical_artifacts_state: self.0.critical_artifacts_state.clone(), + artifacts: self.0.artifacts.clone(), + }; + wire.serialize(serializer) + } +} + +impl<'de> Deserialize<'de> for V2AttemptRecord { + fn deserialize>(deserializer: D) -> Result { + let wire = V2AttemptWire::deserialize(deserializer)?; + if wire.schema_version != SchemaVersion::V2 { + return Err(serde::de::Error::custom( + "v2 attempt record requires schemaVersion maka.eval.v2", + )); + } + Self::try_new(V2AttemptParts { + cell_id: wire.cell_id, + sequence: wire.sequence, + started_at: wire.started_at, + completed_at: wire.completed_at, + durations: wire.durations, + trial_outcome: wire.trial_outcome, + metering: wire.metering, + cleanup_state: wire.cleanup_state, + audit_state: wire.audit_state, + critical_artifacts_state: wire.critical_artifacts_state, + artifacts: wire.artifacts, + }) + .map_err(serde::de::Error::custom) + } +} + +/// Append-only post-hoc evidence recovery event applied on top of an immutable `V2AttemptRecord`. +#[derive(Debug, Clone, PartialEq)] +pub struct PostHocRecovery { + recovered_at: u64, + metering: Option, + cleanup_state: Option, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +struct PostHocRecoveryWire { + recovered_at: u64, + #[serde(default, skip_serializing_if = "Option::is_none")] + metering: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + cleanup_state: Option, +} + +impl PostHocRecovery { + pub fn try_new( + recovered_at: u64, + metering: Option, + cleanup_state: Option, + ) -> Result { + if metering.is_none() && cleanup_state.is_none() { + return Err(ContractError::new( + "post-hoc recovery must provide metering or cleanupState", + )); + } + if let Some(m) = &metering { + m.validate()?; + } + if let Some(c) = &cleanup_state { + validate_cleanup_state(c)?; + } + Ok(Self { + recovered_at, + metering, + cleanup_state, + }) + } + + pub fn recovered_at(&self) -> u64 { + self.recovered_at + } + + pub fn metering(&self) -> Option<&AttemptMetering> { + self.metering.as_ref() + } + + pub fn cleanup_state(&self) -> Option<&CleanupState> { + self.cleanup_state.as_ref() + } +} + +impl Serialize for PostHocRecovery { + fn serialize(&self, serializer: S) -> Result { + PostHocRecoveryWire { + recovered_at: self.recovered_at, + metering: self.metering.clone(), + cleanup_state: self.cleanup_state.clone(), + } + .serialize(serializer) + } +} + +impl<'de> Deserialize<'de> for PostHocRecovery { + fn deserialize>(deserializer: D) -> Result { + let wire = PostHocRecoveryWire::deserialize(deserializer)?; + Self::try_new(wire.recovered_at, wire.metering, wire.cleanup_state) + .map_err(serde::de::Error::custom) + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum SelectedStatus { + Completed, + SubjectFailed, +} + +#[derive(Debug, Clone, PartialEq)] +pub enum V2Disposition { + Selected { + status: SelectedStatus, + score: Option, + failure_reason: Option, + }, + RetryableInfra { + failure_reason: String, + }, + BlockedIndeterminate { + failure_reason: String, + }, +} + +/// Authoritative `v2` cell result selector: picks the earliest attempt by sequence +/// whose `V2Disposition` is `Selected`. +pub fn select_v2_cell_result(attempts: &[V2AttemptRecord]) -> Option<&V2AttemptRecord> { + attempts + .iter() + .filter(|a| matches!(a.disposition(), V2Disposition::Selected { .. })) + .min_by_key(|a| a.sequence()) +} + +/// Authoritative `v2` retry admission check: returns `false` if any attempt is already +/// `Selected` or `BlockedIndeterminate`, or if `attempts.len() > max_infra_retries as usize`. +pub fn can_admit_v2_attempt(attempts: &[V2AttemptRecord], max_infra_retries: u32) -> bool { + !attempts.iter().any(|a| { + matches!( + a.disposition(), + V2Disposition::Selected { .. } | V2Disposition::BlockedIndeterminate { .. } + ) + }) && attempts.len() <= max_infra_retries as usize +} + +fn validate_trial_outcome(outcome: &TrialOutcome) -> Result<(), ContractError> { + match outcome { + TrialOutcome::InfraFailed { reason } | TrialOutcome::Indeterminate { reason } => { + if reason.is_empty() { + return Err(ContractError::new("trialOutcome.reason must not be empty")); + } + } + TrialOutcome::CancelledBeforeVerification { reason: Some(r) } => { + if r.is_empty() { + return Err(ContractError::new("trialOutcome.reason must not be empty")); + } + } + TrialOutcome::CancelledBeforeVerification { reason: None } => {} + TrialOutcome::Evaluated { + subject, verifier, .. + } => { + match subject { + SubjectOutcome::Completed {} => {} + SubjectOutcome::Failed { reason: Some(r) } + | SubjectOutcome::Timeout { reason: Some(r) } => { + if r.is_empty() { + return Err(ContractError::new("subject.reason must not be empty")); + } + } + SubjectOutcome::Failed { reason: None } + | SubjectOutcome::Timeout { reason: None } => {} + } + match verifier { + VerifierOutcome::Verified { score } => { + if !score.is_finite() { + return Err(ContractError::new("verifier score must be finite")); + } + } + VerifierOutcome::SubjectFailed { score, reason } => { + if score.is_some_and(|s| !s.is_finite()) { + return Err(ContractError::new("verifier score must be finite")); + } + if reason.as_deref().is_some_and(str::is_empty) { + return Err(ContractError::new("verifier.reason must not be empty")); + } + } + VerifierOutcome::MissingReward { reason } + | VerifierOutcome::InfraFailed { reason } => { + if reason.is_empty() { + return Err(ContractError::new("verifier.reason must not be empty")); + } + } + } + } + } + Ok(()) +} + +fn validate_cleanup_state(state: &CleanupState) -> Result<(), ContractError> { + if let CleanupState::Unconfirmed { reason } = state + && reason.is_empty() + { + return Err(ContractError::new("cleanupState.reason must not be empty")); + } + Ok(()) +} + +fn validate_audit_state(state: &EgressAuditState) -> Result<(), ContractError> { + match state { + EgressAuditState::NotRequired {} | EgressAuditState::Missing {} => {} + EgressAuditState::Recorded { sha256, .. } => { + if !is_lower_hex_sha256(sha256) { + return Err(ContractError::new( + "auditState.sha256 must be 64 lowercase hex characters", + )); + } + } + EgressAuditState::Unreadable { reason } => { + if reason.is_empty() { + return Err(ContractError::new("auditState.reason must not be empty")); + } + } + } + Ok(()) +} + +fn validate_critical_artifacts_state(state: &CriticalArtifactsState) -> Result<(), ContractError> { + if let CriticalArtifactsState::Missing { paths } = state + && (paths.is_empty() || paths.iter().any(String::is_empty)) + { + return Err(ContractError::new( + "criticalArtifactsState.paths must be non-empty strings", + )); + } + Ok(()) +} + +fn push_artifact_marker(artifacts: &mut Vec, kind: &str) { + if !artifacts + .iter() + .any(|a| a.get("kind").and_then(Value::as_str) == Some(kind)) + { + let mut marker = Map::new(); + marker.insert("kind".into(), Value::String(kind.into())); + artifacts.push(marker); + } +} + +fn append_egress_audit_artifact(artifacts: &mut Vec, audit_state: &EgressAuditState) { + match audit_state { + EgressAuditState::NotRequired {} => {} + EgressAuditState::Recorded { + bytes, + sha256, + truncated, + policy_error_count, + malformed_line_count, + } => { + if !artifacts + .iter() + .any(|a| a.get("kind").and_then(Value::as_str) == Some("egress-audit")) + { + let mut audit_obj = Map::new(); + audit_obj.insert("kind".into(), Value::String("egress-audit".into())); + audit_obj.insert( + "path".into(), + Value::String(EGRESS_AUDIT_ARTIFACT_PATH.into()), + ); + audit_obj.insert("bytes".into(), Value::from(*bytes)); + audit_obj.insert("sha256".into(), Value::String(format!("sha256:{sha256}"))); + audit_obj.insert("truncated".into(), Value::Bool(*truncated)); + audit_obj.insert("policyErrorCount".into(), Value::from(*policy_error_count)); + audit_obj.insert( + "malformedLineCount".into(), + Value::from(*malformed_line_count), + ); + artifacts.push(audit_obj); + } + } + EgressAuditState::Missing {} => { + push_path_artifact( + artifacts, + "egress-audit-missing", + EGRESS_AUDIT_ARTIFACT_PATH, + ); + } + EgressAuditState::Unreadable { .. } => { + push_path_artifact( + artifacts, + "egress-audit-unreadable", + EGRESS_AUDIT_ARTIFACT_PATH, + ); + } + } +} + +fn push_path_artifact(artifacts: &mut Vec, kind: &str, path: &str) { + if !artifacts + .iter() + .any(|a| a.get("kind").and_then(Value::as_str) == Some(kind)) + { + let mut obj = Map::new(); + obj.insert("kind".into(), Value::String(kind.into())); + obj.insert("path".into(), Value::String(path.into())); + artifacts.push(obj); + } +} + +pub(crate) fn exact_object<'a>( + value: &'a Value, + where_label: &str, + required: &[&str], + optional: &[&str], +) -> Result<&'a JsonObject, ContractError> { + let obj = value + .as_object() + .ok_or_else(|| ContractError::new(format!("{where_label} must be an object")))?; + for key in obj.keys() { + if !required.contains(&key.as_str()) && !optional.contains(&key.as_str()) { + return Err(ContractError::new(format!( + "{where_label} has unsupported fields" + ))); + } + } + for req in required { + if !obj.contains_key(*req) { + return Err(ContractError::new(format!( + "{where_label}.{req} is required" + ))); + } + } + Ok(obj) +} + +fn json_object(value: &Value, where_label: &str) -> Result { + value + .as_object() + .cloned() + .ok_or_else(|| ContractError::new(format!("{where_label} must be an object"))) +} + +fn identifier(value: &Value, where_label: &str) -> Result { + let text = nonempty_string(value, where_label)?; + let mut chars = text.chars(); + let Some(first) = chars.next() else { + return Err(ContractError::new(format!("{where_label} is required"))); + }; + if !first.is_ascii_alphanumeric() + || !chars.all(|c| c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | '-')) + { + return Err(ContractError::new(format!("{where_label} is invalid"))); + } + Ok(text) +} + +fn nonempty_string(value: &Value, where_label: &str) -> Result { + match value.as_str() { + Some(s) if !s.is_empty() => Ok(s.to_owned()), + _ => Err(ContractError::new(format!("{where_label} is required"))), + } +} + +fn unique_strings(value: &Value, where_label: &str) -> Result, ContractError> { + let arr = value + .as_array() + .ok_or_else(|| ContractError::new(format!("{where_label} must be an array")))?; + let mut seen = BTreeSet::new(); + let mut out = Vec::with_capacity(arr.len()); + for item in arr { + let s = nonempty_string(item, where_label)?; + if !seen.insert(s.clone()) { + return Err(ContractError::new(format!( + "{where_label} contains duplicates" + ))); + } + out.push(s); + } + Ok(out) +} + +fn ensure_unique_ids<'a>( + ids: impl Iterator, + label: &str, +) -> Result<(), ContractError> { + let mut seen = BTreeSet::new(); + for id in ids { + if !seen.insert(id) { + return Err(ContractError::new(format!("duplicate {label} id"))); + } + } + Ok(()) +} + +fn is_lower_hex_sha256(s: &str) -> bool { + s.len() == 64 + && s.bytes() + .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b)) +} + +pub(crate) fn safe_u64(value: &Value, where_label: &str) -> Result { + if let Some(n) = value.as_u64() { + return Ok(n); + } + match value.as_f64() { + Some(f) + if f.is_finite() && (0.0..=MAX_SAFE_INTEGER_F64).contains(&f) && f.fract() == 0.0 => + { + Ok(f as u64) + } + _ => Err(ContractError::new(format!("{where_label} is invalid"))), + } +} + +fn positive_nonzero_u32(value: &Value, where_label: &str) -> Result { + let n = safe_u64(value, where_label)?; + let raw = u32::try_from(n) + .ok() + .and_then(NonZeroU32::new) + .ok_or_else(|| ContractError::new(format!("{where_label} must be positive")))?; + Ok(raw) +} + +fn nullable_f64(value: &Value, where_label: &str) -> Result, ContractError> { + if value.is_null() { + return Ok(None); + } + match value.as_f64() { + Some(f) if f.is_finite() => Ok(Some(f)), + _ => Err(ContractError::new(format!("{where_label} is invalid"))), + } +} + +fn nullable_nonnegative_f64( + value: &Value, + where_label: &str, +) -> Result, ContractError> { + if value.is_null() { + return Ok(None); + } + match value.as_f64() { + Some(f) if f.is_finite() && f >= 0.0 => Ok(Some(f)), + _ => Err(ContractError::new(format!("{where_label} is invalid"))), + } +} diff --git a/products/eval/src/lib.rs b/products/eval/src/lib.rs new file mode 100644 index 0000000000..237bb1fa2d --- /dev/null +++ b/products/eval/src/lib.rs @@ -0,0 +1,23 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +//! Measurement and control plane for multi-arm Maka evaluation experiments. + +pub mod contract; +pub mod measurement; diff --git a/products/eval/src/measurement.rs b/products/eval/src/measurement.rs new file mode 100644 index 0000000000..7790dea1bd --- /dev/null +++ b/products/eval/src/measurement.rs @@ -0,0 +1,159 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +//! Exact 6-counter token usage preservation and standardized model pricing. + +use crate::contract::{ContractError, exact_object, safe_u64}; +use serde::{Deserialize, Deserializer, Serialize}; +use serde_json::Value; + +pub const COST_USD_TOLERANCE: f64 = 1e-9; + +/// Exact 6-counter normalized token usage. +/// +/// `input_tokens` is normalized to total input tokens **inclusive** of `cache_read_tokens` +/// and `cache_write_tokens`; `total_tokens` is preserved as reported and is never +/// synthesized or overwritten by summing the other five counters. +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct NormalizedUsage { + pub input_tokens: u64, + pub output_tokens: u64, + pub cache_read_tokens: u64, + pub cache_write_tokens: u64, + pub reasoning_tokens: u64, + pub total_tokens: u64, +} + +impl<'de> Deserialize<'de> for NormalizedUsage { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + let value = Value::deserialize(deserializer)?; + decode_usage(&value, "usage").map_err(serde::de::Error::custom) + } +} + +pub(crate) fn decode_usage( + value: &Value, + where_label: &str, +) -> Result { + let obj = exact_object( + value, + where_label, + &[ + "inputTokens", + "outputTokens", + "cacheReadTokens", + "cacheWriteTokens", + "reasoningTokens", + "totalTokens", + ], + &[], + )?; + Ok(NormalizedUsage { + input_tokens: safe_u64( + obj.get("inputTokens").unwrap(), + &format!("{where_label}.inputTokens"), + )?, + output_tokens: safe_u64( + obj.get("outputTokens").unwrap(), + &format!("{where_label}.outputTokens"), + )?, + cache_read_tokens: safe_u64( + obj.get("cacheReadTokens").unwrap(), + &format!("{where_label}.cacheReadTokens"), + )?, + cache_write_tokens: safe_u64( + obj.get("cacheWriteTokens").unwrap(), + &format!("{where_label}.cacheWriteTokens"), + )?, + reasoning_tokens: safe_u64( + obj.get("reasoningTokens").unwrap(), + &format!("{where_label}.reasoningTokens"), + )?, + total_tokens: safe_u64( + obj.get("totalTokens").unwrap(), + &format!("{where_label}.totalTokens"), + )?, + }) +} + +impl NormalizedUsage { + /// Accumulates another `NormalizedUsage` record counter-by-counter without + /// synthesizing `total_tokens` from the component fields. + pub fn accumulate_exact(&mut self, other: &Self) { + self.input_tokens = self.input_tokens.saturating_add(other.input_tokens); + self.output_tokens = self.output_tokens.saturating_add(other.output_tokens); + self.cache_read_tokens = self + .cache_read_tokens + .saturating_add(other.cache_read_tokens); + self.cache_write_tokens = self + .cache_write_tokens + .saturating_add(other.cache_write_tokens); + self.reasoning_tokens = self.reasoning_tokens.saturating_add(other.reasoning_tokens); + self.total_tokens = self.total_tokens.saturating_add(other.total_tokens); + } +} + +#[derive(Debug, Clone, Copy, PartialEq)] +pub struct PricingRates { + pub input_per_million: f64, + pub output_per_million: f64, + pub cache_read_per_million: f64, + pub cache_write_per_million: f64, +} + +impl PricingRates { + pub fn for_model(model: &str) -> Option { + match model { + "deepseek-v4-flash" => Some(DEEPSEEK_V4_FLASH_COST), + _ => None, + } + } + + pub fn cost_usd(&self, usage: &NormalizedUsage) -> f64 { + let cached = usage + .cache_read_tokens + .saturating_add(usage.cache_write_tokens); + let uncached = usage.input_tokens.saturating_sub(cached); + ((uncached as f64) * self.input_per_million + + (usage.cache_read_tokens as f64) * self.cache_read_per_million + + (usage.cache_write_tokens as f64) * self.cache_write_per_million + + (usage.output_tokens as f64) * self.output_per_million) + / 1_000_000.0 + } +} + +/// Published DeepSeek V4 Flash prices in USD per million tokens (off-peak band). +pub const DEEPSEEK_V4_FLASH_COST: PricingRates = PricingRates { + input_per_million: 0.22, + output_per_million: 0.66, + cache_read_per_million: 0.007, + cache_write_per_million: 0.22, +}; + +pub fn costs_match_within_tolerance(left: Option, right: Option) -> bool { + match (left, right) { + (None, None) => true, + (Some(a), Some(b)) => (a - b).abs() <= COST_USD_TOLERANCE, + _ => false, + } +} diff --git a/products/eval/tests/contract.rs b/products/eval/tests/contract.rs new file mode 100644 index 0000000000..6206c2bd72 --- /dev/null +++ b/products/eval/tests/contract.rs @@ -0,0 +1,731 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +use maka_eval::contract::{ + AttemptMetering, CleanupState, CriticalArtifactsState, EGRESS_AUDIT_ARTIFACT_PATH, + EgressAuditState, EvalResultStatus, ExperimentSpec, PhaseDurationsMs, PostHocRecovery, + SelectedStatus, SpecFingerprint, SubjectOutcome, TrialOutcome, V1CellAttempt, V2AttemptParts, + V2AttemptRecord, V2Disposition, VerifierOutcome, can_admit_v2_attempt, expand_experiment, + select_cell_result, select_v2_cell_result, +}; +use maka_eval::measurement::{NormalizedUsage, PricingRates, costs_match_within_tolerance}; +use serde_json::{Map, Value, json}; +use std::num::NonZeroU32; + +const SAMPLE_SHA256: &str = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"; + +fn nz(v: u32) -> NonZeroU32 { + NonZeroU32::new(v).unwrap() +} + +fn sample_usage() -> NormalizedUsage { + NormalizedUsage { + input_tokens: 100, + output_tokens: 20, + cache_read_tokens: 0, + cache_write_tokens: 0, + reasoning_tokens: 0, + total_tokens: 120, + } +} + +fn sample_two_arm_spec_json() -> Value { + json!({ + "schemaVersion": "maka.eval.v1", + "id": "tb21-deepseek-v4-flash-two-arm", + "benchmark": { + "id": "terminal-bench-2.1", + "version": "d49e28f1e4ddd13d289e85a5f312a66750951932", + "config": { + "repository": "https://github.com/harbor-framework/terminal-bench-2-1" + } + }, + "executor": { + "kind": "harbor", + "config": { + "frameworkVersion": "0.20.0", + "pythonPathEnv": "MAKA_EVAL_HARBOR_PYTHON", + "trialsRootEnv": "MAKA_EVAL_TRIALS_DIR", + "environment": { "type": "docker", "delete": true } + } + }, + "execution": { + "maxConcurrentTaskGroups": 2 + }, + "subjects": [ + { + "id": "maka", + "kind": "maka", + "credentials": ["DEEPSEEK_API_KEY"], + "config": { + "model": "deepseek-v4-flash", + "sandboxMode": "workspace-write" + } + }, + { + "id": "external-cli", + "kind": "external", + "credentials": ["DEEPSEEK_API_KEY"], + "config": { + "command": "/opt/maka-node-toolchain/bin/node", + "args": ["harbor-external-subject.js", "{{task.input}}"] + } + } + ], + "tasks": [ + { + "id": "task-alpha", + "input": "Fix alpha", + "config": { "harbor": { "path": "tasks/task-alpha" } } + }, + { + "id": "task-beta", + "input": "Fix beta", + "config": { "harbor": { "path": "tasks/task-beta" } } + } + ], + "repetitions": 1, + "budget": { + "timeoutMultiplier": 1, + "maxSteps": 100 + }, + "verifier": { + "reward": "reward" + } + }) +} + +fn base_v2_parts() -> V2AttemptParts { + V2AttemptParts { + cell_id: "t1::1::maka".into(), + sequence: nz(1), + started_at: 10, + completed_at: 20, + durations: PhaseDurationsMs { + queue_wait_ms: 1, + prepare_ms: 2, + subject_ms: 5, + verify_ms: 1, + cleanup_ms: 1, + total_ms: 10, + }, + trial_outcome: TrialOutcome::Evaluated { + subject: SubjectOutcome::Completed {}, + verifier: VerifierOutcome::Verified { score: 1.0 }, + cancelled_after_verification: false, + }, + metering: AttemptMetering::Complete { + usage: sample_usage(), + model: Some("deepseek-v4-flash".into()), + }, + cleanup_state: CleanupState::Confirmed {}, + audit_state: EgressAuditState::NotRequired {}, + critical_artifacts_state: CriticalArtifactsState::Present {}, + artifacts: Vec::new(), + } +} + +#[test] +fn experiment_spec_validation_and_cell_expansion_match_v1_contract() { + let spec_json = sample_two_arm_spec_json(); + let spec: ExperimentSpec = serde_json::from_value(spec_json.clone()).unwrap(); + let cells = expand_experiment(&spec); + + assert_eq!(cells.len(), 4); + assert_eq!(cells[0].id, "task-alpha::1::maka"); + assert_eq!(cells[1].id, "task-alpha::1::external-cli"); + assert_eq!(cells[2].id, "task-beta::1::maka"); + assert_eq!(cells[3].id, "task-beta::1::external-cli"); + + // Frozen TS safe-integer parity: 1.0 is accepted for integer counts. + let mut float_int_spec = spec_json.clone(); + float_int_spec["repetitions"] = json!(1.0); + float_int_spec["execution"]["maxConcurrentTaskGroups"] = json!(1.0); + let parsed_float_int: ExperimentSpec = serde_json::from_value(float_int_spec).unwrap(); + assert_eq!(parsed_float_int.repetitions, nz(1)); + assert_eq!(parsed_float_int.execution.max_concurrent_task_groups, nz(1)); + + // Reject v2 schemaVersion on ExperimentSpec, unknown fields, 0, fractional, and oversized floats. + let mut v2_spec = spec_json.clone(); + v2_spec["schemaVersion"] = json!("maka.eval.v2"); + assert!(serde_json::from_value::(v2_spec).is_err()); + + let mut unknown = spec_json.clone(); + unknown + .as_object_mut() + .unwrap() + .insert("unknownField".into(), json!(123)); + assert!(serde_json::from_value::(unknown).is_err()); + + for bad_num in [json!(0), json!(1.5), json!(1e30)] { + let mut bad_rep = spec_json.clone(); + bad_rep["repetitions"] = bad_num.clone(); + assert!(serde_json::from_value::(bad_rep).is_err()); + + let mut bad_conc = spec_json.clone(); + bad_conc["execution"]["maxConcurrentTaskGroups"] = bad_num; + assert!(serde_json::from_value::(bad_conc).is_err()); + } + + // Concurrency changes execution fingerprint but preserves semantic fingerprint. + let mut higher_concurrency_json = spec_json.clone(); + higher_concurrency_json["execution"]["maxConcurrentTaskGroups"] = json!(16); + let higher_spec: ExperimentSpec = serde_json::from_value(higher_concurrency_json).unwrap(); + + let fp1 = SpecFingerprint::compute(&spec); + let fp2 = SpecFingerprint::compute(&higher_spec); + assert_eq!(fp1.semantic_sha256, fp2.semantic_sha256); + assert_ne!(fp1.execution_sha256, fp2.execution_sha256); + + // Changing experiment.id, float/integer equivalence (1.0 vs 1), and nested object + // key insertion order preserve semantic_sha256 even under preserve_order. + let mut reordered_json = spec_json; + reordered_json["id"] = json!("tb21-deepseek-v4-flash-rerun"); + let mut reordered_budget = Map::new(); + reordered_budget.insert("maxSteps".into(), json!(100)); + reordered_budget.insert("timeoutMultiplier".into(), json!(1.0)); + reordered_json["budget"] = Value::Object(reordered_budget); + let reordered_spec: ExperimentSpec = serde_json::from_value(reordered_json).unwrap(); + let fp3 = SpecFingerprint::compute(&reordered_spec); + assert_eq!(fp1.semantic_sha256, fp3.semantic_sha256); +} + +#[test] +fn six_counter_usage_and_authoritative_model_pricing() { + let u1: NormalizedUsage = serde_json::from_value(json!({ + "inputTokens": 1000.0, + "outputTokens": 250, + "cacheReadTokens": 400, + "cacheWriteTokens": 100, + "reasoningTokens": 80, + "totalTokens": 9999 + })) + .unwrap(); + let u2 = NormalizedUsage { + input_tokens: 500, + output_tokens: 150, + cache_read_tokens: 200, + cache_write_tokens: 0, + reasoning_tokens: 20, + total_tokens: 7777, + }; + + for invalid in [ + json!({ + "inputTokens": 10, + "outputTokens": 5, + "cacheReadTokens": 0, + "cacheWriteTokens": 0, + "reasoningTokens": 0 + }), + json!({ + "inputTokens": -1, + "outputTokens": 5, + "cacheReadTokens": 0, + "cacheWriteTokens": 0, + "reasoningTokens": 0, + "totalTokens": 4 + }), + json!({ + "inputTokens": 1e30, + "outputTokens": 5, + "cacheReadTokens": 0, + "cacheWriteTokens": 0, + "reasoningTokens": 0, + "totalTokens": 15 + }), + json!({ + "inputTokens": 10, + "outputTokens": 5, + "cacheReadTokens": 0, + "cacheWriteTokens": 0, + "reasoningTokens": 0, + "totalTokens": 15, + "extra": 1 + }), + ] { + assert!(serde_json::from_value::(invalid).is_err()); + } + + let mut combined = u1.clone(); + combined.accumulate_exact(&u2); + assert_eq!(combined.input_tokens, 1500); + assert_eq!(combined.output_tokens, 400); + assert_eq!(combined.cache_read_tokens, 600); + assert_eq!(combined.cache_write_tokens, 100); + assert_eq!(combined.reasoning_tokens, 100); + assert_eq!(combined.total_tokens, 17776); + + // uncached = 1000 - 400 - 100 = 500 + // (500 * 0.22 + 400 * 0.007 + 100 * 0.22 + 250 * 0.66) / 1_000_000 = 0.0002998 + let rates = PricingRates::for_model("deepseek-v4-flash").unwrap(); + let cost = rates.cost_usd(&u1); + assert!(costs_match_within_tolerance(Some(cost), Some(0.0002998))); + assert!(!costs_match_within_tolerance( + Some(cost), + Some(0.0002998 + 1e-8) + )); + assert!(PricingRates::for_model("custom-local").is_none()); +} + +#[test] +fn strict_boundary_decoding_rejects_unknown_fields_and_illegal_states() { + let v2_base = json!({ + "schemaVersion": "maka.eval.v2", + "cellId": "t1::1::maka", + "sequence": 1, + "startedAt": 10, + "completedAt": 20, + "durations": { + "queueWaitMs": 1, + "prepareMs": 2, + "subjectMs": 5, + "verifyMs": 1, + "cleanupMs": 1, + "totalMs": 10 + }, + "trialOutcome": { + "stage": "evaluated", + "subject": { "kind": "completed" }, + "verifier": { "kind": "verified", "score": 1.0 }, + "cancelledAfterVerification": false + }, + "metering": { + "state": "complete", + "usage": { + "inputTokens": 10, + "outputTokens": 5, + "cacheReadTokens": 0, + "cacheWriteTokens": 0, + "reasoningTokens": 0, + "totalTokens": 15 + }, + "model": "deepseek-v4-flash" + }, + "cleanupState": { "kind": "confirmed" }, + "auditState": { "kind": "not_required" }, + "criticalArtifactsState": { "kind": "present" }, + "artifacts": [] + }); + let parsed_v2: V2AttemptRecord = serde_json::from_value(v2_base.clone()).unwrap(); + let encoded_v2 = serde_json::to_value(&parsed_v2).unwrap(); + assert_eq!( + serde_json::from_value::(encoded_v2).unwrap(), + parsed_v2 + ); + + // Every empty-struct tagged variant and metering.unknown must reject extra fields. + let mut bad_subject = v2_base.clone(); + bad_subject["trialOutcome"]["subject"] = json!({ "kind": "completed", "extra": 1 }); + assert!(serde_json::from_value::(bad_subject).is_err()); + + let mut bad_cleanup = v2_base.clone(); + bad_cleanup["cleanupState"] = json!({ "kind": "confirmed", "extra": true }); + assert!(serde_json::from_value::(bad_cleanup).is_err()); + + let mut bad_audit_nr = v2_base.clone(); + bad_audit_nr["auditState"] = json!({ "kind": "not_required", "extra": "x" }); + assert!(serde_json::from_value::(bad_audit_nr).is_err()); + + let mut bad_audit_missing = v2_base.clone(); + bad_audit_missing["auditState"] = json!({ "kind": "missing", "extra": "x" }); + assert!(serde_json::from_value::(bad_audit_missing).is_err()); + + let mut bad_artifacts = v2_base.clone(); + bad_artifacts["criticalArtifactsState"] = json!({ "kind": "present", "extra": 1 }); + assert!(serde_json::from_value::(bad_artifacts).is_err()); + + let mut bad_metering_unknown = v2_base.clone(); + bad_metering_unknown["metering"] = json!({ "state": "unknown", "model": "deepseek-v4-flash" }); + assert!(serde_json::from_value::(bad_metering_unknown).is_err()); + + // Illegal v2 invariants on wire & Rust boundary. + let mut wrong_schema = v2_base.clone(); + wrong_schema["schemaVersion"] = json!("maka.eval.v1"); + assert!(serde_json::from_value::(wrong_schema).is_err()); + + let mut reversed_times = v2_base.clone(); + reversed_times["completedAt"] = json!(5); + assert!(serde_json::from_value::(reversed_times).is_err()); + + let mut empty_missing_paths = v2_base.clone(); + empty_missing_paths["criticalArtifactsState"] = json!({ "kind": "missing", "paths": [] }); + assert!(serde_json::from_value::(empty_missing_paths).is_err()); + + let mut empty_reason = v2_base.clone(); + empty_reason["trialOutcome"] = json!({ "stage": "infra_failed", "reason": "" }); + assert!(serde_json::from_value::(empty_reason).is_err()); + + let mut bad_sha = v2_base; + bad_sha["auditState"] = json!({ "kind": "recorded", "bytes": 12, "sha256": "NOT-HEX" }); + assert!(serde_json::from_value::(bad_sha).is_err()); + + let mut non_finite = base_v2_parts(); + non_finite.trial_outcome = TrialOutcome::Evaluated { + subject: SubjectOutcome::Completed {}, + verifier: VerifierOutcome::Verified { score: f64::NAN }, + cancelled_after_verification: false, + }; + assert!(V2AttemptRecord::try_new(non_finite).is_err()); + + // V1CellAttempt: safe integer parity (1.0 ok, 1e30 & 0 rejected), lossless round-trip. + let valid_v1 = json!({ + "cellId": "t1::1::maka", + "sequence": 1.0, + "startedAt": 10, + "completedAt": 20, + "result": { + "score": null, + "usage": null, + "costUsd": null, + "durationMs": 10, + "status": "subject_failed", + "failureReason": null, + "artifacts": [] + } + }); + let parsed_v1: V1CellAttempt = serde_json::from_value(valid_v1.clone()).unwrap(); + assert_eq!(parsed_v1.sequence, nz(1)); + let encoded_v1 = serde_json::to_value(&parsed_v1).unwrap(); + assert_eq!( + serde_json::from_value::(encoded_v1).unwrap(), + parsed_v1 + ); + + let mut zero_seq = valid_v1.clone(); + zero_seq["sequence"] = json!(0); + assert!(serde_json::from_value::(zero_seq).is_err()); + + let mut saturated_ts = valid_v1; + saturated_ts["startedAt"] = json!(1e30); + assert!(serde_json::from_value::(saturated_ts).is_err()); +} + +#[test] +fn three_gate_v2_funnel_selection_and_v1_projections() { + // Gate 1 dominance: Indeterminate overrides missing critical artifacts and missing audit. + let mut indet_parts = base_v2_parts(); + indet_parts.trial_outcome = TrialOutcome::Indeterminate { + reason: "relay disconnected".into(), + }; + indet_parts.metering = AttemptMetering::Unknown; + indet_parts.critical_artifacts_state = CriticalArtifactsState::Missing { + paths: vec!["transcript.json".into()], + }; + indet_parts.audit_state = EgressAuditState::Missing {}; + let indet_rec = V2AttemptRecord::try_new(indet_parts).unwrap(); + assert_eq!( + indet_rec.disposition(), + V2Disposition::BlockedIndeterminate { + failure_reason: "relay disconnected".into(), + } + ); + assert_eq!( + indet_rec.project_v1().result.status, + EvalResultStatus::Indeterminate + ); + assert!(!can_admit_v2_attempt(&[indet_rec], 3)); + + // Gate 2 root-cause preservation: pre-verification InfraFailed keeps its own reason + // even with Unknown metering and missing artifacts, and admits retries within budget. + let mut infra_parts = base_v2_parts(); + infra_parts.trial_outcome = TrialOutcome::InfraFailed { + reason: "docker pull failed".into(), + }; + infra_parts.metering = AttemptMetering::Unknown; + infra_parts.critical_artifacts_state = CriticalArtifactsState::Missing { + paths: vec!["transcript.json".into()], + }; + let infra_rec = V2AttemptRecord::try_new(infra_parts).unwrap(); + assert_eq!( + infra_rec.disposition(), + V2Disposition::RetryableInfra { + failure_reason: "docker pull failed".into(), + } + ); + assert!(!can_admit_v2_attempt(std::slice::from_ref(&infra_rec), 0)); + assert!(can_admit_v2_attempt(std::slice::from_ref(&infra_rec), 1)); + + // Metering gates on Completed: Unknown -> BlockedIndeterminate; LowerBound -> RetryableInfra. + let mut completed_unknown = base_v2_parts(); + completed_unknown.metering = AttemptMetering::Unknown; + let completed_unknown_rec = V2AttemptRecord::try_new(completed_unknown).unwrap(); + assert!(matches!( + completed_unknown_rec.disposition(), + V2Disposition::BlockedIndeterminate { .. } + )); + + let mut completed_lb = base_v2_parts(); + completed_lb.metering = AttemptMetering::LowerBound { + usage: sample_usage(), + }; + let completed_lb_rec = V2AttemptRecord::try_new(completed_lb).unwrap(); + assert!(matches!( + completed_lb_rec.disposition(), + V2Disposition::RetryableInfra { .. } + )); + + // Scenario 1: Timeout + Verified(0.75) + LowerBound -> Selected(SubjectFailed, Some(0.75)), + // costUsd == None, blocks resampling; Complete derives standardized cost via PricingRates. + let mut timeout_lb_parts = base_v2_parts(); + timeout_lb_parts.trial_outcome = TrialOutcome::Evaluated { + subject: SubjectOutcome::Timeout { + reason: Some("timed out".into()), + }, + verifier: VerifierOutcome::Verified { score: 0.75 }, + cancelled_after_verification: false, + }; + timeout_lb_parts.metering = AttemptMetering::LowerBound { + usage: sample_usage(), + }; + let timeout_lb_rec = V2AttemptRecord::try_new(timeout_lb_parts.clone()).unwrap(); + assert_eq!( + timeout_lb_rec.disposition(), + V2Disposition::Selected { + status: SelectedStatus::SubjectFailed, + score: Some(0.75), + failure_reason: Some("timed out".into()), + } + ); + assert!(!can_admit_v2_attempt( + std::slice::from_ref(&timeout_lb_rec), + 3 + )); + let timeout_lb_v1 = timeout_lb_rec.project_v1(); + assert_eq!(timeout_lb_v1.result.status, EvalResultStatus::SubjectFailed); + assert_eq!(timeout_lb_v1.result.score, Some(0.75)); + assert_eq!(timeout_lb_v1.result.usage, Some(sample_usage())); + assert_eq!(timeout_lb_v1.result.cost_usd, None); + + timeout_lb_parts.metering = AttemptMetering::Complete { + usage: sample_usage(), + model: Some("deepseek-v4-flash".into()), + }; + let timeout_complete_v1 = V2AttemptRecord::try_new(timeout_lb_parts) + .unwrap() + .project_v1(); + assert!(costs_match_within_tolerance( + timeout_complete_v1.result.cost_usd, + Some(0.0000352) + )); + + // Scenario 2: EgressAuditState::Recorded keeps Completed(1.0) and projects full audit fields; + // Missing -> RetryableInfra + egress-audit-missing artifact. + let mut audit_parts = base_v2_parts(); + audit_parts.audit_state = EgressAuditState::Recorded { + bytes: 512, + sha256: SAMPLE_SHA256.into(), + truncated: true, + policy_error_count: 2, + malformed_line_count: 1, + }; + let audit_v1 = V2AttemptRecord::try_new(audit_parts.clone()) + .unwrap() + .project_v1(); + assert_eq!(audit_v1.result.status, EvalResultStatus::Completed); + assert_eq!(audit_v1.result.score, Some(1.0)); + let audit_art = audit_v1 + .result + .artifacts + .iter() + .find(|a| a.get("kind").and_then(Value::as_str) == Some("egress-audit")) + .unwrap(); + assert_eq!( + audit_art.get("path").and_then(Value::as_str), + Some(EGRESS_AUDIT_ARTIFACT_PATH) + ); + assert_eq!(audit_art.get("bytes").and_then(Value::as_u64), Some(512)); + assert_eq!( + audit_art.get("sha256").and_then(Value::as_str), + Some(format!("sha256:{SAMPLE_SHA256}").as_str()) + ); + assert_eq!( + audit_art.get("truncated").and_then(Value::as_bool), + Some(true) + ); + assert_eq!( + audit_art.get("policyErrorCount").and_then(Value::as_u64), + Some(2) + ); + assert_eq!( + audit_art.get("malformedLineCount").and_then(Value::as_u64), + Some(1) + ); + + audit_parts.audit_state = EgressAuditState::Missing {}; + let missing_audit_v1 = V2AttemptRecord::try_new(audit_parts).unwrap().project_v1(); + assert_eq!( + missing_audit_v1.result.status, + EvalResultStatus::InfraFailed + ); + assert_eq!(missing_audit_v1.result.score, None); + assert!(missing_audit_v1.result.artifacts.iter().any(|a| { + a.get("kind").and_then(Value::as_str) == Some("egress-audit-missing") + && a.get("path").and_then(Value::as_str) == Some(EGRESS_AUDIT_ARTIFACT_PATH) + })); + + // Scenario 3: CriticalArtifactsState::Missing -> RetryableInfra (score: None). + let mut missing_art_parts = base_v2_parts(); + missing_art_parts.critical_artifacts_state = CriticalArtifactsState::Missing { + paths: vec!["transcript.json".into()], + }; + let missing_art_v1 = V2AttemptRecord::try_new(missing_art_parts) + .unwrap() + .project_v1(); + assert_eq!(missing_art_v1.result.status, EvalResultStatus::InfraFailed); + assert_eq!(missing_art_v1.result.score, None); + + // Scenario 4: CancelledBeforeVerification -> BlockedIndeterminate; + // Evaluated { cancelled_after_verification: true } -> Selected(Completed) + marker. + let mut cancel_before_parts = base_v2_parts(); + cancel_before_parts.trial_outcome = TrialOutcome::CancelledBeforeVerification { reason: None }; + let cancel_before_v1 = V2AttemptRecord::try_new(cancel_before_parts) + .unwrap() + .project_v1(); + assert_eq!( + cancel_before_v1.result.status, + EvalResultStatus::Indeterminate + ); + assert_eq!(cancel_before_v1.result.score, None); + + let mut cancel_after_parts = base_v2_parts(); + cancel_after_parts.trial_outcome = TrialOutcome::Evaluated { + subject: SubjectOutcome::Completed {}, + verifier: VerifierOutcome::Verified { score: 1.0 }, + cancelled_after_verification: true, + }; + let cancel_after_v1 = V2AttemptRecord::try_new(cancel_after_parts) + .unwrap() + .project_v1(); + assert_eq!(cancel_after_v1.result.status, EvalResultStatus::Completed); + assert_eq!(cancel_after_v1.result.score, Some(1.0)); + assert!( + cancel_after_v1 + .result + .artifacts + .iter() + .any(|a| a.get("kind").and_then(Value::as_str) == Some("cancelled-after-verification")) + ); + + // Scenario 5: PostHocRecovery upgrades immutable record monotonically via projection; + // MissingReward -> RetryableInfra (score: None); genuine 0.0 is terminal and earliest-selected. + let mut unrecovered_parts = base_v2_parts(); + unrecovered_parts.sequence = nz(2); + unrecovered_parts.cleanup_state = CleanupState::Unconfirmed { + reason: "cleanup pending".into(), + }; + unrecovered_parts.metering = AttemptMetering::Unknown; + let unrecovered_rec = V2AttemptRecord::try_new(unrecovered_parts).unwrap(); + assert!(matches!( + unrecovered_rec.disposition(), + V2Disposition::BlockedIndeterminate { .. } + )); + + let recovery = PostHocRecovery::try_new( + 30, + Some(AttemptMetering::Complete { + usage: sample_usage(), + model: Some("deepseek-v4-flash".into()), + }), + Some(CleanupState::Confirmed {}), + ) + .unwrap(); + let recovered_disp = unrecovered_rec.disposition_with_recovery(std::slice::from_ref(&recovery)); + assert_eq!( + recovered_disp, + V2Disposition::Selected { + status: SelectedStatus::Completed, + score: Some(1.0), + failure_reason: None, + } + ); + let recovered_v1 = unrecovered_rec.project_v1_with_recovery(&[recovery]); + assert_eq!(recovered_v1.sequence, nz(2)); + assert_eq!(recovered_v1.result.status, EvalResultStatus::Completed); + assert_eq!(recovered_v1.result.usage, Some(sample_usage())); + assert!(costs_match_within_tolerance( + recovered_v1.result.cost_usd, + Some(0.0000352) + )); + assert!( + recovered_v1 + .result + .artifacts + .iter() + .any(|a| a.get("kind").and_then(Value::as_str) == Some("post-hoc-evidence-recovery")) + ); + // Original record stays immutable. + assert!(matches!( + unrecovered_rec.cleanup_state(), + CleanupState::Unconfirmed { .. } + )); + + let mut missing_reward_parts = base_v2_parts(); + missing_reward_parts.trial_outcome = TrialOutcome::Evaluated { + subject: SubjectOutcome::Completed {}, + verifier: VerifierOutcome::MissingReward { + reason: "reward.json missing".into(), + }, + cancelled_after_verification: false, + }; + let missing_reward_v1 = V2AttemptRecord::try_new(missing_reward_parts) + .unwrap() + .project_v1(); + assert_eq!( + missing_reward_v1.result.status, + EvalResultStatus::InfraFailed + ); + assert_eq!(missing_reward_v1.result.score, None); + + // Selection picks earliest valid attempt (sequence 2, score 0.0) over sequence 1 (InfraFailed) + // and sequence 3 (Completed, score 1.0) in both v2 and v1. + let mut zero_seq2_parts = base_v2_parts(); + zero_seq2_parts.sequence = nz(2); + zero_seq2_parts.trial_outcome = TrialOutcome::Evaluated { + subject: SubjectOutcome::Completed {}, + verifier: VerifierOutcome::Verified { score: 0.0 }, + cancelled_after_verification: false, + }; + let zero_seq2 = V2AttemptRecord::try_new(zero_seq2_parts).unwrap(); + + let mut win_seq3_parts = base_v2_parts(); + win_seq3_parts.sequence = nz(3); + let win_seq3 = V2AttemptRecord::try_new(win_seq3_parts).unwrap(); + + let v2_attempts = [win_seq3.clone(), infra_rec.clone(), zero_seq2.clone()]; + let selected_v2 = select_v2_cell_result(&v2_attempts).unwrap(); + assert_eq!(selected_v2.sequence(), nz(2)); + assert_eq!( + selected_v2.disposition(), + V2Disposition::Selected { + status: SelectedStatus::Completed, + score: Some(0.0), + failure_reason: None, + } + ); + + let v1_attempts = [ + win_seq3.project_v1(), + infra_rec.project_v1(), + zero_seq2.project_v1(), + ]; + let selected_v1 = select_cell_result(&v1_attempts).unwrap(); + assert_eq!(selected_v1.sequence, nz(2)); + assert_eq!(selected_v1.result.score, Some(0.0)); +}