/`.
+
+| Path relative to the page run | Contents |
+| --- | --- |
+| `README.md` | Review instructions and report command. |
+| `review.csv` | Original human evidence, approved decisions, and editable LLM review columns. |
+| `snapshot.html` | Copied page HTML used for the evaluation. |
+| `run.json` | Configuration, identities, prompts, skipped prompts, filters, completion, stopping reason, format failures, usage, and estimated cost. |
+| `normalized-llm-findings.json` | LLM findings and their run-specific IDs, saved during execution. |
+| `report.html` | Generated offline page report. |
+| `artifacts/human-finding-mapping.json` | Frozen approved mapping, with checksum and attribution in the manifest. |
+| `artifacts/human-findings.json` | Canonical imported local and Global human findings. |
+| `artifacts/normalized-programmatic-findings.json` | Checker findings with run-specific IDs. |
+| `artifacts/raw-programmatic-findings.json` | Original checker output. |
+| `artifacts/raw-llm-responses.json` | Model replies, request outcomes, raw usage, billable categories, and request cost estimates. |
+| `artifacts/pricing.json` | Saved model rates, version, and capture date. |
+| `artifacts/prompts/*.json` | Exact `prompt_text` and filtered `payload_slice` for each prepared request. These contain no model replies. |
+| `artifacts/payloads/cl01_payload.json` | Extracted semantic structure before filtering. |
+| `artifacts/payloads/cl02_payload.json` | Extracted forms, fields, and labels before filtering. |
+| `artifacts/payloads/cl03_payload.json` | Extracted images, SVGs, icons, and media before filtering. |
+
+Previews have no LLM response or normalized LLM findings files. Prompt files
+show what is prepared for the model; raw response records show which requests
+actually ran. Check prompts when relevant evidence seems missing and raw
+responses when an answer is absent or malformed. The pipeline's temporary
+`artifacts/manifest.json` is removed; evaluation metadata lives in `run.json`.
+
+A full parent also contains `README.md`, `full.json`, `human-findings.json`, and
+`artifacts/human-finding-mapping.json`. `full.json` records ordered child
+membership and execution totals. Parent human findings retain unique originals
+and page assignments. Reporting adds the combined `report.html`.
+
+## Troubleshooting and compatibility
+
+| Symptom | Action |
+| --- | --- |
+| Approved mapping required | Complete mapping CSV preparation and approval, then pass its JSON path explicitly. Previews also require it. |
+| Audit checksum or page assignment mismatch | Use the same human audit file and its matching snapshot mapping. For a changed human audit file, capture and approve new inputs. Check local findings' reproduction URLs. |
+| Expected checker evidence resolves to zero or several findings | Inspect the named page, rule, and evidence. Reuse the reviewed snapshots and checker code, or prepare and approve a new mapping for changed evidence. |
+| Capture or review directory already exists | Choose a new directory. These commands preserve previous work. |
+| Approval output already exists | Choose a new output filename. Approval refuses to overwrite mappings. |
+| Live key missing | Set the selected provider's key in the environment or repository-root `.env`. |
+| Live execution declined in a script | Use `--live --approve-live` for explicit noninteractive approval. |
+| Incomplete execution or exhausted allowance | Read the stopping reason in `run.json` or `full.json`. Address it and start a fresh live invocation, which repeats paid requests. Completed children can still be reported individually. |
+| Complete run with missing LLM findings | Check `format_failures`, raw replies, skipped prompts, and prepared payload slices. Unreadable responses yield no findings but do not stop execution. |
+| Approved column changed | Restore the saved decision. To change responsibilities, approve a new mapping and create new runs. |
+| Unknown or reused finding ID | Copy IDs from the correct page and source file. Use each ID for at most one human finding in that page's review. |
+| Report disappears after failed regeneration | Correct the reported validation error and rerun reporting. The prior report is removed to prevent stale final results. |
+
+Saved runs must have the current directory layout, mapping artifact, and manifest
+fields. There is no automatic migration or compatibility fallback. Moving files
+into `artifacts/` alone cannot update every older run format. Prepare new runs
+when old evidence lacks required fields; retain earlier files as historical
+artifacts. The former `correct-all` command is removed.
+
+Earlier private experiments may contain `preflight.json`,
+`format-diagnostics.json`, or `review-validation.json`. Standard evaluator
+commands do not create those files or perform those extra experimental checks.
diff --git a/vision_aid/evaluation/__init__.py b/vision_aid/evaluation/__init__.py
new file mode 100644
index 0000000..c734d16
--- /dev/null
+++ b/vision_aid/evaluation/__init__.py
@@ -0,0 +1 @@
+"""Private model-evaluation workflows."""
diff --git a/vision_aid/evaluation/cli.py b/vision_aid/evaluation/cli.py
new file mode 100644
index 0000000..179e165
--- /dev/null
+++ b/vision_aid/evaluation/cli.py
@@ -0,0 +1,372 @@
+"""Prepare and execute fixed-model evaluations, then score human decisions."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+import uuid
+from collections.abc import Sequence
+from datetime import UTC, datetime
+from decimal import Decimal
+from pathlib import Path
+from typing import Any
+
+from dotenv import load_dotenv
+
+from vision_aid.evaluation.execution.llm_audit import (
+ DEFAULT_MODEL,
+ MODEL_PRESETS,
+ LLMRequestConfig,
+ evaluation_configuration,
+ execute,
+ positive_cost,
+)
+from vision_aid.evaluation.execution.pricing import PriceSchedule
+from vision_aid.evaluation.full import (
+ capture_snapshots,
+ execute_full_evaluation,
+ load_full_inputs,
+ prepare_full_evaluation,
+ report_full_evaluation,
+)
+from vision_aid.evaluation.human_finding_mapping import select_human_finding_mapping
+from vision_aid.evaluation.human_finding_mapping_review import (
+ approve_human_finding_mapping_review,
+ prepare_human_finding_mapping_review,
+)
+from vision_aid.evaluation.inputs.human_audit import import_page_human_findings
+from vision_aid.evaluation.page import prepare_page, report_page
+
+PROJECT_ROOT = Path(__file__).resolve().parents[2]
+DEFAULT_HUMAN_AUDIT = PROJECT_ROOT / "Pristine Accessibility Defect Report.xlsx"
+
+
+def build_parser() -> argparse.ArgumentParser:
+ """Expose the supported evaluation operations."""
+ parser = argparse.ArgumentParser(prog="visionaid-evaluate")
+ commands = parser.add_subparsers(dest="command", required=True)
+ mapping_prepare = commands.add_parser(
+ "human-finding-mapping-prepare",
+ help="prepare Human finding eligibility and Programmatic match review CSVs",
+ )
+ mapping_prepare.add_argument("--human-audit", type=Path, default=DEFAULT_HUMAN_AUDIT)
+ mapping_prepare.add_argument("--snapshots", type=Path, required=True)
+ mapping_prepare.add_argument("--output-dir", type=Path, required=True)
+ mapping_approve = commands.add_parser(
+ "human-finding-mapping-approve",
+ help="validate reviewed CSVs and export an approved Human finding mapping",
+ )
+ mapping_approve.add_argument("--human-audit", type=Path, default=DEFAULT_HUMAN_AUDIT)
+ mapping_approve.add_argument("--snapshots", type=Path, required=True)
+ mapping_approve.add_argument("--review-dir", type=Path, required=True)
+ mapping_approve.add_argument("--reviewer", required=True)
+ mapping_approve.add_argument(
+ "--output", type=Path,
+ help="new JSON filename; defaults to review-dir/human-finding-mapping.json",
+ )
+ capture = commands.add_parser(
+ "capture", help="download each Human audit URL to a saved Benchmark snapshot",
+ description=(
+ "Download listed URLs in source order, following redirects. "
+ "No model key or approval is required."
+ ),
+ )
+ capture.add_argument("--human-audit", type=Path, default=DEFAULT_HUMAN_AUDIT)
+ capture.add_argument(
+ "--output-dir", type=Path, required=True, help="new snapshot directory"
+ )
+ run = commands.add_parser(
+ "run", help="preview or execute an Evaluation run with one selected model",
+ description=(
+ "Haiku 5.5 (default), Opus 5.5 "
+ "and Sonnet 5.5 use adaptive thinking with medium effort. GPT models "
+ "use Chat Completions with medium reasoning effort. All presets omit "
+ "temperature, disable summaries, and cap output at 24,192 tokens "
+ "including reasoning/thinking. Live GPT execution requires OPENAI_API_KEY; "
+ "live Claude execution requires ANTHROPIC_API_KEY. Previews need no key."
+ ),
+ )
+ run.add_argument(
+ "--model", choices=MODEL_PRESETS, default=DEFAULT_MODEL,
+ help=f"fixed model preset (default: {DEFAULT_MODEL})",
+ )
+ run.add_argument(
+ "--human-audit", type=Path, default=DEFAULT_HUMAN_AUDIT, help="Human audit"
+ )
+ run.add_argument(
+ "--human-finding-mapping", type=Path,
+ help="approved eligibility classifications and Programmatic matches; defaults to this audit's local Human finding mapping",
+ )
+ run.add_argument("--html", type=Path, required=True, help="local Benchmark HTML")
+ run.add_argument("--source-url", required=True, help="URL represented by the HTML")
+ run.add_argument(
+ "--max-cost-usd", required=True,
+ help="positive between-request limit; a request underway can exceed it",
+ )
+ run.add_argument("--live", action="store_true", help="enable billable execution")
+ run.add_argument(
+ "--approve-live",
+ action="store_true",
+ help="explicitly approve billable execution without a prompt",
+ )
+ run_all = commands.add_parser(
+ "run-all", help="preview or execute every listed page with one shared allowance",
+ description=run.description,
+ )
+ run_all.add_argument(
+ "--model", choices=MODEL_PRESETS, default=DEFAULT_MODEL,
+ help=f"fixed model preset (default: {DEFAULT_MODEL})",
+ )
+ run_all.add_argument(
+ "--human-audit", type=Path, default=DEFAULT_HUMAN_AUDIT, help="Human audit"
+ )
+ run_all.add_argument(
+ "--human-finding-mapping", type=Path,
+ help="approved eligibility classifications and Programmatic matches; defaults to this audit's local Human finding mapping",
+ )
+ run_all.add_argument(
+ "--snapshots", type=Path, required=True,
+ help="saved snapshots.json mapping from capture; resolves HTML beside it",
+ )
+ run_all.add_argument(
+ "--max-cost-usd", required=True,
+ help=(
+ "positive total allowance shared across pages; checked between requests; "
+ "a request underway can exceed it"
+ ),
+ )
+ run_all.add_argument("--live", action="store_true", help="enable billable execution")
+ run_all.add_argument(
+ "--approve-live", action="store_true",
+ help="explicitly approve all displayed pages without a prompt; requires --live",
+ )
+ report = commands.add_parser(
+ "report", help="score completed run and edited review.csv"
+ )
+ report.add_argument("--run-dir", type=Path, required=True)
+ report.add_argument("--reviewer", required=True, help="who made the review decisions")
+ report_all = commands.add_parser(
+ "report-all", help="validate every page review and score unique Human findings",
+ description=(
+ "Final combined reporting requires every listed execution to complete "
+ "and all page reviews to validate. Each original Human finding counts "
+ "once. A Global finding earns source credit through complete detection "
+ "on any single page; testable Global classifications must agree. "
+ "Individual completed page reports remain available after parent failure."
+ ),
+ )
+ report_all.add_argument("--run-dir", type=Path, required=True, help="Full evaluation directory")
+ report_all.add_argument("--reviewer", required=True, help="who made all page review decisions")
+ return parser
+
+
+def _new_run_directory(kind: str = "runs") -> Path:
+ """Create a unique private directory under the containing worktree root."""
+ current = Path.cwd()
+ worktree = next(
+ (
+ parent
+ for parent in (current, *current.parents)
+ if (parent / ".git").exists()
+ ),
+ current,
+ )
+ root = worktree / ".model-evaluation" / kind
+ root.mkdir(parents=True, exist_ok=True)
+ run_id = datetime.now(UTC).strftime("%Y%m%dT%H%M%S") + "-" + uuid.uuid4().hex[:8]
+ run = root / run_id
+ run.mkdir()
+ return run
+
+
+def _required_human_finding_mapping(path: Path | None, audit_sha256: str) -> dict[str, Any]:
+ """Require approved decisions before preparing any new Evaluation run."""
+ mapping = select_human_finding_mapping(path, audit_sha256)
+ if mapping is None:
+ raise ValueError(
+ "New Evaluation runs require an approved Human finding mapping. Use "
+ "human-finding-mapping-prepare and human-finding-mapping-approve to review "
+ "this audit, then supply --human-finding-mapping PATH."
+ )
+ return mapping
+
+
+def _run(args: argparse.Namespace) -> int:
+ """Prepare a fresh run and execute only after explicit approval."""
+ request_config = MODEL_PRESETS[args.model]
+ configuration = evaluation_configuration(request_config)
+ limit = positive_cost(args.max_cost_usd)
+ if args.approve_live and not args.live:
+ raise ValueError("--approve-live requires --live")
+ if not args.html.is_file():
+ raise ValueError(f"Benchmark HTML not found: {args.html}")
+ human_audit = import_page_human_findings(args.human_audit, args.source_url)
+ mapping = _required_human_finding_mapping(args.human_finding_mapping, human_audit.human_audit_sha256)
+ run = _new_run_directory()
+ manifest = prepare_page(
+ run, args.html, args.source_url, human_audit, request_config, limit, mapping
+ )
+ snapshot_hash = manifest["snapshot_sha256"]
+ schedule = PriceSchedule.load(run / "artifacts" / "pricing.json")
+ print(f"Evaluation run: {run}")
+ _display_model(
+ request_config, configuration, schedule, len(manifest["prompts"]), limit
+ )
+ print(
+ f"Snapshot SHA-256: {snapshot_hash}; "
+ f"Human findings: {len(human_audit.human_findings)}"
+ )
+ print(f"Review CSV: {run / 'review.csv'}")
+ if not args.live:
+ print("Preview complete; no paid requests made.")
+ return 0
+ api_key = _approved_api_key(args, configuration)
+ if api_key is None:
+ return 0
+ complete = execute(run, manifest, api_key, request_config)
+ print(f"Run {'complete' if complete else 'incomplete'}: {run}")
+ return 0 if complete else 1
+
+
+def _approved_api_key(args: argparse.Namespace, configuration: dict[str, Any]) -> str | None:
+ """Select the provider's credentials and ask once for the displayed scope."""
+ load_dotenv(PROJECT_ROOT / ".env")
+ key_name = (
+ "OPENAI_API_KEY" if configuration["provider"] == "openai" else "ANTHROPIC_API_KEY"
+ )
+ api_key = os.getenv(key_name)
+ if not api_key:
+ raise ValueError(f"Live execution requires {key_name}")
+ if not args.approve_live and (
+ not sys.stdin.isatty()
+ or input("Type 'yes' to approve paid execution: ") != "yes"
+ ):
+ print("Live execution declined; no paid requests made.")
+ return None
+ return api_key
+
+
+def _display_model(
+ request_config: LLMRequestConfig,
+ configuration: dict[str, Any],
+ schedule: PriceSchedule,
+ prompt_count: int,
+ limit: Decimal,
+) -> None:
+ """Display fixed request settings, recorded prices and the shared guardrail."""
+ print(
+ f"Model: {request_config.model}; provider: {configuration['provider']}; "
+ f"endpoint: {configuration['endpoint']}; "
+ f"thinking: {configuration.get('thinking', {}).get('type', 'none')}; "
+ f"effort: {request_config.reasoning_effort or 'omitted'}; "
+ f"thinking budget: {request_config.thinking_budget_tokens or 'none'}; "
+ f"output cap: {request_config.max_output_tokens}"
+ )
+ print(
+ "Temperature: omitted; summaries: disabled; "
+ f"prompts: {prompt_count}"
+ )
+ print(f"Pricing: {schedule.version} ({schedule.identity})")
+ rates = schedule.data["models"][request_config.model]
+ tiers = [("", rates)]
+ if "long_context" in rates:
+ upper = rates["long_context"]
+ threshold = upper["above_input_tokens"]
+ tiers = [
+ (f"up to {threshold:,} total input tokens per request", rates),
+ (f"above {threshold:,} total input tokens per request", upper),
+ ]
+ cache_write = (
+ "5-minute cache write" if configuration["provider"] == "anthropic" else "cache write"
+ )
+ for description, tier in tiers:
+ print(
+ f"Rates per million tokens (USD){f' ({description})' if description else ''}: "
+ f"input ${tier['input_per_million_usd']}; "
+ f"output ${tier['output_per_million_usd']}; "
+ f"cache read ${tier['cached_input_per_million_usd']}; "
+ f"{cache_write} ${tier['cache_creation_input_per_million_usd']}"
+ )
+ print(
+ f"Cost guardrail: ${limit}; checked before each request; "
+ "a request underway can exceed the limit"
+ )
+
+
+def _run_all(args: argparse.Namespace) -> int:
+ """Prepare all listed pages and approve the entire scope once."""
+ request_config = MODEL_PRESETS[args.model]
+ limit = positive_cost(args.max_cost_usd)
+ if args.approve_live and not args.live:
+ raise ValueError("--approve-live requires --live")
+ audits, snapshots = load_full_inputs(args.human_audit, args.snapshots)
+ mapping = _required_human_finding_mapping(args.human_finding_mapping, next(iter(audits.values())).human_audit_sha256)
+ run = _new_run_directory("full-runs")
+ print(f"Full evaluation: {run}")
+ manifest = prepare_full_evaluation(run, audits, snapshots, request_config, limit, mapping)
+ prompt_count = 0
+ for page in manifest["pages"]:
+ child = run / page["run_dir"]
+ child_manifest = json.loads((child / "run.json").read_text())
+ prompt_count += len(child_manifest["prompts"])
+ print(f"Page: {page['source_url']}; Evaluation run: {child}")
+ print(f"Review CSV: {child / 'review.csv'}")
+ first_child = run / manifest["pages"][0]["run_dir"]
+ schedule = PriceSchedule.load(first_child / "artifacts/pricing.json")
+ _display_model(
+ request_config, manifest["configuration"], schedule, prompt_count, limit
+ )
+ if not args.live:
+ print("Preview complete; no paid requests made.")
+ return 0
+ api_key = _approved_api_key(args, manifest["configuration"])
+ if api_key is None:
+ return 0
+ complete = execute_full_evaluation(run, manifest, api_key, request_config)
+ print(f"Full evaluation {'complete' if complete else 'incomplete'}: {run}")
+ return 0 if complete else 1
+
+
+def _report(args: argparse.Namespace) -> int:
+ """Validate human decisions and report a completed run."""
+ run = args.run_dir.resolve()
+ report_page(run, args.reviewer)
+ print(f"HTML report: {run / 'report.html'}")
+ return 0
+
+
+def main(argv: Sequence[str] | None = None) -> int:
+ """Dispatch the requested CLI operation."""
+ parser = build_parser()
+ args = parser.parse_args(argv)
+ try:
+ if args.command == "human-finding-mapping-prepare":
+ prepare_human_finding_mapping_review(args.human_audit, args.snapshots, args.output_dir)
+ print(f"Human finding mapping review CSVs and instructions: {args.output_dir.resolve()}")
+ return 0
+ if args.command == "human-finding-mapping-approve":
+ mapping = approve_human_finding_mapping_review(
+ args.review_dir, args.human_audit, args.snapshots, args.reviewer, args.output,
+ )
+ print(f"Approved Human finding mapping: {mapping.resolve()}")
+ return 0
+ if args.command == "capture":
+ mapping = capture_snapshots(args.human_audit, args.output_dir)
+ print(f"Snapshot mapping: {mapping}")
+ return 0
+ if args.command == "run-all":
+ return _run_all(args)
+ if args.command == "report-all":
+ run = args.run_dir.resolve()
+ report_full_evaluation(run, args.reviewer)
+ print(f"Combined HTML report: {run / 'report.html'}")
+ return 0
+ return _run(args) if args.command == "run" else _report(args)
+ except (OSError, ValueError, KeyError) as error:
+ parser.error(str(error))
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
diff --git a/vision_aid/evaluation/execution/__init__.py b/vision_aid/evaluation/execution/__init__.py
new file mode 100644
index 0000000..c6c6131
--- /dev/null
+++ b/vision_aid/evaluation/execution/__init__.py
@@ -0,0 +1 @@
+"""Execute LLM requests, normalize findings, and estimate run cost."""
diff --git a/vision_aid/evaluation/execution/llm_audit.py b/vision_aid/evaluation/execution/llm_audit.py
new file mode 100644
index 0000000..f0c6e89
--- /dev/null
+++ b/vision_aid/evaluation/execution/llm_audit.py
@@ -0,0 +1,180 @@
+"""Fixed evaluation model presets and sequential paid execution."""
+
+from __future__ import annotations
+
+import json
+from dataclasses import asdict
+from decimal import Decimal
+from pathlib import Path
+from typing import Any
+
+from processing_scripts.llm_client.audit import (
+ AuditRequestClient as LLMRequestClient,
+)
+from processing_scripts.llm_client.audit import (
+ AuditRequestConfig as LLMRequestConfig,
+)
+from vision_aid.evaluation.execution.normalization import normalize_prompt_response
+from vision_aid.evaluation.execution.pricing import PriceSchedule
+
+DEFAULT_MODEL = "claude-haiku-5-5"
+OPENAI_MODELS = ("gpt-6-luna", "gpt-6.1-sol")
+MODEL_PRESETS = {
+ **{
+ model: LLMRequestConfig(
+ model=model,
+ adaptive_thinking=True,
+ reasoning_effort="medium",
+ max_output_tokens=24192,
+ temperature=None,
+ )
+ for model in (DEFAULT_MODEL, "claude-opus-5-5", "claude-sonnet-5-5")
+ },
+ **{
+ model: LLMRequestConfig(
+ model=model,
+ reasoning_effort="medium",
+ max_output_tokens=24192,
+ temperature=None,
+ )
+ for model in OPENAI_MODELS
+ },
+}
+USAGE_KEYS = (
+ "input_tokens",
+ "cached_input_tokens",
+ "cache_creation_input_tokens",
+ "output_tokens",
+)
+
+
+def evaluation_configuration(config: LLMRequestConfig) -> dict[str, Any]:
+ """Record the selected preset and evaluation-only request settings."""
+ provider = "openai" if config.model in OPENAI_MODELS else "anthropic"
+ metadata = config.as_metadata()
+ return {
+ "model": config.model,
+ "provider": provider,
+ "endpoint": "chat.completions" if provider == "openai" else "messages",
+ **({"thinking": metadata["thinking"]} if "thinking" in metadata else {}),
+ "thinking_budget_tokens": config.thinking_budget_tokens,
+ "reasoning_effort": config.reasoning_effort,
+ "max_output_tokens": config.max_output_tokens,
+ "temperature": "omitted",
+ "summaries": "disabled",
+ }
+
+
+def write_json(path: Path, value: object) -> None:
+ """Save one readable private evidence artifact."""
+ path.write_text(
+ json.dumps(value, indent=2, ensure_ascii=False, sort_keys=True) + "\n",
+ encoding="utf-8",
+ )
+
+
+def positive_cost(value: str) -> Decimal:
+ """Parse a strictly positive cost guardrail."""
+ try:
+ amount = Decimal(value)
+ except Exception as error:
+ raise ValueError("Cost guardrail must be a positive decimal") from error
+ if not amount.is_finite() or amount <= 0:
+ raise ValueError("Cost guardrail must be a positive decimal")
+ return amount
+
+
+def billable_usage(model: str, reported: dict[str, Any]) -> dict[str, int]:
+ """Separate OpenAI cached subsets without changing reported evidence."""
+ usage = {key: int(reported.get(key) or 0) for key in USAGE_KEYS}
+ if model in OPENAI_MODELS:
+ raw = reported.get("provider_usage") or {}
+ details = raw.get("prompt_tokens_details") or {}
+ usage["cached_input_tokens"] = int(details.get("cached_tokens") or 0)
+ usage["cache_creation_input_tokens"] = int(details.get("cache_write_tokens") or 0)
+ usage["input_tokens"] -= (
+ usage["cached_input_tokens"] + usage["cache_creation_input_tokens"]
+ )
+ return usage
+
+
+def execute(
+ run: Path, manifest: dict[str, Any], api_key: str,
+ request_config: LLMRequestConfig,
+) -> bool:
+ """Make each paid request once, saving reported usage and responses as received."""
+ artifacts = run / "artifacts"
+ schedule = PriceSchedule.load(artifacts / "pricing.json")
+ client = LLMRequestClient(
+ api_key=api_key,
+ request_config=request_config,
+ max_retries=0,
+ )
+ usage: dict[str, int] = {key: 0 for key in USAGE_KEYS}
+ estimated_cost = Decimal(0)
+ responses: list[dict[str, Any]] = []
+ findings: list[dict[str, Any]] = []
+ format_failures: list[str] = []
+ complete = True
+ reason = None
+ for prompt in manifest["prompts"]:
+ if estimated_cost >= Decimal(manifest["max_cost_usd"]):
+ complete, reason = False, "cost guardrail exhausted"
+ break
+ name = prompt["name"]
+ prompt_text = json.loads((artifacts / "prompts" / f"{name}.json").read_text())[
+ "prompt_text"
+ ]
+ result: dict[str, Any]
+ try:
+ result = client.call(prompt_text)
+ except Exception as error: # noqa: BLE001
+ # Retain an unexpected request failure as run evidence.
+ result = {
+ "success": False,
+ "response": None,
+ "usage": {},
+ "error": str(error),
+ }
+ reported: dict[str, Any] = result.get("usage") or {}
+ for key in USAGE_KEYS:
+ usage[key] += int(reported.get(key) or 0)
+ billable = billable_usage(request_config.model, reported)
+ request_cost = schedule.estimate(request_config.model, billable)
+ estimated_cost += Decimal(request_cost)
+ responses.append({
+ "prompt": name, "result": result, "billable_usage": billable,
+ "estimated_cost_usd": request_cost,
+ })
+ if result.get("success"):
+ normalized = normalize_prompt_response(
+ name,
+ str(result.get("response") or ""),
+ run_id=manifest["run_id"],
+ model=request_config.model,
+ page_url=manifest["source_url"],
+ )
+ findings.extend(asdict(finding) for finding in normalized.findings)
+ if normalized.parse_status == "malformed":
+ format_failures.append(name)
+ else:
+ complete, reason = False, str(result.get("error") or "request failed")
+ manifest.update(
+ usage=usage,
+ estimated_cost_usd=format(estimated_cost.normalize(), "f"),
+ complete=False,
+ incomplete_reason=reason,
+ format_failures=format_failures,
+ )
+ write_json(artifacts / "raw-llm-responses.json", responses)
+ write_json(run / "normalized-llm-findings.json", findings)
+ write_json(run / "run.json", manifest)
+ if not complete:
+ break
+ manifest.update(
+ complete=complete,
+ incomplete_reason=reason,
+ estimated_cost_usd=format(estimated_cost.normalize(), "f"),
+ )
+ write_json(run / "run.json", manifest)
+ return complete
diff --git a/vision_aid/evaluation/execution/normalization.py b/vision_aid/evaluation/execution/normalization.py
new file mode 100644
index 0000000..7ab336b
--- /dev/null
+++ b/vision_aid/evaluation/execution/normalization.py
@@ -0,0 +1,278 @@
+"""Lossless deterministic normalization of raw LLM outcomes."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import re
+from collections.abc import Callable, Iterable, Sequence
+from dataclasses import dataclass
+from typing import Any, NamedTuple
+
+from entry_points.generate_report import NORMALIZERS, ReportRow, safe_parse_json
+from processing_scripts.llm.registry import PROMPT_REGISTRY, PromptSpec
+from vision_aid.evaluation.schemas import CanonicalFinding
+
+_PROMPTS = {item.name: item for item in PROMPT_REGISTRY if not item.is_summary}
+_PROBLEM_KEYS = (
+ "problem",
+ "issue",
+ "issue_title",
+ "description",
+ "finding",
+ "reason",
+ "actual_result",
+ "summary",
+)
+_LOCATION_KEYS = (
+ "location",
+ "location_hint",
+ "selector",
+ "xpath",
+ "text",
+ "link_text",
+ "src",
+)
+_ELEMENT_KEYS = ("element", "element_name", "tag", "html", "selector")
+_WCAG_KEYS = ("wcag", "wcag_sc", "wcag_criteria", "criterion", "criteria")
+# Checks without a tool normalizer only reach users on screen; these fields name their element.
+_SCREEN_ONLY_ELEMENT_KEYS = ("field_id", "src", "legend", "placeholder")
+
+
+@dataclass(frozen=True)
+class NormalizationResult:
+ """Findings plus the parse outcome for one successful raw response."""
+
+ prompt: str
+ parse_status: str
+ raw_response: str
+ findings: tuple[CanonicalFinding, ...]
+ error: str | None = None
+
+
+def _first_text(item: dict[str, Any], keys: Iterable[str]) -> str:
+ """Return the first populated evidence field as stable text."""
+ for key in keys:
+ value = item.get(key)
+ if value is None or value == "":
+ continue
+ if isinstance(value, (dict, list)):
+ return json.dumps(value, ensure_ascii=False, sort_keys=True)
+ return str(value)
+ return ""
+
+
+def _wcag(item: dict[str, Any], defaults: Sequence[str]) -> tuple[str, ...]:
+ """Extract ordered WCAG identifiers or use the prompt defaults."""
+ values: list[str] = []
+ for key in _WCAG_KEYS:
+ value = item.get(key)
+ if isinstance(value, list):
+ values.extend(str(part) for part in value)
+ elif value:
+ values.append(str(value))
+ criteria = re.findall(r"\b\d+\.\d+\.\d+\b", " ".join(values))
+ return tuple(dict.fromkeys(criteria or defaults))
+
+
+class _Source(NamedTuple):
+ """One LLM finding's text, with the response part it came from."""
+
+ position: int
+ source: Any
+ problem: str
+ element: str
+ location: str
+
+
+def _issue_list(item: dict[str, Any], key: str) -> list[Any]:
+ """Copy a listed-problems field, rejecting text that would split per character."""
+ value = item.get(key)
+ if value is None:
+ return []
+ if not isinstance(value, list):
+ raise TypeError(f"Expected {key!r} to be a JSON array.")
+ return list(value)
+
+
+def _item_list(parsed: Any) -> list[dict[str, Any]]:
+ """Return an array response's items, rejecting a shape no rule can read."""
+ if not isinstance(parsed, list) or not all(isinstance(item, dict) for item in parsed):
+ raise ValueError("Expected a JSON array response.")
+ for item in parsed:
+ _issue_list(item, "issues")
+ return parsed
+
+
+def _tool_findings(
+ normalizer: Callable[..., list[ReportRow]], parsed: Any, spec: PromptSpec
+) -> list[_Source]:
+ """Read findings with the tool's report rule, pairing each row with its source."""
+ wcag = ", ".join(spec.wcag_criteria)
+ if spec.output_type == "object":
+ if not isinstance(parsed, dict):
+ raise ValueError("Expected a JSON object response.")
+ # Page-level rules write one row per issue, then one per vague heading.
+ issues = _issue_list(parsed, "issues") + _issue_list(parsed, "vague_headings")
+ rows = normalizer(parsed, wcag=wcag)
+ sources = [{"response": parsed, "issue": issue} for issue in issues]
+ paired = list(enumerate(zip(sources, rows)))
+ else:
+ # One item at a time, so each row keeps the item it came from.
+ paired = [
+ (index, (item, row))
+ for index, item in enumerate(_item_list(parsed))
+ for row in normalizer([item], wcag=wcag)
+ ]
+ return [
+ _Source(
+ position,
+ source,
+ row.actual_result or row.issue_title,
+ row.element_name,
+ row.steps_to_reproduce,
+ )
+ for position, (source, row) in paired
+ ]
+
+
+def _screen_only_findings(prompt_name: str, parsed: Any) -> list[_Source]:
+ """Read a check the tool's report skips: a finding needs a listed problem."""
+ if prompt_name == "table_semantics" and isinstance(parsed, dict):
+ # The table prompt asks for one object rather than a list.
+ parsed = [parsed]
+ findings = []
+ for index, item in enumerate(_item_list(parsed)):
+ issues = _issue_list(item, "issues")
+ if prompt_name == "table_semantics":
+ issues += _issue_list(item, "header_clarity_issues")
+ # Placeholder-only fields are the problem; that check lists no issues.
+ if not issues and prompt_name != "placeholder_as_label":
+ continue
+ problem = "; ".join(str(issue) for issue in issues)
+ findings.append(_Source(
+ index,
+ item,
+ problem or str(item.get("reason") or "") or prompt_name,
+ _first_text(item, _SCREEN_ONLY_ELEMENT_KEYS),
+ str(item.get("location_hint") or ""),
+ ))
+ return findings
+
+
+def normalize_prompt_response(
+ prompt_name: str,
+ raw_response: str,
+ *,
+ run_id: str,
+ model: str,
+ page_url: str,
+) -> NormalizationResult:
+ """Read one response with the tool's rules, without its filter or deduplication."""
+ spec = _PROMPTS.get(prompt_name)
+ if spec is None:
+ return NormalizationResult(
+ prompt=prompt_name,
+ parse_status="unknown_prompt",
+ raw_response=raw_response,
+ findings=(),
+ error=(f"No fixed evaluation normalizer exists for {prompt_name!r}."),
+ )
+ try:
+ # Parse exactly as the API's report does, so only responses it skips are unreadable.
+ parsed = safe_parse_json(raw_response)
+ except (json.JSONDecodeError, ValueError) as error:
+ return NormalizationResult(
+ prompt=prompt_name,
+ parse_status="malformed",
+ raw_response=raw_response,
+ findings=(),
+ error=str(error),
+ )
+ normalizer = NORMALIZERS.get(prompt_name)
+ try:
+ if normalizer is None:
+ sources = _screen_only_findings(prompt_name, parsed)
+ else:
+ sources = _tool_findings(normalizer, parsed, spec)
+ except Exception as error: # noqa: BLE001
+ # A shape the check's rule cannot read is unreadable, not fatal to the run.
+ return NormalizationResult(
+ prompt=prompt_name,
+ parse_status="malformed",
+ raw_response=raw_response,
+ findings=(),
+ error=str(error) or type(error).__name__,
+ )
+ findings = []
+ for position, source, problem, element, location in sources:
+ canonical_source = json.dumps(
+ source,
+ ensure_ascii=False,
+ sort_keys=True,
+ separators=(",", ":"),
+ )
+ identity = f"{run_id}\0{prompt_name}\0{position}\0{canonical_source}".encode()
+ findings.append(
+ CanonicalFinding(
+ finding_id="llm-" + hashlib.sha256(identity).hexdigest()[:20],
+ source="llm",
+ run_id=run_id,
+ model=model,
+ prompt=prompt_name,
+ checklist=spec.checklist,
+ page_url=page_url,
+ problem_family=prompt_name,
+ problem=problem,
+ element=element,
+ location=location,
+ wcag_evidence=tuple(spec.wcag_criteria),
+ raw_source=source,
+ parse_status="parsed",
+ screen_only=normalizer is None,
+ )
+ )
+ return NormalizationResult(
+ prompt=prompt_name,
+ parse_status="parsed" if findings else "empty",
+ raw_response=raw_response,
+ findings=tuple(findings),
+ )
+
+
+def normalize_programmatic_findings(
+ raw_findings: Sequence[dict[str, Any]],
+ *,
+ run_id: str,
+ page_url: str,
+) -> tuple[CanonicalFinding, ...]:
+ """Normalize checker output while retaining each complete source object."""
+ normalized = []
+ for index, item in enumerate(raw_findings):
+ canonical_item = json.dumps(
+ item,
+ ensure_ascii=False,
+ sort_keys=True,
+ separators=(",", ":"),
+ )
+ identity = f"{run_id}\0programmatic\0{index}\0{canonical_item}".encode()
+ rule = str(item.get("rule_id") or item.get("rule") or "programmatic")
+ normalized.append(
+ CanonicalFinding(
+ finding_id="programmatic-" + hashlib.sha256(identity).hexdigest()[:20],
+ source="programmatic",
+ run_id=run_id,
+ model=None,
+ prompt=rule,
+ checklist=str(item.get("checklist") or "programmatic"),
+ page_url=page_url,
+ problem_family=rule,
+ problem=_first_text(item, _PROBLEM_KEYS + ("message",)) or rule,
+ element=_first_text(item, _ELEMENT_KEYS),
+ location=_first_text(item, _LOCATION_KEYS),
+ wcag_evidence=_wcag(item, ()),
+ raw_source=item,
+ parse_status="parsed",
+ )
+ )
+ return tuple(normalized)
diff --git a/vision_aid/evaluation/execution/pricing.json b/vision_aid/evaluation/execution/pricing.json
new file mode 100644
index 0000000..170f8c4
--- /dev/null
+++ b/vision_aid/evaluation/execution/pricing.json
@@ -0,0 +1,61 @@
+{
+ "models": {
+ "claude-opus-5-5": {
+ "input_per_million_usd": "4",
+ "output_per_million_usd": "20",
+ "cached_input_per_million_usd": "0.20",
+ "cache_creation_input_per_million_usd": "5"
+ },
+ "claude-sonnet-5-5": {
+ "input_per_million_usd": "2",
+ "output_per_million_usd": "10",
+ "cached_input_per_million_usd": "0.20",
+ "cache_creation_input_per_million_usd": "2.50"
+ },
+ "gpt-6-luna": {
+ "input_per_million_usd": "0.10",
+ "output_per_million_usd": "0.50",
+ "cached_input_per_million_usd": "0.01",
+ "cache_creation_input_per_million_usd": "0.125",
+ "long_context": {
+ "above_input_tokens": 272000,
+ "input_per_million_usd": "0.20",
+ "output_per_million_usd": "0.75",
+ "cached_input_per_million_usd": "0.02",
+ "cache_creation_input_per_million_usd": "0.25"
+ }
+ },
+ "gpt-6.1-sol": {
+ "input_per_million_usd": "2.00",
+ "output_per_million_usd": "10.00",
+ "cached_input_per_million_usd": "0.10",
+ "cache_creation_input_per_million_usd": "2.50",
+ "long_context": {
+ "above_input_tokens": 272000,
+ "input_per_million_usd": "4.00",
+ "output_per_million_usd": "15.00",
+ "cached_input_per_million_usd": "0.20",
+ "cache_creation_input_per_million_usd": "5.00"
+ }
+ },
+ "claude-haiku-5-5": {
+ "input_per_million_usd": "0.10",
+ "output_per_million_usd": "0.50",
+ "cached_input_per_million_usd": "0.01",
+ "cache_creation_input_per_million_usd": "0.125",
+ "long_context": {
+ "above_input_tokens": 100000,
+ "input_per_million_usd": "0.50",
+ "output_per_million_usd": "2.50",
+ "cached_input_per_million_usd": "0.05",
+ "cache_creation_input_per_million_usd": "0.625"
+ }
+ }
+ },
+ "captured_at": "2026-10-09",
+ "sources": [
+ "https://developers.openai.com/api/docs/pricing",
+ "https://platform.claude.com/docs/en/about-claude/pricing"
+ ],
+ "version": "public-standard-2026-10-09"
+}
diff --git a/vision_aid/evaluation/execution/pricing.py b/vision_aid/evaluation/execution/pricing.py
new file mode 100644
index 0000000..d4850c4
--- /dev/null
+++ b/vision_aid/evaluation/execution/pricing.py
@@ -0,0 +1,69 @@
+"""Reproducible cost estimates from a versioned public price schedule."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from collections.abc import Mapping
+from dataclasses import dataclass
+from decimal import Decimal
+from pathlib import Path
+from typing import Any
+
+DEFAULT_SCHEDULE = Path(__file__).with_name("pricing.json")
+MILLION = Decimal(1_000_000)
+
+
+@dataclass(frozen=True)
+class PriceSchedule:
+ """Immutable model rates and their content identity."""
+
+ data: dict[str, Any]
+ identity: str
+
+ @classmethod
+ def load(cls, path: Path) -> PriceSchedule:
+ """Load and content-address one JSON price schedule."""
+ raw = path.read_bytes()
+ data = json.loads(raw)
+ if not isinstance(data, dict) or not isinstance(data.get("models"), dict):
+ raise TypeError("Price schedule must contain a models object.")
+ return cls(data=data, identity=hashlib.sha256(raw).hexdigest())
+
+ @property
+ def version(self) -> str:
+ """Return the curator-assigned price-schedule version."""
+ return str(self.data["version"])
+
+ @property
+ def price_description(self) -> str:
+ """Distinguish captured public rates from older publication metadata."""
+ if "captured_at" in self.data:
+ return f"published prices captured on {self.data['captured_at']}"
+ return f"published prices dated {self.data['published_at']}"
+
+ def estimate(self, model: str, usage: Mapping[str, int | None]) -> str:
+ """Price one request's disjoint categories at its input-size tier."""
+ try:
+ rates = self.data["models"][model]
+ except KeyError as error:
+ raise ValueError(
+ f"No versioned pricing exists for model {model!r}."
+ ) from error
+ input_tokens = Decimal(usage.get("input_tokens", 0) or 0)
+ cached_tokens = Decimal(usage.get("cached_input_tokens", 0) or 0)
+ cache_creation_tokens = Decimal(
+ usage.get("cache_creation_input_tokens", 0) or 0
+ )
+ output_tokens = Decimal(usage.get("output_tokens", 0) or 0)
+ upper = rates.get("long_context")
+ if upper and input_tokens + cached_tokens + cache_creation_tokens > upper["above_input_tokens"]:
+ rates = upper
+ cost = (
+ input_tokens * Decimal(rates["input_per_million_usd"])
+ + cached_tokens * Decimal(rates["cached_input_per_million_usd"])
+ + cache_creation_tokens
+ * Decimal(rates["cache_creation_input_per_million_usd"])
+ + output_tokens * Decimal(rates["output_per_million_usd"])
+ ) / MILLION
+ return format(cost.normalize(), "f")
diff --git a/vision_aid/evaluation/full.py b/vision_aid/evaluation/full.py
new file mode 100644
index 0000000..146cc64
--- /dev/null
+++ b/vision_aid/evaluation/full.py
@@ -0,0 +1,398 @@
+"""Coordinate Full evaluation preparation and saved Benchmark snapshots."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import shlex
+import uuid
+from decimal import Decimal
+from pathlib import Path
+from typing import Any
+
+import requests
+
+from vision_aid.evaluation.execution.llm_audit import (
+ USAGE_KEYS,
+ LLMRequestConfig,
+ evaluation_configuration,
+ execute,
+ write_json,
+)
+from vision_aid.evaluation.human_finding_mapping import (
+ load_human_finding_mapping_snapshot,
+ save_human_finding_mapping_snapshot,
+ validate_human_finding_mapping,
+)
+from vision_aid.evaluation.inputs.human_audit import (
+ import_all_human_findings,
+ list_audit_urls,
+)
+from vision_aid.evaluation.page import PageReview, prepare_page, validated_page_review
+from vision_aid.evaluation.reporting.html_report import (
+ write_full_report,
+ write_page_report,
+)
+from vision_aid.evaluation.schemas import HumanAudit
+
+
+def capture_snapshots(human_audit: Path, output_dir: Path) -> Path:
+ """Download only listed pages and publish their mapping after all succeed."""
+ urls = list_audit_urls(human_audit)
+ audit_checksum = hashlib.sha256(human_audit.read_bytes()).hexdigest()
+ output_dir.mkdir(parents=True)
+ snapshots = []
+ for number, source_url in enumerate(urls, start=1):
+ response = requests.get(source_url, allow_redirects=True)
+ response.raise_for_status()
+ html = f"page-{number:03d}.html"
+ (output_dir / html).write_bytes(response.content)
+ snapshots.append({"source_url": source_url, "html": html})
+ mapping = output_dir / "snapshots.json"
+ write_json(mapping, {
+ "human_audit_filename": human_audit.name,
+ "human_audit_sha256": audit_checksum,
+ "snapshots": snapshots,
+ })
+ return mapping
+
+
+def load_full_inputs(
+ human_audit: Path, mapping_path: Path,
+) -> tuple[dict[str, HumanAudit], dict[str, Path]]:
+ """Load listed page assignments and saved snapshots without downloading."""
+ audits = import_all_human_findings(human_audit)
+ mapping = json.loads(mapping_path.read_text(encoding="utf-8"))
+ checksum = next(iter(audits.values())).human_audit_sha256
+ if not isinstance(mapping, dict) or mapping.get("human_audit_sha256") != checksum:
+ raise ValueError(
+ "Snapshot mapping Human audit identity differs from the supplied audit"
+ )
+ entries = mapping.get("snapshots")
+ if not isinstance(entries, list) or any(
+ not isinstance(entry, dict)
+ or not isinstance(entry.get("source_url"), str)
+ or not isinstance(entry.get("html"), str)
+ for entry in entries
+ ):
+ raise ValueError(
+ "Snapshot mapping must contain source URLs and relative HTML filenames"
+ )
+ if [entry["source_url"] for entry in entries] != list(audits):
+ raise ValueError(
+ "Snapshot mapping must match every listed target once in source order"
+ )
+ snapshots = {}
+ directory = mapping_path.parent.resolve()
+ for entry in entries:
+ filename = Path(entry["html"])
+ html = (directory / filename).resolve()
+ if filename.is_absolute() or not html.is_relative_to(directory):
+ raise ValueError(
+ "Snapshot mapping HTML files must be relative files beside the mapping"
+ )
+ if not html.is_file():
+ raise ValueError(f"Snapshot mapping Benchmark HTML not found: {filename}")
+ snapshots[entry["source_url"]] = html
+ return audits, snapshots
+
+
+def prepare_full_evaluation(
+ run: Path,
+ audits: dict[str, HumanAudit],
+ snapshots: dict[str, Path],
+ request_config: LLMRequestConfig,
+ limit: Decimal,
+ mapping: dict[str, Any],
+) -> dict[str, Any]:
+ """Prepare all listed pages, retaining membership and progress on failure."""
+ first_audit = next(iter(audits.values()))
+ pages = []
+ originals = {}
+ assignments = {}
+ for number, (source_url, audit) in enumerate(audits.items(), start=1):
+ child_id = f"{run.name}-page-{number:03d}-{uuid.uuid4().hex[:8]}"
+ human_ids = []
+ for finding in audit.to_dict()["human_findings"]:
+ identity = finding["human_finding_id"]
+ originals[identity] = finding
+ human_ids.append(identity)
+ if finding["page_scope"] != "global":
+ assignments[identity] = source_url
+ pages.append({
+ "source_url": source_url,
+ "run_id": child_id,
+ "run_dir": f"pages/{child_id}",
+ "human_finding_ids": human_ids,
+ "status": "unprepared",
+ "complete": False,
+ })
+ manifest = {
+ "run_id": run.name,
+ "human_audit_filename": first_audit.human_audit_filename,
+ "human_audit_sha256": first_audit.human_audit_sha256,
+ "configuration": evaluation_configuration(request_config),
+ "max_cost_usd": str(limit),
+ "estimated_cost_usd": "0",
+ "usage": {key: 0 for key in USAGE_KEYS},
+ "complete": False,
+ "incomplete_reason": "preparing pages",
+ "pages": pages,
+ }
+ write_json(run / "human-findings.json", {
+ "human_audit_filename": first_audit.human_audit_filename,
+ "human_audit_sha256": first_audit.human_audit_sha256,
+ "human_findings": sorted(originals.values(), key=lambda item: item["source_row"]),
+ "page_assignments": assignments,
+ "source_urls": list(audits),
+ })
+ write_json(run / "full.json", manifest)
+ (run / "README.md").write_text(
+ "# Review this Full evaluation\n\n"
+ "Check `full.json` for page directories, completion, and any stopping reason. "
+ "Add `--live` to `run-all` to call the LLMs. Without `--live`, the command "
+ "runs in preview mode, generating artifacts without calling the LLMs.\n\n"
+ "1. After every page completes, open each directory under `pages/`. Follow its "
+ "README to review LLM matches and save `review.csv`. Review Global findings on "
+ "every page. Keep approved classifications and Programmatic fields unchanged.\n"
+ "2. Generate all page reports and the combined report from the repository root:\n\n"
+ "```bash\n"
+ f"uv run visionaid-evaluate report-all --run-dir {shlex.quote(str(run.resolve()))} "
+ '--reviewer "Your name and review method"\n'
+ "```\n\n"
+ "Open this directory's `report.html` for combined results. Each page also has "
+ "its own `report.html`. Incomplete runs cannot produce a combined report.\n",
+ encoding="utf-8",
+ )
+ save_full_human_finding_mapping(run, mapping)
+ manifest = json.loads((run / "full.json").read_text())
+ pages = manifest["pages"]
+ for page in pages:
+ child = run / page["run_dir"]
+ try:
+ child.mkdir(parents=True)
+ child_manifest = prepare_page(
+ child, snapshots[page["source_url"]], page["source_url"],
+ audits[page["source_url"]], request_config, limit, mapping,
+ )
+ except (OSError, ValueError, KeyError) as error:
+ page["status"] = "preparation_failed"
+ manifest["incomplete_reason"] = (
+ f"page preparation failed for {page['source_url']}: {error}"
+ )
+ write_json(run / "full.json", manifest)
+ raise
+ page.update(
+ status="prepared", snapshot_sha256=child_manifest["snapshot_sha256"]
+ )
+ write_json(run / "full.json", manifest)
+ manifest["incomplete_reason"] = "preview"
+ write_json(run / "full.json", manifest)
+ return manifest
+
+
+def save_full_human_finding_mapping(run: Path, mapping: dict[str, Any]) -> None:
+ """Pin the approved source shared by all children during preparation."""
+ manifest = json.loads((run / "full.json").read_text())
+ validate_human_finding_mapping(mapping, manifest["human_audit_sha256"])
+ original = json.loads((run / "human-findings.json").read_text())
+ if {item["human_finding_id"] for item in mapping["decisions"]} != {
+ item["human_finding_id"] for item in original["human_findings"]
+ }:
+ raise ValueError("Full evaluation Human finding mapping must contain each original Human finding exactly once")
+ save_human_finding_mapping_snapshot(run, manifest, mapping)
+ write_json(run / "full.json", manifest)
+
+
+def execute_full_evaluation(
+ run: Path, manifest: dict[str, Any], api_key: str,
+ request_config: LLMRequestConfig,
+) -> bool:
+ """Execute prepared pages sequentially with one remaining cost allowance."""
+ total_cost = Decimal(0)
+ usage = {key: 0 for key in USAGE_KEYS}
+ limit = Decimal(manifest["max_cost_usd"])
+ manifest["incomplete_reason"] = None
+ for number, page in enumerate(manifest["pages"]):
+ remaining = limit - total_cost
+ if remaining <= 0:
+ manifest["incomplete_reason"] = "cost guardrail exhausted before next page"
+ for unfinished in manifest["pages"][number:]:
+ unfinished["status"] = "not_executed"
+ write_json(run / "full.json", manifest)
+ return False
+ child = run / page["run_dir"]
+ child_manifest = json.loads((child / "run.json").read_text())
+ child_manifest["max_cost_usd"] = str(remaining)
+ write_json(child / "run.json", child_manifest)
+ try:
+ complete = execute(child, child_manifest, api_key, request_config)
+ except Exception as error: # noqa: BLE001
+ # Provider setup can fail before the executor saves a request outcome.
+ child_manifest.update(complete=False, incomplete_reason=str(error))
+ write_json(child / "run.json", child_manifest)
+ complete = False
+ total_cost += Decimal(child_manifest["estimated_cost_usd"])
+ for key in USAGE_KEYS:
+ usage[key] += child_manifest["usage"][key]
+ page.update(status="complete" if complete else "incomplete", complete=complete)
+ manifest.update(estimated_cost_usd=str(total_cost), usage=usage)
+ if not complete:
+ manifest["incomplete_reason"] = (
+ f"page execution failed for {page['source_url']}: "
+ f"{child_manifest['incomplete_reason']}"
+ )
+ for unfinished in manifest["pages"][number + 1:]:
+ unfinished["status"] = "not_executed"
+ write_json(run / "full.json", manifest)
+ if not complete:
+ return False
+ manifest.update(complete=True, incomplete_reason=None)
+ write_json(run / "full.json", manifest)
+ return True
+
+
+def _pooled_rows(
+ originals: list[dict[str, Any]],
+ pages: list[dict[str, Any]],
+ reviews: list[PageReview],
+) -> list[dict[str, Any]]:
+ """Keep each original once, using complete source matches on any one page."""
+ grouped: dict[str, list[dict[str, Any]]] = {}
+ for page, review in zip(pages, reviews, strict=True):
+ for row in review.rows:
+ grouped.setdefault(row["human_finding"]["human_finding_id"], []).append({
+ "source_url": page["source_url"],
+ "decision": row["decision"],
+ "matches": row["matches"],
+ })
+ pooled = []
+ for finding in originals:
+ page_rows = grouped[finding["human_finding_id"]]
+ classifications = {
+ row["decision"]["classification"].strip() for row in page_rows
+ if row["decision"]["classification"].strip() in {"llm_eligible", "programmatic"}
+ }
+ if len(classifications) > 1:
+ raise ValueError(
+ f"Global Human finding {finding['human_finding_id']} has conflicting "
+ "testable classifications; all testable pages must agree"
+ )
+ classification = (
+ next(iter(classifications)) if classifications
+ else page_rows[0]["decision"]["classification"].strip()
+ )
+ pooled.append({
+ "human_finding": finding,
+ "decision": {"classification": classification},
+ "matches": {
+ kind: tuple(identity for row in page_rows for identity in row["matches"][kind])
+ for kind in ("llm", "programmatic")
+ },
+ "page_reviews": page_rows,
+ })
+ return pooled
+
+
+def _validated_full_reviews(
+ run: Path, manifest: dict[str, Any], original: dict[str, Any],
+) -> list[PageReview]:
+ """Check ordered membership, original evidence and ownership for every child."""
+ if manifest["run_id"] != run.name:
+ raise ValueError("Full evaluation evidence does not belong to this directory")
+ if not manifest["complete"]:
+ raise ValueError("Incomplete Full evaluations cannot produce a final report")
+ urls = original["source_urls"]
+ if not urls or len(urls) != len(set(urls)) or [
+ page["source_url"] for page in manifest["pages"]
+ ] != urls:
+ raise ValueError("Full evaluation membership differs from the original ordered targets")
+ if original["human_audit_sha256"] != manifest["human_audit_sha256"]:
+ raise ValueError("Human audit identity differs from the Full evaluation")
+ findings = original["human_findings"]
+ identities = [item["human_finding_id"] for item in findings]
+ if not identities or len(identities) != len(set(identities)):
+ raise ValueError("Full evaluation must preserve each original Human finding once")
+ assignments = original["page_assignments"]
+ locals_ = {item["human_finding_id"] for item in findings if item["page_scope"] != "global"}
+ if set(assignments) != locals_ or any(url not in urls for url in assignments.values()):
+ raise ValueError("Full evaluation has invalid original page assignments")
+ mapping = load_human_finding_mapping_snapshot(run, manifest, "Saved Full evaluation Human finding mapping")
+ mapping_digest = manifest["human_finding_mapping_sha256"]
+ if {item["human_finding_id"] for item in mapping["decisions"]} != set(identities):
+ raise ValueError("Full evaluation Human finding mapping differs from original Human findings")
+ reviews = []
+ child_ids: set[str] = set()
+ tool_ids: set[str] = set()
+ for page in manifest["pages"]:
+ child = (run / page["run_dir"]).resolve()
+ if (
+ page["run_id"] in child_ids
+ or page["run_dir"] != f"pages/{page['run_id']}"
+ or child.parent != run / "pages"
+ or child.name != page["run_id"]
+ ):
+ raise ValueError("Child Evaluation run identity or directory ownership differs")
+ child_ids.add(page["run_id"])
+ if not page["complete"] or page["status"] != "complete":
+ raise ValueError("Every listed page must complete execution before combined reporting")
+ expected = [
+ item for item in findings if item["page_scope"] == "global"
+ or assignments[item["human_finding_id"]] == page["source_url"]
+ ]
+ if page["human_finding_ids"] != [item["human_finding_id"] for item in expected]:
+ raise ValueError("Child Human finding membership differs from its original assignments")
+ evidence = json.loads((child / "artifacts/human-findings.json").read_text())
+ if evidence["human_findings"] != expected or evidence["homepage_url"] != page["source_url"]:
+ raise ValueError("Child Human finding evidence differs from the original page selection")
+ review = validated_page_review(child)
+ saved = review.manifest
+ if saved.get("human_finding_mapping_sha256") != mapping_digest:
+ raise ValueError(f"Child Human finding mapping differs from the Full evaluation: {page['source_url']}")
+ for field, expected_value in (
+ ("run_id", page["run_id"]), ("source_url", page["source_url"]),
+ ("human_audit_sha256", manifest["human_audit_sha256"]),
+ ("configuration", manifest["configuration"]),
+ ("snapshot_sha256", page["snapshot_sha256"]),
+ ):
+ if saved[field] != expected_value:
+ raise ValueError(f"Child {field} differs from the Full evaluation")
+ if hashlib.sha256((child / "snapshot.html").read_bytes()).hexdigest() != saved["snapshot_sha256"]:
+ raise ValueError("Child Benchmark snapshot checksum differs from the saved evidence")
+ if review.schedule.identity != saved["pricing_sha256"] or review.schedule.version != saved["pricing_version"]:
+ raise ValueError("Child pricing evidence differs from its recorded schedule")
+ for filename in ("raw-llm-responses.json", "raw-programmatic-findings.json"):
+ json.loads((child / "artifacts" / filename).read_text())
+ for prompt in saved["prompts"]:
+ json.loads((child / "artifacts/prompts" / f"{prompt['name']}.json").read_text())
+ if any(item["model"] != manifest["configuration"]["model"] for item in review.llm):
+ raise ValueError("LLM findings do not belong to the selected model")
+ ids = [item["finding_id"] for item in review.llm + review.programmatic]
+ if len(ids) != len(set(ids)) or tool_ids.intersection(ids):
+ raise ValueError("Tool finding identities must be unique across all page runs")
+ tool_ids.update(ids)
+ reviews.append(review)
+ cost = sum((Decimal(review.manifest["estimated_cost_usd"]) for review in reviews), Decimal(0))
+ usage = {key: sum(review.manifest["usage"][key] for review in reviews) for key in USAGE_KEYS}
+ if cost != Decimal(manifest["estimated_cost_usd"]) or usage != manifest["usage"]:
+ raise ValueError("Full evaluation totals differ from authoritative child cost and usage")
+ return reviews
+
+
+def report_full_evaluation(run: Path, reviewer: str) -> None:
+ """Validate every page review before writing page and combined reports."""
+ if not reviewer.strip():
+ raise ValueError("A reviewer description is required")
+ run = run.resolve()
+ # Regeneration cannot leave a previous final report beside invalid decisions.
+ (run / "report.html").unlink(missing_ok=True)
+ (run / "report.md").unlink(missing_ok=True)
+ manifest = json.loads((run / "full.json").read_text())
+ original = json.loads((run / "human-findings.json").read_text())
+ reviews = _validated_full_reviews(run, manifest, original)
+ rows = _pooled_rows(original["human_findings"], manifest["pages"], reviews)
+ for page, review in zip(manifest["pages"], reviews, strict=True):
+ child = run / page["run_dir"]
+ write_page_report(child / "report.html", review, reviewer)
+ (child / "report.md").unlink(missing_ok=True)
+ write_full_report(run / "report.html", manifest, rows, reviews, reviewer)
diff --git a/vision_aid/evaluation/human_finding_mapping.py b/vision_aid/evaluation/human_finding_mapping.py
new file mode 100644
index 0000000..0f20d7a
--- /dev/null
+++ b/vision_aid/evaluation/human_finding_mapping.py
@@ -0,0 +1,218 @@
+"""Approved audit decisions, resolved against each page's actual checker evidence."""
+
+from __future__ import annotations
+
+import csv
+import hashlib
+import json
+from pathlib import Path
+from typing import Any
+
+from vision_aid.evaluation.inputs.human_audit import ELIGIBILITY_VALUES
+
+HUMAN_FINDING_MAPPING_DIRECTORY = Path(__file__).with_name("human_finding_mappings")
+HUMAN_FINDING_MAPPING_ARTIFACT = "artifacts/human-finding-mapping.json"
+LOCKED_COLUMNS = (
+ "classification", "classification_reason", "programmatic_finding_id",
+ "programmatic_match_reason",
+)
+
+
+def _text(value: Any) -> bool:
+ """Identify populated human-authored text without coercing invalid values."""
+ return isinstance(value, str) and bool(value.strip())
+
+
+def validate_human_finding_mapping(mapping: dict[str, Any], audit_sha256: str) -> None:
+ """Reject unrelated audits and incomplete or conflicting approved decisions."""
+ if not isinstance(mapping, dict) or mapping.get("schema_version") != 1:
+ raise ValueError("Human finding mapping must use schema_version 1")
+ if mapping.get("human_audit_sha256") != audit_sha256:
+ raise ValueError("Human finding mapping Human audit identity differs from the run")
+ if not _text(mapping.get("approved_by")):
+ raise ValueError("Human finding mapping needs approved_by attribution")
+ decisions = mapping.get("decisions")
+ if not isinstance(decisions, list) or not decisions:
+ raise ValueError("Human finding mapping needs approved decisions")
+ seen = set()
+ for decision in decisions:
+ if not isinstance(decision, dict):
+ raise ValueError("Human finding mapping decision must be an object") # noqa: TRY004 - CLI input validation
+ identity = decision.get("human_finding_id")
+ if not _text(identity) or identity in seen:
+ raise ValueError(f"Human finding mapping has unknown or repeated Human finding: {identity}")
+ seen.add(identity)
+ prefix = f"Human finding mapping Human finding {identity}"
+ if decision.get("classification") not in ELIGIBILITY_VALUES:
+ raise ValueError(f"{prefix}: invalid classification")
+ if not _text(decision.get("classification_reason")):
+ raise ValueError(f"{prefix}: needs approved classification_reason")
+ pages = decision.get("programmatic_by_page")
+ if not isinstance(pages, dict) or not pages:
+ raise ValueError(f"{prefix}: needs explicit Programmatic decisions per page")
+ for url, page in pages.items():
+ if not _text(url) or not isinstance(page, dict):
+ raise ValueError(f"{prefix}: invalid Programmatic page decision")
+ expectations = page.get("expectations")
+ if not isinstance(expectations, list) or not _text(page.get("reason")):
+ raise ValueError(f"{prefix}, {url}: needs expectations and approved reason")
+ classification = decision["classification"]
+ if classification == "programmatic" and not expectations:
+ raise ValueError(f"{prefix}, {url}: Programmatic responsibility requires expected coverage")
+ if classification in {"unavailable_evidence", "ambiguous"} and expectations:
+ raise ValueError(f"{prefix}, {url}: not testable; cannot claim Programmatic coverage")
+ for expectation in expectations:
+ if not isinstance(expectation, dict) or not _text(expectation.get("rule")):
+ raise ValueError(f"{prefix}, {url}: expectation needs an existing rule and evidence")
+ evidence = expectation.get("evidence")
+ if (
+ not isinstance(evidence, dict)
+ or not evidence
+ or set(evidence) - {"element", "location", "problem", "raw_source"}
+ or not any(value for value in evidence.values())
+ ):
+ raise ValueError(f"{prefix}, {url}: expectation needs relevant element/page evidence")
+ for field, value in evidence.items():
+ if field == "raw_source":
+ if not isinstance(value, dict) or not value:
+ raise ValueError(f"{prefix}, {url}: raw_source evidence must be a nonempty object")
+ elif not isinstance(value, str):
+ raise ValueError(f"{prefix}, {url}: {field} evidence must be text")
+
+
+def load_human_finding_mapping(path: Path, audit_sha256: str) -> dict[str, Any]:
+ """Read explicitly selected approved decisions for exactly this Human audit."""
+ mapping = json.loads(path.read_text(encoding="utf-8"))
+ validate_human_finding_mapping(mapping, audit_sha256)
+ return mapping
+
+
+def select_human_finding_mapping(path: Path | None, audit_sha256: str) -> dict[str, Any] | None:
+ """Select explicit decisions or a locally stored Human finding mapping for this audit checksum."""
+ if path is None:
+ path = HUMAN_FINDING_MAPPING_DIRECTORY / f"{audit_sha256}.json"
+ if not path.is_file():
+ return None
+ return load_human_finding_mapping(path, audit_sha256)
+
+
+def _contains(actual: Any, expected: Any) -> bool:
+ """Match exact evidence values, allowing a selected subset of raw object fields."""
+ if isinstance(expected, dict):
+ return isinstance(actual, dict) and all(
+ key in actual and _contains(actual[key], value)
+ for key, value in expected.items()
+ )
+ return actual == expected
+
+
+def resolved_decisions(
+ mapping: dict[str, Any], human_audit: dict[str, Any], source_url: str,
+ programmatic: list[dict[str, Any]],
+) -> dict[str, dict[str, str]]:
+ """Resolve each approved complete match uniquely without run-dependent keys."""
+ validate_human_finding_mapping(mapping, human_audit["human_audit_sha256"])
+ decisions = {item["human_finding_id"]: item for item in mapping["decisions"]}
+ resolved = {}
+ used = set()
+ for human in human_audit["human_findings"]:
+ identity = human["human_finding_id"]
+ prefix = f"Human finding mapping Human finding {identity}, {source_url}"
+ if identity not in decisions:
+ raise ValueError(f"{prefix}: missing approved decision")
+ decision = decisions[identity]
+ if source_url not in decision["programmatic_by_page"]:
+ raise ValueError(f"{prefix}: missing explicit Programmatic page decision")
+ page = decision["programmatic_by_page"][source_url]
+ ids = []
+ for expectation in page["expectations"]:
+ candidates = [
+ item for item in programmatic
+ if item["problem_family"] == expectation["rule"]
+ and _contains(item, expectation["evidence"])
+ ]
+ if len(candidates) != 1:
+ raise ValueError(
+ f"{prefix}: expected Programmatic rule {expectation['rule']} "
+ f"with evidence {json.dumps(expectation['evidence'], ensure_ascii=False, sort_keys=True)} "
+ f"resolved to {len(candidates)} findings; expected exactly one. "
+ "Inspect the saved checker evidence and approved expectation."
+ )
+ finding_id = candidates[0]["finding_id"]
+ if finding_id in used:
+ raise ValueError(f"{prefix}: conflicting Programmatic finding ID reuse: {finding_id}")
+ used.add(finding_id)
+ ids.append(finding_id)
+ resolved[identity] = {
+ "classification": decision["classification"],
+ "classification_reason": decision["classification_reason"],
+ "programmatic_finding_id": ";".join(ids),
+ "programmatic_match_reason": page["reason"],
+ }
+ return resolved
+
+
+def save_human_finding_mapping_snapshot(
+ run: Path, manifest: dict[str, Any], mapping: dict[str, Any],
+) -> None:
+ """Save approved decisions and pin their bytes and attribution in a manifest."""
+ path = run / HUMAN_FINDING_MAPPING_ARTIFACT
+ path.parent.mkdir(exist_ok=True)
+ path.write_text(json.dumps(mapping, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+ manifest["human_finding_mapping_sha256"] = hashlib.sha256(path.read_bytes()).hexdigest()
+ manifest["human_finding_mapping_approved_by"] = mapping["approved_by"]
+
+
+def load_human_finding_mapping_snapshot(
+ run: Path, manifest: dict[str, Any], label: str = "Saved Human finding mapping",
+) -> dict[str, Any]:
+ """Load a pinned approved source only after verifying its bytes and attribution."""
+ path = run / HUMAN_FINDING_MAPPING_ARTIFACT
+ digest = manifest.get("human_finding_mapping_sha256")
+ if not digest:
+ raise ValueError(f"{label} is missing its run identity; prepare a new run with --human-finding-mapping PATH")
+ if not path.is_file() or hashlib.sha256(path.read_bytes()).hexdigest() != digest:
+ raise ValueError(f"{label} is missing or its checksum changed")
+ mapping = load_human_finding_mapping(path, manifest["human_audit_sha256"])
+ if manifest.get("human_finding_mapping_approved_by") != mapping["approved_by"]:
+ raise ValueError(f"{label} approval attribution changed")
+ return mapping
+
+
+def apply_human_finding_mapping(run: Path, mapping: dict[str, Any]) -> None:
+ """Populate fixed decisions and save their source, preserving per-run LLM review."""
+ manifest = json.loads((run / "run.json").read_text())
+ human = json.loads((run / "artifacts/human-findings.json").read_text())
+ programmatic = json.loads((run / "artifacts/normalized-programmatic-findings.json").read_text())
+ if human["human_audit_sha256"] != manifest["human_audit_sha256"]:
+ raise ValueError("Human audit identity differs from the run")
+ decisions = resolved_decisions(mapping, human, manifest["source_url"], programmatic)
+ path = run / "review.csv"
+ with path.open(encoding="utf-8-sig", newline="") as stream:
+ reader = csv.DictReader(stream)
+ columns = list(reader.fieldnames or ())
+ rows = list(reader)
+ if (
+ len(rows) != len(decisions)
+ or {row.get("human_finding_id") for row in rows} != set(decisions)
+ ):
+ raise ValueError("Review CSV must contain each Human finding exactly once")
+ if "programmatic_match_reason" not in columns:
+ columns.append("programmatic_match_reason")
+ for row in rows:
+ row.update(decisions[row["human_finding_id"]])
+ with path.open("w", encoding="utf-8", newline="") as stream:
+ writer = csv.DictWriter(stream, fieldnames=columns)
+ writer.writeheader()
+ writer.writerows(rows)
+ save_human_finding_mapping_snapshot(run, manifest, mapping)
+ (run / "run.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+
+
+def saved_human_finding_mapping_decisions(
+ run: Path, manifest: dict[str, Any], human: dict[str, Any],
+ programmatic: list[dict[str, Any]],
+) -> dict[str, dict[str, str]]:
+ """Load the run's required frozen decisions and verify their saved identity."""
+ mapping = load_human_finding_mapping_snapshot(run, manifest)
+ return resolved_decisions(mapping, human, manifest["source_url"], programmatic)
diff --git a/vision_aid/evaluation/human_finding_mapping_review.py b/vision_aid/evaluation/human_finding_mapping_review.py
new file mode 100644
index 0000000..5bc2f16
--- /dev/null
+++ b/vision_aid/evaluation/human_finding_mapping_review.py
@@ -0,0 +1,316 @@
+"""Prepare readable audit decisions and export an approved Human finding mapping."""
+
+from __future__ import annotations
+
+import csv
+import json
+from dataclasses import asdict
+from pathlib import Path
+from typing import Any
+
+from processing_scripts.programmatic.forms_checklist_02 import audit_forms
+from processing_scripts.programmatic.nontext_checklist_03 import audit_nontext
+from processing_scripts.programmatic.semantic_checklist_01 import audit_html_file
+from vision_aid.evaluation.execution.normalization import (
+ normalize_programmatic_findings,
+)
+from vision_aid.evaluation.full import load_full_inputs
+from vision_aid.evaluation.human_finding_mapping import (
+ resolved_decisions,
+ validate_human_finding_mapping,
+)
+from vision_aid.evaluation.inputs.human_audit import ELIGIBILITY_VALUES, HEADERS
+
+IDENTITY_COLUMNS = (
+ "human_finding_id",
+ "source_sheet",
+ "source_row",
+ "page_scope",
+ "source_element",
+ *HEADERS,
+)
+CLASSIFICATION_COLUMNS = (*IDENTITY_COLUMNS, "classification", "classification_reason")
+MATCH_COLUMNS = (
+ "source_url",
+ *IDENTITY_COLUMNS,
+ "programmatic_finding_refs",
+ "programmatic_reason",
+)
+FINDING_COLUMNS = (
+ "finding_ref",
+ "source_url",
+ "rule",
+ "problem",
+ "element",
+ "text",
+ "target",
+ "location",
+)
+
+
+def _write_csv(
+ path: Path, columns: tuple[str, ...], rows: list[dict[str, Any]]
+) -> None:
+ """Write a UTF-8 CSV that Excel can open with multiline evidence intact."""
+ with path.open("w", encoding="utf-8-sig", newline="") as stream:
+ writer = csv.DictWriter(stream, fieldnames=columns, restval="")
+ writer.writeheader()
+ writer.writerows(rows)
+
+
+def _human_columns(finding: dict[str, Any]) -> dict[str, Any]:
+ """Flatten original evidence beside generated identities for human review."""
+ return {
+ key: finding[key] for key in IDENTITY_COLUMNS if key not in HEADERS
+ } | finding["raw_evidence"]
+
+
+def _review_sources(
+ human_audit: Path, snapshots: Path,
+) -> tuple[dict[str, Any], list[dict[str, Any]]]:
+ """Rebuild original findings and the checker catalog from the supplied saved inputs."""
+ audits, html_files = load_full_inputs(human_audit, snapshots)
+ originals = {}
+ pages = []
+ catalog = []
+ for number, (url, audit) in enumerate(audits.items(), start=1):
+ saved = html_files[url]
+ raw = (
+ audit_html_file(str(saved))
+ + audit_forms(str(saved))
+ + audit_nontext(str(saved))
+ )
+ findings = normalize_programmatic_findings(
+ raw, run_id="human-finding-mapping-review", page_url=url
+ )
+ evidence = {}
+ for position, item in enumerate(findings, start=1):
+ ref = f"P{number:03d}-F{position:03d}"
+ finding = asdict(item)
+ evidence[ref] = finding
+ location = finding["raw_source"].get("location") or {}
+ attributes = location.get("attributes") or {}
+ catalog.append(
+ {
+ "finding_ref": ref,
+ "source_url": url,
+ "rule": finding["problem_family"],
+ "problem": finding["problem"],
+ "element": location.get("tag", finding["element"]),
+ "text": location.get("text_preview")
+ or attributes.get("aria-label")
+ or attributes.get("alt")
+ or attributes.get("title", ""),
+ "target": location.get("href")
+ or location.get("src")
+ or attributes.get("id", ""),
+ "location": location.get("css_path", finding["location"]),
+ }
+ )
+ humans = list(audit.to_dict()["human_findings"])
+ for finding in humans:
+ originals[finding["human_finding_id"]] = finding
+ pages.append(
+ {
+ "source_url": url,
+ "human_finding_ids": [
+ finding["human_finding_id"] for finding in humans
+ ],
+ "findings": evidence,
+ }
+ )
+ humans = sorted(originals.values(), key=lambda finding: finding["source_row"])
+ evidence = {
+ "human_audit_sha256": next(iter(audits.values())).human_audit_sha256,
+ "human_findings": humans,
+ "pages": pages,
+ }
+ return evidence, catalog
+
+
+def _match_rows(evidence: dict[str, Any]) -> list[dict[str, Any]]:
+ """List every original Human finding on each applicable page."""
+ humans = {finding["human_finding_id"]: finding for finding in evidence["human_findings"]}
+ return [
+ {"source_url": page["source_url"], **_human_columns(humans[identity])}
+ for page in evidence["pages"]
+ for identity in page["human_finding_ids"]
+ ]
+
+
+def prepare_human_finding_mapping_review(human_audit: Path, snapshots: Path, output: Path) -> None:
+ """Prepare two editable CSVs and a checker catalog using the original saved inputs."""
+ if output.exists():
+ raise FileExistsError(f"Review directory already exists: {output}")
+ evidence, catalog = _review_sources(human_audit, snapshots)
+ output.mkdir(parents=True)
+ _write_csv(
+ output / "classifications.csv", CLASSIFICATION_COLUMNS,
+ [_human_columns(finding) for finding in evidence["human_findings"]],
+ )
+ _write_csv(output / "programmatic-review.csv", MATCH_COLUMNS, _match_rows(evidence))
+ _write_csv(output / "programmatic-findings.csv", FINDING_COLUMNS, catalog)
+ (output / "README.md").write_text(
+ "# Review the Human finding mapping\n\n"
+ "1. Open classifications.csv in Excel or another CSV editor. For every original Human finding, "
+ "fill classification and classification_reason. Classify each original once, including Globals. "
+ "Use programmatic only when existing checker findings fully cover the reported defect; "
+ "llm_eligible for semantic judgment or checks absent from existing rules; unavailable_evidence "
+ "when saved HTML cannot establish the defect; ambiguous when the Human finding is unclear.\n\n"
+ "2. Open programmatic-review.csv and programmatic-findings.csv side by side. "
+ "For each Human finding and page, enter complete matching finding_ref values in "
+ "programmatic_finding_refs, separated by semicolons, and explain them in programmatic_reason. "
+ "Multiple selected findings must jointly cover the full defect. A shared WCAG criterion alone "
+ "does not establish a match. Select only findings from that page, and use each reference once. "
+ "Leave references blank for no complete match and explain the absence. Programmatic "
+ "classifications require complete matches on every applicable page. Not-testable classifications "
+ "cannot claim matches. LLM classifications may have Programmatic cross-catches.\n\n"
+ "3. Save both edited files as CSV. Keep original evidence, IDs, headers, and every row unchanged. "
+ "Row order may change. Inspect the original saved snapshots when needed. The findings catalog is "
+ "read-only reference material; approval compares it with fresh checker results from those snapshots.\n\n"
+ "4. Run visionaid-evaluate human-finding-mapping-approve --human-audit ORIGINAL_AUDIT.xlsx "
+ "--snapshots ORIGINAL_SNAPSHOTS.json --review-dir THIS_DIRECTORY "
+ "--reviewer 'Your name'. Keep the original human audit Excel file "
+ "and captured snapshots available. "
+ "The CLI reruns the checks, validates the decisions, and writes human-finding-mapping.json. "
+ "Use that file with run or run-all --human-finding-mapping PATH. Approval records your decisions; it does "
+ "not automatically judge whether the selected evidence semantically covers a defect.\n",
+ encoding="utf-8",
+ )
+
+
+def _read_review(
+ path: Path,
+ columns: tuple[str, ...],
+ expected: list[dict[str, Any]],
+ keys: tuple[str, ...],
+ label: str = "Human finding",
+) -> dict[tuple[str, ...], dict[str, str]]:
+ """Accept reordered rows while rejecting missing rows and altered original evidence."""
+ with path.open(encoding="utf-8-sig", newline="") as stream:
+ reader = csv.DictReader(stream)
+ if len(reader.fieldnames or []) != len(columns) or set(
+ reader.fieldnames or []
+ ) != set(columns):
+ raise ValueError(f"{path.name}: keep every original column exactly once")
+ entered = list(reader)
+ originals = {tuple(str(row[key]) for key in keys): row for row in expected}
+ if len(entered) != len(originals):
+ raise ValueError(f"{path.name}: keep every original review row exactly once")
+ result = {}
+ for line, row in enumerate(entered, start=2):
+ key = tuple(row[column] for column in keys)
+ if key not in originals or key in result:
+ raise ValueError(
+ f"{path.name} row {line}: unknown or repeated {label}/page {key}"
+ )
+ if any(
+ row.get(column) != str(value) for column, value in originals[key].items()
+ ):
+ raise ValueError(
+ f"{path.name} row {line}: original {label} evidence changed for {key}"
+ )
+ if None in row or any(value is None for value in row.values()):
+ raise ValueError(
+ f"{path.name} row {line}: invalid CSV row; preserve all columns"
+ )
+ result[key] = row
+ return result
+
+
+def approve_human_finding_mapping_review(
+ directory: Path, human_audit: Path, snapshots: Path,
+ reviewer: str, output: Path | None = None,
+) -> Path:
+ """Validate human CSV decisions and generate portable approved evidence expectations."""
+ if not reviewer.strip():
+ raise ValueError("A Human finding mapping reviewer description is required")
+ evidence, catalog = _review_sources(human_audit, snapshots)
+ _read_review(
+ directory / "programmatic-findings.csv", FINDING_COLUMNS, catalog,
+ ("finding_ref",), label="Programmatic finding",
+ )
+ humans = evidence["human_findings"]
+ by_id = {finding["human_finding_id"]: finding for finding in humans}
+ classifications = _read_review(
+ directory / "classifications.csv",
+ CLASSIFICATION_COLUMNS,
+ [_human_columns(finding) for finding in humans],
+ ("human_finding_id",),
+ )
+ matches = _read_review(
+ directory / "programmatic-review.csv",
+ MATCH_COLUMNS,
+ _match_rows(evidence),
+ ("source_url", "human_finding_id"),
+ )
+ decisions = {}
+ for identity in by_id:
+ row = classifications[(identity,)]
+ if (
+ row["classification"].strip() not in ELIGIBILITY_VALUES
+ or not row["classification_reason"].strip()
+ ):
+ raise ValueError(
+ f"classifications.csv Human finding {identity}: fill a valid classification and classification_reason"
+ )
+ decisions[identity] = {
+ "human_finding_id": identity,
+ "classification": row["classification"].strip(),
+ "classification_reason": row["classification_reason"].strip(),
+ "programmatic_by_page": {},
+ }
+ for page in evidence["pages"]:
+ for identity in page["human_finding_ids"]:
+ row = matches[(page["source_url"], identity)]
+ prefix = f"programmatic-review.csv Human finding {identity} on {page['source_url']}"
+ reason = row["programmatic_reason"].strip()
+ if not reason:
+ raise ValueError(
+ f"{prefix}: explain the complete match or deliberate absence in programmatic_reason"
+ )
+ expectations = []
+ for ref in (
+ part.strip()
+ for part in row["programmatic_finding_refs"].split(";")
+ if part.strip()
+ ):
+ if ref not in page["findings"]:
+ raise ValueError(
+ f"{prefix}: unknown or wrong-page Programmatic finding reference {ref}"
+ )
+ finding = page["findings"][ref]
+ expectations.append(
+ {
+ "rule": finding["problem_family"],
+ "evidence": {"raw_source": finding["raw_source"]},
+ }
+ )
+ decisions[identity]["programmatic_by_page"][page["source_url"]] = {
+ "expectations": expectations,
+ "reason": reason,
+ }
+ mapping = {
+ "schema_version": 1,
+ "human_audit_sha256": evidence["human_audit_sha256"],
+ "approved_by": reviewer.strip(),
+ "decisions": list(decisions.values()),
+ }
+ validate_human_finding_mapping(mapping, evidence["human_audit_sha256"])
+ for page in evidence["pages"]:
+ resolved_decisions(
+ mapping,
+ {
+ "human_audit_sha256": evidence["human_audit_sha256"],
+ "human_findings": [
+ by_id[identity] for identity in page["human_finding_ids"]
+ ],
+ },
+ page["source_url"],
+ list(page["findings"].values()),
+ )
+ output = output or directory / "human-finding-mapping.json"
+ output.parent.mkdir(parents=True, exist_ok=True)
+ with output.open("x", encoding="utf-8") as stream:
+ stream.write(json.dumps(mapping, ensure_ascii=False, indent=2) + "\n")
+ return output
diff --git a/vision_aid/evaluation/inputs/__init__.py b/vision_aid/evaluation/inputs/__init__.py
new file mode 100644
index 0000000..57a4d3c
--- /dev/null
+++ b/vision_aid/evaluation/inputs/__init__.py
@@ -0,0 +1 @@
+"""Import Human audit evidence for evaluation workflows."""
diff --git a/vision_aid/evaluation/inputs/human_audit.py b/vision_aid/evaluation/inputs/human_audit.py
new file mode 100644
index 0000000..1ae550a
--- /dev/null
+++ b/vision_aid/evaluation/inputs/human_audit.py
@@ -0,0 +1,167 @@
+"""Preserve Human findings and select them for one page or every listed page."""
+
+from __future__ import annotations
+
+import hashlib
+import re
+from pathlib import Path
+from typing import Literal
+from urllib.parse import urlsplit
+
+import openpyxl
+
+from vision_aid.evaluation.schemas import HumanAudit, HumanFinding
+
+SHEET = "Defect report"
+HOMEPAGE_SCOPES: set[Literal["home", "global"]] = {"home", "global"}
+HEADERS = (
+ "Sr. #",
+ "element name",
+ "Browser Combination",
+ "Page name",
+ "Issue Title",
+ "Steps to Reproduce",
+ "actual result",
+ "expected result",
+ "Recommendation for Fix",
+ "WCAG Sc",
+ "Type of Change",
+ "comment",
+)
+ELIGIBILITY_VALUES = (
+ "llm_eligible",
+ "programmatic",
+ "unavailable_evidence",
+ "ambiguous",
+)
+
+
+def _source_findings(
+ path: Path, *, homepage_only: bool
+) -> tuple[str, tuple[HumanFinding, ...]]:
+ """Read original Human finding evidence with stable source identities."""
+ checksum = hashlib.sha256(path.read_bytes()).hexdigest()
+ human_audit_excel = openpyxl.load_workbook(path, read_only=True, data_only=True)
+ try:
+ sheet = human_audit_excel[SHEET]
+ rows = sheet.iter_rows(values_only=True)
+ headers = tuple(str(value or "") for value in next(rows))
+ if headers[: len(HEADERS)] != HEADERS:
+ raise ValueError("Human audit has unexpected Defect report columns")
+ human_findings = []
+ for number, values in enumerate(rows, start=2):
+ raw = {
+ header: "" if value is None else str(value)
+ for header, value in zip(headers, values, strict=False)
+ if header
+ }
+ scope = raw["Page name"].strip().casefold()
+ if homepage_only and scope not in HOMEPAGE_SCOPES:
+ continue
+ if not raw["Issue Title"].strip():
+ raise ValueError(f"Defect report row {number} has no Issue Title")
+ identity = f"{checksum}:{SHEET}:{number}".encode()
+ human_findings.append(
+ HumanFinding(
+ human_finding_id="human-" + hashlib.sha256(identity).hexdigest()[:16],
+ source_sheet=SHEET,
+ source_row=number,
+ page_scope=scope if scope in HOMEPAGE_SCOPES else "page",
+ source_element=raw["element name"],
+ problem=raw["Issue Title"],
+ raw_evidence=raw,
+ wcag_evidence=tuple(
+ dict.fromkeys(re.findall(r"\b\d+\.\d+\.\d+\b", raw["WCAG Sc"]))
+ ),
+ )
+ )
+ finally:
+ human_audit_excel.close()
+ return checksum, tuple(human_findings)
+
+
+def import_homepage_human_findings(path: Path, homepage_url: str) -> HumanAudit:
+ """Preserve the one-page command's Home and Global source selection."""
+ checksum, human_findings = _source_findings(path, homepage_only=True)
+ if not human_findings:
+ raise ValueError("Human audit has no Home or Global defects")
+ return HumanAudit(path.name, checksum, homepage_url, human_findings)
+
+
+def import_all_human_findings(path: Path) -> dict[str, HumanAudit]:
+ """Assign local findings by reproduction URL and repeat Globals per page."""
+ urls = list_audit_urls(path)
+ if len(urls) != len(set(urls)):
+ raise ValueError("Human audit URL list contains repeated target URLs")
+ checksum, findings = _source_findings(path, homepage_only=False)
+ if not findings:
+ raise ValueError("Human audit has no Human findings")
+ by_page: dict[str, list[HumanFinding]] = {url: [] for url in urls}
+ for finding in findings:
+ if finding.page_scope == "global":
+ for page_findings in by_page.values():
+ page_findings.append(finding)
+ continue
+ steps = finding.raw_evidence["Steps to Reproduce"]
+ reproduced = {
+ match.rstrip(".,;:)]}")
+ for match in re.findall(r'https?://[^\s<>"\']+', steps)
+ }
+ assigned = set()
+ for url in reproduced:
+ if url in by_page:
+ assigned.add(url)
+ continue
+ parsed = urlsplit(url)
+ # This is the audit's known alias, not general path case folding.
+ if parsed.path == "/Contact/":
+ url = parsed._replace(path="/contact/").geturl()
+ if url in by_page:
+ assigned.add(url)
+ if len(assigned) != 1:
+ raise ValueError(
+ f"Defect report row {finding.source_row} must map to exactly "
+ "one listed target URL in Steps to Reproduce"
+ )
+ by_page[assigned.pop()].append(finding)
+ return {
+ url: HumanAudit(path.name, checksum, url, tuple(page_findings))
+ for url, page_findings in by_page.items()
+ }
+
+
+def list_audit_urls(path: Path) -> tuple[str, ...]:
+ """Read the Human audit's explicit target URLs in source order."""
+ human_audit_excel = openpyxl.load_workbook(path, read_only=True, data_only=True)
+ try:
+ sheet = human_audit_excel["URL list"]
+ rows = sheet.iter_rows(values_only=True)
+ headers = tuple(str(value or "").strip() for value in next(rows))
+ if "URL" not in headers:
+ raise ValueError("Human audit URL list has no URL column")
+ index = headers.index("URL")
+ urls = tuple(
+ str(row[index]).strip()
+ for row in rows
+ if len(row) > index and row[index] is not None and str(row[index]).strip()
+ )
+ finally:
+ human_audit_excel.close()
+ if not urls:
+ raise ValueError("Human audit URL list has no target URLs")
+ return urls
+
+
+def import_page_human_findings(path: Path, source_url: str) -> HumanAudit:
+ """Select a listed page's local findings and Globals, retaining older Home audits."""
+ human_audit_excel = openpyxl.load_workbook(path, read_only=True, data_only=True)
+ try:
+ has_urls = "URL list" in human_audit_excel.sheetnames
+ finally:
+ human_audit_excel.close()
+ if not has_urls:
+ return import_homepage_human_findings(path, source_url)
+ pages = import_all_human_findings(path)
+ if source_url not in pages:
+ raise ValueError(f"Source URL is not listed in the Human audit: {source_url}")
+ return pages[source_url]
diff --git a/vision_aid/evaluation/page.py b/vision_aid/evaluation/page.py
new file mode 100644
index 0000000..d0c2b8e
--- /dev/null
+++ b/vision_aid/evaluation/page.py
@@ -0,0 +1,181 @@
+"""Shared preparation and human-reviewed evidence for one evaluated page."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import shlex
+import shutil
+from dataclasses import asdict, dataclass
+from decimal import Decimal
+from pathlib import Path
+from typing import Any
+
+from entry_points.run_pipeline import run_pipeline
+from vision_aid.evaluation.execution.llm_audit import (
+ USAGE_KEYS,
+ LLMRequestConfig,
+ evaluation_configuration,
+ write_json,
+)
+from vision_aid.evaluation.execution.normalization import (
+ normalize_programmatic_findings,
+)
+from vision_aid.evaluation.execution.pricing import DEFAULT_SCHEDULE, PriceSchedule
+from vision_aid.evaluation.human_finding_mapping import (
+ apply_human_finding_mapping,
+ saved_human_finding_mapping_decisions,
+)
+from vision_aid.evaluation.reporting.html_report import write_page_report
+from vision_aid.evaluation.review import create_review, reviewed_rows
+from vision_aid.evaluation.schemas import HumanAudit
+
+
+@dataclass(frozen=True)
+class PageReview:
+ """A completed page's validated review and authoritative saved evidence."""
+
+ manifest: dict[str, Any]
+ rows: list[dict[str, Any]]
+ llm: list[dict[str, Any]]
+ programmatic: list[dict[str, Any]]
+ schedule: PriceSchedule
+
+
+def prepare_page(
+ run: Path,
+ html: Path,
+ source_url: str,
+ human_audit: HumanAudit,
+ request_config: LLMRequestConfig,
+ limit: Decimal,
+ mapping: dict[str, Any],
+) -> dict[str, Any]:
+ """Save a page's inputs, checks, prompts and blank review without paid requests.
+
+ The caller supplies an existing fresh run directory and applicable Human
+ findings, retaining the same operations for one-page and Full evaluations.
+ """
+ if not html.is_file():
+ raise ValueError(f"Benchmark HTML not found: {html}")
+ configuration = evaluation_configuration(request_config)
+ artifacts = run / "artifacts"
+ artifacts.mkdir()
+ shutil.copyfile(html, run / "snapshot.html")
+ shutil.copyfile(DEFAULT_SCHEDULE, artifacts / "pricing.json")
+ snapshot_hash = hashlib.sha256((run / "snapshot.html").read_bytes()).hexdigest()
+ write_json(artifacts / "human-findings.json", human_audit.to_dict())
+ production = run_pipeline(
+ html_path=str(run / "snapshot.html"),
+ output_dir=artifacts,
+ api_key=None,
+ model=request_config.model,
+ dry_run=True,
+ include_summaries=False,
+ request_config=request_config,
+ )
+ (artifacts / "manifest.json").unlink()
+ raw_path = artifacts / "programmatic_findings.json"
+ raw_programmatic = json.loads(raw_path.read_text())
+ raw_path.rename(artifacts / "raw-programmatic-findings.json")
+ findings = normalize_programmatic_findings(
+ raw_programmatic,
+ run_id=run.name,
+ page_url=source_url,
+ )
+ write_json(
+ artifacts / "normalized-programmatic-findings.json",
+ [asdict(item) for item in findings],
+ )
+ create_review(
+ run / "review.csv", list(human_audit.to_dict()["human_findings"])
+ )
+ schedule = PriceSchedule.load(artifacts / "pricing.json")
+ manifest = {
+ "run_id": run.name,
+ "source_url": source_url,
+ "human_audit_filename": human_audit.human_audit_filename,
+ "human_audit_sha256": human_audit.human_audit_sha256,
+ "snapshot_sha256": snapshot_hash,
+ "configuration": configuration,
+ "pricing_version": schedule.version,
+ "pricing_sha256": schedule.identity,
+ "max_cost_usd": str(limit),
+ "complete": False,
+ "incomplete_reason": "preview",
+ "estimated_cost_usd": "0",
+ "usage": {key: 0 for key in USAGE_KEYS},
+ "format_failures": [],
+ "prompts": [
+ {"name": entry["name"], "checklist": entry["checklist"]}
+ for entry in production["prompts_dry_run"]
+ ],
+ "filters": production["pass1_filters_active"],
+ "skipped_prompts": production["prompts_skipped"],
+ }
+ write_json(run / "run.json", manifest)
+ apply_human_finding_mapping(run, mapping)
+ (run / "README.md").write_text(
+ "# Review this page run\n\n"
+ "Check `run.json` for completion and any stopping reason. "
+ "Add `--live` to `run` or `run-all` to call the LLMs. Without `--live`, the command "
+ "runs in preview mode, generating artifacts without calling the LLMs.\n\n"
+ "1. After execution completes, compare `normalized-llm-findings.json` with the Human "
+ "findings in `review.csv`. Use `snapshot.html` to check the page evidence.\n"
+ "2. Edit only `llm_finding_id`, `match_reason`, and `review_notes`. Separate joint "
+ "matches with semicolons. Select findings only when they fully cover the Human "
+ "finding; otherwise leave the LLM ID blank. Keep approved fields unchanged.\n"
+ "3. Save the CSV, then run from the repository root:\n\n"
+ "```bash\n"
+ f"uv run visionaid-evaluate report --run-dir {shlex.quote(str(run.resolve()))} "
+ '--reviewer "Your name and review method"\n'
+ "```\n\n"
+ "Open `report.html` in a browser. Supporting evidence, including "
+ "`normalized-programmatic-findings.json`, is in `artifacts/`.\n",
+ encoding="utf-8",
+ )
+ manifest = json.loads((run / "run.json").read_text())
+ return manifest
+
+
+def validated_page_review(run: Path) -> PageReview:
+ """Load and validate a completed page's evidence and human decisions."""
+ run = run.resolve()
+ artifacts = run / "artifacts"
+ manifest = json.loads((run / "run.json").read_text())
+ if manifest["run_id"] != run.name:
+ raise ValueError("Run evidence does not belong to this directory")
+ if not manifest["complete"]:
+ raise ValueError("Incomplete Evaluation runs cannot produce a final report")
+ human_audit = json.loads((artifacts / "human-findings.json").read_text())
+ if human_audit["human_audit_sha256"] != manifest["human_audit_sha256"]:
+ raise ValueError("Human audit identity differs from the run")
+ llm = json.loads((run / "normalized-llm-findings.json").read_text())
+ programmatic_file = artifacts / "normalized-programmatic-findings.json"
+ programmatic = json.loads(programmatic_file.read_text())
+ for kind, collection in (("llm", llm), ("programmatic", programmatic)):
+ if any(
+ item["run_id"] != run.name
+ or item["source"] != kind
+ or item["page_url"] != manifest["source_url"]
+ for item in collection
+ ):
+ label = kind.upper() if kind == "llm" else kind.title()
+ raise ValueError(f"{label} findings do not belong to this run")
+ approved = saved_human_finding_mapping_decisions(run, manifest, human_audit, programmatic)
+ rows = reviewed_rows(
+ run / "review.csv", human_audit["human_findings"], llm, programmatic, approved
+ )
+ schedule = PriceSchedule.load(artifacts / "pricing.json")
+ return PageReview(manifest, rows, llm, programmatic, schedule)
+
+
+def report_page(run: Path, reviewer: str) -> PageReview:
+ """Validate human decisions and write the page report with attribution."""
+ (run / "report.html").unlink(missing_ok=True)
+ if not reviewer.strip():
+ raise ValueError("A reviewer description is required")
+ review = validated_page_review(run)
+ write_page_report(run / "report.html", review, reviewer)
+ (run / "report.md").unlink(missing_ok=True)
+ return review
diff --git a/vision_aid/evaluation/reporting/__init__.py b/vision_aid/evaluation/reporting/__init__.py
new file mode 100644
index 0000000..46ce280
--- /dev/null
+++ b/vision_aid/evaluation/reporting/__init__.py
@@ -0,0 +1 @@
+"""Render self-contained Evaluation reports and their packaged assets."""
diff --git a/vision_aid/evaluation/reporting/css/charts.css b/vision_aid/evaluation/reporting/css/charts.css
new file mode 100644
index 0000000..d401ae5
--- /dev/null
+++ b/vision_aid/evaluation/reporting/css/charts.css
@@ -0,0 +1,27 @@
+.performance-heading { margin-bottom: 1.5rem; }
+.performance-heading h2 { margin: 0 0 1rem; }
+.cost-overview { display: grid; grid-template-columns: minmax(0, .8fr) minmax(0, 1.2fr); gap: 1.5rem; align-items: start; }
+.cost-overview > * { min-width: 0; }
+@media (min-width: 60rem) { .cost-overview { align-items: stretch; } }
+.run-cost { margin: 0; padding: .75rem 1rem; max-width: 100%; border: 3px solid #164e88; background: #edf2f6; }
+.cost-label { display: block; margin-bottom: .4rem; font-weight: 650; }
+.cost-value { display: block; font-size: clamp(2rem, 4vw, 3rem); line-height: 1.1; font-variant-numeric: tabular-nums; overflow-wrap: anywhere; }
+.cost-currency { font-size: 1rem; font-weight: 650; }
+.token-usage { display: grid; grid-template-columns: minmax(0, 1fr) auto; gap: .4rem 1rem; margin: 1.25rem 0 0; padding-top: .75rem; border-top: 1px solid #687f93; }
+.token-usage dd { margin: 0; text-align: right; font-variant-numeric: tabular-nums; }
+.model-pricing h3 { margin: 0; }
+.model-pricing table { min-width: 26rem; margin-bottom: 0; }
+.model-pricing th { overflow-wrap: normal; }
+.model-pricing td { white-space: nowrap; }
+@media (max-width: 60rem) { .cost-overview { grid-template-columns: minmax(0, 1fr); } }
+.detection-chart { margin: 1rem 0; }
+.chart-bars { list-style: none; padding: 0; }
+.chart-bars li { margin-block: 1rem; padding: .75rem; border: 1px solid #687f93; overflow-wrap: anywhere; }
+.chart-bars .chart-primary { border: 3px solid #164e88; background: #edf2f6; }
+.chart-primary .chart-value { font-size: 1.2rem; }
+.chart-value { margin: 0 0 .5rem; }
+.chart-percentage { text-decoration: underline; text-underline-offset: .15em; }
+.chart-track { width: 100%; height: 1.4rem; background: #edf2f6; border: 1px solid #687f93; }
+.chart-fill { display: block; height: 100%; background: #164e88; }
+@media (forced-colors: active) { .chart-fill { background: Highlight; } }
+@media print { .chart-track, .chart-fill { print-color-adjust: exact; } .chart-bars li { break-inside: avoid; } }
diff --git a/vision_aid/evaluation/reporting/css/print.css b/vision_aid/evaluation/reporting/css/print.css
new file mode 100644
index 0000000..5c4f3db
--- /dev/null
+++ b/vision_aid/evaluation/reporting/css/print.css
@@ -0,0 +1,11 @@
+#print-report { font: inherit; color: #164e88; background: #edf2f6; border: 1px solid #687f93; border-radius: .2rem; padding: .5rem .75rem; margin-bottom: 1rem; cursor: pointer; }
+@page { margin: 12mm; }
+@media print {
+ body { max-width: none; padding: 0; color: #000; font-size: 11pt; line-height: 1.4; }
+ .skip, nav, #print-report, main > :not(#performance) { display: none !important; }
+ .table-scroll { overflow: visible; }
+ table { table-layout: fixed; }
+ thead { display: table-header-group; }
+ th, td { padding: .35rem; }
+ h1, h2, h3, caption { break-after: avoid; }
+}
diff --git a/vision_aid/evaluation/reporting/css/report.css b/vision_aid/evaluation/reporting/css/report.css
new file mode 100644
index 0000000..405c193
--- /dev/null
+++ b/vision_aid/evaluation/reporting/css/report.css
@@ -0,0 +1,57 @@
+:root { color-scheme: light; font-family: system-ui, sans-serif; color: #182838; background: #fff; line-height: 1.6; }
+* { box-sizing: border-box; }
+body { margin: 0 auto; max-width: 90rem; padding: 1.25rem; }
+h1, h2, h3 { line-height: 1.25; }
+h1 { font-size: clamp(1.8rem, 5vw, 2.7rem); }
+a { color: #164e88; text-underline-offset: .18em; }
+a:focus-visible, summary:focus-visible, button:focus-visible, input:focus-visible, select:focus-visible { outline: 3px solid #8c3500; outline-offset: 4px; }
+.skip { position: absolute; top: -10rem; padding: .7rem; background: #fff; }
+.skip:focus { top: .5rem; }
+nav { display: flex; flex-wrap: wrap; gap: 1.25rem; border-block: 1px solid #8090a0; padding-block: .8rem; }
+section { margin-block: 2rem; }
+.table-scroll { overflow-x: auto; }
+table { width: 100%; border-collapse: collapse; margin-block: 1rem; }
+caption { font-weight: 700; text-align: left; margin-bottom: .5rem; }
+th, td { text-align: left; vertical-align: top; border: 1px solid #8a99a6; padding: .65rem; overflow-wrap: anywhere; }
+thead { background: #edf2f6; }
+summary { cursor: pointer; color: #164e88; font-weight: 600; }
+details[open] > summary { margin-bottom: 1rem; }
+.evidence { min-width: 16rem; max-width: 50rem; }
+dl { margin-block: 1rem; }
+dt { font-weight: 650; }
+dd { margin: 0 0 .8rem; white-space: pre-wrap; overflow-wrap: anywhere; }
+.matched { border-left: 3px solid #687f93; padding-left: 1rem; margin-block: 1rem; }
+@media (max-width: 40rem) { body { padding: .8rem; } .evidence { min-width: 13rem; } }
+@media print { body { max-width: none; } .skip, nav { display: none; } .table-scroll { overflow: visible; } }
+
+#finding-controls fieldset { border: 1px solid #8a99a6; padding: 1rem; }
+#finding-controls label { display: block; font-weight: 650; margin-top: .8rem; }
+#finding-controls select { display: block; width: 100%; max-width: 42rem; font: inherit; padding: .5rem; color: inherit; background: #fff; border: 1px solid #687f93; }
+#finding-controls button { font: inherit; color: #164e88; background: #fff; border: 1px solid #687f93; border-radius: .2rem; padding: .5rem .75rem; margin: .5rem .5rem .5rem 0; cursor: pointer; }
+[data-selected-page-result] { border-left: 3px solid #687f93; padding-left: .7rem; }
+
+@media screen {
+ .table-scroll > table { min-width: 32rem; }
+ .table-scroll thead th { overflow-wrap: normal; }
+ #findings { min-width: 64rem; }
+ #findings > tbody > tr > th, #findings > tbody > tr > td:last-child { min-width: 16rem; }
+ .findings-panel { border: 1px solid #8a99a6; border-bottom: 0; }
+ .findings-panel > .table-scroll { margin-inline: -1px; }
+ .findings-panel > .table-scroll > table { margin: 0; }
+ #finding-controls { background: #edf2f6; padding-top: 1rem; }
+ #finding-controls fieldset { min-width: 0; margin: 0; padding: 0 1rem 1rem; border: 0; }
+ #finding-controls legend { padding: 0; font-weight: 650; margin-bottom: .75rem; }
+ .finding-filter-fields { display: grid; grid-template-columns: minmax(0, 1fr); gap: 1rem; }
+ .finding-filter { min-width: 0; }
+ #finding-controls label { margin-top: 0; }
+ #finding-controls select { max-width: none; }
+ .finding-filter-actions { display: flex; flex-wrap: wrap; gap: .75rem; margin-top: .75rem; }
+ #finding-controls button { margin: 0; }
+ #finding-count { margin: 0; padding: .75rem 1rem; background: #fff; border-top: 1px solid #8a99a6; }
+}
+@media screen and (min-width: 64rem) {
+ .finding-filter-fields { grid-template-columns: repeat(auto-fit, minmax(18rem, 1fr)); }
+}
+@media screen and (max-width: 40rem) {
+ .evidence { width: 16rem; }
+}
diff --git a/vision_aid/evaluation/reporting/html_report.py b/vision_aid/evaluation/reporting/html_report.py
new file mode 100644
index 0000000..cce4fbc
--- /dev/null
+++ b/vision_aid/evaluation/reporting/html_report.py
@@ -0,0 +1,236 @@
+"""Prepare reader-facing data and publish self-contained evaluation reports."""
+
+from __future__ import annotations
+
+import os
+import tempfile
+from decimal import Decimal
+from importlib.resources import files
+from pathlib import Path
+from typing import TYPE_CHECKING, Any
+
+from jinja2 import (
+ Environment,
+ PackageLoader,
+ StrictUndefined,
+ TemplateError,
+ select_autoescape,
+)
+
+from vision_aid.evaluation.review import detection_counts
+
+if TYPE_CHECKING:
+ from vision_aid.evaluation.execution.pricing import PriceSchedule
+ from vision_aid.evaluation.page import PageReview
+
+EXPECTED = {
+ "llm_eligible": "LLM",
+ "programmatic": "Programmatic checks",
+ "unavailable_evidence": "Not testable from the page file",
+ "ambiguous": "Not testable from the page file",
+}
+
+
+def _display_model(model: str) -> str:
+ """Use uppercase GPT in report labels while preserving saved model IDs."""
+ return "GPT-" + model[4:].replace("-", " ").title() if model.startswith("gpt-") else model
+
+
+def _cost_details(manifest: dict[str, Any], schedules: list[PriceSchedule]) -> dict[str, Any]:
+ """Present saved usage and model prices without recalculating the run cost."""
+ usage = manifest["usage"]
+ model = manifest["configuration"].get("model")
+ if model is None:
+ raise ValueError("Cannot render complete report: model is missing")
+ token_usage = [
+ {"label": label, "count": f"{usage[key]:,}"}
+ for key, label in (
+ ("input_tokens", "Input tokens"),
+ ("output_tokens", "Output tokens"),
+ ("cached_input_tokens", "Cached input tokens"),
+ ("cache_creation_input_tokens", "Cache-write input tokens"),
+ )
+ if key in {"input_tokens", "output_tokens"} or usage[key]
+ ]
+ model_prices = []
+ seen = set()
+ for schedule in schedules:
+ if schedule.identity in seen:
+ continue
+ seen.add(schedule.identity)
+ rates = schedule.data["models"][model]
+ upper = rates.get("long_context")
+ tiers = [{"label": "All requests", "rates": rates}]
+ if upper:
+ threshold = f"{upper['above_input_tokens']:,}"
+ tiers = [
+ {"label": f"Up to {threshold}", "rates": rates},
+ {"label": f"Over {threshold}", "rates": upper},
+ ]
+ model_prices.append({
+ "model": _display_model(model), "description": schedule.price_description,
+ "tiers": tiers,
+ "show_cache_read": bool(usage["cached_input_tokens"]),
+ "show_cache_write": bool(usage["cache_creation_input_tokens"]),
+ })
+ return {
+ "cost": f"{Decimal(manifest['estimated_cost_usd']):.6f}",
+ "model_name": _display_model(model),
+ "token_usage": token_usage, "model_prices": model_prices,
+ }
+
+
+def _chart_rates(counts: dict[str, dict[str, int]]) -> list[dict[str, Any]]:
+ """Prepare chart labels and widths from the existing deterministic counts."""
+ rates = []
+ for kind, label in (
+ ("llm", "LLM"),
+ ("programmatic", "Programmatic"),
+ ("overall", "Overall"),
+ ):
+ caught, eligible = counts[kind]["caught"], counts[kind]["eligible"]
+ percentage = f"{caught / eligible * 100:.1f}" if eligible else None
+ rates.append({
+ "label": label, **counts[kind],
+ "bar_percent": percentage,
+ "percentage": percentage + "%" if percentage is not None else (
+ "Not defined: no imported Human findings" if kind == "overall"
+ else "Not defined: no eligible Human findings"
+ ),
+ })
+ return rates
+
+
+def page_data(review: PageReview, reviewer: str) -> dict[str, Any]:
+ """Resolve validated evidence and deterministic scores before rendering."""
+ manifest = review.manifest
+ findings = {item["finding_id"]: item for item in review.llm + review.programmatic}
+ counts = detection_counts(review.rows)
+ rows = []
+ for row in review.rows:
+ human, decision = row["human_finding"], row["decision"]
+ classification = decision["classification"].strip()
+ matched = []
+ for kind in ("llm", "programmatic"):
+ for identity in row["matches"][kind]:
+ finding = findings[identity]
+ check = finding["prompt"].replace("_", " ") + " check"
+ if finding.get("screen_only"):
+ check += "; shown on screen only, not in the tool's CSV report"
+ matched.append(
+ {
+ "id": finding["finding_id"],
+ "source": "LLM" if kind == "llm" else "Programmatic checks",
+ "check": check,
+ "check_id": finding["prompt"],
+ "checklist": finding["checklist"],
+ "problem": finding["problem"],
+ "element": finding["element"],
+ "location": finding["location"],
+ }
+ )
+ rows.append(
+ {
+ "human": human,
+ "human_finding_mapping_approved_by": manifest["human_finding_mapping_approved_by"],
+ "decision": decision,
+ "source_url": manifest["source_url"],
+ "run_id": manifest["run_id"],
+ "expected": EXPECTED[classification],
+ "llm_outcome": "Caught" if row["matches"]["llm"] else "Not caught",
+ "programmatic_outcome": "Caught"
+ if row["matches"]["programmatic"]
+ else "Not caught",
+ "matched": matched,
+ }
+ )
+ rates = _chart_rates(counts)
+ return {
+ "title": "LLM Accessibility Evaluation",
+ "manifest": manifest,
+ "reviewer": reviewer,
+ "rows": rows,
+ "rates": rates,
+ **_cost_details(manifest, [review.schedule]),
+ }
+
+
+def full_data(
+ manifest: dict[str, Any], pooled: list[dict[str, Any]],
+ reviews: list[PageReview], reviewer: str,
+) -> dict[str, Any]:
+ """Combine validated presentation rows without changing pooled scoring."""
+ pages = [page_data(review, reviewer) for review in reviews]
+ grouped: dict[str, list[dict[str, Any]]] = {}
+ for page in pages:
+ for row in page["rows"]:
+ grouped.setdefault(row["human"]["human_finding_id"], []).append({
+ **row, "source_url": page["manifest"]["source_url"],
+ "run_id": page["manifest"]["run_id"],
+ })
+ counts = detection_counts(pooled)
+ rows = []
+ for row in pooled:
+ rows.append({
+ "human": row["human_finding"],
+ "expected": EXPECTED[row["decision"]["classification"]],
+ "llm_outcome": "Caught" if row["matches"]["llm"] else "Not caught",
+ "programmatic_outcome": "Caught" if row["matches"]["programmatic"] else "Not caught",
+ "pages": grouped[row["human_finding"]["human_finding_id"]],
+ })
+ return {
+ "title": "Full Accessibility Evaluation", "full": True,
+ "manifest": manifest, "reviewer": reviewer, "rows": rows, "pages": pages,
+ "rates": _chart_rates(counts),
+ **_cost_details(manifest, [review.schedule for review in reviews]),
+ }
+
+
+def write_full_report(
+ path: Path, manifest: dict[str, Any], pooled: list[dict[str, Any]],
+ reviews: list[PageReview], reviewer: str,
+) -> None:
+ """Render every accepted page decision into the standalone parent."""
+ _write_html(path, "full.html.jinja", full_data(manifest, pooled, reviews, reviewer))
+
+
+def write_page_report(path: Path, review: PageReview, reviewer: str) -> None:
+ """Publish a standalone individual Evaluation report."""
+ _write_html(path, "page.html.jinja", page_data(review, reviewer))
+
+
+def _write_html(path: Path, template: str, data: dict[str, Any]) -> None:
+ """Render completely, then atomically replace the report in its directory."""
+ environment = Environment(
+ loader=PackageLoader("vision_aid.evaluation.reporting", "templates"),
+ autoescape=select_autoescape(("html.jinja",)),
+ undefined=StrictUndefined,
+ )
+ # Only trusted package assets bypass escaping; review data stays escaped.
+ assets = files("vision_aid.evaluation.reporting")
+ styles = "".join(
+ assets.joinpath("css", name).read_text(encoding="utf-8")
+ for name in ("report.css", "charts.css", "print.css")
+ )
+ scripts = {
+ name: assets.joinpath("js", f"{name}.js").read_text(encoding="utf-8")
+ for name in ("filters", "print")
+ }
+ try:
+ content = environment.get_template(template).render(
+ **data, styles=styles, scripts=scripts,
+ )
+ except TemplateError as error:
+ raise ValueError(f"Cannot render complete report: {error}") from error
+ temporary = None
+ try:
+ with tempfile.NamedTemporaryFile(
+ mode="w", encoding="utf-8", dir=path.parent,
+ prefix=f".{path.name}.", suffix=".tmp", delete=False,
+ ) as stream:
+ temporary = Path(stream.name)
+ stream.write(content)
+ os.replace(temporary, path)
+ finally:
+ if temporary is not None:
+ temporary.unlink(missing_ok=True)
diff --git a/vision_aid/evaluation/reporting/js/filters.js b/vision_aid/evaluation/reporting/js/filters.js
new file mode 100644
index 0000000..31bd093
--- /dev/null
+++ b/vision_aid/evaluation/reporting/js/filters.js
@@ -0,0 +1,60 @@
+/* All evidence is already in this document; no network or runtime dependencies. */
+(() => {
+ 'use strict';
+ const controls = document.getElementById('finding-controls');
+ const pageFilter = document.getElementById('finding-page');
+ const outcome = document.getElementById('finding-outcome');
+ const status = document.getElementById('finding-count');
+ const rows = [...document.querySelectorAll('#findings > tbody > tr')].map(row => {
+ return {row, decisions: [...row.querySelectorAll('[data-page-decision]')],
+ selected: row.querySelector('[data-selected-page-result]')};
+ });
+ function matchesOutcome(data) {
+ const llm = data.llm === 'Caught';
+ const programmatic = data.programmatic === 'Caught';
+ switch (outcome.value) {
+ case 'llm-caught': return llm;
+ case 'llm-missed': return data.expected === 'LLM' && !llm;
+ case 'programmatic-caught': return programmatic;
+ case 'programmatic-missed': return data.expected === 'Programmatic checks' && !programmatic;
+ case 'overall-caught': return llm || programmatic;
+ case 'overall-missed': return !llm && !programmatic;
+ case 'not-testable': return data.expected === 'Not testable from the page file';
+ default: return true;
+ }
+ }
+ function update() {
+ const selectedPage = pageFilter?.value || '';
+ let count = 0;
+ for (const item of rows) {
+ const pageDecision = selectedPage ? item.decisions.find(panel => panel.dataset.page === selectedPage) : null;
+ item.row.hidden = Boolean((selectedPage && !pageDecision) ||
+ !matchesOutcome(pageDecision?.dataset || item.row.dataset));
+ item.selected.hidden = !selectedPage || !pageDecision;
+ item.selected.textContent = pageDecision ? `Selected page: ${selectedPage}. Expected source: ${pageDecision.dataset.expected}. LLM: ${pageDecision.dataset.llm}. Programmatic checks: ${pageDecision.dataset.programmatic}. Combined results remain in the overview columns.` : '';
+ for (const panel of item.decisions) panel.hidden = Boolean(selectedPage && panel.dataset.page !== selectedPage);
+ for (const tr of item.row.querySelectorAll('[data-page-table-row]')) tr.hidden = Boolean(selectedPage && tr.dataset.pageTableRow !== selectedPage);
+ if (!item.row.hidden) count++;
+ }
+ status.textContent = `Filtered findings: ${count} of ${rows.length} original Human findings.${count ? '' : ' No matching findings.'}`;
+ }
+ pageFilter?.addEventListener('change', update);
+ outcome.addEventListener('change', update);
+ document.getElementById('clear-filters').addEventListener('click', () => {
+ if (pageFilter) pageFilter.value = '';
+ outcome.value = 'all';
+ update();
+ });
+ for (const [id, open] of [['expand-visible', true], ['collapse-visible', false]]) {
+ document.getElementById(id).addEventListener('click', () => {
+ for (const {row} of rows) {
+ if (row.hidden) continue;
+ for (const panel of row.querySelectorAll('details')) {
+ if (!panel.closest('[hidden]')) panel.open = open;
+ }
+ }
+ });
+ }
+ update();
+ controls.hidden = false;
+})();
diff --git a/vision_aid/evaluation/reporting/js/print.js b/vision_aid/evaluation/reporting/js/print.js
new file mode 100644
index 0000000..24a1a61
--- /dev/null
+++ b/vision_aid/evaluation/reporting/js/print.js
@@ -0,0 +1,29 @@
+/* Print styling limits output to the report header, cost and detection charts. */
+(() => {
+ 'use strict';
+ const button = document.getElementById('print-report');
+ const printMedia = window.matchMedia('print');
+ let readingState = null;
+ function restoreFocus() {
+ if (!readingState || printMedia.matches) return;
+ const {focused, scrollX, scrollY} = readingState;
+ readingState = null;
+ if (focused?.isConnected) focused.focus({preventScroll: true});
+ window.scrollTo(scrollX, scrollY);
+ }
+ printMedia.addEventListener('change', restoreFocus);
+ window.addEventListener('afterprint', restoreFocus);
+ button.addEventListener('click', () => {
+ if (readingState) return;
+ readingState = {
+ focused: document.activeElement, scrollX: window.scrollX, scrollY: window.scrollY,
+ };
+ try {
+ window.print();
+ } catch (error) {
+ restoreFocus();
+ throw error;
+ }
+ });
+ button.hidden = false;
+})();
diff --git a/vision_aid/evaluation/reporting/templates/base.html.jinja b/vision_aid/evaluation/reporting/templates/base.html.jinja
new file mode 100644
index 0000000..5ddbbef
--- /dev/null
+++ b/vision_aid/evaluation/reporting/templates/base.html.jinja
@@ -0,0 +1,25 @@
+
+
+
+
+
+{% block document_title %}{% endblock %}
+
+
+
+Skip to report content
+
+
+{% block content %}{% endblock %}
+
+
+
+
+
diff --git a/vision_aid/evaluation/reporting/templates/charts.html.jinja b/vision_aid/evaluation/reporting/templates/charts.html.jinja
new file mode 100644
index 0000000..f7831ee
--- /dev/null
+++ b/vision_aid/evaluation/reporting/templates/charts.html.jinja
@@ -0,0 +1,44 @@
+{% macro performance_header(cost, token_usage, model_prices) %}
+
+{% endmacro %}
+
+{% macro rate_bar(label, rate, primary=false) %}
+
+{{ label }}: {{ rate.caught }} caught / {{ rate.eligible }} eligible; {{ rate.percentage }}
+{% if rate.bar_percent is not none %}
{% endif %}
+
+{% endmacro %}
+
+{% macro detection_overview(rates) %}
+
+Detection overview
+{% for rate in rates %}{{ rate_bar(rate.label, rate, loop.first) }}{% endfor %}
+Overall includes all Human findings, including those the tool cannot detect from the saved HTML and those too ambiguous to evaluate. These findings are left out of the LLM and Programmatic eligible counts, so the Overall count can exceed their combined total.
+
+{% endmacro %}
+
+{% macro page_comparison(pages) %}
+
+LLM detection by page
+Repeated Global findings contribute to each page's results but only once to combined results.
+
+{% for page in pages %}{{ rate_bar(page.manifest.source_url, page.rates[0]) }}{% endfor %}
+
+{% endmacro %}
diff --git a/vision_aid/evaluation/reporting/templates/controls.html.jinja b/vision_aid/evaluation/reporting/templates/controls.html.jinja
new file mode 100644
index 0000000..884e39a
--- /dev/null
+++ b/vision_aid/evaluation/reporting/templates/controls.html.jinja
@@ -0,0 +1,32 @@
+
+
+
+
diff --git a/vision_aid/evaluation/reporting/templates/evidence.html.jinja b/vision_aid/evaluation/reporting/templates/evidence.html.jinja
new file mode 100644
index 0000000..9ebd0d5
--- /dev/null
+++ b/vision_aid/evaluation/reporting/templates/evidence.html.jinja
@@ -0,0 +1,16 @@
+{% macro human_evidence(row) %}
+Original Human finding
+- Human finding ID
- {{ row.human.human_finding_id }}
- Source
- {{ row.human.source_sheet }}, row {{ row.human.source_row }}; scope: {{ row.human.page_scope }}; element: {{ row.human.source_element }}
+{% for label, value in row.human.raw_evidence.items() %}- {{ label }}
- {{ value }}
{% endfor %}
+{% endmacro %}
+{% macro decision_evidence(row) %}
+{{ row.source_url }}
+Run ID: {{ row.run_id }}
+- Eligibility classification
- {{ row.expected }}
- Recorded explanation
- {{ row.decision.classification_reason }}
- Human finding mapping approved by
- {{ row.human_finding_mapping_approved_by }}
- Approved Programmatic explanation
- {{ row.decision.programmatic_match_reason }}
{% if row.decision.llm_finding_id %}- LLM matching rationale
- {{ row.decision.match_reason }}
{% endif %}{% if row.matched or row.decision.review_notes %}- VA-DAT Matched Findings
- {{ row.decision.review_notes }}
{% endif %}
+{% for finding in row.matched %}{{ 'LLM Finding' if finding.source == 'LLM' else 'Programmatic Finding' }}
{{ finding.source }} ({{ finding.check }}): {{ finding.problem }}
- Finding ID
- {{ finding.id }}
- Check
- {{ finding.check_id }}; {{ finding.checklist }}
{% if finding.source == 'LLM' %}- Element
- {{ finding.element }}
{% endif %}- Location
- {{ finding.location }}
{% endfor %}
+{% endmacro %}
+{% macro page_evidence(row) %}
+View page decisions: {{ row.source_url }}
+{{ decision_evidence(row) }}
+
+{% endmacro %}
diff --git a/vision_aid/evaluation/reporting/templates/full.html.jinja b/vision_aid/evaluation/reporting/templates/full.html.jinja
new file mode 100644
index 0000000..9965ede
--- /dev/null
+++ b/vision_aid/evaluation/reporting/templates/full.html.jinja
@@ -0,0 +1,39 @@
+{% extends 'base.html.jinja' %}
+{% from 'evidence.html.jinja' import human_evidence, page_evidence %}
+{% from 'charts.html.jinja' import performance_header, detection_overview, page_comparison %}
+
+{% block document_title %}{{ title }}: {{ model_name }}; {{ manifest.run_id }}{% endblock %}
+{% block content %}
+
+
+Tool Findings and Human Finding Matches
+Each original Human finding appears once in source order. Open "View details" to read the original Human finding, then "View page decisions" to read the review decision and matched tool findings. Global findings include a separate page decisions dropdown for every page. Details and page decisions start collapsed.
+
+{% include 'controls.html.jinja' %}
+
+
+
+{% endblock %}
diff --git a/vision_aid/evaluation/reporting/templates/page.html.jinja b/vision_aid/evaluation/reporting/templates/page.html.jinja
new file mode 100644
index 0000000..c4e076b
--- /dev/null
+++ b/vision_aid/evaluation/reporting/templates/page.html.jinja
@@ -0,0 +1,35 @@
+{% extends 'base.html.jinja' %}
+{% from 'evidence.html.jinja' import human_evidence, page_evidence %}
+{% from 'charts.html.jinja' import performance_header, detection_overview, page_comparison %}
+
+{% block document_title %}{{ title }}: {{ model_name }}; {{ manifest.source_url }}{% endblock %}
+{% block scope %}{{ manifest.source_url }}
{% endblock %}
+{% block content %}
+
+
+Tool Findings and Human Finding Matches
+Open View details to read the original Human finding, then View page decisions to read the review decision and matched tool findings. Details and page decisions start collapsed.
+
+{% include 'controls.html.jinja' %}
+
+
+{% if not rows %}No Human findings were imported for this page.
{% endif %}
+
+{% endblock %}
diff --git a/vision_aid/evaluation/review.py b/vision_aid/evaluation/review.py
new file mode 100644
index 0000000..40f421c
--- /dev/null
+++ b/vision_aid/evaluation/review.py
@@ -0,0 +1,166 @@
+"""Validate editable Human review decisions and calculate deterministic counts."""
+
+from __future__ import annotations
+
+import csv
+from pathlib import Path
+from typing import Any
+
+from vision_aid.evaluation.human_finding_mapping import LOCKED_COLUMNS
+from vision_aid.evaluation.inputs.human_audit import ELIGIBILITY_VALUES, HEADERS
+
+DECISION_COLUMNS = (
+ "classification",
+ "classification_reason",
+ "llm_finding_id",
+ "programmatic_finding_id",
+ "match_reason",
+ "review_notes",
+)
+REVIEW_COLUMNS = (
+ "human_finding_id",
+ "source_sheet",
+ "source_row",
+ "page_scope",
+ "source_element",
+ *HEADERS,
+ *DECISION_COLUMNS,
+)
+
+
+def create_review(path: Path, human_findings: list[dict[str, Any]]) -> None:
+ """Create one editable row per Human finding with blank decisions."""
+ with path.open("w", encoding="utf-8", newline="") as stream:
+ writer = csv.DictWriter(stream, fieldnames=REVIEW_COLUMNS)
+ writer.writeheader()
+ for item in human_findings:
+ writer.writerow(
+ {
+ "human_finding_id": item["human_finding_id"],
+ "source_sheet": item["source_sheet"],
+ "source_row": item["source_row"],
+ "page_scope": item["page_scope"],
+ "source_element": item["source_element"],
+ **item["raw_evidence"],
+ **{column: "" for column in DECISION_COLUMNS},
+ }
+ )
+
+
+def _ids(value: str) -> tuple[str, ...]:
+ """Semicolons join multiple finding IDs in either finding column."""
+ parts = tuple(part.strip() for part in value.split(";") if part.strip())
+ if len(parts) != len(set(parts)):
+ raise ValueError("A finding ID is repeated within one review row")
+ return parts
+
+
+def reviewed_rows(
+ path: Path,
+ human_findings: list[dict[str, Any]],
+ llm: list[dict[str, Any]],
+ programmatic: list[dict[str, Any]],
+ approved: dict[str, dict[str, str]],
+) -> list[dict[str, Any]]:
+ """Validate completeness and finding kind before granting any row credit."""
+ with path.open(encoding="utf-8-sig", newline="") as stream:
+ reader = csv.DictReader(stream)
+ if not set(REVIEW_COLUMNS).issubset(reader.fieldnames or ()):
+ raise ValueError("Review CSV is missing required columns")
+ if not set(LOCKED_COLUMNS).issubset(reader.fieldnames or ()):
+ raise ValueError("Review CSV is missing approved Human finding mapping columns")
+ entered = list(reader)
+ expected = {item["human_finding_id"]: item for item in human_findings}
+ if len(entered) != len(expected):
+ raise ValueError("Review CSV must contain each Human finding exactly once")
+ by_kind = {
+ "llm": {item["finding_id"]: item for item in llm},
+ "programmatic": {item["finding_id"]: item for item in programmatic},
+ }
+ used: dict[str, set[str]] = {"llm": set(), "programmatic": set()}
+ result: dict[str, dict[str, Any]] = {}
+ for row in entered:
+ human_finding_id = row["human_finding_id"]
+ if human_finding_id not in expected or human_finding_id in result:
+ raise ValueError(f"Unknown or repeated Human finding: {human_finding_id}")
+ human_finding = expected[human_finding_id]
+ for field in ("source_sheet", "source_row", "source_element"):
+ if str(row[field]) != str(human_finding[field]):
+ raise ValueError(
+ f"Human finding identity changed for {human_finding_id}: {field}"
+ )
+ if (
+ row["page_scope"] != human_finding["page_scope"] or any(
+ row[field] != human_finding["raw_evidence"].get(field, "") for field in HEADERS
+ )
+ ):
+ raise ValueError(f"Human finding evidence changed for {human_finding_id}")
+ for column in LOCKED_COLUMNS:
+ actual, required = row[column], approved[human_finding_id][column]
+ if column == "programmatic_finding_id":
+ unchanged = set(_ids(actual)) == set(_ids(required))
+ else:
+ unchanged = actual == required
+ if not unchanged:
+ raise ValueError(
+ f"Human finding {human_finding_id}: approved Human finding mapping "
+ f"decision changed: {column}. Restore the saved decision; "
+ "review LLM matches in this run, or approve a new Human finding mapping CSV review for future runs."
+ )
+ classification = row["classification"].strip()
+ if classification not in ELIGIBILITY_VALUES:
+ raise ValueError(f"Human finding {human_finding_id}: invalid classification")
+ matched: dict[str, tuple[str, ...]] = {}
+ for kind, column, label in (
+ ("llm", "llm_finding_id", "LLM"),
+ ("programmatic", "programmatic_finding_id", "Programmatic"),
+ ):
+ ids = _ids(row[column])
+ unknown = set(ids) - set(by_kind[kind])
+ if unknown:
+ raise ValueError(
+ f"Unknown {label} finding ID: {', '.join(sorted(unknown))}"
+ )
+ reused = set(ids) & used[kind]
+ if reused:
+ raise ValueError(
+ f"Conflicting {label} finding ID: {', '.join(sorted(reused))}"
+ )
+ used[kind].update(ids)
+ matched[kind] = ids
+ if matched["llm"] and not row["match_reason"].strip():
+ raise ValueError(f"Human finding {human_finding_id} needs a match reason")
+ if classification in {"unavailable_evidence", "ambiguous"} and any(
+ matched.values()
+ ):
+ raise ValueError(f"Human finding {human_finding_id} is not testable; cannot claim coverage")
+ result[human_finding_id] = {
+ "human_finding": human_finding, "decision": row, "matches": matched,
+ }
+ return [result[identity] for identity in expected]
+
+
+def detection_counts(rows: list[dict[str, Any]]) -> dict[str, dict[str, int]]:
+ """Count validated catches and denominators without averaging page rates.
+
+ Each row represents one Human finding. Cross-catches earn Overall credit,
+ while source rates include only catches by the expected source. Not-testable
+ findings remain in the Overall denominator.
+ """
+ counts = {}
+ for classification, kind in (
+ ("llm_eligible", "llm"), ("programmatic", "programmatic")
+ ):
+ eligible = [
+ row for row in rows
+ if row["decision"]["classification"].strip() == classification
+ ]
+ counts[kind] = {
+ "caught": sum(bool(row["matches"][kind]) for row in eligible),
+ "eligible": len(eligible),
+ }
+ counts["overall"] = {
+ "caught": sum(any(row["matches"].values()) for row in rows),
+ "eligible": len(rows),
+ }
+ return counts
diff --git a/vision_aid/evaluation/schemas.py b/vision_aid/evaluation/schemas.py
new file mode 100644
index 0000000..646a9af
--- /dev/null
+++ b/vision_aid/evaluation/schemas.py
@@ -0,0 +1,50 @@
+"""Evidence records for page evaluations against the Human audit."""
+
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass
+from typing import Any, Literal
+
+
+@dataclass(frozen=True)
+class HumanFinding:
+ human_finding_id: str
+ source_sheet: str
+ source_row: int
+ page_scope: Literal["home", "global", "page"]
+ source_element: str
+ problem: str
+ raw_evidence: dict[str, str]
+ wcag_evidence: tuple[str, ...]
+
+
+@dataclass(frozen=True)
+class HumanAudit:
+ human_audit_filename: str
+ human_audit_sha256: str
+ homepage_url: str
+ human_findings: tuple[HumanFinding, ...]
+
+ def to_dict(self) -> dict[str, Any]:
+ """Return JSON-compatible source evidence."""
+ return asdict(self)
+
+
+@dataclass(frozen=True)
+class CanonicalFinding:
+ finding_id: str
+ source: Literal["llm", "programmatic"]
+ run_id: str
+ model: str | None
+ prompt: str
+ checklist: str
+ page_url: str
+ problem_family: str
+ problem: str
+ element: str
+ location: str
+ wcag_evidence: tuple[str, ...]
+ raw_source: Any
+ parse_status: str
+ # True when the tool shows this finding on screen but leaves it out of its CSV report.
+ screen_only: bool = False
diff --git a/vision_aid/ingestion/file_crawler.py b/vision_aid/ingestion/file_crawler.py
index 71049c8..a15928f 100644
--- a/vision_aid/ingestion/file_crawler.py
+++ b/vision_aid/ingestion/file_crawler.py
@@ -1,12 +1,13 @@
-import requests
-from datetime import datetime
-from typing import Optional, List, Set
-from urllib.parse import urljoin, urlparse
import re
import time
+from datetime import datetime
+from urllib.parse import urljoin, urlparse
+
+import requests
+
-def download_html(url: str, filename: Optional[str] = None, depth: int = 1,
- visited: Optional[Set[str]] = None, base_domain: Optional[str] = None) -> List[str]:
+def download_html(url: str, filename: str | None = None, depth: int = 1,
+ visited: set[str] | None = None, base_domain: str | None = None) -> list[str]:
"""
Download HTML from a URL and recursively follow links to nested pages.
@@ -97,7 +98,7 @@ def download_html(url: str, filename: Optional[str] = None, depth: int = 1,
print(f"Error downloading {url}: {e}")
return saved_files
-def extract_links(html_content: str, base_url: str) -> Set[str]:
+def extract_links(html_content: str, base_url: str) -> set[str]:
"""Extract all href links from HTML content."""
links = set()
@@ -181,11 +182,11 @@ def fetch_pages_nested(url: str, max_depth: int = 1,
"""
url = _normalize_url(url)
base_domain = urlparse(url).netloc
- visited: Set[str] = set()
- parts: List[str] = []
+ visited: set[str] = set()
+ parts: list[str] = []
tree: dict = {} # parent_url -> [child_urls]
- def _crawl(current_url: str, depth: int, parent_url: str = None) -> None:
+ def _crawl(current_url: str, depth: int, parent_url: str | None = None) -> None:
current_url = _normalize_url(current_url)
if current_url in visited or depth < 0:
return
@@ -226,4 +227,4 @@ def _crawl(current_url: str, depth: int, parent_url: str = None) -> None:
print(f"\nDownloaded {len(saved_files)} files:")
for file in saved_files:
- print(f" - {file}")
\ No newline at end of file
+ print(f" - {file}")
diff --git a/vision_aid/ingestion/pull_html.py b/vision_aid/ingestion/pull_html.py
index 6c40bb0..dae2b14 100644
--- a/vision_aid/ingestion/pull_html.py
+++ b/vision_aid/ingestion/pull_html.py
@@ -1,8 +1,9 @@
-import requests
from datetime import datetime
-from typing import Optional
-def download_html(url:str, filename:Optional[str]):
+import requests
+
+
+def download_html(url:str, filename:str | None):
try:
response = requests.get(url, headers={'User-Agent': 'Mozilla/5.0'})
response.raise_for_status()
@@ -23,4 +24,4 @@ def download_html(url:str, filename:Optional[str]):
return None
if __name__=='__main__':
- download_html("https://visionaid.org/")
\ No newline at end of file
+ download_html("https://visionaid.org/")