From fe25e86746fa3ada3ba3a3a32de5ccec8ef321da Mon Sep 17 00:00:00 2001 From: yuqei Date: Tue, 6 Oct 2026 23:02:05 +0800 Subject: [PATCH] fix(rc06): bind compensation evidence and sealing --- ...006_0031_add_rc06_compensation_evidence.py | 13 +- .../yuqei_ai_platform_api/rc06_publication.py | 95 +++++++++++- .../tests/test_rc06_migrations.py | 48 ++++++ .../tests/test_rc06_publication.py | 137 ++++++++++++++++++ 4 files changed, 287 insertions(+), 6 deletions(-) diff --git a/yuqei-ai-platform/services/ai-platform-api/migrations/versions/20261006_0031_add_rc06_compensation_evidence.py b/yuqei-ai-platform/services/ai-platform-api/migrations/versions/20261006_0031_add_rc06_compensation_evidence.py index f81a4b7c..29458f46 100644 --- a/yuqei-ai-platform/services/ai-platform-api/migrations/versions/20261006_0031_add_rc06_compensation_evidence.py +++ b/yuqei-ai-platform/services/ai-platform-api/migrations/versions/20261006_0031_add_rc06_compensation_evidence.py @@ -27,18 +27,23 @@ def _assert_no_unverifiable_compensation() -> None: ) -def _assert_no_compensation_evidence() -> None: +def _assert_safe_downgrade() -> None: bind = op.get_bind() + if bind.dialect.name == "postgresql": + bind.execute(sa.text( + "LOCK TABLE rc06_prompt_publications IN ACCESS EXCLUSIVE MODE" + )) count = bind.execute(sa.text( "SELECT COUNT(*) FROM rc06_prompt_publications " - "WHERE compensated_from_state IS NOT NULL " + "WHERE state != 'prepared' " + "OR compensated_from_state IS NOT NULL " "OR compensation_item_results_json IS NOT NULL " "OR runtime_snapshot_compensation_results_json IS NOT NULL " "OR compensation_evidence_sha256 IS NOT NULL" )).scalar_one() if count: raise RuntimeError( - "RC06 Prompt migration 0031 downgrade refuses to discard compensation evidence; " + "RC06 Prompt migration 0031 downgrade refuses active or evidenced transactions; " "back up and explicitly retire those rows first." ) @@ -68,7 +73,7 @@ def upgrade() -> None: def downgrade() -> None: - _assert_no_compensation_evidence() + _assert_safe_downgrade() with op.batch_alter_table("rc06_prompt_publications") as batch: batch.drop_constraint( "ck_rc06_prompt_publications_compensation_evidence", diff --git a/yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/rc06_publication.py b/yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/rc06_publication.py index b80f2e2b..cae445dc 100644 --- a/yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/rc06_publication.py +++ b/yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/rc06_publication.py @@ -47,6 +47,28 @@ EXPECTED_MODULE_COUNT = 14 EXPECTED_RUNTIME_COUNT = 14 NEW_RUNTIME_COUNT = 9 TERMINAL_STATES = {"sealed", "compensated", "compensation_failed"} +PROMPT_ROW_KEYS = { + "id", + "name", + "version", + "content", + "enabled", + "status", + "rollback_from_version", + "published_at", + "updated_at", +} +RUNTIME_SNAPSHOT_KEYS = { + "name", + "version", + "content_sha256", + "source_kind", + "prompt_template_id", + "status", + "enabled", + "online_content_sha256", + "verified", +} def is_rc06_publication_path(path: str, api_prefix: str) -> bool: @@ -196,8 +218,14 @@ def _frozen_request(transaction: Rc06PromptPublicationModel) -> PromptPrepareReq if _sha(payload.model_dump(mode="json")) != transaction.request_sha256: raise Rc06PublicationError("Frozen RC06 Prompt request hash drifted.") if ( - payload.candidate_commit != transaction.candidate_commit + payload.transaction_id != transaction.transaction_id + or payload.idempotency_key != transaction.idempotency_key + or payload.candidate_commit != transaction.candidate_commit or payload.release_image_digest != transaction.release_image_digest + or payload.candidate_tag != transaction.candidate_tag + or payload.batch_manifest_raw_sha256 != transaction.batch_manifest_raw_sha256 + or payload.batch_manifest_payload_sha256 != transaction.batch_manifest_payload_sha256 + or payload.approval_receipt_sha256 != transaction.approval_receipt_sha256 or payload.compensation_capability_sha256 != transaction.compensation_capability_sha256 ): raise Rc06PublicationError("Frozen RC06 Prompt deployment identity drifted.") @@ -462,6 +490,7 @@ class Rc06PromptPublicationService: def seal(self, transaction_id: str, capability: str) -> PromptTransactionStatus: frozen = self._load_frozen_request(transaction_id) + self._assert_current_deployment(frozen) expected_modules = sorted( [item.model_dump(mode="json") for item in frozen.modules], key=lambda item: (item["category"], item["key"]), @@ -1006,10 +1035,37 @@ class Rc06PromptPublicationService: return { "schema_version": "rc06-ai-compensation-evidence/v1", "transaction_id": transaction.transaction_id, + "idempotency_key": transaction.idempotency_key, + "request_sha256": transaction.request_sha256, "candidate_commit": transaction.candidate_commit, "candidate_tag": transaction.candidate_tag, "release_image_digest": transaction.release_image_digest, + "batch_manifest_raw_sha256": transaction.batch_manifest_raw_sha256, + "batch_manifest_payload_sha256": transaction.batch_manifest_payload_sha256, + "approval_receipt_sha256": transaction.approval_receipt_sha256, "compensated_from_state": transaction.compensated_from_state, + "committed_item_results_sha256": _sha( + _decode(transaction.item_results_json, []) + ), + "committed_runtime_snapshots_sha256": _sha( + _decode(transaction.runtime_snapshots_json, []) + ), + "before_prompts_sha256": _sha( + _decode(transaction.before_prompts_json, []) + ), + "after_prompts_sha256": _sha( + _decode(transaction.after_prompts_json, []) + ), + "committed_at": ( + _as_utc(transaction.committed_at).isoformat() + if transaction.committed_at is not None + else None + ), + "sealed_at": ( + _as_utc(transaction.sealed_at).isoformat() + if transaction.sealed_at is not None + else None + ), "compensation_item_results": item_results, "runtime_snapshot_compensation_results": runtime_results, "restored_item_count": transaction.restored_item_count, @@ -1214,10 +1270,20 @@ class Rc06PromptPublicationService: raise Rc06PublicationError( "Preserved committed Prompt module evidence drifted after compensation." ) + if ( + transaction.committed_at is None + or transaction.sealed_at is not None + or transaction.after_prompts_json is None + ): + raise Rc06PublicationError( + "Preserved committed Prompt state markers drifted after compensation." + ) snapshots = _decode(transaction.runtime_snapshots_json, []) + after_rows = _decode(transaction.after_prompts_json, []) expected_by_name = {item["name"]: item for item in identities} if ( not isinstance(snapshots, list) + or not isinstance(after_rows, list) or len(snapshots) != EXPECTED_RUNTIME_COUNT or [item.get("name") for item in snapshots] != sorted(expected_by_name) @@ -1225,8 +1291,26 @@ class Rc06PromptPublicationService: raise Rc06PublicationError( "Preserved committed runtime evidence is incomplete after compensation." ) + if any( + not isinstance(item, dict) or set(item) != PROMPT_ROW_KEYS + for item in after_rows + ): + raise Rc06PublicationError( + "Preserved committed Prompt after-image schema drifted." + ) + after_by_id = {item["id"]: item for item in after_rows} + if len(after_by_id) != len(after_rows): + raise Rc06PublicationError( + "Preserved committed Prompt after-image contains duplicate row identities." + ) for snapshot in snapshots: + if not isinstance(snapshot, dict) or set(snapshot) != RUNTIME_SNAPSHOT_KEYS: + raise Rc06PublicationError( + "Preserved committed runtime evidence schema drifted." + ) identity = expected_by_name[snapshot["name"]] + prompt_template_id = snapshot.get("prompt_template_id") + after = after_by_id.get(prompt_template_id) if ( snapshot.get("version") != identity["version"] or snapshot.get("source_kind") != identity["source_kind"] @@ -1235,7 +1319,14 @@ class Rc06PromptPublicationService: or snapshot.get("status") != "published" or snapshot.get("enabled") is not True or snapshot.get("verified") is not True - or not isinstance(snapshot.get("prompt_template_id"), str) + or not isinstance(prompt_template_id, str) + or not prompt_template_id + or after is None + or after["name"] != identity["name"] + or after["version"] != identity["version"] + or prompt_content_hash(after["content"]) != identity["content_sha256"] + or after["status"] != "published" + or after["enabled"] is not True ): raise Rc06PublicationError( f"Preserved committed runtime evidence drifted: {identity['name']}" diff --git a/yuqei-ai-platform/services/ai-platform-api/tests/test_rc06_migrations.py b/yuqei-ai-platform/services/ai-platform-api/tests/test_rc06_migrations.py index d39a0e31..63536c94 100644 --- a/yuqei-ai-platform/services/ai-platform-api/tests/test_rc06_migrations.py +++ b/yuqei-ai-platform/services/ai-platform-api/tests/test_rc06_migrations.py @@ -246,6 +246,54 @@ def test_compensation_evidence_migration_rejects_unverifiable_old_rows(tmp_path) engine.dispose() +def test_compensation_evidence_migration_rejects_active_committed_downgrade(tmp_path): + engine = sa.create_engine( + f"sqlite+pysqlite:///{(tmp_path / 'ai-active-committed.sqlite3').as_posix()}" + ) + migration = _load("20261006_0031_add_rc06_compensation_evidence.py") + with engine.begin() as connection: + _hardened_head(connection) + _run(connection, migration, "upgrade") + now = datetime.now(UTC) + table = sa.Table( + "rc06_prompt_publications", + sa.MetaData(), + autoload_with=connection, + ) + connection.execute(table.insert().values( + transaction_id="active-committed-transaction", + idempotency_key="active-committed-idempotency", + request_sha256="1" * 64, + candidate_commit="2" * 40, + release_image_digest="sha256:" + "3" * 64, + compensation_capability_sha256="4" * 64, + candidate_tag="active-committed-candidate", + batch_manifest_raw_sha256="5" * 64, + batch_manifest_payload_sha256="6" * 64, + approval_receipt_sha256="7" * 64, + state="committed", + request_json="{}", + before_prompts_json="[]", + after_prompts_json="[]", + item_results_json="[]", + runtime_snapshots_json="[]", + created_at=now, + updated_at=now, + committed_at=now, + )) + + with pytest.raises(RuntimeError, match="refuses active or evidenced transactions"): + _run(connection, migration, "downgrade") + assert "compensation_evidence_sha256" in { + item["name"] + for item in sa.inspect(connection).get_columns("rc06_prompt_publications") + } + assert connection.exec_driver_sql( + "SELECT state FROM rc06_prompt_publications" + ).scalar_one() == "committed" + engine.dispose() + + def test_fresh_alembic_upgrade_then_old_head_round_trip(tmp_path, monkeypatch): database_path = tmp_path / "ai-fresh-full-chain.sqlite3" monkeypatch.setenv( diff --git a/yuqei-ai-platform/services/ai-platform-api/tests/test_rc06_publication.py b/yuqei-ai-platform/services/ai-platform-api/tests/test_rc06_publication.py index cf20c4f4..5360b711 100644 --- a/yuqei-ai-platform/services/ai-platform-api/tests/test_rc06_publication.py +++ b/yuqei-ai-platform/services/ai-platform-api/tests/test_rc06_publication.py @@ -311,6 +311,77 @@ def test_compensated_status_rejects_missing_duplicate_or_tampered_evidence( repository._engine.dispose() +@pytest.mark.parametrize("tamper", ["prompt_template_id", "extra_field"]) +def test_compensated_status_rejects_tampered_preserved_runtime_schema( + tmp_path, + tamper, +): + repository = _repository(tmp_path) + try: + _seed_domain_dependencies(repository) + service = _service(repository) + request = _request(f"rc06-ai-preserved-runtime-{tamper}") + service.prepare(request) + service.commit(request.transaction_id) + service.compensate(request.transaction_id, COMPENSATION_CAPABILITY) + + with repository._session_factory.begin() as session: + transaction = session.get(Rc06PromptPublicationModel, request.transaction_id) + snapshots = json.loads(transaction.runtime_snapshots_json) + if tamper == "prompt_template_id": + snapshots[0]["prompt_template_id"] = "forged-nonexistent-prompt-id" + else: + snapshots[0]["attacker_field"] = "must-not-be-accepted" + transaction.runtime_snapshots_json = json.dumps( + snapshots, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ) + transaction.compensation_evidence_sha256 = hashlib.sha256( + json.dumps( + service._compensation_evidence_payload(transaction), + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ).encode("utf-8") + ).hexdigest() + + with pytest.raises(Rc06PublicationError, match="Preserved committed runtime"): + service.status(request.transaction_id) + finally: + repository._engine.dispose() + + +@pytest.mark.parametrize("field", ["committed_at", "sealed_at"]) +def test_compensated_status_rejects_impossible_committed_state_markers(tmp_path, field): + repository = _repository(tmp_path) + try: + _seed_domain_dependencies(repository) + service = _service(repository) + request = _request(f"rc06-ai-compensated-marker-{field}") + service.prepare(request) + service.commit(request.transaction_id) + service.compensate(request.transaction_id, COMPENSATION_CAPABILITY) + + with repository._session_factory.begin() as session: + transaction = session.get(Rc06PromptPublicationModel, request.transaction_id) + setattr(transaction, field, None if field == "committed_at" else datetime.now(UTC)) + transaction.compensation_evidence_sha256 = hashlib.sha256( + json.dumps( + service._compensation_evidence_payload(transaction), + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ).encode("utf-8") + ).hexdigest() + + with pytest.raises(Rc06PublicationError, match="state markers drifted"): + service.status(request.transaction_id) + finally: + repository._engine.dispose() + + def test_prepare_fails_closed_when_a_preexisting_domain_dependency_is_missing(tmp_path): repository = _repository(tmp_path) try: @@ -394,6 +465,47 @@ def test_status_returns_persisted_deployment_identity_and_rejects_tampering(tmp_ repository._engine.dispose() +@pytest.mark.parametrize( + ("field", "replacement"), + [ + ("candidate_tag", "forged-candidate-tag"), + ("batch_manifest_raw_sha256", "8" * 64), + ("batch_manifest_payload_sha256", "9" * 64), + ("approval_receipt_sha256", "f" * 64), + ], +) +def test_compensated_status_rejects_frozen_release_binding_tampering( + tmp_path, + field, + replacement, +): + repository = _repository(tmp_path) + try: + _seed_domain_dependencies(repository) + service = _service(repository) + request = _request(f"rc06-ai-release-binding-{field}") + service.prepare(request) + service.commit(request.transaction_id) + service.compensate(request.transaction_id, COMPENSATION_CAPABILITY) + + with repository._session_factory.begin() as session: + transaction = session.get(Rc06PromptPublicationModel, request.transaction_id) + setattr(transaction, field, replacement) + transaction.compensation_evidence_sha256 = hashlib.sha256( + json.dumps( + service._compensation_evidence_payload(transaction), + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ).encode("utf-8") + ).hexdigest() + + with pytest.raises(Rc06PublicationError, match="deployment identity drifted"): + service.status(request.transaction_id) + finally: + repository._engine.dispose() + + def test_compensation_error_is_persisted_as_terminal_no_go(tmp_path, monkeypatch): repository = _repository(tmp_path) try: @@ -627,6 +739,31 @@ def test_capability_is_required_and_seal_permanently_disables_compensation(tmp_p repository._engine.dispose() +def test_seal_rejects_current_deployment_drift_after_commit(tmp_path): + repository = _repository(tmp_path) + try: + _seed_domain_dependencies(repository) + service = _service(repository) + request = _request("rc06-ai-seal-deployment-drift") + service.prepare(request) + service.commit(request.transaction_id) + + drifted_service = _service( + repository, + frozen_candidate_commit="b" * 40, + release_image_digest="sha256:" + "9" * 64, + ) + with pytest.raises(Rc06Conflict, match="does not match the frozen candidate"): + drifted_service.seal(request.transaction_id, COMPENSATION_CAPABILITY) + + status = service.status(request.transaction_id) + assert status.state == "committed" + assert status.no_go is True + assert status.sealed_at is None + finally: + repository._engine.dispose() + + def test_http_compensation_requires_service_auth_and_capability(tmp_path): repository = _repository(tmp_path) try: