yuqei/ai-contract-platform
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

fix(rc05): harden canary and resume controls

Browse Source
This commit is contained in:
yuqei
2026-10-06 21:51:03 +08:00
parent 65ee37a527
commit f7d5946c42
6 changed files with 621 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
yuqei-infra/scripts/multiscene_call_ledger.py
+293 -15
View File
@@ -25,6 +25,22 @@ TERMINAL_STATUSES = frozenset({"succeeded", "failed", "cancelled"})
MAX_PREALLOCATED_ATTEMPTS = 280
MAX_REPAIR_ATTEMPTS = 16
REPAIR_OPERATIONS = frozenset({"repair_draft", "repair_review"})
REAL_PROVIDER_NAME = "codex-cli-development"
REAL_PROVIDER_MODEL = "gpt-5.6-sol"
AFTER_FILES = {
"draft": frozenset({"01-draft-after.json"}),
"semantic_review": frozenset({"02-semantic-after.json"}),
"repair_draft": frozenset({"03-repair-after.json"}),
"repair_review": frozenset({"04-post-review-after.json"}),
"specialist_review": frozenset({"02-specialist-after.json", "02-after.json"}),
"intake": frozenset({"02-after.json"}),
}
JOB_OPERATIONS = {
"contract_draft": frozenset({"draft", "repair_draft"}),
"contract_draft_semantic_review": frozenset({"semantic_review", "repair_review"}),
"contract_review": frozenset({"specialist_review"}),
"contract_intake": frozenset({"intake"}),
}
LEDGER_SCHEMA = "multiscene-call-ledger.v2"
CAMPAIGN_SCHEMA = "multiscene-call-campaign.v2"
CONTROL_SCHEMA = "multiscene-release-control-state.v1"
@@ -209,6 +225,7 @@ class CallLedger:
def control_snapshot(self) -> dict[str, object]:
with self._locked():
self._materialize_attempt_breakers_unlocked()
return self._control_snapshot_unlocked()
def authorize_selection(self, selected: list[str], *, resume: bool,
@@ -258,6 +275,7 @@ class CallLedger:
if state["blocked_failure_classes"]:
raise CallLedgerError("failed or unknown canary cannot authorize full dispatch")
records = self._records()
binding_sha256s: dict[str, str] = {}
for case_id in canaries:
relevant = [record for (current, _operation), record in records.items()
if current == case_id]
@@ -265,12 +283,15 @@ class CallLedger:
or record["finish"]["status"] != "succeeded"
for record in relevant):
raise CallLedgerError("canary provider attempts are incomplete or unsuccessful")
binding = _read_json(self.case_outcomes / f"{case_id}.json")
binding = self._validated_persisted_outcome_unlocked(
case_id, records, require_success=True)
if binding.get("outcome_sha256") != normalized[case_id]:
raise CallLedgerError("canary outcome does not match immutable binding")
binding_sha256s[case_id] = _digest(binding)
payload = {"schema_version": CONTROL_SCHEMA,
"event": "canary_to_full", "ledger_identity_sha256": self.identity_sha256,
"canary_case_ids": canaries, "outcome_sha256s": normalized}
"canary_case_ids": canaries, "outcome_sha256s": normalized,
"case_outcome_binding_sha256s": binding_sha256s}
transition = self.control_events / "canary-to-full.json"
if transition.exists():
if _read_json(transition) != payload:
@@ -302,11 +323,29 @@ class CallLedger:
def _materialize_attempt_breakers_unlocked(self) -> None:
if not self.controls_enabled:
return
for (case_id, operation), record in self._records().items():
status = (record["finish"]["status"] if record["finish"] is not None
else "unknown")
if status == "succeeded":
continue
records = self._records()
cases = sorted({case_id for case_id, _operation in records})
for case_id in cases:
relevant = [record for (current, _operation), record in records.items()
if current == case_id]
unfinished = any(record["finish"] is None for record in relevant)
unsuccessful = next((record["finish"]["status"] for record in relevant
if record["finish"] is not None
and record["finish"]["status"] != "succeeded"), None)
if unfinished:
status, reason = "unknown", "ledger has an unfinished provider attempt"
elif unsuccessful is not None:
status, reason = unsuccessful, f"provider attempt finished as {unsuccessful}"
else:
try:
self._validated_persisted_outcome_unlocked(
case_id, records, require_success=True)
except (OSError, UnicodeError, TypeError, ValueError,
RuntimeError, CallLedgerError) as error:
status = "unknown"
reason = f"successful provider attempt has invalid case outcome: {error}"
else:
continue
failure_class = self.release_controls["failure_classes"][case_id]
path = self.control_events / f"failure-class--{failure_class}.json"
if not path.exists():
@@ -314,7 +353,7 @@ class CallLedger:
"event": "failure_class_blocked",
"ledger_identity_sha256": self.identity_sha256,
"failure_class": failure_class, "case_id": case_id,
"status": status, "reason": f"ledger attempt {operation} is {status}"})
"status": status, "reason": reason[:1000]})
def _control_snapshot_unlocked(self) -> dict[str, object]:
if not self.controls_enabled:
@@ -323,6 +362,7 @@ class CallLedger:
"canary_transition_sha256": None}
blocked: list[str] = []
transition_sha = None
records = self._records()
for path in self.control_events.iterdir():
_reject_link_or_reparse(path, "control event")
if not path.is_file():
@@ -333,11 +373,48 @@ class CallLedger:
or payload.get("ledger_identity_sha256") != self.identity_sha256):
raise CallLedgerError("control event is invalid")
if path.name == "canary-to-full.json" and payload.get("event") == "canary_to_full":
expected_keys = {
"schema_version", "event", "ledger_identity_sha256",
"canary_case_ids", "outcome_sha256s",
"case_outcome_binding_sha256s",
}
canaries = self.release_controls["canary_case_ids"]
outcomes = payload.get("outcome_sha256s")
bindings = payload.get("case_outcome_binding_sha256s")
if (set(payload) != expected_keys
or payload.get("canary_case_ids") != canaries
or not isinstance(outcomes, dict) or set(outcomes) != set(canaries)
or not isinstance(bindings, dict) or set(bindings) != set(canaries)):
raise CallLedgerError("canary transition shape or inventory is invalid")
for case_id in canaries:
outcome_sha = _sha(outcomes[case_id], "canary transition outcome SHA-256")
binding_sha = _sha(
bindings[case_id], "canary transition binding SHA-256")
binding = self._validated_persisted_outcome_unlocked(
case_id, records, require_success=True)
if (binding["outcome_sha256"] != outcome_sha
or _digest(binding) != binding_sha):
raise CallLedgerError(
"canary transition differs from immutable outcome binding")
transition_sha = _digest(payload)
elif (re.fullmatch(r"failure-class--[A-Za-z0-9._-]+\.json", path.name)
and payload.get("event") == "failure_class_blocked"):
expected_keys = {
"schema_version", "event", "ledger_identity_sha256",
"failure_class", "case_id", "status", "reason",
}
if set(payload) != expected_keys:
raise CallLedgerError("failure-class control event shape is invalid")
failure_class = _identifier(payload.get("failure_class"), "failure class")
if failure_class not in set(self.release_controls["failure_classes"].values()):
case_id = _identifier(payload.get("case_id"), "failure event case ID")
expected_name = f"failure-class--{failure_class}.json"
if (path.name != expected_name or case_id not in self.case_limits
or self.release_controls["failure_classes"][case_id] != failure_class
or payload.get("status") not in {"failed", "cancelled", "unknown"}
or not isinstance(payload.get("reason"), str)
or not payload["reason"] or len(payload["reason"]) > 1000
or failure_class not in set(
self.release_controls["failure_classes"].values())):
raise CallLedgerError("control event failure class is outside frozen controls")
blocked.append(failure_class)
else:
@@ -454,11 +531,41 @@ class CallLedger:
def snapshot(self) -> dict:
with self._locked():
self._materialize_attempt_breakers_unlocked()
records = self._records()
successful_case_evidence: dict[str, bool] = {}
for case_id in self.case_limits if self.controls_enabled else ():
relevant = [record for (current, _operation), record in records.items()
if current == case_id]
if not relevant or any(record["finish"] is None
or record["finish"]["status"] != "succeeded"
for record in relevant):
continue
try:
self._validated_persisted_outcome_unlocked(
case_id, records, require_success=True)
except (OSError, UnicodeError, TypeError, ValueError,
RuntimeError, CallLedgerError):
successful_case_evidence[case_id] = False
else:
successful_case_evidence[case_id] = True
def evidence_terminal(case_id: str, record: dict) -> bool:
finish = record["finish"]
if finish is None:
return False
if not self.controls_enabled:
return True
if finish["status"] != "succeeded":
return True
return successful_case_evidence.get(case_id, False)
operations = {}
for operation in OPERATIONS:
relevant = [record for (case, op), record in records.items() if op == operation]
terminal = sum(record["finish"] is not None for record in relevant)
relevant = [(case, record) for (case, op), record in records.items()
if op == operation]
terminal = sum(evidence_terminal(case, record)
for case, record in relevant)
operations[operation] = {
"limit": self.operation_limits.get(operation, 0),
"attempted": len(relevant), "terminal": terminal,
@@ -467,14 +574,39 @@ class CallLedger:
cases = {}
for case_id, limit in sorted(self.case_limits.items()):
relevant = [record for (case, _op), record in records.items() if case == case_id]
terminal = sum(record["finish"] is not None for record in relevant)
terminal = sum(evidence_terminal(case_id, record) for record in relevant)
cases[case_id] = {"limit": limit, "attempted": len(relevant),
"terminal": terminal, "unknown": len(relevant) - terminal}
attempted = len(records)
provider_terminal = sum(
record["finish"] is not None for record in records.values())
terminal = sum(evidence_terminal(case_id, record)
for (case_id, _operation), record in records.items())
campaign_records = self._campaign_records() if self.registry_root is not None else records
campaign_attempted = len(campaign_records)
campaign_terminal = sum(
campaign_provider_terminal = sum(
record["finish"] is not None for record in campaign_records.values())
def campaign_evidence_terminal(record: dict) -> bool:
finish = record["finish"]
if finish is None:
return False
if finish["status"] != "succeeded":
return True
claim = record["claim"]
if self.registry_root is None:
return evidence_terminal(claim["case_id"], record)
if claim["plan_sha256"] != self.plan_sha256:
# Foreign-plan attempts already fail the current candidate
# gate and their output directory is deliberately not read.
return True
local = records.get((claim["case_id"], claim["operation"]))
return bool(local is not None
and evidence_terminal(claim["case_id"], local))
campaign_terminal = sum(
campaign_evidence_terminal(record)
for record in campaign_records.values())
if self.registry_root is not None:
current_plan_attempted = sum(
record["claim"]["plan_sha256"] == self.plan_sha256
@@ -544,14 +676,16 @@ class CallLedger:
"campaign": {"attempted": campaign_attempted,
"terminal": campaign_terminal,
"unknown": campaign_attempted - campaign_terminal,
"provider_terminal": campaign_provider_terminal,
"current_plan_attempted": current_plan_attempted,
"foreign_plan_attempted": (
campaign_attempted - current_plan_attempted
),
"provider_outcomes": campaign_provider_outcomes},
"attempted": attempted,
"terminal": sum(record["finish"] is not None for record in records.values()),
"unknown": sum(record["finish"] is None for record in records.values()),
"terminal": terminal,
"unknown": attempted - terminal,
"provider_terminal": provider_terminal,
"release_controls": controls}
def attempt_records(self) -> list[dict[str, object]]:
@@ -659,6 +793,150 @@ class CallLedger:
if payload != expected:
raise CallLedgerError("case outcome or evidence changed after binding")
def _validated_persisted_outcome_unlocked(
self,
case_id: str,
records: Mapping[tuple[str, str], dict],
*,
require_success: bool,
) -> dict[str, object]:
"""Validate a persisted case outcome and every file it binds.
This is intentionally independent of the caller-provided hashes used by
``verify_case_outcome``. Resume authorization must rediscover the files
from the ledger and fail closed after a crash or later mutation.
"""
case_id = _identifier(case_id, "persisted outcome case ID")
binding_path = self.case_outcomes / f"{case_id}.json"
_reject_link_or_reparse(binding_path, "case outcome binding")
if not binding_path.is_file():
raise CallLedgerError("case outcome binding is missing")
payload = _read_json(binding_path)
expected_keys = {
"schema_version", "ledger_identity_sha256", "case_id",
"outcome_file", "outcome_sha256", "case_attempts_sha256",
"expected_operations", "case_operation_projection",
"evidence_sha256s",
}
if not isinstance(payload, dict) or set(payload) != expected_keys:
raise CallLedgerError("case outcome binding shape is invalid")
outcome_file = payload.get("outcome_file")
if outcome_file not in {"result.json", "failure.json"}:
raise CallLedgerError("case outcome binding file is invalid")
outcome_sha = _sha(payload.get("outcome_sha256"), "case outcome SHA-256")
expected_operations = _operation_projection(payload.get("expected_operations"))
evidence = _evidence_hashes(payload.get("evidence_sha256s"))
projection = _case_attempt_projection(records, case_id)
expected = {
"schema_version": "multiscene-case-outcome.v2",
"ledger_identity_sha256": self.identity_sha256,
"case_id": case_id,
"outcome_file": outcome_file,
"outcome_sha256": outcome_sha,
"expected_operations": expected_operations,
"case_operation_projection": projection,
"case_attempts_sha256": _digest(projection),
"evidence_sha256s": evidence,
}
if payload != expected:
raise CallLedgerError("case outcome binding no longer matches its attempts")
case_dir = self.output_dir / case_id
_reject_link_or_reparse(case_dir, "case outcome directory")
if not case_dir.is_dir():
raise CallLedgerError("case outcome directory is missing")
for candidate_name in ("result.json", "failure.json"):
_reject_link_or_reparse(case_dir / candidate_name, "case outcome artifact")
candidates = [case_dir / name for name in ("result.json", "failure.json")
if (case_dir / name).is_file()]
if len(candidates) != 1 or candidates[0].name != outcome_file:
raise CallLedgerError("case outcome artifact is missing or ambiguous")
outcome_path = candidates[0]
if hashlib.sha256(outcome_path.read_bytes()).hexdigest() != outcome_sha:
raise CallLedgerError("case outcome artifact digest mismatch")
outcome = _read_json(outcome_path)
if not isinstance(outcome, dict):
raise CallLedgerError("case outcome artifact must be an object")
calls = outcome.get("provider_call_count")
if (outcome.get("case_id") != case_id
or not isinstance(calls, int) or isinstance(calls, bool)
or calls != len(projection)):
raise CallLedgerError("case outcome artifact identity or call count is invalid")
actual_after_names: set[str] = set()
for child in case_dir.iterdir():
_reject_link_or_reparse(child, "case evidence artifact")
if child.name.endswith("-after.json"):
if not child.is_file():
raise CallLedgerError("case evidence artifact is not a regular file")
actual_after_names.add(child.name)
if actual_after_names != set(evidence):
raise CallLedgerError("case evidence inventory differs from immutable binding")
evidence_by_operation: dict[str, dict] = {}
for name, digest in evidence.items():
evidence_path = case_dir / name
if (not evidence_path.is_file()
or hashlib.sha256(evidence_path.read_bytes()).hexdigest() != digest):
raise CallLedgerError("case evidence artifact digest mismatch")
evidence_payload = _read_json(evidence_path)
if not isinstance(evidence_payload, dict):
raise CallLedgerError("case evidence artifact must be an object")
operation = evidence_payload.get("operation")
if operation not in OPERATIONS or operation in evidence_by_operation:
raise CallLedgerError("case evidence operation is invalid or duplicate")
if name not in AFTER_FILES[operation]:
raise CallLedgerError("case evidence file name does not match its operation")
evidence_by_operation[operation] = evidence_payload
if require_success:
if (outcome_file != "result.json" or outcome.get("status") != "completed"
or outcome.get("passed") is not True
or outcome.get("automated_quality_passed") is not True
or outcome.get("automated_model_quality_release_passed") is not True
or outcome.get("content_quality_release_passed") is not True):
raise CallLedgerError("successful case outcome quality gate is incomplete")
if any(item["status"] != "succeeded" for item in projection):
raise CallLedgerError("successful case outcome has a non-successful provider attempt")
actual_operations = [item["operation"] for item in projection]
actual_set, expected_set = set(actual_operations), set(expected_operations)
repair_actual = actual_set & REPAIR_OPERATIONS
mandatory = expected_set - REPAIR_OPERATIONS
if (not projection or len(actual_operations) != len(actual_set)
or not actual_set <= expected_set or not mandatory <= actual_set
or repair_actual not in (set(), set(REPAIR_OPERATIONS))
or set(evidence_by_operation) != actual_set):
raise CallLedgerError("successful case outcome operation coverage is incomplete")
for item in projection:
after = evidence_by_operation[item["operation"]]
elapsed = after.get("execution_elapsed_seconds")
job = after.get("job")
audit = after.get("audit")
if (not isinstance(elapsed, (int, float)) or isinstance(elapsed, bool)
or elapsed < 0 or not isinstance(job, dict)
or not isinstance(audit, dict)):
raise CallLedgerError("case evidence measurements are invalid")
job_id = job.get("id")
audit_id = audit.get("id")
audit_run_id = job.get("audit_run_id")
job_operation = job.get("operation")
if (after.get("case_id") != case_id
or after.get("request_sha256") != item["request_sha256"]
or after.get("response_sha256") != item["response_sha256"]
or after.get("provider_invocation_count") != 1
or job.get("status") != "succeeded"
or job.get("provider_status") != "succeeded"
or job_operation not in JOB_OPERATIONS
or item["operation"] not in JOB_OPERATIONS[job_operation]
or not isinstance(job_id, str) or not job_id
or not isinstance(audit_id, str) or not audit_id
or not isinstance(audit_run_id, str) or not audit_run_id
or audit_id != audit_run_id
or audit.get("provider") != REAL_PROVIDER_NAME
or audit.get("model") != REAL_PROVIDER_MODEL
or audit.get("provider_status") != "succeeded"):
raise CallLedgerError("case evidence differs from immutable provider attempt")
return payload
def _records(self) -> dict[tuple[str, str], dict]:
_reject_link_or_reparse(self.root / "identity.json", "ledger identity")
expected_identity = _read_json(self.root / "identity.json")
yuqei-infra/scripts/multiscene_eval_plan.py
+19 -3
View File
@@ -117,9 +117,25 @@ def _artifact_bytes(path: Path, root: Path, label: str) -> bytes:
def _git(root: Path, *args: str, binary: bool = False):
environment = {**os.environ, "GIT_CONFIG_NOSYSTEM": "1",
"GIT_CONFIG_GLOBAL": os.devnull, "GIT_OPTIONAL_LOCKS": "0",
"GIT_NO_REPLACE_OBJECTS": "1"}
# Git's repository-selection variables override ``-C``. A release shell
# that happens to export GIT_DIR/GIT_WORK_TREE (or Git's counted config
# injection variables) must never redirect candidate verification to a
# different clean checkout. Build a small execution environment instead
# of trying to maintain an incomplete deny-list of GIT_* variables.
inherited = (
"PATH", "PATHEXT", "SystemRoot", "SYSTEMROOT", "WINDIR", "COMSPEC",
"TEMP", "TMP", "TMPDIR",
)
environment = {name: os.environ[name] for name in inherited if name in os.environ}
environment.update({
"GIT_CONFIG_NOSYSTEM": "1",
"GIT_CONFIG_GLOBAL": os.devnull,
"GIT_OPTIONAL_LOCKS": "0",
"GIT_NO_REPLACE_OBJECTS": "1",
"GIT_TERMINAL_PROMPT": "0",
"LANG": "C",
"LC_ALL": "C",
})
completed = subprocess.run(
["git", "-C", str(root), *args], capture_output=True,
text=not binary, check=False, env=environment,
yuqei-infra/scripts/multiscene_eval_runner.py
+18 -13
View File
@@ -1152,19 +1152,6 @@ def execute(runner, plan_path, expected_sha, output_dir, factory, *, resume=Fals
ledger.trip_failure_class(key, "failed", failure["error"][:1000])
_bind_case_outcome(runner, out, ledger, case, "failure.json")
break # Do not consume the remainder after a newly observed failure.
if real and control_before["phase"] == "canary_required":
canaries = plan["release_controls"]["canary_case_ids"]
canary_rows = [_existing_case(runner, out, inventory[key], ledger.snapshot(), ledger)
for key in canaries]
if all(row.get("status") == "completed" and row.get("passed") is True
and row.get("automated_quality_passed") is True
and row.get("automated_model_quality_release_passed") is True
and row.get("content_quality_release_passed") is True
for row in canary_rows):
ledger.complete_canary({
key: runner.sha256((out / key / "result.json").read_bytes())
for key in canaries
})
# A drift during the final case must not be hidden by a green summary.
if real:
final_binding, final_source_index = load_candidate_binding(
@@ -1238,6 +1225,24 @@ def execute(runner, plan_path, expected_sha, output_dir, factory, *, resume=Fals
)
historical_release_eligible, historical_remediation = _historical_remediation(
out, ledger, rows)
canary_gate_passed = None
if real and control_before["phase"] == "canary_required":
canaries = plan["release_controls"]["canary_case_ids"]
canary_gate_passed = bool(
selected == canaries
and current_selection_passed
and selected_release_quality_passed
)
# Persist authorization only after provider/business outcomes, immutable
# result/evidence bindings, mandatory operation coverage, and every
# quality gate above have all been evaluated. A failing summary must
# never leave behind a green transition for the next resume.
if canary_gate_passed:
ledger.complete_canary({
key: runner.sha256((out / key / "result.json").read_bytes())
for key in canaries
})
snapshot = ledger.snapshot()
overall_release_passed = (matrix_execution_passed
and automated_model_quality_release_passed
and historical_release_eligible)
yuqei-infra/tests/test_multiscene_call_ledger.py
+227 -6
View File
@@ -1,3 +1,4 @@
import hashlib
import importlib.util
import json
import subprocess
@@ -78,6 +79,61 @@ def _controls(case_classes=None, canaries=None):
"failure_classes": case_classes}
def _write_json(path, value):
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(value, ensure_ascii=False, sort_keys=True,
separators=(",", ":")), encoding="utf-8")
return hashlib.sha256(path.read_bytes()).hexdigest()
def _bind_success(book, root, case_id, expected_operations):
records = [record for record in book.attempt_records()
if record["case_id"] == case_id]
evidence = {}
names = {
"draft": "01-draft-after.json",
"semantic_review": "02-semantic-after.json",
"repair_draft": "03-repair-after.json",
"repair_review": "04-post-review-after.json",
"specialist_review": "02-specialist-after.json",
"intake": "02-after.json",
}
jobs = {
"draft": "contract_draft",
"semantic_review": "contract_draft_semantic_review",
"repair_draft": "contract_draft",
"repair_review": "contract_draft_semantic_review",
"specialist_review": "contract_review",
"intake": "contract_intake",
}
for index, record in enumerate(records, 1):
name = names[record["operation"]]
run_id = f"audit-{index}"
evidence[name] = _write_json(root / case_id / name, {
"case_id": case_id,
"operation": record["operation"],
"request_sha256": record["request_sha256"],
"response_sha256": record["response_sha256"],
"provider_invocation_count": 1,
"execution_elapsed_seconds": 0.25,
"job": {"id": f"job-{index}", "audit_run_id": run_id,
"operation": jobs[record["operation"]], "status": "succeeded",
"provider_status": "succeeded"},
"audit": {"id": run_id, "provider": "codex-cli-development",
"model": "gpt-5.6-sol", "provider_status": "succeeded"},
})
outcome_sha = _write_json(root / case_id / "result.json", {
"case_id": case_id, "status": "completed", "passed": True,
"provider_call_count": len(records),
"automated_quality_passed": True,
"automated_model_quality_release_passed": True,
"content_quality_release_passed": True,
})
book.bind_case_outcome(
case_id, "result.json", outcome_sha, evidence, expected_operations)
return outcome_sha
def test_claim_precedes_call_and_finish_is_immutable(tmp_path):
calls = []
book = ledger(tmp_path)
@@ -106,12 +162,10 @@ def test_canary_transition_is_explicit_persisted_and_required_before_full(tmp_pa
assert state["phase"] == "canary_required"
token = book.claim("purchase", "draft", SHA_A)
book.finish(token, "succeeded", SHA_B)
outcome = tmp_path / "purchase-result.json"
outcome.write_text("{}", encoding="utf-8")
book.bind_case_outcome("purchase", "result.json", SHA_A, {}, ["draft"])
outcome_sha = _bind_success(book, tmp_path, "purchase", ["draft"])
with pytest.raises(LEDGER.CallLedgerError, match="outcome does not match"):
book.complete_canary({"purchase": SHA_B})
book.complete_canary({"purchase": SHA_A})
book.complete_canary({"purchase": outcome_sha})
assert book.control_snapshot()["phase"] == "full_authorized"
resumed = ledger(tmp_path, release_controls=_controls())
assert resumed.control_snapshot()["phase"] == "full_authorized"
@@ -126,8 +180,8 @@ def test_failure_class_breaker_persists_and_unknown_stays_in_denominator(tmp_pat
book.authorize_selection(["purchase"], resume=False, explicit_selection=True)
token = book.claim("purchase", "draft", SHA_A)
book.finish(token, "succeeded", SHA_B)
book.bind_case_outcome("purchase", "result.json", SHA_A, {}, ["draft"])
book.complete_canary({"purchase": SHA_A})
outcome_sha = _bind_success(book, tmp_path, "purchase", ["draft"])
book.complete_canary({"purchase": outcome_sha})
book.claim("service", "draft", SHA_A) # interrupted provider call => unknown
resumed = ledger(tmp_path, case_limits={key: 1 for key in classes},
operation_limits={"draft": 3},
@@ -139,6 +193,173 @@ def test_failure_class_breaker_persists_and_unknown_stays_in_denominator(tmp_pat
assert snapshot["release_controls"]["blocked_failure_classes"] == ["specialist"]
def test_forged_canary_transition_with_missing_fields_is_rejected(tmp_path):
book = ledger(tmp_path, release_controls=_controls())
(book.control_events / "canary-to-full.json").write_text(json.dumps({
"schema_version": LEDGER.CONTROL_SCHEMA,
"event": "canary_to_full",
"ledger_identity_sha256": book.identity_sha256,
}), encoding="utf-8")
with pytest.raises(LEDGER.CallLedgerError, match="shape or inventory"):
book.control_snapshot()
def test_hash_consistent_canary_transition_rejects_non_terminal_business_evidence(tmp_path):
book = ledger(tmp_path, release_controls=_controls())
token = book.claim("purchase", "draft", SHA_A)
book.finish(token, "succeeded", SHA_B)
outcome_sha = _bind_success(book, tmp_path, "purchase", ["draft"])
binding_path = tmp_path / ".call-ledger/case-outcomes/purchase.json"
binding = json.loads(binding_path.read_text(encoding="utf-8"))
evidence_path = tmp_path / "purchase/01-draft-after.json"
after = json.loads(evidence_path.read_text(encoding="utf-8"))
after["job"]["status"] = "failed"
evidence_sha = _write_json(evidence_path, after)
binding["evidence_sha256s"][evidence_path.name] = evidence_sha
_write_json(binding_path, binding)
transition = {
"schema_version": LEDGER.CONTROL_SCHEMA,
"event": "canary_to_full",
"ledger_identity_sha256": book.identity_sha256,
"canary_case_ids": ["purchase"],
"outcome_sha256s": {"purchase": outcome_sha},
"case_outcome_binding_sha256s": {
"purchase": LEDGER._digest(binding),
},
}
_write_json(tmp_path / ".call-ledger/control-events/canary-to-full.json",
transition)
with pytest.raises(LEDGER.CallLedgerError, match="immutable provider attempt"):
book.control_snapshot()
def test_failure_class_event_requires_exact_shape_and_case_mapping(tmp_path):
book = ledger(tmp_path, release_controls=_controls())
(book.control_events / "failure-class--contract.json").write_text(json.dumps({
"schema_version": LEDGER.CONTROL_SCHEMA,
"event": "failure_class_blocked",
"ledger_identity_sha256": book.identity_sha256,
"failure_class": "contract", "case_id": "service",
"status": "unknown", "reason": "forged", "unexpected": True,
}), encoding="utf-8")
with pytest.raises(LEDGER.CallLedgerError):
book.control_snapshot()
@pytest.mark.parametrize("mutation", ["extra", "order", "outcome", "binding"])
def test_persisted_canary_transition_is_strict_and_revalidated(tmp_path, mutation):
root = tmp_path / mutation
classes = {"purchase": "contract", "service": "specialist"}
book = ledger(root, operation_limits={"draft": 2},
case_limits={key: 1 for key in classes},
release_controls=_controls(classes, ["purchase", "service"]))
book.authorize_selection(["purchase", "service"], resume=False,
explicit_selection=True)
outcomes = {}
for case_id in ("purchase", "service"):
token = book.claim(case_id, "draft", SHA_A)
book.finish(token, "succeeded", SHA_B)
outcomes[case_id] = _bind_success(book, root, case_id, ["draft"])
book.complete_canary(outcomes)
transition_path = book.control_events / "canary-to-full.json"
transition = json.loads(transition_path.read_text(encoding="utf-8"))
if mutation == "extra":
transition["unexpected"] = True
transition_path.write_text(json.dumps(transition), encoding="utf-8")
elif mutation == "order":
transition["canary_case_ids"].reverse()
transition_path.write_text(json.dumps(transition), encoding="utf-8")
elif mutation == "outcome":
transition["outcome_sha256s"]["purchase"] = SHA_A
transition_path.write_text(json.dumps(transition), encoding="utf-8")
else:
binding_path = book.case_outcomes / "purchase.json"
binding = json.loads(binding_path.read_text(encoding="utf-8"))
binding["outcome_sha256"] = SHA_A
binding_path.write_text(json.dumps(binding), encoding="utf-8")
with pytest.raises(LEDGER.CallLedgerError):
book.control_snapshot()
def test_canary_missing_mandatory_operation_cannot_authorize_full(tmp_path):
book = ledger(tmp_path, operation_limits={"draft": 1, "semantic_review": 1},
case_limits={"purchase": 2},
release_controls=_controls({"purchase": "contract"}, ["purchase"]))
book.authorize_selection(["purchase"], resume=False, explicit_selection=True)
token = book.claim("purchase", "draft", SHA_A)
book.finish(token, "succeeded", SHA_B)
outcome_sha = _bind_success(
book, tmp_path, "purchase", ["draft", "semantic_review"])
with pytest.raises(LEDGER.CallLedgerError, match="failed or unknown canary"):
book.complete_canary({"purchase": outcome_sha})
assert not (book.control_events / "canary-to-full.json").exists()
assert book.snapshot()["unknown"] == 1
@pytest.mark.parametrize("mutation", ["result", "evidence", "binding"])
def test_resume_materializes_invalid_success_outcome_as_unknown_breaker(
tmp_path, mutation,
):
root = tmp_path / mutation
classes = {"purchase": "canary", "service": "specialist", "third": "specialist"}
values = dict(operation_limits={"draft": 3},
case_limits={key: 1 for key in classes},
release_controls=_controls(classes, ["purchase"]))
book = ledger(root, **values)
book.authorize_selection(["purchase"], resume=False, explicit_selection=True)
token = book.claim("purchase", "draft", SHA_A)
book.finish(token, "succeeded", SHA_B)
canary_sha = _bind_success(book, root, "purchase", ["draft"])
book.complete_canary({"purchase": canary_sha})
token = book.claim("service", "draft", SHA_A)
book.finish(token, "succeeded", SHA_B)
_bind_success(book, root, "service", ["draft"])
if mutation == "result":
(root / "service/result.json").write_text("{}", encoding="utf-8")
elif mutation == "evidence":
next((root / "service").glob("*-after.json")).unlink()
else:
binding_path = book.case_outcomes / "service.json"
binding = json.loads(binding_path.read_text(encoding="utf-8"))
binding["case_attempts_sha256"] = SHA_A
binding_path.write_text(json.dumps(binding), encoding="utf-8")
resumed = ledger(root, **values)
with pytest.raises(LEDGER.CallLedgerError, match="blocked failure classes"):
resumed.authorize_selection(["third"], resume=True, explicit_selection=True)
snapshot = resumed.snapshot()
assert snapshot["unknown"] == 1
assert snapshot["terminal"] == 1
assert snapshot["provider_terminal"] == 2
assert snapshot["release_controls"]["blocked_failure_classes"] == ["specialist"]
def test_concurrent_resume_fails_closed_between_provider_finish_and_outcome_binding(tmp_path):
classes = {"purchase": "canary", "service": "specialist", "third": "specialist"}
values = dict(operation_limits={"draft": 3},
case_limits={key: 1 for key in classes},
release_controls=_controls(classes, ["purchase"]))
first = ledger(tmp_path, **values)
first.authorize_selection(["purchase"], resume=False, explicit_selection=True)
token = first.claim("purchase", "draft", SHA_A)
first.finish(token, "succeeded", SHA_B)
canary_sha = _bind_success(first, tmp_path, "purchase", ["draft"])
first.complete_canary({"purchase": canary_sha})
token = first.claim("service", "draft", SHA_A)
first.finish(token, "succeeded", SHA_B)
concurrent = ledger(tmp_path, **values)
with pytest.raises(LEDGER.CallLedgerError, match="blocked failure classes"):
concurrent.authorize_selection(["third"], resume=True, explicit_selection=True)
assert concurrent.snapshot()["unknown"] == 1
# Even if the first writer subsequently finishes its local artifacts, the
# persisted breaker cannot be erased by a race.
_bind_success(first, tmp_path, "service", ["draft"])
assert first.control_snapshot()["blocked_failure_classes"] == ["specialist"]
def test_legacy_ledger_schema_is_explicitly_read_only(tmp_path):
root = tmp_path / ".call-ledger"
root.mkdir()
yuqei-infra/tests/test_multiscene_eval_plan.py
+29
View File
@@ -1,5 +1,6 @@
import hashlib
import json
from types import SimpleNamespace
import pytest
@@ -135,6 +136,34 @@ def test_git_candidate_gate_requires_exact_head_tag_and_clean_tree(monkeypatch,
plan_module._verify_git_candidate(tmp_path, commit, "rc05-candidate", "test")
@pytest.mark.parametrize("injected", [
{"GIT_DIR": "redirected.git", "GIT_WORK_TREE": "redirected-tree",
"GIT_COMMON_DIR": "redirected-common"},
{"GIT_INDEX_FILE": "alternate-index", "GIT_OBJECT_DIRECTORY": "alternate-objects",
"GIT_ALTERNATE_OBJECT_DIRECTORIES": "more-objects"},
{"GIT_CONFIG_COUNT": "1", "GIT_CONFIG_KEY_0": "core.bare",
"GIT_CONFIG_VALUE_0": "true"},
])
def test_git_candidate_commands_drop_repository_and_config_injection(
monkeypatch, tmp_path, injected,
):
captured = {}
for name, value in injected.items():
monkeypatch.setenv(name, value)
def run(argv, **kwargs):
captured.update(kwargs["env"])
return SimpleNamespace(returncode=0, stdout="", stderr="")
monkeypatch.setattr(plan_module.subprocess, "run", run)
assert plan_module._git(tmp_path, "status", "--porcelain=v1") == ""
assert not set(injected) & set(captured)
assert captured["GIT_CONFIG_NOSYSTEM"] == "1"
assert captured["GIT_CONFIG_GLOBAL"] == plan_module.os.devnull
assert captured["GIT_NO_REPLACE_OBJECTS"] == "1"
assert captured["GIT_TERMINAL_PROMPT"] == "0"
def test_source_pins_are_repo_relative_existing_files_without_runtime_or_secrets(plan):
assert plan["source_pins"]
for relative, digest in plan["source_pins"].items():
yuqei-infra/tests/test_multiscene_eval_runner.py
+35
View File
@@ -438,6 +438,41 @@ def test_real_execution_rejects_unbound_candidate_and_initial_full_dispatch(
assert calls == []
def test_real_canary_does_not_transition_before_operation_and_evidence_gates(
monkeypatch, tmp_path,
):
plan = _plan(tmp_path, count=1)
plan["cases"][0].update(
provider_call_budget=2,
operations=["draft", "semantic_review"],
)
plan["operation_limits"] = {"draft": 1, "semantic_review": 1}
calls = _install_modules(monkeypatch, plan)
registry = tmp_path.parent / f"{tmp_path.name}-rc05-canary-registry"
registry.mkdir()
monkeypatch.setenv(EVAL.REAL_RUN_REGISTRY_ENV, str(registry))
out = tmp_path / "canary"
summary = EVAL.execute(
_runner(tmp_path), tmp_path / "plan.json", "a" * 64, out, RealFactory,
historical_usage=_historical_usage(), execution_mode="real",
run_registry_dir=registry, case_ids=["case-0"],
candidate_inputs=_candidate_inputs(),
)
assert calls == ["case-0"]
assert summary["case_operation_coverage_passed"] is False
assert summary["ledger"]["unknown"] == 1
assert summary["ledger"]["campaign"]["unknown"] == 1
assert summary["ledger"]["release_controls"]["phase"] == "canary_required"
assert summary["ledger"]["release_controls"]["blocked_failure_classes"] == ["draft"]
with pytest.raises(EVAL.CallLedgerError, match="blocked by a persisted failure class"):
EVAL.execute(
_runner(tmp_path), tmp_path / "plan.json", "a" * 64, out, RealFactory,
historical_usage=_historical_usage(), execution_mode="real",
run_registry_dir=registry, case_ids=["case-0"], resume=True,
candidate_inputs=_candidate_inputs(),
)
def test_authoritative_campaign_registry_requires_explicit_empty_directory(
monkeypatch, tmp_path,
):