ss-tools/scripts/dr_recovery_contract.py

# #region Tooling.DrRecovery.Contract [C:4] [TYPE Module] [SEMANTICS dr,recovery,git,holds]
# @BRIEF Closed recovery sidecar supplements the unchanged v1 manifest with exact source records.
# @RELATION DEPENDS_ON -> [Tooling.DrRecovery.Contract.Primitives]
# @RELATION DEPENDS_ON -> [Tooling.DrRecovery.Contract.Scalars]
# @PRE Inputs are reviewed captures; validation never discovers a configured source or queries a DB.
# @POST Every required component is explicit; malformed/missing authority refuses without a default.
# @RELATION DEPENDS_ON -> [Storage.NativeResultHold]
# @RELATION DEPENDS_ON -> [ScenarioExecution.PublicationConfig.Destination]
# @RATIONALE A separate format preserves the accepted exact manifest schema and its old assertions.
# @REJECTED Optional recovery components or guessed clone paths would turn unavailable evidence into PASS.
from datetime import datetime, timedelta
from pathlib import Path
import sys
from typing import Annotated, Literal

from pydantic import Field, field_validator, model_validator

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "backend"))
from src.schemas.dashboard_testing.result_descriptor import Sha256 as Sha256, UuidString  # noqa: E402
from src.services.dashboard_testing.execution.publication_config import (
    PublicationDestination,
)  # noqa: E402
from src.services.storage_holds import HoldIndex  # noqa: E402

from dr_recovery_contract_types import Text as Text, Commit as Commit  # noqa: E402


from dr_recovery_contract_base import Closed as Closed, Stat as Stat, Artifact as Artifact, aware  # noqa: E402


# #region Tooling.DrRecovery.Contract.Catalog [C:1] [TYPE Class] [SEMANTICS dr,recovery]
class Catalog(Closed):
    catalog_id: UuidString
    catalog_revision_id: UuidString
    catalog_digest: Sha256
    integrity_sha256: Sha256
# #endregion Tooling.DrRecovery.Contract.Catalog


# #region Tooling.DrRecovery.Contract.Cataloghead [C:1] [TYPE Class] [SEMANTICS dr,recovery]
class CatalogHead(Closed):
    catalog_id: UuidString
    repository: Text
# #endregion Tooling.DrRecovery.Contract.Cataloghead


# #region Tooling.DrRecovery.Contract.Approval [C:1] [TYPE Class] [SEMANTICS dr,recovery]
class Approval(Closed):
    receipt_id: UuidString
    receipt_sha256: Sha256
# #endregion Tooling.DrRecovery.Contract.Approval


# #region Tooling.DrRecovery.Contract.Receipt [C:1] [TYPE Class] [SEMANTICS dr,recovery]
class Receipt(Closed):
    catalog_revision_id: UuidString
    integrity_sha256: Sha256
    render_sha256: Sha256
    commit_sha: Commit
    head_after: Commit
    verified_at: Text

    _aware = field_validator("verified_at")(aware)
# #endregion Tooling.DrRecovery.Contract.Receipt


# #region Tooling.DrRecovery.Contract.Publication [C:3] [TYPE Class]
class Publication(Closed):
    id: UuidString
    catalog_id: UuidString
    catalog_revision_id: UuidString
    repository: Text
    destination: PublicationDestination
    catalog_digest: Sha256
    integrity_sha256: Sha256
    render_sha256: Sha256
    expected_branch_head: Commit
    state: Literal["published"]
    error_code: None
    commit_sha: Commit
    published_receipt: Receipt

    # #region Tooling.DrRecovery.Contract.Publication.ExactBinding [C:3] [TYPE Function] [SEMANTICS dr,recovery]
    @model_validator(mode="after")
    def exact_binding(self):
        expected = dict(
            catalog_revision_id=self.catalog_revision_id,
            integrity_sha256=self.integrity_sha256,
            render_sha256=self.render_sha256,
            commit_sha=self.commit_sha,
            head_after=self.commit_sha,
            verified_at=self.published_receipt.verified_at,
        )
        if (
            self.published_receipt.model_dump() != expected
            or self.destination.path
            != f"catalogs/{self.catalog_id}/{self.catalog_revision_id}.yaml"
        ):
            raise ValueError("receipt binding")
        return self
    # #endregion Tooling.DrRecovery.Contract.Publication.ExactBinding


# #endregion Tooling.DrRecovery.Contract.Publication


# #region Tooling.DrRecovery.Contract.Holdreasons [C:3] [TYPE Class] [SEMANTICS dr,recovery]
class HoldReasons(Closed):
    reasons: tuple[
        Literal[
            "approved_baseline",
            "active_operation",
            "investigation_case",
            "analytics_window",
            "shared_reference",
        ],
        ...,
    ]
    at: Text
    _aware = field_validator("at")(aware)

    # #region Tooling.DrRecovery.Contract.Holdreasons.DistinctReasons [C:3] [TYPE Function] [SEMANTICS dr,recovery]
    @field_validator("reasons")
    @classmethod
    def distinct_reasons(cls, value):
        if len(set(value)) != len(value):
            raise ValueError("duplicate reason")
        return value
    # #endregion Tooling.DrRecovery.Contract.Holdreasons.DistinctReasons
# #endregion Tooling.DrRecovery.Contract.Holdreasons


# #region Tooling.DrRecovery.Contract.Retention [C:3] [TYPE Class] [SEMANTICS dr,recovery]
class Retention(Closed):
    id: UuidString
    target_type: Literal["artifact", "run_metadata", "step_metrics", "notification"]
    target_id: Text
    content_ref: str | None
    scenario_id: UuidString | None
    run_id: UuidString | None
    baseline_pin: dict | None
    retention_class: Text
    state: Literal["deletion_pending", "eligible", "deleted", "tombstoned"]
    holds_snapshot: HoldReasons
    attempts: int = Field(ge=0)
    error_code: str | None
    verified_absent_at: str | None
    tombstoned_at: str | None
    created_at: Text
    updated_at: Text
    _created = field_validator("created_at")(aware)
    _updated = field_validator("updated_at")(aware)

    # #region Tooling.DrRecovery.Contract.Retention.OptionalTimestamp [C:3] [TYPE Function] [SEMANTICS dr,recovery]
    @field_validator("verified_absent_at", "tombstoned_at")
    @classmethod
    def optional_timestamp(cls, value):
        return aware(value) if value is not None else None
    # #endregion Tooling.DrRecovery.Contract.Retention.OptionalTimestamp
# #endregion Tooling.DrRecovery.Contract.Retention


# #region Tooling.DrRecovery.Contract.State [C:3] [TYPE Class] [SEMANTICS dr,recovery]
class State(Closed):
    target: dict[str, str]
    alembic_version: Text
    catalog_heads: tuple[CatalogHead, ...] = Field(min_length=1, max_length=10000)
    catalog_revisions: tuple[Catalog, ...] = Field(min_length=1, max_length=10000)
    approval_receipts: tuple[Approval, ...] = Field(min_length=1, max_length=10000)
    publication_operations: tuple[Publication, ...] = Field(
        min_length=1, max_length=10000
    )
    retention_deletions: tuple[Retention, ...] = Field(min_length=1, max_length=10000)

    # #region Tooling.DrRecovery.Contract.State.ExactTarget [C:3] [TYPE Function] [SEMANTICS dr,recovery]
    @field_validator("target")
    @classmethod
    def exact_target(cls, value):
        from dr_backup_preflight import _target_ok

        if not _target_ok(value):
            raise ValueError("invalid target")
        return value
    # #endregion Tooling.DrRecovery.Contract.State.ExactTarget
# #endregion Tooling.DrRecovery.Contract.State


# #region Tooling.DrRecovery.Contract.Ref [C:1] [TYPE Class] [SEMANTICS dr,recovery]
class Ref(Closed):
    name: Annotated[str, Field(pattern=r"^refs/heads/[A-Za-z0-9][A-Za-z0-9._/-]*$")]
    commit_sha: Commit
# #endregion Tooling.DrRecovery.Contract.Ref


# #region Tooling.DrRecovery.Contract.Source [C:1] [TYPE Class] [SEMANTICS dr,recovery]
class Source(Closed):
    source_id: UuidString
    base_url: Text
    repository: Text
    reviewed_by: Text
    approved_at: Text
    object_format: Literal["sha1", "sha256"]
    bundle: Artifact
    refs: tuple[Ref, ...] = Field(min_length=1, max_length=256)
    prerequisites: tuple[Commit, ...] = Field(max_length=256)
    operation_ids: tuple[UuidString, ...] = Field(min_length=1, max_length=10000)
    _aware = field_validator("approved_at")(aware)
# #endregion Tooling.DrRecovery.Contract.Source


# #region Tooling.DrRecovery.Contract.Render [C:1] [TYPE Class] [SEMANTICS dr,recovery]
class Render(Closed):
    operation_id: UuidString
    artifact: Artifact
# #endregion Tooling.DrRecovery.Contract.Render


# #region Tooling.DrRecovery.Contract.Holds [C:1] [TYPE Class] [SEMANTICS dr,recovery]
class Holds(Closed):
    activated: Stat
    index: Stat
    records: HoldIndex
# #endregion Tooling.DrRecovery.Contract.Holds


# #region Tooling.DrRecovery.Contract.Sidecar [C:3] [TYPE Class]
class Sidecar(Closed):
    schema_version: Literal[1]
    manifest_sha256: Sha256
    t_snap: Text
    source_state: State
    git_sources: tuple[Source, ...] = Field(min_length=1, max_length=256)
    renders: tuple[Render, ...] = Field(min_length=1, max_length=10000)
    native_holds: Holds
    _aware = field_validator("t_snap")(aware)

    # #region Tooling.DrRecovery.Contract.Sidecar.UtcSnapshot [C:3] [TYPE Function] [SEMANTICS dr,recovery]
    @field_validator("t_snap")
    @classmethod
    def utc_snapshot(cls, value):
        if datetime.fromisoformat(value).utcoffset() != timedelta(0):
            raise ValueError("UTC snapshot required")
        return value
    # #endregion Tooling.DrRecovery.Contract.Sidecar.UtcSnapshot

    # #region Tooling.DrRecovery.Contract.Sidecar.NonemptyNativeHolds [C:3] [TYPE Function] [SEMANTICS dr,recovery]
    @model_validator(mode="after")
    def nonempty_native_holds(self):
        holds = self.native_holds.records.holds
        if not holds or len({hold.hold_id for hold in holds}) != len(holds):
            raise ValueError("nonempty unique holds required")
        return self
    # #endregion Tooling.DrRecovery.Contract.Sidecar.NonemptyNativeHolds


# #endregion Tooling.DrRecovery.Contract.Sidecar
# #endregion Tooling.DrRecovery.Contract