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

fix(rc06): bind compensation evidence and sealing

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