diff --git a/.github/workflows/build-test-quality-gate.yml b/.github/workflows/build-test-quality-gate.yml index cbd72c1c..de0b5253 100644 --- a/.github/workflows/build-test-quality-gate.yml +++ b/.github/workflows/build-test-quality-gate.yml @@ -203,10 +203,13 @@ jobs: CONTRACT_API_DATABASE_URL: sqlite+pysqlite:////tmp/yuqei-contract-api-ci.sqlite3 CONTRACT_API_DATABASE_AUTO_CREATE: "false" run: | + expected_head="$(python -m alembic heads | sed -nE 's/^([^ ]+) \(head\)$/\1/p')" + test "$(printf '%s\n' "$expected_head" | sed '/^$/d' | wc -l)" -eq 1 python -m alembic upgrade head python -m alembic downgrade 20260710_0030 python -m alembic upgrade head - python -m alembic current | grep -F "20260909_0079 (head)" + actual_head="$(python -m alembic current | sed -nE 's/^([^ ]+) \(head\)$/\1/p')" + test "$actual_head" = "$expected_head" - name: Verify AI Platform API empty-database migrations working-directory: yuqei-ai-platform/services/ai-platform-api @@ -215,5 +218,8 @@ jobs: AI_PLATFORM_DATABASE_AUTO_CREATE: "false" AI_PLATFORM_REPOSITORY_BACKEND: sqlalchemy run: | + expected_head="$(python -m alembic heads | sed -nE 's/^([^ ]+) \(head\)$/\1/p')" + test "$(printf '%s\n' "$expected_head" | sed '/^$/d' | wc -l)" -eq 1 python -m alembic upgrade head - python -m alembic current | grep -F "20260826_0028 (head)" + actual_head="$(python -m alembic current | sed -nE 's/^([^ ]+) \(head\)$/\1/p')" + test "$actual_head" = "$expected_head" diff --git a/yuqei-infra/scripts/multiscene_call_ledger.py b/yuqei-infra/scripts/multiscene_call_ledger.py index 683ce839..93beede2 100644 --- a/yuqei-infra/scripts/multiscene_call_ledger.py +++ b/yuqei-infra/scripts/multiscene_call_ledger.py @@ -25,6 +25,9 @@ TERMINAL_STATUSES = frozenset({"succeeded", "failed", "cancelled"}) MAX_PREALLOCATED_ATTEMPTS = 280 MAX_REPAIR_ATTEMPTS = 16 REPAIR_OPERATIONS = frozenset({"repair_draft", "repair_review"}) +LEDGER_SCHEMA = "multiscene-call-ledger.v2" +CAMPAIGN_SCHEMA = "multiscene-call-campaign.v2" +CONTROL_SCHEMA = "multiscene-release-control-state.v1" _ID_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,127}\Z") _SHA_RE = re.compile(r"[0-9a-f]{64}\Z") @@ -46,7 +49,9 @@ class CallLedger: operation_limits: Mapping[str, int], case_limits: Mapping[str, int], source_pins: Mapping[str, str], real_provider: bool, historical_usage: Mapping[str, object] | None = None, - registry_root: str | Path | None = None) -> None: + registry_root: str | Path | None = None, + candidate_binding: Mapping[str, object] | None = None, + release_controls: Mapping[str, object] | None = None) -> None: requested_output = Path(output_dir).absolute() _reject_path_chain(requested_output, "output directory") self.output_dir = requested_output.resolve() @@ -71,6 +76,9 @@ class CallLedger: self.source_pins[name] = _sha(digest, "source pin") self.real_provider = real_provider self.historical_usage = _historical_usage(historical_usage) + self.candidate_binding = _json_mapping(candidate_binding, "candidate binding") + self.release_controls = _release_controls(release_controls, self.case_limits) + self.controls_enabled = self.release_controls is not None if registry_root is not None: if not real_provider or historical_usage is None: raise ValueError("global registry is only valid for a receipt-bound real run") @@ -85,8 +93,8 @@ class CallLedger: }) self.root = self.registry_root / run_key self.campaign_identity = { - "schema_version": "multiscene-call-campaign.v1", - "campaign_id": "rc05-real-usage-v1", + "schema_version": CAMPAIGN_SCHEMA, + "campaign_id": "rc05-real-usage-v2", "historical_receipt_sha256": self.historical_usage["receipt_sha256"], "historical_attempts_sha256": self.historical_usage["attempts_sha256"], "historical_provider_call_count": self.historical_usage["provider_call_count"], @@ -116,15 +124,18 @@ class CallLedger: } self.attempts = self.root / "attempts" self.case_outcomes = self.root / "case-outcomes" + self.control_events = self.root / "control-events" self.lock_path = ((self.registry_root / ".lock") if self.registry_root is not None else self.root / ".lock") self.identity = { - "schema_version": "multiscene-call-ledger.v1", + "schema_version": LEDGER_SCHEMA, "plan_sha256": self.plan_sha256, "operation_limits": {name: self.operation_limits.get(name, 0) for name in OPERATIONS}, "case_limits": dict(sorted(self.case_limits.items())), "source_pins": dict(sorted(self.source_pins.items())), "real_provider": real_provider, + "candidate_binding": self.candidate_binding, + "release_controls": self.release_controls, "authority": authority, "integrity_not_authentication": True, } @@ -145,7 +156,12 @@ class CallLedger: campaign_identity_path = self.registry_root / "campaign-identity.json" _reject_link_or_reparse(campaign_identity_path, "campaign identity") if campaign_identity_path.exists(): - if _read_json(campaign_identity_path) != self.campaign_identity: + stored_campaign = _read_json(campaign_identity_path) + if stored_campaign.get("schema_version") != CAMPAIGN_SCHEMA: + raise LedgerIdentityError( + "legacy campaign schema is read-only diagnostic material; execution refused" + ) + if stored_campaign != self.campaign_identity: raise LedgerIdentityError( "global campaign is already bound to a different historical receipt" ) @@ -168,7 +184,12 @@ class CallLedger: identity_path = self.root / "identity.json" _reject_link_or_reparse(identity_path, "ledger identity") if identity_path.exists(): - if _read_json(identity_path) != self.identity: + stored_identity = _read_json(identity_path) + if stored_identity.get("schema_version") != LEDGER_SCHEMA: + raise LedgerIdentityError( + "legacy ledger schema is read-only diagnostic material; execution refused" + ) + if stored_identity != self.identity: raise LedgerIdentityError("ledger identity does not match the frozen run") else: _write_x(identity_path, self.identity) @@ -178,11 +199,155 @@ class CallLedger: _reject_link_or_reparse(self.case_outcomes, "case outcome directory") self.case_outcomes.mkdir(exist_ok=True) _reject_link_or_reparse(self.case_outcomes, "case outcome directory") + _reject_link_or_reparse(self.control_events, "control event directory") + self.control_events.mkdir(exist_ok=True) + _reject_link_or_reparse(self.control_events, "control event directory") # Create the output only after the global identity accepted this exact # path. A competing fresh output therefore cannot leave a fork behind. self.output_dir.mkdir(parents=True, exist_ok=True) _reject_link_or_reparse(self.output_dir, "output directory") + def control_snapshot(self) -> dict[str, object]: + with self._locked(): + return self._control_snapshot_unlocked() + + def authorize_selection(self, selected: list[str], *, resume: bool, + explicit_selection: bool) -> dict[str, object]: + """Fail closed before dispatch unless persisted release controls permit it.""" + if not self.controls_enabled: + raise CallLedgerError("release controls are required for controlled execution") + if (not isinstance(selected, list) or not selected + or len(selected) != len(set(selected)) + or any(case_id not in self.case_limits for case_id in selected)): + raise CallLedgerError("selection is outside frozen release controls") + with self._locked(): + self._materialize_attempt_breakers_unlocked() + state = self._control_snapshot_unlocked() + canaries = self.release_controls["canary_case_ids"] + if state["phase"] == "canary_required": + if not explicit_selection or selected != canaries: + raise CallLedgerError( + "canary phase requires the explicit predefined canary case sequence" + ) + if state["blocked_failure_classes"]: + raise CallLedgerError("canary phase is blocked by a persisted failure class") + records = self._records() + blocked = set(state["blocked_failure_classes"]) + future = [case_id for case_id in selected + if not any(key[0] == case_id for key in records)] + forbidden = sorted({self.release_controls["failure_classes"][case_id] + for case_id in future} & blocked) + if forbidden: + raise CallLedgerError( + f"selection contains persisted blocked failure classes: {forbidden}" + ) + return state + + def complete_canary(self, outcome_sha256s: Mapping[str, str]) -> None: + """Persist the single immutable transition that authorizes full dispatch.""" + if not self.controls_enabled: + raise CallLedgerError("release controls are not configured") + canaries = self.release_controls["canary_case_ids"] + if not isinstance(outcome_sha256s, Mapping) or set(outcome_sha256s) != set(canaries): + raise CallLedgerError("canary outcome set is incomplete") + normalized = {case_id: _sha(outcome_sha256s[case_id], "canary outcome SHA-256") + for case_id in canaries} + with self._locked(): + self._materialize_attempt_breakers_unlocked() + state = self._control_snapshot_unlocked() + if state["blocked_failure_classes"]: + raise CallLedgerError("failed or unknown canary cannot authorize full dispatch") + records = self._records() + for case_id in canaries: + 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): + raise CallLedgerError("canary provider attempts are incomplete or unsuccessful") + binding = _read_json(self.case_outcomes / f"{case_id}.json") + if binding.get("outcome_sha256") != normalized[case_id]: + raise CallLedgerError("canary outcome does not match immutable binding") + payload = {"schema_version": CONTROL_SCHEMA, + "event": "canary_to_full", "ledger_identity_sha256": self.identity_sha256, + "canary_case_ids": canaries, "outcome_sha256s": normalized} + transition = self.control_events / "canary-to-full.json" + if transition.exists(): + if _read_json(transition) != payload: + raise CallLedgerError("canary transition is already bound differently") + else: + _write_x(transition, payload) + + def trip_failure_class(self, case_id: str, status: str, reason: str) -> None: + if not self.controls_enabled: + return + case_id = _identifier(case_id, "failure case_id") + if case_id not in self.case_limits or status not in {"failed", "cancelled", "unknown"}: + raise CallLedgerError("failure circuit event is invalid") + failure_class = self.release_controls["failure_classes"][case_id] + if not isinstance(reason, str) or not reason or len(reason) > 1000: + raise CallLedgerError("failure circuit reason is invalid") + payload = {"schema_version": CONTROL_SCHEMA, "event": "failure_class_blocked", + "ledger_identity_sha256": self.identity_sha256, "failure_class": failure_class, + "case_id": case_id, "status": status, "reason": reason} + path = self.control_events / f"failure-class--{failure_class}.json" + with self._locked(): + if path.exists(): + existing = _read_json(path) + if existing.get("failure_class") != failure_class: + raise CallLedgerError("failure circuit event is inconsistent") + else: + _write_x(path, payload) + + 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 + failure_class = self.release_controls["failure_classes"][case_id] + path = self.control_events / f"failure-class--{failure_class}.json" + if not path.exists(): + _write_x(path, {"schema_version": CONTROL_SCHEMA, + "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}"}) + + def _control_snapshot_unlocked(self) -> dict[str, object]: + if not self.controls_enabled: + return {"schema_version": CONTROL_SCHEMA, "phase": "not_configured", + "canary_case_ids": [], "blocked_failure_classes": [], + "canary_transition_sha256": None} + blocked: list[str] = [] + transition_sha = None + for path in self.control_events.iterdir(): + _reject_link_or_reparse(path, "control event") + if not path.is_file(): + raise CallLedgerError("control event journal contains a non-file entry") + payload = _read_json(path) + if (not isinstance(payload, dict) + or payload.get("schema_version") != CONTROL_SCHEMA + 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": + transition_sha = _digest(payload) + elif (re.fullmatch(r"failure-class--[A-Za-z0-9._-]+\.json", path.name) + and payload.get("event") == "failure_class_blocked"): + failure_class = _identifier(payload.get("failure_class"), "failure class") + if 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: + raise CallLedgerError("control event journal contains an unknown entry") + return {"schema_version": CONTROL_SCHEMA, + "phase": "full_authorized" if transition_sha else "canary_required", + "canary_case_ids": self.release_controls["canary_case_ids"], + "blocked_failure_classes": sorted(set(blocked)), + "canary_transition_sha256": transition_sha} + def claim(self, case_id: str, operation: str, request_sha256: str) -> str: case_id = _identifier(case_id, "case_id") if case_id not in self.case_limits: @@ -192,6 +357,15 @@ class CallLedger: request_sha256 = _sha(request_sha256, "request_sha256") stem = f"{case_id}--{operation}" with self._locked(): + if self.controls_enabled: + self._materialize_attempt_breakers_unlocked() + state = self._control_snapshot_unlocked() + failure_class = self.release_controls["failure_classes"][case_id] + if failure_class in state["blocked_failure_classes"]: + raise CallLedgerError("failure-class circuit breaker is open") + if (state["phase"] == "canary_required" + and case_id not in self.release_controls["canary_case_ids"]): + raise CallLedgerError("full dispatch is forbidden before canary transition") records = self._records() campaign_records = self._campaign_records() if self.registry_root is not None else records if (case_id, operation) in records: @@ -327,7 +501,8 @@ class CallLedger: repair_attempted = sum( 1 for record in campaign_records.values() if record["claim"]["operation"] in REPAIR_OPERATIONS) - return {"schema_version": "multiscene-call-ledger-snapshot.v1", + controls = self._control_snapshot_unlocked() + return {"schema_version": "multiscene-call-ledger-snapshot.v2", "ledger_identity_sha256": self.identity_sha256, "real_provider": self.real_provider, "authority": self.identity["authority"], @@ -376,7 +551,8 @@ class CallLedger: "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())} + "unknown": sum(record["finish"] is None for record in records.values()), + "release_controls": controls} def attempt_records(self) -> list[dict[str, object]]: """Return immutable attempt outcomes for evidence aggregation.""" @@ -486,6 +662,10 @@ class CallLedger: 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") + if expected_identity.get("schema_version") != LEDGER_SCHEMA: + raise LedgerIdentityError( + "legacy ledger schema is read-only diagnostic material; execution refused" + ) if expected_identity != self.identity: raise LedgerIdentityError("ledger identity changed") records: dict[tuple[str, str], dict] = {} @@ -536,7 +716,12 @@ class CallLedger: raise CallLedgerError("campaign records require a global registry") _reject_link_or_reparse(self.registry_root / "campaign-identity.json", "campaign identity") - if _read_json(self.registry_root / "campaign-identity.json") != self.campaign_identity: + stored_campaign = _read_json(self.registry_root / "campaign-identity.json") + if stored_campaign.get("schema_version") != CAMPAIGN_SCHEMA: + raise LedgerIdentityError( + "legacy campaign schema is read-only diagnostic material; execution refused" + ) + if stored_campaign != self.campaign_identity: raise LedgerIdentityError("global campaign identity changed") records: dict[str, dict] = {} for path in self.campaign_attempts.iterdir(): @@ -633,6 +818,41 @@ def _limits(value: Mapping[str, int], *, operations: bool) -> dict[str, int]: return result +def _json_mapping(value: Mapping[str, object] | None, label: str) -> dict | None: + if value is None: + return None + if not isinstance(value, Mapping): + raise TypeError(f"{label} must be a mapping") + try: + return json.loads(json.dumps(dict(value), ensure_ascii=False, sort_keys=True, + separators=(",", ":"), allow_nan=False)) + except (TypeError, ValueError) as error: + raise ValueError(f"{label} must be canonical JSON data") from error + + +def _release_controls(value: Mapping[str, object] | None, + case_limits: Mapping[str, int]) -> dict | None: + normalized = _json_mapping(value, "release controls") + if normalized is None: + return None + if set(normalized) != {"schema_version", "canary_case_ids", "failure_classes"}: + raise ValueError("release controls shape is invalid") + if normalized["schema_version"] != "multiscene-release-controls.v1": + raise ValueError("release controls schema is unsupported") + canaries = normalized["canary_case_ids"] + classes = normalized["failure_classes"] + if (not isinstance(canaries, list) or not canaries + or len(canaries) != len(set(canaries)) + or any(case_id not in case_limits for case_id in canaries)): + raise ValueError("release control canary inventory is invalid") + if not isinstance(classes, dict) or set(classes) != set(case_limits): + raise ValueError("release control failure-class inventory is invalid") + for case_id, failure_class in classes.items(): + _identifier(case_id, "release control case ID") + _identifier(failure_class, "release control failure class") + return normalized + + def _case_attempt_projection(records: Mapping[tuple[str, str], dict], case_id: str) -> list[dict]: return [ { diff --git a/yuqei-infra/scripts/multiscene_eval_plan.py b/yuqei-infra/scripts/multiscene_eval_plan.py index a9414deb..9aecdfe3 100644 --- a/yuqei-infra/scripts/multiscene_eval_plan.py +++ b/yuqei-infra/scripts/multiscene_eval_plan.py @@ -4,8 +4,11 @@ from __future__ import annotations from datetime import UTC, datetime from hashlib import sha256 import json -from pathlib import Path -from typing import Any +import os +import re +import subprocess +from pathlib import Path, PurePath +from typing import Any, Mapping from yuqei_contract_api.contract_templates import template_document_kind from yuqei_contract_api.drafting_workspace import ( @@ -20,7 +23,23 @@ from multiscene_supplemental_cases import build_supplemental_cases ROOT = Path(__file__).resolve().parents[2] -SCHEMA_VERSION = "multiscene-evaluation-plan.v1" +SCHEMA_VERSION = "multiscene-evaluation-plan.v2" +CANDIDATE_SCHEMA_VERSION = "multiscene-candidate-binding.v1" +RELEASE_CONTROLS_SCHEMA_VERSION = "multiscene-release-controls.v1" +REQUIRED_IMAGE_ROLES = ( + "contract-web", "contract-api", "ai-platform-api", "ai-worker", +) +CANARY_CASE_IDS = ( + "base--TPL-AGENCY-DISTRIBUTION-001--normal", + "base--TPL-DEMAND-LETTER-001--normal", + "base--TPL-PURCHASE-GOODS-001--normal", + "scene--TPL-AMENDMENT-GENERAL-001--opposite-adversarial", + "specialist-trade_procurement-first", + "intake-tpl-purchase-goods-001", +) +_SHA256_RE = re.compile(r"[0-9a-f]{64}\Z") +_COMMIT_RE = re.compile(r"[0-9a-f]{40}\Z") +_TAG_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._/-]{0,127}\Z") EXPECTED_OPERATIONS = { "draft": 120, "semantic_review": 30, "repair_draft": 8, "repair_review": 8, "specialist_review": 106, "intake": 8, @@ -55,6 +74,262 @@ def _canonical(value: Any) -> str: separators=(",", ":"), allow_nan=False) +def _strict_json(raw: bytes, label: str) -> Any: + def pairs(items): + result = {} + for key, value in items: + if key in result: + raise ValueError(f"{label} contains a duplicate JSON key") + result[key] = value + return result + + def constant(value): + raise ValueError(f"{label} contains non-finite JSON value {value}") + + try: + return json.loads(raw, object_pairs_hook=pairs, parse_constant=constant) + except (UnicodeDecodeError, json.JSONDecodeError) as error: + raise ValueError(f"{label} is not valid strict JSON") from error + + +def _require_sha(value: object, label: str) -> str: + if not isinstance(value, str) or not _SHA256_RE.fullmatch(value): + raise ValueError(f"{label} must be a lowercase SHA-256 value") + return value + + +def _artifact_bytes(path: Path, root: Path, label: str) -> bytes: + supplied = Path(path) + if not supplied.is_absolute(): + raise ValueError(f"{label} path must be absolute") + absolute = supplied.absolute() + current = absolute + while True: + junction = getattr(current, "is_junction", lambda: False) + if current.is_symlink() or junction(): + raise ValueError(f"{label} path must not traverse links or junctions") + if current.parent == current: + break + current = current.parent + if not absolute.is_file() or absolute.resolve().is_relative_to(root.resolve()): + raise ValueError(f"{label} must be a regular file outside the source worktree") + return absolute.read_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"} + completed = subprocess.run( + ["git", "-C", str(root), *args], capture_output=True, + text=not binary, check=False, env=environment, + ) + if completed.returncode: + error = completed.stderr if isinstance(completed.stderr, str) else completed.stderr.decode("utf-8", "replace") + raise ValueError(f"git candidate verification failed: {error.strip()}") + return completed.stdout if binary else completed.stdout.strip() + + +def _verify_git_candidate(root: Path, commit: str, tag: str, stage: str) -> None: + if not _COMMIT_RE.fullmatch(commit): + raise ValueError("candidate commit must be a full lowercase SHA-1 object ID") + if (not _TAG_RE.fullmatch(tag) or ".." in tag or "//" in tag + or tag.endswith((".", "/")) or tag.endswith(".lock")): + raise ValueError("candidate tag is unsafe") + head = _git(root, "rev-parse", "--verify", "HEAD^{commit}") + tagged = _git(root, "rev-parse", "--verify", f"refs/tags/{tag}^{{commit}}") + if head != commit or tagged != commit: + raise ValueError(f"candidate HEAD/tag mismatch {stage}") + if _git(root, "status", "--porcelain=v1", "-z", "--untracked-files=all"): + raise ValueError(f"candidate worktree is not clean {stage}") + + +def _source_manifest(payload: Any, commit: str) -> tuple[dict[str, str], str]: + expected = {"schema_version", "candidate", "files", "summaries", + "claims_not_made", "manifest_payload_sha256"} + if not isinstance(payload, dict) or set(payload) != expected: + raise ValueError("source manifest shape is invalid") + if payload["schema_version"] != "yuqei.release-source-manifest.v1": + raise ValueError("unsupported source manifest schema") + embedded = _require_sha(payload["manifest_payload_sha256"], "source manifest payload SHA-256") + projection = {key: value for key, value in payload.items() + if key != "manifest_payload_sha256"} + if _hash_bytes(_canonical(projection).encode("utf-8")) != embedded: + raise ValueError("source manifest embedded payload SHA-256 mismatch") + candidate = payload["candidate"] + if (not isinstance(candidate, dict) or candidate.get("commit") != commit + or candidate.get("worktree_was_clean") is not True + or candidate.get("identity_policy") != "full_commit_oid_equals_HEAD"): + raise ValueError("source manifest candidate identity mismatch") + files = payload["files"] + if not isinstance(files, dict) or not isinstance(files.get("items"), list): + raise ValueError("source manifest file inventory is invalid") + if files.get("count") != len(files["items"]): + raise ValueError("source manifest file count is inconsistent") + index: dict[str, str] = {} + for row in files["items"]: + if not isinstance(row, dict) or set(row) != { + "path", "mode", "git_blob_oid", "size", "sha256", + }: + raise ValueError("source manifest file row is invalid") + relative = row["path"] + if (not isinstance(relative, str) or relative.startswith(("/", "\\")) + or ".." in PurePath(relative).parts or relative in index): + raise ValueError("source manifest contains an unsafe or duplicate path") + index[relative] = _require_sha(row["sha256"], f"source manifest {relative}") + return index, embedded + + +def _image_identities(payload: Any, commit: str, tag: str) -> tuple[list[dict[str, str]], str, str]: + if not isinstance(payload, dict): + raise ValueError("image identities must be an object") + schema = payload.get("schema_version") + if schema == "yuqei.rc06-image-identities.v1": + if set(payload) != {"schema_version", "commit", "tag", "images"}: + raise ValueError("image identities shape is invalid") + document_commit, document_tag, images = payload["commit"], payload["tag"], payload["images"] + elif schema == "yuqei.installable-release-manifest.v1": + candidate = payload.get("candidate") + if not isinstance(candidate, dict): + raise ValueError("installable image candidate is missing") + document_commit, document_tag, images = candidate.get("commit"), candidate.get("tag"), payload.get("images") + else: + raise ValueError("unsupported image identities schema") + if document_commit != commit or document_tag != tag: + raise ValueError("image identities candidate does not match commit/tag") + if not isinstance(images, list) or len(images) != len(REQUIRED_IMAGE_ROLES): + raise ValueError("image identities must contain four application images") + normalized, roles, manifests = [], set(), set() + for row in images: + if not isinstance(row, dict): + raise ValueError("image identity row is invalid") + if (schema == "yuqei.rc06-image-identities.v1" + and set(row) != {"role", "repository", "manifest_digest", "config_digest"}): + raise ValueError("compact image identity row shape is invalid") + role = row.get("role") + repository = row.get("repository") if schema == "yuqei.rc06-image-identities.v1" else (row.get("image_name") or row.get("registry_reference")) + manifest, config = row.get("manifest_digest"), row.get("config_digest") + if (not isinstance(role, str) or not isinstance(repository, str) or not repository + or not isinstance(manifest, str) or not manifest.startswith("sha256:") + or not isinstance(config, str) or not config.startswith("sha256:")): + raise ValueError("image identity row is incomplete") + _require_sha(manifest.removeprefix("sha256:"), f"image {role} manifest digest") + _require_sha(config.removeprefix("sha256:"), f"image {role} config digest") + if role in roles or manifest in manifests: + raise ValueError("duplicate image role or manifest digest") + roles.add(role) + manifests.add(manifest) + normalized.append({"role": role, "repository": repository, + "manifest_digest": manifest, "config_digest": config}) + if roles != set(REQUIRED_IMAGE_ROLES): + raise ValueError("image roles mismatch") + normalized.sort(key=lambda item: REQUIRED_IMAGE_ROLES.index(item["role"])) + return normalized, _hash_bytes(_canonical(normalized).encode("utf-8")), schema + + +def load_candidate_binding(runner, *, commit: str, tag: str, + source_manifest_path: Path, source_manifest_sha256: str, + source_manifest_payload_sha256: str, image_identities_path: Path, + image_identities_sha256: str, image_identity_set_sha256: str, + stage: str = "candidate preflight") -> tuple[dict[str, Any], dict[str, str]]: + """Load and reverify a built immutable candidate without creating evidence.""" + root = Path(runner.ROOT).resolve() + _verify_git_candidate(root, commit, tag, stage) + source_raw = _artifact_bytes(source_manifest_path, root, "source manifest") + image_raw = _artifact_bytes(image_identities_path, root, "image identities") + if _hash_bytes(source_raw) != _require_sha(source_manifest_sha256, "source manifest raw SHA-256"): + raise ValueError("source manifest raw SHA-256 mismatch") + if _hash_bytes(image_raw) != _require_sha(image_identities_sha256, "image identities raw SHA-256"): + raise ValueError("image identities raw SHA-256 mismatch") + source_index, embedded = _source_manifest(_strict_json(source_raw, "source manifest"), commit) + if embedded != _require_sha(source_manifest_payload_sha256, "source manifest payload SHA-256"): + raise ValueError("source manifest payload SHA-256 argument mismatch") + images, identity_set, image_schema = _image_identities( + _strict_json(image_raw, "image identities"), commit, tag) + if identity_set != _require_sha(image_identity_set_sha256, "image identity-set SHA-256"): + raise ValueError("image identity-set SHA-256 mismatch") + binding = { + "schema_version": CANDIDATE_SCHEMA_VERSION, "commit": commit, "tag": tag, + "source_manifest": {"schema_version": "yuqei.release-source-manifest.v1", + "raw_sha256": source_manifest_sha256, + "payload_sha256": source_manifest_payload_sha256}, + "image_identities": {"schema_version": image_schema, + "raw_sha256": image_identities_sha256, + "identity_set_sha256": image_identity_set_sha256, + "images": images}, + } + return binding, source_index + + +def _candidate_binding(value: Mapping[str, Any]) -> dict[str, Any]: + if not isinstance(value, Mapping) or value.get("schema_version") != CANDIDATE_SCHEMA_VERSION: + raise ValueError("candidate binding is missing or unsupported") + result = json.loads(_canonical(dict(value))) + if set(result) != {"schema_version", "commit", "tag", "source_manifest", "image_identities"}: + raise ValueError("candidate binding shape is invalid") + if not _COMMIT_RE.fullmatch(str(result.get("commit", ""))): + raise ValueError("candidate binding commit is invalid") + if (not isinstance(result.get("tag"), str) or not _TAG_RE.fullmatch(result["tag"]) + or ".." in result["tag"] or "//" in result["tag"]): + raise ValueError("candidate binding tag is invalid") + source, images = result.get("source_manifest"), result.get("image_identities") + if not isinstance(source, dict) or not isinstance(images, dict): + raise ValueError("candidate binding artifacts are invalid") + if set(source) != {"schema_version", "raw_sha256", "payload_sha256"}: + raise ValueError("candidate source binding shape is invalid") + if source["schema_version"] != "yuqei.release-source-manifest.v1": + raise ValueError("candidate source binding schema is invalid") + if set(images) != {"schema_version", "raw_sha256", "identity_set_sha256", "images"}: + raise ValueError("candidate image binding shape is invalid") + if images["schema_version"] not in { + "yuqei.rc06-image-identities.v1", "yuqei.installable-release-manifest.v1", + }: + raise ValueError("candidate image binding schema is invalid") + _require_sha(source.get("raw_sha256"), "candidate source raw SHA-256") + _require_sha(source.get("payload_sha256"), "candidate source payload SHA-256") + _require_sha(images.get("raw_sha256"), "candidate image raw SHA-256") + _require_sha(images.get("identity_set_sha256"), "candidate image set SHA-256") + if not isinstance(images.get("images"), list) or len(images["images"]) != 4: + raise ValueError("candidate binding must include four image identities") + normalized, roles, manifests = [], set(), set() + for row in images["images"]: + if not isinstance(row, dict) or set(row) != { + "role", "repository", "manifest_digest", "config_digest", + }: + raise ValueError("candidate normalized image row is invalid") + role, manifest, config = row["role"], row["manifest_digest"], row["config_digest"] + if role not in REQUIRED_IMAGE_ROLES or role in roles or manifest in manifests: + raise ValueError("candidate image role or manifest digest is invalid") + if not isinstance(row["repository"], str) or not row["repository"]: + raise ValueError("candidate image repository is invalid") + if not isinstance(manifest, str) or not manifest.startswith("sha256:"): + raise ValueError("candidate image manifest digest is invalid") + if not isinstance(config, str) or not config.startswith("sha256:"): + raise ValueError("candidate image config digest is invalid") + _require_sha(manifest.removeprefix("sha256:"), "candidate image manifest digest") + _require_sha(config.removeprefix("sha256:"), "candidate image config digest") + roles.add(role); manifests.add(manifest); normalized.append(row) + if roles != set(REQUIRED_IMAGE_ROLES): + raise ValueError("candidate image roles mismatch") + normalized.sort(key=lambda item: REQUIRED_IMAGE_ROLES.index(item["role"])) + if images["images"] != normalized: + raise ValueError("candidate image identities are not canonically ordered") + if _hash_bytes(_canonical(normalized).encode("utf-8")) != images["identity_set_sha256"]: + raise ValueError("candidate image identity-set digest is invalid") + return result + + +def verify_source_pins_at_candidate(runner, source_pins: Mapping[str, str], + source_index: Mapping[str, str], commit: str) -> None: + root = Path(runner.ROOT).resolve() + for relative, digest in source_pins.items(): + if source_index.get(relative) != digest: + raise ValueError(f"source pin is absent from or differs from manifest: {relative}") + committed = _git(root, "show", f"{commit}:{relative}", binary=True) + if _hash_bytes(committed) != digest: + raise ValueError(f"source pin differs from candidate commit: {relative}") + + def draft_request_projection(request: ContractDraftRequest | dict[str, Any]) -> dict[str, Any]: """Return stable model facts and template identities; omit workspace/lifecycle data.""" request = ContractDraftRequest.model_validate(request) @@ -137,7 +412,10 @@ def _normalized_cases() -> list[dict[str, Any]]: projection, template = _prepared_projection(case, repository) policy = {"track_changes": True, "semantic_review": case["mode"] == "contract", "auto_repair": case["auto_repair"]} - drafting.append({**case, "candidate_policy": policy, + failure_class = ("scene_drafting" if case["mode"] == "scene_evaluation" + else "contract_drafting") + drafting.append({**case, "failure_class": failure_class, + "candidate_policy": policy, "template_content_sha256": projection["template_content_sha256"], "template_input_schema_sha256": projection["template_schema_sha256"], "values_sha256": draft_digest(case["values"]), @@ -148,6 +426,7 @@ def _normalized_cases() -> list[dict[str, Any]]: "case_id": case["case_id"], "source_case_id": case["case_id"], "kind": case["operation"], "mode": case["operation"], "operation": case["operation"], "operations": [case["operation"]], "provider_call_budget": 1, + "failure_class": case["operation"], "request": case["request"].model_dump(mode="json"), "request_sha256": case["request_sha256"], "expectations": case["expectations"], **({"module": case["module"], "party_position": case["party_position"]} @@ -164,35 +443,54 @@ def _operation_counts(cases: list[dict[str, Any]]) -> dict[str, int]: return result -def build_plan(runner) -> dict[str, Any]: +def build_plan(runner, candidate_binding: Mapping[str, Any]) -> dict[str, Any]: cases = _normalized_cases() counts = _operation_counts(cases) if len(cases) != 144 or counts != EXPECTED_OPERATIONS: raise ValueError(f"case or operation inventory mismatch: {len(cases)}, {counts}") if sum(case["provider_call_budget"] for case in cases) != EXPECTED_BUDGET: raise ValueError("global provider call budget mismatch") + inventory = {case["case_id"]: case for case in cases} + if any(case_id not in inventory for case_id in CANARY_CASE_IDS): + raise ValueError("predefined RC-05 canary inventory drift") return { "schema_version": SCHEMA_VERSION, "prepared_at": datetime.now(UTC).isoformat(), "model": "gpt-5.6-sol", "reasoning_effort": "medium", "timeout_seconds": 300, "case_count": 144, "drafting_case_count": 120, "supplemental_case_count": 24, "operation_limits": counts, "total_provider_call_budget": EXPECTED_BUDGET, + "candidate": _candidate_binding(candidate_binding), + "release_controls": { + "schema_version": RELEASE_CONTROLS_SCHEMA_VERSION, + "canary_case_ids": list(CANARY_CASE_IDS), + "failure_classes": {case["case_id"]: case["failure_class"] for case in cases}, + }, "source_pins": _source_pins(runner), "cases": cases, "limitations": ["合成隔离评测，不是生产发布或法律批准。", "deterministic projection equality does not establish semantic or legal correctness."], } -def validate_plan(runner, path: Path, expected_sha: str | None = None) -> tuple[dict[str, Any], str]: +def validate_plan(runner, path: Path, expected_sha: str | None = None, + candidate_binding: Mapping[str, Any] | None = None, + source_manifest_index: Mapping[str, str] | None = None) -> tuple[dict[str, Any], str]: raw = Path(path).read_bytes() actual = _hash_bytes(raw) if expected_sha is not None and actual != expected_sha: raise ValueError("plan raw SHA-256 mismatch") - plan = json.loads(raw) - expected = build_plan(runner) + plan = _strict_json(raw, "multiscene plan") if plan.get("schema_version") != SCHEMA_VERSION: - raise ValueError("unsupported multiscene plan") + raise ValueError("legacy/unsupported multiscene plan is read-only diagnostic material; execution refused") + frozen_candidate = _candidate_binding(plan.get("candidate")) + if candidate_binding is not None and _canonical(frozen_candidate) != _canonical( + _candidate_binding(candidate_binding)): + raise ValueError("multiscene plan candidate binding mismatch") + expected = build_plan(runner, frozen_candidate) left = {key: value for key, value in plan.items() if key != "prepared_at"} right = {key: value for key, value in expected.items() if key != "prepared_at"} if _canonical(left) != _canonical(right): raise ValueError("multiscene plan content or source pins drift") + if source_manifest_index is not None: + verify_source_pins_at_candidate( + runner, plan["source_pins"], source_manifest_index, + frozen_candidate["commit"]) return plan, actual diff --git a/yuqei-infra/scripts/multiscene_eval_runner.py b/yuqei-infra/scripts/multiscene_eval_runner.py index a8e36f9c..bc328a97 100644 --- a/yuqei-infra/scripts/multiscene_eval_runner.py +++ b/yuqei-infra/scripts/multiscene_eval_runner.py @@ -18,7 +18,7 @@ from pathlib import Path from time import perf_counter from uuid import uuid4 -from multiscene_call_ledger import CallLedger, OPERATIONS +from multiscene_call_ledger import CallLedger, CallLedgerError, OPERATIONS HISTORICAL_USAGE_SCHEMA = "rc05-historical-provider-usage.v3" @@ -1041,18 +1041,12 @@ def _authoritative_registry(runner, supplied) -> Path: def execute(runner, plan_path, expected_sha, output_dir, factory, *, resume=False, case_ids=None, historical_usage=None, execution_mode=None, - run_registry_dir=None): - from multiscene_eval_plan import validate_plan + run_registry_dir=None, candidate_inputs=None): + from multiscene_eval_plan import ( + load_candidate_binding, validate_plan, verify_source_pins_at_candidate, + ) from multiscene_supplemental_cases import run_supplemental_case - plan, digest = validate_plan(runner, plan_path, expected_sha) - inventory = {case["case_id"]: case for case in plan["cases"]} - selected = list(inventory) if case_ids is None else case_ids - if not selected or len(selected) != len(set(selected)) or set(selected) - set(inventory): - raise ValueError("selection must contain unique case IDs from the frozen plan") - out = Path(output_dir) - if out.exists() and not resume: - raise ValueError("fresh output directory required unless --resume is explicit") if execution_mode not in {"real", "synthetic"}: raise ValueError("execution_mode must explicitly be real or synthetic") real = execution_mode == "real" @@ -1060,6 +1054,24 @@ def execute(runner, plan_path, expected_sha, output_dir, factory, *, resume=Fals raise ValueError("real execution requires the unwrapped frozen real provider factory") if not real and getattr(factory, "_rc05_execution_mode", None) != "synthetic": raise ValueError("synthetic execution requires an attested synthetic provider factory") + if real: + if not isinstance(candidate_inputs, dict): + raise ValueError("real RC-05 execution requires complete immutable candidate inputs") + candidate_binding, source_manifest_index = load_candidate_binding( + runner, **candidate_inputs, stage="before plan acceptance") + plan, digest = validate_plan( + runner, plan_path, expected_sha, candidate_binding, + source_manifest_index) + else: + candidate_binding = source_manifest_index = None + plan, digest = validate_plan(runner, plan_path, expected_sha) + inventory = {case["case_id"]: case for case in plan["cases"]} + selected = list(inventory) if case_ids is None else case_ids + if not selected or len(selected) != len(set(selected)) or set(selected) - set(inventory): + raise ValueError("selection must contain unique case IDs from the frozen plan") + out = Path(output_dir) + if out.exists() and not resume: + raise ValueError("fresh output directory required unless --resume is explicit") if resume and not out.exists(): raise ValueError("resume requires the original evidence output directory") if resume and not real and not (out / ".call-ledger/identity.json").is_file(): @@ -1081,7 +1093,13 @@ def execute(runner, plan_path, expected_sha, output_dir, factory, *, resume=Fals ledger = CallLedger(out, digest, plan["operation_limits"], {key: case["provider_call_budget"] for key, case in inventory.items()}, plan["source_pins"], real, historical_usage=historical_usage, - registry_root=authoritative_registry if real else None) + registry_root=authoritative_registry if real else None, + candidate_binding=plan.get("candidate") if real else None, + release_controls=plan.get("release_controls") if real else None) + control_before = None + if real: + control_before = ledger.authorize_selection( + selected, resume=resume, explicit_selection=case_ids is not None) runner.write_x(out / f"selection-{uuid4().hex}.json", { "plan_raw_sha256": digest, "selected": selected, "resume": resume, "historical_usage": historical_usage, @@ -1091,6 +1109,14 @@ def execute(runner, plan_path, expected_sha, output_dir, factory, *, resume=Fals prior = _existing_case(runner, out, inventory[key], snapshot, ledger) if prior["status"] != "not_attempted": continue + if real: + current_binding, current_source_index = load_candidate_binding( + runner, **candidate_inputs, stage=f"before case {key}") + if current_binding != candidate_binding: + raise ValueError("candidate binding drift before next case") + verify_source_pins_at_candidate( + runner, plan["source_pins"], current_source_index, + candidate_binding["commit"]) # Source drift cannot consume the next case; never update a frozen plan. _verify_source_pins(runner, plan["source_pins"], "before next case") case, state, started = inventory[key], {"provider_call_count": 0}, perf_counter() @@ -1112,6 +1138,8 @@ def execute(runner, plan_path, expected_sha, output_dir, factory, *, resume=Fals runner.write_x(directory / "result.json", result) _bind_case_outcome(runner, out, ledger, case, "result.json") if result.get("passed") is not True: + if real: + ledger.trip_failure_class(key, "failed", "case result did not pass") break except Exception as error: failure = {"case_id": key, "status": "stopped", "passed": False, @@ -1120,9 +1148,32 @@ def execute(runner, plan_path, expected_sha, output_dir, factory, *, resume=Fals "elapsed_seconds": round(perf_counter() - started, 3), "automatic_retry": False, "not_legal_approval": True} runner.write_x(directory / "failure.json", failure) + if real: + 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( + runner, **candidate_inputs, stage="before final summary") + if final_binding != candidate_binding: + raise ValueError("candidate binding drift before final summary") + verify_source_pins_at_candidate( + runner, plan["source_pins"], final_source_index, + candidate_binding["commit"]) _verify_source_pins(runner, plan["source_pins"], "before final summary") snapshot = ledger.snapshot() rows = [_existing_case(runner, out, inventory[key], snapshot, ledger) for key in inventory] @@ -1251,12 +1302,31 @@ def execute(runner, plan_path, expected_sha, output_dir, factory, *, resume=Fals def dispatch(runner, args): - from multiscene_eval_plan import build_plan, validate_plan + from multiscene_eval_plan import ( + build_plan, load_candidate_binding, validate_plan, + verify_source_pins_at_candidate, + ) historical_usage_path = getattr(args, "historical_usage", None) run_registry_dir = getattr(args, "run_registry_dir", None) - if args.inputs is not None or args.source_manifest is not None: - raise SystemExit("multiscene mode derives its complete inventory; no legacy input/manifest override") + if args.inputs is not None: + raise SystemExit("multiscene mode derives its complete inventory; no legacy input override") + candidate_inputs = { + "commit": getattr(args, "candidate_commit", None), + "tag": getattr(args, "candidate_tag", None), + "source_manifest_path": getattr(args, "source_manifest", None), + "source_manifest_sha256": getattr(args, "source_manifest_sha256", None), + "source_manifest_payload_sha256": getattr( + args, "source_manifest_payload_sha256", None), + "image_identities_path": getattr(args, "image_identities", None), + "image_identities_sha256": getattr(args, "image_identities_sha256", None), + "image_identity_set_sha256": getattr( + args, "image_identity_set_sha256", None), + } + candidate_supplied = any(value is not None for value in candidate_inputs.values()) + candidate_complete = all(value is not None for value in candidate_inputs.values()) + if candidate_supplied and not candidate_complete: + raise SystemExit("immutable candidate arguments must be supplied as one complete set") if historical_usage_path is not None and not args.execute: raise SystemExit("--historical-usage is only valid with --execute") if run_registry_dir is not None and not args.execute: @@ -1264,7 +1334,13 @@ def dispatch(runner, args): if args.prepare: if args.resume or args.case_id: raise SystemExit("prepare freezes the complete matrix, not a selection") - plan = build_plan(runner) + if not candidate_complete: + raise SystemExit("RC-05 plan preparation requires a complete built candidate identity") + binding, source_index = load_candidate_binding( + runner, **candidate_inputs, stage="before plan preparation") + plan = build_plan(runner, binding) + verify_source_pins_at_candidate( + runner, plan["source_pins"], source_index, binding["commit"]) runner.write_x(args.plan, plan) return {"prepared": True, "case_count": len(plan["cases"]), "total_provider_call_budget": plan["total_provider_call_budget"], @@ -1280,6 +1356,8 @@ def dispatch(runner, args): raise SystemExit("real RC-05 execution requires --run-registry-dir") historical_usage = load_historical_usage(runner, historical_usage_path) execution_mode = "real" + if not candidate_complete: + raise SystemExit("real RC-05 execution requires a complete immutable candidate identity") else: from multiscene_supplemental_cases import synthetic_adapter_factory factory = synthetic_adapter_factory(runner) @@ -1287,9 +1365,16 @@ def dispatch(runner, args): execution_mode = "synthetic" return execute(runner, args.plan, args.plan_sha256, args.output_dir, factory, resume=args.resume, case_ids=args.case_id, historical_usage=historical_usage, - execution_mode=execution_mode, run_registry_dir=run_registry_dir) + execution_mode=execution_mode, run_registry_dir=run_registry_dir, + candidate_inputs=candidate_inputs if args.execute else None) if args.resume or args.case_id: raise SystemExit("--resume/--case-id require execution mode") - plan, digest = validate_plan(runner, args.plan, args.plan_sha256) + if candidate_complete: + binding, source_index = load_candidate_binding( + runner, **candidate_inputs, stage="during plan validation") + plan, digest = validate_plan( + runner, args.plan, args.plan_sha256, binding, source_index) + else: + plan, digest = validate_plan(runner, args.plan, args.plan_sha256) return {"validated": True, "case_count": len(plan["cases"]), "model_calls": 0, "plan_raw_sha256": digest} diff --git a/yuqei-infra/scripts/run-core-drafting-smoke.py b/yuqei-infra/scripts/run-core-drafting-smoke.py index 6ecfee6b..3fca3a8a 100644 --- a/yuqei-infra/scripts/run-core-drafting-smoke.py +++ b/yuqei-infra/scripts/run-core-drafting-smoke.py @@ -760,6 +760,13 @@ def main() -> int: parser.add_argument("--plan", type=Path, required=True) parser.add_argument("--inputs", type=Path) parser.add_argument("--source-manifest", type=Path) + parser.add_argument("--source-manifest-sha256") + parser.add_argument("--source-manifest-payload-sha256") + parser.add_argument("--candidate-commit") + parser.add_argument("--candidate-tag") + parser.add_argument("--image-identities", type=Path) + parser.add_argument("--image-identities-sha256") + parser.add_argument("--image-identity-set-sha256") parser.add_argument("--prepare", action="store_true") parser.add_argument("--synthetic", action="store_true") parser.add_argument("--execute", action="store_true") @@ -784,6 +791,10 @@ def main() -> int: raise SystemExit("--historical-usage requires --multiscene --execute") if args.run_registry_dir is not None: raise SystemExit("--run-registry-dir requires --multiscene --execute") + if any((args.source_manifest_sha256, args.source_manifest_payload_sha256, + args.candidate_commit, args.candidate_tag, args.image_identities, + args.image_identities_sha256, args.image_identity_set_sha256)): + raise SystemExit("immutable candidate arguments require --multiscene") if args.source_manifest is not None and not args.prepare: raise SystemExit("--source-manifest is only valid with --prepare") if args.prepare: diff --git a/yuqei-infra/tests/test_multiscene_base_journeys.py b/yuqei-infra/tests/test_multiscene_base_journeys.py index a126f5eb..fa0eed71 100644 --- a/yuqei-infra/tests/test_multiscene_base_journeys.py +++ b/yuqei-infra/tests/test_multiscene_base_journeys.py @@ -2,6 +2,7 @@ from __future__ import annotations import importlib.util +import json import tempfile from pathlib import Path @@ -61,7 +62,23 @@ def test_isolated_sqlite_flow_preserves_primary_error_and_still_releases_databas def _base_cases(): - plan = multiscene_eval_plan.build_plan(RUNNER) + images = [{"role": role, "repository": f"registry.example/{role}", + "manifest_digest": f"sha256:{index + 1:064x}", + "config_digest": f"sha256:{index + 11:064x}"} + for index, role in enumerate(multiscene_eval_plan.REQUIRED_IMAGE_ROLES)] + image_set_sha = RUNNER.sha256(json.dumps( + images, ensure_ascii=False, sort_keys=True, separators=(",", ":") + ).encode("utf-8")) + candidate = { + "schema_version": multiscene_eval_plan.CANDIDATE_SCHEMA_VERSION, + "commit": "a" * 40, "tag": "rc05-test-candidate", + "source_manifest": {"schema_version": "yuqei.release-source-manifest.v1", + "raw_sha256": "b" * 64, "payload_sha256": "c" * 64}, + "image_identities": {"schema_version": "yuqei.rc06-image-identities.v1", + "raw_sha256": "d" * 64, "identity_set_sha256": image_set_sha, + "images": images}, + } + plan = multiscene_eval_plan.build_plan(RUNNER, candidate) cases = [case for case in plan["cases"] if case.get("kind") == "drafting" and case.get("mode") == "contract"] assert len(cases) == 30 diff --git a/yuqei-infra/tests/test_multiscene_call_ledger.py b/yuqei-infra/tests/test_multiscene_call_ledger.py index cfc3b670..077b8ab4 100644 --- a/yuqei-infra/tests/test_multiscene_call_ledger.py +++ b/yuqei-infra/tests/test_multiscene_call_ledger.py @@ -71,6 +71,13 @@ def ledger(tmp_path, **overrides): return LEDGER.CallLedger(**values) +def _controls(case_classes=None, canaries=None): + case_classes = case_classes or {"purchase": "contract", "service": "specialist"} + return {"schema_version": "multiscene-release-controls.v1", + "canary_case_ids": canaries or ["purchase"], + "failure_classes": case_classes} + + def test_claim_precedes_call_and_finish_is_immutable(tmp_path): calls = [] book = ledger(tmp_path) @@ -89,6 +96,58 @@ def test_claim_precedes_call_and_finish_is_immutable(tmp_path): book.claim("purchase", "draft", SHA_A) +def test_canary_transition_is_explicit_persisted_and_required_before_full(tmp_path): + book = ledger(tmp_path, release_controls=_controls()) + with pytest.raises(LEDGER.CallLedgerError, match="explicit predefined canary"): + book.authorize_selection(["purchase", "service"], resume=False, + explicit_selection=False) + state = book.authorize_selection(["purchase"], resume=False, + explicit_selection=True) + 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"]) + with pytest.raises(LEDGER.CallLedgerError, match="outcome does not match"): + book.complete_canary({"purchase": SHA_B}) + book.complete_canary({"purchase": SHA_A}) + assert book.control_snapshot()["phase"] == "full_authorized" + resumed = ledger(tmp_path, release_controls=_controls()) + assert resumed.control_snapshot()["phase"] == "full_authorized" + resumed.authorize_selection(["service"], resume=True, explicit_selection=True) + + +def test_failure_class_breaker_persists_and_unknown_stays_in_denominator(tmp_path): + classes = {"purchase": "contract", "service": "specialist", "third": "specialist"} + book = ledger(tmp_path, case_limits={key: 1 for key in classes}, + operation_limits={"draft": 3}, + release_controls=_controls(classes, ["purchase"])) + 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}) + 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}, + release_controls=_controls(classes, ["purchase"])) + 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["release_controls"]["blocked_failure_classes"] == ["specialist"] + + +def test_legacy_ledger_schema_is_explicitly_read_only(tmp_path): + root = tmp_path / ".call-ledger" + root.mkdir() + (root / "identity.json").write_text( + json.dumps({"schema_version": "multiscene-call-ledger.v1"}), encoding="utf-8") + with pytest.raises(LEDGER.LedgerIdentityError, match="read-only diagnostic"): + ledger(tmp_path) + + def test_unknown_claim_blocks_automatic_replay_but_other_operation_is_distinct(tmp_path): book = ledger(tmp_path) book.claim("purchase", "semantic_review", SHA_A) diff --git a/yuqei-infra/tests/test_multiscene_eval_plan.py b/yuqei-infra/tests/test_multiscene_eval_plan.py index ae1810bc..62f1e57e 100644 --- a/yuqei-infra/tests/test_multiscene_eval_plan.py +++ b/yuqei-infra/tests/test_multiscene_eval_plan.py @@ -6,6 +6,23 @@ import pytest import multiscene_eval_plan as plan_module +def _candidate(): + images = [{"role": role, "repository": f"registry.example/{role}", + "manifest_digest": f"sha256:{index + 1:064x}", + "config_digest": f"sha256:{index + 11:064x}"} + for index, role in enumerate(plan_module.REQUIRED_IMAGE_ROLES)] + image_set_sha = hashlib.sha256(json.dumps( + images, ensure_ascii=False, sort_keys=True, separators=(",", ":") + ).encode("utf-8")).hexdigest() + return {"schema_version": plan_module.CANDIDATE_SCHEMA_VERSION, + "commit": "a" * 40, "tag": "rc05-test-candidate", + "source_manifest": {"schema_version": "yuqei.release-source-manifest.v1", + "raw_sha256": "b" * 64, "payload_sha256": "c" * 64}, + "image_identities": {"schema_version": "yuqei.rc06-image-identities.v1", + "raw_sha256": "d" * 64, "identity_set_sha256": image_set_sha, + "images": images}} + + def _write(path, value): path.write_text(json.dumps(value, ensure_ascii=False), encoding="utf-8") return hashlib.sha256(path.read_bytes()).hexdigest() @@ -13,10 +30,13 @@ def _write(path, value): @pytest.fixture(scope="module") def plan(): - return plan_module.build_plan(object()) + return plan_module.build_plan(object(), _candidate()) def test_plan_freezes_144_cases_280_calls_and_real_request_projections(plan): + assert plan["schema_version"] == "multiscene-evaluation-plan.v2" + assert plan["candidate"] == _candidate() + assert plan["release_controls"]["canary_case_ids"] == list(plan_module.CANARY_CASE_IDS) assert plan["case_count"] == 144 assert plan["total_provider_call_budget"] == 280 assert plan["operation_limits"] == plan_module.EXPECTED_OPERATIONS @@ -54,7 +74,7 @@ def test_validate_rebuilds_cases_budget_projection_and_source_pins(plan, tmp_pat assert plan_module.validate_plan(object(), path)[1] == digest -@pytest.mark.parametrize("mutation", ["case", "budget", "projection", "pin", "sha"]) +@pytest.mark.parametrize("mutation", ["case", "budget", "projection", "pin", "candidate", "controls", "sha"]) def test_validate_rejects_inventory_budget_projection_pin_or_user_sha_drift(plan, tmp_path, mutation): changed = json.loads(json.dumps(plan)) if mutation == "case": @@ -67,10 +87,52 @@ def test_validate_rejects_inventory_budget_projection_pin_or_user_sha_drift(plan elif mutation == "pin": first = next(iter(changed["source_pins"])) changed["source_pins"][first] = "0" * 64 + elif mutation == "candidate": + changed["candidate"]["commit"] = "f" * 40 + elif mutation == "controls": + changed["release_controls"]["canary_case_ids"].pop() path = tmp_path / f"{mutation}.json" digest = _write(path, changed) with pytest.raises(ValueError): - plan_module.validate_plan(object(), path, "f" * 64 if mutation == "sha" else digest) + plan_module.validate_plan( + object(), path, "f" * 64 if mutation == "sha" else digest, + _candidate() if mutation == "candidate" else None) + + +def test_legacy_plan_is_explicitly_read_only(plan, tmp_path): + legacy = {**plan, "schema_version": "multiscene-evaluation-plan.v1"} + path = tmp_path / "legacy.json" + _write(path, legacy) + with pytest.raises(ValueError, match="read-only diagnostic"): + plan_module.validate_plan(object(), path) + + +def test_git_candidate_gate_requires_exact_head_tag_and_clean_tree(monkeypatch, tmp_path): + commit = "a" * 40 + + def clean_git(_root, *args, **_kwargs): + if args[0] == "rev-parse": + return commit + if args[0] == "status": + return "" + raise AssertionError(args) + + monkeypatch.setattr(plan_module, "_git", clean_git) + plan_module._verify_git_candidate(tmp_path, commit, "rc05-candidate", "test") + + def dirty_git(_root, *args, **_kwargs): + return " M tracked.py" if args[0] == "status" else commit + + monkeypatch.setattr(plan_module, "_git", dirty_git) + with pytest.raises(ValueError, match="not clean"): + plan_module._verify_git_candidate(tmp_path, commit, "rc05-candidate", "test") + + monkeypatch.setattr( + plan_module, "_git", + lambda _root, *args, **_kwargs: ("b" * 40 if "refs/tags" in args[-1] else commit), + ) + with pytest.raises(ValueError, match="HEAD/tag mismatch"): + plan_module._verify_git_candidate(tmp_path, commit, "rc05-candidate", "test") def test_source_pins_are_repo_relative_existing_files_without_runtime_or_secrets(plan): @@ -83,3 +145,12 @@ def test_source_pins_are_repo_relative_existing_files_without_runtime_or_secrets assert "yuqei-infra/scripts/run-core-drafting-smoke.py" in plan["source_pins"] assert "yuqei-contract-platform/services/contract-api/tests/test_semantic_review_cross_service_flow.py" \ in plan["source_pins"] + + +def test_quality_gate_derives_each_alembic_head_and_checks_upgraded_revision(): + workflow = (plan_module.ROOT / ".github/workflows/build-test-quality-gate.yml").read_text( + encoding="utf-8") + assert "20260909_0079 (head)" not in workflow + assert "20260826_0028 (head)" not in workflow + assert workflow.count('expected_head="$(python -m alembic heads') == 2 + assert workflow.count('test "$actual_head" = "$expected_head"') == 2 diff --git a/yuqei-infra/tests/test_multiscene_eval_runner.py b/yuqei-infra/tests/test_multiscene_eval_runner.py index 9f911a08..6db9d449 100644 --- a/yuqei-infra/tests/test_multiscene_eval_runner.py +++ b/yuqei-infra/tests/test_multiscene_eval_runner.py @@ -43,10 +43,28 @@ def _plan(tmp_path, count=3): "kind": "supplemental", "operation": "draft", "operations": ["draft"]} for index in range(count)] + images = [{"role": role, "repository": f"example/{role}", + "manifest_digest": f"sha256:{index + 1:064x}", + "config_digest": f"sha256:{index + 11:064x}"} + for index, role in enumerate(("contract-web", "contract-api", + "ai-platform-api", "ai-worker"))] + candidate = {"schema_version": "multiscene-candidate-binding.v1", + "commit": "1" * 40, "tag": "test-candidate", + "source_manifest": {"schema_version": "yuqei.release-source-manifest.v1", + "raw_sha256": "2" * 64, "payload_sha256": "3" * 64}, + "image_identities": {"schema_version": "yuqei.rc06-image-identities.v1", + "raw_sha256": "4" * 64, + "identity_set_sha256": _sha(json.dumps( + images, ensure_ascii=False, sort_keys=True, separators=(",", ":") + ).encode("utf-8")), "images": images}} return { "cases": cases, "operation_limits": {"draft": count}, "source_pins": {"source.py": _sha(source.read_bytes())}, + "candidate": candidate, + "release_controls": {"schema_version": "multiscene-release-controls.v1", + "canary_case_ids": ["case-0"], + "failure_classes": {case["case_id"]: "draft" for case in cases}}, } @@ -55,6 +73,8 @@ def _install_modules(monkeypatch, plan, digest="a" * 64, calls=None): plan_module = types.ModuleType("multiscene_eval_plan") plan_module.validate_plan = lambda *_args: (plan, digest) plan_module.build_plan = lambda *_args: plan + plan_module.load_candidate_binding = lambda *_args, **_kwargs: (plan["candidate"], {}) + plan_module.verify_source_pins_at_candidate = lambda *_args, **_kwargs: None supplemental = types.ModuleType("multiscene_supplemental_cases") def run_case(case, directory, _factory, state, *, ledger, runner): @@ -99,6 +119,16 @@ def _runner(tmp_path): ) +def _candidate_inputs(): + return {"commit": "1" * 40, "tag": "test-candidate", + "source_manifest_path": Path("source.json"), + "source_manifest_sha256": "2" * 64, + "source_manifest_payload_sha256": "3" * 64, + "image_identities_path": Path("images.json"), + "image_identities_sha256": "4" * 64, + "image_identity_set_sha256": "5" * 64} + + def test_selected_execution_keeps_full_denominator_and_cannot_make_all_passed(monkeypatch, tmp_path): plan = _plan(tmp_path) calls = _install_modules(monkeypatch, plan) @@ -333,19 +363,23 @@ def test_real_execution_requires_and_binds_known_historical_usage(monkeypatch, t EVAL.execute(_runner(tmp_path), tmp_path / "plan.json", "a" * 64, tmp_path / "wrong-registry", RealFactory, historical_usage=_historical_usage(), execution_mode="real", - run_registry_dir=wrong_registry) + run_registry_dir=wrong_registry, case_ids=["case-0"], + candidate_inputs=_candidate_inputs()) with pytest.raises(ValueError, match="historical usage receipt"): EVAL.execute(_runner(tmp_path), tmp_path / "plan.json", "a" * 64, tmp_path / "missing-history", RealFactory, execution_mode="real", - run_registry_dir=registry) + run_registry_dir=registry, case_ids=["case-0"], + candidate_inputs=_candidate_inputs()) with pytest.raises(ValueError, match="historical usage declaration"): EVAL.execute(_runner(tmp_path), tmp_path / "plan.json", "a" * 64, tmp_path / "wrong-history", RealFactory, historical_usage={**_historical_usage(), "provider_call_count": 3}, - execution_mode="real", run_registry_dir=registry) + execution_mode="real", run_registry_dir=registry, case_ids=["case-0"], + candidate_inputs=_candidate_inputs()) summary = EVAL.execute(_runner(tmp_path), tmp_path / "plan.json", "a" * 64, tmp_path / "bound-history", RealFactory, historical_usage=_historical_usage(), - execution_mode="real", run_registry_dir=registry) + execution_mode="real", run_registry_dir=registry, case_ids=["case-0"], + candidate_inputs=_candidate_inputs()) assert calls == ["case-0"] assert summary["current_selection_passed"] is True assert summary["selected_passed"] is False @@ -378,6 +412,32 @@ def test_real_execution_requires_and_binds_known_historical_usage(monkeypatch, t assert len(identity["historical_usage"]["attempts_sha256"]) == 64 +def test_real_execution_rejects_unbound_candidate_and_initial_full_dispatch( + monkeypatch, tmp_path, +): + plan = _plan(tmp_path, count=2) + calls = _install_modules(monkeypatch, plan) + registry = tmp_path.parent / f"{tmp_path.name}-rc05-preflight-registry" + registry.mkdir() + monkeypatch.setenv(EVAL.REAL_RUN_REGISTRY_ENV, str(registry)) + with pytest.raises(ValueError, match="complete immutable candidate inputs"): + EVAL.execute( + _runner(tmp_path), tmp_path / "plan.json", "a" * 64, + tmp_path / "unbound", RealFactory, + historical_usage=_historical_usage(), execution_mode="real", + run_registry_dir=registry, case_ids=["case-0"], + ) + assert calls == [] + with pytest.raises(EVAL.CallLedgerError, match="explicit predefined canary"): + EVAL.execute( + _runner(tmp_path), tmp_path / "plan.json", "a" * 64, + tmp_path / "initial-full", RealFactory, + historical_usage=_historical_usage(), execution_mode="real", + run_registry_dir=registry, candidate_inputs=_candidate_inputs(), + ) + assert calls == [] + + def test_authoritative_campaign_registry_requires_explicit_empty_directory( monkeypatch, tmp_path, ): @@ -388,7 +448,8 @@ def test_authoritative_campaign_registry_requires_explicit_empty_directory( with pytest.raises(ValueError, match="must be pre-created"): EVAL.execute(_runner(tmp_path), tmp_path / "plan.json", "a" * 64, tmp_path / "out", RealFactory, historical_usage=_historical_usage(), - execution_mode="real", run_registry_dir=registry) + execution_mode="real", run_registry_dir=registry, case_ids=["case-0"], + candidate_inputs=_candidate_inputs()) assert calls == [] @@ -889,6 +950,7 @@ def test_foreign_plan_campaign_attempt_blocks_current_plan_green(monkeypatch, tm _runner(tmp_path), tmp_path / "plan.json", "2" * 64, tmp_path / "current-output", RealFactory, historical_usage=usage, execution_mode="real", run_registry_dir=registry, + case_ids=["case-0"], candidate_inputs=_candidate_inputs(), ) assert summary["ledger"]["campaign"]["attempted"] == 2 assert summary["ledger"]["campaign"]["foreign_plan_attempted"] == 1