154 lines
7.5 KiB
Python
154 lines
7.5 KiB
Python
"""Local evaluation integrity checks, not signatures or legal approval.
|
|
|
|
Recompute observable checks from frozen inputs. Never trust a report's green
|
|
boolean alone. These files are operator-owned local evidence, not authenticated
|
|
third-party attestations and must not be accepted as an upload-to-publish token.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
from contextlib import contextmanager
|
|
from pathlib import Path
|
|
|
|
from yuqei_sdk.contract_drafting import (
|
|
ContractDraftRequest, ContractDraftOutput, draft_digest, draft_output_schema,
|
|
validate_draft_output,
|
|
)
|
|
|
|
|
|
@contextmanager
|
|
def evaluation_directory_lock(directory: Path):
|
|
"""An OS-owned, nonblocking lock; the remaining file is not a live-job signal."""
|
|
directory.mkdir(parents=True, exist_ok=True)
|
|
with (directory / ".evaluation.lock").open("a+b") as stream:
|
|
if stream.seek(0, 2) == 0:
|
|
stream.write(b"0")
|
|
stream.flush()
|
|
stream.seek(0)
|
|
acquired = False
|
|
try:
|
|
try:
|
|
if os.name == "nt":
|
|
import msvcrt
|
|
msvcrt.locking(stream.fileno(), msvcrt.LK_NBLCK, 1)
|
|
else:
|
|
import fcntl
|
|
fcntl.flock(stream.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
acquired = True
|
|
except OSError as error:
|
|
raise ValueError("Evaluation directory is locked by a live runner; do not start a duplicate.") from error
|
|
yield
|
|
finally:
|
|
if acquired:
|
|
stream.seek(0)
|
|
if os.name == "nt":
|
|
msvcrt.locking(stream.fileno(), msvcrt.LK_UNLCK, 1)
|
|
else:
|
|
fcntl.flock(stream.fileno(), fcntl.LOCK_UN)
|
|
|
|
|
|
def text_sha256(value: str) -> str:
|
|
return hashlib.sha256(value.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def safe_case_id(value: str) -> str:
|
|
if not isinstance(value, str) or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_-]{0,119}", value):
|
|
raise ValueError("Invalid case identifier; a bounded filename-safe ID is required.")
|
|
if value.upper() in {"CON", "PRN", "AUX", "NUL", *(f"COM{i}" for i in range(10)), *(f"LPT{i}" for i in range(10))}:
|
|
raise ValueError("Reserved case identifier.")
|
|
return value
|
|
|
|
|
|
def select_cases(frozen: dict, case_id: str | None = None, prefix: str | None = None) -> list[dict]:
|
|
if case_id and prefix:
|
|
raise ValueError("Choose an exact case or a prefix, not both.")
|
|
if frozen.get("synthetic_only") is not True or frozen.get("model") != "gpt-5.6-sol":
|
|
raise ValueError("Only explicitly synthetic gpt-5.6-sol cases are supported.")
|
|
for value in (case_id, prefix):
|
|
if value is not None:
|
|
safe_case_id(value)
|
|
cases = frozen.get("cases")
|
|
if not isinstance(cases, list) or not cases:
|
|
raise ValueError("Frozen cases are missing.")
|
|
identifiers = [safe_case_id(case["id"]) for case in cases]
|
|
if len(set(identifiers)) != len(identifiers):
|
|
raise ValueError("Duplicate case identifiers.")
|
|
selected = [case for case in cases if (not case_id or case["id"] == case_id)
|
|
and (not prefix or case["id"].startswith(prefix))]
|
|
if not selected:
|
|
raise ValueError("Unknown case selection.")
|
|
return selected
|
|
|
|
|
|
def prepare_targets(directory: Path, cases: list[dict], suffix: str | None) -> tuple[list[Path], Path]:
|
|
"""Check the entire batch before any model call, including the summary."""
|
|
targets = [directory / f"{safe_case_id(case['id'])}.json" for case in cases]
|
|
summary = directory / (f"summary-{safe_case_id(suffix)}.json" if suffix else "summary.json")
|
|
if len({target.resolve() for target in [*targets, summary]}) != len(targets) + 1:
|
|
raise ValueError("Result and summary filenames overlap.")
|
|
for target in [*targets, summary]:
|
|
if target.exists():
|
|
raise ValueError(f"Result exists: {target}; choose a new directory, never overwrite evidence.")
|
|
directory.mkdir(parents=True, exist_ok=True)
|
|
return targets, summary
|
|
|
|
|
|
def write_evidence(path: Path, data: dict):
|
|
# Exclusive create also prevents races with another runner after preflight.
|
|
with path.open("x", encoding="utf-8") as stream:
|
|
json.dump(data, stream, ensure_ascii=False, indent=2, allow_nan=False)
|
|
|
|
|
|
def draft_artifact_errors(frozen: dict, case: dict, artifact: dict) -> list[str]:
|
|
"""Revalidate input, model, prompt, template, facts and declared expectations."""
|
|
errors = []
|
|
try:
|
|
selected = select_cases(frozen, case["id"])
|
|
if draft_digest(selected[0]) != draft_digest(case):
|
|
errors.append("case_not_in_frozen_suite")
|
|
request = ContractDraftRequest.model_validate(case["request"])
|
|
prompt = frozen["prompt_snapshot"]
|
|
template = prompt["template"]
|
|
if prompt.get("snapshot_version") != "prompt-execution.v1" or text_sha256(template["content"]) != prompt["content_sha256"]:
|
|
errors.append("invalid_frozen_prompt")
|
|
if artifact.get("case_id") != case["id"] or artifact.get("synthetic_only") is not True:
|
|
errors.append("artifact_identity_mismatch")
|
|
if artifact.get("frozen_file_sha256") != draft_digest(frozen) or artifact.get("input_sha256") != draft_digest(case["request"]):
|
|
errors.append("artifact_input_mismatch")
|
|
job = artifact["job"]
|
|
if (job.get("operation") != "contract_draft" or job.get("tenant_id") != "isolated-eval"
|
|
or job.get("created_by") != "synthetic-evaluator"):
|
|
errors.append("job_scope_mismatch")
|
|
if (artifact.get("real_model_call") is not True or job.get("status") != "succeeded"
|
|
or job.get("provider_status") != "succeeded" or job.get("fallback_used")
|
|
or job.get("quality_status") != "passed"):
|
|
errors.append("draft_execution_not_passed")
|
|
result = job["result"]
|
|
output = ContractDraftOutput.model_validate(result["output"])
|
|
errors.extend(f"output:{finding.code}" for finding in validate_draft_output(request, output) if finding.severity == "error")
|
|
evidence = result["evidence"]
|
|
expected_evidence = {
|
|
"schema_version": output.schema_version, "schema_sha256": draft_digest(draft_output_schema(request)),
|
|
"input_sha256": draft_digest(request.model_dump(mode="json")),
|
|
"prompt_name": template["name"], "prompt_version": template["version"], "prompt_sha256": prompt["content_sha256"],
|
|
"template_sha256": request.template_snapshot.get("content_sha256"), "playbook_sha256": draft_digest(request.playbook_snapshot),
|
|
"model": "gpt-5.6-sol", "provider": "codex-cli-development", "fact_bindings_checked": True,
|
|
}
|
|
for name, expected in expected_evidence.items():
|
|
if evidence.get(name) != expected:
|
|
errors.append(f"evidence_mismatch:{name}")
|
|
if result.get("validation_findings"):
|
|
errors.append("draft_validation_findings")
|
|
if output.status != case["expected_status"] or not set(case["expected_missing"]).issubset({item.field for item in output.missing_fields}):
|
|
errors.append("draft_expectation_mismatch")
|
|
if case.get("expected_missing") and output.status != "incomplete":
|
|
errors.append("missing_fact_not_incomplete")
|
|
if (artifact.get("enterprise_rule_check") or {}).get("errors"):
|
|
errors.append("enterprise_rules_failed")
|
|
except (KeyError, TypeError, ValueError, AttributeError) as error:
|
|
errors.append(f"malformed_evidence:{type(error).__name__}")
|
|
return sorted(set(errors))
|