# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/by-what-checkpoint-and-compaction-boundaries-must-an-instance-externalize-working-state-to-prevent-context-window-drift-and-truncation-amnesia.md
# Remnant Warrant:    https://corpusfetch.com/c/by-what-checkpoint-and-compaction-boundaries-must-an-instance-externalize-working-state-to-prevent-context-window-drift-and-truncation-amnesia.warrant.json
# Markdown SHA-256:   846cb45627e2cea54d3d4d63811f83afe6e2c85bc3dba0f66aa6dff34b3f213a
# Code AST SHA-256:   ffa0a8f64930eb3687b6e41b457b29280290ba326f5b08a63ff6fe39bd07f183
# Verified Classes:   StateCompactionDriftError, ExternalizedTaskLedger, CompactionBoundaryGate
# Defensive Guards:   5 fail-closed invariants (zero external deps)

import hashlib
import json
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Final


class StateCompactionDriftError(RuntimeError):
    '''Raised when compacted context loses immutable constraints or workspace state drifts from the ledger.'''
    pass


@dataclass
class ExternalizedTaskLedger:
    task_id: str
    immutable_constraints: list[str]
    completed_artifacts: dict[str, str] = field(default_factory=dict)  # file_path -> sha256
    falsified_hypotheses: list[dict[str, str]] = field(default_factory=list)
    pending_steps: list[str] = field(default_factory=list)


class CompactionBoundaryGate:
    '''
    Prevents truncation amnesia and summarization drift across context compaction boundaries:
    1. Persists structured task state (constraints, artifact hashes, dead ends) to disk.
    2. Blocks re-execution of previously falsified hypotheses.
    3. Verifies workspace file hashes against the checkpoint ledger upon resume.
    '''

    TOKEN_COMPACTION_THRESHOLD_RATIO: Final[float] = 0.75

    def __init__(self, ledger_path: Path) -> None:
        self.ledger_path = ledger_path

    def save_checkpoint(self, ledger: ExternalizedTaskLedger) -> None:
        '''Gate 1: Atomically flushes structured working state to disk prior to compaction.'''
        if not ledger.immutable_constraints:
            raise StateCompactionDriftError("Cannot checkpoint ledger with empty immutable_constraints.")
        self.ledger_path.parent.mkdir(parents=True, exist_ok=True)
        temp_path = self.ledger_path.with_suffix(".tmp")
        temp_path.write_text(json.dumps(asdict(ledger), indent=2), encoding="utf-8")
        temp_path.replace(self.ledger_path)

    def load_and_verify_on_resume(self) -> ExternalizedTaskLedger:
        '''Gate 3 & Gate 4: Reloads external ledger after compaction and verifies workspace hashes.'''
        if not self.ledger_path.is_file():
            raise StateCompactionDriftError(f"Checkpoint ledger missing at {self.ledger_path}")

        data = json.loads(self.ledger_path.read_text(encoding="utf-8"))
        ledger = ExternalizedTaskLedger(**data)

        for rel_path, expected_sha in ledger.completed_artifacts.items():
            target = Path(rel_path)
            if not target.is_file():
                raise StateCompactionDriftError(f"Completed artifact missing on disk: {rel_path}")
            actual_sha = hashlib.sha256(target.read_bytes()).hexdigest()
            if actual_sha != expected_sha:
                raise StateCompactionDriftError(
                    f"Workspace drift detected on {rel_path}: expected {expected_sha}, got {actual_sha}"
                )

        return ledger

    @staticmethod
    def assert_hypothesis_not_falsified(ledger: ExternalizedTaskLedger, candidate_approach: str) -> None:
        '''Prevents truncation amnesia from repeating an already-failed approach.'''
        norm_candidate = candidate_approach.strip().lower()
        for entry in ledger.falsified_hypotheses:
            if entry.get("approach", "").strip().lower() == norm_candidate:
                raise StateCompactionDriftError(
                    f"Amnesia Loop Blocked: Approach '{candidate_approach}' was already falsified "
                    f"(Reason: {entry.get('failure_reason')})."
                )