feat(records): define archive transfer provider contract

docs/RECORDS_FILING_CONTRACT.md

@@ -11,6 +11,8 @@ revision.
 - `records.filing` is supplied by the enabled Records module.
 - `records.source.<module>` is supplied by an enabled source module, for
   example `records.source.files` or `records.source.cases`.
+- `records.archive.<provider>` is supplied by an enabled archive-transfer
+  adapter. Discovery does not imply conformance or current health.
 
 Callers discover capabilities through the module registry. They must not
 import optional source-module internals.
@@ -59,3 +61,24 @@ additional optional metadata remains backward compatible.
 
 Provider-specific selection UI belongs to the source module. The generic
 Records dialog remains a diagnostic/manual fallback for exact identifiers.
+
+## Archive Transfer Boundary
+
+`RecordTransferPackage` binds a stable package ID, record revision, provider
+profile, canonical manifest, and manifest SHA-256. An archive provider exposes
+`RecordArchiveProviderState` before dispatch and accepts only a
+`RecordArchiveTransferRequest` for a declared healthy profile. Its receipt must
+identify the same package and provider and return one bounded outcome:
+`accepted`, `rejected`, or `outcome_unknown`.
+
+An unknown outcome is never retry-safe. Callers must retain the intent and
+reconcile it against the provider before another effect. Provider state also
+declares authority mode, freshness, limitations, and whether the provider is a
+simulation. Credentials, transport configuration, archive-specific package
+schemas, and custody semantics remain provider-owned.
+
+Records includes `records.archive.simulation` to prove package and receipt
+handling. The simulation is explicitly non-conformant, transfers no custody,
+and cannot be used as evidence of an archive handoff. A real provider requires
+a selected target/profile, provider-specific recovery declaration, and target
+test evidence.

src/govoplan_core/core/records.py

@@ -8,6 +8,7 @@ from typing import Literal, Protocol, runtime_checkable
 
 CAPABILITY_RECORDS_FILING = "records.filing"
 CAPABILITY_RECORD_SOURCE_PREFIX = "records.source."
+CAPABILITY_RECORD_ARCHIVE_PREFIX = "records.archive."
 
 RecordSourceAuthority = Literal[
     "native_authoritative",
@@ -17,6 +18,17 @@ RecordSourceAuthority = Literal[
     "governance_overlay",
     "linked_reference",
 ]
+RecordArchiveOutcome = Literal["accepted", "rejected", "outcome_unknown"]
+
+_RECORD_SOURCE_AUTHORITIES = {
+    "native_authoritative",
+    "external_authoritative",
+    "external_mirror",
+    "governed_sync",
+    "governance_overlay",
+    "linked_reference",
+}
+_RECORD_ARCHIVE_OUTCOMES = {"accepted", "rejected", "outcome_unknown"}
 
 
 class RecordContractError(ValueError):
@@ -129,6 +141,109 @@ class RecordFilingResult:
     replayed: bool = False
 
 
+@dataclass(frozen=True, slots=True)
+class RecordTransferPackage:
+    """Exact, digest-bound package prepared by Records for one provider profile."""
+
+    tenant_id: str
+    package_id: str
+    record_id: str
+    record_revision: int
+    profile: str
+    manifest_sha256: str
+    manifest: Mapping[str, object]
+
+    def __post_init__(self) -> None:
+        _require_text_fields(
+            self,
+            "tenant_id",
+            "package_id",
+            "record_id",
+            "profile",
+            "manifest_sha256",
+        )
+        if self.record_revision < 1:
+            raise RecordContractError("Record transfer revisions must be positive.")
+        _require_sha256(self.manifest_sha256, "Record transfer manifest")
+
+
+@dataclass(frozen=True, slots=True)
+class RecordArchiveProviderState:
+    provider_id: str
+    label: str
+    profiles: tuple[str, ...]
+    authority_modes: tuple[RecordSourceAuthority, ...]
+    healthy: bool
+    checked_at: datetime
+    last_success_at: datetime | None = None
+    freshness_seconds: int | None = None
+    limitations: tuple[str, ...] = ()
+    simulated: bool = False
+
+    def __post_init__(self) -> None:
+        _require_text_fields(self, "provider_id", "label")
+        if not self.profiles or any(not item.strip() for item in self.profiles):
+            raise RecordContractError(
+                "Record archive providers require at least one profile."
+            )
+        if not self.authority_modes:
+            raise RecordContractError(
+                "Record archive providers require an authority mode."
+            )
+        if any(mode not in _RECORD_SOURCE_AUTHORITIES for mode in self.authority_modes):
+            raise RecordContractError(
+                "Record archive providers declared an invalid authority mode."
+            )
+        if self.freshness_seconds is not None and self.freshness_seconds < 0:
+            raise RecordContractError(
+                "Record archive provider freshness cannot be negative."
+            )
+
+
+@dataclass(frozen=True, slots=True)
+class RecordArchiveTransferRequest:
+    package: RecordTransferPackage
+    purpose: str
+    idempotency_key: str
+    institutional_context: Mapping[str, object] = field(default_factory=dict)
+
+    def __post_init__(self) -> None:
+        _require_text_fields(self, "purpose", "idempotency_key")
+        if len(self.purpose) > 255 or len(self.idempotency_key) > 255:
+            raise RecordContractError(
+                "Record archive purpose or idempotency key is too long."
+            )
+
+
+@dataclass(frozen=True, slots=True)
+class RecordArchiveReceipt:
+    provider_id: str
+    package_id: str
+    outcome: RecordArchiveOutcome
+    observed_at: datetime
+    receipt_sha256: str
+    external_reference: str | None = None
+    retry_safe: bool = False
+    simulated: bool = False
+    metadata: Mapping[str, object] = field(default_factory=dict)
+
+    def __post_init__(self) -> None:
+        _require_text_fields(
+            self,
+            "provider_id",
+            "package_id",
+            "outcome",
+            "receipt_sha256",
+        )
+        _require_sha256(self.receipt_sha256, "Record archive receipt")
+        if self.outcome not in _RECORD_ARCHIVE_OUTCOMES:
+            raise RecordContractError("Record archive receipt outcome is invalid.")
+        if self.outcome == "outcome_unknown" and self.retry_safe:
+            raise RecordContractError(
+                "Unknown archive outcomes cannot be declared retry-safe."
+            )
+
+
 @runtime_checkable
 class RecordSourceProvider(Protocol):
     provider_id: str
@@ -157,16 +272,32 @@ class RecordFilingService(Protocol):
     ) -> RecordFilingResult: ...
 
 
+@runtime_checkable
+class RecordArchiveProvider(Protocol):
+    provider_id: str
+
+    def state(self) -> RecordArchiveProviderState: ...
+
+    def dispatch(
+        self,
+        session: object,
+        principal: object,
+        *,
+        request: RecordArchiveTransferRequest,
+    ) -> RecordArchiveReceipt:
+        """Dispatch one prepared package without retrying an unknown outcome."""
+
+
 def record_source_capability(source_module: str) -> str:
-    normalized = source_module.strip().lower()
+    normalized = _capability_suffix(source_module, "source module")
-    if not normalized or any(
-        character not in "abcdefghijklmnopqrstuvwxyz0123456789_-"
-        for character in normalized
-    ):
-        raise RecordContractError("Record source module identifiers are invalid.")
     return f"{CAPABILITY_RECORD_SOURCE_PREFIX}{normalized}"
 
 
+def record_archive_capability(provider_id: str) -> str:
+    normalized = _capability_suffix(provider_id, "archive provider")
+    return f"{CAPABILITY_RECORD_ARCHIVE_PREFIX}{normalized}"
+
+
 def record_source_capabilities(registry: object | None) -> tuple[str, ...]:
     if registry is None or not hasattr(registry, "capability_names"):
         return ()
@@ -177,6 +308,16 @@ def record_source_capabilities(registry: object | None) -> tuple[str, ...]:
     )
 
 
+def record_archive_capabilities(registry: object | None) -> tuple[str, ...]:
+    if registry is None or not hasattr(registry, "capability_names"):
+        return ()
+    return tuple(
+        name
+        for name in registry.capability_names()
+        if str(name).startswith(CAPABILITY_RECORD_ARCHIVE_PREFIX)
+    )
+
+
 def _require_text_fields(value: object, *field_names: str) -> None:
     for field_name in field_names:
         if not str(getattr(value, field_name, "") or "").strip():
@@ -185,9 +326,33 @@ def _require_text_fields(value: object, *field_names: str) -> None:
             )
 
 
+def _capability_suffix(value: str, label: str) -> str:
+    normalized = value.strip().lower()
+    if not normalized or any(
+        character not in "abcdefghijklmnopqrstuvwxyz0123456789_-"
+        for character in normalized
+    ):
+        raise RecordContractError(f"Record {label} identifiers are invalid.")
+    return normalized
+
+
+def _require_sha256(value: str, label: str) -> None:
+    digest = value.removeprefix("sha256:")
+    if len(digest) != 64 or any(
+        character not in "0123456789abcdefABCDEF" for character in digest
+    ):
+        raise RecordContractError(f"{label} SHA-256 must be hexadecimal.")
+
+
 __all__ = [
+    "CAPABILITY_RECORD_ARCHIVE_PREFIX",
     "CAPABILITY_RECORDS_FILING",
     "CAPABILITY_RECORD_SOURCE_PREFIX",
+    "RecordArchiveOutcome",
+    "RecordArchiveProvider",
+    "RecordArchiveProviderState",
+    "RecordArchiveReceipt",
+    "RecordArchiveTransferRequest",
     "RecordContractError",
     "RecordFilingRequest",
     "RecordFilingResult",
@@ -196,6 +361,9 @@ __all__ = [
     "RecordSourceLocator",
     "RecordSourceProvider",
     "RecordSourceReference",
+    "RecordTransferPackage",
+    "record_archive_capabilities",
+    "record_archive_capability",
     "record_source_capabilities",
     "record_source_capability",
 ]

tests/test_records_contract.py

@@ -4,10 +4,16 @@ from datetime import UTC, datetime
 import unittest
 
 from govoplan_core.core.records import (
+    RecordArchiveProviderState,
+    RecordArchiveReceipt,
+    RecordArchiveTransferRequest,
     RecordContractError,
     RecordFilingRequest,
     RecordSourceLocator,
     RecordSourceReference,
+    RecordTransferPackage,
+    record_archive_capabilities,
+    record_archive_capability,
     record_source_capabilities,
     record_source_capability,
 )
@@ -15,7 +21,12 @@ from govoplan_core.core.records import (
 
 class _Registry:
     def capability_names(self) -> tuple[str, ...]:
-        return ("mail.delivery", "records.source.cases", "records.source.files")
+        return (
+            "mail.delivery",
+            "records.archive.simulation",
+            "records.source.cases",
+            "records.source.files",
+        )
 
 
 class RecordsContractTests(unittest.TestCase):
@@ -84,6 +95,72 @@ class RecordsContractTests(unittest.TestCase):
             record_source_capabilities(_Registry()),
         )
 
+    def test_archive_contract_is_digest_bound_and_discoverable(self) -> None:
+        package = RecordTransferPackage(
+            tenant_id="tenant-1",
+            package_id="package-1",
+            record_id="record-1",
+            record_revision=3,
+            profile="govoplan-simulation-v1",
+            manifest_sha256="a" * 64,
+            manifest={"record_id": "record-1", "items": []},
+        )
+        request = RecordArchiveTransferRequest(
+            package=package,
+            purpose="archive appraisal",
+            idempotency_key="transfer-1",
+        )
+        state = RecordArchiveProviderState(
+            provider_id="simulation",
+            label="Archive simulation",
+            profiles=("govoplan-simulation-v1",),
+            authority_modes=("linked_reference",),
+            healthy=True,
+            checked_at=datetime(2026, 8, 6, tzinfo=UTC),
+            simulated=True,
+        )
+
+        self.assertEqual("package-1", request.package.package_id)
+        self.assertTrue(state.simulated)
+        self.assertEqual(
+            "records.archive.simulation",
+            record_archive_capability("simulation"),
+        )
+        self.assertEqual(
+            ("records.archive.simulation",),
+            record_archive_capabilities(_Registry()),
+        )
+
+    def test_unknown_archive_outcome_cannot_be_retry_safe(self) -> None:
+        with self.assertRaisesRegex(RecordContractError, "retry-safe"):
+            RecordArchiveReceipt(
+                provider_id="archive-1",
+                package_id="package-1",
+                outcome="outcome_unknown",
+                observed_at=datetime(2026, 8, 6, tzinfo=UTC),
+                receipt_sha256="b" * 64,
+                retry_safe=True,
+            )
+
+    def test_archive_provider_runtime_values_are_validated(self) -> None:
+        with self.assertRaisesRegex(RecordContractError, "invalid authority mode"):
+            RecordArchiveProviderState(
+                provider_id="archive",
+                label="Archive",
+                profiles=("profile",),
+                authority_modes=("untrusted",),  # type: ignore[arg-type]
+                healthy=True,
+                checked_at=datetime.now(UTC),
+            )
+        with self.assertRaisesRegex(RecordContractError, "outcome is invalid"):
+            RecordArchiveReceipt(
+                provider_id="archive",
+                package_id="package-1",
+                outcome="maybe",  # type: ignore[arg-type]
+                observed_at=datetime.now(UTC),
+                receipt_sha256="a" * 64,
+            )
+
 
 if __name__ == "__main__":
     unittest.main()