From b90fbd01db69312e3e9e9e68e5c8e047cc000334 Mon Sep 17 00:00:00 2001 From: yuqei Date: Tue, 6 Oct 2026 18:11:09 +0800 Subject: [PATCH] feat(release): add fail-closed RC06 apply protocol --- .../scripts/apply-rc06-publication-batch.py | 1013 +++++++++++++++++ .../scripts/build-rc06-publication-batch.py | 254 ++++- .../tests/test_rc06_publication_apply.py | 844 ++++++++++++++ 3 files changed, 2106 insertions(+), 5 deletions(-) create mode 100644 yuqei-infra/scripts/apply-rc06-publication-batch.py create mode 100644 yuqei-infra/tests/test_rc06_publication_apply.py diff --git a/yuqei-infra/scripts/apply-rc06-publication-batch.py b/yuqei-infra/scripts/apply-rc06-publication-batch.py new file mode 100644 index 00000000..eeea0d58 --- /dev/null +++ b/yuqei-infra/scripts/apply-rc06-publication-batch.py @@ -0,0 +1,1013 @@ +#!/usr/bin/env python3 +"""Protocol client for applying one frozen RC-06 batch through a real service. + +The service contract is deliberately fail-closed. A successful return requires +the exact 60 assets and 14 Prompt modules to be committed, read back online, and +covered by the existing independently signed RC-06 apply receipt. This client +never accepts an unsigned HTTP response as publication evidence. + +The legal-source request/decision/keyring checked here is an externally signed +assertion. This program delegates identity, qualification, freshness, status +URL, rationale and exact source hashes to the canonical scene-review validator; +it does not perform legal review or infer professional status. + +This file does not implement the cross-Contract/AI transaction service. RC-06 +remains blocked until a deployed endpoint can truthfully produce the signed +single-service transaction, rollback and online read-back evidence required by +this client; ordinary sequential publish endpoints are intentionally rejected. +""" + +from __future__ import annotations + +import argparse +import base64 +import hashlib +import importlib.util +import json +import os +import sqlite3 +import sys +import urllib.error +import urllib.parse +import urllib.request +from datetime import UTC, datetime +from pathlib import Path +from typing import Any, Mapping + +from cryptography.exceptions import InvalidSignature +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey + + +SCRIPT_DIR = Path(__file__).resolve().parent +REPO_ROOT = SCRIPT_DIR.parents[1] +BUILDER_PATH = SCRIPT_DIR / "build-rc06-publication-batch.py" + + +def _load_builder(): + spec = importlib.util.spec_from_file_location("rc06_publication_batch_builder", BUILDER_PATH) + if spec is None or spec.loader is None: + raise RuntimeError("cannot load RC-06 publication batch builder") + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +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.v1" +APPLY_RESPONSE_SCHEMA_VERSION = "yuqei.rc06-online-apply-response.v1" +APPLY_STATUS_SCHEMA_VERSION = "yuqei.rc06-online-apply-status.v1" +ROLLBACK_REQUEST_SCHEMA_VERSION = "yuqei.rc06-online-rollback-request.v1" +ROLLBACK_RESPONSE_SCHEMA_VERSION = "yuqei.rc06-online-rollback-response.v1" +RUNTIME_RECEIPT_SCHEMA_VERSION = "yuqei.rc06-runtime-snapshot-receipt.v1" +RUNTIME_PAYLOAD_SCHEMA_VERSION = "yuqei.rc06-runtime-snapshot-payload.v1" +RUNTIME_SIGNATURE_DOMAIN = b"YUQEI_RC06_RUNTIME_SNAPSHOT_RECEIPT_V1\0" +EXECUTION_EVIDENCE_SCHEMA_VERSION = "yuqei.rc06-online-apply-evidence.v1" +AUTH_TOKEN_ENV = "RC06_APPLY_BEARER_TOKEN" +MAX_HTTP_BODY_BYTES = 16 * 1024 * 1024 + +class ApplyPublicationError(batch.PublicationBatchError): + """The online transaction could not be proven safe and complete.""" + + +def _require_exact(value: Any, label: str, keys: tuple[str, ...]) -> dict[str, Any]: + if not isinstance(value, dict) or set(value) != set(keys): + raise ApplyPublicationError(f"{label} fields are invalid") + return value + + +def _require_text(value: Any, label: str, *, maximum: int = 512) -> str: + if not isinstance(value, str) or not value.strip() or value != value.strip() or len(value) > maximum: + raise ApplyPublicationError(f"{label} is invalid") + return value + + +def _sha256(value: bytes) -> str: + return hashlib.sha256(value).hexdigest() + + +def _public_raw(key: Ed25519PublicKey) -> bytes: + return key.public_bytes(serialization.Encoding.Raw, serialization.PublicFormat.Raw) + + +def _load_scene_legal_review_module(): + if not SCENE_LEGAL_REVIEW_BUILDER_PATH.is_file(): + raise ApplyPublicationError( + "scene legal review validator is absent from the frozen candidate" + ) + spec = importlib.util.spec_from_file_location( + "rc06_scene_source_legal_review", SCENE_LEGAL_REVIEW_BUILDER_PATH + ) + if spec is None or spec.loader is None: + raise ApplyPublicationError("cannot load scene legal review validator") + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +def validate_scene_legal_review( + manifest: Mapping[str, Any], + approval: Mapping[str, Any], + request: Any, + decision: Any, + keyring: Any, + *, + now: datetime, +) -> tuple[dict[str, Any], str]: + """Delegate the 30-scene/43-decision proof to its canonical validator.""" + + if not isinstance(request, dict) or not isinstance(decision, dict) or not isinstance(keyring, dict): + raise ApplyPublicationError("scene legal review request, decision and keyring are required") + if request.get("candidate_commit") != manifest["candidate"]["commit"]: + raise ApplyPublicationError("scene legal review candidate does not match the publication batch") + scope = request.get("scope") + if not isinstance(scope, dict) or scope.get("scene_count") != 30 or scope.get( + "reference_occurrence_count" + ) != 43: + raise ApplyPublicationError("scene legal review must cover 30 scenes and 43 references") + request_sha256 = request.get("request_sha256") + if not isinstance(request_sha256, str) or not batch.HEX64_RE.fullmatch(request_sha256): + raise ApplyPublicationError("scene legal review request_sha256 is invalid") + if decision.get("request_sha256") != request_sha256: + raise ApplyPublicationError("scene legal review decision is not bound to its request") + validator = _load_scene_legal_review_module() + try: + verified_request = validator.verify_request(request, REPO_ROOT) + summary = validator.verify_signed_decision( + verified_request, + decision, + keyring, + checked_at=now.astimezone(UTC).isoformat(timespec="seconds"), + ) + except Exception as exc: + raise ApplyPublicationError(f"scene legal review validation failed: {exc}") from exc + if ( + summary.get("status") != "verified_qualified_human_review" + or summary.get("request_sha256") != request_sha256 + or summary.get("reviewed_occurrence_count") != 43 + or summary.get("all_current_effectiveness_verified") is not True + or summary.get("all_transaction_applicability_verified") is not True + or summary.get("authorizes_publication_by_itself") is not False + ): + raise ApplyPublicationError("scene legal review verification summary is incomplete") + approval_expires_at = batch._parse_time( + approval["payload"]["expires_at"], "approval expires_at" + ) + legal_expires_at = batch._parse_time(summary.get("expires_at"), "legal review expires_at") + if legal_expires_at > approval_expires_at: + raise ApplyPublicationError("scene legal review validity exceeds the approval window") + binding = { + "request_sha256": request_sha256, + "decision_sha256": batch._sha256_json(decision), + "keyring_sha256": batch._sha256_json(keyring), + } + return summary, batch._sha256_json(binding) + + +def expected_runtime_snapshot_results(manifest: Mapping[str, Any]) -> list[dict[str, Any]]: + """Project candidate-rendered executable hashes into online read-back rows.""" + + dependencies = manifest["runtime_prompt_dependencies"] + if dependencies.get("publication_item_count") != 74 or dependencies.get("count") != 14: + raise ApplyPublicationError( + "manifest must distinguish 74 publication items from 14 runtime snapshots" + ) + results = [ + { + "kind": item["kind"], + "runtime_key": item["runtime_key"], + "version": item["version"], + "expected_content_sha256": item["content_sha256"], + "online_content_sha256": item["content_sha256"], + "status": "published_and_resolved", + "publication_dependency": item["publication_dependency"], + "publication_module_dependencies": item[ + "publication_module_dependencies" + ], + } + for item in dependencies["items"] + ] + if len(results) != 14 or len({item["runtime_key"] for item in results}) != 14: + raise ApplyPublicationError("runtime snapshot inventory must contain 14 identities") + return results + + +def validate_runtime_snapshot_receipt( + manifest: Mapping[str, Any], + manifest_raw_sha256: str, + approval: Mapping[str, Any], + approval_raw_sha256: str, + legal_review_raw_sha256: str, + online_service_public_key: Ed25519PublicKey, + receipt: Any, + *, + environment_id: str, + transaction_id: str, +) -> dict[str, Any]: + """Verify signed online evidence for 1 drafting plus 13 review identities.""" + + envelope = _require_exact( + receipt, + "runtime snapshot receipt", + ("schema_version", "receipt_id", "payload", "signature"), + ) + if envelope["schema_version"] != RUNTIME_RECEIPT_SCHEMA_VERSION: + raise ApplyPublicationError("runtime snapshot receipt schema is unsupported") + payload = _require_exact( + envelope["payload"], + "runtime snapshot payload", + ( + "schema_version", + "key_id", + "batch_manifest_raw_sha256", + "batch_manifest_payload_sha256", + "approval_receipt_sha256", + "legal_source_review_binding_sha256", + "environment_id", + "transaction_id", + "captured_at", + "generated_by", + "transaction_mode", + "publication_item_count", + "runtime_snapshot_dependencies", + "runtime_snapshot_dependency_count", + "status", + ), + ) + if ( + payload["schema_version"] != RUNTIME_PAYLOAD_SCHEMA_VERSION + or payload["key_id"] != batch.TRUSTED_ONLINE_APPLY_SERVICE_KEY_ID + or payload["status"] != "runtime_snapshots_verified" + or payload["batch_manifest_raw_sha256"] != manifest_raw_sha256 + or payload["batch_manifest_payload_sha256"] != manifest["manifest_payload_sha256"] + or payload["approval_receipt_sha256"] != approval_raw_sha256 + or payload["legal_source_review_binding_sha256"] != legal_review_raw_sha256 + or payload["environment_id"] != environment_id + or payload["transaction_id"] != transaction_id + or payload["transaction_mode"] != "single_service_atomic" + or payload["publication_item_count"] != 74 + or payload["runtime_snapshot_dependency_count"] != 14 + or payload["runtime_snapshot_dependencies"] + != expected_runtime_snapshot_results(manifest) + or payload["generated_by"] + != {"type": "system", "component": batch.TRUSTED_ONLINE_APPLY_SERVICE_COMPONENT} + ): + raise ApplyPublicationError("runtime snapshot receipt identity or 14-item evidence drift") + captured_at = batch._parse_time(payload["captured_at"], "runtime snapshot captured_at") + approval_payload = approval["payload"] + if captured_at < batch._parse_time( + approval_payload["approved_at"], "approval approved_at" + ) or captured_at >= batch._parse_time( + approval_payload["expires_at"], "approval expires_at" + ): + raise ApplyPublicationError("runtime snapshots were captured outside the approval window") + if _sha256(_public_raw(online_service_public_key)) != batch.TRUSTED_ONLINE_APPLY_SERVICE_PUBLIC_KEY_SHA256: + raise ApplyPublicationError("runtime snapshot signer does not match the frozen online key pin") + try: + signature = base64.b64decode( + _require_text(envelope["signature"], "runtime snapshot signature", maximum=1024), + validate=True, + ) + online_service_public_key.verify( + signature, + RUNTIME_SIGNATURE_DOMAIN + batch._canonical(payload), + ) + except (TypeError, ValueError, InvalidSignature) as exc: + raise ApplyPublicationError("runtime snapshot service signature verification failed") from exc + expected_id = "rc06-runtime-snapshot-" + _sha256( + batch._canonical(payload) + signature + )[:24] + if envelope["receipt_id"] != expected_id: + raise ApplyPublicationError("runtime snapshot receipt_id does not match signed material") + batch._assert_no_secret_material(envelope, "runtime snapshot receipt") + return envelope + + +class JsonHttpClient: + def __init__( + self, + base_url: str, + bearer_token: str, + *, + timeout_seconds: float = 30.0, + allow_http_loopback: bool = False, + ) -> None: + parsed = urllib.parse.urlsplit(base_url) + if parsed.username or parsed.password or parsed.query or parsed.fragment: + raise ApplyPublicationError("publication service URL contains forbidden components") + loopback = parsed.hostname in {"127.0.0.1", "localhost", "::1"} + if parsed.scheme != "https" and not ( + allow_http_loopback and parsed.scheme == "http" and loopback + ): + raise ApplyPublicationError("publication service requires HTTPS") + if not parsed.netloc or parsed.path not in {"", "/"}: + raise ApplyPublicationError("publication service base URL is invalid") + self.base_url = base_url.rstrip("/") + self.token = _require_text(bearer_token, "publication bearer token", maximum=8192) + self.timeout_seconds = timeout_seconds + + def request( + self, + method: str, + path: str, + *, + payload: Mapping[str, Any] | None, + headers: Mapping[str, str], + ) -> Any: + body = batch._canonical(payload) if payload is not None else None + request_headers = { + "Accept": "application/json", + "Authorization": f"Bearer {self.token}", + **dict(headers), + } + if body is not None: + request_headers["Content-Type"] = "application/json" + request = urllib.request.Request( + self.base_url + path, + data=body, + headers=request_headers, + method=method, + ) + try: + response = urllib.request.urlopen(request, timeout=self.timeout_seconds) + raw = response.read(MAX_HTTP_BODY_BYTES + 1) + status = response.status + content_type = response.headers.get_content_type() + except urllib.error.HTTPError as exc: + raw = exc.read(MAX_HTTP_BODY_BYTES + 1) + detail = f"HTTP {exc.code}" + try: + parsed = json.loads(raw.decode("utf-8")) + if isinstance(parsed, dict) and isinstance(parsed.get("error"), str): + detail += f": {parsed['error'][:512]}" + except (UnicodeDecodeError, json.JSONDecodeError): + pass + raise ApplyPublicationError(f"publication service rejected request ({detail})") from exc + except (OSError, TimeoutError) as exc: + raise ApplyPublicationError("publication service request failed") from exc + if status < 200 or status >= 300 or content_type != "application/json": + raise ApplyPublicationError("publication service returned an invalid HTTP response") + if len(raw) > MAX_HTTP_BODY_BYTES: + raise ApplyPublicationError("publication service response is too large") + try: + return json.loads( + raw.decode("utf-8"), + object_pairs_hook=batch._reject_duplicate_json_pairs, + parse_constant=batch._reject_json_constant, + ) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise ApplyPublicationError("publication service response is not valid JSON") from exc + + +class ApplyLedger: + """Small local recovery ledger; it stores hashes/receipts, never credentials.""" + + def __init__(self, path: Path) -> None: + self.path = path.resolve() + self.path.parent.mkdir(parents=True, exist_ok=True) + self.connection = sqlite3.connect(self.path, timeout=30, isolation_level=None) + self.connection.execute("PRAGMA journal_mode=WAL") + self.connection.execute("PRAGMA synchronous=FULL") + self.connection.executescript( + """ + CREATE TABLE IF NOT EXISTS rc06_apply_operations ( + idempotency_key TEXT PRIMARY KEY, + request_sha256 TEXT NOT NULL, + state TEXT NOT NULL, + apply_receipt_json TEXT, + apply_receipt_sha256 TEXT, + last_error TEXT, + updated_at TEXT NOT NULL + ); + CREATE TABLE IF NOT EXISTS rc06_apply_events ( + sequence INTEGER PRIMARY KEY AUTOINCREMENT, + idempotency_key TEXT NOT NULL, + state TEXT NOT NULL, + evidence_sha256 TEXT NOT NULL, + recorded_at TEXT NOT NULL + ); + """ + ) + + def reserve(self, key: str, request_sha256: str, now: datetime) -> dict[str, Any] | None: + self.connection.execute("BEGIN IMMEDIATE") + try: + row = self.connection.execute( + "SELECT request_sha256, state, apply_receipt_json FROM rc06_apply_operations " + "WHERE idempotency_key = ?", + (key,), + ).fetchone() + if row is not None and row[0] != request_sha256: + raise ApplyPublicationError("idempotency key is bound to another request") + if row is not None and row[1] == "succeeded": + receipt = json.loads(row[2]) + self.connection.execute("COMMIT") + return receipt + if row is not None and row[1] in {"dispatching", "verifying"}: + raise ApplyPublicationError("an apply with this idempotency key needs reconciliation") + timestamp = now.astimezone(UTC).isoformat(timespec="seconds") + self.connection.execute( + "INSERT INTO rc06_apply_operations " + "(idempotency_key, request_sha256, state, updated_at) VALUES (?, ?, 'dispatching', ?) " + "ON CONFLICT(idempotency_key) DO UPDATE SET state='dispatching', last_error=NULL, updated_at=excluded.updated_at", + (key, request_sha256, timestamp), + ) + self._event(key, "dispatching", request_sha256, timestamp) + self.connection.execute("COMMIT") + return None + except Exception: + self.connection.execute("ROLLBACK") + raise + + def state( + self, + key: str, + state: str, + evidence: Any, + now: datetime, + *, + receipt: Mapping[str, Any] | None = None, + error: str | None = None, + ) -> None: + evidence_sha = batch._sha256_json(evidence) + receipt_json = ( + batch._canonical(receipt).decode("utf-8") if receipt is not None else None + ) + receipt_sha = batch._sha256_json(receipt) if receipt is not None else None + timestamp = now.astimezone(UTC).isoformat(timespec="seconds") + self.connection.execute("BEGIN IMMEDIATE") + try: + self.connection.execute( + "UPDATE rc06_apply_operations SET state=?, apply_receipt_json=?, " + "apply_receipt_sha256=?, last_error=?, updated_at=? WHERE idempotency_key=?", + (state, receipt_json, receipt_sha, error, timestamp, key), + ) + self._event(key, state, evidence_sha, timestamp) + self.connection.execute("COMMIT") + except Exception: + self.connection.execute("ROLLBACK") + raise + + def _event(self, key: str, state: str, evidence_sha: str, timestamp: str) -> None: + self.connection.execute( + "INSERT INTO rc06_apply_events " + "(idempotency_key, state, evidence_sha256, recorded_at) VALUES (?, ?, ?, ?)", + (key, state, evidence_sha, timestamp), + ) + + def close(self) -> None: + self.connection.close() + + +def _idempotency_key( + manifest_raw_sha256: str, + approval_raw_sha256: str, + legal_review_raw_sha256: str, + environment_id: str, +) -> str: + identity = { + "operation": batch.APPLY_OPERATION, + "manifest_raw_sha256": manifest_raw_sha256, + "approval_raw_sha256": approval_raw_sha256, + "legal_review_raw_sha256": legal_review_raw_sha256, + "environment_id": environment_id, + } + return _sha256(batch._canonical(identity)) + + +def _transaction_id(idempotency_key: str) -> str: + return f"rc06-{idempotency_key}" + + +def _headers( + key: str, + manifest_raw_sha256: str, + approval_raw_sha256: str, + legal_review_raw_sha256: str, +) -> dict[str, str]: + return { + "Idempotency-Key": key, + "X-RC06-Manifest-SHA256": manifest_raw_sha256, + "X-RC06-Approval-SHA256": approval_raw_sha256, + "X-RC06-Legal-Review-SHA256": legal_review_raw_sha256, + } + + +def _validate_apply_response( + response: Any, + *, + key: str, + request_sha256: str, + legal_review_raw_sha256: str, +) -> dict[str, Any]: + row = _require_exact( + response, + "online apply response", + ( + "schema_version", + "idempotency_key", + "request_sha256", + "legal_source_review_binding_sha256", + "transaction_mode", + "publication_item_count", + "runtime_snapshot_dependency_count", + "apply_receipt", + "runtime_snapshot_receipt", + ), + ) + if ( + row["schema_version"] != APPLY_RESPONSE_SCHEMA_VERSION + or row["idempotency_key"] != key + or row["request_sha256"] != request_sha256 + or row["legal_source_review_binding_sha256"] != legal_review_raw_sha256 + or row["transaction_mode"] != "single_service_atomic" + or row["publication_item_count"] != 74 + or row["runtime_snapshot_dependency_count"] != 14 + ): + raise ApplyPublicationError("online apply response identity or transaction mode drift") + return row + + +def _validate_online_status( + status: Any, + *, + key: str, + transaction_id: str, + legal_review_raw_sha256: str, + apply_receipt: Mapping[str, Any], + runtime_snapshot_receipt: Mapping[str, Any], + expected_publication_results: list[dict[str, Any]], + expected_runtime_results: list[dict[str, Any]], +) -> dict[str, Any]: + row = _require_exact( + status, + "online apply status", + ( + "schema_version", + "idempotency_key", + "transaction_id", + "state", + "legal_source_review_binding_sha256", + "publication_item_results", + "publication_item_count", + "runtime_snapshot_dependencies", + "runtime_snapshot_dependency_count", + "apply_receipt_sha256", + "runtime_snapshot_receipt_sha256", + ), + ) + if ( + row["schema_version"] != APPLY_STATUS_SCHEMA_VERSION + or row["idempotency_key"] != key + or row["transaction_id"] != transaction_id + or row["state"] != "committed_and_verified" + or row["legal_source_review_binding_sha256"] != legal_review_raw_sha256 + or row["publication_item_results"] != expected_publication_results + or row["publication_item_count"] != 74 + or row["runtime_snapshot_dependencies"] != expected_runtime_results + or row["runtime_snapshot_dependency_count"] != 14 + or row["apply_receipt_sha256"] != batch._sha256_json(apply_receipt) + or row["runtime_snapshot_receipt_sha256"] + != batch._sha256_json(runtime_snapshot_receipt) + ): + raise ApplyPublicationError("online read-after-write hash verification failed") + return row + + +def _compensate( + client: JsonHttpClient, + *, + key: str, + transaction_id: str, + headers: Mapping[str, str], + manifest_raw_sha256: str, + approval_raw_sha256: str, + legal_review_raw_sha256: str, +) -> dict[str, Any]: + request = { + "schema_version": ROLLBACK_REQUEST_SCHEMA_VERSION, + "idempotency_key": key, + "transaction_id": transaction_id, + "batch_manifest_raw_sha256": manifest_raw_sha256, + "approval_receipt_sha256": approval_raw_sha256, + "legal_source_review_binding_sha256": legal_review_raw_sha256, + } + response = client.request( + "POST", + f"/internal/rc06/publication-transactions/{transaction_id}/rollback", + payload=request, + headers=headers, + ) + row = _require_exact( + response, + "online rollback response", + ( + "schema_version", + "idempotency_key", + "transaction_id", + "state", + "restored_item_count", + "rollback_evidence_sha256", + ), + ) + if ( + row["schema_version"] != ROLLBACK_RESPONSE_SCHEMA_VERSION + or row["idempotency_key"] != key + or row["transaction_id"] != transaction_id + or row["state"] != "rolled_back" + or row["restored_item_count"] != 0 + or not batch.HEX64_RE.fullmatch(str(row["rollback_evidence_sha256"])) + ): + raise ApplyPublicationError("online rollback evidence is invalid") + return row + + +def apply_publication_batch( + *, + manifest: Mapping[str, Any], + manifest_raw_sha256: str, + approval: Mapping[str, Any], + approval_raw_sha256: str, + approval_keyring: Mapping[str, Any], + approval_root_public_key: Ed25519PublicKey, + apply_service_public_key: Ed25519PublicKey, + online_service_public_key: Ed25519PublicKey, + legal_review_request: Mapping[str, Any], + legal_review_decision: Mapping[str, Any], + legal_review_keyring: Mapping[str, Any], + environment_id: str, + client: JsonHttpClient, + ledger: ApplyLedger, + now: datetime, +) -> dict[str, Any]: + """Perform one authenticated, idempotent and recoverable online apply.""" + + batch.validate_manifest(manifest) + if manifest["runtime_prompt_dependencies"]["items"] != batch._load_executable_prompt_dependencies( + REPO_ROOT + ): + raise ApplyPublicationError( + "runtime Prompt name/version/content hashes differ from candidate executable source" + ) + 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") + environment_id = _require_text(environment_id, "environment_id", maximum=256) + batch.validate_approval_receipt( + manifest, + manifest_raw_sha256, + approval, + approval_keyring, + approval_root_public_key, + ) + _, legal_review_binding_sha256 = validate_scene_legal_review( + manifest, + approval, + legal_review_request, + legal_review_decision, + legal_review_keyring, + now=now, + ) + keys = { + _public_raw(approval_root_public_key), + _public_raw(apply_service_public_key), + _public_raw(online_service_public_key), + } + if len(keys) != 3: + raise ApplyPublicationError("approval, apply and online service keys must be distinct") + key = _idempotency_key( + manifest_raw_sha256, + approval_raw_sha256, + legal_review_binding_sha256, + environment_id, + ) + transaction_id = _transaction_id(key) + request_payload = { + "schema_version": APPLY_REQUEST_SCHEMA_VERSION, + "idempotency_key": key, + "transaction_id": transaction_id, + "environment_id": environment_id, + "batch_manifest_raw_sha256": manifest_raw_sha256, + "batch_manifest_payload_sha256": manifest["manifest_payload_sha256"], + "approval_receipt_sha256": approval_raw_sha256, + "legal_source_review_binding_sha256": legal_review_binding_sha256, + "manifest": manifest, + "approval_receipt": approval, + "legal_source_review_request": legal_review_request, + "legal_source_review_decision": legal_review_decision, + "legal_source_review_keyring": legal_review_keyring, + "expected_publication_item_results": batch._expected_apply_item_results(manifest), + "expected_publication_item_count": 74, + "expected_runtime_snapshot_dependencies": expected_runtime_snapshot_results(manifest), + "expected_runtime_snapshot_dependency_count": 14, + } + request_sha256 = batch._sha256_json(request_payload) + replay = ledger.reserve(key, request_sha256, now) + if replay is not None: + replay = _require_exact( + replay, + "stored online apply evidence", + ( + "schema_version", + "publication_item_count", + "runtime_snapshot_dependency_count", + "apply_receipt", + "runtime_snapshot_receipt", + ), + ) + if ( + replay["schema_version"] != EXECUTION_EVIDENCE_SCHEMA_VERSION + or replay["publication_item_count"] != 74 + or replay["runtime_snapshot_dependency_count"] != 14 + ): + raise ApplyPublicationError("stored online apply evidence count drift") + apply_receipt = replay["apply_receipt"] + batch.validate_apply_receipt( + manifest, + manifest_raw_sha256, + approval, + approval_keyring, + approval_root_public_key, + apply_service_public_key, + online_service_public_key, + approval_raw_sha256, + apply_receipt, + ) + validate_runtime_snapshot_receipt( + manifest, + manifest_raw_sha256, + approval, + approval_raw_sha256, + legal_review_binding_sha256, + online_service_public_key, + replay["runtime_snapshot_receipt"], + environment_id=environment_id, + transaction_id=transaction_id, + ) + if apply_receipt["payload"]["online_transaction_receipt"]["payload"]["transaction_id"] != transaction_id: + raise ApplyPublicationError("stored apply receipt idempotency binding drift") + return replay + headers = _headers( + key, manifest_raw_sha256, approval_raw_sha256, legal_review_binding_sha256 + ) + dispatch_started = False + try: + dispatch_started = True + response = client.request( + "POST", + "/internal/rc06/publication-transactions/apply", + payload=request_payload, + headers={**headers, "X-RC06-Request-SHA256": request_sha256}, + ) + response = _validate_apply_response( + response, + key=key, + request_sha256=request_sha256, + legal_review_raw_sha256=legal_review_binding_sha256, + ) + apply_receipt = response["apply_receipt"] + runtime_receipt = response["runtime_snapshot_receipt"] + batch.validate_apply_receipt( + manifest, + manifest_raw_sha256, + approval, + approval_keyring, + approval_root_public_key, + apply_service_public_key, + online_service_public_key, + approval_raw_sha256, + apply_receipt, + ) + validate_runtime_snapshot_receipt( + manifest, + manifest_raw_sha256, + approval, + approval_raw_sha256, + legal_review_binding_sha256, + online_service_public_key, + runtime_receipt, + environment_id=environment_id, + transaction_id=transaction_id, + ) + online_payload = apply_receipt["payload"]["online_transaction_receipt"]["payload"] + if online_payload["transaction_id"] != transaction_id: + raise ApplyPublicationError("signed apply receipt is not bound to the idempotency identity") + ledger.state(key, "verifying", response, now) + status = client.request( + "GET", + f"/internal/rc06/publication-transactions/{transaction_id}/status", + payload=None, + headers=headers, + ) + status = _validate_online_status( + status, + key=key, + transaction_id=transaction_id, + legal_review_raw_sha256=legal_review_binding_sha256, + apply_receipt=apply_receipt, + runtime_snapshot_receipt=runtime_receipt, + expected_publication_results=batch._expected_apply_item_results(manifest), + expected_runtime_results=expected_runtime_snapshot_results(manifest), + ) + evidence = { + "schema_version": EXECUTION_EVIDENCE_SCHEMA_VERSION, + "publication_item_count": 74, + "runtime_snapshot_dependency_count": 14, + "apply_receipt": apply_receipt, + "runtime_snapshot_receipt": runtime_receipt, + } + ledger.state(key, "succeeded", status, now, receipt=evidence) + return evidence + except Exception as exc: + if not isinstance(exc, ApplyPublicationError): + failure = ApplyPublicationError("unexpected online apply failure") + failure.__cause__ = exc + else: + failure = exc + if dispatch_started: + try: + rollback = _compensate( + client, + key=key, + transaction_id=transaction_id, + headers=headers, + manifest_raw_sha256=manifest_raw_sha256, + approval_raw_sha256=approval_raw_sha256, + legal_review_raw_sha256=legal_review_binding_sha256, + ) + ledger.state(key, "rolled_back", rollback, now, error=str(failure)) + except Exception as rollback_exc: + ledger.state( + key, + "compensation_failed", + {"apply_error": str(failure), "rollback_error": str(rollback_exc)}, + now, + error=f"{failure}; compensation failed: {rollback_exc}", + ) + raise ApplyPublicationError( + "online apply failed and compensation could not be proven" + ) from rollback_exc + else: + ledger.state(key, "failed", {"error": str(failure)}, now, error=str(failure)) + raise failure + + +def _read_public_key(path: Path, expected_sha256: str, label: str) -> Ed25519PublicKey: + return batch._read_pinned_raw_public_key( + path, + expected_sha256=expected_sha256, + label=label, + ) + + +def _outside_repo(path: Path, label: str) -> Path: + resolved = path.resolve() + try: + resolved.relative_to(REPO_ROOT) + except ValueError: + return resolved + raise ApplyPublicationError(f"{label} must be stored outside the source worktree") + + +def _write_receipt(path: Path, receipt: Mapping[str, Any]) -> str: + raw = batch._canonical(receipt) + b"\n" + path.parent.mkdir(parents=True, exist_ok=True) + if path.exists(): + existing = path.read_bytes() + if existing != raw: + raise ApplyPublicationError("refusing to overwrite different online apply evidence") + return _sha256(existing) + temporary = path.with_name(path.name + f".tmp-{os.getpid()}") + try: + with temporary.open("xb") as handle: + handle.write(raw) + handle.flush() + os.fsync(handle.fileno()) + os.replace(temporary, path) + try: + path.chmod(0o600) + except OSError: + pass + finally: + if temporary.exists(): + temporary.unlink() + return _sha256(raw) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--manifest", type=Path, required=True) + parser.add_argument("--manifest-sha256", required=True) + parser.add_argument("--approval", type=Path, required=True) + parser.add_argument("--approval-sha256", required=True) + parser.add_argument("--approval-keyring", type=Path, required=True) + parser.add_argument("--approval-keyring-sha256", required=True) + parser.add_argument("--approval-root-public-key", type=Path, required=True) + parser.add_argument("--apply-service-public-key", type=Path, required=True) + parser.add_argument("--online-service-public-key", type=Path, required=True) + parser.add_argument("--legal-review-request", type=Path, required=True) + parser.add_argument("--legal-review-request-sha256", required=True) + parser.add_argument("--legal-review-decision", type=Path, required=True) + parser.add_argument("--legal-review-decision-sha256", required=True) + parser.add_argument("--legal-review-keyring", type=Path, required=True) + parser.add_argument("--legal-review-keyring-sha256", required=True) + parser.add_argument("--environment-id", required=True) + parser.add_argument("--service-url", required=True) + parser.add_argument("--ledger", type=Path, required=True) + parser.add_argument("--output", type=Path, required=True) + args = parser.parse_args(argv) + ledger: ApplyLedger | None = None + try: + token = os.environ.get(AUTH_TOKEN_ENV, "") + if not token: + raise ApplyPublicationError(f"{AUTH_TOKEN_ENV} is not set") + manifest, manifest_raw = batch.secure_read_json( + args.manifest, "publication manifest", args.manifest_sha256 + ) + approval, approval_raw = batch.secure_read_json( + args.approval, "approval receipt", args.approval_sha256 + ) + keyring, _ = batch.secure_read_json( + args.approval_keyring, + "approval keyring", + args.approval_keyring_sha256, + ) + legal_request, _ = batch.secure_read_json( + args.legal_review_request, + "scene legal review request", + args.legal_review_request_sha256, + ) + legal_decision, _ = batch.secure_read_json( + args.legal_review_decision, + "scene legal review decision", + args.legal_review_decision_sha256, + ) + legal_keyring, _ = batch.secure_read_json( + args.legal_review_keyring, + "scene legal review keyring", + args.legal_review_keyring_sha256, + ) + root_key = _read_public_key( + args.approval_root_public_key, + batch.TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256, + "approval root public key", + ) + apply_key = _read_public_key( + args.apply_service_public_key, + batch.TRUSTED_APPLY_SERVICE_PUBLIC_KEY_SHA256, + "apply service public key", + ) + online_key = _read_public_key( + args.online_service_public_key, + batch.TRUSTED_ONLINE_APPLY_SERVICE_PUBLIC_KEY_SHA256, + "online service public key", + ) + ledger_path = _outside_repo(args.ledger, "apply ledger") + output_path = _outside_repo(args.output, "apply receipt") + ledger = ApplyLedger(ledger_path) + receipt = apply_publication_batch( + manifest=manifest, + manifest_raw_sha256=manifest_raw, + approval=approval, + approval_raw_sha256=approval_raw, + approval_keyring=keyring, + approval_root_public_key=root_key, + apply_service_public_key=apply_key, + online_service_public_key=online_key, + legal_review_request=legal_request, + legal_review_decision=legal_decision, + legal_review_keyring=legal_keyring, + environment_id=args.environment_id, + client=JsonHttpClient(args.service_url, token), + ledger=ledger, + now=datetime.now(UTC), + ) + receipt_sha = _write_receipt(output_path, receipt) + print( + json.dumps( + { + "status": "applied_and_verified", + "apply_evidence": str(output_path), + "apply_evidence_raw_sha256": receipt_sha, + "publication_item_count": 74, + "runtime_snapshot_dependency_count": 14, + }, + ensure_ascii=False, + sort_keys=True, + ) + ) + return 0 + except (batch.PublicationBatchError, OSError, sqlite3.Error) as exc: + print(f"rc06_apply_error: {exc}", file=sys.stderr) + return 2 + finally: + if ledger is not None: + ledger.close() + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/yuqei-infra/scripts/build-rc06-publication-batch.py b/yuqei-infra/scripts/build-rc06-publication-batch.py index 0b6263fc..cc1dec5c 100644 --- a/yuqei-infra/scripts/build-rc06-publication-batch.py +++ b/yuqei-infra/scripts/build-rc06-publication-batch.py @@ -19,6 +19,7 @@ import stat import subprocess import sys import tempfile +import types from datetime import datetime, timedelta from pathlib import Path, PurePath from typing import Any, Iterable, Mapping @@ -28,7 +29,7 @@ from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey -SCHEMA_VERSION = "yuqei.rc06-publication-batch.v3" +SCHEMA_VERSION = "yuqei.rc06-publication-batch.v4" DISPOSITION_SCHEMA_VERSION = "yuqei.rc06-golden-dispositions.v2" IMAGE_SCHEMA_VERSION = "yuqei.rc06-image-identities.v1" APPROVAL_SCHEMA_VERSION = "yuqei.rc06-publication-batch-approval.v5" @@ -74,7 +75,7 @@ RC05_CAMPAIGN_SCHEMA_VERSION = "multiscene-call-campaign.v1" BLOCKED_CONTROLS = ( "single_reviewer_batch_approval_receipt", "system_generated_item_apply_receipts_and_online_snapshots", - "runtime_executable_prompt_snapshots_1_drafting_plus_8_review", + "runtime_executable_prompt_snapshots_1_drafting_plus_13_review", ) # Only the stable key owner identity is frozen in source. A title, organization, # and authorization scope are human assertions for one exact approval and must @@ -106,6 +107,7 @@ EXPECTED_DRAFTING_MODULE_COUNT = 6 EXPECTED_REVIEW_MODULE_COUNT = 8 EXPECTED_MODULE_COUNT = EXPECTED_DRAFTING_MODULE_COUNT + EXPECTED_REVIEW_MODULE_COUNT EXPECTED_DEFAULT_PROMPT_COUNT = 46 +EXPECTED_RUNTIME_PROMPT_COUNT = 14 EXPECTED_GOLDEN_COUNT = 31 EXPECTED_GOLDEN_GENERATION_COUNT = 23 EXPECTED_GOLDEN_CLAUSE_COUNT = 8 @@ -149,6 +151,21 @@ DEFAULT_PROMPT_PATH = ( "yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/" "prompt_assets/v1_phase15/default-prompt-assets.json" ) +RUNTIME_EXISTING_DOMAIN_PROMPTS = ( + "contract_review_channel_franchise", + "contract_review_employment_hr", + "contract_review_investment_finance", + "contract_review_real_estate_property", + "contract_review_tech_saas_data_compliance", +) +DRAFTING_RUNTIME_MODULE_ORDER = ( + "draft_intake", + "template_match", + "missing_facts_questions", + "clause_assembly_plan", + "draft_consistency_check", + "candidate_revision_explanation", +) GOLDEN_INVENTORY_PATH = ( "yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/" "golden_assets/v1_phase13_legal_knowledge_template_library/inventories/" @@ -156,6 +173,8 @@ GOLDEN_INVENTORY_PATH = ( ) FROZEN_IMPLEMENTATION_PATHS = ( "yuqei-infra/scripts/build-rc06-publication-batch.py", + "yuqei-infra/scripts/apply-rc06-publication-batch.py", + "yuqei-infra/scripts/build-scene-source-legal-review.py", "yuqei-infra/scripts/rc06-single-reviewer-signing.py", "yuqei-infra/scripts/rc06-release-trust-bootstrap.py", "yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/contract_templates.py", @@ -172,6 +191,9 @@ CONTRACT_RUNTIME_PACKAGE_ROOT = ( "yuqei-contract-platform/services/contract-api/src/yuqei_contract_api" ) SDK_RUNTIME_PACKAGE_ROOT = "yuqei-ai-sdk-python/src/yuqei_sdk" +AI_RUNTIME_PACKAGE_ROOT = ( + "yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api" +) DISPOSITION_STATUSES = { "named_legal_approval", @@ -493,7 +515,11 @@ def _inventory_source_paths(repo: Path, commit: str | None = None) -> list[str]: raise PublicationBatchError(f"template source must not be a symbolic link: {path}") if path.is_file() and path.suffix.lower() in {".json", ".md"}: paths.append(_relative_source_path(repo, path)) - for package_root in (CONTRACT_RUNTIME_PACKAGE_ROOT, SDK_RUNTIME_PACKAGE_ROOT): + for package_root in ( + CONTRACT_RUNTIME_PACKAGE_ROOT, + SDK_RUNTIME_PACKAGE_ROOT, + AI_RUNTIME_PACKAGE_ROOT, + ): root = repo / package_root _assert_no_symlink_components(root, f"runtime package {package_root}") paths.extend( @@ -513,6 +539,7 @@ def _inventory_source_paths(repo: Path, commit: str | None = None) -> list[str]: ASSET_ROOT, CONTRACT_RUNTIME_PACKAGE_ROOT, SDK_RUNTIME_PACKAGE_ROOT, + AI_RUNTIME_PACKAGE_ROOT, ) ) for relative in listing.splitlines(): @@ -523,6 +550,7 @@ def _inventory_source_paths(repo: Path, commit: str | None = None) -> list[str]: if ( relative.startswith(f"{CONTRACT_RUNTIME_PACKAGE_ROOT}/") or relative.startswith(f"{SDK_RUNTIME_PACKAGE_ROOT}/") + or relative.startswith(f"{AI_RUNTIME_PACKAGE_ROOT}/") ): paths.append(relative) paths.extend(FROZEN_IMPLEMENTATION_PATHS) @@ -713,6 +741,130 @@ def _load_inventory_symbols(repo: Path) -> tuple[Any, Any, Any, Any, Any]: pass +def _load_executable_prompt_dependencies(repo: Path) -> list[dict[str, Any]]: + """Render the candidate's actual runtime Prompt content for 14 identities.""" + + paths = ( + repo / "yuqei-ai-platform/services/ai-platform-api/src", + repo / "yuqei-ai-sdk-python/src", + ) + inserted: list[str] = [] + for path in reversed(paths): + value = str(path) + if value not in sys.path: + sys.path.insert(0, value) + inserted.append(value) + prefixes = ("yuqei_ai_platform_api", "yuqei_sdk") + previous = { + name: module + for name, module in list(sys.modules.items()) + if any(name == prefix or name.startswith(f"{prefix}.") for prefix in prefixes) + } + for name in previous: + sys.modules.pop(name, None) + try: + from yuqei_ai_platform_api import prompts as prompt_routes # type: ignore + from yuqei_ai_platform_api.contract_drafting import ( # type: ignore + DRAFT_PROMPT_CONTENT, + DRAFT_PROMPT_NAMES, + DRAFT_PROMPT_VERSION, + ) + from yuqei_ai_platform_api.drafting_prompt_modules import ( # type: ignore + drafting_module_evidence, + ) + from yuqei_ai_platform_api.packaged_prompt_assets_import import ( # type: ignore + DEFAULT_VERSION, + load_packaged_prompt_assets, + ) + from yuqei_ai_platform_api.review_scene_prompts import ( # type: ignore + review_module_evidence, + scene_review_template, + ) + + drafting_evidence = drafting_module_evidence(DRAFT_PROMPT_CONTENT) + results: list[dict[str, Any]] = [ + { + "kind": "composed_drafting", + "runtime_key": DRAFT_PROMPT_NAMES[0], + "version": DRAFT_PROMPT_VERSION, + "content_sha256": _sha256_bytes(DRAFT_PROMPT_CONTENT.encode("utf-8")), + "publication_dependency": "six_module_composition", + "publication_module_dependencies": drafting_evidence, + } + ] + review_routes = sorted( + item["key"] + for item in json.loads((repo / REVIEW_MODULE_PATH).read_text(encoding="utf-8"))[ + "modules" + ] + ) + original_candidates = prompt_routes.contract_review_prompt_candidates + try: + for route in review_routes: + prompt_routes.contract_review_prompt_candidates = ( + lambda *_args, _route=route, **_kwargs: [_route] + ) + request = types.SimpleNamespace( + contract_type="rc06_runtime_snapshot", + scene_template_id="rc06-runtime-snapshot", + review_profile="scene_evaluation_v4", + ) + template = scene_review_template(request) + module_evidence = review_module_evidence(template) + if len(module_evidence) != 1 or module_evidence[0]["key"] != route: + raise PublicationBatchError( + f"runtime review Prompt does not contain its exact module: {route}" + ) + results.append( + { + "kind": "new_review_module", + "runtime_key": route, + "version": template.version, + "content_sha256": template.content_hash, + "publication_dependency": "published_module", + "publication_module_dependencies": module_evidence, + } + ) + finally: + prompt_routes.contract_review_prompt_candidates = original_candidates + + packaged, _, duplicates = load_packaged_prompt_assets(repo / DEFAULT_PROMPT_PATH) + if duplicates: + raise PublicationBatchError("default Prompt runtime inventory contains duplicates") + by_key = {item.prompt_key: item for item in packaged} + for key in RUNTIME_EXISTING_DOMAIN_PROMPTS: + if key not in by_key: + raise PublicationBatchError(f"existing runtime domain Prompt is absent: {key}") + results.append( + { + "kind": "existing_domain_prompt", + "runtime_key": key, + "version": DEFAULT_VERSION, + "content_sha256": _sha256_bytes(by_key[key].content.encode("utf-8")), + "publication_dependency": "preexisting_domain_prompt", + "publication_module_dependencies": [], + } + ) + results.sort(key=lambda item: (item["kind"], item["runtime_key"])) + if len(results) != EXPECTED_RUNTIME_PROMPT_COUNT or len( + {item["runtime_key"] for item in results} + ) != EXPECTED_RUNTIME_PROMPT_COUNT: + raise PublicationBatchError("runtime Prompt dependency inventory must contain 14 identities") + return results + except (ImportError, OSError, ValueError, KeyError, TypeError) as exc: + raise PublicationBatchError(f"cannot render executable runtime Prompts: {exc}") from exc + finally: + for name in list(sys.modules): + if any(name == prefix or name.startswith(f"{prefix}.") for prefix in prefixes): + sys.modules.pop(name, None) + sys.modules.update(previous) + for value in inserted: + try: + sys.path.remove(value) + except ValueError: # pragma: no cover + pass + + def _asset_hash_projection(asset: Mapping[str, Any]) -> dict[str, Any]: return { "id": asset["template_id"], @@ -911,6 +1063,7 @@ def load_runtime_inventory( return { "assets": assets, "prompt_modules": modules, + "runtime_prompt_dependencies": _load_executable_prompt_dependencies(repo), "default_prompt_boundary": { "included_in_publication_batch": False, "count": len(default_inventory), @@ -2103,6 +2256,12 @@ def assemble_manifest( "items": inventory["prompt_modules"], "inventory_sha256": _sha256_json(inventory["prompt_modules"]), }, + "runtime_prompt_dependencies": { + "publication_item_count": EXPECTED_ASSET_COUNT + EXPECTED_MODULE_COUNT, + "count": len(inventory["runtime_prompt_dependencies"]), + "items": inventory["runtime_prompt_dependencies"], + "inventory_sha256": _sha256_json(inventory["runtime_prompt_dependencies"]), + }, "default_prompt_asset_boundary": inventory["default_prompt_boundary"], "golden_dispositions": { "count": len(dispositions), @@ -2136,7 +2295,7 @@ def assemble_manifest( "No reviewer approval or apply evidence is embedded in this offline manifest.", "One signed batch approval must bind all 60 assets, 14 Prompt modules, and 31 dispositions.", "The apply service must emit 74 machine-verifiable item results and online hashes.", - "One drafting plus eight review executable runtime Prompt snapshots are not implemented and remain blocking.", + "One composed drafting plus thirteen review runtime Prompt snapshots are not implemented and remain blocking.", ], } _assert_no_secret_material(manifest) @@ -2209,7 +2368,8 @@ def validate_manifest(payload: Any) -> dict[str, Any]: "publication batch", ( "schema_version", "candidate", "asset_batch", "prompt_module_batch", - "default_prompt_asset_boundary", "golden_dispositions", "publication_control", + "runtime_prompt_dependencies", "default_prompt_asset_boundary", + "golden_dispositions", "publication_control", "side_effects", "claims_not_made", "manifest_payload_sha256", ), ) @@ -2336,6 +2496,90 @@ def validate_manifest(payload: Any) -> dict[str, Any]: if _sha256_json(modules["items"]) != modules["inventory_sha256"]: raise PublicationBatchError("Prompt module inventory hash drift") + runtime = _require_keys( + manifest["runtime_prompt_dependencies"], + "runtime Prompt dependencies", + ("publication_item_count", "count", "items", "inventory_sha256"), + ) + if ( + runtime["publication_item_count"] != EXPECTED_ASSET_COUNT + EXPECTED_MODULE_COUNT + or runtime["count"] != EXPECTED_RUNTIME_PROMPT_COUNT + or not isinstance(runtime["items"], list) + or len(runtime["items"]) != EXPECTED_RUNTIME_PROMPT_COUNT + ): + raise PublicationBatchError( + "runtime Prompt dependencies must distinguish 74 publication items from 14 snapshots" + ) + runtime_keys: set[str] = set() + runtime_kinds: list[str] = [] + runtime_keys_by_kind: dict[str, set[str]] = {} + publication_modules = { + item["key"]: { + "key": item["key"], + "version": item["version"], + "content_sha256": item["content_sha256"], + } + for item in modules["items"] + } + for item in runtime["items"]: + row = _require_keys( + item, + "runtime Prompt dependency", + ( + "kind", + "runtime_key", + "version", + "content_sha256", + "publication_dependency", + "publication_module_dependencies", + ), + ) + key = _require_string(row["runtime_key"], "runtime Prompt key", max_length=128) + _require_string(row["version"], f"runtime Prompt {key} version", max_length=128) + _require_sha(row["content_sha256"], f"runtime Prompt {key} content SHA-256") + if key in runtime_keys: + raise PublicationBatchError("runtime Prompt dependency keys must be unique") + runtime_keys.add(key) + runtime_kinds.append(row["kind"]) + runtime_keys_by_kind.setdefault(row["kind"], set()).add(key) + expected_dependency = { + "composed_drafting": "six_module_composition", + "new_review_module": "published_module", + "existing_domain_prompt": "preexisting_domain_prompt", + }.get(row["kind"]) + if row["publication_dependency"] != expected_dependency: + raise PublicationBatchError(f"runtime Prompt dependency kind drift: {key}") + dependency_rows = row["publication_module_dependencies"] + if not isinstance(dependency_rows, list): + raise PublicationBatchError( + f"runtime Prompt module dependencies must be a list: {key}" + ) + if row["kind"] == "composed_drafting": + expected_module_keys = list(DRAFTING_RUNTIME_MODULE_ORDER) + elif row["kind"] == "new_review_module": + expected_module_keys = [key] + else: + expected_module_keys = [] + expected_dependency_rows = [ + publication_modules[module_key] for module_key in expected_module_keys + ] + if dependency_rows != expected_dependency_rows: + raise PublicationBatchError( + f"runtime Prompt is not bound to exact publication module identities: {key}" + ) + if ( + runtime_kinds.count("composed_drafting") != 1 + or runtime_kinds.count("new_review_module") != 8 + or runtime_kinds.count("existing_domain_prompt") != 5 + or runtime_keys_by_kind.get("composed_drafting") != {"contract_drafting_v1"} + or runtime_keys_by_kind.get("new_review_module") != set(review_routes) + or runtime_keys_by_kind.get("existing_domain_prompt") + != set(RUNTIME_EXISTING_DOMAIN_PROMPTS) + ): + raise PublicationBatchError("runtime Prompt dependency 1+8+5 inventory drift") + if _sha256_json(runtime["items"]) != runtime["inventory_sha256"]: + raise PublicationBatchError("runtime Prompt dependency inventory hash drift") + boundary = manifest["default_prompt_asset_boundary"] required_boundary = { "included_in_publication_batch", "count", "source_path", "source_raw_sha256", diff --git a/yuqei-infra/tests/test_rc06_publication_apply.py b/yuqei-infra/tests/test_rc06_publication_apply.py new file mode 100644 index 00000000..6a80143a --- /dev/null +++ b/yuqei-infra/tests/test_rc06_publication_apply.py @@ -0,0 +1,844 @@ +from __future__ import annotations + +import base64 +import copy +import importlib.util +import json +import sqlite3 +import sys +import threading +from contextlib import contextmanager +from datetime import UTC, datetime, timedelta +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path + +import pytest +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + + +REPO_ROOT = Path(__file__).resolve().parents[2] +APPLY_SCRIPT = REPO_ROOT / "yuqei-infra" / "scripts" / "apply-rc06-publication-batch.py" +FIXTURE_SCRIPT = REPO_ROOT / "yuqei-infra" / "tests" / "test_rc06_publication_batch.py" + + +def _load(path: Path, name: str): + spec = importlib.util.spec_from_file_location(name, path) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + sys.modules[name] = module + spec.loader.exec_module(module) + return module + + +apply = _load(APPLY_SCRIPT, "apply_rc06_publication_batch") +fixtures = _load(FIXTURE_SCRIPT, "rc06_publication_batch_test_fixtures") + + +def _raw_public(key) -> bytes: + return key.public_bytes(serialization.Encoding.Raw, serialization.PublicFormat.Raw) + + +def _sign_envelope(private_key, *, schema: str, id_prefix: str, domain: bytes, payload: dict) -> dict: + signature = private_key.sign(domain + apply.batch._canonical(payload)) + return { + "schema_version": schema, + "receipt_id": id_prefix + + apply.batch._sha256_bytes(apply.batch._canonical(payload) + signature)[:24], + "payload": payload, + "signature": base64.b64encode(signature).decode("ascii"), + } + + +class _FakeCanonicalLegalReviewValidator: + """Test double for build-scene-source-legal-review.py; never production evidence.""" + + @staticmethod + def verify_request(request: dict, _repo: Path) -> dict: + projection = {key: value for key, value in request.items() if key != "request_sha256"} + if request.get("request_sha256") != apply.batch._sha256_json(projection): + raise ValueError("review request embedded hash drift") + if len(request.get("occurrences", [])) != 43: + raise ValueError("review request must contain 43 occurrences") + return request + + @staticmethod + def verify_signed_decision( + request: dict, decision: dict, _keyring: dict, *, checked_at: str + ) -> dict: + if decision.get("request_sha256") != request["request_sha256"]: + raise ValueError("decision request binding drift") + now = datetime.fromisoformat(checked_at) + expires_at = datetime.fromisoformat(decision["expires_at"]) + if now > expires_at: + raise ValueError("decision expired") + expected = { + item["occurrence_id"]: item["source_record_sha256"] + for item in request["occurrences"] + } + items = decision.get("items") + if not isinstance(items, list) or len(items) != 43: + raise ValueError("decision must contain 43 items") + for item in items: + if item.get("source_record_sha256") != expected.get(item.get("occurrence_id")): + raise ValueError("source record hash mismatch") + if ( + item.get("current_effectiveness_verified") is not True + or item.get("transaction_applicability_verified") is not True + or len(item.get("rationale", "")) < 20 + or not item.get("authoritative_status_url", "").startswith("https://") + ): + raise ValueError("legal source decision is incomplete") + return { + "status": "verified_qualified_human_review", + "candidate_commit": request["candidate_commit"], + "request_sha256": request["request_sha256"], + "reviewer_name": decision["reviewer"]["name"], + "reviewed_occurrence_count": 43, + "all_current_effectiveness_verified": True, + "all_transaction_applicability_verified": True, + "expires_at": decision["expires_at"], + "authorizes_publication_by_itself": False, + } + + +def _material(monkeypatch: pytest.MonkeyPatch, now: datetime) -> dict: + inventory = apply.batch.load_runtime_inventory(REPO_ROOT) + manifest = fixtures._manifest(inventory, resolved=True) + manifest_raw = apply.batch._sha256_bytes(apply.batch._canonical(manifest)) + root_private = Ed25519PrivateKey.generate() + reviewer_private = Ed25519PrivateKey.generate() + apply_private = Ed25519PrivateKey.generate() + online_private = Ed25519PrivateKey.generate() + monkeypatch.setattr( + apply.batch, + "TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256", + apply.batch._sha256_bytes(_raw_public(root_private.public_key())), + ) + monkeypatch.setattr(apply.batch, "FROZEN_APPROVAL_KEYRING_EPOCH", 1) + monkeypatch.setattr( + apply.batch, + "TRUSTED_APPLY_SERVICE_PUBLIC_KEY_SHA256", + apply.batch._sha256_bytes(_raw_public(apply_private.public_key())), + ) + monkeypatch.setattr( + apply.batch, + "TRUSTED_ONLINE_APPLY_SERVICE_PUBLIC_KEY_SHA256", + apply.batch._sha256_bytes(_raw_public(online_private.public_key())), + ) + monkeypatch.setattr( + apply.batch, + "TRUSTED_ONLINE_APPLY_ENVIRONMENT_IDS", + ("rc06-test",), + ) + monkeypatch.setattr( + apply, + "_load_scene_legal_review_module", + lambda: _FakeCanonicalLegalReviewValidator, + ) + reviewer_pem = reviewer_private.public_key().public_bytes( + serialization.Encoding.PEM, + serialization.PublicFormat.SubjectPublicKeyInfo, + ).decode("ascii") + keyring_unsigned = { + "schema_version": apply.batch.APPROVAL_KEYRING_SCHEMA_VERSION, + "root_key_id": apply.batch.TRUSTED_APPROVAL_ROOT_KEY_ID, + "epoch": 1, + "issued_at": (now - timedelta(days=2)).isoformat(timespec="seconds"), + "expires_at": (now + timedelta(days=3)).isoformat(timespec="seconds"), + "authorized_batch_manifest_raw_sha256": manifest_raw, + "keys": [ + { + "key_id": "rc06-release-owner-test", + "algorithm": "Ed25519", + "public_key_pem": reviewer_pem, + **apply.batch.SOLE_RELEASE_REVIEWER, + "not_before": (now - timedelta(days=2)).isoformat(timespec="seconds"), + "not_after": (now + timedelta(days=3)).isoformat(timespec="seconds"), + "revoked": False, + } + ], + } + keyring = { + **keyring_unsigned, + "keyring_payload_sha256": apply.batch._sha256_json(keyring_unsigned), + "root_signature_base64": base64.b64encode( + root_private.sign( + apply.batch.APPROVAL_KEYRING_SIGNATURE_DOMAIN + + apply.batch._canonical(keyring_unsigned) + ) + ).decode("ascii"), + } + approval_payload = apply.batch.build_approval_payload( + manifest, + manifest_raw_sha256=manifest_raw, + key_id="rc06-release-owner-test", + approval_keyring_epoch=1, + approval_keyring_payload_sha256=keyring["keyring_payload_sha256"], + approved_at=now - timedelta(hours=2), + expires_at=now + timedelta(days=2), + approver={ + **apply.batch.SOLE_RELEASE_REVIEWER, + "role": "测试发布负责人", + "organization": "测试组织（非真实法律服务主体）", + "authorization_scope": "仅批准本临时 SQLite/HTTP 测试候选，不授权生产发布", + "human_confirmed": True, + }, + ) + approval_signature = reviewer_private.sign( + apply.batch.APPROVAL_SIGNATURE_DOMAIN + apply.batch._canonical(approval_payload) + ) + approval = { + "schema_version": apply.batch.APPROVAL_SCHEMA_VERSION, + "approval_id": "rc06-batch-approval-" + + apply.batch._sha256_bytes( + apply.batch._canonical(approval_payload) + approval_signature + )[:24], + "payload": approval_payload, + "signature": base64.b64encode(approval_signature).decode("ascii"), + } + approval_raw = apply.batch._sha256_bytes(apply.batch._canonical(approval)) + occurrences = [] + for asset in sorted( + manifest["asset_batch"]["items"], key=lambda item: item["template_id"] + ): + references = ((asset.get("input_schema") or {}).get("scene") or {}).get( + "source_references", [] + ) + for index, source in enumerate(references): + occurrences.append( + { + "occurrence_id": f"fixture:{asset['template_id']}:{index}", + "scene_id": f"fixture-scene-{len(occurrences):02d}", + "template_id": asset["template_id"], + "source_key": f"source-{index}", + "source_pack_path": "fixture-only/not-production.json", + "source_record_sha256": apply.batch._sha256_json(source), + "source": source, + "required_human_decisions": [ + "current_effectiveness_verified", + "transaction_applicability_verified", + ], + } + ) + assert len(occurrences) == 43 + legal_request = { + "schema_version": "yuqei.scene-source-legal-review-request.v1", + "status": "pending_qualified_human_review", + "candidate_commit": manifest["candidate"]["commit"], + "created_at": (now - timedelta(hours=2)).isoformat(timespec="seconds"), + "scope": { + "scene_count": 30, + "reference_occurrence_count": 43, + "distinct_source_record_count": 8, + "distinct_url_count": 7, + }, + "source_files": [], + "occurrences": occurrences, + "guardrails": { + "not_legal_approval": True, + "authorizes_publication": False, + "ai_must_not_supply_human_decisions": True, + "all_occurrences_require_qualified_human_review": True, + }, + } + legal_request["request_sha256"] = apply.batch._sha256_json(legal_request) + legal_decision = { + "schema_version": "yuqei.scene-source-legal-review-decision.v1", + "request_sha256": legal_request["request_sha256"], + "candidate_commit": manifest["candidate"]["commit"], + "reviewed_at": (now - timedelta(hours=1)).isoformat(timespec="seconds"), + "expires_at": (now + timedelta(days=1)).isoformat(timespec="seconds"), + "reviewer": { + "name": "测试法源复核人（非真实身份）", + "organization": "测试组织（非真实法律服务主体）", + "qualification": "测试资格声明（不代表真实法律资质）", + "credential_reference": "TEST-FIXTURE-NOT-A-CREDENTIAL", + }, + "items": [ + { + "occurrence_id": item["occurrence_id"], + "source_record_sha256": item["source_record_sha256"], + "current_effectiveness_verified": True, + "transaction_applicability_verified": True, + "checked_at": (now - timedelta(hours=1)).isoformat(timespec="seconds"), + "authoritative_status_url": "https://example.gov.cn/test-fixture-status", + "rationale": "仅用于验证 fail-closed 协议的合成判断，不构成真实法源复核结论。", + } + for item in occurrences + ], + "signature": { + "algorithm": "Ed25519", + "key_id": "fixture-legal-reviewer", + "value_base64": base64.b64encode(b"fixture-signature").decode("ascii"), + }, + } + legal_keyring = { + "schema_version": "yuqei.scene-source-legal-review-keyring.v1", + "keys": [{"key_id": "fixture-legal-reviewer", "status": "trusted"}], + } + return { + "now": now, + "manifest": manifest, + "manifest_raw": manifest_raw, + "approval": approval, + "approval_raw": approval_raw, + "keyring": keyring, + "legal_request": legal_request, + "legal_decision": legal_decision, + "legal_keyring": legal_keyring, + "root_private": root_private, + "apply_private": apply_private, + "online_private": online_private, + } + + +class FakePublicationService: + def __init__( + self, + database: Path, + material: dict, + *, + fail_after: int | None = None, + tamper_status: bool = False, + tamper_runtime_status: bool = False, + ) -> None: + self.database = database + self.material = material + self.fail_after = fail_after + self.tamper_status = tamper_status + self.tamper_runtime_status = tamper_runtime_status + self.apply_calls = 0 + with sqlite3.connect(database) as connection: + connection.executescript( + """ + CREATE TABLE transactions ( + transaction_id TEXT PRIMARY KEY, + idempotency_key TEXT NOT NULL UNIQUE, + state TEXT NOT NULL, + legal_review_sha256 TEXT NOT NULL, + response_json TEXT + ); + CREATE TABLE publication_items ( + transaction_id TEXT NOT NULL, + kind TEXT NOT NULL, + item_id TEXT NOT NULL, + content_sha256 TEXT NOT NULL, + PRIMARY KEY(transaction_id, kind, item_id) + ); + CREATE TABLE runtime_snapshots ( + transaction_id TEXT NOT NULL, + kind TEXT NOT NULL, + runtime_key TEXT NOT NULL, + content_sha256 TEXT NOT NULL, + status TEXT NOT NULL, + PRIMARY KEY(transaction_id, runtime_key) + ); + """ + ) + + def _receipt_pair(self, request: dict) -> tuple[dict, dict]: + material = self.material + now = material["now"] + challenge_payload = { + "schema_version": apply.batch.APPLY_CHALLENGE_PAYLOAD_SCHEMA_VERSION, + "nonce_base64": base64.b64encode(b"q" * 32).decode("ascii"), + "candidate_commit": request["manifest"]["candidate"]["commit"], + "candidate_tag": request["manifest"]["candidate"]["tag"], + "batch_manifest_raw_sha256": request["batch_manifest_raw_sha256"], + "batch_manifest_payload_sha256": request["batch_manifest_payload_sha256"], + "approval_receipt_sha256": request["approval_receipt_sha256"], + "operation": apply.batch.APPLY_OPERATION, + "environment_id": request["environment_id"], + "actor": { + "type": "system", + "component": apply.batch.TRUSTED_ONLINE_APPLY_SERVICE_COMPONENT, + }, + "issued_at": (now - timedelta(minutes=1)).isoformat(timespec="seconds"), + "expires_at": (now + timedelta(minutes=5)).isoformat(timespec="seconds"), + } + challenge = _sign_envelope( + material["apply_private"], + schema=apply.batch.APPLY_CHALLENGE_SCHEMA_VERSION, + id_prefix="rc06-apply-challenge-", + domain=apply.batch.APPLY_CHALLENGE_SIGNATURE_DOMAIN, + payload=challenge_payload, + ) + challenge["challenge_id"] = challenge.pop("receipt_id") + committed_at = now.isoformat(timespec="seconds") + online_payload = { + "schema_version": apply.batch.ONLINE_TRANSACTION_PAYLOAD_SCHEMA_VERSION, + "key_id": apply.batch.TRUSTED_ONLINE_APPLY_SERVICE_KEY_ID, + "batch_manifest_payload_sha256": request["batch_manifest_payload_sha256"], + "approval_receipt_sha256": request["approval_receipt_sha256"], + "environment_id": request["environment_id"], + "transaction_id": request["transaction_id"], + "database_commit_id": "sqlite-" + request["idempotency_key"][:24], + "committed_at": committed_at, + "authority_challenge": challenge, + "generated_by": { + "type": "system", + "component": apply.batch.TRUSTED_ONLINE_APPLY_SERVICE_COMPONENT, + }, + "item_results": request["expected_publication_item_results"], + "write_count": 74, + "status": "committed_and_verified", + } + online_receipt = _sign_envelope( + material["online_private"], + schema=apply.batch.ONLINE_TRANSACTION_RECEIPT_SCHEMA_VERSION, + id_prefix="rc06-online-transaction-", + domain=apply.batch.ONLINE_TRANSACTION_SIGNATURE_DOMAIN, + payload=online_payload, + ) + apply_payload = { + "schema_version": apply.batch.APPLY_PAYLOAD_SCHEMA_VERSION, + "batch_manifest_payload_sha256": request["batch_manifest_payload_sha256"], + "approval_receipt_sha256": request["approval_receipt_sha256"], + "environment_id": request["environment_id"], + "applied_at": committed_at, + "generated_by": {"type": "system", "component": "rc06-apply-authority"}, + "item_results": request["expected_publication_item_results"], + "write_count": 74, + "status": "applied_and_verified", + "online_transaction_receipt": online_receipt, + } + apply_receipt = _sign_envelope( + material["apply_private"], + schema=apply.batch.APPLY_SCHEMA_VERSION, + id_prefix="rc06-apply-receipt-", + domain=apply.batch.APPLY_SIGNATURE_DOMAIN, + payload=apply_payload, + ) + runtime_payload = { + "schema_version": apply.RUNTIME_PAYLOAD_SCHEMA_VERSION, + "key_id": apply.batch.TRUSTED_ONLINE_APPLY_SERVICE_KEY_ID, + "batch_manifest_raw_sha256": request["batch_manifest_raw_sha256"], + "batch_manifest_payload_sha256": request["batch_manifest_payload_sha256"], + "approval_receipt_sha256": request["approval_receipt_sha256"], + "legal_source_review_binding_sha256": request[ + "legal_source_review_binding_sha256" + ], + "environment_id": request["environment_id"], + "transaction_id": request["transaction_id"], + "captured_at": committed_at, + "generated_by": { + "type": "system", + "component": apply.batch.TRUSTED_ONLINE_APPLY_SERVICE_COMPONENT, + }, + "transaction_mode": "single_service_atomic", + "publication_item_count": 74, + "runtime_snapshot_dependencies": request[ + "expected_runtime_snapshot_dependencies" + ], + "runtime_snapshot_dependency_count": 14, + "status": "runtime_snapshots_verified", + } + runtime_receipt = _sign_envelope( + material["online_private"], + schema=apply.RUNTIME_RECEIPT_SCHEMA_VERSION, + id_prefix="rc06-runtime-snapshot-", + domain=apply.RUNTIME_SIGNATURE_DOMAIN, + payload=runtime_payload, + ) + return apply_receipt, runtime_receipt + + def apply(self, request: dict, request_sha256: str) -> tuple[int, dict]: + self.apply_calls += 1 + with sqlite3.connect(self.database) as connection: + existing = connection.execute( + "SELECT response_json FROM transactions WHERE idempotency_key=? AND state='committed_and_verified'", + (request["idempotency_key"],), + ).fetchone() + if existing is not None: + return 200, json.loads(existing[0]) + try: + connection.execute("BEGIN IMMEDIATE") + connection.execute( + "INSERT OR REPLACE INTO transactions VALUES (?, ?, 'preparing', ?, NULL)", + ( + request["transaction_id"], + request["idempotency_key"], + request["legal_source_review_binding_sha256"], + ), + ) + for index, item in enumerate(request["expected_publication_item_results"]): + connection.execute( + "INSERT INTO publication_items VALUES (?, ?, ?, ?)", + ( + request["transaction_id"], + item["kind"], + item["item_id"], + item["online_content_sha256"], + ), + ) + if self.fail_after is not None and index + 1 == self.fail_after: + raise RuntimeError("injected transaction failure") + for item in request["expected_runtime_snapshot_dependencies"]: + connection.execute( + "INSERT INTO runtime_snapshots VALUES (?, ?, ?, ?, ?)", + ( + request["transaction_id"], + item["kind"], + item["runtime_key"], + item["online_content_sha256"], + item["status"], + ), + ) + apply_receipt, runtime_receipt = self._receipt_pair(request) + response = { + "schema_version": apply.APPLY_RESPONSE_SCHEMA_VERSION, + "idempotency_key": request["idempotency_key"], + "request_sha256": request_sha256, + "legal_source_review_binding_sha256": request[ + "legal_source_review_binding_sha256" + ], + "transaction_mode": "single_service_atomic", + "publication_item_count": 74, + "runtime_snapshot_dependency_count": 14, + "apply_receipt": apply_receipt, + "runtime_snapshot_receipt": runtime_receipt, + } + connection.execute( + "UPDATE transactions SET state='committed_and_verified', response_json=? WHERE transaction_id=?", + (json.dumps(response, ensure_ascii=False), request["transaction_id"]), + ) + connection.commit() + return 200, response + except RuntimeError: + connection.rollback() + return 409, {"error": "injected transaction failure; SQLite rolled back"} + + def status(self, transaction_id: str) -> tuple[int, dict]: + with sqlite3.connect(self.database) as connection: + transaction = connection.execute( + "SELECT idempotency_key, state, legal_review_sha256, response_json FROM transactions WHERE transaction_id=?", + (transaction_id,), + ).fetchone() + if transaction is None or transaction[1] == "rolled_back": + return 404, {"error": "transaction is not committed"} + response = json.loads(transaction[3]) + publication = response["apply_receipt"]["payload"]["item_results"] + runtime = response["runtime_snapshot_receipt"]["payload"][ + "runtime_snapshot_dependencies" + ] + if self.tamper_status: + publication = copy.deepcopy(publication) + publication[0]["online_content_sha256"] = "f" * 64 + if self.tamper_runtime_status: + runtime = copy.deepcopy(runtime) + runtime[0]["online_content_sha256"] = "f" * 64 + return 200, { + "schema_version": apply.APPLY_STATUS_SCHEMA_VERSION, + "idempotency_key": transaction[0], + "transaction_id": transaction_id, + "state": "committed_and_verified", + "legal_source_review_binding_sha256": transaction[2], + "publication_item_results": publication, + "publication_item_count": len(publication), + "runtime_snapshot_dependencies": runtime, + "runtime_snapshot_dependency_count": len(runtime), + "apply_receipt_sha256": apply.batch._sha256_json(response["apply_receipt"]), + "runtime_snapshot_receipt_sha256": apply.batch._sha256_json( + response["runtime_snapshot_receipt"] + ), + } + + def rollback(self, transaction_id: str, request: dict) -> tuple[int, dict]: + with sqlite3.connect(self.database) as connection: + connection.execute("BEGIN IMMEDIATE") + connection.execute( + "DELETE FROM publication_items WHERE transaction_id=?", (transaction_id,) + ) + connection.execute( + "DELETE FROM runtime_snapshots WHERE transaction_id=?", (transaction_id,) + ) + connection.execute( + "INSERT OR REPLACE INTO transactions " + "(transaction_id,idempotency_key,state,legal_review_sha256,response_json) " + "VALUES (?,?,'rolled_back',?,NULL)", + ( + transaction_id, + request["idempotency_key"], + request["legal_source_review_binding_sha256"], + ), + ) + connection.commit() + evidence = { + "transaction_id": transaction_id, + "idempotency_key": request["idempotency_key"], + "restored_item_count": 0, + } + return 200, { + "schema_version": apply.ROLLBACK_RESPONSE_SCHEMA_VERSION, + "idempotency_key": request["idempotency_key"], + "transaction_id": transaction_id, + "state": "rolled_back", + "restored_item_count": 0, + "rollback_evidence_sha256": apply.batch._sha256_json(evidence), + } + + def counts(self) -> tuple[int, int]: + with sqlite3.connect(self.database) as connection: + return ( + connection.execute("SELECT count(*) FROM publication_items").fetchone()[0], + connection.execute("SELECT count(*) FROM runtime_snapshots").fetchone()[0], + ) + + +@contextmanager +def _serve(service: FakePublicationService): + class Handler(BaseHTTPRequestHandler): + def log_message(self, _format, *_args): + return + + def _send(self, status: int, payload: dict) -> None: + raw = json.dumps(payload, ensure_ascii=False).encode("utf-8") + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(raw))) + self.end_headers() + self.wfile.write(raw) + + def _authorized(self) -> bool: + if self.headers.get("Authorization") != "Bearer test-token": + self._send(401, {"error": "unauthorized"}) + return False + return True + + def _body(self) -> dict: + return json.loads(self.rfile.read(int(self.headers["Content-Length"]))) + + def do_POST(self): + if not self._authorized(): + return + if self.path == "/internal/rc06/publication-transactions/apply": + request = self._body() + if ( + self.headers.get("Idempotency-Key") != request["idempotency_key"] + or self.headers.get("X-RC06-Request-SHA256") + != apply.batch._sha256_json(request) + ): + self._send(400, {"error": "identity header mismatch"}) + return + status, payload = service.apply( + request, self.headers["X-RC06-Request-SHA256"] + ) + self._send(status, payload) + return + if self.path.endswith("/rollback"): + request = self._body() + transaction_id = self.path.split("/")[-2] + status, payload = service.rollback(transaction_id, request) + self._send(status, payload) + return + self._send(404, {"error": "not found"}) + + def do_GET(self): + if not self._authorized(): + return + if self.path.endswith("/status"): + transaction_id = self.path.split("/")[-2] + status, payload = service.status(transaction_id) + self._send(status, payload) + return + self._send(404, {"error": "not found"}) + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + yield f"http://127.0.0.1:{server.server_port}" + finally: + server.shutdown() + thread.join(timeout=5) + server.server_close() + + +def _execute(material: dict, client, ledger) -> dict: + return apply.apply_publication_batch( + manifest=material["manifest"], + manifest_raw_sha256=material["manifest_raw"], + approval=material["approval"], + approval_raw_sha256=material["approval_raw"], + approval_keyring=material["keyring"], + 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(), + legal_review_request=material["legal_request"], + legal_review_decision=material["legal_decision"], + legal_review_keyring=material["legal_keyring"], + environment_id="rc06-test", + client=client, + ledger=ledger, + now=material["now"], + ) + + +def test_authenticated_apply_is_atomic_hash_verified_and_idempotent( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + material = _material(monkeypatch, datetime(2026, 10, 6, 8, 0, tzinfo=UTC)) + runtime = apply.expected_runtime_snapshot_results(material["manifest"]) + assert len(runtime) == 14 + drafting = next(item for item in runtime if item["kind"] == "composed_drafting") + assert drafting["runtime_key"] == "contract_drafting_v1" + assert drafting["version"] == "tp06-23-20260909-1" + assert sum(item["kind"] == "new_review_module" for item in runtime) == 8 + assert sum(item["kind"] == "existing_domain_prompt" for item in runtime) == 5 + service = FakePublicationService(tmp_path / "remote.sqlite3", material) + with _serve(service) as url: + client = apply.JsonHttpClient( + url, "test-token", allow_http_loopback=True + ) + ledger = apply.ApplyLedger(tmp_path / "ledger.sqlite3") + try: + evidence = _execute(material, client, ledger) + replay = _execute(material, client, ledger) + finally: + ledger.close() + assert replay == evidence + assert evidence["publication_item_count"] == 74 + assert evidence["runtime_snapshot_dependency_count"] == 14 + assert service.counts() == (74, 14) + assert service.apply_calls == 1 + + second_ledger = apply.ApplyLedger(tmp_path / "second-ledger.sqlite3") + try: + assert _execute(material, client, second_ledger) == evidence + finally: + second_ledger.close() + assert service.counts() == (74, 14) + assert service.apply_calls == 2 + + +@pytest.mark.parametrize("mode", ["missing", "expired", "hash_mismatch"]) +def test_legal_source_review_is_fail_closed_before_http( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + mode: str, +) -> None: + material = _material(monkeypatch, datetime(2026, 10, 6, 8, 0, tzinfo=UTC)) + if mode == "missing": + material["legal_request"] = {} + elif mode == "expired": + material["legal_decision"]["reviewed_at"] = "2026-10-04T08:00:00+00:00" + material["legal_decision"]["expires_at"] = "2026-10-05T08:00:00+00:00" + else: + material["legal_decision"]["items"][0]["source_record_sha256"] = "f" * 64 + service = FakePublicationService(tmp_path / "remote.sqlite3", 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.ApplyPublicationError, match="scene legal review"): + _execute(material, client, ledger) + finally: + ledger.close() + assert service.apply_calls == 0 + assert service.counts() == (0, 0) + + +@pytest.mark.parametrize("field", ["version", "content_sha256"]) +def test_forged_executable_runtime_identity_is_rejected_before_http( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + field: str, +) -> None: + material = _material(monkeypatch, datetime(2026, 10, 6, 8, 0, tzinfo=UTC)) + manifest = material["manifest"] + drafting = next( + item + for item in manifest["runtime_prompt_dependencies"]["items"] + if item["kind"] == "composed_drafting" + ) + drafting[field] = "forged-version" if field == "version" else "f" * 64 + manifest["runtime_prompt_dependencies"]["inventory_sha256"] = apply.batch._sha256_json( + manifest["runtime_prompt_dependencies"]["items"] + ) + manifest["manifest_payload_sha256"] = apply.batch._sha256_json( + {key: value for key, value in manifest.items() if key != "manifest_payload_sha256"} + ) + material["manifest_raw"] = apply.batch._sha256_bytes(apply.batch._canonical(manifest)) + service = FakePublicationService(tmp_path / "remote.sqlite3", 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.ApplyPublicationError, match="candidate executable source"): + _execute(material, client, ledger) + finally: + ledger.close() + assert service.apply_calls == 0 + assert service.counts() == (0, 0) + + +def test_runtime_snapshot_module_hash_must_match_publication_item( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + material = _material(monkeypatch, datetime(2026, 10, 6, 8, 0, tzinfo=UTC)) + manifest = material["manifest"] + drafting = next( + item + for item in manifest["runtime_prompt_dependencies"]["items"] + if item["kind"] == "composed_drafting" + ) + drafting["publication_module_dependencies"][0]["content_sha256"] = "f" * 64 + manifest["runtime_prompt_dependencies"]["inventory_sha256"] = apply.batch._sha256_json( + manifest["runtime_prompt_dependencies"]["items"] + ) + manifest["manifest_payload_sha256"] = apply.batch._sha256_json( + {key: value for key, value in manifest.items() if key != "manifest_payload_sha256"} + ) + material["manifest_raw"] = apply.batch._sha256_bytes(apply.batch._canonical(manifest)) + service = FakePublicationService(tmp_path / "remote.sqlite3", 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="exact publication module identities", + ): + _execute(material, client, ledger) + finally: + ledger.close() + assert service.apply_calls == 0 + assert service.counts() == (0, 0) + + +@pytest.mark.parametrize("failure", ["write", "readback_hash", "runtime_hash"]) +def test_failure_never_leaves_partial_or_false_success( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + failure: str, +) -> None: + material = _material(monkeypatch, datetime(2026, 10, 6, 8, 0, tzinfo=UTC)) + service = FakePublicationService( + tmp_path / "remote.sqlite3", + material, + fail_after=17 if failure == "write" else None, + tamper_status=failure == "readback_hash", + tamper_runtime_status=failure == "runtime_hash", + ) + 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.ApplyPublicationError): + _execute(material, client, ledger) + state = ledger.connection.execute( + "SELECT state FROM rc06_apply_operations" + ).fetchone()[0] + finally: + ledger.close() + assert state == "rolled_back" + assert service.counts() == (0, 0) + + +def test_http_and_auth_fail_closed() -> None: + with pytest.raises(apply.ApplyPublicationError, match="HTTPS"): + apply.JsonHttpClient("http://example.com", "token") + with pytest.raises(apply.ApplyPublicationError, match="bearer token"): + apply.JsonHttpClient("https://example.com", "")