diff --git a/yuqei-infra/scripts/multiscene_call_ledger.py b/yuqei-infra/scripts/multiscene_call_ledger.py index 93beede2..676b9795 100644 --- a/yuqei-infra/scripts/multiscene_call_ledger.py +++ b/yuqei-infra/scripts/multiscene_call_ledger.py @@ -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") diff --git a/yuqei-infra/scripts/multiscene_eval_plan.py b/yuqei-infra/scripts/multiscene_eval_plan.py index 9aecdfe3..96f1e4b1 100644 --- a/yuqei-infra/scripts/multiscene_eval_plan.py +++ b/yuqei-infra/scripts/multiscene_eval_plan.py @@ -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, diff --git a/yuqei-infra/scripts/multiscene_eval_runner.py b/yuqei-infra/scripts/multiscene_eval_runner.py index bc328a97..12c7eb61 100644 --- a/yuqei-infra/scripts/multiscene_eval_runner.py +++ b/yuqei-infra/scripts/multiscene_eval_runner.py @@ -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) diff --git a/yuqei-infra/tests/test_multiscene_call_ledger.py b/yuqei-infra/tests/test_multiscene_call_ledger.py index 077b8ab4..82b18b77 100644 --- a/yuqei-infra/tests/test_multiscene_call_ledger.py +++ b/yuqei-infra/tests/test_multiscene_call_ledger.py @@ -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() diff --git a/yuqei-infra/tests/test_multiscene_eval_plan.py b/yuqei-infra/tests/test_multiscene_eval_plan.py index 62f1e57e..b3299256 100644 --- a/yuqei-infra/tests/test_multiscene_eval_plan.py +++ b/yuqei-infra/tests/test_multiscene_eval_plan.py @@ -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(): diff --git a/yuqei-infra/tests/test_multiscene_eval_runner.py b/yuqei-infra/tests/test_multiscene_eval_runner.py index 6db9d449..c813c87f 100644 --- a/yuqei-infra/tests/test_multiscene_eval_runner.py +++ b/yuqei-infra/tests/test_multiscene_eval_runner.py @@ -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, ):