likang/cdsl-cad
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

wip: preserve local engine and agent workflow changes before ganjihong refactor integration

Browse Source
This commit is contained in:
李康
2026-09-09 17:08:24 +08:00
parent a5c3ec1f36
commit 387b99e255
112 changed files with 3543 additions and 12177 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/cad_agent/evals/TOKEN_BASELINE.md
-21
View File
@@ -1,21 +0,0 @@
# Token Baseline Format
`python -m app.cad_agent.evals.live --suite release --require-live` requires
`--baseline-report` to point at a measured protocol-2 report. It is not a
fixture or an estimate. Record three independent runs of every release
scenario with the same author request configuration and runtime profile.
| Field | Requirement |
| --- | --- |
| `schema_version` | `cad.token-baseline.v1` |
| `protocol_version` | `2.0` |
| `author` | Exact `provider`, `model`, `api_style`, `reasoning_effort`, and `sampling` object from the v3 report |
| `runtime_profile_sha256` | Exact v3 `runtime_profile_sha256` |
| `measurement` | `prompt_tokens` when provider usage exists; otherwise `context_chars` |
| `scenarios[]` | One entry for every versioned release scenario, including its request SHA-256 |
| `repetitions[]` | Entries `1`, `2`, and `3`, with non-negative `author_metric` and `plan_review_metric`, plus completion, final-review, and deterministic-claim booleans |
The comparison uses `author_metric + plan_review_metric` for the v2 median and
the v3 author metric only. It fails closed when provenance differs, any
repetition is absent, completion/review/claim rates drop, plan-review calls
remain, or the median reduction is below 30%.
backend/app/cad_agent/evals/__init__.py
+1 -1
View File
@@ -1 +1 @@
"""Live, non-mocked protocol v3 release evaluations."""
"""Local evaluation helpers for the single-stage Authoring protocol."""
backend/app/cad_agent/evals/create_isolated_task.py
+4 -4
View File
@@ -1,4 +1,4 @@
"""Create a resumable live-evaluation task without starting its workflow."""
"""Create an isolated single-stage task without starting its workflow."""
from __future__ import annotations
@@ -9,12 +9,12 @@ from pathlib import Path
import secrets
import sys
from app.cad_agent.composition import compose_v3
from app.cad_agent.composition import compose_cad_services
from app.settings import get_settings
def _arguments() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Create one isolated live CAD task for resume_one_step.")
parser = argparse.ArgumentParser(description="Create one isolated Authoring CDSL evaluation task.")
parser.add_argument("--report-root", required=True, type=Path)
parser.add_argument("--prompt", required=True)
parser.add_argument("--task-id")
@@ -25,7 +25,7 @@ def main() -> int:
arguments = _arguments()
root = arguments.report_root.resolve()
settings = get_settings()
services = compose_v3(replace(
services = compose_cad_services(replace(
settings,
task_root=root / "artifacts",
conversation_root=root / "conversations",
backend/app/cad_agent/evals/fixtures/comprehensive.json
-423
View File
@@ -1,423 +0,0 @@
{
"schema_version": "cad.comprehensive-prompt-fixtures.v1",
"source_document": "docs/cad-agent-v3-comprehensive-prompt-test-target.md",
"source_document_sha256": "79fbfeba41b2bf1fa37d2e90df1c4f5f06460f725f8898ae4238676bfea82d8b",
"scenarios": [
{
"id": "rectangular_mounting_plate",
"units": "mm",
"request": "生成一个CNC矩形安装板，长100毫米，宽60毫米，厚8毫米，四角圆角R6，四角各有一个直径8毫米的贯穿孔，孔中心距左右边10毫米、上下边10毫米，中心有一个直径30毫米的贯穿孔。",
"expected_phase": "COMPLETED",
"required_claim_kinds": ["solid_count_equals", "bbox_dimension_mm", "through_cylindrical_bore", "rectangular_corner_through_bore_pattern"],
"required_claims": [
{"claim_kind": "solid_count_equals", "expected": {"value": 1}},
{"claim_kind": "bbox_dimension_mm", "expected": {"axis": "x", "value": 100}},
{"claim_kind": "bbox_dimension_mm", "expected": {"axis": "y", "value": 60}},
{"claim_kind": "bbox_dimension_mm", "expected": {"axis": "z", "value": 8}},
{"claim_kind": "through_cylindrical_bore", "expected": {"diameter_mm": 8, "count": 4}},
{"claim_kind": "through_cylindrical_bore", "expected": {"diameter_mm": 30, "count": 1}},
{"claim_kind": "rectangular_corner_through_bore_pattern", "expected": {"diameter_mm": 8, "count": 4, "edge_offset_mm": 10}}
],
"required_atomic_ids": ["extrude_add_blind"],
"required_any_atomic_ids": ["hole_blind", "extrude_cut_blind"],
"validation_capability_gaps": [
{"id": "corner_radius", "description": "R6 outer corner radii"}
],
"max_author_turns": 56, "max_reviewer_turns": 20, "max_total_calls": 76, "max_wall_seconds": 1500, "max_total_tokens": 220000
},
{
"id": "circular_flange_pcd",
"units": "mm",
"request": "生成一个圆形法兰盘，外径120毫米，厚度12毫米，中心贯穿孔直径40毫米，在直径90毫米的分度圆上均布6个直径8毫米的贯穿孔。",
"expected_phase": "COMPLETED",
"required_claim_kinds": ["solid_count_equals", "bbox_dimension_mm", "through_cylindrical_bore", "circular_hole_pattern"],
"required_claims": [
{"claim_kind": "solid_count_equals", "expected": {"value": 1}},
{"claim_kind": "bbox_dimension_mm", "expected": {"axis": "x", "value": 120}},
{"claim_kind": "bbox_dimension_mm", "expected": {"axis": "y", "value": 120}},
{"claim_kind": "bbox_dimension_mm", "expected": {"axis": "z", "value": 12}},
{"claim_kind": "through_cylindrical_bore", "expected": {"diameter_mm": 40, "count": 1}},
{"claim_kind": "through_cylindrical_bore", "expected": {"diameter_mm": 8, "count": 6}},
{"claim_kind": "circular_hole_pattern", "expected": {"diameter_mm": 8, "count": 6, "pitch_radius_mm": 45, "concentric_bore_diameter_mm": 40}}
],
"required_atomic_ids": [],
"required_any_atomic_ids": ["hole_blind", "extrude_cut_blind", "pattern_linear"],
"required_any_atomic_id_groups": [["extrude_add_blind", "revolve_add"]],
"validation_capability_gaps": [],
"max_author_turns": 52, "max_reviewer_turns": 20, "max_total_calls": 72, "max_wall_seconds": 1500, "max_total_tokens": 320000
},
{
"id": "square_flange",
"units": "mm",
"request": "生成一个方形安装法兰，长100毫米，宽100毫米，厚度12毫米，四角圆角R8，中心贯穿孔直径45毫米，四角各有一个直径10毫米的安装孔，孔中心距相邻两边各15毫米。",
"expected_phase": "COMPLETED",
"required_claim_kinds": ["solid_count_equals", "bbox_dimension_mm", "through_cylindrical_bore", "rectangular_corner_through_bore_pattern"],
"required_claims": [
{"claim_kind": "solid_count_equals", "expected": {"value": 1}},
{"claim_kind": "bbox_dimension_mm", "expected": {"axis": "x", "value": 100}},
{"claim_kind": "bbox_dimension_mm", "expected": {"axis": "y", "value": 100}},
{"claim_kind": "bbox_dimension_mm", "expected": {"axis": "z", "value": 12}},
{"claim_kind": "through_cylindrical_bore", "expected": {"diameter_mm": 45, "count": 1}},
{"claim_kind": "through_cylindrical_bore", "expected": {"diameter_mm": 10, "count": 4}},
{"claim_kind": "rectangular_corner_through_bore_pattern", "expected": {"diameter_mm": 10, "count": 4, "edge_offset_mm": 15}}
],
"required_atomic_ids": ["extrude_add_blind"],
"required_any_atomic_ids": ["hole_blind", "extrude_cut_blind"],
"validation_capability_gaps": [
{"id": "corner_radius", "description": "R8 outer corner radii"}
],
"max_author_turns": 56, "max_reviewer_turns": 20, "max_total_calls": 76, "max_wall_seconds": 1500, "max_total_tokens": 220000
},
{
"id": "counterbored_mounting_plate",
"units": "mm",
"request": "生成一个矩形安装板，长120毫米，宽80毫米，厚度15毫米，四角各有一个直径9毫米的贯穿孔，每个孔顶部带直径16毫米、深5毫米的圆柱沉孔，孔中心距相邻边各12毫米。",
"expected_phase": "COMPLETED",
"required_claim_kinds": ["solid_count_equals", "bbox_dimension_mm", "through_cylindrical_bore", "cylindrical_bore", "cylindrical_bore_depth", "rectangular_corner_through_bore_pattern"],
"required_claims": [
{"claim_kind": "solid_count_equals", "expected": {"value": 1}},
{"claim_kind": "bbox_dimension_mm", "expected": {"axis": "x", "value": 120}},
{"claim_kind": "bbox_dimension_mm", "expected": {"axis": "y", "value": 80}},
{"claim_kind": "bbox_dimension_mm", "expected": {"axis": "z", "value": 15}},
{"claim_kind": "through_cylindrical_bore", "expected": {"diameter_mm": 9, "count": 4}},
{"claim_kind": "cylindrical_bore", "expected": {"diameter_mm": 16, "count": 4}},
{"claim_kind": "cylindrical_bore_depth", "expected": {"diameter_mm": 16, "count": 4, "depth_mm": 5}},
{"claim_kind": "rectangular_corner_through_bore_pattern", "expected": {"diameter_mm": 9, "count": 4, "edge_offset_mm": 12}}
],
"required_atomic_ids": ["extrude_add_blind"],
"required_any_atomic_ids": ["hole_counterbore", "hole_wizard"],
"validation_capability_gaps": [],
"max_author_turns": 64, "max_reviewer_turns": 24, "max_total_calls": 88, "max_wall_seconds": 1800, "max_total_tokens": 250000
},
{
"id": "countersunk_cover_plate",
"units": "mm",
"request": "生成一个盖板，长100毫米，宽70毫米，厚度8毫米，四角圆角R5，四角各有一个直径6.5毫米的贯穿孔，孔顶部带90度沉头，沉头最大直径12毫米。",
"expected_phase": "COMPLETED",
"required_claim_kinds": ["solid_count_equals", "bbox_dimension_mm", "through_cylindrical_bore", "conical_bore"],
"required_claims": [
{"claim_kind": "solid_count_equals", "expected": {"value": 1}},
{"claim_kind": "bbox_dimension_mm", "expected": {"axis": "x", "value": 100}},
{"claim_kind": "bbox_dimension_mm", "expected": {"axis": "y", "value": 70}},
{"claim_kind": "bbox_dimension_mm", "expected": {"axis": "z", "value": 8}},
{"claim_kind": "through_cylindrical_bore", "expected": {"diameter_mm": 6.5, "count": 4}},
{"claim_kind": "conical_bore", "expected": {"small_diameter_mm": 6.5, "large_diameter_mm": 12, "included_angle_deg": 90, "count": 4}}
],
"required_atomic_ids": ["extrude_add_blind"],
"required_any_atomic_ids": ["hole_countersink", "hole_wizard"],
"validation_capability_gaps": [
{"id": "corner_radius", "description": "R5 outer corner radii"}
],
"max_author_turns": 64, "max_reviewer_turns": 24, "max_total_calls": 88, "max_wall_seconds": 1800, "max_total_tokens": 250000
},
{
"id": "obround_slot_plate",
"units": "mm",
"request": "生成一个连接板，长140毫米，宽50毫米，厚度10毫米，两端各有一个长圆形贯穿槽，槽总长30毫米、宽12毫米，槽中心距板端20毫米，槽的长轴沿板长度方向。",
"expected_phase": "COMPLETED",
"required_claim_kinds": ["solid_count_equals", "bbox_dimension_mm", "visual"],
"required_claims": [
{"claim_kind": "solid_count_equals", "expected": {"value": 1}},
{"claim_kind": "bbox_dimension_mm", "expected": {"axis": "x", "value": 140}},
{"claim_kind": "bbox_dimension_mm", "expected": {"axis": "y", "value": 50}},
{"claim_kind": "bbox_dimension_mm", "expected": {"axis": "z", "value": 10}}
],
"required_atomic_ids": ["extrude_add_blind"],
"required_any_atomic_ids": ["extrude_cut_blind"],
"validation_capability_gaps": [
{"id": "slot_dimensions", "description": "two through obround slots with 30 x 12 dimensions"},
{"id": "slot_position_orientation", "description": "20 mm end offset and lengthwise major-axis orientation"}
],
"max_author_turns": 64, "max_reviewer_turns": 24, "max_total_calls": 88, "max_wall_seconds": 1800, "max_total_tokens": 250000
},
{
"id": "t_slot_test_block",
"units": "mm",
"request": "生成一个T形槽试块，长100毫米，宽50毫米，高25毫米，在顶面中心沿长度方向加工一条T形槽，槽口宽10毫米、深8毫米，槽底宽20毫米、总深15毫米，槽贯穿试块两端。",
"expected_phase": "COMPLETED",
"required_claim_kinds": ["solid_count_equals", "bbox_dimension_mm", "visual"],
"required_claims": [
{"claim_kind": "solid_count_equals", "expected": {"value": 1}},
{"claim_kind": "bbox_dimension_mm", "expected": {"axis": "x", "value": 100}},
{"claim_kind": "bbox_dimension_mm", "expected": {"axis": "y", "value": 50}},
{"claim_kind": "bbox_dimension_mm", "expected": {"axis": "z", "value": 25}}
],
"required_atomic_ids": ["extrude_add_blind"],
"required_any_atomic_ids": ["extrude_cut_blind"],
"validation_capability_gaps": [
{"id": "t_slot_cross_section", "description": "10 x 8 slot mouth, 20 mm lower width and 15 mm total depth"},
{"id": "t_slot_through_orientation", "description": "top-centred T-slot through both lengthwise ends"}
],
"max_author_turns": 72, "max_reviewer_turns": 28, "max_total_calls": 100, "max_wall_seconds": 2100, "max_total_tokens": 280000
},
{
"id": "rounded_rectangular_pocket",
"units": "mm",
"request": "生成一个矩形底板，长120毫米，宽80毫米，厚20毫米，在顶面中心加工一个长80毫米、宽45毫米、深12毫米的矩形口袋，口袋四角圆角R6，底部保留8毫米厚度。",
"expected_phase": "COMPLETED",
"required_claim_kinds": ["solid_count_equals", "bbox_dimension_mm", "visual"],
"required_claims": [
{"claim_kind": "solid_count_equals", "expected": {"value": 1}},
{"claim_kind": "bbox_dimension_mm", "expected": {"axis": "x", "value": 120}},
{"claim_kind": "bbox_dimension_mm", "expected": {"axis": "y", "value": 80}},
{"claim_kind": "bbox_dimension_mm", "expected": {"axis": "z", "value": 20}}
],
"required_atomic_ids": ["extrude_add_blind"],
"required_any_atomic_ids": ["extrude_cut_blind"],
"validation_capability_gaps": [
{"id": "pocket_dimensions", "description": "centred 80 x 45 x 12 pocket with 8 mm remaining floor"},
{"id": "pocket_corner_radius", "description": "R6 internal pocket corners"}
],
"max_author_turns": 64, "max_reviewer_turns": 24, "max_total_calls": 88, "max_wall_seconds": 1800, "max_total_tokens": 250000
},
{
"id": "two_level_pocket_plate",
"units": "mm",
"request": "生成一个CNC加工板，长140毫米，宽100毫米，厚25毫米。顶面中心先加工一个长100毫米、宽70毫米、深8毫米的矩形口袋，再在第一级口袋中心加工一个长60毫米、宽35毫米、额外深7毫米的第二级口袋，所有内角圆角R5。",
"expected_phase": "COMPLETED",
"required_claim_kinds": ["solid_count_equals", "bbox_dimension_mm", "visual"],
"required_claims": [
{"claim_kind": "solid_count_equals", "expected": {"value": 1}},
{"claim_kind": "bbox_dimension_mm", "expected": {"axis": "x", "value": 140}},
{"claim_kind": "bbox_dimension_mm", "expected": {"axis": "y", "value": 100}},
{"claim_kind": "bbox_dimension_mm", "expected": {"axis": "z", "value": 25}}
],
"required_atomic_ids": ["extrude_add_blind"],
"required_any_atomic_ids": ["extrude_cut_blind"],
"validation_capability_gaps": [
{"id": "pocket_levels", "description": "100 x 70 x 8 first pocket and centred 60 x 35 x 7 additional-depth second pocket"},
{"id": "pocket_corner_radii", "description": "R5 on every internal corner"}
],
"max_author_turns": 80, "max_reviewer_turns": 30, "max_total_calls": 110, "max_wall_seconds": 2400, "max_total_tokens": 320000
},
{
"id": "cross_drilled_valve_block",
"units": "mm",
"request": "生成一个阀块试件，长80毫米，宽60毫米，高50毫米。沿长度方向加工一个直径20毫米的贯穿孔，沿宽度方向加工一个直径12毫米的贯穿孔，两个孔的轴线在零件中心相交。",
"expected_phase": "COMPLETED",
"required_claim_kinds": ["solid_count_equals", "bbox_dimension_mm", "through_cylindrical_bore", "orthogonal_intersecting_through_bores"],
"required_claims": [
{"claim_kind": "solid_count_equals", "expected": {"value": 1}},
{"claim_kind": "bbox_dimension_mm", "expected": {"axis": "x", "value": 80}},
{"claim_kind": "bbox_dimension_mm", "expected": {"axis": "y", "value": 60}},
{"claim_kind": "bbox_dimension_mm", "expected": {"axis": "z", "value": 50}},
{"claim_kind": "through_cylindrical_bore", "expected": {"diameter_mm": 20, "count": 1}},
{"claim_kind": "through_cylindrical_bore", "expected": {"diameter_mm": 12, "count": 1}},
{"claim_kind": "orthogonal_intersecting_through_bores", "expected": {"first_diameter_mm": 20, "second_diameter_mm": 12, "first_axis": "x", "second_axis": "y"}}
],
"required_atomic_ids": ["extrude_add_blind"],
"required_any_atomic_ids": ["hole_blind", "extrude_cut_blind"],
"validation_capability_gaps": [],
"max_author_turns": 72, "max_reviewer_turns": 28, "max_total_calls": 100, "max_wall_seconds": 2100, "max_total_tokens": 280000
},
{
"id": "double_hole_linkage_arm",
"units": "mm",
"request": "生成一个机械臂双孔连杆，两个销孔中心距120毫米，连杆厚度12毫米，两端外圆直径40毫米，两个销孔直径16毫米，中间杆身最小宽度24毫米，轮廓平滑相切，中部设置三个直径14毫米的减重贯穿孔。",
"expected_phase": "COMPLETED",
"required_claim_kinds": ["solid_count_equals", "bbox_rank_dimension_mm", "through_cylindrical_bore", "collinear_through_bore_chain", "visual"],
"required_claims": [
{"claim_kind": "solid_count_equals", "expected": {"value": 1}},
{"claim_kind": "bbox_rank_dimension_mm", "expected": {"rank": "minimum", "value": 12}},
{"claim_kind": "through_cylindrical_bore", "expected": {"diameter_mm": 16, "count": 2}},
{"claim_kind": "through_cylindrical_bore", "expected": {"diameter_mm": 14, "count": 3}},
{"claim_kind": "collinear_through_bore_chain", "expected": {"diameter_mm": 16, "adjacent_distances_mm": [120], "tolerance_mm": 0.1}},
{"claim_kind": "visual", "expected": {}}
],
"required_atomic_ids": ["extrude_add_blind"],
"required_any_atomic_ids": ["hole_blind", "extrude_cut_blind"],
"validation_capability_gaps": [
{"id": "linkage_outline", "description": "40 mm end circles, 24 mm minimum web width and smooth tangency"},
{"id": "lightening_holes", "description": "three 14 mm central through holes"}
],
"max_author_turns": 84, "max_reviewer_turns": 32, "max_total_calls": 116, "max_wall_seconds": 2400, "max_total_tokens": 340000
},
{
"id": "three_hole_linkage",
"units": "mm",
"request": "生成一个三孔机械连杆，三个孔的中心位于同一直线上，相邻孔中心距分别为60毫米和80毫米，三个孔直径均为12毫米，连杆厚度10毫米，每个孔周围外圆直径32毫米，各段外轮廓平滑连接。",
"expected_phase": "COMPLETED",
"required_claim_kinds": ["solid_count_equals", "bbox_rank_dimension_mm", "through_cylindrical_bore", "collinear_through_bore_chain", "visual"],
"required_claims": [
{"claim_kind": "solid_count_equals", "expected": {"value": 1}},
{"claim_kind": "bbox_rank_dimension_mm", "expected": {"rank": "minimum", "value": 10}},
{"claim_kind": "through_cylindrical_bore", "expected": {"diameter_mm": 12, "count": 3}},
{"claim_kind": "collinear_through_bore_chain", "expected": {"diameter_mm": 12, "adjacent_distances_mm": [60, 80], "tolerance_mm": 0.1}},
{"claim_kind": "visual", "expected": {}}
],
"required_atomic_ids": ["extrude_add_blind"],
"required_any_atomic_ids": ["hole_blind", "extrude_cut_blind"],
"validation_capability_gaps": [
{"id": "linkage_outer_circles", "description": "32 mm outer circles with smoothly connected segments"}
],
"max_author_turns": 84, "max_reviewer_turns": 32, "max_total_calls": 116, "max_wall_seconds": 2400, "max_total_tokens": 340000
},
{
"id": "l_bracket",
"units": "mm",
"request": "生成一个整体式L形角码，水平底板长80毫米、宽50毫米、厚8毫米，竖直板高60毫米、宽50毫米、厚8毫米，两板成90度。水平板上有两个直径8毫米贯穿孔，竖直板上有两个直径8毫米贯穿孔，孔左右对称。",
"expected_phase": "COMPLETED",
"required_claim_kinds": ["single_connected_body", "through_cylindrical_bore", "visual"],
"required_claims": [
{"claim_kind": "single_connected_body", "expected": {}},
{"claim_kind": "through_cylindrical_bore", "expected": {"diameter_mm": 8, "count": 4}},
{"claim_kind": "visual", "expected": {}}
],
"required_atomic_ids": ["extrude_add_blind"],
"required_any_atomic_ids": ["hole_blind", "extrude_cut_blind"],
"validation_capability_gaps": [
{"id": "l_bracket_dimensions", "description": "base and upright dimensions, thicknesses and 90 degree relationship"},
{"id": "l_bracket_hole_symmetry", "description": "two symmetric 8 mm through holes on each plate"}
],
"max_author_turns": 88, "max_reviewer_turns": 32, "max_total_calls": 120, "max_wall_seconds": 2700, "max_total_tokens": 360000
},
{
"id": "ribbed_l_bracket",
"units": "mm",
"request": "生成一个整体式L形机械支架，底板长100毫米、宽60毫米、厚10毫米，竖板高80毫米、宽60毫米、厚10毫米，两板成90度。底板和竖板之间设置两个厚度8毫米的三角加强筋。底板有四个直径9毫米贯穿孔，竖板中心有一个直径30毫米贯穿孔。",
"expected_phase": "COMPLETED",
"required_claim_kinds": ["single_connected_body", "through_cylindrical_bore", "visual"],
"required_claims": [
{"claim_kind": "single_connected_body", "expected": {}},
{"claim_kind": "through_cylindrical_bore", "expected": {"diameter_mm": 9, "count": 4}},
{"claim_kind": "through_cylindrical_bore", "expected": {"diameter_mm": 30, "count": 1}},
{"claim_kind": "visual", "expected": {}}
],
"required_atomic_ids": ["extrude_add_blind"],
"required_any_atomic_ids": ["hole_blind", "extrude_cut_blind"],
"validation_capability_gaps": [
{"id": "bracket_dimensions", "description": "100 x 60 x 10 base and 80 x 60 x 10 upright at 90 degrees"},
{"id": "triangular_ribs", "description": "two triangular 8 mm thick ribs"},
{"id": "bracket_hole_placement", "description": "four base 9 mm holes and centred upright 30 mm hole"}
],
"max_author_turns": 96, "max_reviewer_turns": 36, "max_total_calls": 132, "max_wall_seconds": 3000, "max_total_tokens": 400000
},
{
"id": "u_bearing_support",
"units": "mm",
"request": "生成一个U形轴承支座，底板长100毫米、宽60毫米、厚12毫米，两侧竖耳厚12毫米、高55毫米，两个竖耳内侧间距40毫米。两个竖耳上各有一个直径20毫米的同轴贯穿孔，孔轴线距底板上表面35毫米。底板四角各有一个直径8毫米安装孔。",
"expected_phase": "COMPLETED",
"required_claim_kinds": ["single_connected_body", "through_cylindrical_bore", "coaxial_through_bore_group", "visual"],
"required_claims": [
{"claim_kind": "single_connected_body", "expected": {}},
{"claim_kind": "through_cylindrical_bore", "expected": {"diameter_mm": 20, "count": 2}},
{"claim_kind": "through_cylindrical_bore", "expected": {"diameter_mm": 8, "count": 4}},
{"claim_kind": "coaxial_through_bore_group", "expected": {"diameter_mm": 20, "count": 2}},
{"claim_kind": "visual", "expected": {}}
],
"required_atomic_ids": ["extrude_add_blind"],
"required_any_atomic_ids": ["hole_blind", "extrude_cut_blind"],
"validation_capability_gaps": [
{"id": "u_support_dimensions", "description": "base and ear dimensions, ear spacing and 35 mm axis height"},
{"id": "base_hole_positions", "description": "four 8 mm base mounting-hole positions"}
],
"max_author_turns": 96, "max_reviewer_turns": 36, "max_total_calls": 132, "max_wall_seconds": 3000, "max_total_tokens": 400000
},
{
"id": "double_lug_mount",
"units": "mm",
"request": "生成一个双耳连接座，底座长90毫米、宽60毫米、厚12毫米，底座上有两个平行耳板，每个耳板厚10毫米、高50毫米，两耳板内侧间距30毫米。两个耳板上各有一个直径16毫米的同轴贯穿销孔，孔中心距底座上表面32毫米。",
"expected_phase": "COMPLETED",
"required_claim_kinds": ["single_connected_body", "through_cylindrical_bore", "coaxial_through_bore_group", "visual"],
"required_claims": [
{"claim_kind": "single_connected_body", "expected": {}},
{"claim_kind": "through_cylindrical_bore", "expected": {"diameter_mm": 16, "count": 2}},
{"claim_kind": "coaxial_through_bore_group", "expected": {"diameter_mm": 16, "count": 2}},
{"claim_kind": "visual", "expected": {}}
],
"required_atomic_ids": ["extrude_add_blind"],
"required_any_atomic_ids": ["hole_blind", "extrude_cut_blind"],
"validation_capability_gaps": [
{"id": "lug_dimensions", "description": "base, lug thickness/height, 30 mm inside spacing and 32 mm bore-axis height"}
],
"max_author_turns": 92, "max_reviewer_turns": 34, "max_total_calls": 126, "max_wall_seconds": 2700, "max_total_tokens": 380000
},
{
"id": "stepped_shaft",
"units": "mm",
"request": "生成一根阶梯轴，总长120毫米。第一段直径30毫米、长40毫米；第二段直径24毫米、长50毫米；第三段直径18毫米、长30毫米。所有轴肩过渡圆角R2，两端倒角1毫米乘45度。",
"expected_phase": "COMPLETED",
"required_claim_kinds": ["single_connected_body", "bbox_rank_dimension_mm", "outer_cylindrical_surface", "visual"],
"required_claims": [
{"claim_kind": "single_connected_body", "expected": {}},
{"claim_kind": "bbox_rank_dimension_mm", "expected": {"rank": "maximum", "value": 120}},
{"claim_kind": "outer_cylindrical_surface", "expected": {"diameter_mm": 30, "count": 1}},
{"claim_kind": "outer_cylindrical_surface", "expected": {"diameter_mm": 24, "count": 1}},
{"claim_kind": "outer_cylindrical_surface", "expected": {"diameter_mm": 18, "count": 1}},
{"claim_kind": "visual", "expected": {}}
],
"required_atomic_ids": ["revolve_add"],
"required_any_atomic_ids": ["fillet", "chamfer"],
"validation_capability_gaps": [
{"id": "shaft_segment_lengths", "description": "the 30 mm, 24 mm and 18 mm shaft sections have axial lengths 40 mm, 50 mm and 30 mm"},
{"id": "shaft_finish_features", "description": "R2 shoulders and 1 x 45 degree end chamfers"}
],
"max_author_turns": 72, "max_reviewer_turns": 28, "max_total_calls": 100, "max_wall_seconds": 2100, "max_total_tokens": 280000
},
{
"id": "keyed_stepped_shaft",
"units": "mm",
"request": "生成一根阶梯传动轴，总长140毫米，中间轴段直径30毫米、长70毫米，两端轴段直径20毫米、各长35毫米。中间轴段沿轴向加工一条平键槽，键槽宽8毫米、深3.3毫米、长50毫米，所有轴肩圆角R2。",
"expected_phase": "COMPLETED",
"required_claim_kinds": ["single_connected_body", "bbox_rank_dimension_mm", "outer_cylindrical_surface", "visual"],
"required_claims": [
{"claim_kind": "single_connected_body", "expected": {}},
{"claim_kind": "bbox_rank_dimension_mm", "expected": {"rank": "maximum", "value": 140}},
{"claim_kind": "outer_cylindrical_surface", "expected": {"diameter_mm": 30, "count": 1}},
{"claim_kind": "outer_cylindrical_surface", "expected": {"diameter_mm": 20, "count": 2}},
{"claim_kind": "visual", "expected": {}}
],
"required_atomic_ids": ["revolve_add"],
"required_any_atomic_ids": ["extrude_cut_blind", "fillet"],
"validation_capability_gaps": [
{"id": "shaft_segment_lengths", "description": "the 20 mm end sections are 35 mm each and the 30 mm middle section is 70 mm"},
{"id": "keyway", "description": "8 x 3.3 x 50 axial keyway on the middle shaft section"},
{"id": "shoulder_fillet", "description": "R2 on all shaft shoulders"}
],
"max_author_turns": 84, "max_reviewer_turns": 32, "max_total_calls": 116, "max_wall_seconds": 2400, "max_total_tokens": 340000
},
{
"id": "flanged_sleeve",
"units": "mm",
"request": "生成一个机械套筒，外径50毫米，内孔直径30毫米，总长60毫米。套筒一端带外径70毫米、厚度10毫米的法兰，法兰上在直径56毫米分度圆上均布4个直径7毫米贯穿孔。",
"expected_phase": "COMPLETED",
"required_claim_kinds": ["single_connected_body", "bbox_rank_dimension_mm", "through_cylindrical_bore", "circular_hole_pattern", "outer_cylindrical_surface"],
"required_claims": [
{"claim_kind": "single_connected_body", "expected": {}},
{"claim_kind": "bbox_rank_dimension_mm", "expected": {"rank": "minimum", "value": 60}},
{"claim_kind": "bbox_rank_dimension_mm", "expected": {"rank": "maximum", "value": 70}},
{"claim_kind": "outer_cylindrical_surface", "expected": {"diameter_mm": 50, "count": 1, "axial_span_mm": 50}},
{"claim_kind": "outer_cylindrical_surface", "expected": {"diameter_mm": 70, "count": 1, "axial_span_mm": 10}},
{"claim_kind": "through_cylindrical_bore", "expected": {"diameter_mm": 30, "count": 1}},
{"claim_kind": "through_cylindrical_bore", "expected": {"diameter_mm": 7, "count": 4}},
{"claim_kind": "circular_hole_pattern", "expected": {"diameter_mm": 7, "count": 4, "pitch_radius_mm": 28, "concentric_bore_diameter_mm": 30}}
],
"required_atomic_ids": ["revolve_add"],
"required_any_atomic_ids": ["hole_blind", "extrude_cut_blind"],
"validation_capability_gaps": [
{"id": "flange_at_one_end", "description": "the 70 mm flange occurs at exactly one end of the coaxial sleeve, not at an interior position"}
],
"max_author_turns": 84, "max_reviewer_turns": 32, "max_total_calls": 116, "max_wall_seconds": 2400, "max_total_tokens": 340000
},
{
"id": "chamfered_bushing",
"units": "mm",
"request": "生成一个圆柱轴套，外径40毫米，内孔直径25毫米，长度50毫米，两端外边缘倒角1.5毫米乘45度，两端内孔边缘倒角1毫米乘45度。",
"expected_phase": "COMPLETED",
"required_claim_kinds": ["single_connected_body", "bbox_rank_dimension_mm", "through_cylindrical_bore", "visual"],
"required_claims": [
{"claim_kind": "single_connected_body", "expected": {}},
{"claim_kind": "bbox_rank_dimension_mm", "expected": {"rank": "minimum", "value": 40}},
{"claim_kind": "bbox_rank_dimension_mm", "expected": {"rank": "maximum", "value": 50}},
{"claim_kind": "through_cylindrical_bore", "expected": {"diameter_mm": 25, "count": 1}},
{"claim_kind": "visual", "expected": {}}
],
"required_atomic_ids": ["revolve_add"],
"required_any_atomic_ids": ["chamfer"],
"validation_capability_gaps": [
{"id": "inside_outside_chamfers", "description": "both 1.5 x 45 degree external and 1 x 45 degree internal chamfers"}
],
"max_author_turns": 72, "max_reviewer_turns": 28, "max_total_calls": 100, "max_wall_seconds": 2100, "max_total_tokens": 280000
}
]
}
backend/app/cad_agent/evals/fixtures/release.json
-78
View File
@@ -1,78 +0,0 @@
{
"schema_version": "cad.live-eval-fixtures.v1",
"scenarios": [
{
"id": "rectangular_plate",
"units": "mm",
"request": "Create one connected rectangular plate, 80 mm long, 50 mm wide, and 8 mm thick. Use millimetres. Verify the single solid and all three bounding-box dimensions.",
"expected_phase": "COMPLETED",
"required_claim_kinds": ["solid_count_equals", "bbox_dimension_mm"],
"required_claims": [
{"claim_kind": "solid_count_equals", "expected": {"value": 1}},
{"claim_kind": "bbox_dimension_mm", "expected": {"axis": "x", "value": 80, "tolerance_mm": 0.01}},
{"claim_kind": "bbox_dimension_mm", "expected": {"axis": "y", "value": 50, "tolerance_mm": 0.01}},
{"claim_kind": "bbox_dimension_mm", "expected": {"axis": "z", "value": 8, "tolerance_mm": 0.01}}
],
"required_atomic_ids": ["extrude_add_blind"],
"max_author_turns": 32,
"max_reviewer_turns": 12,
"max_total_calls": 44,
"max_wall_seconds": 900,
"max_total_tokens": 120000
},
{
"id": "simple_flange",
"units": "mm",
"request": "Create one connected round flange in millimetres: outer diameter 120 mm, thickness 12 mm, and a centered 40 mm through bore. Verify the single solid, thickness, and through bore topology.",
"expected_phase": "COMPLETED",
"required_claim_kinds": ["solid_count_equals", "bbox_dimension_mm", "through_cylindrical_bore"],
"required_claims": [
{"claim_kind": "solid_count_equals", "expected": {"value": 1}},
{"claim_kind": "bbox_dimension_mm", "expected": {"axis": "x", "value": 120, "tolerance_mm": 0.01}},
{"claim_kind": "bbox_dimension_mm", "expected": {"axis": "y", "value": 120, "tolerance_mm": 0.01}},
{"claim_kind": "bbox_dimension_mm", "expected": {"axis": "z", "value": 12, "tolerance_mm": 0.01}},
{"claim_kind": "through_cylindrical_bore", "expected": {"diameter_mm": 40, "count": 1, "tolerance_mm": 0.01}}
],
"required_atomic_ids": ["extrude_add_blind"],
"required_any_atomic_ids": ["hole_blind", "extrude_cut_blind"],
"max_author_turns": 40,
"max_reviewer_turns": 16,
"max_total_calls": 56,
"max_wall_seconds": 1200,
"max_total_tokens": 160000
},
{
"id": "ribbed_mounting_plate",
"units": "mm",
"request": "Create one connected mounting plate in millimetres with a rectangular base, a vertical reinforcing rib, and four equally spaced mounting through holes. State reasonable dimensions as assumptions, then verify a single connected solid and the four-hole pattern.",
"expected_phase": "COMPLETED",
"required_claim_kinds": ["solid_count_equals", "circular_hole_pattern", "through_cylindrical_bore"],
"required_claims": [
{"claim_kind": "solid_count_equals", "expected": {"value": 1}}
],
"required_atomic_ids": ["extrude_add_blind", "hole_blind"],
"max_author_turns": 56,
"max_reviewer_turns": 20,
"max_total_calls": 76,
"max_wall_seconds": 1500,
"max_total_tokens": 200000
},
{
"id": "selector_finish",
"units": "mm",
"request": "Create one connected rectangular plate with an added rounded or chamfered edge selected from the current topology. Use millimetres, state assumptions, and verify the final single solid.",
"expected_phase": "COMPLETED",
"required_claim_kinds": ["solid_count_equals"],
"required_claims": [
{"claim_kind": "solid_count_equals", "expected": {"value": 1}}
],
"required_atomic_ids": ["extrude_add_blind"],
"required_any_atomic_ids": ["fillet", "chamfer"],
"max_author_turns": 40,
"max_reviewer_turns": 16,
"max_total_calls": 56,
"max_wall_seconds": 1200,
"max_total_tokens": 160000
}
]
}
backend/app/cad_agent/evals/live.py
-990
View File
@@ -1,990 +0,0 @@
"""Run the real provider, engine, reviewer and v3 state machine end to end.
``--require-live`` deliberately fails closed: missing credentials, blocked
network, a skipped model capability, a timeout, or any scenario failure emits
``LIVE_EVAL_BLOCKED``/failure details and exits non-zero.
"""
from __future__ import annotations
import argparse
import asyncio
from dataclasses import replace
from datetime import datetime, timezone
from hashlib import sha256
import json
from pathlib import Path
import secrets
from statistics import median
import subprocess
import sys
from typing import Any
from app.cad_agent.application.capabilities import verify_model_capability
from app.cad_agent.application.workflow import ModelIdentity
from app.cad_agent.composition import compose_v3
from app.cad_agent.domain.claim_matching import contains_expected
from app.cad_agent.domain.operation_contract import canonical_hash
from app.cad_agent.domain.verifier_registry import default_registry
from app.cad_agent.evals.token_baseline import (
TokenBaselineError,
author_request_identity,
compare_token_baseline,
load_token_baseline,
profile_sha256,
validate_token_baseline_provenance,
)
from app.settings import BACKEND_ROOT, get_settings
_LEGACY_CANDIDATE_EVIDENCE_FILES = frozenset({
"candidate.json",
"candidate-review.json",
"model.cdsl.json",
"model.step",
"model.glb",
"model.topology.json",
"rebuild-report.json",
"renders/render-manifest.json",
"renders/contact-sheet.jpg",
})
# A v3.2 Feature DAG publishes one atomic checkpoint after local verification.
# It deliberately has no candidate review or technical render bundle; the
# final review owns that later evidence. GLB is optional because a preview
# conversion outage must not invalidate an otherwise sound STEP checkpoint.
_FEATURE_NODE_EVIDENCE_FILES = frozenset({
"input.json",
"model.cdsl.json",
"model.step",
"model.topology.json",
"node-verification.json",
"rebuild-report.json",
})
_FAILURE_LAYERS = frozenset({
"model_format_or_decision",
"v3_contract_or_verifier",
"cdsl_expression",
"engine_execution",
"independent_visual_review",
"configuration_or_network",
})
_ERROR_FAILURE_LAYERS = {
"AUTHOR_FORMAT_INVALID": "model_format_or_decision",
"AUTHOR_DECISION_REJECTED": "model_format_or_decision",
"STALE_WORKING_HEAD": "model_format_or_decision",
"FAILED_AUTHOR_FORMAT": "model_format_or_decision",
"MODEL_STRUCTURED_OUTPUT_UNSUPPORTED": "model_format_or_decision",
"RUNTIME_PRECONDITION_FAILED": "v3_contract_or_verifier",
"RUNTIME_CONTRACT_INVALID": "v3_contract_or_verifier",
"VERIFIER_UNAVAILABLE": "v3_contract_or_verifier",
"CLAIM_VERIFICATION_FAILED": "cdsl_expression",
"CANDIDATE_BUILD_FAILED": "engine_execution",
"CANDIDATE_REVIEW_REJECTED": "independent_visual_review",
"AUTHOR_TRANSPORT_UNAVAILABLE": "configuration_or_network",
"REVIEW_SERVICE_UNAVAILABLE": "configuration_or_network",
"RENDER_SERVICE_UNAVAILABLE": "configuration_or_network",
"STORAGE_FAILURE": "configuration_or_network",
"LIVE_EVAL_TIMEOUT": "configuration_or_network",
"FAILED_INTERNAL": "engine_execution",
}
def _arguments() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Run live protocol-v3 CAD evaluations.")
parser.add_argument("--suite", choices=("smoke", "release", "comprehensive"), default="smoke")
parser.add_argument("--require-live", action="store_true")
parser.add_argument("--allow-skip", action="store_true")
parser.add_argument("--author-provider")
parser.add_argument("--author-model")
parser.add_argument("--review-provider")
parser.add_argument("--review-model")
parser.add_argument("--scenario", action="append", dest="scenarios", help="Run a fixture scenario by stable ID. Repeat this option to select a comparison set.")
parser.add_argument("--repetitions", type=int, help="Run every selected scenario this many times.")
parser.add_argument("--author-guidance", choices=("on", "off"), help="Override CDSL author guidance for this run.")
parser.add_argument("--compare-guidance-reports", nargs=2, type=Path, metavar=("CONTROL", "TREATMENT"), help="Compare matched --author-guidance off/on report.json files without invoking providers.")
parser.add_argument("--baseline-report", type=Path, help="Measured pre-v3 token baseline JSON for a release run.")
return parser.parse_args()
def _fixture(suite: str, scenario_ids: list[str] | None = None) -> list[dict[str, Any]]:
fixture_name = "comprehensive.json" if suite == "comprehensive" else "release.json"
value = json.loads((Path(__file__).parent / "fixtures" / fixture_name).read_text(encoding="utf-8"))
if fixture_name == "comprehensive.json":
source_document = str(value.get("source_document") or "")
expected_digest = str(value.get("source_document_sha256") or "")
source_path = (BACKEND_ROOT.parent / source_document).resolve()
workspace_root = BACKEND_ROOT.parent.resolve()
if (
not source_document
or workspace_root not in source_path.parents
or not source_path.is_file()
or sha256(source_path.read_bytes()).hexdigest() != expected_digest
):
raise ValueError("Comprehensive fixture is not synchronized with its source document.")
values = [item for item in value.get("scenarios") or () if isinstance(item, dict)]
values = values[:2] if suite == "smoke" else values
if not scenario_ids:
return values
requested = list(dict.fromkeys(scenario_ids))
available = {str(item.get("id") or "") for item in values}
unknown = [scenario_id for scenario_id in requested if scenario_id not in available]
if unknown:
raise ValueError(f"Unknown scenario {unknown[0]!r} for suite {suite!r}")
selected = [item for item in values if str(item.get("id") or "") in set(requested)]
return selected
def _rejection_codes(events: list[dict[str, Any]]) -> list[str]:
codes: list[str] = []
for event in events:
payload = event.get("payload") if isinstance(event.get("payload"), dict) else {}
result = payload.get("result") if isinstance(payload.get("result"), dict) else payload
code = result.get("code") if isinstance(result, dict) else None
if code in {"AUTHOR_FORMAT_INVALID", "AUTHOR_DECISION_REJECTED", "STALE_WORKING_HEAD"}:
codes.append(str(code))
return codes
def _acceptance_coverage(
scenario: dict[str, Any],
requirements_contract: dict[str, Any] | None,
) -> dict[str, Any]:
"""Assess whether a fixture's acceptance contract is representable today.
Comprehensive prompts deliberately include manufacturing relationships that
may not yet have a deterministic verifier. A healthy-looking solid is
not evidence for those relationships, so they remain explicit capability
gaps instead of silently passing a scenario.
"""
actual = [
{
"claim_kind": str(claim.get("claim_kind") or ""),
"expected": claim.get("expected") if isinstance(claim.get("expected"), dict) else {},
}
for requirement in (requirements_contract or {}).get("requirements") or ()
if isinstance(requirement, dict)
for claim in requirement.get("acceptance_claims") or ()
if isinstance(claim, dict)
]
required = {str(value) for value in scenario.get("required_claim_kinds") or ()}
expected_claims = [
{
"claim_kind": str(item.get("claim_kind") or ""),
"expected": item.get("expected") if isinstance(item.get("expected"), dict) else {},
}
for item in scenario.get("required_claims") or ()
if isinstance(item, dict) and isinstance(item.get("claim_kind"), str)
]
expected_kinds = {claim["claim_kind"] for claim in expected_claims}
expected_claims.extend(
{"claim_kind": claim_kind, "expected": {}}
for claim_kind in sorted(required - expected_kinds)
)
gaps = [
{"id": str(item.get("id") or ""), "description": str(item.get("description") or "")}
for item in scenario.get("validation_capability_gaps") or ()
if isinstance(item, dict)
]
missing = [
claim for claim in expected_claims
if not any(
actual_claim["claim_kind"] == claim["claim_kind"]
and _contains_business_expected(actual_claim["expected"], claim["expected"])
for actual_claim in actual
)
]
return {
"required_claim_kinds": sorted(required),
"covered_claim_kinds": sorted(required.intersection({claim["claim_kind"] for claim in actual})),
"required_claims": expected_claims,
"missing_claims": missing,
"validation_capability_gaps": gaps,
"complete": not missing and not gaps,
}
def _contains_business_expected(actual: Any, required: Any) -> bool:
"""Match fixture business values without coupling to verifier tolerances.
Tolerances are executable verifier parameters chosen within the schema's
safe range. They are not a separate user requirement and should not make
a valid generated contract fail release evaluation merely because the
author used the registry default instead of the fixture's tighter value.
"""
if isinstance(actual, dict) and isinstance(required, dict):
return all(
key in actual
and _contains_business_expected(actual[key], value)
for key, value in required.items()
if key not in {"tolerance_mm", "tolerance"}
)
return contains_expected(actual, required)
def _contains_expected(actual: Any, required: Any) -> bool:
"""Match canonical claim values in release evaluation."""
return contains_expected(actual, required)
def _capability_block_reason(*reports: dict[str, Any]) -> tuple[str, str]:
"""Classify a failed conformance probe without confusing outages for gaps.
A provider transport failure means the probe did not establish either
support or non-support. Only a complete response that violates one of the
exposed tool contracts is evidence of a model structured-output limit.
"""
messages = [
str(failure.get("message") or "").casefold()
for report in reports
for failure in report.get("failures") or ()
if isinstance(failure, dict)
]
unavailable_markers = (
"transport unavailable",
"connection",
"network",
"timeout",
"timed out",
"temporarily unavailable",
)
if any(any(marker in message for marker in unavailable_markers) for message in messages):
return "MODEL_CAPABILITY_PROBE_UNAVAILABLE", "configuration_or_network"
return "MODEL_STRUCTURED_OUTPUT_UNSUPPORTED", "model_format_or_decision"
def _artifact_evidence_complete(
artifact_root: Path,
revisions: list[str],
active_revision: str,
artifact_manifest: dict[str, Any] | None,
ledger: list[dict[str, Any]],
) -> bool:
"""Require every published CAD decision to retain its reviewable evidence.
A non-empty report manifest is not enough: a task could otherwise report
only a rendered contract view while silently losing the STEP, topology, or
node verification used to accept a revision. Checkpoint manifests protect
immutable build inputs and outputs; the report manifest additionally
protects the frozen contract and final independent review written later.
"""
if not revisions or not active_revision:
return False
report_files = {
str(item.get("path") or "")
for item in (artifact_manifest or {}).get("files") or ()
if isinstance(item, dict)
}
required_report_files = {
"requirements-contract.json",
f"reviews/final/{active_revision}/final-review.json",
}
if artifact_manifest is not None and not required_report_files.issubset(report_files):
return False
if not all((artifact_root / path).is_file() for path in required_report_files):
return False
feature_revisions = {
str(item.get("revision_id") or "")
for item in ledger
if isinstance(item, dict) and item.get("event") == "feature_node_verified"
}
for revision_id in sorted(set(revisions)):
revision_root = artifact_root / "revisions" / revision_id
evidence_files = (
_FEATURE_NODE_EVIDENCE_FILES
if revision_id in feature_revisions
else _LEGACY_CANDIDATE_EVIDENCE_FILES
)
required_paths = {revision_root / relative for relative in evidence_files}
manifest_path = revision_root / "manifest.json"
if not manifest_path.is_file() or not all(path.is_file() for path in required_paths):
return False
try:
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError):
return False
declared = manifest.get("files") if isinstance(manifest, dict) else None
if not isinstance(declared, dict) or not evidence_files.issubset(declared):
return False
for relative, digest in declared.items():
path = revision_root / str(relative)
if not isinstance(digest, str) or len(digest) != 64 or not path.is_file():
return False
if sha256(path.read_bytes()).hexdigest() != digest:
return False
if artifact_manifest is not None:
expected_report_paths = {f"revisions/{revision_id}/{relative}" for relative in evidence_files}
if not expected_report_paths.issubset(report_files):
return False
return True
def _failure_attribution(
*,
checks: dict[str, bool],
outcome: str,
events: list[dict[str, Any]],
projection: dict[str, Any],
) -> dict[str, Any] | None:
"""Return one evidence-backed failure layer required by the test target."""
if outcome == "passed":
return None
codes: list[str] = []
for event in events:
payload = event.get("payload") if isinstance(event, dict) and isinstance(event.get("payload"), dict) else {}
result = payload.get("result") if isinstance(payload.get("result"), dict) else payload
code = result.get("code") if isinstance(result, dict) else ""
if isinstance(code, str) and code:
codes.append(code)
last_error = projection.get("last_error") if isinstance(projection, dict) else ""
if isinstance(last_error, str) and last_error:
codes.append(last_error)
for code in reversed(codes):
layer = _ERROR_FAILURE_LAYERS.get(code)
if layer:
return {
"layer": layer,
"reason_code": code,
"evidence_codes": list(dict.fromkeys(codes)),
"failed_checks": [name for name, passed in checks.items() if not passed],
}
failed_checks = [name for name, passed in checks.items() if not passed]
if outcome == "validation_capability_gap" or any(name in {"required_claims", "acceptance_contract_coverage"} for name in failed_checks):
layer, reason = "v3_contract_or_verifier", "VALIDATION_CAPABILITY_GAP"
elif any(name in {"deterministic_claims_pass", "required_operations", "required_operation_alternative"} for name in failed_checks):
layer, reason = "cdsl_expression", "CAD_ACCEPTANCE_GATE_FAILED"
elif any(name in {"token_budget", "call_budget", "author_turn_budget", "reviewer_turn_budget"} for name in failed_checks):
layer, reason = "configuration_or_network", "EVALUATION_BUDGET_EXCEEDED"
elif "raw_argument_audit" in failed_checks:
layer, reason = "v3_contract_or_verifier", "RAW_ARGUMENT_AUDIT_FAILED"
elif any(name in {"immutable_artifacts", "artifact_manifest", "required_artifact_evidence", "action_ledger"} for name in failed_checks):
layer, reason = "configuration_or_network", "ARTIFACT_EVIDENCE_INCOMPLETE"
else:
layer, reason = "model_format_or_decision", "WORKFLOW_TERMINAL_STATE_MISMATCH"
return {
"layer": layer,
"reason_code": reason,
"evidence_codes": list(dict.fromkeys(codes)),
"failed_checks": failed_checks,
}
def _run_checks(
scenario: dict[str, Any],
projection: dict[str, Any],
usage: dict[str, Any],
requirements_contract: dict[str, Any] | None,
events: list[dict[str, Any]],
artifact_root: Path,
ledger: list[dict[str, Any]] | None = None,
tool_audits: list[dict[str, Any]] | None = None,
artifact_manifest: dict[str, Any] | None = None,
) -> dict[str, bool]:
records = usage.get("records") if isinstance(usage.get("records"), list) else []
author_calls = [item for item in records if isinstance(item, dict) and item.get("role") != "reviewer"]
reviewer_calls = [item for item in records if isinstance(item, dict) and item.get("role") == "reviewer"]
total_tokens = int(usage.get("prompt_tokens") or 0) + int(usage.get("completion_tokens") or 0)
audit_ledger = ledger if ledger is not None else [item for item in projection.get("action_ledger_summary") or () if isinstance(item, dict)]
published = [
item for item in audit_ledger
if isinstance(item, dict) and item.get("event") in {"accepted", "feature_node_verified"}
]
revisions = [str(item.get("revision_id") or "") for item in published]
scheduled_atomic_ids = {
str(item.get("node_id") or ""): str(item.get("atomic_id") or "")
for item in audit_ledger
if isinstance(item, dict) and item.get("event") == "feature_node_scheduled"
}
operation_ids = {
str(
item.get("actual_atomic_id")
or item.get("atomic_id")
or scheduled_atomic_ids.get(str(item.get("node_id") or ""), "")
)
for item in published
}
required_operation_groups = [
{str(atomic_id) for atomic_id in group if isinstance(atomic_id, str) and atomic_id}
for group in scenario.get("required_any_atomic_id_groups") or ()
if isinstance(group, list)
]
active_revision = str(projection.get("active_revision") or "")
manifest = artifact_root / "revisions" / active_revision / "manifest.json"
jsonl = artifact_root / "actions" / "action-ledger.jsonl"
final_claims = next((item.get("claim_results") for item in reversed(audit_ledger) if isinstance(item, dict) and item.get("event") == "completed" and isinstance(item.get("claim_results"), list)), [])
audit_records = tool_audits if isinstance(tool_audits, list) else []
audit_valid = bool(audit_records) and all(
isinstance(item, dict)
and isinstance(item.get("raw_arguments_hash"), str)
and len(item["raw_arguments_hash"]) == 64
and item.get("canonical_schema_valid") is True
and isinstance(item.get("state_binding"), dict)
and bool(item["state_binding"].get("phase"))
and item["state_binding"].get("binding_valid") is True
and item.get("single_allowed_call") is True
for item in audit_records
)
acceptance = _acceptance_coverage(scenario, requirements_contract)
artifact_evidence = _artifact_evidence_complete(
artifact_root,
revisions,
active_revision,
artifact_manifest,
audit_ledger,
)
return {
"expected_terminal_phase": projection.get("phase") == scenario.get("expected_phase", "COMPLETED"),
"token_budget": total_tokens <= int(scenario["max_total_tokens"]),
"call_budget": len(records) <= int(scenario["max_total_calls"]),
"author_turn_budget": len(author_calls) <= int(scenario["max_author_turns"]),
"reviewer_turn_budget": len(reviewer_calls) <= int(scenario["max_reviewer_turns"]),
"required_claims": not acceptance["missing_claims"],
"acceptance_contract_coverage": acceptance["complete"],
"required_operations": {str(value) for value in scenario.get("required_atomic_ids") or ()}.issubset(operation_ids),
"required_operation_alternative": (
(not scenario.get("required_any_atomic_ids") or bool({str(value) for value in scenario["required_any_atomic_ids"]}.intersection(operation_ids)))
and all(group.intersection(operation_ids) for group in required_operation_groups)
),
"immutable_artifacts": bool(active_revision) and manifest.is_file(),
"required_artifact_evidence": artifact_evidence,
"artifact_manifest": (
artifact_manifest is None
or (
isinstance(artifact_manifest.get("files"), list)
and bool(artifact_manifest["files"])
and all(
isinstance(item, dict)
and isinstance(item.get("path"), str)
and isinstance(item.get("sha256"), str)
and len(item["sha256"]) == 64
for item in artifact_manifest["files"]
)
)
),
"action_ledger": jsonl.is_file(),
"unique_revisions": len(revisions) == len(set(revisions)),
"deterministic_claims_pass": bool(final_claims) and all(
not isinstance(item, dict) or not item.get("deterministic") or item.get("status") == "pass"
for item in final_claims
),
"raw_argument_audit": audit_valid if tool_audits is not None else bool(records) and all(isinstance(item, dict) and isinstance(item.get("raw_arguments_hash"), str) and len(item["raw_arguments_hash"]) == 64 for item in records),
"no_schema_or_decision_rejections": not _rejection_codes(events),
}
def _git_revision() -> str:
try:
completed = subprocess.run(
["git", "rev-parse", "HEAD"], cwd=BACKEND_ROOT.parent, capture_output=True,
text=True, check=True, timeout=5,
)
return completed.stdout.strip()
except (OSError, subprocess.SubprocessError):
return "unknown"
def _event_audit(events: list[dict[str, Any]]) -> list[dict[str, Any]]:
"""Persist event metadata without prompts, tool arguments, or tool facts."""
audit: list[dict[str, Any]] = []
for item in events:
payload = item.get("payload") if isinstance(item.get("payload"), dict) else {}
result = payload.get("result") if isinstance(payload.get("result"), dict) else {}
usage = payload.get("usage") if isinstance(payload.get("usage"), dict) else {}
audit.append({
"name": str(item.get("name") or ""), "tool": str(payload.get("tool") or ""),
"status": str(payload.get("status") or ""), "code": str(result.get("code") or payload.get("code") or ""),
"raw_arguments_hash": str(usage.get("raw_arguments_hash") or ""),
"field_errors": result.get("field_errors") if isinstance(result.get("field_errors"), list) else [],
})
return audit
def _report_tool_audits(audits: list[dict[str, Any]]) -> list[dict[str, Any]]:
"""Return report-safe audit metadata without argument values or prompts."""
return [
{
"audit_id": item.get("audit_id"),
"at": item.get("at"),
"actor": item.get("actor"),
"tool": item.get("tool"),
"raw_arguments_hash": item.get("raw_arguments_hash"),
"canonical_schema_valid": item.get("canonical_schema_valid"),
"field_errors": item.get("field_errors") if isinstance(item.get("field_errors"), list) else [],
"state_binding": item.get("state_binding") if isinstance(item.get("state_binding"), dict) else {},
"returned_tool": item.get("returned_tool"),
"single_allowed_call": item.get("single_allowed_call"),
}
for item in audits
if isinstance(item, dict)
]
def _ledger_identifiers(ledger: list[dict[str, Any]]) -> dict[str, list[str]]:
"""Expose audit identifiers explicitly without copying provider payloads."""
candidate_ids = sorted({
str(item.get("candidate_id"))
for item in ledger
if isinstance(item.get("candidate_id"), str) and item.get("candidate_id")
})
revision_ids = sorted({
str(item.get("revision_id"))
for item in ledger
if isinstance(item.get("revision_id"), str) and item.get("revision_id")
})
return {"candidate_ids": candidate_ids, "revision_ids": revision_ids}
def _final_claim_results(ledger: list[dict[str, Any]]) -> list[dict[str, Any]]:
for item in reversed(ledger):
results = item.get("claim_results")
if item.get("event") == "completed" and isinstance(results, list):
return [value for value in results if isinstance(value, dict)]
return []
def _redacted_correlation_ids(task_id: str, invocations: list[dict[str, Any]]) -> list[str]:
"""Keep run linkage without publishing task or provider correlation values."""
return [
canonical_hash({"task_id": task_id, "invocation_id": str(item.get("invocation_id") or "")})[:20]
for item in invocations
if isinstance(item, dict) and item.get("invocation_id")
]
def _safe_artifact_manifest(artifact_root: Path) -> dict[str, Any]:
"""Hash reviewable CAD evidence without copying source or prompt content."""
allowed_exact = {
"requirements-contract.json",
"requirements.md",
"completion-target.md",
"completion-result.md",
}
allowed_prefixes = ("actions/", "revisions/", "reviews/", "documents/requirements-")
files: list[dict[str, str]] = []
if artifact_root.is_dir():
for path in sorted(artifact_root.rglob("*")):
if not path.is_file():
continue
relative = path.relative_to(artifact_root).as_posix()
if relative not in allowed_exact and not relative.startswith(allowed_prefixes):
continue
files.append({"path": relative, "sha256": sha256(path.read_bytes()).hexdigest()})
return {"schema_version": "cad.live-eval-artifact-manifest.v1", "artifact_root": str(artifact_root), "files": files}
def _guidance_metadata(usage: dict[str, Any]) -> dict[str, Any]:
"""Summarize author-only guidance audit metadata without retaining prompts."""
records = [
item for item in usage.get("records") or ()
if isinstance(item, dict) and item.get("role") != "reviewer"
]
sections = sorted({
section_id
for item in records
for section_id in item.get("guidance_section_ids") or ()
if isinstance(section_id, str)
})
versions = sorted({
str(item.get("guidance_version") or "")
for item in records
if str(item.get("guidance_version") or "")
})
return {
"enabled": bool(records) and any(item.get("guidance_enabled") is True for item in records),
"versions": versions,
"section_ids": sections,
"chars_total": sum(int(item.get("guidance_chars") or 0) for item in records),
"fallback_reasons": sorted({
str(item.get("guidance_fallback_reason") or "")
for item in records
if str(item.get("guidance_fallback_reason") or "")
}),
}
def _has_unsupported_capability(row: dict[str, Any]) -> bool:
"""Recognize engine-declared unsupported capability without hiding model errors."""
for event in row.get("ledger") or ():
if not isinstance(event, dict):
continue
for failure in event.get("operation_failures") or ():
if isinstance(failure, dict) and "unsupported_" in str(failure.get("message") or ""):
return True
if "unsupported_" in str(event.get("message") or ""):
return True
return False
def _guidance_metric_rows(report: dict[str, Any]) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
results = [item for item in report.get("results") or () if isinstance(item, dict)]
capability_gaps = [
item for item in results
if item.get("outcome") == "validation_capability_gap" or _has_unsupported_capability(item)
]
return [item for item in results if item not in capability_gaps], capability_gaps
def _guidance_metrics(rows: list[dict[str, Any]]) -> dict[str, Any]:
if not rows:
return {"eligible_runs": 0}
author_calls = [
sum(1 for item in (row.get("usage") or {}).get("records") or () if isinstance(item, dict) and item.get("role") != "reviewer")
for row in rows
]
context_chars = [
sum(int(item.get("context_chars") or 0) for item in (row.get("usage") or {}).get("records") or () if isinstance(item, dict) and item.get("role") != "reviewer")
for row in rows
]
prompt_tokens = [
sum(int(item.get("prompt_tokens") or 0) for item in (row.get("usage") or {}).get("records") or () if isinstance(item, dict) and item.get("role") != "reviewer")
for row in rows
]
failure_layers: dict[str, int] = {}
for row in rows:
layer = str((row.get("failure_attribution") or {}).get("layer") or "passed")
failure_layers[layer] = failure_layers.get(layer, 0) + 1
def rate(predicate: Any) -> float:
return sum(1 for row in rows if predicate(row)) / len(rows)
return {
"eligible_runs": len(rows),
"executable_checkpoint_rate": rate(lambda row: bool(row.get("revision_ids"))),
"completion_rate": rate(lambda row: str((row.get("projection") or {}).get("phase") or "") == "COMPLETED"),
"deterministic_claim_success_rate": rate(lambda row: bool((row.get("checks") or {}).get("deterministic_claims_pass"))),
"schema_or_decision_rejections": sum(int(row.get("schema_rejection_count") or 0) for row in rows),
"cdsl_expression_failures": sum(1 for row in rows if str((row.get("failure_attribution") or {}).get("layer") or "") == "cdsl_expression"),
"median_author_calls": median(author_calls),
"median_author_context_chars": median(context_chars),
"total_author_prompt_tokens": sum(prompt_tokens),
"median_author_prompt_tokens": median(prompt_tokens),
"failure_layers": dict(sorted(failure_layers.items())),
}
def compare_guidance_reports(control: dict[str, Any], treatment: dict[str, Any]) -> dict[str, Any]:
"""Compare paired guidance-off/on runs without treating engine gaps as prompt results."""
control_rows, control_gaps = _guidance_metric_rows(control)
treatment_rows, treatment_gaps = _guidance_metric_rows(treatment)
control_by_key = {(str(row.get("scenario") or ""), int(row.get("repetition") or 0)): row for row in control_rows}
treatment_by_key = {(str(row.get("scenario") or ""), int(row.get("repetition") or 0)): row for row in treatment_rows}
paired = sorted(set(control_by_key).intersection(treatment_by_key))
control_only = sorted(set(control_by_key).difference(treatment_by_key))
treatment_only = sorted(set(treatment_by_key).difference(control_by_key))
control_pairs = [control_by_key[key] for key in paired]
treatment_pairs = [treatment_by_key[key] for key in paired]
control_metrics = _guidance_metrics(control_pairs)
treatment_metrics = _guidance_metrics(treatment_pairs)
same_runtime = (
control.get("author") == treatment.get("author")
and control.get("reviewer") == treatment.get("reviewer")
and control.get("runtime_profile_sha256") == treatment.get("runtime_profile_sha256")
and control.get("operation_contracts") == treatment.get("operation_contracts")
)
control_guidance = bool((control.get("author_guidance") or {}).get("enabled"))
treatment_guidance = bool((treatment.get("author_guidance") or {}).get("enabled"))
same_budgets = all(
control_by_key[key].get("scenario_budget") == treatment_by_key[key].get("scenario_budget")
for key in paired
)
calls_control = control_metrics.get("median_author_calls")
calls_treatment = treatment_metrics.get("median_author_calls")
calls_within_limit = (
isinstance(calls_control, (int, float))
and isinstance(calls_treatment, (int, float))
and calls_treatment <= calls_control * 1.10
)
improved = (
treatment_metrics.get("schema_or_decision_rejections", 0) < control_metrics.get("schema_or_decision_rejections", 0)
or treatment_metrics.get("cdsl_expression_failures", 0) < control_metrics.get("cdsl_expression_failures", 0)
)
gates = {
"complete_pairing": bool(paired) and not control_only and not treatment_only,
"control_off_treatment_on": not control_guidance and treatment_guidance,
"same_author_reviewer_runtime_and_contracts": same_runtime,
"same_per_scenario_budgets": same_budgets,
"checkpoint_rate_not_lower": treatment_metrics.get("executable_checkpoint_rate", -1) >= control_metrics.get("executable_checkpoint_rate", 0),
"completion_rate_not_lower": treatment_metrics.get("completion_rate", -1) >= control_metrics.get("completion_rate", 0),
"median_author_calls_within_ten_percent": calls_within_limit,
"model_or_cdsl_failure_improved": improved,
}
return {
"schema_version": "cad.author-guidance-comparison.v1",
"status": "passed" if all(gates.values()) else "failed",
"gates": gates,
"paired_runs": [{"scenario": scenario, "repetition": repetition} for scenario, repetition in paired],
"unpaired_runs": {
"control_only": [{"scenario": scenario, "repetition": repetition} for scenario, repetition in control_only],
"treatment_only": [{"scenario": scenario, "repetition": repetition} for scenario, repetition in treatment_only],
},
"control": control_metrics,
"treatment": treatment_metrics,
"excluded_capability_gaps": {
"control": [{"scenario": item.get("scenario"), "repetition": item.get("repetition")} for item in control_gaps],
"treatment": [{"scenario": item.get("scenario"), "repetition": item.get("repetition")} for item in treatment_gaps],
},
}
def compare_guidance_report_paths(control_path: Path, treatment_path: Path) -> dict[str, Any]:
try:
control = json.loads(control_path.read_text(encoding="utf-8"))
treatment = json.loads(treatment_path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as error:
return {"status": "failed", "error": f"GUIDANCE_COMPARISON_INPUT_INVALID: {type(error).__name__}"}
if not isinstance(control, dict) or not isinstance(treatment, dict):
return {"status": "failed", "error": "GUIDANCE_COMPARISON_INPUT_INVALID: report must be an object"}
return compare_guidance_reports(control, treatment)
async def _run(arguments: argparse.Namespace, report_root: Path) -> dict[str, Any]:
try:
scenarios = _fixture(arguments.suite, arguments.scenarios)
except ValueError as error:
return {"status": "LIVE_EVAL_BLOCKED", "error": str(error)}
repetitions = arguments.repetitions if arguments.repetitions is not None else 3 if arguments.suite == "release" else 1
if repetitions < 1:
return {"status": "LIVE_EVAL_BLOCKED", "error": "--repetitions must be at least 1"}
settings = get_settings()
if arguments.author_guidance is not None:
settings = replace(settings, agent_author_guidance_enabled=arguments.author_guidance == "on")
try:
author_provider, author_model = settings.resolve_model(arguments.author_provider, arguments.author_model)
if arguments.review_provider or arguments.review_model:
review_provider, review_model = settings.resolve_model(arguments.review_provider, arguments.review_model)
if review_provider.id == author_provider.id and review_model.id == author_model.id:
raise ValueError("Reviewer must differ from author")
else:
review_provider, review_model = settings.resolve_independent_review_model(author_provider, author_model)
except ValueError as error:
return {"status": "LIVE_EVAL_BLOCKED", "error": str(error)}
baseline_path = arguments.baseline_report.resolve() if arguments.baseline_report else None
if arguments.suite == "release" and baseline_path is None:
return {"status": "LIVE_EVAL_BLOCKED", "error": "TOKEN_BASELINE_REQUIRED: release requires a provenance-checked pre-v3 token baseline report."}
try:
baseline = load_token_baseline(baseline_path) if baseline_path else None
current_author_identity = author_request_identity(author_provider, author_model)
runtime_profile_hash = profile_sha256(settings.engine_root / "profile_schema.json")
except TokenBaselineError as error:
return {"status": "LIVE_EVAL_BLOCKED", "error": f"TOKEN_BASELINE_INVALID: {error}"}
if baseline is not None:
baseline_errors = validate_token_baseline_provenance(
baseline,
scenarios=scenarios,
author_identity=current_author_identity,
runtime_profile_hash=runtime_profile_hash,
)
if baseline_errors:
return {"status": "LIVE_EVAL_BLOCKED", "error": f"TOKEN_BASELINE_INVALID: {baseline_errors[0]}", "baseline_errors": baseline_errors}
isolated = replace(settings, task_root=report_root / "artifacts", conversation_root=report_root / "conversations")
services = compose_v3(isolated)
contracts = [
{
"atomic_id": atomic_id, "contract_hash": services.workflow.runtime.operation_contract(atomic_id)["contract_hash"],
"contract_version": services.workflow.runtime.operation_contract(atomic_id)["contract_version"],
"registry_revision": services.workflow.runtime.operation_contract(atomic_id)["registry_revision"],
}
for atomic_id in services.workflow.runtime.supported_atomic_ids()
]
verifier_schema_hash = canonical_hash(default_registry().expected_one_of_schema())
try:
author_capability = await verify_model_capability(services.repository, services.workflow.runtime, services.models, provider_id=author_provider.id, model_id=author_model.id, role="author", force=True)
reviewer_capability = await verify_model_capability(services.repository, services.workflow.runtime, services.models, provider_id=review_provider.id, model_id=review_model.id, role="reviewer", force=True)
except Exception as error:
return {"status": "LIVE_EVAL_BLOCKED", "error": str(error)[:1000]}
if not author_capability.get("supported") or not reviewer_capability.get("supported"):
error, failure_layer = _capability_block_reason(author_capability, reviewer_capability)
return {
"status": "LIVE_EVAL_BLOCKED",
"error": error,
"failure_layer": failure_layer,
"author_capability": author_capability,
"reviewer_capability": reviewer_capability,
}
results: list[dict[str, Any]] = []
for scenario in scenarios:
for repetition in range(1, repetitions + 1):
services.workflow.config = replace(
services.workflow.config,
# State transitions include local candidate recovery, so the
# loop guard is deliberately independent from the externally
# measured model-call budgets below.
max_turns=max(8, int(scenario["max_total_calls"]) * 3),
max_author_turns=int(scenario["max_author_turns"]),
max_reviewer_turns=int(scenario["max_reviewer_turns"]),
max_model_calls=int(scenario["max_total_calls"]),
)
task_id = f"cad_{secrets.token_hex(6)}"
oracle_claims = [item for item in scenario.get("required_claims") or () if isinstance(item, dict)]
if oracle_claims:
services.workflow.requirements.register_evaluation_contract_oracle(
task_id,
oracle_claims,
validation_capability_gaps=[item for item in scenario.get("validation_capability_gaps") or () if isinstance(item, dict)],
)
services.workflow.create_task(task_id, str(scenario["request"]))
events: list[dict[str, Any]] = []
started = datetime.now(timezone.utc)
try:
async with asyncio.timeout(int(scenario["max_wall_seconds"])):
async for name, payload in services.workflow.run(
task_id=task_id,
author=ModelIdentity(author_provider.id, author_model.id),
reviewer=ModelIdentity(review_provider.id, review_model.id),
):
events.append({"name": name, "payload": payload})
except TimeoutError:
events.append({"name": "timeout", "payload": {"code": "LIVE_EVAL_TIMEOUT"}})
projection = services.repository.get_task_projection(task_id) or {}
usage = services.repository.usage_summary(task_id)
ledger = services.repository.ledger_events(task_id)
invocations = services.repository.invocation_records(task_id)
tool_audits = services.repository.tool_audits(task_id)
terminal = next((item["payload"] for item in reversed(events) if item["name"] == "task_terminal"), {})
artifact_root = (isolated.task_root / task_id).resolve()
artifact_manifest = _safe_artifact_manifest(artifact_root)
finished = datetime.now(timezone.utc)
checks = _run_checks(
scenario,
projection,
usage,
services.artifacts.read_requirements_contract(
task_id,
services.repository.get_state(task_id).requirements_contract_path
if services.repository.get_state(task_id) is not None
else "",
),
events,
artifact_root,
ledger,
tool_audits,
artifact_manifest,
)
success = all(checks.values())
acceptance = _acceptance_coverage(
scenario,
services.artifacts.read_requirements_contract(
task_id,
services.repository.get_state(task_id).requirements_contract_path
if services.repository.get_state(task_id) is not None
else "",
),
)
outcome = (
"passed"
if success
else "validation_capability_gap"
if not acceptance["complete"]
and all(value for key, value in checks.items() if key != "acceptance_contract_coverage")
else "failed"
)
failure_attribution = _failure_attribution(
checks=checks,
outcome=outcome,
events=events,
projection=projection,
)
results.append({
"scenario": scenario["id"], "repetition": repetition, "success": success, "outcome": outcome,
"started_at": started.isoformat(), "finished_at": finished.isoformat(),
"duration_ms": round((finished - started).total_seconds() * 1000),
"terminal": terminal, "projection": projection, "usage": usage, "checks": checks,
"acceptance_coverage": acceptance,
"failure_attribution": failure_attribution,
"guidance": _guidance_metadata(usage),
"scenario_budget": {
"max_author_turns": int(scenario["max_author_turns"]),
"max_reviewer_turns": int(scenario["max_reviewer_turns"]),
"max_total_calls": int(scenario["max_total_calls"]),
"max_total_tokens": int(scenario["max_total_tokens"]),
},
"rejection_codes": _rejection_codes(events),
"schema_rejection_count": len(_rejection_codes(events)),
"retry_count": sum(1 for event in events if (event.get("payload") or {}).get("status") == "error"),
"rollback_count": sum(1 for item in ledger if item.get("event") == "rollback"),
**_ledger_identifiers(ledger),
"final_claim_results": _final_claim_results(ledger),
"redacted_correlation_ids": _redacted_correlation_ids(task_id, invocations),
"tool_audits": _report_tool_audits(tool_audits),
"artifact_root": str(artifact_root), "action_ledger_path": str(artifact_root / "actions" / "action-ledger.jsonl"),
"artifact_manifest": artifact_manifest,
"ledger": ledger, "invocations": invocations, "event_audit": _event_audit(events),
})
token_comparison = (
compare_token_baseline(
baseline,
scenarios=scenarios,
v3_results=results,
author_identity=current_author_identity,
runtime_profile_hash=runtime_profile_hash,
)
if baseline is not None
else {"schema_version": "cad.token-comparison.v1", "status": "not_required"}
)
token_gate = all(token_comparison.get("checks", {}).values()) if baseline is not None else True
return {
"status": "passed" if results and all(item["success"] for item in results) and token_gate else "failed",
"author": {"provider": author_provider.id, "model": author_model.id},
"author_request_identity": current_author_identity,
"reviewer": {"provider": review_provider.id, "model": review_model.id},
"author_guidance": {
"enabled": settings.agent_author_guidance_enabled,
"max_chars": settings.agent_author_guidance_max_chars,
},
"author_capability": author_capability,
"reviewer_capability": reviewer_capability,
"structured_output_mode": {
"author": str(author_capability.get("mode") or ""),
"reviewer": str(reviewer_capability.get("mode") or ""),
},
"git_revision": _git_revision(),
"protocol_version": "3.2",
"runtime_profile_sha256": runtime_profile_hash,
"operation_contracts": contracts,
"verifier_registry_version": "cad.verifier-registry.v1",
"verifier_schema_hash": verifier_schema_hash,
"token_comparison": token_comparison,
"baseline_report": str(baseline_path) if baseline_path else "",
"repetitions": repetitions,
"results": results,
}
def main() -> int:
arguments = _arguments()
report_root = BACKEND_ROOT / "live-evals" / datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
report_root.mkdir(parents=True, exist_ok=True)
try:
result = (
compare_guidance_report_paths(*arguments.compare_guidance_reports)
if arguments.compare_guidance_reports
else asyncio.run(_run(arguments, report_root))
)
except KeyboardInterrupt:
# Let an explicit operator interruption retain its normal CLI
# semantics. An external kill cannot be reported reliably either.
raise
except BaseException as error:
# A live evaluation may fail before it creates a task (for example
# during conformance). Its report is still the release gate's audit
# artifact, so an unexpected evaluator failure must not disappear.
result = {
"status": "LIVE_EVAL_BLOCKED",
"error": f"UNEXPECTED_LIVE_EVAL_ERROR: {type(error).__name__}: {str(error)[:900]}",
"failure_layer": "configuration_or_network",
}
if result["status"] == "LIVE_EVAL_BLOCKED" and arguments.allow_skip and not arguments.require_live:
result["skip_reason"] = result.get("error", "live provider access is unavailable")
result["status"] = "skipped"
result.update({"suite": arguments.suite, "require_live": arguments.require_live, "report_root": str(report_root.resolve())})
(report_root / "report.json").write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
print(json.dumps({"status": result["status"], "report": str((report_root / "report.json").resolve())}, ensure_ascii=False))
if result["status"] == "passed":
return 0
if result["status"] == "skipped":
return 0
return 2
if __name__ == "__main__":
sys.exit(main())
backend/app/cad_agent/evals/resume_one_step.py
-90
View File
@@ -1,90 +0,0 @@
"""Advance one persisted event for a task in an isolated live-evaluation root.
Long real-provider evaluations can outlive a command host's execution window.
This utility takes exactly one event from ``WorkflowCoordinator.run`` and
closes the async generator after that event has been durably handled. Repeated
invocations therefore resume the same task without re-running already
persisted nodes.
"""
from __future__ import annotations
import argparse
import asyncio
from dataclasses import replace
import json
from pathlib import Path
import sys
from typing import Any
from app.cad_agent.application.workflow import ModelIdentity
from app.cad_agent.composition import compose_v3
from app.settings import get_settings
def _arguments() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Advance one event for an isolated live CAD evaluation task.")
parser.add_argument("--report-root", required=True, type=Path)
parser.add_argument("--task-id", required=True)
parser.add_argument("--author-provider")
parser.add_argument("--author-model")
parser.add_argument("--review-provider")
parser.add_argument("--review-model")
return parser.parse_args()
async def _advance(arguments: argparse.Namespace) -> dict[str, Any]:
settings = get_settings()
author_provider, author_model = settings.resolve_model(arguments.author_provider, arguments.author_model)
if arguments.review_provider or arguments.review_model:
review_provider, review_model = settings.resolve_model(arguments.review_provider, arguments.review_model)
else:
review_provider, review_model = settings.resolve_independent_review_model(author_provider, author_model)
report_root = arguments.report_root.resolve()
services = compose_v3(replace(
settings,
task_root=report_root / "artifacts",
conversation_root=report_root / "conversations",
))
before = services.repository.get_state(arguments.task_id)
if before is None:
raise ValueError(f"Unknown task {arguments.task_id!r} in {report_root}")
runner = services.workflow.run(
task_id=arguments.task_id,
author=ModelIdentity(author_provider.id, author_model.id),
reviewer=ModelIdentity(review_provider.id, review_model.id),
)
try:
name, payload = await anext(runner)
except StopAsyncIteration:
name, payload = "workflow_exhausted", {}
finally:
await runner.aclose()
after = services.repository.get_state(arguments.task_id)
return {
"task_id": arguments.task_id,
"event": {"name": name, "payload": payload},
"before": {"phase": before.phase.value, "version": before.version},
"after": {
"phase": after.phase.value if after is not None else "",
"version": after.version if after is not None else -1,
"active_revision": after.active_revision if after is not None else "",
"last_error": after.last_error.value if after is not None and after.last_error else "",
},
}
def main() -> int:
arguments = _arguments()
try:
result = asyncio.run(_advance(arguments))
except BaseException as error:
result = {"error": f"{type(error).__name__}: {str(error)[:1000]}"}
print(json.dumps(result, ensure_ascii=False))
return 2
print(json.dumps(result, ensure_ascii=False))
return 0
if __name__ == "__main__":
sys.exit(main())
backend/app/cad_agent/evals/single_stage.py
+55
View File
@@ -0,0 +1,55 @@
"""Aggregate local results for the `cad.single-stage.v1` evaluation suite.
The evaluator accepts already-recorded task projections and never calls a
model. This keeps rate reporting reproducible and separates it from live
provider experiments.
"""
from __future__ import annotations
from typing import Any, Iterable
def summarize(records: Iterable[dict[str, Any]]) -> dict[str, Any]:
"""Return first-pass, final, compliance, and cost counters.
A record may contain ``attempts`` with ``schema_valid`` and
``executable`` booleans, final lifecycle/revision information, requirement
target states, usage records, and an elapsed duration in milliseconds.
Missing fields are reported as zero rather than guessed.
"""
values = [item for item in records if isinstance(item, dict)]
first_schema = first_executable = final_executable = 0
targets: dict[str, int] = {key: 0 for key in ("pass", "fail", "pending", "not_applicable")}
calls = prompt_tokens = completion_tokens = duration_ms = 0
for record in values:
attempts = record.get("attempts") if isinstance(record.get("attempts"), list) else []
first = attempts[0] if attempts and isinstance(attempts[0], dict) else {}
first_schema += int(bool(first.get("schema_valid")))
first_executable += int(bool(first.get("executable")))
final_executable += int(bool(record.get("published_revision") or record.get("active_revision")))
for target in record.get("requirement_targets") or ():
state = str(target.get("status") or "pending") if isinstance(target, dict) else "pending"
targets[state if state in targets else "pending"] += 1
usage = record.get("usage") if isinstance(record.get("usage"), dict) else {}
usage_records = usage.get("records") if isinstance(usage.get("records"), list) else []
calls += len(usage_records)
for item in usage_records:
if not isinstance(item, dict):
continue
prompt_tokens += int(item.get("prompt_tokens") or 0)
completion_tokens += int(item.get("completion_tokens") or 0)
duration_ms += int(record.get("duration_ms") or 0)
total = len(values)
rate = lambda count: count / total if total else 0.0
return {
"schema_version": "cad.single-stage.eval-summary.v1",
"tasks": total,
"first_pass_schema_rate": rate(first_schema),
"first_pass_executable_rate": rate(first_executable),
"final_executable_rate": rate(final_executable),
"requirement_targets": targets,
"calls": calls,
"prompt_tokens": prompt_tokens,
"completion_tokens": completion_tokens,
"duration_ms": duration_ms,
}
backend/app/cad_agent/evals/token_baseline.py
-297
View File
@@ -1,297 +0,0 @@
"""Provenance-checked token baseline comparison for the v3 release gate.
The target intentionally does not accept estimated tokens or a hand-written
percentage. A v2 run must record the same fixed requests and author request
configuration before it can be compared with a v3 release report.
"""
from __future__ import annotations
from hashlib import sha256
import json
from pathlib import Path
from statistics import median
from typing import Any
from app.settings import ProviderConfig, ProviderModel
BASELINE_SCHEMA_VERSION = "cad.token-baseline.v1"
BASELINE_PROTOCOL_VERSION = "2.0"
MINIMUM_REPETITIONS = 3
TOKEN_REDUCTION_TARGET = 0.30
MEASUREMENTS = {"prompt_tokens", "context_chars"}
class TokenBaselineError(ValueError):
"""A baseline cannot prove the release token target."""
def request_sha256(request: str) -> str:
return sha256(request.encode("utf-8")).hexdigest()
def profile_sha256(profile_path: Path) -> str:
try:
return sha256(profile_path.read_bytes()).hexdigest()
except OSError as error:
raise TokenBaselineError("Runtime profile is unavailable for token-baseline provenance.") from error
def author_request_identity(provider: ProviderConfig, model: ProviderModel) -> dict[str, Any]:
"""Capture every author setting that can affect prompt-token comparison."""
return {
"provider": provider.id,
"model": model.id,
"api_style": provider.api_style,
"reasoning_effort": provider.reasoning_effort,
# Chat Completions is explicitly deterministic in StructuredModelGateway.
# Responses models use their configured/provider-default sampling mode.
"sampling": {"temperature": 0 if provider.api_style == "chat_completions" else None},
}
def load_token_baseline(path: Path) -> dict[str, Any]:
try:
payload = json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as error:
raise TokenBaselineError("Token baseline report is missing or is not valid JSON.") from error
if not isinstance(payload, dict):
raise TokenBaselineError("Token baseline report must be a JSON object.")
if payload.get("schema_version") != BASELINE_SCHEMA_VERSION:
raise TokenBaselineError(f"Token baseline must use {BASELINE_SCHEMA_VERSION}.")
if payload.get("protocol_version") != BASELINE_PROTOCOL_VERSION:
raise TokenBaselineError("Token baseline must be a measured pre-v3 (2.0) report.")
if payload.get("measurement") not in MEASUREMENTS:
raise TokenBaselineError("Token baseline measurement must be prompt_tokens or context_chars.")
return payload
def validate_token_baseline_provenance(
baseline: dict[str, Any],
*,
scenarios: list[dict[str, Any]],
author_identity: dict[str, Any],
runtime_profile_hash: str,
) -> list[str]:
"""Reject an incomparable baseline before any billable v3 API call."""
errors: list[str] = []
if baseline.get("schema_version") != BASELINE_SCHEMA_VERSION:
errors.append(f"schema_version must equal {BASELINE_SCHEMA_VERSION}")
if baseline.get("protocol_version") != BASELINE_PROTOCOL_VERSION:
errors.append("protocol_version must equal 2.0")
if baseline.get("author") != author_identity:
errors.append("author provider/model/API mode/reasoning/sampling does not match the v3 release")
if baseline.get("runtime_profile_sha256") != runtime_profile_hash:
errors.append("runtime profile hash does not match the v3 release")
measurement = str(baseline.get("measurement") or "")
if measurement not in MEASUREMENTS:
errors.append("measurement must be prompt_tokens or context_chars")
baseline_scenarios = baseline.get("scenarios")
if not isinstance(baseline_scenarios, list):
return [*errors, "scenarios must be an array"]
by_scenario = {
str(item.get("scenario") or ""): item
for item in baseline_scenarios
if isinstance(item, dict) and item.get("scenario")
}
for scenario in scenarios:
scenario_id = str(scenario.get("id") or "")
expected_request_hash = request_sha256(str(scenario.get("request") or ""))
baseline_scenario = by_scenario.get(scenario_id)
if not isinstance(baseline_scenario, dict):
errors.append(f"missing baseline scenario {scenario_id}")
continue
if baseline_scenario.get("request_sha256") != expected_request_hash:
errors.append(f"baseline request hash differs for {scenario_id}")
repetitions = baseline_scenario.get("repetitions")
if not isinstance(repetitions, list):
errors.append(f"baseline {scenario_id} repetitions must be an array")
continue
by_repetition = {
int(item.get("repetition")): item
for item in repetitions
if isinstance(item, dict) and isinstance(item.get("repetition"), int)
}
expected_repetitions = set(range(1, MINIMUM_REPETITIONS + 1))
if not expected_repetitions.issubset(by_repetition):
errors.append(f"baseline {scenario_id} is missing repetitions 1..{MINIMUM_REPETITIONS}")
continue
for repetition in expected_repetitions:
run = by_repetition[repetition]
if _non_negative_int(run.get("author_metric")) is None or _non_negative_int(run.get("plan_review_metric")) is None:
errors.append(f"baseline {scenario_id} repetition {repetition} has invalid {measurement} metrics")
for field in ("completed", "final_review_passed", "deterministic_claims_pass"):
if not isinstance(run.get(field), bool):
errors.append(f"baseline {scenario_id} repetition {repetition} has non-boolean {field}")
return errors
def compare_token_baseline(
baseline: dict[str, Any],
*,
scenarios: list[dict[str, Any]],
v3_results: list[dict[str, Any]],
author_identity: dict[str, Any],
runtime_profile_hash: str,
) -> dict[str, Any]:
"""Return a complete, fail-closed comparison for the release report."""
errors = validate_token_baseline_provenance(
baseline, scenarios=scenarios, author_identity=author_identity,
runtime_profile_hash=runtime_profile_hash,
)
measurement = str(baseline.get("measurement") or "")
baseline_scenarios = baseline.get("scenarios")
if not isinstance(baseline_scenarios, list):
errors.append("scenarios must be an array")
baseline_scenarios = []
by_scenario = {
str(item.get("scenario") or ""): item
for item in baseline_scenarios
if isinstance(item, dict) and item.get("scenario")
}
current_by_scenario: dict[str, list[dict[str, Any]]] = {}
for result in v3_results:
if isinstance(result, dict) and isinstance(result.get("scenario"), str):
current_by_scenario.setdefault(result["scenario"], []).append(result)
baseline_tokens: list[int] = []
current_tokens: list[int] = []
baseline_completed: list[bool] = []
current_completed: list[bool] = []
baseline_final_review: list[bool] = []
current_final_review: list[bool] = []
baseline_deterministic: list[bool] = []
current_deterministic: list[bool] = []
per_scenario: list[dict[str, Any]] = []
for scenario in scenarios:
scenario_id = str(scenario.get("id") or "")
expected_request_hash = request_sha256(str(scenario.get("request") or ""))
baseline_scenario = by_scenario.get(scenario_id)
if not isinstance(baseline_scenario, dict):
errors.append(f"missing baseline scenario {scenario_id}")
continue
if baseline_scenario.get("request_sha256") != expected_request_hash:
errors.append(f"baseline request hash differs for {scenario_id}")
continue
repetitions = baseline_scenario.get("repetitions")
if not isinstance(repetitions, list) or len(repetitions) < MINIMUM_REPETITIONS:
errors.append(f"baseline {scenario_id} needs at least {MINIMUM_REPETITIONS} repetitions")
continue
current = current_by_scenario.get(scenario_id, [])
if len(current) < MINIMUM_REPETITIONS:
errors.append(f"v3 {scenario_id} needs at least {MINIMUM_REPETITIONS} repetitions")
continue
baseline_by_repetition = {
int(item.get("repetition")) : item
for item in repetitions
if isinstance(item, dict) and isinstance(item.get("repetition"), int)
}
current_by_repetition = {
int(item.get("repetition")): item
for item in current
if isinstance(item.get("repetition"), int)
}
expected_repetitions = set(range(1, MINIMUM_REPETITIONS + 1))
if not expected_repetitions.issubset(baseline_by_repetition):
errors.append(f"baseline {scenario_id} is missing repetitions 1..{MINIMUM_REPETITIONS}")
continue
if not expected_repetitions.issubset(current_by_repetition):
errors.append(f"v3 {scenario_id} is missing repetitions 1..{MINIMUM_REPETITIONS}")
continue
scenario_baseline_tokens: list[int] = []
scenario_current_tokens: list[int] = []
for repetition in sorted(expected_repetitions):
baseline_run = baseline_by_repetition[repetition]
author_tokens = _non_negative_int(baseline_run.get("author_metric"))
plan_review_tokens = _non_negative_int(baseline_run.get("plan_review_metric"))
if author_tokens is None or plan_review_tokens is None:
errors.append(f"baseline {scenario_id} repetition {repetition} has invalid {measurement} metrics")
continue
current_run = current_by_repetition[repetition]
usage = current_run.get("usage") if isinstance(current_run.get("usage"), dict) else {}
v3_author_tokens = _v3_author_metric(usage, measurement)
if v3_author_tokens is None:
errors.append(f"v3 {scenario_id} repetition {repetition} has invalid author {measurement} usage")
continue
scenario_baseline_tokens.append(author_tokens + plan_review_tokens)
scenario_current_tokens.append(v3_author_tokens)
baseline_completed.append(bool(baseline_run.get("completed")))
current_completed.append(str((current_run.get("projection") or {}).get("phase") or "") == "COMPLETED")
baseline_final_review.append(bool(baseline_run.get("final_review_passed")))
current_final_review.append(str((current_run.get("terminal") or {}).get("lifecycle") or "") == "completed")
baseline_deterministic.append(bool(baseline_run.get("deterministic_claims_pass")))
checks = current_run.get("checks") if isinstance(current_run.get("checks"), dict) else {}
current_deterministic.append(bool(checks.get("deterministic_claims_pass")))
if len(scenario_baseline_tokens) == MINIMUM_REPETITIONS and len(scenario_current_tokens) == MINIMUM_REPETITIONS:
baseline_tokens.extend(scenario_baseline_tokens)
current_tokens.extend(scenario_current_tokens)
per_scenario.append({
"scenario": scenario_id,
"baseline_median_metric": median(scenario_baseline_tokens),
"v3_median_author_metric": median(scenario_current_tokens),
})
baseline_median = median(baseline_tokens) if baseline_tokens else None
current_median = median(current_tokens) if current_tokens else None
reduction = (1 - current_median / baseline_median) if baseline_median and current_median is not None else None
baseline_valid = not errors
checks = {
"baseline_valid": baseline_valid,
"median_metric_reduction": baseline_valid and reduction is not None and reduction >= TOKEN_REDUCTION_TARGET,
"completion_rate_not_lower": baseline_valid and _rate(current_completed) >= _rate(baseline_completed) if baseline_completed and current_completed else False,
"final_review_rate_not_lower": baseline_valid and _rate(current_final_review) >= _rate(baseline_final_review) if baseline_final_review and current_final_review else False,
"deterministic_claim_rate_not_lower": baseline_valid and _rate(current_deterministic) >= _rate(baseline_deterministic) if baseline_deterministic and current_deterministic else False,
"plan_review_calls_removed": baseline_valid and all(
not str(record.get("tool") or "").startswith("modeling_plan")
for result in v3_results if isinstance(result, dict)
for record in ((result.get("usage") or {}).get("records") or [])
if isinstance(record, dict)
),
}
return {
"schema_version": "cad.token-comparison.v1",
"baseline_protocol_version": baseline.get("protocol_version"),
"measurement": measurement,
"baseline_median_metric": baseline_median,
"v3_median_author_metric": current_median,
"median_metric_reduction": reduction,
"target_median_metric_reduction": TOKEN_REDUCTION_TARGET,
"baseline_completion_rate": _rate(baseline_completed),
"v3_completion_rate": _rate(current_completed),
"baseline_final_review_rate": _rate(baseline_final_review),
"v3_final_review_rate": _rate(current_final_review),
"baseline_deterministic_claim_rate": _rate(baseline_deterministic),
"v3_deterministic_claim_rate": _rate(current_deterministic),
"per_scenario": per_scenario,
"errors": errors,
"checks": checks,
}
def _non_negative_int(value: Any) -> int | None:
return value if isinstance(value, int) and value >= 0 else None
def _v3_author_metric(usage: dict[str, Any], measurement: str) -> int | None:
records = usage.get("records")
if not isinstance(records, list):
return None
author_records = [item for item in records if isinstance(item, dict) and item.get("role") != "reviewer"]
if not author_records:
return None
if measurement == "prompt_tokens" and not all(item.get("usage_available") is True for item in author_records):
return None
field = "prompt_tokens" if measurement == "prompt_tokens" else "context_chars"
values = [
_non_negative_int(item.get(field))
for item in author_records
]
return sum(value for value in values if value is not None) if all(value is not None for value in values) else None
def _rate(values: list[bool]) -> float | None:
return sum(values) / len(values) if values else None
backend/app/cad_agent/evals/usable_smoke.py
-164
View File
@@ -1,164 +0,0 @@
"""Run real-provider CAD tasks with a usable-model success criterion.
This evaluator is intentionally narrower than the release gate in ``live.py``.
It answers one operational question: can the current workflow reliably finish
ordinary prompts with a downloadable/previewable checkpoint, even if some
acceptance claims remain best-effort warnings.
"""
from __future__ import annotations
import argparse
import asyncio
from dataclasses import replace
from datetime import datetime, timezone
import json
from pathlib import Path
import secrets
import sys
from typing import Any
from app.cad_agent.application.workflow import ModelIdentity
from app.cad_agent.composition import compose_v3
from app.settings import BACKEND_ROOT, get_settings
DEFAULT_PROMPTS = (
"生成一个 80 mm x 50 mm x 8 mm 的简单矩形板，使用毫米，输出一个单一实体。",
"生成一个简单法兰，外径 100 mm，厚度 10 mm，中间有 30 mm 通孔，使用毫米。",
"生成一个圆柱垫块，直径 60 mm，高度 20 mm，中间有 20 mm 通孔，使用毫米。",
)
def _arguments() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Run real CAD generation prompts and require usable model artifacts.")
parser.add_argument("--author-provider")
parser.add_argument("--author-model")
parser.add_argument("--review-provider")
parser.add_argument("--review-model")
parser.add_argument("--prompt", action="append", help="Prompt to run. Repeat for multiple prompts. Defaults to three simple CAD prompts.")
parser.add_argument("--max-wall-seconds", type=int, default=1200)
parser.add_argument("--max-model-calls", type=int, default=60)
return parser.parse_args()
def _result_code(events: list[dict[str, Any]], projection: dict[str, Any]) -> str:
terminal = next((item["payload"] for item in reversed(events) if item.get("name") == "task_terminal"), {})
for source in (terminal, projection):
code = source.get("code") or source.get("last_error") if isinstance(source, dict) else ""
if isinstance(code, str) and code:
return code
return ""
def _artifact_ok(artifact_root: Path, revision_id: str) -> bool:
if not revision_id:
return False
revision_root = artifact_root / "revisions" / revision_id
return all((revision_root / name).is_file() for name in ("model.step", "model.glb", "model.cdsl.json", "rebuild-report.json"))
def _classify_failure(code: str, projection: dict[str, Any]) -> str:
if str(projection.get("lifecycle") or "") == "waiting_retry":
return "configuration_or_service"
if code in {"FAILED_INTERNAL", "RUNTIME_CONTRACT_INVALID", "REQUIREMENTS_SPEC_INVALID", "STORAGE_FAILURE", "RENDER_SERVICE_UNAVAILABLE"}:
return "code_or_flow"
if code in {"AUTHOR_TRANSPORT_UNAVAILABLE", "REVIEW_SERVICE_UNAVAILABLE", "MODEL_PROTOCOL_CHECK_PENDING"}:
return "configuration_or_service"
if code in {"AUTHOR_FORMAT_INVALID", "AUTHOR_DECISION_REJECTED", "CANDIDATE_BUILD_FAILED", "CLAIM_VERIFICATION_FAILED", "CANDIDATE_REVIEW_REJECTED", "NO_PROGRESS_LIMIT", "BEST_EFFORT_COMPLETED"}:
return "model_output_or_best_effort"
return "unknown"
async def _run(arguments: argparse.Namespace, report_root: Path) -> dict[str, Any]:
settings = get_settings()
try:
author_provider, author_model = settings.resolve_model(arguments.author_provider, arguments.author_model)
if arguments.review_provider or arguments.review_model:
review_provider, review_model = settings.resolve_model(arguments.review_provider, arguments.review_model)
else:
review_provider, review_model = settings.resolve_independent_review_model(author_provider, author_model)
except ValueError as error:
return {"status": "blocked", "error": str(error), "results": []}
isolated = replace(settings, task_root=report_root / "artifacts", conversation_root=report_root / "conversations")
services = compose_v3(isolated)
services.workflow.config = replace(
services.workflow.config,
max_turns=max(8, arguments.max_model_calls * 3),
max_model_calls=arguments.max_model_calls,
)
prompts = tuple(arguments.prompt or DEFAULT_PROMPTS)
results: list[dict[str, Any]] = []
for index, prompt in enumerate(prompts, start=1):
task_id = f"cad_{secrets.token_hex(6)}"
services.workflow.create_task(task_id, prompt)
events: list[dict[str, Any]] = []
started = datetime.now(timezone.utc)
try:
async with asyncio.timeout(arguments.max_wall_seconds):
async for name, payload in services.workflow.run(
task_id=task_id,
author=ModelIdentity(author_provider.id, author_model.id),
reviewer=ModelIdentity(review_provider.id, review_model.id),
):
events.append({"name": name, "payload": payload})
except TimeoutError:
events.append({"name": "timeout", "payload": {"code": "LIVE_EVAL_TIMEOUT", "message": "Task timed out."}})
projection = services.repository.get_task_projection(task_id) or {}
artifact_root = (isolated.task_root / task_id).resolve()
revision_id = str(projection.get("active_revision") or projection.get("current_revision") or "")
code = _result_code(events, projection)
usable = str(projection.get("lifecycle") or "") == "completed" and _artifact_ok(artifact_root, revision_id)
results.append({
"index": index,
"task_id": task_id,
"prompt": prompt,
"success": usable,
"failure_layer": "" if usable else _classify_failure(code, projection),
"code": code,
"phase": projection.get("phase"),
"lifecycle": projection.get("lifecycle"),
"active_revision": revision_id,
"verification_status": projection.get("verification_status"),
"artifact_root": str(artifact_root),
"duration_ms": round((datetime.now(timezone.utc) - started).total_seconds() * 1000),
"usage": services.repository.usage_summary(task_id),
"event_audit": [
{
"name": item.get("name"),
"status": (item.get("payload") or {}).get("status"),
"lifecycle": (item.get("payload") or {}).get("lifecycle"),
"code": ((item.get("payload") or {}).get("result") or {}).get("code") if isinstance((item.get("payload") or {}).get("result"), dict) else (item.get("payload") or {}).get("code"),
"tool": (item.get("payload") or {}).get("tool"),
}
for item in events
],
})
return {
"status": "passed" if results and all(item["success"] for item in results) else "failed",
"schema_version": "cad.usable-smoke.v1",
"author": {"provider": author_provider.id, "model": author_model.id},
"reviewer": {"provider": review_provider.id, "model": review_model.id},
"results": results,
}
def main() -> int:
arguments = _arguments()
report_root = BACKEND_ROOT / "live-evals" / datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
report_root.mkdir(parents=True, exist_ok=True)
try:
result = asyncio.run(_run(arguments, report_root))
except KeyboardInterrupt:
raise
except BaseException as error:
result = {"status": "blocked", "error": f"UNEXPECTED_USABLE_SMOKE_ERROR: {type(error).__name__}: {str(error)[:1000]}", "results": []}
result.update({"report_root": str(report_root.resolve())})
report_path = report_root / "usable-smoke-report.json"
report_path.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
print(json.dumps({"status": result["status"], "report": str(report_path.resolve())}, ensure_ascii=False))
return 0 if result["status"] == "passed" else 2
if __name__ == "__main__":
sys.exit(main())