fix(rc06): recover sealed prompts after authority expiry

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}

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: