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 f0b73648..f4eb83f4 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 @@ -13,7 +13,7 @@ import hashlib import hmac import json import re -from datetime import UTC, datetime +from datetime import UTC, datetime, timedelta from pathlib import Path from typing import Any, Literal from uuid import uuid4 @@ -45,6 +45,7 @@ IMAGE_DIGEST = r"^sha256:[a-f0-9]{64}$" EXPECTED_MODULE_COUNT = 14 EXPECTED_RUNTIME_COUNT = 14 NEW_RUNTIME_COUNT = 9 +AUTHORITY_COMMIT_MARGIN = timedelta(seconds=5) TERMINAL_STATES = {"sealed", "compensated", "compensation_failed"} PROMPT_ROW_KEYS = { "id", @@ -235,9 +236,26 @@ def _database_wall_clock(session: Any) -> datetime: return normalized -def _assert_authority_active(session: Any, authority_expires_at: datetime) -> None: - if _database_wall_clock(session) >= authority_expires_at.astimezone(UTC): +def _assert_authority_active(session: Any, authority_expires_at: datetime) -> datetime: + boundary_at = _database_wall_clock(session) + if boundary_at + AUTHORITY_COMMIT_MARGIN >= authority_expires_at.astimezone(UTC): raise Rc06Conflict("RC06 publication authority expired before AI persistence.") + return boundary_at + + +def _assert_committed_within_authority( + transaction: Rc06PromptPublicationModel, +) -> None: + committed_at = _as_utc(transaction.committed_at) + authority_expires_at = _as_utc(transaction.authority_expires_at) + if ( + committed_at is None + or authority_expires_at is None + or committed_at + AUTHORITY_COMMIT_MARGIN >= authority_expires_at + ): + raise Rc06PublicationError( + "RC06 Prompt durable commit is not proven within its authority window." + ) def _prompt_rows(session: Any, names: list[str]) -> list[dict[str, Any]]: @@ -512,10 +530,15 @@ class Rc06PromptPublicationService: transaction.after_prompts_json = _encode(after) transaction.state = "committed" transaction.error_message = None - transaction.committed_at = now transaction.updated_at = now session.flush() - _assert_authority_active(session, frozen.authority_expires_at) + boundary_at = _assert_authority_active( + session, + frozen.authority_expires_at, + ) + transaction.committed_at = boundary_at + transaction.updated_at = boundary_at + session.flush() return self._status_row(transaction) except (KeyError, Rc06PublicationError, Rc06Conflict): raise @@ -546,14 +569,12 @@ class Rc06PromptPublicationService: return self._status_row(transaction) if transaction.state != "committed": raise Rc06Conflict(f"Prompt transaction cannot seal from {transaction.state}.") - _assert_authority_active(session, frozen.authority_expires_at) self._verify_committed(session, transaction, expected_runtime, expected_modules) now = datetime.now(UTC) transaction.state = "sealed" transaction.sealed_at = now transaction.updated_at = now session.flush() - _assert_authority_active(session, frozen.authority_expires_at) return self._status_row(transaction) def compensate(self, transaction_id: str, capability: str) -> PromptTransactionStatus: @@ -799,6 +820,7 @@ class Rc06PromptPublicationService: identities: list[dict[str, Any]], modules: list[dict[str, Any]], ) -> None: + _assert_committed_within_authority(transaction) snapshots = self._runtime_snapshots(session, identities) if snapshots != _decode(transaction.runtime_snapshots_json, []): raise Rc06Conflict("Committed Prompt runtime state drifted; automatic compensation refused.") @@ -1314,6 +1336,7 @@ class Rc06PromptPublicationService: raise Rc06PublicationError( "Preserved committed Prompt state markers drifted after compensation." ) + _assert_committed_within_authority(transaction) snapshots = _decode(transaction.runtime_snapshots_json, []) after_rows = _decode(transaction.after_prompts_json, []) expected_by_name = {item["name"]: item for item in identities} 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 ec9169df..af421ad7 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 @@ -129,7 +129,7 @@ def test_commit_rolls_back_all_ai_writes_when_authority_expires_before_precommit request = _request("rc06-ai-expiry-crossing") service.prepare(request) expiry = request.authority_expires_at - readings = iter((expiry - timedelta(seconds=1), expiry)) + readings = iter((expiry - timedelta(seconds=10), expiry - timedelta(seconds=5))) monkeypatch.setattr( rc06_publication, "_database_wall_clock", @@ -154,6 +154,63 @@ def test_commit_rolls_back_all_ai_writes_when_authority_expires_before_precommit repository._engine.dispose() +def test_commit_persists_final_database_authority_boundary(tmp_path, monkeypatch): + repository = _repository(tmp_path) + try: + _seed_domain_dependencies(repository) + service = _service(repository) + request = _request("rc06-ai-database-boundary") + service.prepare(request) + expiry = request.authority_expires_at + prewrite = expiry - timedelta(seconds=20) + precommit = expiry - timedelta(seconds=10) + readings = iter((prewrite, precommit)) + monkeypatch.setattr( + rc06_publication, + "_database_wall_clock", + lambda _session: next(readings), + ) + + committed = service.commit(request.transaction_id) + + assert committed.state == "committed" + assert committed.committed_at == precommit + with repository._session_factory() as session: + row = session.get(Rc06PromptPublicationModel, request.transaction_id) + assert row.committed_at.replace(tzinfo=UTC) == precommit + finally: + repository._engine.dispose() + + +def test_seal_after_expiry_recovers_proven_preexpiry_durable_commit( + tmp_path, + monkeypatch, +): + repository = _repository(tmp_path) + try: + _seed_domain_dependencies(repository) + service = _service(repository) + request = _request("rc06-ai-expired-seal-recovery") + service.prepare(request) + committed = service.commit(request.transaction_id) + monkeypatch.setattr( + rc06_publication, + "_database_wall_clock", + lambda _session: request.authority_expires_at + timedelta(days=1), + ) + + sealed = service.seal(request.transaction_id, COMPENSATION_CAPABILITY) + + assert sealed.state == "sealed" + assert sealed.committed_at == committed.committed_at + assert service.seal( + request.transaction_id, + COMPENSATION_CAPABILITY, + ) == sealed + finally: + repository._engine.dispose() + + def test_sqlite_prompt_participant_is_durable_idempotent_and_compensates(tmp_path): repository = _repository(tmp_path) try: