fix(rc06): bind prompt writes to live authority

yuqei-ai-platform/services/ai-platform-api/migrations/versions/20261007_0032_bind_rc06_authority_expiry.py

@@ -0,0 +1,56 @@
+"""bind RC06 AI transactions to the authority expiry
+
+Revision ID: 20261007_0032
+Revises: 20261006_0031
+"""
+from __future__ import annotations
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "20261007_0032"
+down_revision = "20261006_0031"
+branch_labels = None
+depends_on = None
+
+
+def _lock_and_assert_empty(action: str) -> 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"
+    )).scalar_one()
+    if count:
+        raise RuntimeError(
+            f"RC06 Prompt migration 0032 {action} refuses nonempty publication "
+            "journals because authority expiry cannot be inferred or discarded safely; "
+            "back up and explicitly retire those rows first."
+        )
+
+
+def upgrade() -> None:
+    _lock_and_assert_empty("upgrade")
+    with op.batch_alter_table("rc06_prompt_publications") as batch:
+        batch.add_column(sa.Column(
+            "authority_expires_at",
+            sa.DateTime(timezone=True),
+            nullable=False,
+        ))
+        batch.create_check_constraint(
+            "ck_rc06_prompt_publications_authority_window",
+            "authority_expires_at > created_at",
+        )
+
+
+def downgrade() -> None:
+    _lock_and_assert_empty("downgrade")
+    with op.batch_alter_table("rc06_prompt_publications") as batch:
+        batch.drop_constraint(
+            "ck_rc06_prompt_publications_authority_window",
+            type_="check",
+        )
+        batch.drop_column("authority_expires_at")

yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/models.py

@@ -194,6 +194,10 @@ class Rc06PromptPublicationModel(Base):
             "compensated_at IS NOT NULL)",
             name="ck_rc06_prompt_publications_compensation_evidence",
         ),
+        CheckConstraint(
+            "authority_expires_at > created_at",
+            name="ck_rc06_prompt_publications_authority_window",
+        ),
     )
 
     transaction_id: Mapped[str] = mapped_column(String(64), primary_key=True)
@@ -206,6 +210,10 @@ class Rc06PromptPublicationModel(Base):
     batch_manifest_raw_sha256: Mapped[str] = mapped_column(String(64), nullable=False)
     batch_manifest_payload_sha256: Mapped[str] = mapped_column(String(64), nullable=False)
     approval_receipt_sha256: Mapped[str] = mapped_column(String(64), nullable=False)
+    authority_expires_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        nullable=False,
+    )
     state: Mapped[str] = mapped_column(String(32), nullable=False, index=True)
     request_json: Mapped[str] = mapped_column(Text, nullable=False)
     before_prompts_json: Mapped[str] = mapped_column(Text, nullable=False)

yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/rc06_publication.py

@@ -15,13 +15,12 @@ import json
 import re
 from datetime import UTC, datetime
 from pathlib import Path
-from types import SimpleNamespace
 from typing import Any, Literal
 from uuid import uuid4
 
 from fastapi import FastAPI, Header, HTTPException
-from pydantic import BaseModel, ConfigDict, Field
-from sqlalchemy import select
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+from sqlalchemy import select, text
 
 from yuqei_sdk.prompt_source_baseline import prompt_source_baseline
 
@@ -36,7 +35,7 @@ from .models import (
 from .packaged_prompt_assets_import import DEFAULT_SOURCE_FILE, DEFAULT_VERSION, load_packaged_prompt_assets
 from .persistence.prompts import encode_model_json, prompt_template_from_model, prompt_write_transaction
 from .prompt_compiler import prompt_content_hash, validate_prompt_content
-from .review_scene_prompts import review_modules, scene_review_template
+from .review_scene_prompts import review_modules, scene_review_template_for_route
 from .service_auth import is_valid_service_key
 
 
@@ -123,10 +122,18 @@ class PromptPrepareRequest(BaseModel):
     batch_manifest_raw_sha256: str = Field(pattern=HEX64)
     batch_manifest_payload_sha256: str = Field(pattern=HEX64)
     approval_receipt_sha256: str = Field(pattern=HEX64)
+    authority_expires_at: datetime
     compensation_capability_sha256: str = Field(pattern=HEX64)
     modules: list[PromptModuleInput]
     runtime_identities: list[RuntimeIdentityInput]
 
+    @field_validator("authority_expires_at")
+    @classmethod
+    def require_aware_authority_expiry(cls, value: datetime) -> datetime:
+        if value.tzinfo is None or value.utcoffset() is None:
+            raise ValueError("authority_expires_at must include a timezone")
+        return value.astimezone(UTC)
+
 
 class PromptTransactionStatus(BaseModel):
     transaction_id: str
@@ -137,6 +144,7 @@ class PromptTransactionStatus(BaseModel):
     batch_manifest_raw_sha256: str
     batch_manifest_payload_sha256: str
     approval_receipt_sha256: str
+    authority_expires_at: datetime
     item_results: list[dict[str, Any]] = Field(default_factory=list)
     runtime_snapshots: list[dict[str, Any]] = Field(default_factory=list)
     write_count: int = 0
@@ -203,6 +211,35 @@ def _as_utc(value: datetime | None) -> datetime | None:
     return value.replace(tzinfo=UTC) if value.tzinfo is None else value.astimezone(UTC)
 
 
+def _database_wall_clock(session: Any) -> datetime:
+    """Read a participant-local database clock, never the application clock.
+
+    PostgreSQL's CURRENT_TIMESTAMP is fixed at transaction start, so RC06 uses
+    clock_timestamp() there. SQLite is only supported for local tests and
+    embedded development; each CURRENT_TIMESTAMP statement observes its live
+    database clock.
+    """
+    statement = (
+        "SELECT clock_timestamp()"
+        if session.get_bind().dialect.name == "postgresql"
+        else "SELECT CURRENT_TIMESTAMP"
+    )
+    value = session.execute(text(statement)).scalar_one()
+    if isinstance(value, str):
+        value = datetime.fromisoformat(value.replace(" ", "T"))
+    if not isinstance(value, datetime):
+        raise Rc06PublicationError("RC06 AI database clock returned an invalid value.")
+    normalized = _as_utc(value)
+    if normalized is None:  # pragma: no cover - guarded by the type check above
+        raise Rc06PublicationError("RC06 AI database clock returned no value.")
+    return normalized
+
+
+def _assert_authority_active(session: Any, authority_expires_at: datetime) -> None:
+    if _database_wall_clock(session) >= authority_expires_at.astimezone(UTC):
+        raise Rc06Conflict("RC06 publication authority expired before AI persistence.")
+
+
 def _prompt_rows(session: Any, names: list[str]) -> list[dict[str, Any]]:
     rows = session.scalars(
         select(PromptTemplateModel).where(PromptTemplateModel.name.in_(names))
@@ -226,6 +263,7 @@ def _frozen_request(transaction: Rc06PromptPublicationModel) -> PromptPrepareReq
         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.authority_expires_at != _as_utc(transaction.authority_expires_at)
         or payload.compensation_capability_sha256 != transaction.compensation_capability_sha256
     ):
         raise Rc06PublicationError("Frozen RC06 Prompt deployment identity drifted.")
@@ -265,22 +303,12 @@ def _source_inventory() -> tuple[list[dict[str, Any]], list[dict[str, Any]], dic
     modules.sort(key=lambda item: (item["category"], item["key"]))
 
     runtime_contents: dict[str, str] = {DRAFT_PROMPT_NAMES[0]: DRAFT_PROMPT_CONTENT}
-    route_inputs = {
-        "contract_review_trade_procurement": ("purchase", "scene_evaluation_v3"),
-        "contract_review_services": ("service", "scene_evaluation_v3"),
-        "contract_review_transport_warehousing": ("transport", "scene_evaluation_v4"),
-        "contract_review_manufacturing": ("manufacturing", "scene_evaluation_v4"),
-        "contract_review_construction": ("construction", "scene_evaluation_v3"),
-        "contract_review_marketing": ("marketing", "scene_evaluation_v4"),
-        "contract_review_consumer_services": ("consumer_service", "scene_evaluation_v3"),
-        "contract_review_notices_amendments": ("amendment", "scene_evaluation_v4"),
-    }
-    for route, (contract_type, profile) in route_inputs.items():
-        template = scene_review_template(SimpleNamespace(
-            contract_type=contract_type,
-            scene_template_id=None,
-            review_profile=profile,
-        ))
+    for route in reviews:
+        template = scene_review_template_for_route(
+            route,
+            review_profile="scene_evaluation_v4",
+            scene_template_id="rc06-runtime-snapshot",
+        )
         if template.name != route:
             raise Rc06PublicationError(f"Runtime route drift: {route}")
         runtime_contents[route] = template.content
@@ -372,6 +400,8 @@ class Rc06PromptPublicationService:
             if session.get(Rc06PromptPublicationModel, payload.transaction_id) is not None:
                 raise Rc06Conflict("Transaction ID was already used with a different idempotency key.")
 
+            _assert_authority_active(session, payload.authority_expires_at)
+
             rows = session.scalars(select(PromptTemplateModel).where(PromptTemplateModel.name.in_(names))).all()
             before = sorted((_prompt_row(row) for row in rows), key=lambda item: (item["name"], item["id"]))
             self._assert_domain_dependencies(session, expected_runtime)
@@ -387,6 +417,7 @@ class Rc06PromptPublicationService:
                 batch_manifest_raw_sha256=payload.batch_manifest_raw_sha256,
                 batch_manifest_payload_sha256=payload.batch_manifest_payload_sha256,
                 approval_receipt_sha256=payload.approval_receipt_sha256,
+                authority_expires_at=payload.authority_expires_at,
                 state="prepared",
                 request_json=_encode(request_payload),
                 before_prompts_json=_encode(before),
@@ -423,6 +454,7 @@ class Rc06PromptPublicationService:
                     updated_at=now,
                 ))
             session.flush()
+            _assert_authority_active(session, payload.authority_expires_at)
             return self._status_row(transaction)
 
     def commit(self, transaction_id: str) -> PromptTransactionStatus:
@@ -452,6 +484,7 @@ class Rc06PromptPublicationService:
                 if transaction.state != "prepared":
                     raise Rc06Conflict(f"Prompt transaction cannot commit from {transaction.state}.")
                 self._assert_current_deployment(frozen)
+                _assert_authority_active(session, frozen.authority_expires_at)
                 before = _decode(transaction.before_prompts_json, [])
                 current = _prompt_rows(session, names)
                 if current != before:
@@ -482,6 +515,7 @@ class Rc06PromptPublicationService:
                 transaction.committed_at = now
                 transaction.updated_at = now
                 session.flush()
+                _assert_authority_active(session, frozen.authority_expires_at)
                 return self._status_row(transaction)
         except (KeyError, Rc06PublicationError, Rc06Conflict):
             raise
@@ -512,12 +546,14 @@ 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:
@@ -1472,6 +1508,7 @@ class Rc06PromptPublicationService:
             batch_manifest_raw_sha256=row.batch_manifest_raw_sha256,
             batch_manifest_payload_sha256=row.batch_manifest_payload_sha256,
             approval_receipt_sha256=row.approval_receipt_sha256,
+            authority_expires_at=_as_utc(row.authority_expires_at),
             item_results=results,
             runtime_snapshots=snapshots,
             write_count=len(results),

yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/review_scene_prompts.py

@@ -108,10 +108,16 @@ def _module_block(item):
     return f'<review-module id="{item["key"]}" version="{item["version"]}" sha256="{item["content_sha256"]}">\n{item["text"]}\n</review-module>'
 
 
-def scene_review_template(request) -> PromptTemplate:
-    from .prompts import contract_review_prompt_candidates
-    route = contract_review_prompt_candidates(request.contract_type, request.scene_template_id)[0]
+def scene_review_template_for_route(
+    route: str,
+    *,
+    review_profile: str,
+    scene_template_id: str | None = None,
+) -> PromptTemplate:
+    """Render one explicit review route without mutating global routing state."""
     modules = review_modules()
+    if route not in modules and route not in REVIEW_FOCUS:
+        raise PromptRenderError(f"未知专项审查路由：{route}")
     selected = [modules[route]] if route in modules else []
     focus = REVIEW_FOCUS.get(route, REVIEW_FOCUS["contract_review_default"])
     # One (name, version) must have one content identity in the shared registry.
@@ -123,7 +129,7 @@ def scene_review_template(request) -> PromptTemplate:
             + "其他场景不据此推定劳动关系或套用劳动专项结论。")
     # Lightweight manifest builders created before v3 only carry the route
     # fields; absence of an explicit profile must retain the historical v2 asset.
-    request_profile = getattr(request, "review_profile", "scene_evaluation")
+    request_profile = review_profile
     uses_compact_context = request_profile == "scene_evaluation_v4"
     is_v3 = request_profile in {"scene_evaluation_v3", "scene_evaluation_v4"}
     output_version = REVIEW_CONTRACT_V3_VERSION if is_v3 else REVIEW_CONTRACT_VERSION
@@ -160,6 +166,16 @@ def scene_review_template(request) -> PromptTemplate:
         content=content, status="builtin_evaluation", updated_at=datetime(2026, 8, 28, tzinfo=UTC))
 
 
+def scene_review_template(request) -> PromptTemplate:
+    from .prompts import contract_review_prompt_candidates
+    route = contract_review_prompt_candidates(request.contract_type, request.scene_template_id)[0]
+    return scene_review_template_for_route(
+        route,
+        review_profile=getattr(request, "review_profile", "scene_evaluation"),
+        scene_template_id=getattr(request, "scene_template_id", None),
+    )
+
+
 def capture_review_prompt_snapshot(store, request):
     from .prompts import contract_review_prompt_candidates
     if request.review_profile not in {"scene_evaluation", "scene_evaluation_v3", "scene_evaluation_v4"}:

yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/schema_check.py

@@ -137,6 +137,7 @@ RC06_COLUMNS_BY_TABLE = {
         "batch_manifest_raw_sha256",
         "batch_manifest_payload_sha256",
         "approval_receipt_sha256",
+        "authority_expires_at",
         "request_json",
         "before_prompts_json",
         "after_prompts_json",
@@ -171,6 +172,7 @@ RC06_CONSTRAINTS_BY_TABLE = {
         "ux_rc06_prompt_publications_idempotency",
         "ck_rc06_prompt_publications_state",
         "ck_rc06_prompt_publications_compensation_evidence",
+        "ck_rc06_prompt_publications_authority_window",
     ),
     "rc06_prompt_module_publications": (
         "ux_rc06_prompt_module_publications_transaction_key",

yuqei-ai-platform/services/ai-platform-api/tests/test_health.py

@@ -139,10 +139,14 @@ def test_rc06_schema_contract_includes_durable_compensation_evidence() -> None:
         "removed_item_count",
         "unchanged_item_count",
         "compensated_at",
+        "authority_expires_at",
     } <= set(RC06_COLUMNS_BY_TABLE["rc06_prompt_publications"])
     assert "ck_rc06_prompt_publications_compensation_evidence" in set(
         RC06_CONSTRAINTS_BY_TABLE["rc06_prompt_publications"]
     )
+    assert "ck_rc06_prompt_publications_authority_window" in set(
+        RC06_CONSTRAINTS_BY_TABLE["rc06_prompt_publications"]
+    )
 
 
 def test_runtime_schema_check_errors_for_missing_core_tables(tmp_path) -> None:

yuqei-ai-platform/services/ai-platform-api/tests/test_rc06_migrations.py

@@ -66,6 +66,15 @@ def _hardened_head(connection) -> None:
     )
 
 
+def _compensation_head(connection) -> None:
+    _hardened_head(connection)
+    _run(
+        connection,
+        _load("20261006_0031_add_rc06_compensation_evidence.py"),
+        "upgrade",
+    )
+
+
 def test_empty_old_head_upgrades_and_round_trips_without_rewriting_0029(tmp_path):
     engine = sa.create_engine(f"sqlite+pysqlite:///{(tmp_path / 'ai-migration.sqlite3').as_posix()}")
     migration = _load("20261006_0030_harden_rc06_prompt_publication.py")
@@ -294,6 +303,72 @@ def test_compensation_evidence_migration_rejects_active_committed_downgrade(tmp_
     engine.dispose()
 
 
+def test_authority_expiry_migration_requires_empty_journal_and_round_trips(tmp_path):
+    engine = sa.create_engine(
+        f"sqlite+pysqlite:///{(tmp_path / 'ai-authority-expiry.sqlite3').as_posix()}"
+    )
+    migration = _load("20261007_0032_bind_rc06_authority_expiry.py")
+    with engine.begin() as connection:
+        _compensation_head(connection)
+        _run(connection, migration, "upgrade")
+        inspector = sa.inspect(connection)
+        assert "authority_expires_at" in {
+            item["name"]
+            for item in inspector.get_columns("rc06_prompt_publications")
+        }
+        assert "ck_rc06_prompt_publications_authority_window" in {
+            item["name"]
+            for item in inspector.get_check_constraints("rc06_prompt_publications")
+        }
+        _run(connection, migration, "downgrade")
+        assert "authority_expires_at" not in {
+            item["name"]
+            for item in sa.inspect(connection).get_columns("rc06_prompt_publications")
+        }
+        _run(connection, migration, "upgrade")
+    engine.dispose()
+
+
+def test_authority_expiry_migration_rejects_nonempty_journal_before_schema_change(tmp_path):
+    engine = sa.create_engine(
+        f"sqlite+pysqlite:///{(tmp_path / 'ai-authority-nonempty.sqlite3').as_posix()}"
+    )
+    migration = _load("20261007_0032_bind_rc06_authority_expiry.py")
+    with engine.begin() as connection:
+        _compensation_head(connection)
+        now = datetime.now(UTC)
+        table = sa.Table(
+            "rc06_prompt_publications",
+            sa.MetaData(),
+            autoload_with=connection,
+        )
+        connection.execute(table.insert().values(
+            transaction_id="authority-migration-transaction",
+            idempotency_key="authority-migration-idempotency",
+            request_sha256="1" * 64,
+            candidate_commit="2" * 40,
+            release_image_digest="sha256:" + "3" * 64,
+            compensation_capability_sha256="4" * 64,
+            candidate_tag="authority-migration-candidate",
+            batch_manifest_raw_sha256="5" * 64,
+            batch_manifest_payload_sha256="6" * 64,
+            approval_receipt_sha256="7" * 64,
+            state="prepared",
+            request_json="{}",
+            before_prompts_json="[]",
+            created_at=now,
+            updated_at=now,
+        ))
+
+        with pytest.raises(RuntimeError, match="refuses nonempty publication journals"):
+            _run(connection, migration, "upgrade")
+        assert "authority_expires_at" not in {
+            item["name"]
+            for item in sa.inspect(connection).get_columns("rc06_prompt_publications")
+        }
+    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(
@@ -308,7 +383,7 @@ def test_fresh_alembic_upgrade_then_old_head_round_trip(tmp_path, monkeypatch):
         with engine.connect() as connection:
             assert connection.exec_driver_sql(
                 "SELECT version_num FROM ai_platform_alembic_version"
-            ).scalar_one() == "20261006_0031"
+            ).scalar_one() == "20261007_0032"
         engine.dispose()
 
         command.downgrade(config, "20261006_0029")
@@ -318,6 +393,7 @@ def test_fresh_alembic_upgrade_then_old_head_round_trip(tmp_path, monkeypatch):
             assert {
                 "sealed_at",
                 "compensation_evidence_sha256",
+                "authority_expires_at",
             } <= {
                 item["name"] for item in sa.inspect(connection).get_columns("rc06_prompt_publications")
             }

yuqei-ai-platform/services/ai-platform-api/tests/test_rc06_publication.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 import hashlib
 import json
-from datetime import UTC, datetime
+from datetime import UTC, datetime, timedelta
 
 import pytest
 from fastapi.testclient import TestClient
@@ -11,6 +11,7 @@ from sqlalchemy.dialects import postgresql, sqlite
 from sqlalchemy.schema import CreateTable
 
 from yuqei_sdk.prompt_source_baseline import DOMAIN_REVIEW_RUNTIME_IDENTITIES
+import yuqei_ai_platform_api.rc06_publication as rc06_publication
 from yuqei_ai_platform_api.config import AiPlatformSettings
 from yuqei_ai_platform_api.main import create_app
 from yuqei_ai_platform_api.models import (
@@ -85,6 +86,7 @@ def _request(transaction_id: str = "rc06-ai-transaction-0001") -> PromptPrepareR
         batch_manifest_raw_sha256="d" * 64,
         batch_manifest_payload_sha256="b" * 64,
         approval_receipt_sha256="c" * 64,
+        authority_expires_at=datetime.now(UTC) + timedelta(days=1),
         compensation_capability_sha256=hashlib.sha256(
             COMPENSATION_CAPABILITY.encode("utf-8")
         ).hexdigest(),
@@ -93,6 +95,65 @@ def _request(transaction_id: str = "rc06-ai-transaction-0001") -> PromptPrepareR
     )
 
 
+def test_prepare_rejects_expired_authority_without_journal_or_prompt_writes(tmp_path):
+    repository = _repository(tmp_path)
+    try:
+        _seed_domain_dependencies(repository)
+        service = _service(repository)
+        request = PromptPrepareRequest.model_validate({
+            **_request("rc06-ai-expired-prepare").model_dump(mode="python"),
+            "authority_expires_at": datetime.now(UTC) - timedelta(seconds=1),
+        })
+
+        with pytest.raises(Rc06Conflict, match="authority expired"):
+            service.prepare(request)
+
+        with repository._session_factory() as session:
+            assert session.scalar(select(Rc06PromptPublicationModel)) is None
+            assert session.scalars(select(Rc06PromptModulePublicationModel)).all() == []
+            assert {
+                row.name for row in session.scalars(select(PromptTemplateModel)).all()
+            } == set(DOMAIN_REVIEW_RUNTIME_IDENTITIES)
+    finally:
+        repository._engine.dispose()
+
+
+def test_commit_rolls_back_all_ai_writes_when_authority_expires_before_precommit(
+    tmp_path,
+    monkeypatch,
+):
+    repository = _repository(tmp_path)
+    try:
+        _seed_domain_dependencies(repository)
+        service = _service(repository)
+        request = _request("rc06-ai-expiry-crossing")
+        service.prepare(request)
+        expiry = request.authority_expires_at
+        readings = iter((expiry - timedelta(seconds=1), expiry))
+        monkeypatch.setattr(
+            rc06_publication,
+            "_database_wall_clock",
+            lambda _session: next(readings),
+        )
+
+        with pytest.raises(Rc06Conflict, match="authority expired"):
+            service.commit(request.transaction_id)
+
+        with repository._session_factory() as session:
+            transaction = session.get(Rc06PromptPublicationModel, request.transaction_id)
+            assert transaction.state == "prepared"
+            assert transaction.item_results_json is None
+            assert transaction.runtime_snapshots_json is None
+            modules = session.scalars(select(Rc06PromptModulePublicationModel)).all()
+            assert len(modules) == 14
+            assert {item.state for item in modules} == {"prepared"}
+            assert {
+                row.name for row in session.scalars(select(PromptTemplateModel)).all()
+            } == set(DOMAIN_REVIEW_RUNTIME_IDENTITIES)
+    finally:
+        repository._engine.dispose()
+
+
 def test_sqlite_prompt_participant_is_durable_idempotent_and_compensates(tmp_path):
     repository = _repository(tmp_path)
     try:
@@ -106,6 +167,7 @@ def test_sqlite_prompt_participant_is_durable_idempotent_and_compensates(tmp_pat
         assert committed.state == "committed"
         assert committed.write_count == 14
         assert committed.runtime_snapshot_count == 14
+        assert committed.authority_expires_at == request.authority_expires_at
         assert sum(item["source_kind"] == "composed_drafting_modules" for item in committed.runtime_snapshots) == 1
         assert sum(item["source_kind"] == "new_review_module_composite" for item in committed.runtime_snapshots) == 8
         assert sum(item["source_kind"] == "existing_domain_prompt_asset" for item in committed.runtime_snapshots) == 5
@@ -708,6 +770,8 @@ def test_prompt_journal_ddl_has_same_constraints_on_sqlite_and_postgresql():
         assert "compensation_evidence_sha256 VARCHAR(64)" in transaction_sql
         assert "restored_item_count INTEGER" in transaction_sql
         assert "ck_rc06_prompt_publications_compensation_evidence" in transaction_sql
+        assert "authority_expires_at" in transaction_sql
+        assert "ck_rc06_prompt_publications_authority_window" in transaction_sql
         assert "sealed" in transaction_sql
         assert "ux_rc06_prompt_module_publications_transaction_key" in module_sql
         assert "ck_rc06_prompt_module_state" in module_sql

yuqei-ai-platform/services/ai-platform-api/tests/test_scene_runtime_manifest.py

@@ -32,7 +32,11 @@ from yuqei_ai_platform_api.packaged_prompt_assets_import import (
     load_packaged_prompt_assets,
 )
 from yuqei_ai_platform_api.repository import PromptTemplate
-from yuqei_ai_platform_api.review_scene_prompts import review_modules, scene_review_template
+from yuqei_ai_platform_api.review_scene_prompts import (
+    review_modules,
+    scene_review_template,
+    scene_review_template_for_route,
+)
 from yuqei_ai_platform_api.scene_runtime_manifest import build_scene_runtime_manifest
 from yuqei_contract_api.contract_templates import get_contract_template
 from yuqei_ai_platform_api.scene_evaluation_gate import (
@@ -231,23 +235,13 @@ def test_prompt_source_catalog_is_recomputed_from_the_actual_module_composers_an
         for item in REVIEW_MODULES
     ] == list(REVIEW_MODULES)
 
-    route_inputs = {
-        "contract_review_trade_procurement": ("purchase", "scene_evaluation_v3"),
-        "contract_review_services": ("service", "scene_evaluation_v3"),
-        "contract_review_transport_warehousing": ("transport", "scene_evaluation_v4"),
-        "contract_review_manufacturing": ("manufacturing", "scene_evaluation_v4"),
-        "contract_review_construction": ("construction", "scene_evaluation_v3"),
-        "contract_review_marketing": ("marketing", "scene_evaluation_v4"),
-        "contract_review_consumer_services": ("consumer_service", "scene_evaluation_v3"),
-        "contract_review_notices_amendments": ("amendment", "scene_evaluation_v4"),
-    }
     recomputed_new_review = {}
-    for route, (contract_type, profile) in route_inputs.items():
-        template = scene_review_template(SimpleNamespace(
-            contract_type=contract_type,
-            scene_template_id=None,
-            review_profile=profile,
-        ))
+    for route in actual_review_modules:
+        template = scene_review_template_for_route(
+            route,
+            review_profile="scene_evaluation_v4",
+            scene_template_id="rc06-runtime-snapshot",
+        )
         assert template.name == route
         recomputed_new_review[route] = {
             "name": template.name,

yuqei-ai-sdk-python/src/yuqei_sdk/prompt_source_baseline.py

@@ -13,7 +13,7 @@ from .review_routing import REVIEW_MODULE_KEYS, SCENE_REVIEW_ROUTES
 
 
 BASELINE_SCHEMA_VERSION = "prompt-source-baseline.v1"
-BASELINE_VERSION = "rc05-current-source-20261006"
+BASELINE_VERSION = "rc05-current-source-20261007"
 
 DRAFTING_MODULES = (
     {"key": "draft_intake", "version": "tp06-22-20260909-1", "content_sha256": "64519f6e5e95d9f1408e8b3db4feb818dec7cb0386de7983f591947f1cb53983"},
@@ -46,13 +46,13 @@ DRAFT_RUNTIME_IDENTITY = {
 # A route has one release-target identity even when older evaluation evidence is
 # still bound to an earlier profile.  That evidence is not rewritten here.
 NEW_REVIEW_RUNTIME_IDENTITIES = {
-    "contract_review_trade_procurement": {"name": "contract_review_trade_procurement", "version": "tp06-2-20260831-1", "content_sha256": "71956dde40698f3a7488da8f8b8214db8ca7bc8ebeddda6c4b965d74185bb457", "source_kind": "new_review_module_composite"},
-    "contract_review_services": {"name": "contract_review_services", "version": "tp06-2-20260831-1", "content_sha256": "21ce149263a42f8cdd76f11800aedb66f5ee8e4b43cd5e41696092d96b31114d", "source_kind": "new_review_module_composite"},
+    "contract_review_trade_procurement": {"name": "contract_review_trade_procurement", "version": "tp06-8-20260909-1", "content_sha256": "abd850ec3a28b1c1bc1f636ba3690b84f4e7c5afb56cdd681a77d7d8eb548b2a", "source_kind": "new_review_module_composite"},
+    "contract_review_services": {"name": "contract_review_services", "version": "tp06-8-20260909-1", "content_sha256": "848d6b33eebf8b3a8bcee9330e26b8b8487fa0469ac2b9ad6790a3f1559711a3", "source_kind": "new_review_module_composite"},
     "contract_review_transport_warehousing": {"name": "contract_review_transport_warehousing", "version": "tp06-8-20260909-1", "content_sha256": "950d6a02476bf37e6c2a733710bdfdfefe474b44743bd9d7448bbf95c01c6b14", "source_kind": "new_review_module_composite"},
     "contract_review_manufacturing": {"name": "contract_review_manufacturing", "version": "tp06-8-20260909-1", "content_sha256": "97ba767d8d47166697d7c0790fa4e17cb7b7c72950e149d6bbf4bda9bfb3d5c4", "source_kind": "new_review_module_composite"},
-    "contract_review_construction": {"name": "contract_review_construction", "version": "tp06-2-20260831-1", "content_sha256": "444552139eeec9b210f652c132ffb1ea42697e28be0c8fe33e2b46a663238cbc", "source_kind": "new_review_module_composite"},
+    "contract_review_construction": {"name": "contract_review_construction", "version": "tp06-8-20260909-1", "content_sha256": "44f52d9d292232012e8c9712ac714ac8d00f65cff7e4ddb07f66b67076499d46", "source_kind": "new_review_module_composite"},
     "contract_review_marketing": {"name": "contract_review_marketing", "version": "tp06-8-20260909-1", "content_sha256": "261fb010dcab3d20f6f0751df941b3be11a1db06be5e4f2e43122ea2e538fcf5", "source_kind": "new_review_module_composite"},
-    "contract_review_consumer_services": {"name": "contract_review_consumer_services", "version": "tp06-2-20260831-1", "content_sha256": "c42da047fae64cf91cf544f3c694ef3ff99226dd8af0d4465b71f272d3b3f603", "source_kind": "new_review_module_composite"},
+    "contract_review_consumer_services": {"name": "contract_review_consumer_services", "version": "tp06-8-20260909-1", "content_sha256": "1d15d83ccb6fd247e459974d64acedc860b02188f0053bd79b5d0e6edab443dd", "source_kind": "new_review_module_composite"},
     "contract_review_notices_amendments": {"name": "contract_review_notices_amendments", "version": "tp06-8-20260909-1", "content_sha256": "19fbfc866d6a67c5da9aa18e55fcd239b3d8d9915a5fa1076e8befb64c64d6e3", "source_kind": "new_review_module_composite"},
 }
 

yuqei-infra/scripts/build-rc06-publication-batch.py

@@ -20,7 +20,6 @@ import stat
 import subprocess
 import sys
 import tempfile
-import types
 from datetime import UTC, datetime, timedelta
 from pathlib import Path, PurePath
 from typing import Any, Iterable, Mapping
@@ -798,7 +797,6 @@ def _load_executable_prompt_dependencies(repo: Path) -> list[dict[str, Any]]:
     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,
@@ -813,7 +811,7 @@ def _load_executable_prompt_dependencies(repo: Path) -> list[dict[str, Any]]:
         )
         from yuqei_ai_platform_api.review_scene_prompts import (  # type: ignore
             review_module_evidence,
-            scene_review_template,
+            scene_review_template_for_route,
         )
 
         drafting_evidence = drafting_module_evidence(DRAFT_PROMPT_CONTENT)
@@ -833,35 +831,27 @@ def _load_executable_prompt_dependencies(repo: Path) -> list[dict[str, Any]]:
                 "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]
+        for route in review_routes:
+            template = scene_review_template_for_route(
+                route,
+                review_profile="scene_evaluation_v4",
+                scene_template_id="rc06-runtime-snapshot",
+            )
+            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}"
                 )
-                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
+            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,
+                }
+            )
 
         packaged, _, duplicates = load_packaged_prompt_assets(repo / DEFAULT_PROMPT_PATH)
         if duplicates: