"""Backend-neutral patch plan models for verified tabular repairs.""" from __future__ import annotations import hashlib import json import secrets from datetime import UTC, datetime from typing import Literal from pydantic import BaseModel, ConfigDict, Field from dataforge.transactions.txn import CellFix PATCH_PLAN_SCHEMA_VERSION: Literal["patch_plan_v1"] = "patch_plan_v1" class RowIdentity(BaseModel): """Stable row locator required before a table-store repair can apply.""" kind: Literal["csv_position", "column_values", "unavailable"] columns: tuple[str, ...] = Field(default_factory=tuple) values: dict[str, str] = Field(default_factory=dict) stable: bool = False reason: str = Field(min_length=1) model_config = ConfigDict(extra="forbid", frozen=True) class PatchOperation(BaseModel): """One cell-level mutation in a backend-neutral repair plan.""" operation: Literal["update"] = "update" relation: str = Field(min_length=1) row: int = Field(ge=0) row_identity: RowIdentity column: str = Field(min_length=1) old_value: str new_value: str detector_id: str = Field(min_length=1) reason: str = Field(min_length=1) confidence: float = Field(ge=0.0, le=1.0) provenance: str = Field(min_length=1) precondition_sql: str | None = None forward_sql: str | None = None rollback_sql: str | None = None verification_sql: tuple[str, ...] = Field(default_factory=tuple) model_config = ConfigDict(extra="forbid", frozen=True) @classmethod def from_cell_fix( cls, fix: CellFix, *, relation: str, row_identity: RowIdentity, reason: str, confidence: float, provenance: str, precondition_sql: str | None = None, forward_sql: str | None = None, rollback_sql: str | None = None, verification_sql: tuple[str, ...] = (), ) -> PatchOperation: """Build an operation from an existing DataForge cell fix.""" if fix.operation != "update": raise ValueError("PatchPlan v1 supports cell updates only.") return cls( relation=relation, row=fix.row, row_identity=row_identity, column=fix.column, old_value=fix.old_value, new_value=fix.new_value, detector_id=fix.detector_id, reason=reason, confidence=confidence, provenance=provenance, precondition_sql=precondition_sql, forward_sql=forward_sql, rollback_sql=rollback_sql, verification_sql=verification_sql, ) class CostEstimate(BaseModel): """Small, backend-agnostic estimate surfaced before mutation.""" rows_scanned: int = Field(ge=0) rows_written: int = Field(ge=0) bytes_scanned: int | None = Field(default=None, ge=0) quota_units: float = Field(default=0.0, ge=0.0) model_config = ConfigDict(extra="forbid", frozen=True) class PatchPlan(BaseModel): """The only write contract accepted by non-CSV DataForge stores.""" schema_version: Literal["patch_plan_v1"] = PATCH_PLAN_SCHEMA_VERSION plan_id: str = Field(pattern=r"^plan-[0-9a-f]{12}$") created_at: datetime backend: str = Field(min_length=1) target: str = Field(min_length=1) relation: str = Field(min_length=1) row_identity_columns: tuple[str, ...] = Field(default_factory=tuple) stable_row_identity: bool operations: tuple[PatchOperation, ...] = Field(default_factory=tuple) forward_sql: tuple[str, ...] = Field(default_factory=tuple) rollback_sql: tuple[str, ...] = Field(default_factory=tuple) preflight_probes: tuple[str, ...] = Field(default_factory=tuple) verification_queries: tuple[str, ...] = Field(default_factory=tuple) touched_constraints: tuple[str, ...] = Field(default_factory=tuple) smt_obligations: tuple[str, ...] = Field(default_factory=tuple) cost_estimate: CostEstimate safety_verdict: str = Field(min_length=1) reversible: bool apply_supported: bool apply_requires_approval: bool = True audit_metadata: dict[str, str] = Field(default_factory=dict) reason: str = Field(min_length=1) model_config = ConfigDict(extra="forbid", frozen=True) @classmethod def new( cls, *, backend: str, target: str, relation: str, row_identity_columns: tuple[str, ...], operations: tuple[PatchOperation, ...], safety_verdict: str, rows_scanned: int, reason: str, touched_constraints: tuple[str, ...] = (), smt_obligations: tuple[str, ...] = (), audit_metadata: dict[str, str] | None = None, apply_supported: bool | None = None, reversible: bool | None = None, ) -> PatchPlan: """Construct a stable plan with derived SQL and support flags.""" stable = bool(row_identity_columns) and all(op.row_identity.stable for op in operations) has_operations = bool(operations) supported = stable and has_operations if apply_supported is None else apply_supported is_reversible = supported if reversible is None else reversible return cls( plan_id=f"plan-{secrets.token_hex(6)}", created_at=datetime.now(UTC), backend=backend, target=target, relation=relation, row_identity_columns=row_identity_columns, stable_row_identity=stable, operations=operations, forward_sql=tuple(sql for op in operations if (sql := op.forward_sql)), rollback_sql=tuple(sql for op in operations if (sql := op.rollback_sql)), preflight_probes=tuple(sql for op in operations if (sql := op.precondition_sql)), verification_queries=tuple(query for op in operations for query in op.verification_sql), touched_constraints=touched_constraints, smt_obligations=smt_obligations, cost_estimate=CostEstimate(rows_scanned=rows_scanned, rows_written=len(operations)), safety_verdict=safety_verdict, reversible=is_reversible, apply_supported=supported, audit_metadata=audit_metadata or {}, reason=reason, ) def canonical_json(self) -> str: """Return deterministic JSON suitable for audit hashing.""" return json.dumps(self.model_dump(mode="json"), sort_keys=True, separators=(",", ":")) def sha256(self) -> str: """Return a SHA-256 digest of the canonical plan.""" return hashlib.sha256(self.canonical_json().encode("utf-8")).hexdigest()