diff --git a/yuqei-infra/scripts/apply-rc06-publication-batch.py b/yuqei-infra/scripts/apply-rc06-publication-batch.py index e294c747..99de5e26 100644 --- a/yuqei-infra/scripts/apply-rc06-publication-batch.py +++ b/yuqei-infra/scripts/apply-rc06-publication-batch.py @@ -57,7 +57,7 @@ batch = _load_builder() SCENE_LEGAL_REVIEW_BUILDER_PATH = SCRIPT_DIR / "build-scene-source-legal-review.py" -APPLY_REQUEST_SCHEMA_VERSION = "yuqei.rc06-online-apply-request.v2" +APPLY_REQUEST_SCHEMA_VERSION = "yuqei.rc06-online-apply-request.v3" APPLY_RESPONSE_SCHEMA_VERSION = "yuqei.rc06-online-apply-response.v2" APPLY_STATUS_SCHEMA_VERSION = "yuqei.rc06-online-apply-status.v2" ROLLBACK_REQUEST_SCHEMA_VERSION = "yuqei.rc06-online-rollback-request.v2" @@ -975,6 +975,7 @@ def execute_online_publication_batch( approval: Mapping[str, Any], approval_raw_sha256: str, approval_keyring: Mapping[str, Any], + approval_keyring_raw_sha256: str, approval_root_public_key: Ed25519PublicKey, apply_service_public_key: Ed25519PublicKey, online_service_public_key: Any, @@ -1003,6 +1004,11 @@ def execute_online_publication_batch( ) manifest_raw_sha256 = batch._require_sha(manifest_raw_sha256, "manifest raw SHA-256") approval_raw_sha256 = batch._require_sha(approval_raw_sha256, "approval raw SHA-256") + approval_keyring_raw_sha256 = batch._require_sha( + approval_keyring_raw_sha256, + "approval keyring raw SHA-256", + ) + approval_keyring_canonical_sha256 = batch._sha256_json(approval_keyring) environment_id = _require_text(environment_id, "environment_id", maximum=256) tenant_id = _require_text(tenant_id, "tenant_id", maximum=64) batch.validate_approval_receipt( @@ -1074,6 +1080,9 @@ def execute_online_publication_batch( "batch_manifest_raw_sha256": manifest_raw_sha256, "batch_manifest_payload_sha256": manifest["manifest_payload_sha256"], "approval_receipt_sha256": approval_raw_sha256, + "approval_keyring": approval_keyring, + "approval_keyring_raw_sha256": approval_keyring_raw_sha256, + "approval_keyring_canonical_sha256": approval_keyring_canonical_sha256, "legal_source_review_binding_sha256": legal_review_binding_sha256, "legal_source_review_expires_at": legal_review_expires_at, "authority_challenge": authority_challenge, @@ -1679,7 +1688,7 @@ def main(argv: list[str] | None = None) -> int: approval, approval_raw = batch.secure_read_json( args.approval, "approval receipt", args.approval_sha256 ) - keyring, _ = batch.secure_read_json( + keyring, keyring_raw = batch.secure_read_json( args.approval_keyring, "approval keyring", args.approval_keyring_sha256, @@ -1738,6 +1747,7 @@ def main(argv: list[str] | None = None) -> int: approval=approval, approval_raw_sha256=approval_raw, approval_keyring=keyring, + approval_keyring_raw_sha256=keyring_raw, approval_root_public_key=root_key, apply_service_public_key=apply_key, online_service_public_key=online_key, diff --git a/yuqei-infra/scripts/build-rc06-publication-batch.py b/yuqei-infra/scripts/build-rc06-publication-batch.py index f988d917..bae482ee 100644 --- a/yuqei-infra/scripts/build-rc06-publication-batch.py +++ b/yuqei-infra/scripts/build-rc06-publication-batch.py @@ -89,7 +89,7 @@ TRUSTED_SCENE_LEGAL_REVIEW_KEYRING_SHA256 = "" RC05_SUMMARY_SCHEMA_VERSION = "multiscene-evaluation-run.v1" RC05_PLAN_SCHEMA_VERSION = "multiscene-evaluation-plan.v2" RC05_CANDIDATE_SCHEMA_VERSION = "multiscene-candidate-binding.v1" -RC05_RELEASE_CONTROLS_SCHEMA_VERSION = "multiscene-release-controls.v1" +RC05_RELEASE_CONTROLS_SCHEMA_VERSION = "multiscene-release-controls.v2" RC05_LEDGER_SCHEMA_VERSION = "multiscene-call-ledger.v2" RC05_LEDGER_SNAPSHOT_SCHEMA_VERSION = "multiscene-call-ledger-snapshot.v2" RC05_CAMPAIGN_SCHEMA_VERSION = "multiscene-call-campaign.v2" @@ -1291,16 +1291,26 @@ def _validate_rc05_release_controls( controls = _require_keys( value, "RC-05 release controls", - ("schema_version", "canary_case_ids", "failure_classes"), + ( + "schema_version", "canary_case_ids", "failure_classes", + "expected_operations", + ), ) canaries = controls["canary_case_ids"] failure_classes = controls["failure_classes"] + expected_operations = controls["expected_operations"] case_ids = [case["case_id"] for case in cases] + expected_operation_map = { + case["case_id"]: list(case["operations"]) + for case in cases + } if ( controls["schema_version"] != RC05_RELEASE_CONTROLS_SCHEMA_VERSION or canaries != list(EXPECTED_RC05_CANARY_CASE_IDS) or not isinstance(failure_classes, dict) or set(failure_classes) != set(case_ids) + or not isinstance(expected_operations, dict) + or expected_operations != expected_operation_map ): raise PublicationBatchError("RC-05 release controls identity or coverage drift") for case in cases: @@ -1541,20 +1551,52 @@ def validate_rc05_plan_identity(value: Any) -> dict[str, Any]: controls = _require_keys( identity["release_controls"], "RC-05 plan identity release controls", - ("schema_version", "canary_case_ids", "failure_classes"), + ( + "schema_version", "canary_case_ids", "failure_classes", + "expected_operations", + ), ) + failure_classes = controls["failure_classes"] + expected_operations = controls["expected_operations"] + operation_counts = { + operation: 0 for operation in EXPECTED_RC05_OPERATION_LIMITS + } + expected_operations_valid = ( + isinstance(expected_operations, dict) + and isinstance(failure_classes, dict) + and set(expected_operations) == set(failure_classes) + ) + if expected_operations_valid: + for case_id, operations in expected_operations.items(): + if ( + not isinstance(case_id, str) + or not STABLE_ID_RE.fullmatch(case_id) + or not isinstance(operations, list) + or not operations + or len(operations) != len(set(operations)) + or any( + operation not in EXPECTED_RC05_OPERATION_LIMITS + for operation in operations + ) + ): + expected_operations_valid = False + break + for operation in operations: + operation_counts[operation] += 1 if ( controls["schema_version"] != RC05_RELEASE_CONTROLS_SCHEMA_VERSION or controls["canary_case_ids"] != list(EXPECTED_RC05_CANARY_CASE_IDS) - or not isinstance(controls["failure_classes"], dict) - or len(controls["failure_classes"]) != 144 + or not isinstance(failure_classes, dict) + or len(failure_classes) != 144 or any( not isinstance(key, str) or not STABLE_ID_RE.fullmatch(key) or not isinstance(value, str) or not STABLE_ID_RE.fullmatch(value) - for key, value in controls["failure_classes"].items() + for key, value in failure_classes.items() ) + or not expected_operations_valid + or operation_counts != EXPECTED_RC05_OPERATION_LIMITS ): raise PublicationBatchError("RC-05 plan identity release controls drift") return dict(identity) @@ -1956,16 +1998,32 @@ def validate_rc05_terminal_evidence( expected_observed=len(result_cases), ) - snapshot = run["ledger"] - if not isinstance(snapshot, dict): - raise PublicationBatchError("RC-05 terminal ledger snapshot is missing") + snapshot = _require_keys( + run["ledger"], + "RC-05 terminal ledger snapshot", + ( + "schema_version", "ledger_identity_sha256", "real_provider", + "authority", "budget", "operations", "cases", "campaign", + "attempted", "terminal", "unknown", "provider_terminal", + "release_controls", + ), + ) + snapshot_attempted = _require_count( + snapshot["attempted"], "RC-05 ledger attempted", maximum=280 + ) + snapshot_terminal = _require_count( + snapshot["terminal"], "RC-05 ledger terminal", maximum=280 + ) + snapshot_unknown = _require_count( + snapshot["unknown"], "RC-05 ledger unknown", maximum=280 + ) if ( snapshot.get("schema_version") != RC05_LEDGER_SNAPSHOT_SCHEMA_VERSION or snapshot.get("ledger_identity_sha256") != ledger_identity_sha256 or snapshot.get("real_provider") is not True or snapshot.get("authority") != authority - or snapshot.get("unknown") != 0 - or snapshot.get("terminal") != snapshot.get("attempted") + or snapshot_unknown != 0 + or snapshot_terminal != snapshot_attempted ): raise PublicationBatchError("RC-05 terminal ledger snapshot identity or terminal counts drift") control_state = _require_keys( @@ -2007,6 +2065,11 @@ def validate_rc05_terminal_evidence( }: raise PublicationBatchError(f"RC-05 ledger case counts drift: {case_id}") current_attempted = sum(row["provider_call_count"] for row in result_cases.values()) + provider_terminal = _require_count( + snapshot["provider_terminal"], + "RC-05 ledger provider-terminal count", + maximum=280, + ) repair_case_calls = sum( result_cases[case_id]["provider_call_count"] - sum( @@ -2016,7 +2079,8 @@ def validate_rc05_terminal_evidence( for case_id, plan_case in plan_cases.items() ) if ( - snapshot["attempted"] != current_attempted + snapshot_attempted != current_attempted + or provider_terminal != current_attempted or run["provider_call_count"] != current_attempted or run["current_plan_provider_call_count"] != current_attempted or run["provider_call_count_scope"] != "current_plan" @@ -2233,13 +2297,63 @@ def validate_rc05_terminal_evidence( }: raise PublicationBatchError("RC-05 historical_by_operation projection drift") - campaign_counts = snapshot.get("campaign") + campaign_counts = _require_keys( + snapshot["campaign"], + "RC-05 terminal campaign counts", + ( + "attempted", "terminal", "unknown", "provider_terminal", + "current_plan_attempted", "foreign_plan_attempted", + "provider_outcomes", + ), + ) + campaign_attempted = _require_count( + campaign_counts["attempted"], "RC-05 campaign attempted", maximum=280 + ) + campaign_terminal = _require_count( + campaign_counts["terminal"], "RC-05 campaign terminal", maximum=280 + ) + campaign_unknown = _require_count( + campaign_counts["unknown"], "RC-05 campaign unknown", maximum=280 + ) + campaign_provider_terminal = _require_count( + campaign_counts["provider_terminal"], + "RC-05 campaign provider-terminal count", + maximum=280, + ) + campaign_current_plan = _require_count( + campaign_counts["current_plan_attempted"], + "RC-05 campaign current-plan attempted", + maximum=280, + ) + campaign_foreign_plan = _require_count( + campaign_counts["foreign_plan_attempted"], + "RC-05 campaign foreign-plan attempted", + maximum=280, + ) + campaign_provider_outcomes = _require_keys( + campaign_counts["provider_outcomes"], + "RC-05 campaign provider outcomes", + ("succeeded", "failed", "cancelled"), + ) + for status, value in campaign_provider_outcomes.items(): + _require_count( + value, f"RC-05 campaign provider outcome {status}", maximum=280 + ) combined_counts = run["combined_call_counts"] if ( - not isinstance(campaign_counts, dict) - or campaign_counts.get("attempted") != current_attempted - or campaign_counts.get("terminal") != current_attempted - or campaign_counts.get("unknown") != 0 + campaign_attempted != current_attempted + or campaign_terminal != current_attempted + or campaign_unknown != 0 + or campaign_provider_terminal != current_attempted + or campaign_provider_terminal != provider_terminal + or campaign_current_plan != current_attempted + or campaign_foreign_plan != 0 + or campaign_current_plan + campaign_foreign_plan != campaign_attempted + or campaign_provider_outcomes != { + "succeeded": current_attempted, + "failed": 0, + "cancelled": 0, + } or run["campaign_new_provider_call_count"] != current_attempted or combined_counts != { "attempted": combined_attempted, diff --git a/yuqei-infra/tests/test_rc06_publication_apply.py b/yuqei-infra/tests/test_rc06_publication_apply.py index b1ca2ca2..bbd4f30d 100644 --- a/yuqei-infra/tests/test_rc06_publication_apply.py +++ b/yuqei-infra/tests/test_rc06_publication_apply.py @@ -452,6 +452,9 @@ def _material(monkeypatch: pytest.MonkeyPatch, now: datetime) -> dict: "approval": approval, "approval_raw": approval_raw, "approval_keyring": approval_keyring, + "approval_keyring_raw": apply.batch._sha256_bytes( + apply.batch._canonical(approval_keyring) + ), "legal_request": legal_request, "legal_decision": legal_decision, "legal_keyring": legal_keyring, @@ -880,6 +883,7 @@ def _execute(material: dict, client, ledger) -> dict: approval=material["approval"], approval_raw_sha256=material["approval_raw"], approval_keyring=material["approval_keyring"], + approval_keyring_raw_sha256=material["approval_keyring_raw"], approval_root_public_key=material["root_private"].public_key(), apply_service_public_key=material["apply_private"].public_key(), online_service_public_key=material["online_private"].public_key(), @@ -1101,6 +1105,17 @@ def test_online_execute_then_offline_countersign_is_durable_and_idempotent( ledger = apply.ApplyLedger(tmp_path / "ledger.sqlite3") try: execution = _execute(material, client, ledger) + assert service.request is not None + assert service.request["schema_version"] == ( + "yuqei.rc06-online-apply-request.v3" + ) + assert service.request["approval_keyring"] == material["approval_keyring"] + assert service.request["approval_keyring_raw_sha256"] == material[ + "approval_keyring_raw" + ] + assert service.request["approval_keyring_canonical_sha256"] == ( + apply.batch._sha256_json(material["approval_keyring"]) + ) assert _execute(material, client, ledger) == execution state = ledger.connection.execute( "SELECT state FROM rc06_apply_operations" @@ -1207,6 +1222,26 @@ def test_expired_challenge_is_rejected_before_initial_dispatch( assert service.apply_calls == 0 +def test_approval_keyring_raw_digest_is_required_before_dispatch( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + material = _material(monkeypatch, datetime(2026, 10, 6, 8, 0, tzinfo=UTC)) + material["approval_keyring_raw"] = "not-a-sha256" + service = _service(material) + with _serve(service) as url: + client = apply.JsonHttpClient(url, "test-token", allow_http_loopback=True) + ledger = apply.ApplyLedger(tmp_path / "ledger.sqlite3") + try: + with pytest.raises( + apply.batch.PublicationBatchError, + match="approval keyring raw SHA-256", + ): + _execute(material, client, ledger) + finally: + ledger.close() + assert service.apply_calls == 0 + + def test_expired_approval_is_rejected_independently_before_http( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: diff --git a/yuqei-infra/tests/test_rc06_publication_batch.py b/yuqei-infra/tests/test_rc06_publication_batch.py index f0db6130..e82bca66 100644 --- a/yuqei-infra/tests/test_rc06_publication_batch.py +++ b/yuqei-infra/tests/test_rc06_publication_batch.py @@ -196,6 +196,10 @@ def _rc05_plan() -> dict: "schema_version": batch.RC05_RELEASE_CONTROLS_SCHEMA_VERSION, "canary_case_ids": list(batch.EXPECTED_RC05_CANARY_CASE_IDS), "failure_classes": failure_classes, + "expected_operations": { + case["case_id"]: list(case["operations"]) + for case in cases + }, }, "source_pins": {"source.py": "0" * 64}, "cases": cases, @@ -436,10 +440,23 @@ def _terminal_documents(plan: dict, plan_sha: str = "3" * 64): }, "operations": operation_snapshot, "cases": case_counts, - "campaign": {"attempted": current_calls, "terminal": current_calls, "unknown": 0}, + "campaign": { + "attempted": current_calls, + "terminal": current_calls, + "unknown": 0, + "provider_terminal": current_calls, + "current_plan_attempted": current_calls, + "foreign_plan_attempted": 0, + "provider_outcomes": { + "succeeded": current_calls, + "failed": 0, + "cancelled": 0, + }, + }, "attempted": current_calls, "terminal": current_calls, "unknown": 0, + "provider_terminal": current_calls, "release_controls": { "schema_version": batch.RC05_CONTROL_STATE_SCHEMA_VERSION, "phase": "full_authorized", @@ -2626,6 +2643,26 @@ def test_rc05_plan_requires_exact_budget_operations_and_unique_case_inventory() with pytest.raises(batch.PublicationBatchError, match="release controls"): batch.validate_rc05_plan(control_drift) + legacy_controls = copy.deepcopy(plan) + legacy_controls["release_controls"]["schema_version"] = ( + "multiscene-release-controls.v1" + ) + with pytest.raises(batch.PublicationBatchError, match="release controls"): + batch.validate_rc05_plan(legacy_controls) + + missing_expected_operations = copy.deepcopy(plan) + missing_expected_operations["release_controls"].pop("expected_operations") + with pytest.raises(batch.PublicationBatchError, match="keys invalid"): + batch.validate_rc05_plan(missing_expected_operations) + + operation_binding_drift = copy.deepcopy(plan) + first_case_id = operation_binding_drift["cases"][0]["case_id"] + operation_binding_drift["release_controls"]["expected_operations"][ + first_case_id + ] = ["intake"] + with pytest.raises(batch.PublicationBatchError, match="release controls"): + batch.validate_rc05_plan(operation_binding_drift) + drifted = copy.deepcopy(plan) drifted["total_provider_call_budget"] = 281 with pytest.raises(batch.PublicationBatchError, match="budget drift"): @@ -2653,6 +2690,13 @@ def test_rc05_plan_requires_exact_budget_operations_and_unique_case_inventory() with pytest.raises(batch.PublicationBatchError, match="keys invalid"): batch.validate_rc05_plan_identity(invalid_identity) + invalid_control_identity = copy.deepcopy(identity) + invalid_control_identity["release_controls"]["expected_operations"][ + first_case_id + ] = ["intake"] + with pytest.raises(batch.PublicationBatchError, match="release controls"): + batch.validate_rc05_plan_identity(invalid_control_identity) + def test_secure_json_rejects_duplicate_keys_and_non_finite_values(tmp_path: Path) -> None: duplicate = tmp_path / "duplicate.json" @@ -2741,6 +2785,239 @@ def test_rc05_terminal_evidence_requires_full_real_green_chain_and_remediation() ) +@pytest.mark.parametrize( + ("path", "value"), + ( + (("provider_terminal",), 263), + (("campaign", "provider_terminal"), 263), + (("campaign", "current_plan_attempted"), 263), + (("campaign", "foreign_plan_attempted"), 1), + ( + ("campaign", "provider_outcomes"), + {"succeeded": 263, "failed": 1, "cancelled": 0}, + ), + ), +) +def test_rc05_terminal_rejects_rehashed_provider_and_campaign_count_tamper( + path: tuple[str, ...], value, +) -> None: + plan = _rc05_plan() + plan_raw_sha256 = batch._sha256_json(plan) + summary, ledger, campaign = _terminal_documents(plan, plan_raw_sha256) + target = summary["ledger"] + for key in path[:-1]: + target = target[key] + target[path[-1]] = value + + # Model an attacker who rewrites every supplied document digest after + # changing a semantically authoritative terminal counter. + with pytest.raises(batch.PublicationBatchError): + batch.validate_rc05_terminal_evidence( + plan, + plan_raw_sha256, + summary, + batch._sha256_json(summary), + ledger, + batch._sha256_json(ledger), + campaign, + batch._sha256_json(campaign), + ) + + +@pytest.mark.parametrize( + ("section", "field"), + ( + ("ledger", "provider_terminal"), + ("campaign", "provider_terminal"), + ("campaign", "current_plan_attempted"), + ("campaign", "foreign_plan_attempted"), + ("provider_outcomes", "cancelled"), + ), +) +def test_rc05_terminal_requires_exact_provider_campaign_snapshot_shape( + section: str, field: str, +) -> None: + plan = _rc05_plan() + summary, ledger, campaign = _terminal_documents(plan) + if section == "ledger": + summary["ledger"].pop(field) + elif section == "campaign": + summary["ledger"]["campaign"].pop(field) + else: + summary["ledger"]["campaign"]["provider_outcomes"].pop(field) + with pytest.raises(batch.PublicationBatchError, match="keys invalid"): + batch.validate_rc05_terminal_evidence( + plan, "3" * 64, summary, batch._sha256_json(summary), + ledger, batch._sha256_json(ledger), + campaign, batch._sha256_json(campaign), + ) + + +def test_actual_rc05_v2_ledger_snapshot_feeds_rc06_terminal_builder( + tmp_path: Path, +) -> None: + ledger_script = ( + REPO_ROOT / "yuqei-infra" / "scripts" / "multiscene_call_ledger.py" + ) + spec = importlib.util.spec_from_file_location( + "rc05_ledger_producer_contract_test", ledger_script + ) + assert spec is not None and spec.loader is not None + producer = importlib.util.module_from_spec(spec) + spec.loader.exec_module(producer) + if getattr(producer, "LEDGER_SCHEMA", None) != batch.RC05_LEDGER_SCHEMA_VERSION: + pytest.skip("requires the integrated RC-05 v2 producer commit") + + plan = _rc05_plan() + plan_raw_sha256 = batch._sha256_json(plan) + summary, fixture_ledger, _fixture_campaign = _terminal_documents( + plan, plan_raw_sha256 + ) + historical_input = { + key: value + for key, value in fixture_ledger["historical_usage"].items() + if key != "attempts_sha256" + } + registry = tmp_path / "registry" + registry.mkdir() + ledger = producer.CallLedger( + tmp_path / "output", + plan_raw_sha256, + plan["operation_limits"], + { + case["case_id"]: case["provider_call_budget"] + for case in plan["cases"] + }, + plan["source_pins"], + True, + historical_usage=historical_input, + registry_root=registry, + candidate_binding=plan["candidate"], + release_controls=plan["release_controls"], + ) + + after_names = { + "draft": "01-draft-after.json", + "semantic_review": "02-semantic-after.json", + "specialist_review": "02-specialist-after.json", + "intake": "02-after.json", + } + job_operations = { + "draft": "contract_draft", + "semantic_review": "contract_draft_semantic_review", + "specialist_review": "contract_review", + "intake": "contract_intake", + } + + def write_json(path: Path, payload: dict) -> str: + raw = json.dumps( + payload, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode("utf-8") + path.write_bytes(raw) + return hashlib.sha256(raw).hexdigest() + + def complete_case(case: dict) -> str: + case_id = case["case_id"] + case_dir = ledger.output_dir / case_id + case_dir.mkdir() + evidence_sha256s = {} + actual_operations = [ + operation + for operation in case["operations"] + if operation not in {"repair_draft", "repair_review"} + ] + for operation in actual_operations: + request_sha256 = batch._sha256_json( + {"case_id": case_id, "operation": operation, "kind": "request"} + ) + response_sha256 = batch._sha256_json( + {"case_id": case_id, "operation": operation, "kind": "response"} + ) + token = ledger.claim(case_id, operation, request_sha256) + ledger.finish(token, "succeeded", response_sha256) + audit_run_id = f"audit-{case_id}-{operation}" + evidence_sha256s[after_names[operation]] = write_json( + case_dir / after_names[operation], + { + "case_id": case_id, + "operation": operation, + "request_sha256": request_sha256, + "response_sha256": response_sha256, + "provider_invocation_count": 1, + "execution_elapsed_seconds": 1.0, + "job": { + "id": f"job-{case_id}-{operation}", + "status": "succeeded", + "provider_status": "succeeded", + "audit_run_id": audit_run_id, + "operation": job_operations[operation], + }, + "audit": { + "id": audit_run_id, + "provider": "codex-cli-development", + "model": "gpt-5.6-sol", + "provider_status": "succeeded", + }, + }, + ) + result_sha256 = write_json( + case_dir / "result.json", + { + "case_id": case_id, + "status": "completed", + "passed": True, + "provider_call_count": len(actual_operations), + "automated_quality_passed": True, + "automated_model_quality_release_passed": True, + "content_quality_release_passed": True, + }, + ) + ledger.bind_case_outcome( + case_id, + "result.json", + result_sha256, + evidence_sha256s, + case["operations"], + ) + return result_sha256 + + case_index = {case["case_id"]: case for case in plan["cases"]} + canary_ids = plan["release_controls"]["canary_case_ids"] + assert ledger.authorize_selection( + canary_ids, resume=False, explicit_selection=True + )["phase"] == "canary_required" + canary_outcomes = { + case_id: complete_case(case_index[case_id]) for case_id in canary_ids + } + ledger.complete_canary(canary_outcomes) + remaining = [case for case in plan["cases"] if case["case_id"] not in canary_ids] + assert ledger.authorize_selection( + [case["case_id"] for case in remaining], + resume=True, + explicit_selection=True, + )["phase"] == "full_authorized" + for case in remaining: + complete_case(case) + + summary["ledger"] = ledger.snapshot() + terminal_identity = batch.validate_rc05_terminal_evidence( + plan, + plan_raw_sha256, + summary, + batch._sha256_json(summary), + ledger.identity, + batch._sha256_json(ledger.identity), + ledger.campaign_identity, + batch._sha256_json(ledger.campaign_identity), + ) + assert terminal_identity["complete_real_green"] is True + assert terminal_identity["current_plan_provider_call_count"] == 264 + + @pytest.mark.parametrize( ("mutation", "message"), (