From 8204c75256658b16387ecf10d8bde12c117665b7 Mon Sep 17 00:00:00 2001 From: Guangyu Date: Thu, 20 Aug 2026 08:56:37 +0000 Subject: [PATCH 1/6] feat: add provisional publication contracts --- reflexio/models/api_schema/domain/entities.py | 5 + .../server/services/playbook/publication.py | 215 ++++++++++++- .../storage/storage_base/playbook/_user.py | 34 +++ .../test_provisional_publication_contract.py | 289 ++++++++++++++++++ .../playbook/test_publication_models.py | 54 ++++ 5 files changed, 593 insertions(+), 4 deletions(-) create mode 100644 tests/server/services/playbook/test_provisional_publication_contract.py diff --git a/reflexio/models/api_schema/domain/entities.py b/reflexio/models/api_schema/domain/entities.py index fdb4cf533..403a2da68 100644 --- a/reflexio/models/api_schema/domain/entities.py +++ b/reflexio/models/api_schema/domain/entities.py @@ -114,6 +114,7 @@ "PlaybookAggregationChangeLog", "PlaybookAggregationChangeLogResponse", "OptimizerKind", + "OpenWorldDeploymentLifecycleState", "OptimizationJobStage", "OptimizationTerminalOutcome", "OptimizationArtifactKind", @@ -416,6 +417,10 @@ class AgentPlaybook(BaseModel): "optimizer_legacy_unknown", ] +OpenWorldDeploymentLifecycleState = Literal[ + "provisional", "confirmed", "restored", "displaced", "erased" +] + OptimizationJobStage = Literal[ "evidence_frozen", "discovery_analyzed", diff --git a/reflexio/server/services/playbook/publication.py b/reflexio/server/services/playbook/publication.py index 5032b4a7c..9890c738f 100644 --- a/reflexio/server/services/playbook/publication.py +++ b/reflexio/server/services/playbook/publication.py @@ -9,14 +9,19 @@ from hashlib import sha256 from typing import Literal, Protocol -from reflexio.models.api_schema.domain.entities import OptimizerKind +from reflexio.models.api_schema.domain.entities import OptimizerKind, UserPlaybook PublicationOutcome = Literal["applied", "incumbent_changed"] -PublishableOptimizerKind = Literal["gepa", "offline_tuner_replay"] +PublishableOptimizerKind = Literal[ + "gepa", "offline_tuner_replay", "offline_tuner_open_world" +] PublicationSource = Literal["gepa", "offline_optimizer"] -_PUBLISHABLE_OPTIMIZERS = frozenset({"gepa", "offline_tuner_replay"}) +_PUBLISHABLE_OPTIMIZERS = frozenset( + {"gepa", "offline_tuner_replay", "offline_tuner_open_world"} +) _PROJECTION_SCHEMA_VERSION = "offline-tuner-candidate-search-projection-v1" +_USER_PLAYBOOK_FULL_VERSION_SCHEMA = "user-playbook-full-version-v1" _CANONICAL_DECIMAL = re.compile(r"-?(?:0|[1-9][0-9]*)(?:\.[0-9]*[1-9])?\Z") PUBLICATION_SUBJECT_EPOCHS_METADATA_KEY = "publication_subject_epochs" PUBLICATION_PROOF_JSON_METADATA_KEY = "publication_proof_json" @@ -115,7 +120,7 @@ def publication_source_for_optimizer( optimizer_kind: OptimizerKind, ) -> PublicationSource: _validate_optimizer(optimizer_kind) - return "offline_optimizer" if optimizer_kind == "offline_tuner_replay" else "gepa" + return "offline_optimizer" if optimizer_kind != "gepa" else "gepa" def incumbent_user_playbook_semantic_digest( @@ -324,6 +329,188 @@ def __post_init__(self) -> None: subject_refs.add(subject_ref) +@dataclass(frozen=True) +class QualificationAuthorityRef: + epoch: int + authority_digest: str + discovery_component_identity_digest: str + discovery_qualification_suite_digest: str + discovery_qualification_result_digest: str + held_out_component_identity_digest: str + held_out_qualification_suite_digest: str + held_out_qualification_result_digest: str + candidate_generator_identity_digest: str + candidate_generator_authorization_digest: str + + def __post_init__(self) -> None: + if type(self.epoch) is not int or self.epoch <= 0: + raise ValueError("qualification authority epoch must be positive") + for field in ( + "authority_digest", + "discovery_component_identity_digest", + "discovery_qualification_suite_digest", + "discovery_qualification_result_digest", + "held_out_component_identity_digest", + "held_out_qualification_suite_digest", + "held_out_qualification_result_digest", + "candidate_generator_identity_digest", + "candidate_generator_authorization_digest", + ): + _require_digest(f"qualification authority {field}", getattr(self, field)) + + +def _validated_full_version_snapshot( + *, + incumbent_snapshot_json: str, + incumbent_full_version_fingerprint: str, + incumbent_user_playbook_id: int, +) -> UserPlaybook: + _require_digest( + "incumbent full version fingerprint", incumbent_full_version_fingerprint + ) + payload = _canonical_payload("incumbent_snapshot_json", incumbent_snapshot_json) + if sha256(incumbent_snapshot_json.encode("utf-8")).hexdigest() != ( + incumbent_full_version_fingerprint + ): + raise ValueError("incumbent full version fingerprint does not match snapshot") + if ( + not isinstance(payload, dict) + or set(payload) != {"schema_version", "user_playbook"} + or payload.get("schema_version") != _USER_PLAYBOOK_FULL_VERSION_SCHEMA + or not isinstance(payload.get("user_playbook"), dict) + ): + raise ValueError("incumbent snapshot is not a full user playbook version") + snapshot = UserPlaybook.model_validate(payload["user_playbook"]) + expected_playbook = snapshot.model_dump(mode="json", exclude={"embedding"}) | { + "governance_subject_ref": snapshot.governance_subject_ref, + "retired_at": snapshot.retired_at, + } + if payload["user_playbook"] != expected_playbook: + raise ValueError("incumbent snapshot is not a full user playbook version") + if snapshot.user_playbook_id != incumbent_user_playbook_id: + raise ValueError("incumbent_user_playbook_id does not match snapshot") + return snapshot + + +@dataclass(frozen=True) +class ProvisionalPublicationRequest: + optimizer_kind: Literal["offline_tuner_open_world"] + job_id: int + attempt_key: str + publication_claim: PublicationClaim + worker_fence: int + incumbent_user_playbook_id: int + incumbent_full_version_fingerprint: str + incumbent_snapshot_json: str + revised_content: str + projection: PublicationSearchProjection + decision_proof: DecisionProofEnvelope + subject_epochs_json: str + qualification_authority: QualificationAuthorityRef + evidence_bundle_digest: str + candidate_digest: str + + def __post_init__(self) -> None: + if self.optimizer_kind != "offline_tuner_open_world": + raise ValueError("optimizer_kind must be offline_tuner_open_world") + if type(self.job_id) is not int or self.job_id <= 0: + raise ValueError("provisional publication job_id must be positive") + _require_text("provisional publication attempt_key", self.attempt_key) + if self.publication_claim.job_id != self.job_id: + raise ValueError("publication claim job_id must match request job_id") + if type(self.worker_fence) is not int or self.worker_fence <= 0: + raise ValueError("worker_fence must be positive") + if ( + type(self.incumbent_user_playbook_id) is not int + or self.incumbent_user_playbook_id <= 0 + ): + raise ValueError("incumbent_user_playbook_id must be positive") + snapshot = _validated_full_version_snapshot( + incumbent_snapshot_json=self.incumbent_snapshot_json, + incumbent_full_version_fingerprint=self.incumbent_full_version_fingerprint, + incumbent_user_playbook_id=self.incumbent_user_playbook_id, + ) + _require_text("revised_content", self.revised_content) + if self.revised_content == snapshot.content: + raise ValueError("revised_content must differ from incumbent content") + if self.projection.preserved_trigger != snapshot.trigger: + raise ValueError("search projection must preserve incumbent trigger") + if self.decision_proof.optimizer_kind != self.optimizer_kind: + raise ValueError("decision proof optimizer_kind must match request") + if sha256(self.revised_content.encode("utf-8")).hexdigest() != ( + self.projection.candidate_content_digest + ): + raise ValueError("revised content digest must match search projection") + epochs = _canonical_payload("subject_epochs_json", self.subject_epochs_json) + if ( + not isinstance(epochs, dict) + or set(epochs) != {"subjects"} + or not isinstance(epochs.get("subjects"), list) + or not epochs["subjects"] + ): + raise ValueError("subject epochs must contain a non-empty subjects list") + subject_refs: set[str] = set() + for item in epochs["subjects"]: + if not isinstance(item, dict): + raise ValueError("subject epochs must contain objects") + if set(item) != {"ref", "epoch"}: + raise ValueError("subject epochs must use ref and epoch fields") + subject_ref = item["ref"] + epoch = item["epoch"] + if ( + not isinstance(subject_ref, str) + or not subject_ref + or type(epoch) is not int + or epoch < 0 + ): + raise ValueError("subject epochs contain an invalid identity or epoch") + if subject_ref in subject_refs: + raise ValueError("subject epochs must contain unique subject refs") + subject_refs.add(subject_ref) + if not isinstance(self.qualification_authority, QualificationAuthorityRef): + raise ValueError("qualification authority is invalid") + _require_digest("evidence bundle digest", self.evidence_bundle_digest) + _require_digest("candidate digest", self.candidate_digest) + + +@dataclass(frozen=True) +class ProvisionalPublicationResult: + job_id: int + outcome: Literal["provisional", "incumbent_changed"] + successor_user_playbook_id: int | None + deployment_lifecycle_id: int | None + observation_deadline: int | None + + def __post_init__(self) -> None: + if type(self.job_id) is not int or self.job_id <= 0: + raise ValueError("provisional publication result job_id must be positive") + if self.outcome not in {"provisional", "incumbent_changed"}: + raise ValueError("provisional publication result outcome is invalid") + if self.outcome == "provisional": + if ( + type(self.successor_user_playbook_id) is not int + or self.successor_user_playbook_id <= 0 + or type(self.deployment_lifecycle_id) is not int + or self.deployment_lifecycle_id <= 0 + or type(self.observation_deadline) is not int + or self.observation_deadline <= 0 + ): + raise ValueError( + "provisional publication requires successor lifecycle and deadline" + ) + elif any( + value is not None + for value in ( + self.successor_user_playbook_id, + self.deployment_lifecycle_id, + self.observation_deadline, + ) + ): + raise ValueError( + "incumbent_changed provisional publication cannot have successor state" + ) + + @dataclass(frozen=True) class PublicationResult: job_id: int @@ -368,6 +555,22 @@ def load_user_playbook_publication_result( self, job_id: int ) -> PublicationResult | None: ... + def claim_user_playbook_provisional_publication( + self, *, job_id: int, owner: str, worker_fence: int + ) -> PublicationClaim: ... + + def stage_user_playbook_provisional_publication( + self, request: ProvisionalPublicationRequest + ) -> None: ... + + def commit_user_playbook_provisional_publication( + self, request: ProvisionalPublicationRequest + ) -> ProvisionalPublicationResult: ... + + def load_user_playbook_provisional_publication_result( + self, job_id: int + ) -> ProvisionalPublicationResult | None: ... + class UserPlaybookPublicationService: """Coordinates proof verification with durable staging and atomic commit.""" @@ -418,10 +621,14 @@ def publish_user_playbook_successor( "PublicationRequest", "PublicationResult", "PublicationSearchProjection", + "ProvisionalPublicationRequest", + "ProvisionalPublicationResult", + "PublishableOptimizerKind", "PUBLICATION_INCUMBENT_CONTENT_DIGEST_METADATA_KEY", "PUBLICATION_INCUMBENT_SEMANTIC_DIGEST_METADATA_KEY", "PUBLICATION_INCUMBENT_TRIGGER_METADATA_KEY", "PUBLICATION_SUBJECT_EPOCHS_METADATA_KEY", + "QualificationAuthorityRef", "UserPlaybookPublicationService", "UserPlaybookPublicationStore", "canonical_json_bytes", diff --git a/reflexio/server/services/storage/storage_base/playbook/_user.py b/reflexio/server/services/storage/storage_base/playbook/_user.py index bd47f1f4e..298b49f6b 100644 --- a/reflexio/server/services/storage/storage_base/playbook/_user.py +++ b/reflexio/server/services/storage/storage_base/playbook/_user.py @@ -11,6 +11,8 @@ if TYPE_CHECKING: from reflexio.server.services.playbook.publication import ( + ProvisionalPublicationRequest, + ProvisionalPublicationResult, PublicationClaim, PublicationRequest, PublicationResult, @@ -58,6 +60,38 @@ def load_user_playbook_publication_result( "Storage backend does not support atomic user-playbook publication" ) + def claim_user_playbook_provisional_publication( + self, *, job_id: int, owner: str, worker_fence: int + ) -> "PublicationClaim": + """Claim one open-world provisional publication attempt.""" + raise NotImplementedError( + "Storage backend does not support provisional user-playbook publication" + ) + + def stage_user_playbook_provisional_publication( + self, request: "ProvisionalPublicationRequest" + ) -> None: + """Persist one open-world provisional payload outside visible tables.""" + raise NotImplementedError( + "Storage backend does not support provisional user-playbook publication" + ) + + def commit_user_playbook_provisional_publication( + self, request: "ProvisionalPublicationRequest" + ) -> "ProvisionalPublicationResult": + """Atomically commit one provisional successor or changed incumbent.""" + raise NotImplementedError( + "Storage backend does not support provisional user-playbook publication" + ) + + def load_user_playbook_provisional_publication_result( + self, job_id: int + ) -> "ProvisionalPublicationResult | None": + """Load the immutable terminal provisional publication result.""" + raise NotImplementedError( + "Storage backend does not support provisional user-playbook publication" + ) + @abstractmethod def save_user_playbooks( self, diff --git a/tests/server/services/playbook/test_provisional_publication_contract.py b/tests/server/services/playbook/test_provisional_publication_contract.py new file mode 100644 index 000000000..4ed00f6e9 --- /dev/null +++ b/tests/server/services/playbook/test_provisional_publication_contract.py @@ -0,0 +1,289 @@ +from __future__ import annotations + +import json +from dataclasses import replace +from hashlib import sha256 +from inspect import signature + +import pytest + +from reflexio.models.api_schema.domain import UserPlaybook +from reflexio.server.services.playbook.publication import ( + DecisionProofEnvelope, + ProvisionalPublicationRequest, + ProvisionalPublicationResult, + PublicationClaim, + PublicationSearchProjection, + QualificationAuthorityRef, + UserPlaybookPublicationStore, +) + + +def _canonical(payload: dict[str, object]) -> str: + return json.dumps( + payload, ensure_ascii=False, separators=(",", ":"), sort_keys=True + ) + + +def _digest(value: str) -> str: + return sha256(value.encode()).hexdigest() + + +def _incumbent_snapshot(*, trigger: str | None = "refund") -> str: + incumbent = UserPlaybook( + user_playbook_id=101, + user_id="user-1", + agent_version="agent-v1", + request_id="request-1", + playbook_name="Refund handling", + created_at=123, + content="old content", + trigger=trigger, + rationale="Original rationale", + source="manual", + source_interaction_ids=[1, 2], + source_span="support", + notes="Keep concise", + reader_angle="operator", + tags=["billing"], + governance_subject_ref="subject-1", + ) + payload = { + "schema_version": "user-playbook-full-version-v1", + "user_playbook": incumbent.model_dump(mode="json", exclude={"embedding"}) + | { + "governance_subject_ref": incumbent.governance_subject_ref, + "retired_at": incumbent.retired_at, + }, + } + return _canonical(payload) + + +def _authority() -> QualificationAuthorityRef: + return QualificationAuthorityRef( + epoch=7, + authority_digest="a" * 64, + discovery_component_identity_digest="b" * 64, + discovery_qualification_suite_digest="c" * 64, + discovery_qualification_result_digest="d" * 64, + held_out_component_identity_digest="e" * 64, + held_out_qualification_suite_digest="f" * 64, + held_out_qualification_result_digest="0" * 64, + candidate_generator_identity_digest="1" * 64, + candidate_generator_authorization_digest="2" * 64, + ) + + +def _projection() -> PublicationSearchProjection: + canonical = _canonical( + { + "candidate_content_digest": _digest("new content"), + "embedding": ["0.125", "0.5"], + "embedding_model_id": "test-embedding-v1", + "expanded_terms": ["refund", "escalation"], + "lexical_document": "refund escalation exact projection", + "preserved_trigger": "refund", + "projector_code_digest": "3" * 64, + "projector_id": "reflexio.search.user-playbook", + "projector_version": "1", + "schema_version": "offline-tuner-candidate-search-projection-v1", + } + ) + return PublicationSearchProjection( + schema_version="offline-tuner-candidate-search-projection-v1", + canonical_json=canonical, + digest=_digest(canonical), + projector_id="reflexio.search.user-playbook", + projector_version="1", + projector_code_digest="3" * 64, + candidate_content_digest=_digest("new content"), + preserved_trigger="refund", + embedding_model_id="test-embedding-v1", + embedding=("0.125", "0.5"), + expanded_terms=("refund", "escalation"), + lexical_document="refund escalation exact projection", + ) + + +def _proof() -> DecisionProofEnvelope: + canonical = _canonical( + { + "decision": "apply", + "optimizer_kind": "offline_tuner_open_world", + "schema_version": "offline-tuner-open-world-publication-proof-v1", + } + ) + return DecisionProofEnvelope( + optimizer_kind="offline_tuner_open_world", + schema_version="offline-tuner-open-world-publication-proof-v1", + canonical_json=canonical, + digest=_digest(canonical), + decision="apply", + ) + + +def _request(**changes: object) -> ProvisionalPublicationRequest: + snapshot = _incumbent_snapshot() + values: dict[str, object] = { + "optimizer_kind": "offline_tuner_open_world", + "job_id": 7, + "attempt_key": "attempt-7", + "publication_claim": PublicationClaim(job_id=7, owner="worker-a", fence=3), + "worker_fence": 11, + "incumbent_user_playbook_id": 101, + "incumbent_full_version_fingerprint": _digest(snapshot), + "incumbent_snapshot_json": snapshot, + "revised_content": "new content", + "projection": _projection(), + "decision_proof": _proof(), + "subject_epochs_json": _canonical( + {"subjects": [{"epoch": 0, "ref": "subject:1"}]} + ), + "qualification_authority": _authority(), + "evidence_bundle_digest": "4" * 64, + "candidate_digest": "5" * 64, + } + values.update(changes) + return ProvisionalPublicationRequest(**values) # type: ignore[arg-type] + + +def test_provisional_publication_contract_accepts_exact_content_only_bindings() -> None: + request = _request() + + assert request.optimizer_kind == "offline_tuner_open_world" + assert request.incumbent_full_version_fingerprint == _digest( + request.incumbent_snapshot_json + ) + assert request.qualification_authority.epoch == 7 + + +def test_provisional_publication_result_accepts_complete_terminal_shapes() -> None: + assert ( + ProvisionalPublicationResult( + job_id=7, + outcome="provisional", + successor_user_playbook_id=10, + deployment_lifecycle_id=9, + observation_deadline=1_209_600, + ).successor_user_playbook_id + == 10 + ) + assert ( + ProvisionalPublicationResult( + job_id=7, + outcome="incumbent_changed", + successor_user_playbook_id=None, + deployment_lifecycle_id=None, + observation_deadline=None, + ).outcome + == "incumbent_changed" + ) + + +@pytest.mark.parametrize( + "field", + [ + "authority_digest", + "discovery_component_identity_digest", + "discovery_qualification_suite_digest", + "discovery_qualification_result_digest", + "held_out_component_identity_digest", + "held_out_qualification_suite_digest", + "held_out_qualification_result_digest", + "candidate_generator_identity_digest", + "candidate_generator_authorization_digest", + ], +) +def test_qualification_authority_requires_exact_digest_references(field: str) -> None: + with pytest.raises(ValueError, match="lowercase SHA-256"): + replace(_authority(), **{field: "invalid"}) + + +@pytest.mark.parametrize("optimizer_kind", ["gepa", "offline_tuner_replay"]) +def test_provisional_publication_rejects_non_open_world_optimizers( + optimizer_kind: str, +) -> None: + with pytest.raises(ValueError, match="offline_tuner_open_world"): + _request(optimizer_kind=optimizer_kind) + + +def test_provisional_publication_rejects_claim_projection_and_authority_drift() -> None: + with pytest.raises(ValueError, match="claim job_id"): + _request( + publication_claim=PublicationClaim(job_id=8, owner="worker-a", fence=3) + ) + + with pytest.raises(ValueError, match="revised content digest"): + _request(revised_content="different content") + + with pytest.raises(ValueError, match="qualification authority"): + _request(qualification_authority=replace(_authority(), epoch=0)) + + +def test_provisional_publication_rejects_full_snapshot_fingerprint_and_field_drift() -> ( + None +): + snapshot = _incumbent_snapshot() + with pytest.raises(ValueError, match="full version fingerprint"): + _request(incumbent_full_version_fingerprint="6" * 64) + + with pytest.raises(ValueError, match="incumbent_user_playbook_id"): + _request(incumbent_user_playbook_id=102) + + changed_trigger_snapshot = _incumbent_snapshot(trigger="billing") + with pytest.raises(ValueError, match="preserve incumbent trigger"): + _request( + incumbent_snapshot_json=changed_trigger_snapshot, + incumbent_full_version_fingerprint=_digest(changed_trigger_snapshot), + ) + + changed_snapshot = snapshot.replace("old content", "drifted content") + with pytest.raises(ValueError, match="full version fingerprint"): + _request(incumbent_snapshot_json=changed_snapshot) + + +@pytest.mark.parametrize( + ("outcome", "successor_id", "lifecycle_id", "deadline"), + [ + ("provisional", None, 9, 1_209_600), + ("provisional", 10, None, 1_209_600), + ("provisional", 10, 9, None), + ("incumbent_changed", 10, None, None), + ("incumbent_changed", None, 9, None), + ("incumbent_changed", None, None, 1_209_600), + ], +) +def test_provisional_publication_result_rejects_incomplete_terminal_shapes( + outcome: str, + successor_id: int | None, + lifecycle_id: int | None, + deadline: int | None, +) -> None: + with pytest.raises(ValueError): + ProvisionalPublicationResult( + job_id=7, + outcome=outcome, # type: ignore[arg-type] + successor_user_playbook_id=successor_id, + deployment_lifecycle_id=lifecycle_id, + observation_deadline=deadline, + ) + + +def test_provisional_publication_store_exposes_distinct_durable_operations() -> None: + assert tuple( + signature( + UserPlaybookPublicationStore.claim_user_playbook_provisional_publication + ).parameters + ) == ("self", "job_id", "owner", "worker_fence") + assert callable( + UserPlaybookPublicationStore.claim_user_playbook_provisional_publication + ) + assert callable( + UserPlaybookPublicationStore.stage_user_playbook_provisional_publication + ) + assert callable( + UserPlaybookPublicationStore.commit_user_playbook_provisional_publication + ) + assert callable( + UserPlaybookPublicationStore.load_user_playbook_provisional_publication_result + ) diff --git a/tests/server/services/playbook/test_publication_models.py b/tests/server/services/playbook/test_publication_models.py index 4eb5d150d..10e48e04b 100644 --- a/tests/server/services/playbook/test_publication_models.py +++ b/tests/server/services/playbook/test_publication_models.py @@ -2,20 +2,74 @@ import json from hashlib import sha256 +from typing import get_args import pytest +from reflexio.models.api_schema.domain import ( + OpenWorldDeploymentLifecycleState, + UserPlaybook, +) from reflexio.server.services.playbook.publication import ( DecisionProofEnvelope, PublicationClaim, PublicationRequest, PublicationSearchProjection, + PublishableOptimizerKind, UserPlaybookPublicationService, canonical_json_bytes, incumbent_user_playbook_semantic_digest, ) +def test_open_world_publication_literals_and_user_playbook_field_partition() -> None: + assert get_args(PublishableOptimizerKind) == ( + "gepa", + "offline_tuner_replay", + "offline_tuner_open_world", + ) + assert get_args(OpenWorldDeploymentLifecycleState) == ( + "provisional", + "confirmed", + "restored", + "displaced", + "erased", + ) + + tunable = {"content"} + preserved = { + "user_id", + "agent_version", + "request_id", + "playbook_name", + "trigger", + "rationale", + "blocking_issue", + "source", + "source_interaction_ids", + "source_span", + "notes", + "reader_angle", + "tags", + "governance_subject_ref", + } + version_mechanics = { + "user_playbook_id", + "created_at", + "status", + "merged_into", + "superseded_by", + "retired_at", + } + search_derived = {"embedding", "expanded_terms"} + partitions = (tunable, preserved, version_mechanics, search_derived) + + assert len(set().union(*partitions)) == sum( + len(partition) for partition in partitions + ) + assert set().union(*partitions) == set(UserPlaybook.model_fields) + + def _canonical(payload: dict[str, object]) -> str: return json.dumps( payload, ensure_ascii=False, separators=(",", ":"), sort_keys=True From 0091512ad6003757ce04ed8f0445680a910a3dbb Mon Sep 17 00:00:00 2001 From: Guangyu Date: Thu, 20 Aug 2026 09:12:44 +0000 Subject: [PATCH 2/6] fix: isolate provisional publication contracts --- .../server/services/playbook/publication.py | 24 +++++++++++---- .../test_provisional_publication_contract.py | 30 +++++++++++++++++++ .../playbook/test_publication_models.py | 24 +++++++++++++++ 3 files changed, 72 insertions(+), 6 deletions(-) diff --git a/reflexio/server/services/playbook/publication.py b/reflexio/server/services/playbook/publication.py index 9890c738f..d1bbba493 100644 --- a/reflexio/server/services/playbook/publication.py +++ b/reflexio/server/services/playbook/publication.py @@ -17,7 +17,8 @@ ] PublicationSource = Literal["gepa", "offline_optimizer"] -_PUBLISHABLE_OPTIMIZERS = frozenset( +_LEGACY_PUBLICATION_OPTIMIZERS = frozenset({"gepa", "offline_tuner_replay"}) +_DECISION_PROOF_OPTIMIZERS = frozenset( {"gepa", "offline_tuner_replay", "offline_tuner_open_world"} ) _PROJECTION_SCHEMA_VERSION = "offline-tuner-candidate-search-projection-v1" @@ -111,15 +112,20 @@ def _canonical_payload(name: str, value: str) -> object: return payload -def _validate_optimizer(value: object) -> None: - if value not in _PUBLISHABLE_OPTIMIZERS: +def _validate_legacy_publication_optimizer(value: object) -> None: + if value not in _LEGACY_PUBLICATION_OPTIMIZERS: + raise ValueError("optimizer_kind is not publishable") + + +def _validate_decision_proof_optimizer(value: object) -> None: + if value not in _DECISION_PROOF_OPTIMIZERS: raise ValueError("optimizer_kind is not publishable") def publication_source_for_optimizer( optimizer_kind: OptimizerKind, ) -> PublicationSource: - _validate_optimizer(optimizer_kind) + _validate_legacy_publication_optimizer(optimizer_kind) return "offline_optimizer" if optimizer_kind != "gepa" else "gepa" @@ -161,7 +167,7 @@ class DecisionProofEnvelope: decision: Literal["apply"] def __post_init__(self) -> None: - _validate_optimizer(self.optimizer_kind) + _validate_decision_proof_optimizer(self.optimizer_kind) _require_text("decision proof schema_version", self.schema_version) _require_digest("decision proof digest", self.digest) if self.decision != "apply": @@ -270,7 +276,7 @@ class PublicationRequest: request_id: str def __post_init__(self) -> None: - _validate_optimizer(self.optimizer_kind) + _validate_legacy_publication_optimizer(self.optimizer_kind) if type(self.job_id) is not int or self.job_id <= 0: raise ValueError("publication job_id must be positive") _require_text("publication attempt_key", self.attempt_key) @@ -413,6 +419,12 @@ class ProvisionalPublicationRequest: def __post_init__(self) -> None: if self.optimizer_kind != "offline_tuner_open_world": raise ValueError("optimizer_kind must be offline_tuner_open_world") + if type(self.publication_claim) is not PublicationClaim: + raise ValueError("publication_claim must be PublicationClaim") + if type(self.projection) is not PublicationSearchProjection: + raise ValueError("projection must be PublicationSearchProjection") + if type(self.decision_proof) is not DecisionProofEnvelope: + raise ValueError("decision_proof must be DecisionProofEnvelope") if type(self.job_id) is not int or self.job_id <= 0: raise ValueError("provisional publication job_id must be positive") _require_text("provisional publication attempt_key", self.attempt_key) diff --git a/tests/server/services/playbook/test_provisional_publication_contract.py b/tests/server/services/playbook/test_provisional_publication_contract.py index 4ed00f6e9..dd5abac8d 100644 --- a/tests/server/services/playbook/test_provisional_publication_contract.py +++ b/tests/server/services/playbook/test_provisional_publication_contract.py @@ -147,6 +147,21 @@ def _request(**changes: object) -> ProvisionalPublicationRequest: return ProvisionalPublicationRequest(**values) # type: ignore[arg-type] +class _PublicationClaimImpostor: + job_id = 7 + owner = "worker-a" + fence = 3 + + +class _PublicationSearchProjectionImpostor: + preserved_trigger = "refund" + candidate_content_digest = _digest("new content") + + +class _DecisionProofEnvelopeImpostor: + optimizer_kind = "offline_tuner_open_world" + + def test_provisional_publication_contract_accepts_exact_content_only_bindings() -> None: request = _request() @@ -157,6 +172,21 @@ def test_provisional_publication_contract_accepts_exact_content_only_bindings() assert request.qualification_authority.epoch == 7 +@pytest.mark.parametrize( + ("field", "value"), + [ + ("publication_claim", _PublicationClaimImpostor()), + ("projection", _PublicationSearchProjectionImpostor()), + ("decision_proof", _DecisionProofEnvelopeImpostor()), + ], +) +def test_provisional_publication_rejects_structurally_matching_subcontract_impostors( + field: str, value: object +) -> None: + with pytest.raises(ValueError, match=f"{field} must be"): + _request(**{field: value}) + + def test_provisional_publication_result_accepts_complete_terminal_shapes() -> None: assert ( ProvisionalPublicationResult( diff --git a/tests/server/services/playbook/test_publication_models.py b/tests/server/services/playbook/test_publication_models.py index 10e48e04b..228f9a626 100644 --- a/tests/server/services/playbook/test_publication_models.py +++ b/tests/server/services/playbook/test_publication_models.py @@ -269,6 +269,30 @@ def test_publication_request_binds_content_optimizer_and_canonical_epochs() -> N ) +def test_legacy_publication_request_rejects_open_world_optimizer() -> None: + with pytest.raises(ValueError, match="optimizer_kind is not publishable"): + PublicationRequest( + optimizer_kind="offline_tuner_open_world", + job_id=7, + attempt_key="attempt-7", + publication_claim=PublicationClaim(job_id=7, owner="worker-a", fence=3), + worker_fence=11, + incumbent_user_playbook_id=101, + incumbent_content_digest=_digest("old content"), + incumbent_trigger="refund", + incumbent_semantic_digest=incumbent_user_playbook_semantic_digest( + content_digest=_digest("old content"), trigger="refund" + ), + revised_content="new content", + projection=_projection(), + decision_proof=_proof(), + subject_epochs_json=_canonical( + {"subjects": [{"epoch": 0, "ref": "subject:1"}]} + ), + request_id="request-7", + ) + + def test_publication_request_rejects_wrong_claim_kind_and_non_apply_decision() -> None: canonical = _canonical( { From 948f48b1f26e8baa7581ce285087e16fa94492eb Mon Sep 17 00:00:00 2001 From: Guangyu Date: Fri, 21 Aug 2026 04:26:09 +0000 Subject: [PATCH 3/6] fix: seal provisional publication contracts --- .../server/services/playbook/publication.py | 15 ++- .../test_provisional_publication_contract.py | 91 +++++++++++++++++-- 2 files changed, 97 insertions(+), 9 deletions(-) diff --git a/reflexio/server/services/playbook/publication.py b/reflexio/server/services/playbook/publication.py index d1bbba493..e1e2bec4e 100644 --- a/reflexio/server/services/playbook/publication.py +++ b/reflexio/server/services/playbook/publication.py @@ -391,7 +391,9 @@ def _validated_full_version_snapshot( "governance_subject_ref": snapshot.governance_subject_ref, "retired_at": snapshot.retired_at, } - if payload["user_playbook"] != expected_playbook: + if canonical_json_bytes(payload["user_playbook"]) != canonical_json_bytes( + expected_playbook + ): raise ValueError("incumbent snapshot is not a full user playbook version") if snapshot.user_playbook_id != incumbent_user_playbook_id: raise ValueError("incumbent_user_playbook_id does not match snapshot") @@ -479,8 +481,10 @@ def __post_init__(self) -> None: if subject_ref in subject_refs: raise ValueError("subject epochs must contain unique subject refs") subject_refs.add(subject_ref) - if not isinstance(self.qualification_authority, QualificationAuthorityRef): - raise ValueError("qualification authority is invalid") + if type(self.qualification_authority) is not QualificationAuthorityRef: + raise ValueError( + "qualification authority must be QualificationAuthorityRef" + ) _require_digest("evidence bundle digest", self.evidence_bundle_digest) _require_digest("candidate digest", self.candidate_digest) @@ -567,6 +571,10 @@ def load_user_playbook_publication_result( self, job_id: int ) -> PublicationResult | None: ... + +class UserPlaybookProvisionalPublicationStore(UserPlaybookPublicationStore, Protocol): + """Durable Phase 4 provisional publication operations.""" + def claim_user_playbook_provisional_publication( self, *, job_id: int, owner: str, worker_fence: int ) -> PublicationClaim: ... @@ -641,6 +649,7 @@ def publish_user_playbook_successor( "PUBLICATION_INCUMBENT_TRIGGER_METADATA_KEY", "PUBLICATION_SUBJECT_EPOCHS_METADATA_KEY", "QualificationAuthorityRef", + "UserPlaybookProvisionalPublicationStore", "UserPlaybookPublicationService", "UserPlaybookPublicationStore", "canonical_json_bytes", diff --git a/tests/server/services/playbook/test_provisional_publication_contract.py b/tests/server/services/playbook/test_provisional_publication_contract.py index dd5abac8d..8d4d6466d 100644 --- a/tests/server/services/playbook/test_provisional_publication_contract.py +++ b/tests/server/services/playbook/test_provisional_publication_contract.py @@ -15,6 +15,7 @@ PublicationClaim, PublicationSearchProjection, QualificationAuthorityRef, + UserPlaybookProvisionalPublicationStore, UserPlaybookPublicationStore, ) @@ -162,6 +163,14 @@ class _DecisionProofEnvelopeImpostor: optimizer_kind = "offline_tuner_open_world" +class _QualificationAuthorityRefImpostor: + epoch = 7 + + +class _QualificationAuthorityRefSubclass(QualificationAuthorityRef): + pass + + def test_provisional_publication_contract_accepts_exact_content_only_bindings() -> None: request = _request() @@ -250,6 +259,20 @@ def test_provisional_publication_rejects_claim_projection_and_authority_drift() _request(qualification_authority=replace(_authority(), epoch=0)) +@pytest.mark.parametrize( + "authority", + [ + _QualificationAuthorityRefImpostor(), + _QualificationAuthorityRefSubclass(**_authority().__dict__), + ], +) +def test_provisional_publication_requires_an_exact_qualification_authority_ref( + authority: object, +) -> None: + with pytest.raises(ValueError, match="qualification authority must be"): + _request(qualification_authority=authority) + + def test_provisional_publication_rejects_full_snapshot_fingerprint_and_field_drift() -> ( None ): @@ -272,6 +295,41 @@ def test_provisional_publication_rejects_full_snapshot_fingerprint_and_field_dri _request(incumbent_snapshot_json=changed_snapshot) +@pytest.mark.parametrize( + ("field", "value", "message"), + [ + ("created_at", True, "full user playbook version"), + ("created_at", 123.0, "canonical JSON"), + ], +) +def test_provisional_publication_rejects_coerced_full_snapshot_scalars( + field: str, value: object, message: str +) -> None: + payload = json.loads(_incumbent_snapshot()) + payload["user_playbook"][field] = value + snapshot = _canonical(payload) + + with pytest.raises(ValueError, match=message): + _request( + incumbent_snapshot_json=snapshot, + incumbent_full_version_fingerprint=_digest(snapshot), + ) + + +def test_provisional_publication_accepts_serialized_enum_full_snapshot() -> None: + payload = json.loads(_incumbent_snapshot()) + payload["user_playbook"]["status"] = "archived" + snapshot = _canonical(payload) + + assert ( + _request( + incumbent_snapshot_json=snapshot, + incumbent_full_version_fingerprint=_digest(snapshot), + ).incumbent_snapshot_json + == snapshot + ) + + @pytest.mark.parametrize( ("outcome", "successor_id", "lifecycle_id", "deadline"), [ @@ -299,21 +357,42 @@ def test_provisional_publication_result_rejects_incomplete_terminal_shapes( ) -def test_provisional_publication_store_exposes_distinct_durable_operations() -> None: +def test_provisional_publication_store_keeps_legacy_and_provisional_contracts_separate() -> ( + None +): + provisional_operations = ( + "claim_user_playbook_provisional_publication", + "stage_user_playbook_provisional_publication", + "commit_user_playbook_provisional_publication", + "load_user_playbook_provisional_publication_result", + ) + + assert all( + operation not in UserPlaybookPublicationStore.__dict__ + for operation in provisional_operations + ) + assert ( + UserPlaybookPublicationStore + in UserPlaybookProvisionalPublicationStore.__bases__ + ) + assert ( + "load_open_world_provisional_publication_retry" + not in UserPlaybookProvisionalPublicationStore.__dict__ + ) assert tuple( signature( - UserPlaybookPublicationStore.claim_user_playbook_provisional_publication + UserPlaybookProvisionalPublicationStore.claim_user_playbook_provisional_publication ).parameters ) == ("self", "job_id", "owner", "worker_fence") assert callable( - UserPlaybookPublicationStore.claim_user_playbook_provisional_publication + UserPlaybookProvisionalPublicationStore.claim_user_playbook_provisional_publication ) assert callable( - UserPlaybookPublicationStore.stage_user_playbook_provisional_publication + UserPlaybookProvisionalPublicationStore.stage_user_playbook_provisional_publication ) assert callable( - UserPlaybookPublicationStore.commit_user_playbook_provisional_publication + UserPlaybookProvisionalPublicationStore.commit_user_playbook_provisional_publication ) assert callable( - UserPlaybookPublicationStore.load_user_playbook_provisional_publication_result + UserPlaybookProvisionalPublicationStore.load_user_playbook_provisional_publication_result ) From d198fcc411b2a4651e7daa741c37141a83f40782 Mon Sep 17 00:00:00 2001 From: Guangyu Date: Sat, 22 Aug 2026 01:40:26 +0000 Subject: [PATCH 4/6] refactor: share subject epoch validation --- .../server/services/playbook/publication.py | 83 +++++++------------ 1 file changed, 31 insertions(+), 52 deletions(-) diff --git a/reflexio/server/services/playbook/publication.py b/reflexio/server/services/playbook/publication.py index e1e2bec4e..5c3bf2392 100644 --- a/reflexio/server/services/playbook/publication.py +++ b/reflexio/server/services/playbook/publication.py @@ -112,6 +112,35 @@ def _canonical_payload(name: str, value: str) -> object: return payload +def _validate_subject_epochs_json(value: str) -> None: + epochs = _canonical_payload("subject_epochs_json", value) + if ( + not isinstance(epochs, dict) + or set(epochs) != {"subjects"} + or not isinstance(epochs.get("subjects"), list) + or not epochs["subjects"] + ): + raise ValueError("subject epochs must contain a non-empty subjects list") + subject_refs: set[str] = set() + for item in epochs["subjects"]: + if not isinstance(item, dict): + raise ValueError("subject epochs must contain objects") + if set(item) != {"ref", "epoch"}: + raise ValueError("subject epochs must use ref and epoch fields") + subject_ref = item["ref"] + epoch = item["epoch"] + if ( + not isinstance(subject_ref, str) + or not subject_ref + or type(epoch) is not int + or epoch < 0 + ): + raise ValueError("subject epochs contain an invalid identity or epoch") + if subject_ref in subject_refs: + raise ValueError("subject epochs must contain unique subject refs") + subject_refs.add(subject_ref) + + def _validate_legacy_publication_optimizer(value: object) -> None: if value not in _LEGACY_PUBLICATION_OPTIMIZERS: raise ValueError("optimizer_kind is not publishable") @@ -307,32 +336,7 @@ def __post_init__(self) -> None: self.projection.candidate_content_digest ): raise ValueError("revised content digest must match search projection") - epochs = _canonical_payload("subject_epochs_json", self.subject_epochs_json) - if ( - not isinstance(epochs, dict) - or set(epochs) != {"subjects"} - or not isinstance(epochs.get("subjects"), list) - or not epochs["subjects"] - ): - raise ValueError("subject epochs must contain a non-empty subjects list") - subject_refs: set[str] = set() - for item in epochs["subjects"]: - if not isinstance(item, dict): - raise ValueError("subject epochs must contain objects") - if set(item) != {"ref", "epoch"}: - raise ValueError("subject epochs must use ref and epoch fields") - subject_ref = item["ref"] - epoch = item["epoch"] - if ( - not isinstance(subject_ref, str) - or not subject_ref - or type(epoch) is not int - or epoch < 0 - ): - raise ValueError("subject epochs contain an invalid identity or epoch") - if subject_ref in subject_refs: - raise ValueError("subject epochs must contain unique subject refs") - subject_refs.add(subject_ref) + _validate_subject_epochs_json(self.subject_epochs_json) @dataclass(frozen=True) @@ -455,32 +459,7 @@ def __post_init__(self) -> None: self.projection.candidate_content_digest ): raise ValueError("revised content digest must match search projection") - epochs = _canonical_payload("subject_epochs_json", self.subject_epochs_json) - if ( - not isinstance(epochs, dict) - or set(epochs) != {"subjects"} - or not isinstance(epochs.get("subjects"), list) - or not epochs["subjects"] - ): - raise ValueError("subject epochs must contain a non-empty subjects list") - subject_refs: set[str] = set() - for item in epochs["subjects"]: - if not isinstance(item, dict): - raise ValueError("subject epochs must contain objects") - if set(item) != {"ref", "epoch"}: - raise ValueError("subject epochs must use ref and epoch fields") - subject_ref = item["ref"] - epoch = item["epoch"] - if ( - not isinstance(subject_ref, str) - or not subject_ref - or type(epoch) is not int - or epoch < 0 - ): - raise ValueError("subject epochs contain an invalid identity or epoch") - if subject_ref in subject_refs: - raise ValueError("subject epochs must contain unique subject refs") - subject_refs.add(subject_ref) + _validate_subject_epochs_json(self.subject_epochs_json) if type(self.qualification_authority) is not QualificationAuthorityRef: raise ValueError( "qualification authority must be QualificationAuthorityRef" From b76bfb7d42aee8d72ffba335c5594a5c32ff0d4d Mon Sep 17 00:00:00 2001 From: Guangyu Date: Fri, 28 Aug 2026 00:25:35 +0000 Subject: [PATCH 5/6] fix: fence provisional publication claims on incumbent --- reflexio/server/services/playbook/publication.py | 13 +++++++++++-- .../storage/storage_base/playbook/_user.py | 15 +++++++++++---- .../test_provisional_publication_contract.py | 10 +++++++++- 3 files changed, 31 insertions(+), 7 deletions(-) diff --git a/reflexio/server/services/playbook/publication.py b/reflexio/server/services/playbook/publication.py index 5c3bf2392..03dc6ab4c 100644 --- a/reflexio/server/services/playbook/publication.py +++ b/reflexio/server/services/playbook/publication.py @@ -555,12 +555,21 @@ class UserPlaybookProvisionalPublicationStore(UserPlaybookPublicationStore, Prot """Durable Phase 4 provisional publication operations.""" def claim_user_playbook_provisional_publication( - self, *, job_id: int, owner: str, worker_fence: int + self, + *, + job_id: int, + owner: str, + worker_fence: int, + incumbent_user_playbook_id: int, + incumbent_full_version_fingerprint: str, + incumbent_snapshot_json: str, ) -> PublicationClaim: ... def stage_user_playbook_provisional_publication( self, request: ProvisionalPublicationRequest - ) -> None: ... + ) -> bool: + """Stage the request, or return false after cleaning stale-CAS residue.""" + ... def commit_user_playbook_provisional_publication( self, request: ProvisionalPublicationRequest diff --git a/reflexio/server/services/storage/storage_base/playbook/_user.py b/reflexio/server/services/storage/storage_base/playbook/_user.py index 298b49f6b..c6dd0b765 100644 --- a/reflexio/server/services/storage/storage_base/playbook/_user.py +++ b/reflexio/server/services/storage/storage_base/playbook/_user.py @@ -61,17 +61,24 @@ def load_user_playbook_publication_result( ) def claim_user_playbook_provisional_publication( - self, *, job_id: int, owner: str, worker_fence: int + self, + *, + job_id: int, + owner: str, + worker_fence: int, + incumbent_user_playbook_id: int, + incumbent_full_version_fingerprint: str, + incumbent_snapshot_json: str, ) -> "PublicationClaim": - """Claim one open-world provisional publication attempt.""" + """Claim only while the frozen incumbent is still current.""" raise NotImplementedError( "Storage backend does not support provisional user-playbook publication" ) def stage_user_playbook_provisional_publication( self, request: "ProvisionalPublicationRequest" - ) -> None: - """Persist one open-world provisional payload outside visible tables.""" + ) -> bool: + """Stage a payload, or clean a stale claim and return false.""" raise NotImplementedError( "Storage backend does not support provisional user-playbook publication" ) diff --git a/tests/server/services/playbook/test_provisional_publication_contract.py b/tests/server/services/playbook/test_provisional_publication_contract.py index 8d4d6466d..44db97f2c 100644 --- a/tests/server/services/playbook/test_provisional_publication_contract.py +++ b/tests/server/services/playbook/test_provisional_publication_contract.py @@ -383,7 +383,15 @@ def test_provisional_publication_store_keeps_legacy_and_provisional_contracts_se signature( UserPlaybookProvisionalPublicationStore.claim_user_playbook_provisional_publication ).parameters - ) == ("self", "job_id", "owner", "worker_fence") + ) == ( + "self", + "job_id", + "owner", + "worker_fence", + "incumbent_user_playbook_id", + "incumbent_full_version_fingerprint", + "incumbent_snapshot_json", + ) assert callable( UserPlaybookProvisionalPublicationStore.claim_user_playbook_provisional_publication ) From 7554c98a3664dccae13cce7851a0f428dec413bf Mon Sep 17 00:00:00 2001 From: Guangyu Date: Fri, 28 Aug 2026 01:08:46 +0000 Subject: [PATCH 6/6] fix: expose fenced provisional cleanup contract --- reflexio/server/services/playbook/publication.py | 6 ++++++ .../services/storage/storage_base/playbook/_user.py | 8 ++++++++ .../playbook/test_provisional_publication_contract.py | 9 +++++++++ 3 files changed, 23 insertions(+) diff --git a/reflexio/server/services/playbook/publication.py b/reflexio/server/services/playbook/publication.py index 03dc6ab4c..e0c474184 100644 --- a/reflexio/server/services/playbook/publication.py +++ b/reflexio/server/services/playbook/publication.py @@ -554,6 +554,12 @@ def load_user_playbook_publication_result( class UserPlaybookProvisionalPublicationStore(UserPlaybookPublicationStore, Protocol): """Durable Phase 4 provisional publication operations.""" + def cleanup_user_playbook_provisional_stale_attempt( + self, *, job_id: int, owner: str, worker_fence: int + ) -> bool: + """Clean only matching uncommitted residue; false means ownership changed.""" + ... + def claim_user_playbook_provisional_publication( self, *, diff --git a/reflexio/server/services/storage/storage_base/playbook/_user.py b/reflexio/server/services/storage/storage_base/playbook/_user.py index c6dd0b765..f8e25dc41 100644 --- a/reflexio/server/services/storage/storage_base/playbook/_user.py +++ b/reflexio/server/services/storage/storage_base/playbook/_user.py @@ -75,6 +75,14 @@ def claim_user_playbook_provisional_publication( "Storage backend does not support provisional user-playbook publication" ) + def cleanup_user_playbook_provisional_stale_attempt( + self, *, job_id: int, owner: str, worker_fence: int + ) -> bool: + """Clean only this worker's uncommitted provisional publication residue.""" + raise NotImplementedError( + "Storage backend does not support provisional user-playbook publication" + ) + def stage_user_playbook_provisional_publication( self, request: "ProvisionalPublicationRequest" ) -> bool: diff --git a/tests/server/services/playbook/test_provisional_publication_contract.py b/tests/server/services/playbook/test_provisional_publication_contract.py index 44db97f2c..4a79d2d1c 100644 --- a/tests/server/services/playbook/test_provisional_publication_contract.py +++ b/tests/server/services/playbook/test_provisional_publication_contract.py @@ -361,6 +361,7 @@ def test_provisional_publication_store_keeps_legacy_and_provisional_contracts_se None ): provisional_operations = ( + "cleanup_user_playbook_provisional_stale_attempt", "claim_user_playbook_provisional_publication", "stage_user_playbook_provisional_publication", "commit_user_playbook_provisional_publication", @@ -392,6 +393,14 @@ def test_provisional_publication_store_keeps_legacy_and_provisional_contracts_se "incumbent_full_version_fingerprint", "incumbent_snapshot_json", ) + assert callable( + UserPlaybookProvisionalPublicationStore.cleanup_user_playbook_provisional_stale_attempt + ) + assert tuple( + signature( + UserPlaybookProvisionalPublicationStore.cleanup_user_playbook_provisional_stale_attempt + ).parameters + ) == ("self", "job_id", "owner", "worker_fence") assert callable( UserPlaybookProvisionalPublicationStore.claim_user_playbook_provisional_publication )