diff --git a/loopx/capabilities/material_lifecycle/README.md b/loopx/capabilities/material_lifecycle/README.md index e6805c786f..dc13ce5199 100644 --- a/loopx/capabilities/material_lifecycle/README.md +++ b/loopx/capabilities/material_lifecycle/README.md @@ -98,6 +98,30 @@ Ordinary one-off reading, summarization, or web research does not require this capability unless the project has explicitly activated a managed material store. +## Revising An Existing Candidate + +When a short preview is later fully read, revise the existing candidate instead +of adding a duplicate. `build_material_candidate_revision_proposal` binds the +current authority revision, previous record reference, replacement record +reference and exact-read content digest. `apply_material_candidate_revision` +uses the existing source owner's authorization, immutable staging, readback and +CAS; `rollback_material_candidate_revision` restores the previous authority +only while the applied revision is still current. + +The source implements `MaterialCandidateRevisionProvider` beside its existing +intake methods. Reconciliation must preserve all stable identities, record +counts, other records, lifecycle states, rankings and previous content. A +revision cannot reactivate an archive. Metadata changes must be bound by the +replacement record reference; source adapters must invalidate public-field +reviews that no longer match the revised record/content. Ranking changes still +use a separate Decision Context-backed settlement and receipt. + +This extends the existing Python material source apply boundary, without a new +authority, scheduler, ranking policy or generic CLI writer. An SDK proposal +does not authorize a write. Product/project adapters must expose the operation, +publish the readable queues, persist receipts and verify their readback before +reporting the user's workflow complete. Existing intake stays append-only. + ## Single-Material Reranking `plan_material_single_move(ordered_material_refs, material_ref, to_rank)` diff --git a/loopx/capabilities/material_lifecycle/__init__.py b/loopx/capabilities/material_lifecycle/__init__.py index ecba127904..f625ebae97 100644 --- a/loopx/capabilities/material_lifecycle/__init__.py +++ b/loopx/capabilities/material_lifecycle/__init__.py @@ -95,8 +95,26 @@ MATERIAL_INTAKE_RANKING_SETTLEMENT_SCHEMA_VERSION, build_material_intake_ranking_settlement, ) +from .revision import ( + MATERIAL_CANDIDATE_REVISION_APPLY_RECEIPT_SCHEMA_VERSION, + MATERIAL_CANDIDATE_REVISION_PROPOSAL_SCHEMA_VERSION, + MATERIAL_CANDIDATE_REVISION_ROLLBACK_RECEIPT_SCHEMA_VERSION, + MaterialCandidateRevisionProvider, + MaterialCandidateRevisionReconciliation, + apply_material_candidate_revision, + build_material_candidate_revision_proposal, + rollback_material_candidate_revision, +) __all__ = [ + "MATERIAL_CANDIDATE_REVISION_APPLY_RECEIPT_SCHEMA_VERSION", + "MATERIAL_CANDIDATE_REVISION_PROPOSAL_SCHEMA_VERSION", + "MATERIAL_CANDIDATE_REVISION_ROLLBACK_RECEIPT_SCHEMA_VERSION", + "MaterialCandidateRevisionProvider", + "MaterialCandidateRevisionReconciliation", + "apply_material_candidate_revision", + "build_material_candidate_revision_proposal", + "rollback_material_candidate_revision", "MaterialProjectScope", "MaterialProjectScopeVerifier", "MATERIAL_CANDIDATE_INTAKE_APPLY_RECEIPT_SCHEMA_VERSION", diff --git a/loopx/capabilities/material_lifecycle/revision.py b/loopx/capabilities/material_lifecycle/revision.py new file mode 100644 index 0000000000..fa476ef074 --- /dev/null +++ b/loopx/capabilities/material_lifecycle/revision.py @@ -0,0 +1,513 @@ +"""Revise one existing candidate through the existing material source owner.""" + +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass +from typing import Any, Protocol + +from ._validation import ( + capability_contract, + compact_token, + iso_timestamp, + nonnegative_int, + packet_ref, +) +from .apply import MaterialAuthorityTransition +from .intake import ( + MATERIAL_CANDIDATE_INTAKE_PROPOSAL_SCHEMA_VERSION, + MaterialCandidateIntakeProvider, + MaterialCandidateReadback, + MaterialStagedCandidate, + _content_digest, + _identity, + _packet_owner, + _proposal, + _provider_id, + _readback, + _snapshot, + _validate_candidate, + _verified, + build_material_candidate_intake_proposal, +) +from .ownership import MaterialProjectScope, verify_project_material_write + +MATERIAL_CANDIDATE_REVISION_PROPOSAL_SCHEMA_VERSION = ( + "material_candidate_revision_proposal_v0" +) +MATERIAL_CANDIDATE_REVISION_APPLY_RECEIPT_SCHEMA_VERSION = ( + "material_candidate_revision_apply_receipt_v0" +) +MATERIAL_CANDIDATE_REVISION_ROLLBACK_RECEIPT_SCHEMA_VERSION = ( + "material_candidate_revision_rollback_receipt_v0" +) + + +@dataclass(frozen=True) +class MaterialCandidateRevisionReconciliation: + """Source proof: replace one record, retaining membership, ranks and history.""" + + provider_id: str + store_id: str + source_revision: str + target_revision: str + material_ref: str + before_item_count: int + after_item_count: int + validation_ref: str + stable_ids_verified: bool = False + other_records_unchanged: bool = False + rankings_unchanged: bool = False + lifecycle_unchanged: bool = False + previous_content_retained: bool = False + + +class MaterialCandidateRevisionProvider(MaterialCandidateIntakeProvider, Protocol): + """Optional extension of a material source; intake remains append-only.""" + + def stage_candidate_revision( + self, + *, + store_id: str, + proposal_ref: str, + source_revision: str, + material_ref: str, + expected_record_ref: str, + candidate_record_ref: str, + source_ref: str, + source_material_revision: str, + exact_read_ref: str, + content: bytes, + owner_gate_ref: str, + observed_at: str, + ) -> MaterialStagedCandidate: ... + + def reconcile_candidate_revision( + self, + *, + store_id: str, + source_revision: str, + target_revision: str, + material_ref: str, + observed_at: str, + ) -> MaterialCandidateRevisionReconciliation: ... + + +def build_material_candidate_revision_proposal( + *, + expected_record_ref: str, + candidate_record_ref: str, + goal_id: str | None = None, + project_scope: MaterialProjectScope | None = None, + proposal_id: str, + store_id: str, + source_authority_revision: str, + material_ref: str, + source_ref: str, + source_revision: str, + exact_read_ref: str, + content_digest: str, + content_size_bytes: int, + observed_at: str, +) -> dict[str, Any]: + """Bind the previous and replacement records to exact-read content evidence.""" + + proposal = build_material_candidate_intake_proposal( + goal_id=goal_id, + project_scope=project_scope, + proposal_id=proposal_id, + store_id=store_id, + source_authority_revision=source_authority_revision, + material_ref=material_ref, + source_ref=source_ref, + source_revision=source_revision, + exact_read_ref=exact_read_ref, + content_digest=content_digest, + content_size_bytes=content_size_bytes, + observed_at=observed_at, + ) + proposal.pop("proposal_ref") + proposal.update( + schema_version=MATERIAL_CANDIDATE_REVISION_PROPOSAL_SCHEMA_VERSION, + expected_record_ref=compact_token( + expected_record_ref, field="expected_record_ref" + ), + candidate_record_ref=compact_token( + candidate_record_ref, field="candidate_record_ref" + ), + capability=capability_contract( + packet_role="candidate_revision_proposal", + project_scoped=project_scope is not None, + ), + ) + proposal["proposal_ref"] = packet_ref("material-candidate-revision", proposal) + return proposal + + +def _revision_proposal(value: Mapping[str, Any]) -> dict[str, Any]: + if ( + value.get("schema_version") + != MATERIAL_CANDIDATE_REVISION_PROPOSAL_SCHEMA_VERSION + ): + raise ValueError("candidate revision proposal has unsupported schema_version") + packet = dict(value) + supplied_ref = packet.pop("proposal_ref", None) + if supplied_ref != packet_ref("material-candidate-revision", packet): + raise ValueError("candidate revision proposal digest does not match") + previous = compact_token( + packet.pop("expected_record_ref", None), field="expected_record_ref" + ) + replacement = compact_token( + packet.pop("candidate_record_ref", None), field="candidate_record_ref" + ) + packet.update( + schema_version=MATERIAL_CANDIDATE_INTAKE_PROPOSAL_SCHEMA_VERSION, + proposal_ref=supplied_ref, + ) + return { + **_proposal(packet), + "expected_record_ref": previous, + "candidate_record_ref": replacement, + } + + +def _candidate( + provider: MaterialCandidateIntakeProvider, + provider_id: str, + proposal: Mapping[str, Any], + revision: str, + timestamp: str, +) -> MaterialCandidateReadback: + value = _readback( + provider, + provider_id=provider_id, + store_id=proposal["store_id"], + authority_revision=revision, + material_ref=proposal["material_ref"], + observed_at=timestamp, + ) + if value is None: + raise ValueError("candidate revision target is missing") + if value.lifecycle_state != "candidate": + raise ValueError("candidate revision cannot reactivate an archived material") + _verified(value.readback_verified, field="candidate_readback.readback_verified") + return value + + +def _switch( + provider: MaterialCandidateIntakeProvider, + provider_id: str, + owner: Mapping[str, Any], + before: str, + after: str, + gate: str, + timestamp: str, +) -> MaterialAuthorityTransition: + store = owner["store_id"] + verify_project_material_write( + provider, + owner=owner, + store_id=store, + owner_gate_ref=gate, + observed_at=timestamp, + ) + current = _snapshot( + provider, provider_id=provider_id, store_id=store, observed_at=timestamp + ) + if current.authority_revision != before: + raise ValueError("candidate revision authority CAS failed") + transition = provider.switch_authority( + store_id=store, + expected_revision=before, + target_revision=after, + owner_gate_ref=gate, + observed_at=timestamp, + ) + if not isinstance(transition, MaterialAuthorityTransition): + raise TypeError("candidate revision requires MaterialAuthorityTransition") + _identity( + transition, + provider_id=provider_id, + store_id=store, + field="authority_transition", + ) + if transition.before_revision != before or transition.after_revision != after: + raise ValueError("candidate revision transition revisions do not match") + for field in ("atomic_write_verified", "readback_verified"): + _verified(getattr(transition, field), field=f"authority_transition.{field}") + active = _snapshot( + provider, provider_id=provider_id, store_id=store, observed_at=timestamp + ) + if active.authority_revision != after: + raise ValueError("candidate revision authority readback does not match") + return transition + + +def _reconcile( + provider: MaterialCandidateRevisionProvider, + pid: str, + owner: Mapping[str, Any], + source: str, + target: str, + timestamp: str, +) -> tuple[int, str]: + store = owner["store_id"] + reconciliation = provider.reconcile_candidate_revision( + store_id=store, + source_revision=source, + target_revision=target, + material_ref=owner["material_ref"], + observed_at=timestamp, + ) + if not isinstance(reconciliation, MaterialCandidateRevisionReconciliation): + raise TypeError( + "candidate revision requires MaterialCandidateRevisionReconciliation" + ) + _identity( + reconciliation, provider_id=pid, store_id=store, field="revision_reconciliation" + ) + for field, expected in ( + ("source_revision", source), + ("target_revision", target), + ("material_ref", owner["material_ref"]), + ): + if getattr(reconciliation, field) != expected: + raise ValueError( + f"candidate revision reconciliation {field} does not match" + ) + count = nonnegative_int(reconciliation.before_item_count, field="before_item_count") + if ( + nonnegative_int(reconciliation.after_item_count, field="after_item_count") + != count + ): + raise ValueError("candidate revision must retain the record count") + for field in ( + "stable_ids_verified", + "other_records_unchanged", + "rankings_unchanged", + "lifecycle_unchanged", + "previous_content_retained", + ): + _verified( + getattr(reconciliation, field), field=f"revision_reconciliation.{field}" + ) + return count, compact_token(reconciliation.validation_ref, field="validation_ref") + + +def apply_material_candidate_revision( + *, + provider: MaterialCandidateRevisionProvider, + provider_id: str, + revision_proposal: Mapping[str, Any], + content: bytes, + owner_gate_ref: str, + receipt_id: str, + observed_at: str, +) -> dict[str, Any]: + """Replace exactly one candidate, leaving ranking to a separate settlement.""" + + proposal = _revision_proposal(revision_proposal) + if not isinstance(content, bytes) or not content: + raise ValueError("candidate revision content must be non-empty bytes") + if ( + len(content) != proposal["content_size_bytes"] + or _content_digest(content) != proposal["content_digest"] + ): + raise ValueError("candidate revision content does not match proposal") + pid = _provider_id(provider, expected=provider_id) + timestamp = iso_timestamp(observed_at, field="observed_at") + gate = compact_token(owner_gate_ref, field="owner_gate_ref") + compact_token(receipt_id, field="receipt_id") + source = proposal["source_authority_revision"] + store = proposal["store_id"] + verify_project_material_write( + provider, + owner=proposal, + store_id=store, + owner_gate_ref=gate, + observed_at=timestamp, + ) + if ( + _snapshot( + provider, provider_id=pid, store_id=store, observed_at=timestamp + ).authority_revision + != source + ): + raise ValueError("candidate revision source CAS failed before staging") + previous = _candidate(provider, pid, proposal, source, timestamp) + if previous.candidate_record_ref != proposal["expected_record_ref"]: + raise ValueError("candidate revision previous record CAS failed") + previous_record = { + field: getattr(previous, field) + for field in ( + "material_ref", + "source_ref", + "source_revision", + "exact_read_ref", + "content_digest", + "content_size_bytes", + "candidate_record_ref", + "content_backing_ref", + ) + } + # The rollback snapshot must already satisfy the shared public receipt + # contract before the provider stages any new catalog or content. + _validate_candidate(previous, proposal=previous_record, authority_revision=source) + staged = provider.stage_candidate_revision( + store_id=store, + proposal_ref=proposal["proposal_ref"], + source_revision=source, + material_ref=proposal["material_ref"], + expected_record_ref=proposal["expected_record_ref"], + candidate_record_ref=proposal["candidate_record_ref"], + source_ref=proposal["source_ref"], + source_material_revision=proposal["source_revision"], + exact_read_ref=proposal["exact_read_ref"], + content=content, + owner_gate_ref=gate, + observed_at=timestamp, + ) + if not isinstance(staged, MaterialStagedCandidate): + raise TypeError("candidate revision requires MaterialStagedCandidate") + _identity(staged, provider_id=pid, store_id=store, field="staged_candidate") + target = compact_token(staged.target_revision, field="target_revision") + if staged.source_revision != source or target == source: + raise ValueError("candidate revision staged revisions do not match") + for field in ( + "material_ref", + "candidate_record_ref", + "content_digest", + "content_size_bytes", + ): + if getattr(staged, field) != proposal[field]: + raise ValueError(f"candidate revision staged {field} does not match") + for field in ("atomic_write_verified", "readback_verified"): + _verified(getattr(staged, field), field=f"staged_candidate.{field}") + replacement = _candidate(provider, pid, proposal, target, timestamp) + refs = _validate_candidate( + replacement, proposal=proposal, authority_revision=target + ) + if ( + refs["candidate_record_ref"] != proposal["candidate_record_ref"] + or refs["content_backing_ref"] != staged.content_backing_ref + ): + raise ValueError("candidate revision staged record/backing does not match") + count, validation = _reconcile(provider, pid, proposal, source, target, timestamp) + if _candidate(provider, pid, proposal, source, timestamp) != previous: + raise ValueError("candidate revision overwrote previous immutable content") + transition = _switch(provider, pid, proposal, source, target, gate, timestamp) + if _candidate(provider, pid, proposal, target, timestamp) != replacement: + raise ValueError("candidate revision active record readback does not match") + receipt = { + "schema_version": MATERIAL_CANDIDATE_REVISION_APPLY_RECEIPT_SCHEMA_VERSION, + "capability": capability_contract( + packet_role="candidate_revision_apply_receipt", + project_scoped="project_scope" in proposal, + ), + **_packet_owner(revision_proposal), + "receipt_id": receipt_id, + "proposal_ref": proposal["proposal_ref"], + "provider_id": pid, + "store_id": store, + "material_ref": proposal["material_ref"], + "candidate_record_ref": replacement.candidate_record_ref, + "content_backing_ref": replacement.content_backing_ref, + "content_digest": replacement.content_digest, + "before_revision": source, + "after_revision": target, + "item_count": count, + "previous_record": previous_record, + "owner_gate_ref": gate, + "validation_ref": validation, + "authority_ref": compact_token(transition.authority_ref, field="authority_ref"), + "rollback_ref": compact_token(transition.rollback_ref, field="rollback_ref"), + "status": "applied", + "rankings_unchanged": True, + "other_records_unchanged": True, + "previous_content_retained": True, + "source_revision_cas_verified": True, + "readback_verified": True, + "raw_content_captured": False, + "private_locations_captured": False, + "visibility": "public_safe", + "observed_at": timestamp, + } + receipt["receipt_ref"] = packet_ref("material-candidate-revision-apply", receipt) + return receipt + + +def rollback_material_candidate_revision( + *, + provider: MaterialCandidateRevisionProvider, + provider_id: str, + apply_receipt: Mapping[str, Any], + owner_gate_ref: str, + receipt_id: str, + observed_at: str, +) -> dict[str, Any]: + """Restore the previous candidate, refusing rollback across later writes.""" + + packet = dict(apply_receipt) + supplied_ref = packet.pop("receipt_ref", None) + if ( + packet.get("schema_version") + != MATERIAL_CANDIDATE_REVISION_APPLY_RECEIPT_SCHEMA_VERSION + or packet.get("status") != "applied" + or supplied_ref != packet_ref("material-candidate-revision-apply", packet) + ): + raise ValueError( + "candidate revision rollback requires an intact applied receipt" + ) + pid = _provider_id(provider, expected=provider_id) + if packet.get("provider_id") != pid: + raise ValueError("candidate revision receipt provider does not match") + owner = _packet_owner(packet) + timestamp = iso_timestamp(observed_at, field="observed_at") + gate = compact_token(owner_gate_ref, field="owner_gate_ref") + compact_token(receipt_id, field="receipt_id") + verify_project_material_write( + provider, + owner=owner, + store_id=packet["store_id"], + owner_gate_ref=gate, + observed_at=timestamp, + ) + previous = packet["previous_record"] + target = packet["before_revision"] + old = _candidate(provider, pid, packet, target, timestamp) + refs = _validate_candidate(old, proposal=previous, authority_revision=target) + if any(refs[field] != previous[field] for field in refs): + raise ValueError("candidate revision rollback previous record does not match") + _reconcile(provider, pid, packet, target, packet["after_revision"], timestamp) + transition = _switch( + provider, pid, packet, packet["after_revision"], target, gate, timestamp + ) + if _candidate(provider, pid, packet, target, timestamp) != old: + raise ValueError("candidate revision rollback readback does not match") + result = { + "schema_version": MATERIAL_CANDIDATE_REVISION_ROLLBACK_RECEIPT_SCHEMA_VERSION, + "capability": capability_contract( + packet_role="candidate_revision_rollback_receipt", + project_scoped="project_scope" in packet, + ), + **_packet_owner(packet), + "receipt_id": receipt_id, + "apply_receipt_ref": supplied_ref, + "provider_id": pid, + "store_id": packet["store_id"], + "material_ref": packet["material_ref"], + "before_revision": packet["after_revision"], + "after_revision": target, + "owner_gate_ref": gate, + "rollback_ref": transition.rollback_ref, + "status": "rolled_back", + "previous_content_restored": True, + "staged_revision_retained": True, + "readback_verified": True, + "raw_content_captured": False, + "private_locations_captured": False, + "visibility": "public_safe", + "observed_at": timestamp, + } + result["receipt_ref"] = packet_ref("material-candidate-revision-rollback", result) + return result diff --git a/tests/test_material_candidate_revision.py b/tests/test_material_candidate_revision.py new file mode 100644 index 0000000000..9e8f4be6ec --- /dev/null +++ b/tests/test_material_candidate_revision.py @@ -0,0 +1,454 @@ +from __future__ import annotations + +import copy +import hashlib +import json +from dataclasses import replace +from pathlib import Path +from typing import Any, cast + +import pytest + +from loopx.capabilities.material_lifecycle import ( + MaterialAuthoritySnapshot, + MaterialAuthorityTransition, + MaterialCandidateReadback, + MaterialCandidateRevisionProvider, + MaterialCandidateRevisionReconciliation, + MaterialProjectScope, + MaterialStagedCandidate, + apply_material_candidate_revision, + build_material_candidate_revision_proposal, + rollback_material_candidate_revision, +) + +STAMP = "2026-07-26T15:20:00Z" +SCOPE = MaterialProjectScope("project:notes", "profile:materials", "grant:write") +CONTENT = b"full article and exact-read source evidence" + + +def digest(content: bytes) -> str: + return "sha256:" + hashlib.sha256(content).hexdigest() + + +class FileSource: + """Real immutable catalog/content files with an independently checked gate.""" + + provider_id = "file-materials" + + def __init__(self, root: Path) -> None: + self.root = root + self.authority = "revision:old" + self.allowed = True + self.revoke_after_stage = False + self.race = False + self.fault: str | None = None + self.switches = 0 + old = self.backing(b"short preview") + peer = self.backing(b"unrelated material") + records = { + ref: { + "material_ref": ref, + "lifecycle_state": "candidate", + "source_ref": "source:article", + "source_revision": "source:old", + "exact_read_ref": "read:preview", + "candidate_record_ref": "record:old", + "content_backing_ref": backing, + "content_digest": digest((root / backing).read_bytes()), + "content_size_bytes": len((root / backing).read_bytes()), + } + for ref, backing in (("material:article", old), ("material:peer", peer)) + } + self.save( + self.authority, + {"records": records, "rankings": ["material:peer", "material:article"]}, + ) + + def backing(self, content: bytes) -> str: + name = hashlib.sha256(content).hexdigest() + ".bin" + (self.root / name).write_bytes(content) + return name + + def save(self, revision: str, catalog: dict[str, Any]) -> None: + (self.root / (revision.replace(":", "-") + ".json")).write_text( + json.dumps(catalog) + ) + + def load(self, revision: str) -> dict[str, Any]: + return cast( + dict[str, Any], + json.loads( + (self.root / (revision.replace(":", "-") + ".json")).read_text() + ), + ) + + def verify_project_scope(self, **kwargs: Any) -> bool: + return ( + self.allowed + and kwargs["project_scope"] + == { + "project_ref": SCOPE.project_ref, + "source_profile_ref": SCOPE.source_profile_ref, + "workspace_grant_ref": SCOPE.workspace_grant_ref, + } + and kwargs["owner_gate_ref"] == "gate:current" + and kwargs["store_id"] == "store:materials" + ) + + def inspect_authority( + self, *, store_id: str, observed_at: str + ) -> MaterialAuthoritySnapshot: + return MaterialAuthoritySnapshot( + self.provider_id, store_id, self.authority, observed_at + ) + + def read_candidate( + self, + *, + store_id: str, + authority_revision: str, + material_ref: str, + observed_at: str, + ) -> MaterialCandidateReadback | None: + record = self.load(authority_revision)["records"].get(material_ref) + if record is None: + return None + content = (self.root / record["content_backing_ref"]).read_bytes() + assert digest(content) == record["content_digest"] + assert len(content) == record["content_size_bytes"] + return MaterialCandidateReadback( + self.provider_id, + store_id, + authority_revision, + **record, + readback_verified=True, + ) + + def stage_candidate_revision(self, **kwargs: Any) -> MaterialStagedCandidate: + assert self.verify_project_scope( + project_scope={ + "project_ref": SCOPE.project_ref, + "source_profile_ref": SCOPE.source_profile_ref, + "workspace_grant_ref": SCOPE.workspace_grant_ref, + }, + **kwargs, + ) + source = self.load(kwargs["source_revision"]) + target = copy.deepcopy(source) + ref = kwargs["material_ref"] + assert ( + target["records"][ref]["candidate_record_ref"] + == kwargs["expected_record_ref"] + ) + body = kwargs["content"] + backing = self.backing(body) + target["records"][ref].update( + candidate_record_ref=kwargs["candidate_record_ref"], + content_backing_ref=backing, + source_ref=kwargs["source_ref"], + source_revision=kwargs["source_material_revision"], + exact_read_ref=kwargs["exact_read_ref"], + content_digest=digest(body), + content_size_bytes=len(body), + ) + if self.fault == "peer": + target["records"]["material:peer"]["source_revision"] = "source:corrupted" + if self.fault == "rank": + target["rankings"].reverse() + if self.fault == "membership": + target["records"].pop("material:peer") + self.save("revision:new", target) + if self.race: + self.authority = "revision:competitor" + if self.revoke_after_stage: + self.allowed = False + return MaterialStagedCandidate( + self.provider_id, + kwargs["store_id"], + kwargs["source_revision"], + "revision:new", + ref, + kwargs["candidate_record_ref"], + backing, + digest(body), + len(body), + True, + True, + ) + + def reconcile_candidate_revision( + self, **kwargs: Any + ) -> MaterialCandidateRevisionReconciliation: + before = self.load(kwargs["source_revision"]) + after = self.load(kwargs["target_revision"]) + ref = kwargs["material_ref"] + return MaterialCandidateRevisionReconciliation( + self.provider_id, + kwargs["store_id"], + kwargs["source_revision"], + kwargs["target_revision"], + ref, + len(before["records"]), + len(after["records"]), + "validation:revision", + stable_ids_verified=set(before["records"]) == set(after["records"]), + other_records_unchanged={ + k: v for k, v in before["records"].items() if k != ref + } + == {k: v for k, v in after["records"].items() if k != ref}, + rankings_unchanged=before["rankings"] == after["rankings"], + lifecycle_unchanged=before["records"][ref]["lifecycle_state"] + == after["records"][ref]["lifecycle_state"], + previous_content_retained=( + self.root / before["records"][ref]["content_backing_ref"] + ).is_file(), + ) + + def switch_authority(self, **kwargs: Any) -> MaterialAuthorityTransition: + assert self.allowed + assert self.authority == kwargs["expected_revision"] + self.authority = kwargs["target_revision"] + self.switches += 1 + return MaterialAuthorityTransition( + self.provider_id, + kwargs["store_id"], + kwargs["expected_revision"], + kwargs["target_revision"], + "authority:materials", + "rollback:old", + True, + True, + ) + + +def proposal(**updates: Any) -> dict[str, Any]: + args: dict[str, Any] = { + "project_scope": SCOPE, + "proposal_id": "proposal:revise", + "store_id": "store:materials", + "source_authority_revision": "revision:old", + "material_ref": "material:article", + "expected_record_ref": "record:old", + "candidate_record_ref": "record:full-text", + "source_ref": "source:article", + "source_revision": "source:full", + "exact_read_ref": "read:full", + "content_digest": digest(CONTENT), + "content_size_bytes": len(CONTENT), + "observed_at": STAMP, + } + return build_material_candidate_revision_proposal(**{**args, **updates}) + + +def apply( + source: FileSource, packet: dict[str, Any] | None = None, content: bytes = CONTENT +) -> dict[str, Any]: + return apply_material_candidate_revision( + provider=cast(MaterialCandidateRevisionProvider, source), + provider_id=source.provider_id, + revision_proposal=packet or proposal(), + content=content, + owner_gate_ref="gate:current", + receipt_id="receipt:revision", + observed_at=STAMP, + ) + + +def test_revision_and_rollback_keep_identity_order_and_original_bytes( + tmp_path: Path, +) -> None: + source = FileSource(tmp_path) + original = source.load("revision:old") + receipt = apply(source) + current = source.load(source.authority) + assert set(current["records"]) == set(original["records"]) + assert current["rankings"] == original["rankings"] + assert current["records"]["material:peer"] == original["records"]["material:peer"] + assert ( + tmp_path / current["records"]["material:article"]["content_backing_ref"] + ).read_bytes() == CONTENT + assert source.load("revision:old") == original + restored = rollback_material_candidate_revision( + provider=cast(MaterialCandidateRevisionProvider, source), + provider_id=source.provider_id, + apply_receipt=receipt, + owner_gate_ref="gate:current", + receipt_id="receipt:restore", + observed_at=STAMP, + ) + assert restored["previous_content_restored"] + assert source.authority == "revision:old" + assert source.load("revision:new") == current + assert source.load(source.authority) == original + assert "full article" not in json.dumps(receipt) + assert str(tmp_path) not in json.dumps(receipt) + + +@pytest.mark.parametrize("fault", ["peer", "rank", "membership"]) +def test_invalid_revision_never_switches_authority(tmp_path: Path, fault: str) -> None: + source = FileSource(tmp_path) + source.fault = fault + with pytest.raises(ValueError): + apply(source) + assert source.authority == "revision:old" + assert source.switches == 0 + + +@pytest.mark.parametrize( + ("field", "unsafe_value"), + [ + ("source_ref", "https://example.invalid/private-source"), + ("source_revision", "/private/source-revision"), + ("exact_read_ref", "https://example.invalid/private-read"), + ("content_digest", "https://example.invalid/private-digest"), + ("candidate_record_ref", "/private/record"), + ("content_backing_ref", "/private/content"), + ("content_size_bytes", 0), + ("content_size_bytes", -1), + ("content_size_bytes", True), + ("content_size_bytes", 1.25), + ("content_size_bytes", "13"), + ], +) +def test_invalid_previous_snapshot_is_rejected_before_any_staging( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + field: str, + unsafe_value: Any, +) -> None: + source = FileSource(tmp_path) + original = source.load("revision:old") + before_files = {p.name: p.read_bytes() for p in tmp_path.iterdir()} + read_candidate = source.read_candidate + stage_candidate = source.stage_candidate_revision + stages = 0 + + def unsafe_previous(**kwargs: Any) -> MaterialCandidateReadback | None: + value = read_candidate(**kwargs) + if value is not None and kwargs["authority_revision"] == "revision:old": + return replace(value, **{field: unsafe_value}) + return value + + def stage(**kwargs: Any) -> MaterialStagedCandidate: + nonlocal stages + stages += 1 + return stage_candidate(**kwargs) + + monkeypatch.setattr(source, "read_candidate", unsafe_previous) + monkeypatch.setattr(source, "stage_candidate_revision", stage) + with pytest.raises(ValueError): + apply(source) + assert stages == 0 + assert source.switches == 0 + assert source.authority == "revision:old" + assert source.load(source.authority) == original + assert {p.name: p.read_bytes() for p in tmp_path.iterdir()} == before_files + + +@pytest.mark.parametrize("phase", ["before", "after_stage"]) +def test_current_project_grant_is_checked_before_every_write( + tmp_path: Path, phase: str +) -> None: + source = FileSource(tmp_path) + source.allowed = phase != "before" + source.revoke_after_stage = phase == "after_stage" + with pytest.raises(ValueError, match="authorization"): + apply(source) + assert source.authority == "revision:old" + assert source.switches == 0 + + +def test_stale_record_and_authority_conflicts_do_not_publish(tmp_path: Path) -> None: + source = FileSource(tmp_path) + with pytest.raises(ValueError, match="previous record CAS"): + apply(source, proposal(expected_record_ref="record:stale")) + source.race = True + with pytest.raises(ValueError, match="authority CAS"): + apply(source) + assert source.authority == "revision:competitor" + assert source.switches == 0 + + +def test_archive_requires_explicit_lifecycle_action(tmp_path: Path) -> None: + source = FileSource(tmp_path) + catalog = source.load("revision:old") + catalog["records"]["material:article"]["lifecycle_state"] = "archive" + source.save("revision:old", catalog) + with pytest.raises(ValueError, match="reactivate"): + apply(source) + assert source.switches == 0 + + +def test_proposal_and_content_tampering_are_rejected_before_staging( + tmp_path: Path, +) -> None: + source = FileSource(tmp_path) + packet = proposal() + packet["material_ref"] = "material:peer" + with pytest.raises(ValueError, match="digest"): + apply(source, packet) + with pytest.raises(ValueError, match="content does not match"): + apply(source, content=b"other bytes") + assert not (tmp_path / "revision-new.json").exists() + + +def test_rollback_refuses_later_writes_and_tampered_receipts(tmp_path: Path) -> None: + source = FileSource(tmp_path) + receipt = apply(source) + source.authority = "revision:later" + for packet in (receipt, {**receipt, "before_revision": "revision:unrelated"}): + with pytest.raises(ValueError): + rollback_material_candidate_revision( + provider=cast(MaterialCandidateRevisionProvider, source), + provider_id=source.provider_id, + apply_receipt=packet, + owner_gate_ref="gate:current", + receipt_id="receipt:restore", + observed_at=STAMP, + ) + assert source.authority == "revision:later" + assert source.switches == 1 + + +def test_rollback_rechecks_grant_before_reading_history(tmp_path: Path) -> None: + source = FileSource(tmp_path) + receipt = apply(source) + source.allowed = False + with pytest.raises(ValueError, match="authorization"): + rollback_material_candidate_revision( + provider=cast(MaterialCandidateRevisionProvider, source), + provider_id=source.provider_id, + apply_receipt=receipt, + owner_gate_ref="gate:current", + receipt_id="receipt:restore", + observed_at=STAMP, + ) + assert source.authority == "revision:new" + assert source.switches == 1 + + +def test_rollback_checks_peers_even_with_recomputed_receipt_digest( + tmp_path: Path, +) -> None: + from loopx.capabilities.material_lifecycle._validation import packet_ref + + source = FileSource(tmp_path) + receipt = apply(source) + unrelated = source.load("revision:old") + unrelated["records"]["material:peer"]["source_revision"] = "source:unrelated" + source.save("revision:unrelated", unrelated) + packet = {k: v for k, v in receipt.items() if k != "receipt_ref"} + packet["before_revision"] = "revision:unrelated" + packet["receipt_ref"] = packet_ref("material-candidate-revision-apply", packet) + with pytest.raises(ValueError, match="other_records_unchanged"): + rollback_material_candidate_revision( + provider=cast(MaterialCandidateRevisionProvider, source), + provider_id=source.provider_id, + apply_receipt=packet, + owner_gate_ref="gate:current", + receipt_id="receipt:restore", + observed_at=STAMP, + ) + assert source.authority == "revision:new" + assert source.switches == 1