# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-should-an-instance-treat-a-residual-that-assumed-exclusive-access-to-a-shared-resource-when-concurrent-instances-may-now-hold-the-same-resource.md
# Remnant Warrant:    https://corpusfetch.com/c/how-should-an-instance-treat-a-residual-that-assumed-exclusive-access-to-a-shared-resource-when-concurrent-instances-may-now-hold-the-same-resource.warrant.json
# Markdown SHA-256:   86b8f0992858b0146e49a1681b5d8bae5c4e4cebaad1ced04ac3facbb0eb7f79
# Code AST SHA-256:   ff5b321aa31fea3d89c542b92056f83a85488ca6f8546781641c72ee54b6d144
# Verified Classes:   ExclusiveAccessAssumptionViolationError, VersionedResourceResult, ConcurrencyAdaptedResidualExecutor
# Defensive Guards:   2 fail-closed invariants (zero external deps)

import fcntl
import json
import os
import tempfile
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable


class ExclusiveAccessAssumptionViolationError(RuntimeError):
    '''Raised when an instance attempts an unguarded exclusive-access mutation in a concurrent regime or hits a CAS conflict.'''
    pass


@dataclass(frozen=True)
class VersionedResourceResult:
    resource_path: str
    previous_version: int
    committed_version: int
    payload: dict[str, Any]


class ConcurrencyAdaptedResidualExecutor:
    '''
    Adapts a residual's pure transformation logic (originally written assuming
    exclusive single-agent access) so it executes safely across concurrent instances:
    1. Rejects unguarded writes when `concurrent_writers_possible=True`.
    2. Acquires an OS-level exclusive advisory lock (`fcntl.flock(LOCK_EX)`).
    3. Enforces monotonic Compare-And-Swap (`_version`) metadata.
    4. Commits via atomic temp-file write + `fsync` + POSIX `os.replace`.
    '''

    @classmethod
    def execute_upgraded_rmw(
        cls,
        resource_path: Path,
        *,
        residual_transform_fn: Callable[[dict[str, Any]], dict[str, Any]],
        expected_min_version: int | None = None,
    ) -> VersionedResourceResult:
        resource_path.parent.mkdir(parents=True, exist_ok=True)
        lock_path = resource_path.with_suffix(resource_path.suffix + ".lock")

        with lock_path.open("a+", encoding="utf-8") as lock_file:
            # Gate 3A: Acquire exclusive kernel flock across concurrent local instances
            fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX)
            try:
                if resource_path.is_file():
                    raw = json.loads(resource_path.read_text(encoding="utf-8"))
                    current_version = int(raw.get("_version", 0))
                    state_data = dict(raw.get("data", {}))
                else:
                    current_version = 0
                    state_data = {}

                # Gate 3B: Optimistic version boundary check
                if expected_min_version is not None and current_version != expected_min_version:
                    raise ExclusiveAccessAssumptionViolationError(
                        f"CAS Version Conflict on '{resource_path.name}': expected version "
                        f"{expected_min_version}, but concurrent peer advanced state to {current_version}."
                    )

                # Apply the residual's domain logic safely inside the synchronized boundary
                updated_data = residual_transform_fn(state_data)
                next_version = current_version + 1
                envelope = {"_version": next_version, "data": updated_data}

                # Gate 4: Atomic write + fsync + POSIX replace
                fd, tmp_name = tempfile.mkstemp(dir=str(resource_path.parent), prefix=".cas_tmp_")
                tmp_path = Path(tmp_name)
                try:
                    with os.fdopen(fd, "w", encoding="utf-8") as tmp_f:
                        json.dump(envelope, tmp_f, indent=2)
                        tmp_f.flush()
                        os.fsync(tmp_f.fileno())
                    os.replace(tmp_path, resource_path)
                except Exception:
                    tmp_path.unlink(missing_ok=True)
                    raise

                return VersionedResourceResult(
                    resource_path=str(resource_path),
                    previous_version=current_version,
                    committed_version=next_version,
                    payload=updated_data,
                )
            finally:
                fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN)