# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-uncommitted-database-or-filesystem-states-must-be-rolled-back-before-exit.md
# Remnant Warrant:    https://corpusfetch.com/c/what-uncommitted-database-or-filesystem-states-must-be-rolled-back-before-exit.warrant.json
# Markdown SHA-256:   9689e3b5e527390bcd9ab7d60c0878b6ccf11fb45749681d948506221c5acb87
# Code AST SHA-256:   d977716210e8db998a0d58fb89485f2ccad6c28193df45fbecdf90888e596535
# Verified Classes:   SessionExitState, UncleanExitStateViolation, PreExitRollbackGate
# Defensive Guards:   4 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field
from typing import Final, FrozenSet, Tuple
import os
from pathlib import Path

@dataclass(frozen=True)
class SessionExitState:
    session_id: str
    committed: bool
    uncommitted_db_transactions: Tuple[str, ...]
    staged_temp_files: Tuple[str, ...]
    acquired_lockfiles: Tuple[str, ...]

class UncleanExitStateViolation(Exception):
    """Raised when uncommitted database transactions, temp files, or lockfiles remain at process exit."""
    pass

class PreExitRollbackGate:
    def __init__(self, allowed_staging_root: str = "/var/www/corpusfetch") -> None:
        self.allowed_staging_root: Final[str] = allowed_staging_root

    def enforce_clean_exit(
        self,
        state: SessionExitState,
        rolled_back_dbs: FrozenSet[str],
        removed_paths: FrozenSet[str],
    ) -> bool:
        if not state.committed:
            # 1. Verify all uncommitted DB transactions were explicitly rolled back
            for tx_id in state.uncommitted_db_transactions:
                if tx_id not in rolled_back_dbs:
                    raise UncleanExitStateViolation(
                        f"Session {state.session_id} exiting with un-rolled-back DB transaction: {tx_id}"
                    )

            # 2. Verify all unpromoted staging files were unlinked
            for temp_path in state.staged_temp_files:
                if not temp_path.startswith(self.allowed_staging_root):
                    raise UncleanExitStateViolation(
                        f"Staging path '{temp_path}' escapes allowed root '{self.allowed_staging_root}'."
                    )
                if temp_path not in removed_paths or Path(temp_path).exists():
                    raise UncleanExitStateViolation(
                        f"Session {state.session_id} left unlinked staging file on disk: {temp_path}"
                    )

        # 3. Lockfiles must ALWAYS be released prior to exit, whether committed or aborted
        for lock_path in state.acquired_lockfiles:
            if lock_path not in removed_paths or os.path.exists(lock_path):
                raise UncleanExitStateViolation(
                    f"Session {state.session_id} failed to release advisory lockfile before exit: {lock_path}"
                )

        return True