yuqei/ai-contract-platform
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

fix(rc06): harden publication saga evidence

Browse Source
This commit is contained in:
yuqei
2026-10-06 20:23:59 +08:00
parent adcba7c2ef
commit f49a26c61f
10 changed files with 1333 additions and 152 deletions
Download Patch File Download Diff File Expand all files Collapse all files
yuqei-ai-platform/services/ai-platform-api/migrations/versions/20261006_0029_add_rc06_prompt_publication_saga.py
+2
View File
@@ -22,6 +22,7 @@ def upgrade() -> None:
sa.Column("idempotency_key", sa.String(160), nullable=False),
sa.Column("request_sha256", sa.String(64), nullable=False),
sa.Column("candidate_commit", sa.String(64), nullable=False),
sa.Column("release_image_digest", sa.String(71), nullable=False),
sa.Column("candidate_tag", sa.String(128), nullable=False),
sa.Column("batch_manifest_raw_sha256", sa.String(64), nullable=False),
sa.Column("batch_manifest_payload_sha256", sa.String(64), nullable=False),
@@ -29,6 +30,7 @@ def upgrade() -> None:
sa.Column("state", sa.String(32), nullable=False),
sa.Column("request_json", sa.Text(), nullable=False),
sa.Column("before_prompts_json", sa.Text(), nullable=False),
sa.Column("after_prompts_json", sa.Text()),
sa.Column("item_results_json", sa.Text()),
sa.Column("runtime_snapshots_json", sa.Text()),
sa.Column("error_message", sa.Text()),
yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/config.py
+2
View File
@@ -46,6 +46,8 @@ class AiPlatformSettings(BaseSettings):
codex_cli_working_dir: str = ".runtime/codex-model-provider"
service_shared_secret: str = ""
rc06_publication_shared_secret: str = ""
rc06_frozen_candidate_commit: str = ""
rc06_release_image_digest: str = ""
provider_legal_attestation_secret: str = ""
# Ed25519 legal approval is the only supported identity for protected promotion.
# The HMAC setting above remains only for old, non-promotion fixtures.
yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/models.py
+2
View File
@@ -185,6 +185,7 @@ class Rc06PromptPublicationModel(Base):
idempotency_key: Mapped[str] = mapped_column(String(160), nullable=False)
request_sha256: Mapped[str] = mapped_column(String(64), nullable=False)
candidate_commit: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
release_image_digest: Mapped[str] = mapped_column(String(71), nullable=False)
candidate_tag: Mapped[str] = mapped_column(String(128), nullable=False)
batch_manifest_raw_sha256: Mapped[str] = mapped_column(String(64), nullable=False)
batch_manifest_payload_sha256: Mapped[str] = mapped_column(String(64), nullable=False)
@@ -192,6 +193,7 @@ class Rc06PromptPublicationModel(Base):
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)
after_prompts_json: Mapped[str | None] = mapped_column(Text, nullable=True)
item_results_json: Mapped[str | None] = mapped_column(Text, nullable=True)
runtime_snapshots_json: Mapped[str | None] = mapped_column(Text, nullable=True)
error_message: Mapped[str | None] = mapped_column(Text, nullable=True)
yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/rc06_publication.py
+264 -55
View File
@@ -20,7 +20,7 @@ from uuid import uuid4
from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel, ConfigDict, Field
from sqlalchemy import delete, select
from sqlalchemy import select
from yuqei_sdk.prompt_source_baseline import prompt_source_baseline
@@ -41,6 +41,7 @@ from .service_auth import is_valid_service_key
HEX64 = r"^[a-f0-9]{64}$"
CANDIDATE_COMMIT = r"^(?:[a-f0-9]{40}|[a-f0-9]{64})$"
IMAGE_DIGEST = r"^sha256:[a-f0-9]{64}$"
EXPECTED_MODULE_COUNT = 14
EXPECTED_RUNTIME_COUNT = 14
NEW_RUNTIME_COUNT = 9
@@ -94,6 +95,7 @@ class PromptPrepareRequest(BaseModel):
transaction_id: str = Field(pattern=r"^[A-Za-z0-9][A-Za-z0-9._:-]{7,63}$")
idempotency_key: str = Field(min_length=8, max_length=160)
candidate_commit: str = Field(pattern=CANDIDATE_COMMIT)
release_image_digest: str = Field(pattern=IMAGE_DIGEST)
candidate_tag: str = Field(min_length=1, max_length=128)
batch_manifest_raw_sha256: str = Field(pattern=HEX64)
batch_manifest_payload_sha256: str = Field(pattern=HEX64)
@@ -106,6 +108,7 @@ class PromptTransactionStatus(BaseModel):
transaction_id: str
state: str
candidate_commit: str
release_image_digest: str
candidate_tag: str
batch_manifest_raw_sha256: str
batch_manifest_payload_sha256: str
@@ -115,6 +118,7 @@ class PromptTransactionStatus(BaseModel):
write_count: int = 0
runtime_snapshot_count: int = 0
no_go: bool = False
committed_at: datetime | None = None
error: str | None = None
@@ -154,6 +158,34 @@ def _parse_time(value: str | None) -> datetime | None:
return datetime.fromisoformat(value) if value else None
def _as_utc(value: datetime | None) -> datetime | None:
if value is None:
return None
return value.replace(tzinfo=UTC) if value.tzinfo is None else value.astimezone(UTC)
def _prompt_rows(session: Any, names: list[str]) -> list[dict[str, Any]]:
rows = session.scalars(
select(PromptTemplateModel).where(PromptTemplateModel.name.in_(names))
).all()
return sorted((_prompt_row(row) for row in rows), key=lambda item: (item["name"], item["id"]))
def _frozen_request(transaction: Rc06PromptPublicationModel) -> PromptPrepareRequest:
try:
payload = PromptPrepareRequest.model_validate(_decode(transaction.request_json, {}))
except Exception as error:
raise Rc06PublicationError("Frozen RC06 Prompt request is invalid.") from error
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
or payload.release_image_digest != transaction.release_image_digest
):
raise Rc06PublicationError("Frozen RC06 Prompt deployment identity drifted.")
return payload
def _source_inventory() -> tuple[list[dict[str, Any]], list[dict[str, Any]], dict[str, str]]:
"""Recompute the batch from executable packaged sources on every prepare."""
drafting_path = Path(__file__).with_name("drafting_prompt_modules.json")
@@ -234,10 +266,31 @@ def _source_inventory() -> tuple[list[dict[str, Any]], list[dict[str, Any]], dic
class Rc06PromptPublicationService:
def __init__(self, session_factory: Any) -> None:
def __init__(
self,
session_factory: Any,
*,
frozen_candidate_commit: str = "",
release_image_digest: str = "",
) -> None:
self._session_factory = session_factory
self._frozen_candidate_commit = frozen_candidate_commit
self._release_image_digest = release_image_digest
def _assert_current_deployment(self, payload: PromptPrepareRequest) -> None:
if (
re.fullmatch(CANDIDATE_COMMIT, self._frozen_candidate_commit) is None
or re.fullmatch(IMAGE_DIGEST, self._release_image_digest) is None
):
raise Rc06PublicationError("RC06 AI deployment identity is not configured.")
if (
payload.candidate_commit != self._frozen_candidate_commit
or payload.release_image_digest != self._release_image_digest
):
raise Rc06Conflict("RC06 AI deployment identity does not match the frozen candidate.")
def prepare(self, payload: PromptPrepareRequest) -> PromptTransactionStatus:
self._assert_current_deployment(payload)
expected_modules, expected_runtime, runtime_contents = _source_inventory()
supplied_modules = sorted(
[item.model_dump(mode="json") for item in payload.modules],
@@ -261,6 +314,7 @@ class Rc06PromptPublicationService:
if existing is not None:
if existing.request_sha256 != request_sha256 or existing.transaction_id != payload.transaction_id:
raise Rc06Conflict("Idempotency key was already used for different RC06 Prompt material.")
_frozen_request(existing)
return self._status_row(existing)
if session.get(Rc06PromptPublicationModel, payload.transaction_id) is not None:
raise Rc06Conflict("Transaction ID was already used with a different idempotency key.")
@@ -274,6 +328,7 @@ class Rc06PromptPublicationService:
idempotency_key=payload.idempotency_key,
request_sha256=request_sha256,
candidate_commit=payload.candidate_commit,
release_image_digest=payload.release_image_digest,
candidate_tag=payload.candidate_tag,
batch_manifest_raw_sha256=payload.batch_manifest_raw_sha256,
batch_manifest_payload_sha256=payload.batch_manifest_payload_sha256,
@@ -281,6 +336,7 @@ class Rc06PromptPublicationService:
state="prepared",
request_json=_encode(request_payload),
before_prompts_json=_encode(before),
after_prompts_json=None,
item_results_json=None,
runtime_snapshots_json=None,
error_message=None,
@@ -308,23 +364,34 @@ class Rc06PromptPublicationService:
return self._status_row(transaction)
def commit(self, transaction_id: str) -> PromptTransactionStatus:
expected_modules, expected_runtime, runtime_contents = _source_inventory()
frozen = self._load_frozen_request(transaction_id)
expected_modules = sorted(
[item.model_dump(mode="json") for item in frozen.modules],
key=lambda item: (item["category"], item["key"]),
)
expected_runtime = sorted(
[item.model_dump(mode="json") for item in frozen.runtime_identities],
key=lambda item: item["name"],
)
installed_modules, installed_runtime, runtime_contents = _source_inventory()
if installed_modules != expected_modules or installed_runtime != expected_runtime:
raise Rc06Conflict("Installed Prompt source changed after RC06 prepare; commit refused.")
names = [item["name"] for item in expected_runtime]
try:
with prompt_write_transaction(self._session_factory, names) as session:
transaction = session.get(Rc06PromptPublicationModel, transaction_id)
if transaction is None:
raise KeyError(transaction_id)
if _frozen_request(transaction) != frozen:
raise Rc06PublicationError("Frozen RC06 Prompt request changed during commit.")
if transaction.state == "committed":
self._verify_committed(session, transaction, expected_runtime)
self._verify_committed(session, transaction, expected_runtime, expected_modules)
return self._status_row(transaction)
if transaction.state != "prepared":
raise Rc06Conflict(f"Prompt transaction cannot commit from {transaction.state}.")
self._assert_current_deployment(frozen)
before = _decode(transaction.before_prompts_json, [])
current_rows = session.scalars(
select(PromptTemplateModel).where(PromptTemplateModel.name.in_(names))
).all()
current = sorted((_prompt_row(row) for row in current_rows), key=lambda item: (item["name"], item["id"]))
current = _prompt_rows(session, names)
if current != before:
raise Rc06Conflict("Prompt state changed after prepare; commit refused.")
self._assert_domain_dependencies(session, expected_runtime)
@@ -337,26 +404,17 @@ class Rc06PromptPublicationService:
modules = session.scalars(select(Rc06PromptModulePublicationModel).where(
Rc06PromptModulePublicationModel.transaction_id == transaction_id
)).all()
if len(modules) != EXPECTED_MODULE_COUNT:
raise Rc06PublicationError("Prepared Prompt module journal is incomplete.")
self._verify_module_rows(modules, expected_modules, expected_state="prepared")
for module in modules:
module.state = "applied"
module.updated_at = now
snapshots = self._runtime_snapshots(session, expected_runtime)
item_results = [
{
"kind": "prompt_module",
"item_id": item["key"],
"expected_content_sha256": item["content_sha256"],
"online_content_sha256": item["content_sha256"],
"status": "applied_and_verified",
}
for item in expected_modules
]
item_results.sort(key=lambda item: item["item_id"])
item_results = self._module_results(modules, expected_modules)
after = _prompt_rows(session, names)
transaction.item_results_json = _encode(item_results)
transaction.runtime_snapshots_json = _encode(snapshots)
transaction.after_prompts_json = _encode(after)
transaction.state = "committed"
transaction.error_message = None
transaction.committed_at = now
@@ -369,26 +427,46 @@ class Rc06PromptPublicationService:
raise Rc06PublicationError(f"Prompt commit failed: {type(error).__name__}") from error
def compensate(self, transaction_id: str) -> PromptTransactionStatus:
_, expected_runtime, _ = _source_inventory()
frozen = self._load_frozen_request(transaction_id)
expected_modules = sorted(
[item.model_dump(mode="json") for item in frozen.modules],
key=lambda item: (item["category"], item["key"]),
)
expected_runtime = sorted(
[item.model_dump(mode="json") for item in frozen.runtime_identities],
key=lambda item: item["name"],
)
names = [item["name"] for item in expected_runtime]
try:
with prompt_write_transaction(self._session_factory, names) as session:
transaction = session.get(Rc06PromptPublicationModel, transaction_id)
if transaction is None:
raise KeyError(transaction_id)
if _frozen_request(transaction) != frozen:
raise Rc06PublicationError("Frozen RC06 Prompt request changed during compensation.")
if transaction.state == "compensated":
return self._status_row(transaction)
if transaction.state == "compensation_failed":
raise Rc06Conflict("Prompt compensation previously reached terminal NO_GO.")
if transaction.state == "committed":
self._verify_committed(session, transaction, expected_runtime)
self._restore_before_image(session, _decode(transaction.before_prompts_json, []), names)
self._verify_committed(session, transaction, expected_runtime, expected_modules)
self._restore_owned_changes(
session,
before=_decode(transaction.before_prompts_json, []),
after=_decode(transaction.after_prompts_json, []),
names=names,
)
elif transaction.state != "prepared":
raise Rc06Conflict(f"Prompt transaction cannot compensate from {transaction.state}.")
now = datetime.now(UTC)
modules = session.scalars(select(Rc06PromptModulePublicationModel).where(
Rc06PromptModulePublicationModel.transaction_id == transaction_id
)).all()
self._verify_module_rows(
modules,
expected_modules,
expected_state="applied" if transaction.state == "committed" else "prepared",
)
for module in modules:
module.state = "compensated"
module.updated_at = now
@@ -400,17 +478,31 @@ class Rc06PromptPublicationService:
transaction.updated_at = now
session.flush()
return self._status_row(transaction)
except (KeyError, Rc06Conflict):
except KeyError:
raise
except Exception as error:
self._mark_compensation_failed(transaction_id, error)
if isinstance(error, Rc06Conflict):
raise Rc06Conflict("Prompt compensation detected concurrent state drift; transaction is terminal NO_GO.") from error
raise Rc06PublicationError("Prompt compensation failed; transaction is terminal NO_GO.") from error
def status(self, transaction_id: str) -> PromptTransactionStatus:
with self._session_factory() as session:
frozen = self._load_frozen_request(transaction_id)
expected_modules = sorted(
[item.model_dump(mode="json") for item in frozen.modules],
key=lambda item: (item["category"], item["key"]),
)
expected_runtime = sorted(
[item.model_dump(mode="json") for item in frozen.runtime_identities],
key=lambda item: item["name"],
)
names = [item["name"] for item in expected_runtime]
with prompt_write_transaction(self._session_factory, names) as session:
row = session.get(Rc06PromptPublicationModel, transaction_id)
if row is None:
raise KeyError(transaction_id)
if row.state == "committed":
self._verify_committed(session, row, expected_runtime, expected_modules)
return self._status_row(row)
@staticmethod
@@ -492,10 +584,10 @@ class Rc06PromptPublicationService:
PromptTemplateModel.enabled.is_(True),
))
if row is None:
raise Rc06PublicationError(f"Runtime Prompt readback is missing: {identity['name']}")
raise Rc06Conflict(f"Runtime Prompt readback is missing: {identity['name']}")
online_hash = prompt_content_hash(row.content)
if online_hash != identity["content_sha256"]:
raise Rc06PublicationError(f"Runtime Prompt readback hash drift: {identity['name']}")
raise Rc06Conflict(f"Runtime Prompt readback hash drift: {identity['name']}")
snapshots.append({
**identity,
"prompt_template_id": row.id,
@@ -509,50 +601,161 @@ class Rc06PromptPublicationService:
raise Rc06PublicationError("Runtime Prompt readback count must be exactly 14.")
return snapshots
def _verify_committed(self, session: Any, transaction: Rc06PromptPublicationModel, identities: list[dict[str, Any]]) -> None:
def _verify_committed(
self,
session: Any,
transaction: Rc06PromptPublicationModel,
identities: list[dict[str, Any]],
modules: list[dict[str, Any]],
) -> None:
snapshots = self._runtime_snapshots(session, identities)
if snapshots != _decode(transaction.runtime_snapshots_json, []):
raise Rc06Conflict("Committed Prompt runtime state drifted; automatic compensation refused.")
rows = session.scalars(select(Rc06PromptModulePublicationModel).where(
Rc06PromptModulePublicationModel.transaction_id == transaction.transaction_id
)).all()
self._verify_module_rows(rows, modules, expected_state="applied")
if self._module_results(rows, modules) != _decode(transaction.item_results_json, []):
raise Rc06Conflict("Committed Prompt module readback drifted; automatic compensation refused.")
@staticmethod
def _restore_before_image(session: Any, before: list[dict[str, Any]], names: list[str]) -> None:
def _verify_module_rows(
rows: list[Rc06PromptModulePublicationModel],
expected: list[dict[str, Any]],
*,
expected_state: str,
) -> None:
expected_by_key = {item["key"]: item for item in expected}
if len(rows) != EXPECTED_MODULE_COUNT or len(expected_by_key) != EXPECTED_MODULE_COUNT:
raise Rc06PublicationError("RC06 Prompt module journal is incomplete.")
if len({row.module_key for row in rows}) != EXPECTED_MODULE_COUNT:
raise Rc06PublicationError("RC06 Prompt module journal contains duplicate keys.")
for row in rows:
item = expected_by_key.get(row.module_key)
if item is None or (
row.category != item["category"]
or row.version != item["version"]
or row.content != item["content"]
or row.content_sha256 != item["content_sha256"]
or row.module_payload_sha256 != item["module_payload_sha256"]
or prompt_content_hash(row.content) != item["content_sha256"]
or row.state != expected_state
):
raise Rc06Conflict(f"RC06 Prompt module journal drifted: {row.module_key}")
@classmethod
def _module_results(
cls,
rows: list[Rc06PromptModulePublicationModel],
expected: list[dict[str, Any]],
) -> list[dict[str, Any]]:
expected_by_key = {item["key"]: item for item in expected}
results = [
{
"kind": "prompt_module",
"item_id": row.module_key,
"expected_content_sha256": expected_by_key[row.module_key]["content_sha256"],
"online_content_sha256": prompt_content_hash(row.content),
"status": "applied_and_verified",
}
for row in rows
]
return sorted(results, key=lambda item: item["item_id"])
@staticmethod
def _restore_owned_changes(
session: Any,
*,
before: list[dict[str, Any]],
after: list[dict[str, Any]],
names: list[str],
) -> None:
if not after:
raise Rc06PublicationError("Committed Prompt after-image is missing.")
before_by_id = {item["id"]: item for item in before}
after_by_id = {item["id"]: item for item in after}
current = session.scalars(select(PromptTemplateModel).where(PromptTemplateModel.name.in_(names))).all()
current_by_id = {row.id: row for row in current}
touched_ids = {
row_id for row_id, snapshot in after_by_id.items()
if before_by_id.get(row_id) != snapshot
}
for row in current:
snapshot = before_by_id.get(row.id)
if snapshot is None:
if row.id not in after_by_id and (row.status == "published" or row.enabled):
raise Rc06Conflict(
f"A newer active Prompt appeared after RC06 commit: {row.name}:{row.version}"
)
now = datetime.now(UTC)
for row_id in sorted(touched_ids):
after_snapshot = after_by_id[row_id]
row = current_by_id.get(row_id)
before_snapshot = before_by_id.get(row_id)
if row is None:
if before_snapshot is not None:
raise Rc06Conflict(f"An RC06-modified Prompt was deleted concurrently: {row_id}")
continue
if _prompt_row(row) != after_snapshot:
raise Rc06Conflict(f"An RC06-owned Prompt changed after commit: {row.name}:{row.version}")
before_payload = prompt_template_from_model(row).model_dump(mode="json")
if before_snapshot is None:
tombstone = {
**before_payload,
"enabled": False,
"status": "deleted",
"updated_at": now.isoformat(),
}
session.add(PromptTemplateChangeModel(
id=f"prompt-change-{uuid4().hex[:12]}",
prompt_template_id=row.id,
name=row.name,
version=row.version,
operation="rc06_compensate",
before_template_json=encode_model_json(before_payload),
after_template_json=encode_model_json(tombstone),
changed_by="rc06-publication-saga",
change_note="Remove only the Prompt row created by the compensated RC06 transaction.",
created_at=now,
))
session.delete(row)
continue
row.name = snapshot["name"]
row.version = snapshot["version"]
row.content = snapshot["content"]
row.enabled = snapshot["enabled"]
row.status = snapshot["status"]
row.rollback_from_version = snapshot["rollback_from_version"]
row.published_at = _parse_time(snapshot["published_at"])
row.updated_at = _parse_time(snapshot["updated_at"]) or datetime.now(UTC)
existing_ids = {row.id for row in current}
for snapshot in before:
if snapshot["id"] in existing_ids:
continue
session.add(PromptTemplateModel(
id=snapshot["id"],
name=snapshot["name"],
version=snapshot["version"],
content=snapshot["content"],
enabled=snapshot["enabled"],
status=snapshot["status"],
rollback_from_version=snapshot["rollback_from_version"],
published_at=_parse_time(snapshot["published_at"]),
updated_at=_parse_time(snapshot["updated_at"]) or datetime.now(UTC),
row.name = before_snapshot["name"]
row.version = before_snapshot["version"]
row.content = before_snapshot["content"]
row.enabled = before_snapshot["enabled"]
row.status = before_snapshot["status"]
row.rollback_from_version = before_snapshot["rollback_from_version"]
row.published_at = _parse_time(before_snapshot["published_at"])
row.updated_at = _parse_time(before_snapshot["updated_at"]) or now
session.flush()
restored = prompt_template_from_model(row).model_dump(mode="json")
session.add(PromptTemplateChangeModel(
id=f"prompt-change-{uuid4().hex[:12]}",
prompt_template_id=row.id,
name=row.name,
version=row.version,
operation="rc06_compensate",
before_template_json=encode_model_json(before_payload),
after_template_json=encode_model_json(restored),
changed_by="rc06-publication-saga",
change_note="Restore the exact RC06 before-image after coordinated compensation.",
created_at=now,
))
session.flush()
def _load_frozen_request(self, transaction_id: str) -> PromptPrepareRequest:
with self._session_factory() as session:
row = session.get(Rc06PromptPublicationModel, transaction_id)
if row is None:
raise KeyError(transaction_id)
return _frozen_request(row)
def _mark_compensation_failed(self, transaction_id: str, error: Exception) -> None:
with self._session_factory.begin() as session:
row = session.get(Rc06PromptPublicationModel, transaction_id)
if row is None:
return
if row.state == "compensated":
return
row.state = "compensation_failed"
row.error_message = f"{type(error).__name__}: {error}"[:2000]
row.updated_at = datetime.now(UTC)
@@ -565,6 +768,7 @@ class Rc06PromptPublicationService:
transaction_id=row.transaction_id,
state=row.state,
candidate_commit=row.candidate_commit,
release_image_digest=row.release_image_digest,
candidate_tag=row.candidate_tag,
batch_manifest_raw_sha256=row.batch_manifest_raw_sha256,
batch_manifest_payload_sha256=row.batch_manifest_payload_sha256,
@@ -574,6 +778,7 @@ class Rc06PromptPublicationService:
write_count=len(results),
runtime_snapshot_count=len(snapshots),
no_go=row.state == "compensation_failed",
committed_at=_as_utc(row.committed_at),
error=row.error_message,
)
@@ -585,7 +790,11 @@ def register_rc06_publication_routes(
store: Any,
) -> None:
session_factory = getattr(store, "_session_factory", None)
service = Rc06PromptPublicationService(session_factory) if session_factory is not None else None
service = Rc06PromptPublicationService(
session_factory,
frozen_candidate_commit=settings.rc06_frozen_candidate_commit,
release_image_digest=settings.rc06_release_image_digest,
) if session_factory is not None else None
def authorized(candidate: str | None) -> None:
if not is_valid_service_key(candidate, settings.rc06_publication_shared_secret):
yuqei-ai-platform/services/ai-platform-api/tests/test_rc06_publication.py
+208 -5
View File
@@ -12,6 +12,7 @@ from yuqei_sdk.prompt_source_baseline import DOMAIN_REVIEW_RUNTIME_IDENTITIES
from yuqei_ai_platform_api.config import AiPlatformSettings
from yuqei_ai_platform_api.main import create_app
from yuqei_ai_platform_api.models import (
PromptTemplateChangeModel,
PromptTemplateModel,
Rc06PromptModulePublicationModel,
Rc06PromptPublicationModel,
@@ -31,6 +32,10 @@ from yuqei_ai_platform_api.rc06_publication import (
from yuqei_ai_platform_api.repository import SqlAlchemyAiPlatformRepository
CANDIDATE_COMMIT = "a" * 40
AI_IMAGE_DIGEST = "sha256:" + "e" * 64
def _repository(tmp_path):
return SqlAlchemyAiPlatformRepository(
f"sqlite:///{tmp_path / 'ai.sqlite3'}",
@@ -38,6 +43,15 @@ def _repository(tmp_path):
)
def _service(repository, **overrides):
options = {
"frozen_candidate_commit": CANDIDATE_COMMIT,
"release_image_digest": AI_IMAGE_DIGEST,
**overrides,
}
return Rc06PromptPublicationService(repository._session_factory, **options)
def _seed_domain_dependencies(repository) -> None:
assets, _, duplicates = load_packaged_prompt_assets(DEFAULT_SOURCE_FILE)
assert duplicates == []
@@ -62,7 +76,8 @@ def _request(transaction_id: str = "rc06-ai-transaction-0001") -> PromptPrepareR
return PromptPrepareRequest(
transaction_id=transaction_id,
idempotency_key=f"idempotency:{transaction_id}",
candidate_commit="a" * 40,
candidate_commit=CANDIDATE_COMMIT,
release_image_digest=AI_IMAGE_DIGEST,
candidate_tag="rc06-test-candidate",
batch_manifest_raw_sha256="d" * 64,
batch_manifest_payload_sha256="b" * 64,
@@ -76,7 +91,7 @@ def test_sqlite_prompt_participant_is_durable_idempotent_and_compensates(tmp_pat
repository = _repository(tmp_path)
try:
_seed_domain_dependencies(repository)
service = Rc06PromptPublicationService(repository._session_factory)
service = _service(repository)
request = _request()
assert service.prepare(request).state == "prepared"
@@ -109,7 +124,7 @@ def test_sqlite_prompt_participant_is_durable_idempotent_and_compensates(tmp_pat
def test_prepare_fails_closed_when_a_preexisting_domain_dependency_is_missing(tmp_path):
repository = _repository(tmp_path)
try:
service = Rc06PromptPublicationService(repository._session_factory)
service = _service(repository)
with pytest.raises(Rc06Conflict, match="dependency is unavailable"):
service.prepare(_request())
with repository._session_factory() as session:
@@ -118,11 +133,57 @@ def test_prepare_fails_closed_when_a_preexisting_domain_dependency_is_missing(tm
repository._engine.dispose()
@pytest.mark.parametrize(
("overrides", "message"),
[
({"frozen_candidate_commit": ""}, "not configured"),
({"release_image_digest": ""}, "not configured"),
({"frozen_candidate_commit": "b" * 40}, "does not match"),
({"release_image_digest": "sha256:" + "9" * 64}, "does not match"),
],
)
def test_prepare_fails_closed_for_missing_or_mismatched_deployment_identity(
tmp_path,
overrides,
message,
):
repository = _repository(tmp_path)
try:
service = _service(repository, **overrides)
with pytest.raises(Rc06PublicationError, match=message):
service.prepare(_request("rc06-ai-deployment-identity"))
with repository._session_factory() as session:
assert session.scalar(select(Rc06PromptPublicationModel)) is None
finally:
repository._engine.dispose()
def test_status_returns_persisted_deployment_identity_and_rejects_tampering(tmp_path):
repository = _repository(tmp_path)
try:
_seed_domain_dependencies(repository)
service = _service(repository)
request = _request("rc06-ai-deployment-identity-readback")
prepared = service.prepare(request)
assert prepared.candidate_commit == CANDIDATE_COMMIT
assert prepared.release_image_digest == AI_IMAGE_DIGEST
assert service.prepare(request) == prepared
with repository._session_factory.begin() as session:
session.get(Rc06PromptPublicationModel, request.transaction_id).release_image_digest = (
"sha256:" + "8" * 64
)
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:
_seed_domain_dependencies(repository)
service = Rc06PromptPublicationService(repository._session_factory)
service = _service(repository)
request = _request()
service.prepare(request)
service.commit(request.transaction_id)
@@ -130,7 +191,7 @@ def test_compensation_error_is_persisted_as_terminal_no_go(tmp_path, monkeypatch
def fail(*_args, **_kwargs):
raise RuntimeError("injected restore failure")
monkeypatch.setattr(service, "_restore_before_image", fail)
monkeypatch.setattr(service, "_restore_owned_changes", fail)
with pytest.raises(Rc06PublicationError, match="terminal NO_GO"):
service.compensate(request.transaction_id)
status = service.status(request.transaction_id)
@@ -151,6 +212,8 @@ def test_internal_fastapi_routes_require_constant_time_service_auth(tmp_path):
database_auto_create=True,
service_shared_secret="s" * 48,
rc06_publication_shared_secret="r" * 48,
rc06_frozen_candidate_commit=CANDIDATE_COMMIT,
rc06_release_image_digest=AI_IMAGE_DIGEST,
require_service_auth=False,
object_storage_local_dir=str(tmp_path / "objects"),
)
@@ -168,11 +231,151 @@ def test_internal_fastapi_routes_require_constant_time_service_auth(tmp_path):
repository._engine.dispose()
def test_commit_refuses_installed_source_drift_after_prepare(tmp_path, monkeypatch):
repository = _repository(tmp_path)
try:
_seed_domain_dependencies(repository)
service = _service(repository)
request = _request("rc06-ai-source-drift")
service.prepare(request)
import yuqei_ai_platform_api.rc06_publication as publication
original_inventory = publication._source_inventory
def drifted_inventory():
modules, identities, contents = original_inventory()
modules[0] = {**modules[0], "content": modules[0]["content"] + "\nunauthorized"}
return modules, identities, contents
monkeypatch.setattr(publication, "_source_inventory", drifted_inventory)
with pytest.raises(Rc06Conflict, match="changed after RC06 prepare"):
service.commit(request.transaction_id)
with repository._session_factory() as session:
assert session.get(Rc06PromptPublicationModel, request.transaction_id).state == "prepared"
finally:
repository._engine.dispose()
def test_commit_refuses_deployment_identity_drift_after_prepare(tmp_path):
repository = _repository(tmp_path)
try:
_seed_domain_dependencies(repository)
service = _service(repository)
request = _request("rc06-ai-deployment-drift")
service.prepare(request)
service._release_image_digest = "sha256:" + "7" * 64
with pytest.raises(Rc06Conflict, match="deployment identity does not match"):
service.commit(request.transaction_id)
assert service.status(request.transaction_id).state == "prepared"
finally:
repository._engine.dispose()
def test_committed_status_performs_live_runtime_and_module_readback(tmp_path):
repository = _repository(tmp_path)
try:
_seed_domain_dependencies(repository)
service = _service(repository)
request = _request("rc06-ai-live-readback")
service.prepare(request)
service.commit(request.transaction_id)
with repository._session_factory.begin() as session:
module = session.scalar(select(Rc06PromptModulePublicationModel).where(
Rc06PromptModulePublicationModel.transaction_id == request.transaction_id
))
module.content += "\ndrift"
with pytest.raises(Rc06Conflict, match="module journal drifted"):
service.status(request.transaction_id)
finally:
repository._engine.dispose()
def test_committed_status_rejects_live_runtime_prompt_drift(tmp_path):
repository = _repository(tmp_path)
try:
_seed_domain_dependencies(repository)
service = _service(repository)
request = _request("rc06-ai-runtime-readback")
service.prepare(request)
committed = service.commit(request.transaction_id)
target_id = next(
item["prompt_template_id"] for item in committed.runtime_snapshots
if item["source_kind"] == "new_review_module_composite"
)
with repository._session_factory.begin() as session:
session.get(PromptTemplateModel, target_id).content += "\nruntime drift"
with pytest.raises(Rc06Conflict, match="hash drift"):
service.status(request.transaction_id)
finally:
repository._engine.dispose()
def test_compensation_preserves_concurrent_draft_and_keeps_change_history(tmp_path):
repository = _repository(tmp_path)
try:
_seed_domain_dependencies(repository)
service = _service(repository)
request = _request("rc06-ai-preserve-concurrent-draft")
service.prepare(request)
committed = service.commit(request.transaction_id)
runtime_name = next(
item["name"] for item in committed.runtime_snapshots
if item["source_kind"] == "new_review_module_composite"
)
now = datetime.now(UTC)
with repository._session_factory.begin() as session:
session.add(PromptTemplateModel(
id="concurrent-draft-after-rc06",
name=runtime_name,
version="next-draft",
content="concurrent draft content",
enabled=False,
status="draft",
published_at=None,
updated_at=now,
))
assert service.compensate(request.transaction_id).state == "compensated"
with repository._session_factory() as session:
assert session.get(PromptTemplateModel, "concurrent-draft-after-rc06") is not None
changes = session.scalars(select(PromptTemplateChangeModel)).all()
assert sum(item.operation == "rc06_publish" for item in changes) == 9
assert sum(item.operation == "rc06_compensate" for item in changes) == 9
finally:
repository._engine.dispose()
def test_compensation_fails_closed_without_overwriting_concurrent_prompt_change(tmp_path):
repository = _repository(tmp_path)
try:
_seed_domain_dependencies(repository)
service = _service(repository)
request = _request("rc06-ai-concurrent-prompt-drift")
service.prepare(request)
committed = service.commit(request.transaction_id)
target_id = next(
item["prompt_template_id"] for item in committed.runtime_snapshots
if item["source_kind"] == "new_review_module_composite"
)
with repository._session_factory.begin() as session:
target = session.get(PromptTemplateModel, target_id)
target.content += "\nconcurrent edit"
with pytest.raises(Rc06Conflict, match="terminal NO_GO"):
service.compensate(request.transaction_id)
with repository._session_factory() as session:
assert session.get(PromptTemplateModel, target_id).content.endswith("concurrent edit")
assert session.get(Rc06PromptPublicationModel, request.transaction_id).state == "compensation_failed"
finally:
repository._engine.dispose()
def test_prompt_journal_ddl_has_same_constraints_on_sqlite_and_postgresql():
for dialect in (sqlite.dialect(), postgresql.dialect()):
transaction_sql = str(CreateTable(Rc06PromptPublicationModel.__table__).compile(dialect=dialect))
module_sql = str(CreateTable(Rc06PromptModulePublicationModel.__table__).compile(dialect=dialect))
assert "ux_rc06_prompt_publications_idempotency" in transaction_sql
assert "ck_rc06_prompt_publications_state" in transaction_sql
assert "release_image_digest VARCHAR(71) NOT NULL" in transaction_sql
assert "ux_rc06_prompt_module_publications_transaction_key" in module_sql
assert "ck_rc06_prompt_module_state" in module_sql
yuqei-contract-platform/services/contract-api/migrations/versions/20261006_0080_add_rc06_publication_saga.py
+11 -1
View File
@@ -23,6 +23,10 @@ def upgrade() -> None:
sa.Column("idempotency_key", sa.String(160), nullable=False),
sa.Column("request_sha256", sa.String(64), nullable=False),
sa.Column("candidate_commit", sa.String(64), nullable=False),
sa.Column("contract_candidate_commit", sa.String(64), nullable=False),
sa.Column("contract_release_image_digest", sa.String(71), nullable=False),
sa.Column("ai_candidate_commit", sa.String(64), nullable=False),
sa.Column("ai_release_image_digest", sa.String(71), nullable=False),
sa.Column("candidate_tag", sa.String(128), nullable=False),
sa.Column("batch_manifest_raw_sha256", sa.String(64), nullable=False),
sa.Column("batch_manifest_payload_sha256", sa.String(64), nullable=False),
@@ -35,9 +39,14 @@ def upgrade() -> None:
sa.Column("error_message", sa.Text()),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("contract_committed_at", sa.DateTime(timezone=True)),
sa.Column("committed_at", sa.DateTime(timezone=True)),
sa.Column("compensated_at", sa.DateTime(timezone=True)),
sa.UniqueConstraint("idempotency_key", name="ux_rc06_publication_sagas_idempotency"),
sa.UniqueConstraint(
"tenant_id",
"idempotency_key",
name="ux_rc06_publication_sagas_idempotency",
),
sa.CheckConstraint(
"state in ('preparing', 'prepared', 'ai_committed', 'contract_committed', "
"'committed', 'compensating', 'compensated', 'compensation_failed')",
@@ -62,6 +71,7 @@ def upgrade() -> None:
sa.Column("asset_payload_sha256", sa.String(64), nullable=False),
sa.Column("before_head_event_id", sa.String(64)),
sa.Column("lifecycle_event_id", sa.String(64)),
sa.Column("compensation_event_id", sa.String(64)),
sa.Column("snapshot_id", sa.String(64)),
sa.Column("online_content_sha256", sa.String(64)),
sa.Column("state", sa.String(32), nullable=False),
yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/config.py
+2
View File
@@ -26,6 +26,8 @@ class ContractApiSettings(BaseSettings):
ai_platform_rc06_publication_key: str = ""
ai_platform_timeout_seconds: float = Field(default=30.0, ge=1.0, le=600.0)
rc06_publication_shared_secret: str = ""
rc06_frozen_candidate_commit: str = ""
rc06_release_image_digest: str = ""
web_proxy_shared_secret: str = ""
require_web_proxy_auth: bool = False
playbook_direct_mutations_enabled: bool = False
yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/persistence/models.py
+11 -1
View File
@@ -1726,7 +1726,11 @@ class Rc06PublicationSagaModel(Base):
__tablename__ = "rc06_publication_sagas"
__table_args__ = (
UniqueConstraint("idempotency_key", name="ux_rc06_publication_sagas_idempotency"),
UniqueConstraint(
"tenant_id",
"idempotency_key",
name="ux_rc06_publication_sagas_idempotency",
),
CheckConstraint(
"state in ('preparing', 'prepared', 'ai_committed', 'contract_committed', "
"'committed', 'compensating', 'compensated', 'compensation_failed')",
@@ -1739,6 +1743,10 @@ class Rc06PublicationSagaModel(Base):
idempotency_key: Mapped[str] = mapped_column(String(160), nullable=False)
request_sha256: Mapped[str] = mapped_column(String(64), nullable=False)
candidate_commit: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
contract_candidate_commit: Mapped[str] = mapped_column(String(64), nullable=False)
contract_release_image_digest: Mapped[str] = mapped_column(String(71), nullable=False)
ai_candidate_commit: Mapped[str] = mapped_column(String(64), nullable=False)
ai_release_image_digest: Mapped[str] = mapped_column(String(71), nullable=False)
candidate_tag: Mapped[str] = mapped_column(String(128), nullable=False)
batch_manifest_raw_sha256: Mapped[str] = mapped_column(String(64), nullable=False)
batch_manifest_payload_sha256: Mapped[str] = mapped_column(String(64), nullable=False)
@@ -1751,6 +1759,7 @@ class Rc06PublicationSagaModel(Base):
error_message: Mapped[str | None] = mapped_column(Text, nullable=True)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
contract_committed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
committed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
compensated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
@@ -1779,6 +1788,7 @@ class Rc06TemplatePublicationItemModel(Base):
asset_payload_sha256: Mapped[str] = mapped_column(String(64), nullable=False)
before_head_event_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
lifecycle_event_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
compensation_event_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
snapshot_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
online_content_sha256: Mapped[str | None] = mapped_column(String(64), nullable=True)
state: Mapped[str] = mapped_column(String(32), nullable=False, index=True)
yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/rc06_publication.py
+427 -80
View File
@@ -8,6 +8,7 @@ from __future__ import annotations
import hashlib
import json
import re
from datetime import UTC, datetime
from threading import RLock
from typing import Any, Protocol
@@ -16,7 +17,7 @@ from uuid import uuid4
import httpx
from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel, ConfigDict, Field
from sqlalchemy import delete, select, text
from sqlalchemy import select, text
from yuqei_sdk.review_routing import REVIEW_ROUTING_VERSION, review_prompt_routes
@@ -38,6 +39,7 @@ from .tenant_context import get_tenant_context
HEX64 = r"^[a-f0-9]{64}$"
CANDIDATE_COMMIT = r"^(?:[a-f0-9]{40}|[a-f0-9]{64})$"
IMAGE_DIGEST = r"^sha256:[a-f0-9]{64}$"
EXPECTED_ASSET_COUNT = 60
EXPECTED_MODULE_COUNT = 14
EXPECTED_RUNTIME_COUNT = 14
@@ -59,6 +61,8 @@ class Rc06PrepareRequest(BaseModel):
transaction_id: str = Field(pattern=r"^[A-Za-z0-9][A-Za-z0-9._:-]{7,63}$")
idempotency_key: str = Field(min_length=8, max_length=160)
candidate_commit: str = Field(pattern=CANDIDATE_COMMIT)
contract_release_image_digest: str = Field(pattern=IMAGE_DIGEST)
ai_release_image_digest: str = Field(pattern=IMAGE_DIGEST)
candidate_tag: str = Field(min_length=1, max_length=128)
batch_manifest_raw_sha256: str = Field(pattern=HEX64)
batch_manifest_payload_sha256: str = Field(pattern=HEX64)
@@ -72,6 +76,10 @@ class Rc06SagaStatus(BaseModel):
transaction_id: str
state: str
candidate_commit: str
contract_candidate_commit: str
contract_release_image_digest: str
ai_candidate_commit: str
ai_release_image_digest: str
candidate_tag: str
batch_manifest_raw_sha256: str
batch_manifest_payload_sha256: str
@@ -82,6 +90,9 @@ class Rc06SagaStatus(BaseModel):
runtime_snapshot_count: int = 0
no_go: bool = True
cross_database_atomicity: bool = False
ai_committed_at: datetime | None = None
contract_committed_at: datetime | None = None
committed_at: datetime | None = None
error: str | None = None
@@ -166,6 +177,16 @@ def _decode(value: str | None, default: Any) -> Any:
return json.loads(value) if value else default
def _parse_time(value: Any) -> datetime | None:
if isinstance(value, datetime):
result = value
else:
result = datetime.fromisoformat(value) if isinstance(value, str) and value else None
if result is None:
return None
return result.replace(tzinfo=UTC) if result.tzinfo is None else result.astimezone(UTC)
def _finish_asset(asset: dict[str, Any]) -> dict[str, Any]:
asset["body_sha256"] = hashlib.sha256(asset["body_content"].encode("utf-8")).hexdigest()
asset["input_schema_sha256"] = _sha(asset["input_schema"]) if asset["input_schema"] is not None else None
@@ -241,6 +262,8 @@ class Rc06PublicationCoordinator:
repository: Any,
ai_client: AiRc06Client,
authorization_verifier: Rc06AuthorizationVerifier | None = None,
frozen_candidate_commit: str = "",
release_image_digest: str = "",
) -> None:
session_factory = getattr(repository, "_session_factory", None)
if session_factory is None:
@@ -249,8 +272,23 @@ class Rc06PublicationCoordinator:
self._session_factory = session_factory
self._ai = ai_client
self._authorization = authorization_verifier or FailClosedRc06AuthorizationVerifier()
self._frozen_candidate_commit = frozen_candidate_commit
self._release_image_digest = release_image_digest
def _assert_current_deployment(self, payload: Rc06PrepareRequest) -> None:
if (
re.fullmatch(CANDIDATE_COMMIT, self._frozen_candidate_commit) is None
or re.fullmatch(IMAGE_DIGEST, self._release_image_digest) is None
):
raise Rc06PublicationError("RC06 Contract deployment identity is not configured.")
if (
payload.candidate_commit != self._frozen_candidate_commit
or payload.contract_release_image_digest != self._release_image_digest
):
raise Rc06Conflict("RC06 Contract deployment identity does not match the frozen candidate.")
def prepare(self, payload: Rc06PrepareRequest, *, tenant_id: str) -> Rc06SagaStatus:
self._assert_current_deployment(payload)
self._authorization.verify(payload, tenant_id=tenant_id)
installed = contract_source_assets()
supplied = sorted(payload.assets, key=lambda item: item.get("template_id", ""))
@@ -266,6 +304,7 @@ class Rc06PublicationCoordinator:
with _coordinator_lock, self._session_factory.begin() as session:
self._lock_saga(session, payload.transaction_id)
existing = session.scalar(select(Rc06PublicationSagaModel).where(
Rc06PublicationSagaModel.tenant_id == tenant_id,
Rc06PublicationSagaModel.idempotency_key == payload.idempotency_key
))
if existing is not None:
@@ -276,6 +315,7 @@ class Rc06PublicationCoordinator:
):
raise Rc06Conflict("Idempotency key was already used for different RC06 material.")
if existing.state != "preparing":
self._frozen_request(existing)
return self._status_row(existing)
else:
if session.get(Rc06PublicationSagaModel, payload.transaction_id) is not None:
@@ -286,6 +326,10 @@ class Rc06PublicationCoordinator:
idempotency_key=payload.idempotency_key,
request_sha256=request_sha256,
candidate_commit=payload.candidate_commit,
contract_candidate_commit=payload.candidate_commit,
contract_release_image_digest=payload.contract_release_image_digest,
ai_candidate_commit=payload.candidate_commit,
ai_release_image_digest=payload.ai_release_image_digest,
candidate_tag=payload.candidate_tag,
batch_manifest_raw_sha256=payload.batch_manifest_raw_sha256,
batch_manifest_payload_sha256=payload.batch_manifest_payload_sha256,
@@ -298,13 +342,21 @@ class Rc06PublicationCoordinator:
error_message=None,
created_at=now,
updated_at=now,
contract_committed_at=None,
committed_at=None,
compensated_at=None,
)
session.add(existing)
for asset in installed:
head = self._latest_event(session, tenant_id, asset["template_id"])
if head is not None or asset["source_lifecycle_status"] != "internal_trial":
restored_boundary = (
head is not None
and head.action == "rollback"
and head.status == "internal_trial"
and (head.event_metadata or {}).get("rc06_compensation") is True
and (head.event_metadata or {}).get("rc06_restored_boundary") is True
)
if (head is not None and not restored_boundary) or asset["source_lifecycle_status"] != "internal_trial":
raise Rc06Conflict(
f"Template is no longer at the approved internal-trial boundary: {asset['template_id']}"
)
@@ -315,8 +367,9 @@ class Rc06PublicationCoordinator:
template_id=asset["template_id"],
expected_content_sha256=asset["content_sha256"],
asset_payload_sha256=asset["asset_payload_sha256"],
before_head_event_id=None,
before_head_event_id=head.id if head is not None else None,
lifecycle_event_id=None,
compensation_event_id=None,
snapshot_id=None,
online_content_sha256=None,
state="prepared",
@@ -324,15 +377,21 @@ class Rc06PublicationCoordinator:
updated_at=now,
))
ai_payload = self._ai_prepare_payload(payload)
ai_payload = self._ai_prepare_payload(payload, tenant_id=tenant_id)
try:
ai_status = self._ai.prepare(ai_payload)
if ai_status.get("state") != "prepared" or ai_status.get("transaction_id") != payload.transaction_id:
if (
ai_status.get("state") != "prepared"
or ai_status.get("transaction_id") != payload.transaction_id
or ai_status.get("candidate_commit") != payload.candidate_commit
or ai_status.get("release_image_digest") != payload.ai_release_image_digest
):
raise Rc06PublicationError("AI participant did not enter prepared state.")
except Exception as error:
self._record_error(payload.transaction_id, error)
raise
with _coordinator_lock, self._session_factory.begin() as session:
self._lock_saga(session, payload.transaction_id)
row = self._require_saga(session, payload.transaction_id)
if row.state == "preparing":
row.state = "prepared"
@@ -341,13 +400,17 @@ class Rc06PublicationCoordinator:
row.updated_at = datetime.now(UTC)
return self._status_row(row)
def commit(self, transaction_id: str) -> Rc06SagaStatus:
def commit(self, transaction_id: str, *, tenant_id: str | None = None) -> Rc06SagaStatus:
try:
row = self._load_saga(transaction_id)
row = self._load_saga(transaction_id, tenant_id=tenant_id)
if row.state == "committed":
return self._verify_final(row)
return self._verify_final(transaction_id)
if row.state not in {"prepared", "ai_committed", "contract_committed"}:
raise Rc06Conflict(f"RC06 saga cannot commit from {row.state}.")
if row.state in {"prepared", "ai_committed"}:
self._assert_current_deployment(
Rc06PrepareRequest.model_validate(self._frozen_request(row))
)
if row.state == "prepared":
# The previous coordinator process may have crashed after the AI
# database committed but before its local journal advanced. Ask
@@ -365,40 +428,59 @@ class Rc06PublicationCoordinator:
# it by the durable transaction identity before deciding
# whether the saga must compensate.
ai_status = self._ai.status(transaction_id)
self._assert_ai_committed(ai_status, transaction_id)
self._assert_ai_committed(
ai_status,
transaction_id,
self._frozen_request(row),
)
self._advance_ai_committed(transaction_id, ai_status)
row = self._load_saga(transaction_id)
row = self._load_saga(transaction_id, tenant_id=tenant_id)
if row.state == "ai_committed":
self._commit_templates(transaction_id)
row = self._load_saga(transaction_id)
row = self._load_saga(transaction_id, tenant_id=tenant_id)
if row.state == "contract_committed":
ai_status = self._ai.status(transaction_id)
self._assert_ai_committed(ai_status, transaction_id)
self._assert_ai_committed(
ai_status,
transaction_id,
self._frozen_request(row),
)
self._finalize(transaction_id, ai_status)
return self._verify_final(self._load_saga(transaction_id))
return self._verify_final(transaction_id)
except KeyError:
raise
except Exception as error:
try:
state = self._load_saga(transaction_id).state
if state in {"prepared", "ai_committed", "contract_committed"}:
self.compensate(transaction_id)
state = self._load_saga(transaction_id, tenant_id=tenant_id).state
if state in {"prepared", "ai_committed"}:
self.compensate(transaction_id, tenant_id=tenant_id)
elif state == "contract_committed":
# Both participants have durably committed. A failed live
# readback/finalize is evidence-pending NO_GO, not authority
# to undo 60 + 14 already-committed publication writes.
self._record_error(transaction_id, error)
except Exception:
pass
if isinstance(error, Rc06Conflict):
raise
raise Rc06PublicationError(f"RC06 commit failed: {type(error).__name__}") from error
def compensate(self, transaction_id: str) -> Rc06SagaStatus:
row = self._load_saga(transaction_id)
if row.state == "compensated":
return self._status_row(row)
if row.state == "compensation_failed":
raise Rc06Conflict("RC06 compensation previously reached terminal NO_GO.")
if row.state == "committed":
raise Rc06Conflict("A verified completed RC06 saga cannot be compensated automatically.")
def compensate(self, transaction_id: str, *, tenant_id: str | None = None) -> Rc06SagaStatus:
with _coordinator_lock, self._session_factory.begin() as session:
self._lock_saga(session, transaction_id)
current = self._require_saga(session, transaction_id)
self._assert_tenant(current, tenant_id)
self._frozen_request(current)
if current.state == "compensated":
return self._status_row(current)
if current.state == "compensation_failed":
raise Rc06Conflict("RC06 compensation previously reached terminal NO_GO.")
if current.state in {"contract_committed", "committed"}:
raise Rc06Conflict(
"RC06 participants already committed; automatic compensation is forbidden."
)
if current.state not in {"preparing", "prepared", "ai_committed", "compensating"}:
raise Rc06Conflict(f"RC06 saga cannot compensate from {current.state}.")
current.state = "compensating"
current.updated_at = datetime.now(UTC)
errors: list[str] = []
@@ -411,9 +493,20 @@ class Rc06PublicationCoordinator:
if ai_status.get("state") != "compensated":
raise Rc06PublicationError("AI participant did not confirm compensation.")
except Exception as error:
errors.append(f"ai:{type(error).__name__}:{error}")
try:
ai_status = self._ai.status(transaction_id)
if ai_status.get("state") != "compensated":
raise Rc06PublicationError("AI participant compensation outcome is not durable.")
except Exception as status_error:
errors.append(
f"ai:{type(error).__name__}:{error};"
f"status:{type(status_error).__name__}:{status_error}"
)
with _coordinator_lock, self._session_factory.begin() as session:
self._lock_saga(session, transaction_id)
current = self._require_saga(session, transaction_id)
if current.state == "compensated":
return self._status_row(current)
now = datetime.now(UTC)
current.state = "compensation_failed" if errors else "compensated"
current.error_message = "; ".join(errors)[:4000] if errors else None
@@ -424,8 +517,9 @@ class Rc06PublicationCoordinator:
raise Rc06PublicationError("RC06 compensation failed; saga is terminal NO_GO.")
return result
def status(self, transaction_id: str) -> Rc06SagaStatus:
return self._status_row(self._load_saga(transaction_id))
def status(self, transaction_id: str, *, tenant_id: str | None = None) -> Rc06SagaStatus:
row = self._load_saga(transaction_id, tenant_id=tenant_id)
return self._verify_final(transaction_id) if row.state == "committed" else self._status_row(row)
def _commit_templates(self, transaction_id: str) -> None:
with _coordinator_lock, self._session_factory.begin() as session:
@@ -435,7 +529,7 @@ class Rc06PublicationCoordinator:
return
if saga.state != "ai_committed":
raise Rc06Conflict(f"Contract participant cannot commit from {saga.state}.")
request = _decode(saga.request_json, {})
request = self._frozen_request(saga)
assets = {item["template_id"]: item for item in request["assets"]}
items = session.scalars(select(Rc06TemplatePublicationItemModel).where(
Rc06TemplatePublicationItemModel.transaction_id == transaction_id
@@ -443,12 +537,26 @@ class Rc06PublicationCoordinator:
if len(items) != EXPECTED_ASSET_COUNT:
raise Rc06PublicationError("Contract participant journal does not contain 60 items.")
now = datetime.now(UTC)
for index, item in enumerate(sorted(items, key=lambda value: value.template_id)):
for item in sorted(items, key=lambda value: value.template_id):
self._lock_template(session, saga.tenant_id, item.template_id)
head = self._latest_event(session, saga.tenant_id, item.template_id)
if (head.id if head else None) != item.before_head_event_id:
raise Rc06Conflict(f"Template lifecycle changed after prepare: {item.template_id}")
asset = assets[item.template_id]
previous_publications = session.scalars(select(TemplateLifecycleEventModel).where(
TemplateLifecycleEventModel.tenant_id == saga.tenant_id,
TemplateLifecycleEventModel.template_id == item.template_id,
TemplateLifecycleEventModel.version == asset["version"],
TemplateLifecycleEventModel.status == "published",
)).all()
if any(
(previous.event_metadata or {}).get("publication_content_sha256")
!= asset["content_sha256"]
for previous in previous_publications
):
raise Rc06Conflict(
f"Immutable template version has different historical content: {item.template_id}"
)
event_id = f"rc06-event-{hashlib.sha256(f'{transaction_id}:{item.template_id}'.encode()).hexdigest()[:32]}"
snapshot_id = f"rc06-snapshot-{hashlib.sha256(f'{transaction_id}:{item.template_id}'.encode()).hexdigest()[:29]}"
event = TemplateLifecycleEventModel(
@@ -499,47 +607,33 @@ class Rc06PublicationCoordinator:
item.updated_at = now
saga.state = "contract_committed"
saga.error_message = None
saga.contract_committed_at = now
saga.updated_at = now
def _finalize(self, transaction_id: str, ai_status: dict[str, Any]) -> None:
with _coordinator_lock, self._session_factory.begin() as session:
self._lock_saga(session, transaction_id)
saga = self._require_saga(session, transaction_id)
if saga.state == "committed":
return
if saga.state != "contract_committed":
raise Rc06Conflict(f"RC06 saga cannot finalize from {saga.state}.")
request = _decode(saga.request_json, {})
expected = {item["template_id"]: item for item in request["assets"]}
items = session.scalars(select(Rc06TemplatePublicationItemModel).where(
Rc06TemplatePublicationItemModel.transaction_id == transaction_id
)).all()
asset_results = []
for item in items:
event = session.get(TemplateLifecycleEventModel, item.lifecycle_event_id)
snapshot = session.get(TemplatePublishedSnapshotModel, item.snapshot_id)
asset = expected[item.template_id]
if (
event is None
or snapshot is None
or event.status != "published"
or event.event_metadata.get("publication_content_sha256") != asset["content_sha256"]
or snapshot.body_content != asset["body_content"]
or item.online_content_sha256 != asset["content_sha256"]
):
raise Rc06PublicationError(f"Template online hash readback failed: {item.template_id}")
asset_results.append({
"kind": "asset",
"item_id": item.template_id,
"expected_content_sha256": asset["content_sha256"],
"online_content_sha256": item.online_content_sha256,
"status": "applied_and_verified",
})
prompt_results = ai_status.get("item_results")
snapshots = ai_status.get("runtime_snapshots")
if not isinstance(prompt_results, list) or len(prompt_results) != EXPECTED_MODULE_COUNT:
raise Rc06PublicationError("AI participant did not return 14 module apply results.")
if not isinstance(snapshots, list) or len(snapshots) != EXPECTED_RUNTIME_COUNT:
raise Rc06PublicationError("AI participant did not return 14 runtime snapshots.")
request = self._frozen_request(saga)
self._assert_ai_committed(ai_status, transaction_id, request)
persisted_ai_status = _decode(saga.ai_status_json, {})
if (
ai_status.get("candidate_commit") != persisted_ai_status.get("candidate_commit")
or ai_status.get("release_image_digest")
!= persisted_ai_status.get("release_image_digest")
or _parse_time(ai_status.get("committed_at"))
!= _parse_time(persisted_ai_status.get("committed_at"))
):
raise Rc06PublicationError("AI participant durable commit identity drifted.")
for asset in sorted(request["assets"], key=lambda value: value["template_id"]):
self._lock_template(session, saga.tenant_id, asset["template_id"])
asset_results = self._verify_contract_readback(session, saga, request)
prompt_results = ai_status["item_results"]
snapshots = ai_status["runtime_snapshots"]
results = sorted(asset_results + prompt_results, key=lambda item: (item["kind"], item["item_id"]))
if len(results) != EXPECTED_WRITE_COUNT or any(item.get("status") != "applied_and_verified" for item in results):
raise Rc06PublicationError("RC06 online apply readback is incomplete.")
@@ -554,22 +648,86 @@ class Rc06PublicationCoordinator:
def _compensate_templates(self, transaction_id: str) -> None:
with _coordinator_lock, self._session_factory.begin() as session:
self._lock_saga(session, transaction_id)
saga = self._require_saga(session, transaction_id)
items = session.scalars(select(Rc06TemplatePublicationItemModel).where(
Rc06TemplatePublicationItemModel.transaction_id == transaction_id
)).all()
if len(items) != EXPECTED_ASSET_COUNT:
raise Rc06PublicationError("Contract participant journal does not contain 60 items.")
now = datetime.now(UTC)
for item in sorted(items, key=lambda value: value.template_id):
self._lock_template(session, saga.tenant_id, item.template_id)
if item.state == "applied":
head = self._latest_event(session, saga.tenant_id, item.template_id)
if head is None or head.id != item.lifecycle_event_id:
raise Rc06Conflict(f"Template advanced after RC06 commit: {item.template_id}")
session.execute(delete(TemplatePublishedSnapshotModel).where(
TemplatePublishedSnapshotModel.id == item.snapshot_id
))
session.execute(delete(TemplateLifecycleEventModel).where(
TemplateLifecycleEventModel.id == item.lifecycle_event_id
event = session.get(TemplateLifecycleEventModel, item.lifecycle_event_id)
snapshot = session.get(TemplatePublishedSnapshotModel, item.snapshot_id)
if (
event is None
or event.tenant_id != saga.tenant_id
or event.template_id != item.template_id
or event.deleted_at is not None
or (event.event_metadata or {}).get("rc06_transaction_id") != transaction_id
or snapshot is None
or snapshot.tenant_id != saga.tenant_id
or snapshot.template_id != item.template_id
or snapshot.published_event_id != event.id
or snapshot.deleted_at is not None
):
raise Rc06Conflict(f"RC06-owned template evidence drifted: {item.template_id}")
snapshot.deleted_at = now
snapshot.updated_by = "rc06-publication-saga"
restored_head = (
session.get(TemplateLifecycleEventModel, item.before_head_event_id)
if item.before_head_event_id is not None
else None
)
if item.before_head_event_id is not None and restored_head is None:
raise Rc06Conflict(f"RC06 before-boundary is missing: {item.template_id}")
if restored_head is not None and (
restored_head.tenant_id != saga.tenant_id
or restored_head.template_id != item.template_id
or restored_head.action != "rollback"
or restored_head.status != "internal_trial"
or (restored_head.event_metadata or {}).get("rc06_compensation") is not True
or (restored_head.event_metadata or {}).get("rc06_restored_boundary") is not True
):
raise Rc06Conflict(f"RC06 before-boundary drifted: {item.template_id}")
restored_status = restored_head.status if restored_head is not None else "internal_trial"
restored_version = restored_head.version if restored_head is not None else event.version
compensation_event_id = (
"rc06-compensation-"
+ hashlib.sha256(f"{transaction_id}:{item.template_id}".encode()).hexdigest()[:28]
)
session.add(TemplateLifecycleEventModel(
id=compensation_event_id,
tenant_id=saga.tenant_id,
template_id=item.template_id,
action="rollback",
status=restored_status,
version=restored_version,
actor_name="rc06-publication-saga",
summary="RC06 publication compensated; source boundary restored",
event_metadata={
"rc06_transaction_id": transaction_id,
"rc06_compensation": True,
"rc06_restored_boundary": True,
"compensated_lifecycle_event_id": event.id,
"compensated_snapshot_id": snapshot.id,
"restored_head_event_id": item.before_head_event_id,
"restored_status": restored_status,
"restored_version": restored_version,
"batch_manifest_payload_sha256": saga.batch_manifest_payload_sha256,
"approval_receipt_sha256": saga.approval_receipt_sha256,
},
created_at=now,
created_by="rc06-publication-saga",
updated_by="rc06-publication-saga",
deleted_at=None,
))
item.compensation_event_id = compensation_event_id
item.state = "compensated"
item.online_content_sha256 = None
item.updated_at = now
@@ -577,46 +735,213 @@ class Rc06PublicationCoordinator:
saga.runtime_snapshots_json = None
saga.updated_at = now
def _verify_final(self, row: Rc06PublicationSagaModel) -> Rc06SagaStatus:
status = self._status_row(row)
if status.state != "committed" or status.write_count != EXPECTED_WRITE_COUNT or status.runtime_snapshot_count != EXPECTED_RUNTIME_COUNT:
raise Rc06PublicationError("Committed RC06 saga has incomplete durable evidence.")
return status
def _verify_final(self, transaction_id: str) -> Rc06SagaStatus:
ai_status = self._ai.status(transaction_id)
with _coordinator_lock, self._session_factory.begin() as session:
self._lock_saga(session, transaction_id)
row = self._require_saga(session, transaction_id)
if row.state != "committed":
raise Rc06PublicationError("RC06 saga is not committed.")
request = self._frozen_request(row)
self._assert_ai_committed(ai_status, transaction_id, request)
persisted_ai_status = _decode(row.ai_status_json, {})
if (
ai_status.get("candidate_commit") != persisted_ai_status.get("candidate_commit")
or ai_status.get("release_image_digest")
!= persisted_ai_status.get("release_image_digest")
or _parse_time(ai_status.get("committed_at"))
!= _parse_time(persisted_ai_status.get("committed_at"))
):
raise Rc06PublicationError("AI participant durable commit identity drifted.")
for asset in sorted(request["assets"], key=lambda value: value["template_id"]):
self._lock_template(session, row.tenant_id, asset["template_id"])
asset_results = self._verify_contract_readback(session, row, request)
results = sorted(
asset_results + ai_status["item_results"],
key=lambda item: (item["kind"], item["item_id"]),
)
snapshots = ai_status["runtime_snapshots"]
if (
results != _decode(row.item_results_json, [])
or snapshots != _decode(row.runtime_snapshots_json, [])
or len(results) != EXPECTED_WRITE_COUNT
or len(snapshots) != EXPECTED_RUNTIME_COUNT
):
raise Rc06PublicationError("Committed RC06 saga has incomplete or stale durable evidence.")
return self._status_row(row)
@staticmethod
def _assert_ai_committed(payload: dict[str, Any], transaction_id: str) -> None:
def _assert_ai_committed(
payload: dict[str, Any],
transaction_id: str,
request: dict[str, Any],
) -> None:
expected_results = sorted((
{
"kind": "prompt_module",
"item_id": item["key"],
"expected_content_sha256": item["content_sha256"],
"online_content_sha256": item["content_sha256"],
"status": "applied_and_verified",
}
for item in request["prompt_modules"]
), key=lambda item: item["item_id"])
expected_runtime = {item["name"]: item for item in request["runtime_identities"]}
snapshots = payload.get("runtime_snapshots")
snapshot_by_name = {
item.get("name"): item for item in snapshots
} if isinstance(snapshots, list) else {}
if (
payload.get("transaction_id") != transaction_id
or payload.get("state") != "committed"
or payload.get("candidate_commit") != request["candidate_commit"]
or payload.get("release_image_digest") != request["ai_release_image_digest"]
or payload.get("candidate_tag") != request["candidate_tag"]
or payload.get("batch_manifest_raw_sha256") != request["batch_manifest_raw_sha256"]
or payload.get("batch_manifest_payload_sha256") != request["batch_manifest_payload_sha256"]
or payload.get("approval_receipt_sha256") != request["approval_receipt_sha256"]
or not payload.get("committed_at")
or payload.get("write_count") != EXPECTED_MODULE_COUNT
or payload.get("runtime_snapshot_count") != EXPECTED_RUNTIME_COUNT
or payload.get("item_results") != expected_results
or len(snapshot_by_name) != EXPECTED_RUNTIME_COUNT
or set(snapshot_by_name) != set(expected_runtime)
):
raise Rc06PublicationError("AI participant commit/readback is incomplete.")
for name, expected in expected_runtime.items():
actual = snapshot_by_name[name]
if (
any(actual.get(key) != expected[key] for key in (
"name", "version", "content_sha256", "source_kind"
))
or actual.get("online_content_sha256") != expected["content_sha256"]
or actual.get("status") != "published"
or actual.get("enabled") is not True
or actual.get("verified") is not True
or not isinstance(actual.get("prompt_template_id"), str)
or not actual["prompt_template_id"]
):
raise Rc06PublicationError(f"AI runtime Prompt readback drifted: {name}")
@staticmethod
def _verify_contract_readback(
session: Any,
saga: Rc06PublicationSagaModel,
request: dict[str, Any],
) -> list[dict[str, Any]]:
expected = {item["template_id"]: item for item in request["assets"]}
items = session.scalars(select(Rc06TemplatePublicationItemModel).where(
Rc06TemplatePublicationItemModel.transaction_id == saga.transaction_id
)).all()
if len(items) != EXPECTED_ASSET_COUNT or set(expected) != {item.template_id for item in items}:
raise Rc06PublicationError("Contract participant journal does not contain the exact 60 assets.")
results = []
for item in items:
event = session.get(TemplateLifecycleEventModel, item.lifecycle_event_id)
snapshot = session.get(TemplatePublishedSnapshotModel, item.snapshot_id)
asset = expected[item.template_id]
metadata = event.event_metadata if event is not None else {}
if (
item.tenant_id != saga.tenant_id
or item.state != "applied"
or item.expected_content_sha256 != asset["content_sha256"]
or item.asset_payload_sha256 != asset["asset_payload_sha256"]
or item.online_content_sha256 != asset["content_sha256"]
or event is None
or event.deleted_at is not None
or event.tenant_id != saga.tenant_id
or event.template_id != item.template_id
or event.action != "publish"
or event.status != "published"
or event.version != asset["version"]
or metadata.get("rc06_transaction_id") != saga.transaction_id
or metadata.get("batch_manifest_payload_sha256") != saga.batch_manifest_payload_sha256
or metadata.get("approval_receipt_sha256") != saga.approval_receipt_sha256
or metadata.get("asset_payload_sha256") != asset["asset_payload_sha256"]
or metadata.get("publication_content_sha256") != asset["content_sha256"]
or snapshot is None
or snapshot.deleted_at is not None
or snapshot.tenant_id != saga.tenant_id
or snapshot.template_id != item.template_id
or snapshot.version != asset["version"]
or snapshot.published_event_id != event.id
or snapshot.body_content != asset["body_content"]
or snapshot.required_variables != [value["name"] for value in asset["required_variables"]]
or snapshot.optional_variables != [value["name"] for value in asset["optional_variables"]]
or snapshot.validation_rules != asset["validation_rules"]
or snapshot.input_schema != asset["input_schema"]
or snapshot.clause_ids != asset["clause_ids"]
or snapshot.review_rule_ids != asset["review_rule_ids"]
):
raise Rc06PublicationError(f"Template online readback failed: {item.template_id}")
results.append({
"kind": "asset",
"item_id": item.template_id,
"expected_content_sha256": asset["content_sha256"],
"online_content_sha256": item.online_content_sha256,
"status": "applied_and_verified",
})
return results
def _advance_ai_committed(self, transaction_id: str, status: dict[str, Any]) -> None:
with _coordinator_lock, self._session_factory.begin() as session:
self._lock_saga(session, transaction_id)
row = self._require_saga(session, transaction_id)
if row.state == "prepared":
row.state = "ai_committed"
row.ai_status_json = _encode(status)
row.error_message = None
row.updated_at = datetime.now(UTC)
elif row.state != "ai_committed":
raise Rc06Conflict(f"RC06 saga changed state during AI commit: {row.state}")
def _record_error(self, transaction_id: str, error: Exception) -> None:
with self._session_factory.begin() as session:
with _coordinator_lock, self._session_factory.begin() as session:
self._lock_saga(session, transaction_id)
row = session.get(Rc06PublicationSagaModel, transaction_id)
if row is not None:
row.error_message = f"{type(error).__name__}: {error}"[:4000]
row.updated_at = datetime.now(UTC)
def _load_saga(self, transaction_id: str) -> Rc06PublicationSagaModel:
def _load_saga(
self,
transaction_id: str,
*,
tenant_id: str | None = None,
) -> Rc06PublicationSagaModel:
with self._session_factory() as session:
row = session.get(Rc06PublicationSagaModel, transaction_id)
if row is None:
raise KeyError(transaction_id)
self._assert_tenant(row, tenant_id)
self._frozen_request(row)
session.expunge(row)
return row
@staticmethod
def _assert_tenant(row: Rc06PublicationSagaModel, tenant_id: str | None) -> None:
if tenant_id is not None and row.tenant_id != tenant_id:
raise KeyError(row.transaction_id)
@staticmethod
def _frozen_request(row: Rc06PublicationSagaModel) -> dict[str, Any]:
try:
request = Rc06PrepareRequest.model_validate(_decode(row.request_json, {}))
except Exception as error:
raise Rc06PublicationError("Frozen RC06 Contract request is invalid.") from error
payload = request.model_dump(mode="json")
if _sha(payload) != row.request_sha256:
raise Rc06PublicationError("Frozen RC06 Contract request hash drifted.")
if (
row.candidate_commit != request.candidate_commit
or row.contract_candidate_commit != request.candidate_commit
or row.contract_release_image_digest != request.contract_release_image_digest
or row.ai_candidate_commit != request.candidate_commit
or row.ai_release_image_digest != request.ai_release_image_digest
):
raise Rc06PublicationError("Frozen RC06 participant deployment identity drifted.")
return payload
@staticmethod
def _require_saga(session: Any, transaction_id: str) -> Rc06PublicationSagaModel:
row = session.get(Rc06PublicationSagaModel, transaction_id)
@@ -639,9 +964,15 @@ class Rc06PublicationCoordinator:
@staticmethod
def _lock_saga(session: Any, transaction_id: str) -> None:
if session.get_bind().dialect.name == "postgresql":
dialect = session.get_bind().dialect.name
if dialect == "postgresql":
key = int.from_bytes(hashlib.sha256(f"rc06-saga:{transaction_id}".encode()).digest()[:8], "big", signed=True)
session.execute(text("SELECT pg_advisory_xact_lock(:key)"), {"key": key})
elif dialect == "sqlite":
# Acquire SQLite's database writer reservation before any saga or
# lifecycle read so the observed before-boundary cannot race a
# writer in another connection/process.
session.execute(text("BEGIN IMMEDIATE"))
@staticmethod
def _lock_template(session: Any, tenant_id: str, template_id: str) -> None:
@@ -654,11 +985,14 @@ class Rc06PublicationCoordinator:
session.execute(text("SELECT pg_advisory_xact_lock(:key)"), {"key": key})
@staticmethod
def _ai_prepare_payload(payload: Rc06PrepareRequest) -> dict[str, Any]:
def _ai_prepare_payload(payload: Rc06PrepareRequest, *, tenant_id: str) -> dict[str, Any]:
return {
"transaction_id": payload.transaction_id,
"idempotency_key": f"ai:{hashlib.sha256(payload.idempotency_key.encode()).hexdigest()}",
"idempotency_key": "ai:" + hashlib.sha256(
f"{tenant_id}\0{payload.idempotency_key}".encode()
).hexdigest(),
"candidate_commit": payload.candidate_commit,
"release_image_digest": payload.ai_release_image_digest,
"candidate_tag": payload.candidate_tag,
"batch_manifest_raw_sha256": payload.batch_manifest_raw_sha256,
"batch_manifest_payload_sha256": payload.batch_manifest_payload_sha256,
@@ -671,10 +1005,15 @@ class Rc06PublicationCoordinator:
def _status_row(row: Rc06PublicationSagaModel) -> Rc06SagaStatus:
results = _decode(row.item_results_json, [])
snapshots = _decode(row.runtime_snapshots_json, [])
ai_status = _decode(row.ai_status_json, {})
return Rc06SagaStatus(
transaction_id=row.transaction_id,
state=row.state,
candidate_commit=row.candidate_commit,
contract_candidate_commit=row.contract_candidate_commit,
contract_release_image_digest=row.contract_release_image_digest,
ai_candidate_commit=row.ai_candidate_commit,
ai_release_image_digest=row.ai_release_image_digest,
candidate_tag=row.candidate_tag,
batch_manifest_raw_sha256=row.batch_manifest_raw_sha256,
batch_manifest_payload_sha256=row.batch_manifest_payload_sha256,
@@ -685,6 +1024,9 @@ class Rc06PublicationCoordinator:
runtime_snapshot_count=len(snapshots),
no_go=row.state != "committed",
cross_database_atomicity=False,
ai_committed_at=_parse_time(ai_status.get("committed_at")),
contract_committed_at=_parse_time(row.contract_committed_at),
committed_at=_parse_time(row.committed_at),
error=row.error_message,
)
@@ -705,6 +1047,8 @@ def register_rc06_publication_routes(
service_key=settings.ai_platform_rc06_publication_key,
timeout=settings.ai_platform_timeout_seconds,
),
frozen_candidate_commit=settings.rc06_frozen_candidate_commit,
release_image_digest=settings.rc06_release_image_digest,
)
except Rc06PublicationError:
coordinator = None
@@ -736,14 +1080,17 @@ def register_rc06_publication_routes(
@app.post(f"{path}/{{transaction_id}}/commit", response_model=Rc06SagaStatus, tags=["internal", "rc06"])
def commit(transaction_id: str, x_rc06_publication_service_key: str | None = Header(default=None)):
authorized(x_rc06_publication_service_key)
return invoke(lambda: coordinator.commit(transaction_id))
tenant_id = get_tenant_context() or "default"
return invoke(lambda: coordinator.commit(transaction_id, tenant_id=tenant_id))
@app.get(f"{path}/{{transaction_id}}", response_model=Rc06SagaStatus, tags=["internal", "rc06"])
def status(transaction_id: str, x_rc06_publication_service_key: str | None = Header(default=None)):
authorized(x_rc06_publication_service_key)
return invoke(lambda: coordinator.status(transaction_id))
tenant_id = get_tenant_context() or "default"
return invoke(lambda: coordinator.status(transaction_id, tenant_id=tenant_id))
@app.post(f"{path}/{{transaction_id}}/compensate", response_model=Rc06SagaStatus, tags=["internal", "rc06"])
def compensate(transaction_id: str, x_rc06_publication_service_key: str | None = Header(default=None)):
authorized(x_rc06_publication_service_key)
return invoke(lambda: coordinator.compensate(transaction_id))
tenant_id = get_tenant_context() or "default"
return invoke(lambda: coordinator.compensate(transaction_id, tenant_id=tenant_id))
yuqei-contract-platform/services/contract-api/tests/test_rc06_publication_saga.py
+404 -10
View File
@@ -40,6 +40,11 @@ from yuqei_contract_api.rc06_publication import (
from yuqei_contract_api.sqlalchemy_contract_repository import SqlAlchemyContractRepository
CANDIDATE_COMMIT = "a" * 40
CONTRACT_IMAGE_DIGEST = "sha256:" + "e" * 64
AI_IMAGE_DIGEST = "sha256:" + "f" * 64
class TestAuthorizationVerifier:
"""Explicit test-only authority; never constructed by the FastAPI app."""
@@ -67,6 +72,28 @@ class LocalAiClient:
return self.service.compensate(transaction_id).model_dump(mode="json")
def _ai_service(repository):
return Rc06PromptPublicationService(
repository._session_factory,
frozen_candidate_commit=CANDIDATE_COMMIT,
release_image_digest=AI_IMAGE_DIGEST,
)
def _coordinator(*, repository, ai_client, authorization_verifier=None, **overrides):
options = {
"frozen_candidate_commit": CANDIDATE_COMMIT,
"release_image_digest": CONTRACT_IMAGE_DIGEST,
**overrides,
}
return Rc06PublicationCoordinator(
repository=repository,
ai_client=ai_client,
authorization_verifier=authorization_verifier or TestAuthorizationVerifier(),
**options,
)
def _repositories(tmp_path):
ai = SqlAlchemyAiPlatformRepository(f"sqlite:///{tmp_path / 'ai.sqlite3'}", auto_create=True)
contract = SqlAlchemyContractRepository(
@@ -99,7 +126,9 @@ def _request(transaction_id="rc06-saga-transaction-0001"):
return Rc06PrepareRequest(
transaction_id=transaction_id,
idempotency_key=f"idempotency:{transaction_id}",
candidate_commit="a" * 40,
candidate_commit=CANDIDATE_COMMIT,
contract_release_image_digest=CONTRACT_IMAGE_DIGEST,
ai_release_image_digest=AI_IMAGE_DIGEST,
candidate_tag="rc06-test-candidate",
batch_manifest_raw_sha256="d" * 64,
batch_manifest_payload_sha256="b" * 64,
@@ -113,8 +142,8 @@ def _request(transaction_id="rc06-saga-transaction-0001"):
def test_sqlite_saga_commits_exactly_74_items_and_14_runtime_snapshots(tmp_path):
contract, ai = _repositories(tmp_path)
try:
ai_service = Rc06PromptPublicationService(ai._session_factory)
coordinator = Rc06PublicationCoordinator(
ai_service = _ai_service(ai)
coordinator = _coordinator(
repository=contract,
ai_client=LocalAiClient(ai_service),
authorization_verifier=TestAuthorizationVerifier(),
@@ -129,9 +158,20 @@ def test_sqlite_saga_commits_exactly_74_items_and_14_runtime_snapshots(tmp_path)
assert committed.runtime_snapshot_count == 14
assert committed.no_go is False
assert committed.cross_database_atomicity is False
assert committed.contract_candidate_commit == CANDIDATE_COMMIT
assert committed.contract_release_image_digest == CONTRACT_IMAGE_DIGEST
assert committed.ai_candidate_commit == CANDIDATE_COMMIT
assert committed.ai_release_image_digest == AI_IMAGE_DIGEST
assert committed.ai_committed_at is not None
assert committed.contract_committed_at is not None
assert committed.committed_at is not None
assert sum(item["kind"] == "asset" for item in committed.item_results) == 60
assert sum(item["kind"] == "prompt_module" for item in committed.item_results) == 14
assert coordinator.commit(request.transaction_id) == committed
visible = contract.list_templates()
assert visible.total == 60
assert sum(item.lifecycle_status == "published" for item in visible.items) == 60
assert sum(contract.template_drafting_allowed(item) for item in visible.items) == 60
with contract._session_factory() as session:
assert len(session.scalars(select(Rc06TemplatePublicationItemModel)).all()) == 60
@@ -146,7 +186,7 @@ def test_sqlite_saga_commits_exactly_74_items_and_14_runtime_snapshots(tmp_path)
def test_lost_ai_commit_response_is_reconciled_by_transaction_status(tmp_path):
contract, ai = _repositories(tmp_path)
try:
service = Rc06PromptPublicationService(ai._session_factory)
service = _ai_service(ai)
class LostResponseClient(LocalAiClient):
lost = False
@@ -158,7 +198,7 @@ def test_lost_ai_commit_response_is_reconciled_by_transaction_status(tmp_path):
raise OSError("injected response loss after durable AI commit")
return result
coordinator = Rc06PublicationCoordinator(
coordinator = _coordinator(
repository=contract,
ai_client=LostResponseClient(service),
authorization_verifier=TestAuthorizationVerifier(),
@@ -172,10 +212,258 @@ def test_lost_ai_commit_response_is_reconciled_by_transaction_status(tmp_path):
contract.engine.dispose()
def test_lost_ai_compensation_response_is_reconciled_by_transaction_status(tmp_path):
contract, ai = _repositories(tmp_path)
try:
service = _ai_service(ai)
class LostCompensationResponseClient(LocalAiClient):
lost = False
def commit(self, transaction_id):
result = super().commit(transaction_id)
now = datetime.now(UTC)
template_id = contract_source_assets()[0]["template_id"]
with contract._session_factory.begin() as session:
session.add(TemplateLifecycleEventModel(
id="injected-lost-compensation-response-event",
tenant_id="default",
template_id=template_id,
action="offline",
status="offline",
version="concurrent-v1",
actor_name="concurrent-writer",
summary="fault injection",
event_metadata={},
created_at=now,
created_by="concurrent-writer",
updated_by="concurrent-writer",
deleted_at=None,
))
return result
def compensate(self, transaction_id):
result = super().compensate(transaction_id)
if not self.lost:
self.lost = True
raise OSError("injected response loss after durable AI compensation")
return result
coordinator = _coordinator(
repository=contract,
ai_client=LostCompensationResponseClient(service),
)
request = _request("rc06-saga-lost-compensation-response")
coordinator.prepare(request, tenant_id="default")
with pytest.raises(Rc06Conflict, match="changed after prepare"):
coordinator.commit(request.transaction_id)
assert coordinator.status(request.transaction_id).state == "compensated"
assert service.status(request.transaction_id).state == "compensated"
finally:
ai._engine.dispose()
contract.engine.dispose()
def test_finalize_drift_preserves_participant_commits_and_retry_finalizes(tmp_path):
contract, ai = _repositories(tmp_path)
try:
service = _ai_service(ai)
class OneShotDriftedCommittedStatus(LocalAiClient):
drifted = False
def status(self, transaction_id):
result = super().status(transaction_id)
if result["state"] == "committed" and not self.drifted:
self.drifted = True
result["runtime_snapshots"] = [dict(item) for item in result["runtime_snapshots"]]
result["runtime_snapshots"][0]["online_content_sha256"] = "f" * 64
return result
coordinator = _coordinator(
repository=contract,
ai_client=OneShotDriftedCommittedStatus(service),
authorization_verifier=TestAuthorizationVerifier(),
)
request = _request("rc06-saga-live-readback-drift")
coordinator.prepare(request, tenant_id="default")
with pytest.raises(Rc06PublicationError, match="RC06 commit failed"):
coordinator.commit(request.transaction_id)
pending = coordinator.status(request.transaction_id)
assert pending.state == "contract_committed"
assert pending.no_go is True
assert service.status(request.transaction_id).state == "committed"
visible = contract.list_templates()
assert visible.total == 60
assert sum(item.lifecycle_status == "published" for item in visible.items) == 60
assert sum(contract.template_drafting_allowed(item) for item in visible.items) == 60
with contract._session_factory() as session:
events = session.scalars(select(TemplateLifecycleEventModel)).all()
snapshots = session.scalars(select(TemplatePublishedSnapshotModel)).all()
items = session.scalars(select(Rc06TemplatePublicationItemModel)).all()
assert len(events) == 60
assert all(item.action == "publish" for item in events)
assert len(snapshots) == 60
assert all(item.deleted_at is None for item in snapshots)
assert all(item.lifecycle_event_id and item.compensation_event_id is None for item in items)
with pytest.raises(Rc06Conflict, match="already committed"):
coordinator.compensate(request.transaction_id)
committed = coordinator.commit(request.transaction_id)
assert committed.state == "committed"
assert committed.write_count == 74
with contract._session_factory() as session:
assert len(session.scalars(select(TemplateLifecycleEventModel)).all()) == 60
assert len(session.scalars(select(TemplatePublishedSnapshotModel)).all()) == 60
finally:
ai._engine.dispose()
contract.engine.dispose()
def test_transient_ai_status_failure_after_both_commits_never_rolls_back(tmp_path):
contract, ai = _repositories(tmp_path)
try:
service = _ai_service(ai)
class OneShotCommittedStatusOutage(LocalAiClient):
failed = False
def status(self, transaction_id):
result = super().status(transaction_id)
if result["state"] == "committed" and not self.failed:
self.failed = True
raise OSError("injected transient AI status outage")
return result
coordinator = _coordinator(
repository=contract,
ai_client=OneShotCommittedStatusOutage(service),
authorization_verifier=TestAuthorizationVerifier(),
)
request = _request("rc06-saga-transient-ai-status")
coordinator.prepare(request, tenant_id="default")
with pytest.raises(Rc06PublicationError, match="RC06 commit failed"):
coordinator.commit(request.transaction_id)
pending = coordinator.status(request.transaction_id)
assert pending.state == "contract_committed"
assert pending.no_go is True
assert service.status(request.transaction_id).state == "committed"
with contract._session_factory() as session:
assert len(session.scalars(select(TemplateLifecycleEventModel)).all()) == 60
snapshots = session.scalars(select(TemplatePublishedSnapshotModel)).all()
assert len(snapshots) == 60
assert all(item.deleted_at is None for item in snapshots)
committed = coordinator.commit(request.transaction_id)
assert committed.state == "committed"
with contract._session_factory() as session:
assert len(session.scalars(select(TemplateLifecycleEventModel)).all()) == 60
assert len(session.scalars(select(TemplatePublishedSnapshotModel)).all()) == 60
finally:
ai._engine.dispose()
contract.engine.dispose()
def test_lost_finalize_response_is_reconciled_without_duplicate_writes(tmp_path, monkeypatch):
contract, ai = _repositories(tmp_path)
try:
service = _ai_service(ai)
coordinator = _coordinator(
repository=contract,
ai_client=LocalAiClient(service),
authorization_verifier=TestAuthorizationVerifier(),
)
original_finalize = coordinator._finalize
lost = False
def finalize_then_lose_response(transaction_id, ai_status):
nonlocal lost
original_finalize(transaction_id, ai_status)
if not lost:
lost = True
raise OSError("injected response loss after durable coordinator finalize")
monkeypatch.setattr(coordinator, "_finalize", finalize_then_lose_response)
request = _request("rc06-saga-lost-finalize-response")
coordinator.prepare(request, tenant_id="default")
with pytest.raises(Rc06PublicationError, match="RC06 commit failed"):
coordinator.commit(request.transaction_id)
assert coordinator.status(request.transaction_id).state == "committed"
assert coordinator.commit(request.transaction_id).state == "committed"
assert service.status(request.transaction_id).state == "committed"
with contract._session_factory() as session:
assert len(session.scalars(select(TemplateLifecycleEventModel)).all()) == 60
assert len(session.scalars(select(TemplatePublishedSnapshotModel)).all()) == 60
finally:
ai._engine.dispose()
contract.engine.dispose()
def test_internal_contract_recovery_restores_exact_prior_boundary_append_only(tmp_path):
contract, ai = _repositories(tmp_path)
try:
service = _ai_service(ai)
client = LocalAiClient(service)
coordinator = _coordinator(
repository=contract,
ai_client=client,
authorization_verifier=TestAuthorizationVerifier(),
)
def commit_then_invoke_internal_recovery(request):
coordinator.prepare(request, tenant_id="default")
ai_status = client.commit(request.transaction_id)
coordinator._advance_ai_committed(request.transaction_id, ai_status)
coordinator._commit_templates(request.transaction_id)
assert coordinator.status(request.transaction_id).state == "contract_committed"
with pytest.raises(Rc06Conflict, match="already committed"):
coordinator.compensate(request.transaction_id)
# This private primitive is deliberately not exposed by the current
# state machine. It preserves the append-only recovery semantics
# for a future separately-authorized operator workflow.
coordinator._compensate_templates(request.transaction_id)
service.compensate(request.transaction_id)
first = _request("rc06-saga-first-compensation")
commit_then_invoke_internal_recovery(first)
second = _request("rc06-saga-second-compensation")
commit_then_invoke_internal_recovery(second)
template_id = contract_source_assets()[0]["template_id"]
with contract._session_factory() as session:
first_item = session.scalar(select(Rc06TemplatePublicationItemModel).where(
Rc06TemplatePublicationItemModel.transaction_id == first.transaction_id,
Rc06TemplatePublicationItemModel.template_id == template_id,
))
second_item = session.scalar(select(Rc06TemplatePublicationItemModel).where(
Rc06TemplatePublicationItemModel.transaction_id == second.transaction_id,
Rc06TemplatePublicationItemModel.template_id == template_id,
))
first_boundary = session.get(TemplateLifecycleEventModel, first_item.compensation_event_id)
second_boundary = session.get(TemplateLifecycleEventModel, second_item.compensation_event_id)
assert second_item.before_head_event_id == first_boundary.id
assert second_boundary.status == first_boundary.status
assert second_boundary.version == first_boundary.version
assert second_boundary.event_metadata["restored_head_event_id"] == first_boundary.id
assert second_boundary.event_metadata["restored_status"] == first_boundary.status
assert second_boundary.event_metadata["restored_version"] == first_boundary.version
assert second_boundary.event_metadata["rc06_compensation"] is True
assert second_boundary.event_metadata["rc06_restored_boundary"] is True
assert len(session.scalars(select(TemplateLifecycleEventModel)).all()) == 240
snapshots = session.scalars(select(TemplatePublishedSnapshotModel)).all()
assert len(snapshots) == 120
assert all(item.deleted_at is not None for item in snapshots)
finally:
ai._engine.dispose()
contract.engine.dispose()
def test_contract_failure_compensates_ai_and_restores_original_state(tmp_path):
contract, ai = _repositories(tmp_path)
try:
service = Rc06PromptPublicationService(ai._session_factory)
service = _ai_service(ai)
class DriftAfterAiCommit(LocalAiClient):
def commit(self, transaction_id):
@@ -200,7 +488,7 @@ def test_contract_failure_compensates_ai_and_restores_original_state(tmp_path):
))
return result
coordinator = Rc06PublicationCoordinator(
coordinator = _coordinator(
repository=contract,
ai_client=DriftAfterAiCommit(service),
authorization_verifier=TestAuthorizationVerifier(),
@@ -222,7 +510,7 @@ def test_contract_failure_compensates_ai_and_restores_original_state(tmp_path):
def test_compensation_failure_is_terminal_no_go(tmp_path):
contract, ai = _repositories(tmp_path)
try:
service = Rc06PromptPublicationService(ai._session_factory)
service = _ai_service(ai)
class FailingCompensation(LocalAiClient):
def commit(self, transaction_id):
@@ -250,7 +538,7 @@ def test_compensation_failure_is_terminal_no_go(tmp_path):
def compensate(self, transaction_id):
raise OSError("injected AI compensation outage")
coordinator = Rc06PublicationCoordinator(
coordinator = _coordinator(
repository=contract,
ai_client=FailingCompensation(service),
authorization_verifier=TestAuthorizationVerifier(),
@@ -269,6 +557,84 @@ def test_compensation_failure_is_terminal_no_go(tmp_path):
contract.engine.dispose()
@pytest.mark.parametrize(
("overrides", "message"),
[
({"frozen_candidate_commit": ""}, "not configured"),
({"release_image_digest": ""}, "not configured"),
({"frozen_candidate_commit": "b" * 40}, "does not match"),
({"release_image_digest": "sha256:" + "9" * 64}, "does not match"),
],
)
def test_contract_prepare_fails_closed_for_missing_or_mismatched_deployment_identity(
tmp_path,
overrides,
message,
):
contract, ai = _repositories(tmp_path)
try:
coordinator = _coordinator(
repository=contract,
ai_client=LocalAiClient(_ai_service(ai)),
**overrides,
)
with pytest.raises(Rc06PublicationError, match=message):
coordinator.prepare(_request("rc06-contract-deployment-identity"), tenant_id="default")
with contract._session_factory() as session:
assert session.scalar(select(Rc06PublicationSagaModel)) is None
finally:
ai._engine.dispose()
contract.engine.dispose()
def test_ai_prepare_identity_mismatch_is_not_accepted_by_contract(tmp_path):
contract, ai = _repositories(tmp_path)
try:
class MismatchedAiIdentity(LocalAiClient):
def prepare(self, payload):
result = super().prepare(payload)
result["release_image_digest"] = "sha256:" + "7" * 64
return result
coordinator = _coordinator(
repository=contract,
ai_client=MismatchedAiIdentity(_ai_service(ai)),
)
request = _request("rc06-ai-identity-mismatch")
with pytest.raises(Rc06PublicationError, match="did not enter prepared state"):
coordinator.prepare(request, tenant_id="default")
assert coordinator.status(request.transaction_id).state == "preparing"
finally:
ai._engine.dispose()
contract.engine.dispose()
def test_contract_status_returns_persisted_identities_and_rejects_tampering(tmp_path):
contract, ai = _repositories(tmp_path)
try:
coordinator = _coordinator(
repository=contract,
ai_client=LocalAiClient(_ai_service(ai)),
)
request = _request("rc06-contract-identity-readback")
prepared = coordinator.prepare(request, tenant_id="default")
assert prepared.contract_candidate_commit == CANDIDATE_COMMIT
assert prepared.contract_release_image_digest == CONTRACT_IMAGE_DIGEST
assert prepared.ai_candidate_commit == CANDIDATE_COMMIT
assert prepared.ai_release_image_digest == AI_IMAGE_DIGEST
assert coordinator.prepare(request, tenant_id="default") == prepared
with contract._session_factory.begin() as session:
session.get(Rc06PublicationSagaModel, request.transaction_id).ai_release_image_digest = (
"sha256:" + "6" * 64
)
with pytest.raises(Rc06PublicationError, match="deployment identity drifted"):
coordinator.status(request.transaction_id)
finally:
ai._engine.dispose()
contract.engine.dispose()
def test_real_fastapi_endpoint_is_authenticated_and_authorization_fail_closed(tmp_path):
contract, ai = _repositories(tmp_path)
try:
@@ -278,9 +644,11 @@ def test_real_fastapi_endpoint_is_authenticated_and_authorization_fail_closed(tm
seed_demo_data=False,
file_storage_dir=str(tmp_path / "files"),
rc06_publication_shared_secret="r" * 48,
rc06_frozen_candidate_commit=CANDIDATE_COMMIT,
rc06_release_image_digest=CONTRACT_IMAGE_DIGEST,
)
with TestClient(create_app(settings, contract, rc06_ai_client=LocalAiClient(
Rc06PromptPublicationService(ai._session_factory)
_ai_service(ai)
))) as client:
response = client.post("/api/v1/internal/rc06-publications/prepare", json=_request().model_dump(mode="json"))
assert response.status_code == 401
@@ -298,6 +666,28 @@ def test_real_fastapi_endpoint_is_authenticated_and_authorization_fail_closed(tm
contract.engine.dispose()
def test_transaction_lookup_is_tenant_bound(tmp_path):
contract, ai = _repositories(tmp_path)
try:
coordinator = _coordinator(
repository=contract,
ai_client=LocalAiClient(_ai_service(ai)),
authorization_verifier=TestAuthorizationVerifier(),
)
request = _request("rc06-saga-tenant-bound")
coordinator.prepare(request, tenant_id="default")
with pytest.raises(KeyError):
coordinator.status(request.transaction_id, tenant_id="other-tenant")
with pytest.raises(KeyError):
coordinator.commit(request.transaction_id, tenant_id="other-tenant")
with pytest.raises(KeyError):
coordinator.compensate(request.transaction_id, tenant_id="other-tenant")
assert coordinator.status(request.transaction_id, tenant_id="default").state == "prepared"
finally:
ai._engine.dispose()
contract.engine.dispose()
def test_internal_proxy_exemption_matches_only_registered_routes():
assert _is_rc06_internal_path("/api/v1/internal/rc06-publications/prepare", "/api/v1")
assert _is_rc06_internal_path("/api/v1/internal/rc06-publications/rc06-test-0001", "/api/v1")
@@ -313,5 +703,9 @@ def test_rc06_journal_ddl_has_same_constraints_on_sqlite_and_postgresql():
item_sql = str(CreateTable(Rc06TemplatePublicationItemModel.__table__).compile(dialect=dialect))
assert "ux_rc06_publication_sagas_idempotency" in saga_sql
assert "ck_rc06_publication_sagas_state" in saga_sql
assert "contract_candidate_commit VARCHAR(64) NOT NULL" in saga_sql
assert "contract_release_image_digest VARCHAR(71) NOT NULL" in saga_sql
assert "ai_candidate_commit VARCHAR(64) NOT NULL" in saga_sql
assert "ai_release_image_digest VARCHAR(71) NOT NULL" in saga_sql
assert "ux_rc06_template_publication_items_transaction_template" in item_sql
assert "ck_rc06_template_item_state" in item_sql