from contextlib import contextmanager
import ctypes
from dataclasses import dataclass, field
from enum import Enum
import errno
import hashlib
import hmac
import os
import posixpath
import re
import secrets
import stat
import sys
import tempfile
import threading

try:
    import fcntl
except ImportError:  # pragma: no cover - non-Linux import portability
    fcntl = None


MAX_MANAGED_ROOTS = 8
MAX_MANAGED_ROOT_ID_BYTES = 64
MAX_MANAGED_ROOT_PATH_BYTES = 4096
MAX_RELATIVE_PATH_BYTES = 4096
MAX_COMPONENT_BYTES = 255
MAX_PATH_DEPTH = 32
MAX_LISTING_ENTRIES = 1000
MAX_LISTING_BYTES = 1024 * 1024
MAX_FILE_BYTES = 64 * 1024 * 1024
MAX_RESULT_FILE_BYTES = 256 * 1024 * 1024

RUNTIME_LOG_ROOT_ID = 'runtime-logs'
RUNTIME_LOG_ROOT_PATH = '/data/runtime-linux/logs'
RUNTIME_KEYCHECK_ROOT_ID = 'runtime-keychecks'
RUNTIME_KEYCHECK_ROOT_PATH = '/data/runtime-linux/keychecks'
RUNTIME_RESULT_ROOT_ID = 'runtime-results'
RUNTIME_RESULT_ROOT_PATH = '/data/runtime-linux/results'
MANAGED_DATA_ROOT = '/data/managed-files'

_PREDEFINED_READ_ONLY_ROOTS = {
    RUNTIME_KEYCHECK_ROOT_ID: RUNTIME_KEYCHECK_ROOT_PATH,
    RUNTIME_LOG_ROOT_ID: RUNTIME_LOG_ROOT_PATH,
    RUNTIME_RESULT_ROOT_ID: RUNTIME_RESULT_ROOT_PATH,
}

_ROOT_ID = re.compile(r'^[a-z][a-z0-9-]{0,63}$')
_DRIVE_PATH = re.compile(r'^/?[A-Za-z]:')
_DRIVE_COMPONENT = re.compile(r'^[A-Za-z]:')
_HASH = re.compile(r'^[0-9a-f]{64}$')
_RESULT_PROJECTION_FILE = re.compile(
    r'^(?:found_secrets|scan_results)(?:\.g[0-9]{6})?\.jsonl$',
)
_TEMPORARY_PREFIX = '.truf-managed-file-'
_TEMPORARY_ATTEMPTS = 16
_READ_CHUNK_BYTES = 64 * 1024
_RENAME_NOREPLACE = 1
_PERMISSION_KEYS = frozenset({'list', 'read', 'create_replace', 'delete'})
_LIMIT_BOUNDS = {
    'max_relative_path_bytes': MAX_RELATIVE_PATH_BYTES,
    'max_component_bytes': MAX_COMPONENT_BYTES,
    'max_path_depth': MAX_PATH_DEPTH,
    'max_listing_entries': MAX_LISTING_ENTRIES,
    'max_listing_bytes': MAX_LISTING_BYTES,
    'max_file_bytes': MAX_FILE_BYTES,
}


class ManagedFileConfigurationError(ValueError):
    def __init__(self, category, field):
        self.category = category
        self.field = tuple(field)
        super().__init__('managed file root configuration is invalid')


class ManagedFileAccessError(RuntimeError):
    def __init__(self, category):
        self.category = category
        super().__init__('managed file access failed')


class ManagedFileOperation(str, Enum):
    LIST = 'list'
    READ = 'read'
    CREATE_REPLACE = 'create-replace'
    DELETE = 'delete'


@dataclass(frozen=True)
class ManagedFilePermissions:
    allow_list: bool
    allow_read: bool
    allow_create_replace: bool
    allow_delete: bool

    def allows(self, operation):
        if operation is ManagedFileOperation.LIST:
            return self.allow_list
        if operation is ManagedFileOperation.READ:
            return self.allow_read
        if operation is ManagedFileOperation.CREATE_REPLACE:
            return self.allow_create_replace
        if operation is ManagedFileOperation.DELETE:
            return self.allow_delete
        return False


@dataclass(frozen=True)
class ManagedFileLimits:
    max_relative_path_bytes: int
    max_component_bytes: int
    max_path_depth: int
    max_listing_entries: int
    max_listing_bytes: int
    max_file_bytes: int


@dataclass(frozen=True)
class ManagedFileRoot:
    root_id: str
    absolute_path: str = field(repr=False)
    permissions: ManagedFilePermissions
    limits: ManagedFileLimits


@dataclass(frozen=True)
class ManagedFileRootRegistry:
    roots: tuple[ManagedFileRoot, ...] = ()

    def get(self, root_id):
        if type(root_id) is not str:
            return None
        return next((root for root in self.roots if root.root_id == root_id), None)

    def root_ids(self):
        return tuple(root.root_id for root in self.roots)


@dataclass(frozen=True)
class ManagedFileIdentity:
    sha256: str
    byte_count: int


@dataclass(frozen=True)
class ManagedFileDirectoryEntry:
    name: str
    kind: str
    byte_count: int | None


@dataclass(frozen=True)
class ManagedFileListing:
    entries: tuple[ManagedFileDirectoryEntry, ...]
    name_bytes: int


class ManagedFileSnapshot:
    __slots__ = ('_handle', '_release', '_lock')

    def __init__(self, handle, release):
        self._handle = handle
        self._release = release
        self._lock = threading.Lock()

    def chunks(self):
        try:
            while True:
                with self._lock:
                    handle = self._handle
                    chunk = None if handle is None else handle.read(_READ_CHUNK_BYTES)
                if not chunk:
                    break
                yield chunk
        finally:
            self.close()

    def close(self):
        with self._lock:
            handle = self._handle
            release = self._release
            self._handle = None
            self._release = None
        if handle is not None:
            try:
                handle.close()
            except OSError:
                pass
        if release is not None:
            release(self)

    def __repr__(self):
        return '<ManagedFileSnapshot>'


@dataclass(frozen=True)
class ManagedFileDownload:
    identity: ManagedFileIdentity
    content: bytes | None = field(repr=False, default=None)
    snapshot: ManagedFileSnapshot | None = field(repr=False, default=None)

    def __post_init__(self):
        if (self.content is None) == (self.snapshot is None):
            raise ValueError('managed file download requires one body source')


@dataclass(frozen=True)
class ManagedFileMutation:
    before: ManagedFileIdentity | None
    after: ManagedFileIdentity | None
    written: bool


class ManagedFileOpenedTarget:
    __slots__ = ('_descriptor', '_details')

    def __init__(self, descriptor, details):
        self._descriptor = descriptor
        self._details = details

    def fileno(self):
        if self._descriptor is None:
            raise ManagedFileAccessError('closed')
        return self._descriptor

    @property
    def details(self):
        return self._details

    def _close(self):
        descriptor = self._descriptor
        self._descriptor = None
        if descriptor is not None:
            try:
                os.close(descriptor)
            except OSError:
                pass

    def __repr__(self):
        return '<ManagedFileOpenedTarget>'


def _error(category, *field):
    raise ManagedFileConfigurationError(category, ('root',) + field)


def _absolute_root_path(value):
    if type(value) is not str:
        _error('type', 'path')
    try:
        encoded = value.encode('utf-8')
    except UnicodeEncodeError:
        _error('bounds', 'path')
    if (
        not value or len(encoded) > MAX_MANAGED_ROOT_PATH_BYTES
        or '\\' in value or '\x00' in value or _DRIVE_PATH.match(value)
        or not value.startswith('/') or value == '/' or value.endswith('/')
        or '//' in value
    ):
        _error('deployment_path', 'path')
    components = value.split('/')[1:]
    if any(
        component in ('', '.', '..')
        or len(component.encode('utf-8')) > MAX_COMPONENT_BYTES
        for component in components
    ) or posixpath.normpath(value) != value:
        _error('deployment_path', 'path')
    return value


def _permissions(value):
    if type(value) is not dict:
        _error('type', 'permissions')
    if set(value) != _PERMISSION_KEYS:
        category = 'unknown_key' if set(value) - _PERMISSION_KEYS else 'schema'
        _error(category, 'permissions')
    if any(type(value[name]) is not bool for name in _PERMISSION_KEYS):
        _error('type', 'permissions')
    return ManagedFilePermissions(
        allow_list=value['list'],
        allow_read=value['read'],
        allow_create_replace=value['create_replace'],
        allow_delete=value['delete'],
    )


def _limits(value, root_id):
    if type(value) is not dict:
        _error('type', 'limits')
    expected = set(_LIMIT_BOUNDS)
    if set(value) != expected:
        category = 'unknown_key' if set(value) - expected else 'schema'
        _error(category, 'limits')
    for name, maximum in _LIMIT_BOUNDS.items():
        if name == 'max_file_bytes' and root_id == RUNTIME_RESULT_ROOT_ID:
            maximum = MAX_RESULT_FILE_BYTES
        if type(value[name]) is not int or not 1 <= value[name] <= maximum:
            _error('bounds', 'limits', name)
    if (
        value['max_component_bytes'] > value['max_relative_path_bytes']
        or value['max_listing_bytes'] < value['max_component_bytes']
    ):
        _error('bounds', 'limits')
    return ManagedFileLimits(**{name: value[name] for name in _LIMIT_BOUNDS})


def _allowed_root(root_id, path, permissions):
    predefined_path = _PREDEFINED_READ_ONLY_ROOTS.get(root_id)
    predefined_id = next((
        candidate_id for candidate_id, candidate_path
        in _PREDEFINED_READ_ONLY_ROOTS.items()
        if candidate_path == path
    ), None)
    if predefined_path is not None or predefined_id is not None:
        if predefined_path != path or permissions != ManagedFilePermissions(
            allow_list=True,
            allow_read=True,
            allow_create_replace=False,
            allow_delete=False,
        ):
            _error('deployment_path', 'path')
        return
    parent, name = posixpath.split(path)
    if parent != MANAGED_DATA_ROOT or not name:
        _error('deployment_path', 'path')


def normalize_managed_file_roots(value):
    if type(value) is not dict:
        raise ManagedFileConfigurationError('type', ('root',))
    if len(value) > MAX_MANAGED_ROOTS:
        raise ManagedFileConfigurationError('bounds', ('root',))

    if any(type(root_id) is not str for root_id in value):
        _error('bounds', 'id')
    normalized = {}
    roots = []
    paths = set()
    for root_id in sorted(value):
        if (
            type(root_id) is not str or not _ROOT_ID.fullmatch(root_id)
            or len(root_id.encode('utf-8')) > MAX_MANAGED_ROOT_ID_BYTES
        ):
            _error('bounds', 'id')
        raw = value[root_id]
        if type(raw) is not dict:
            _error('type')
        expected = {'path', 'permissions', 'limits'}
        if set(raw) != expected:
            category = 'unknown_key' if set(raw) - expected else 'schema'
            _error(category)
        path = _absolute_root_path(raw['path'])
        permissions = _permissions(raw['permissions'])
        limits = _limits(raw['limits'], root_id)
        _allowed_root(root_id, path, permissions)
        if path in paths:
            _error('deployment_path', 'path')
        paths.add(path)
        root = ManagedFileRoot(root_id, path, permissions, limits)
        roots.append(root)
        normalized[root_id] = {
            'path': path,
            'permissions': {
                'list': permissions.allow_list,
                'read': permissions.allow_read,
                'create_replace': permissions.allow_create_replace,
                'delete': permissions.allow_delete,
            },
            'limits': {name: getattr(limits, name) for name in _LIMIT_BOUNDS},
        }
    return normalized, ManagedFileRootRegistry(tuple(roots))


def managed_file_root_registry_from_config(config):
    if type(config) is not dict:
        raise ManagedFileConfigurationError('type', ('root',))
    supervisor = config.get('supervisor', {})
    if type(supervisor) is not dict:
        raise ManagedFileConfigurationError('type', ('root',))
    worker_api = supervisor.get('worker_api', {})
    if type(worker_api) is not dict:
        raise ManagedFileConfigurationError('type', ('root',))
    admin = worker_api.get('admin', {})
    if admin is None:
        admin = {}
    if type(admin) is not dict:
        raise ManagedFileConfigurationError('type', ('root',))
    return normalize_managed_file_roots(admin.get('managed_file_roots', {}))[1]


def parse_managed_relative_path(value, limits):
    if type(value) is not str or not isinstance(limits, ManagedFileLimits):
        raise ManagedFileAccessError('invalid_path')
    try:
        encoded = value.encode('utf-8')
    except UnicodeEncodeError:
        raise ManagedFileAccessError('invalid_path') from None
    if (
        not value or len(encoded) > limits.max_relative_path_bytes
        or value.startswith('/') or '\\' in value or '\x00' in value
        or _DRIVE_PATH.match(value)
    ):
        raise ManagedFileAccessError('invalid_path')
    components = value.split('/')
    if (
        len(components) > limits.max_path_depth
        or any(
            component in ('', '.', '..')
            or _DRIVE_COMPONENT.match(component)
            or component.startswith(_TEMPORARY_PREFIX)
            or len(component.encode('utf-8')) > limits.max_component_bytes
            for component in components
        )
    ):
        raise ManagedFileAccessError('invalid_path')
    return tuple(components)


def _close_descriptor(descriptor):
    if descriptor is None:
        return
    try:
        os.close(descriptor)
    except OSError:
        pass


def _descriptor_flags():
    required = (
        'O_PATH', 'O_DIRECTORY', 'O_NOFOLLOW', 'O_CLOEXEC',
        'O_NONBLOCK', 'O_NOCTTY', 'O_CREAT', 'O_EXCL',
    )
    if (
        not sys.platform.startswith('linux')
        or os.open not in getattr(os, 'supports_dir_fd', ())
        or any(type(getattr(os, name, None)) is not int for name in required)
    ):
        raise ManagedFileAccessError('filesystem_unavailable')
    return {
        'directory': (
            os.O_PATH | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC
        ),
        'list': os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC,
        'inspect': os.O_PATH | os.O_NOFOLLOW | os.O_CLOEXEC,
        'read': os.O_RDONLY | os.O_NONBLOCK | os.O_NOCTTY | os.O_CLOEXEC,
        'temporary': (
            os.O_RDWR | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW | os.O_CLOEXEC
        ),
    }


def _filesystem_error_category(exc):
    if exc.errno == errno.ENOENT:
        return 'not_found'
    if exc.errno in {
        errno.ELOOP, errno.ENOTDIR, errno.EISDIR, errno.ENXIO,
        errno.ENODEV, errno.EMLINK,
        getattr(errno, 'EOPNOTSUPP', -1),
        getattr(errno, 'ENOTSUP', -1),
    }:
        return 'unsafe_target'
    return 'filesystem_unavailable'


def _file_revision(details):
    return (
        details.st_dev,
        details.st_ino,
        stat.S_IFMT(details.st_mode),
        details.st_nlink,
        details.st_size,
        getattr(details, 'st_mtime_ns', None),
        getattr(details, 'st_ctime_ns', None),
    )


def _regular_single_link(details):
    return stat.S_ISREG(details.st_mode) and details.st_nlink == 1


def _valid_hash(value):
    if type(value) is not str or not _HASH.fullmatch(value):
        raise ManagedFileAccessError('invalid_hash')
    return value


def _rename_noreplace(source, target, directory_descriptor):
    try:
        function = ctypes.CDLL(None, use_errno=True).renameat2
        function.argtypes = (
            ctypes.c_int, ctypes.c_char_p, ctypes.c_int, ctypes.c_char_p,
            ctypes.c_uint,
        )
        function.restype = ctypes.c_int
        ctypes.set_errno(0)
        result = function(
            directory_descriptor, source.encode('utf-8'),
            directory_descriptor, target.encode('utf-8'),
            _RENAME_NOREPLACE,
        )
    except (AttributeError, OSError, TypeError, ValueError):
        raise ManagedFileAccessError('filesystem_unavailable') from None
    if result == 0:
        return
    error_number = ctypes.get_errno()
    if error_number == errno.EEXIST:
        raise ManagedFileAccessError('hash_conflict')
    raise ManagedFileAccessError('filesystem_unavailable')


class ManagedFileTraversal:
    def __init__(self, registry):
        if not isinstance(registry, ManagedFileRootRegistry):
            raise ValueError('managed file root registry is invalid')
        if any(not isinstance(root, ManagedFileRoot) for root in registry.roots):
            raise ValueError('managed file root registry is invalid')
        root_ids = tuple(root.root_id for root in registry.roots)
        try:
            roots_are_valid = all(
                _ROOT_ID.fullmatch(root.root_id)
                and _absolute_root_path(root.absolute_path) == root.absolute_path
                and isinstance(root.permissions, ManagedFilePermissions)
                and isinstance(root.limits, ManagedFileLimits)
                for root in registry.roots
            )
        except (AttributeError, ManagedFileConfigurationError, TypeError):
            roots_are_valid = False
        if not roots_are_valid or len(root_ids) != len(set(root_ids)):
            raise ValueError('managed file root registry is invalid')

        self._registry = registry
        self._lock = threading.Lock()
        self._mutation_lock = threading.Lock()
        self._snapshot_gate = threading.BoundedSemaphore(1)
        self._snapshots = set()
        self._closed = False
        self._roots = {}
        self._flags = None
        if not registry.roots:
            return

        self._flags = _descriptor_flags()
        try:
            for root in registry.roots:
                self._roots[root.root_id] = self._open_root(root.absolute_path)
        except BaseException:
            descriptors = tuple(self._roots.values())
            self._roots.clear()
            self._closed = True
            for descriptor in descriptors:
                _close_descriptor(descriptor)
            raise

    def _open_root(self, absolute_path):
        current = None
        try:
            current = os.open('/', self._flags['directory'])
            if not stat.S_ISDIR(os.fstat(current).st_mode):
                raise ManagedFileAccessError('root_unavailable')
            for component in absolute_path.split('/')[1:]:
                child = None
                try:
                    child = os.open(
                        component, self._flags['directory'], dir_fd=current,
                    )
                    if not stat.S_ISDIR(os.fstat(child).st_mode):
                        raise ManagedFileAccessError('root_unavailable')
                except BaseException:
                    _close_descriptor(child)
                    raise
                previous = current
                current = child
                child = None
                _close_descriptor(previous)
            descriptor = current
            current = None
            return descriptor
        except OSError:
            raise ManagedFileAccessError('root_unavailable') from None
        finally:
            _close_descriptor(current)

    def close(self):
        with self._lock:
            if self._closed:
                return
            self._closed = True
            descriptors = tuple(self._roots.values())
            snapshots = tuple(self._snapshots)
            self._roots.clear()
            self._snapshots.clear()
        for descriptor in descriptors:
            _close_descriptor(descriptor)
        for snapshot in snapshots:
            snapshot.close()

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        self.close()

    def _root_for_operation(self, root_id, operation):
        root = self._registry.get(root_id)
        if root is None:
            raise ManagedFileAccessError('unknown_root')
        if not root.permissions.allows(operation):
            raise ManagedFileAccessError('operation_not_allowed')
        return root

    def _duplicate_root(self, root_id):
        with self._lock:
            if self._closed:
                raise ManagedFileAccessError('closed')
            descriptor = self._roots.get(root_id)
            if descriptor is None:
                raise ManagedFileAccessError('root_unavailable')
            try:
                return os.dup(descriptor)
            except OSError:
                raise ManagedFileAccessError('root_unavailable') from None

    @staticmethod
    def _relative_target_allowed(root_id, components, target_kind):
        if root_id != RUNTIME_RESULT_ROOT_ID:
            return True
        if target_kind == 'directory':
            return not components
        return (
            len(components) == 1
            and _RESULT_PROJECTION_FILE.fullmatch(components[0]) is not None
        )

    def _require_relative_target(self, root_id, components, target_kind):
        if not self._relative_target_allowed(root_id, components, target_kind):
            raise ManagedFileAccessError('not_found')

    def _snapshot_closed(self, snapshot):
        with self._lock:
            self._snapshots.discard(snapshot)
        self._snapshot_gate.release()

    def _new_snapshot_file(self, root_id):
        root_descriptor = self._duplicate_root(root_id)
        try:
            return tempfile.TemporaryFile(
                mode='w+b', dir=f'/proc/self/fd/{root_descriptor}',
            )
        except OSError:
            raise ManagedFileAccessError('filesystem_unavailable') from None
        finally:
            _close_descriptor(root_descriptor)

    @staticmethod
    def _write_snapshot(handle, payload):
        view = memoryview(payload)
        written = 0
        while written < len(view):
            try:
                count = handle.write(view[written:])
            except InterruptedError:
                continue
            except OSError:
                raise ManagedFileAccessError('filesystem_unavailable') from None
            if not count:
                raise ManagedFileAccessError('filesystem_unavailable')
            written += count

    def _snapshot_descriptor(self, root, descriptor):
        handle = None
        try:
            before = os.fstat(descriptor)
            if not _regular_single_link(before):
                raise ManagedFileAccessError('unsafe_target')
            if before.st_size > root.limits.max_file_bytes:
                raise ManagedFileAccessError('limit_exceeded')
            handle = self._new_snapshot_file(root.root_id)
            os.lseek(descriptor, 0, os.SEEK_SET)
            digest = hashlib.sha256()
            byte_count = 0
            while True:
                remaining = root.limits.max_file_bytes - byte_count
                if remaining < 0:
                    raise ManagedFileAccessError('limit_exceeded')
                try:
                    chunk = os.read(
                        descriptor, min(_READ_CHUNK_BYTES, remaining + 1),
                    )
                except InterruptedError:
                    continue
                except OSError:
                    raise ManagedFileAccessError('filesystem_unavailable') from None
                if not chunk:
                    break
                byte_count += len(chunk)
                if byte_count > root.limits.max_file_bytes:
                    raise ManagedFileAccessError('limit_exceeded')
                digest.update(chunk)
                self._write_snapshot(handle, chunk)
            after = os.fstat(descriptor)
            if (
                not _regular_single_link(after)
                or _file_revision(before) != _file_revision(after)
                or byte_count != before.st_size
            ):
                raise ManagedFileAccessError('concurrent_change')
            handle.flush()
            handle.seek(0)
            with self._lock:
                if self._closed:
                    raise ManagedFileAccessError('closed')
                snapshot = ManagedFileSnapshot(handle, self._snapshot_closed)
                self._snapshots.add(snapshot)
                handle = None
            return ManagedFileIdentity(digest.hexdigest(), byte_count), snapshot
        except OSError:
            raise ManagedFileAccessError('filesystem_unavailable') from None
        finally:
            if handle is not None:
                try:
                    handle.close()
                except OSError:
                    pass

    def _open_relative(self, root, components, target_kind):
        current = self._duplicate_root(root.root_id)
        leaf = None
        try:
            for component in components[:-1]:
                child = None
                try:
                    child = os.open(
                        component, self._flags['directory'], dir_fd=current,
                    )
                    if not stat.S_ISDIR(os.fstat(child).st_mode):
                        raise ManagedFileAccessError('unsafe_target')
                except BaseException:
                    _close_descriptor(child)
                    raise
                previous = current
                current = child
                child = None
                _close_descriptor(previous)

            leaf_name = components[-1] if components else '.'
            if target_kind == 'directory':
                leaf = os.open(leaf_name, self._flags['list'], dir_fd=current)
                details = os.fstat(leaf)
                safe = stat.S_ISDIR(details.st_mode)
            else:
                leaf, details = self._open_file_at(current, leaf_name)
                safe = True
            if not safe:
                raise ManagedFileAccessError('unsafe_target')
            target = ManagedFileOpenedTarget(leaf, details)
            leaf = None
            return target
        except OSError as exc:
            raise ManagedFileAccessError(_filesystem_error_category(exc)) from None
        finally:
            _close_descriptor(leaf)
            _close_descriptor(current)

    def _open_file_at(self, directory_descriptor, leaf_name):
        inspected = readable = None
        try:
            inspected = os.open(
                leaf_name, self._flags['inspect'], dir_fd=directory_descriptor,
            )
            inspected_details = os.fstat(inspected)
            if not _regular_single_link(inspected_details):
                raise ManagedFileAccessError('unsafe_target')
            try:
                readable = os.open(
                    f'/proc/self/fd/{inspected}', self._flags['read'],
                )
                readable_details = os.fstat(readable)
            except OSError:
                raise ManagedFileAccessError('filesystem_unavailable') from None
            if (
                not _regular_single_link(readable_details)
                or readable_details.st_dev != inspected_details.st_dev
                or readable_details.st_ino != inspected_details.st_ino
            ):
                raise ManagedFileAccessError('unsafe_target')
            result = readable
            readable = None
            return result, readable_details
        finally:
            _close_descriptor(readable)
            _close_descriptor(inspected)

    @contextmanager
    def _opened_parent(self, root, components):
        current = operational = None
        try:
            current = self._duplicate_root(root.root_id)
            for component in components[:-1]:
                child = None
                try:
                    child = os.open(
                        component, self._flags['directory'], dir_fd=current,
                    )
                    if not stat.S_ISDIR(os.fstat(child).st_mode):
                        raise ManagedFileAccessError('unsafe_target')
                except BaseException:
                    _close_descriptor(child)
                    raise
                previous = current
                current = child
                child = None
                _close_descriptor(previous)
            anchored = os.fstat(current)
            operational = os.open('.', self._flags['list'], dir_fd=current)
            opened = os.fstat(operational)
            if (
                not stat.S_ISDIR(opened.st_mode)
                or opened.st_dev != anchored.st_dev
                or opened.st_ino != anchored.st_ino
            ):
                raise ManagedFileAccessError('unsafe_target')
            descriptor = operational
            operational = None
            try:
                yield descriptor, components[-1]
            finally:
                _close_descriptor(descriptor)
        except OSError as exc:
            raise ManagedFileAccessError(_filesystem_error_category(exc)) from None
        finally:
            _close_descriptor(operational)
            _close_descriptor(current)

    def _read_descriptor(self, descriptor, max_bytes, *, include_content):
        payload = bytearray() if include_content else None
        try:
            before = os.fstat(descriptor)
            if not _regular_single_link(before):
                raise ManagedFileAccessError('unsafe_target')
            if before.st_size > max_bytes:
                raise ManagedFileAccessError('limit_exceeded')
            try:
                os.lseek(descriptor, 0, os.SEEK_SET)
            except OSError:
                raise ManagedFileAccessError('filesystem_unavailable') from None
            digest = hashlib.sha256()
            byte_count = 0
            while True:
                remaining = max_bytes - byte_count
                if remaining < 0:
                    raise ManagedFileAccessError('limit_exceeded')
                try:
                    chunk = os.read(
                        descriptor, min(_READ_CHUNK_BYTES, remaining + 1),
                    )
                except InterruptedError:
                    continue
                except OSError:
                    raise ManagedFileAccessError('filesystem_unavailable') from None
                if not chunk:
                    break
                byte_count += len(chunk)
                if byte_count > max_bytes:
                    raise ManagedFileAccessError('limit_exceeded')
                digest.update(chunk)
                if payload is not None:
                    payload.extend(chunk)
            after = os.fstat(descriptor)
            if (
                not _regular_single_link(after)
                or _file_revision(before) != _file_revision(after)
                or byte_count != before.st_size
            ):
                raise ManagedFileAccessError('concurrent_change')
            identity = ManagedFileIdentity(digest.hexdigest(), byte_count)
            content = bytes(payload) if payload is not None else None
            if payload is not None:
                payload.clear()
            return identity, before, content
        except OSError:
            raise ManagedFileAccessError('filesystem_unavailable') from None
        except BaseException:
            if payload is not None:
                payload.clear()
            raise

    def _read_file_at(self, directory_descriptor, leaf_name, max_bytes, *, content):
        descriptor = None
        try:
            descriptor, details = self._open_file_at(
                directory_descriptor, leaf_name,
            )
            identity, stable, payload = self._read_descriptor(
                descriptor, max_bytes, include_content=content,
            )
            if _file_revision(details) != _file_revision(stable):
                raise ManagedFileAccessError('concurrent_change')
            return identity, stable, payload
        except OSError as exc:
            raise ManagedFileAccessError(_filesystem_error_category(exc)) from None
        finally:
            _close_descriptor(descriptor)

    def _revalidate_named_file(self, directory_descriptor, leaf_name, expected):
        descriptor = None
        try:
            descriptor = os.open(
                leaf_name, self._flags['inspect'], dir_fd=directory_descriptor,
            )
            current = os.fstat(descriptor)
            if (
                not _regular_single_link(current)
                or _file_revision(current) != _file_revision(expected)
            ):
                raise ManagedFileAccessError('concurrent_change')
            return current
        except ManagedFileAccessError:
            raise
        except OSError:
            raise ManagedFileAccessError('concurrent_change') from None
        finally:
            _close_descriptor(descriptor)

    def _new_temporary_file(self, directory_descriptor):
        for _ in range(_TEMPORARY_ATTEMPTS):
            name = f'{_TEMPORARY_PREFIX}{secrets.token_hex(12)}.tmp'
            try:
                descriptor = os.open(
                    name, self._flags['temporary'], 0o600,
                    dir_fd=directory_descriptor,
                )
                return name, descriptor
            except FileExistsError:
                continue
            except OSError:
                raise ManagedFileAccessError('filesystem_unavailable') from None
        raise ManagedFileAccessError('filesystem_unavailable')

    def _stage_temporary_file(self, directory_descriptor, payload, max_bytes):
        name = descriptor = view = None
        try:
            name, descriptor = self._new_temporary_file(directory_descriptor)
            os.fchmod(descriptor, 0o600)
            details = os.fstat(descriptor)
            if (
                not _regular_single_link(details)
                or details.st_uid != os.geteuid()
                or stat.S_IMODE(details.st_mode) != 0o600
            ):
                raise ManagedFileAccessError('unsafe_target')
            view = memoryview(payload)
            written = 0
            while written < len(view):
                try:
                    count = os.write(descriptor, view[written:])
                except InterruptedError:
                    continue
                except OSError:
                    raise ManagedFileAccessError('filesystem_unavailable') from None
                if count <= 0:
                    raise ManagedFileAccessError('filesystem_unavailable')
                written += count
            try:
                os.fsync(descriptor)
            except OSError:
                raise ManagedFileAccessError('filesystem_unavailable') from None
            identity, stable, _ = self._read_descriptor(
                descriptor, max_bytes, include_content=False,
            )
            if (
                identity.byte_count != len(payload)
                or stable.st_uid != os.geteuid()
                or stat.S_IMODE(stable.st_mode) != 0o600
            ):
                raise ManagedFileAccessError('unsafe_target')
            return name, descriptor, identity, stable
        except OSError:
            failure = ManagedFileAccessError('filesystem_unavailable')
            self._release_unpublished_temporary(
                directory_descriptor, name, descriptor, failure,
            )
            descriptor = None
            raise failure from None
        except BaseException as failure:
            self._release_unpublished_temporary(
                directory_descriptor, name, descriptor, failure,
            )
            descriptor = None
            raise
        finally:
            if view is not None:
                view.release()
            payload = None

    def _release_unpublished_temporary(
            self, directory_descriptor, name, descriptor, failure=None):
        cleanup_failure = close_failure = None
        if name is not None:
            try:
                self._cleanup_temporary_file(directory_descriptor, name)
            except BaseException as exc:
                cleanup_failure = exc
        try:
            _close_descriptor(descriptor)
        except BaseException as exc:
            close_failure = exc
        if failure is not None:
            if isinstance(failure, Exception):
                if cleanup_failure is not None:
                    raise cleanup_failure
                if close_failure is not None:
                    raise close_failure
            return
        if cleanup_failure is not None:
            raise cleanup_failure
        if close_failure is not None:
            raise close_failure

    @staticmethod
    def _cleanup_temporary_file(directory_descriptor, name):
        try:
            os.unlink(name, dir_fd=directory_descriptor)
        except FileNotFoundError:
            pass
        except OSError:
            raise ManagedFileAccessError('durability_uncertain') from None
        try:
            os.fsync(directory_descriptor)
        except OSError:
            raise ManagedFileAccessError('durability_uncertain') from None

    @staticmethod
    def _fsync_directory(directory_descriptor):
        try:
            os.fsync(directory_descriptor)
        except OSError:
            raise ManagedFileAccessError('durability_uncertain') from None

    @staticmethod
    def _lock_mutation_directory(directory_descriptor):
        if fcntl is None:
            raise ManagedFileAccessError('filesystem_unavailable')
        while True:
            try:
                fcntl.flock(directory_descriptor, fcntl.LOCK_EX)
                return
            except InterruptedError:
                continue
            except OSError:
                raise ManagedFileAccessError(
                    'filesystem_unavailable',
                ) from None

    @staticmethod
    def _unlock_mutation_directory(directory_descriptor):
        if fcntl is None:
            return
        try:
            fcntl.flock(directory_descriptor, fcntl.LOCK_UN)
        except OSError:
            pass

    @contextmanager
    def _locked_parent(self, root, components):
        with self._opened_parent(root, components) as parent:
            directory_descriptor, _ = parent
            self._lock_mutation_directory(directory_descriptor)
            try:
                yield parent
            finally:
                self._unlock_mutation_directory(directory_descriptor)

    def _publish_temporary(
            self, directory_descriptor, temporary_name, leaf_name, *, create):
        failure = None
        try:
            if create:
                _rename_noreplace(
                    temporary_name, leaf_name, directory_descriptor,
                )
            else:
                try:
                    os.replace(
                        temporary_name, leaf_name,
                        src_dir_fd=directory_descriptor,
                        dst_dir_fd=directory_descriptor,
                    )
                except OSError:
                    raise ManagedFileAccessError(
                        'concurrent_change',
                    ) from None
        except BaseException as exc:
            failure = exc
            raise
        finally:
            try:
                self._fsync_directory(directory_descriptor)
            except BaseException as sync_failure:
                if (
                    failure is None
                    or isinstance(failure, Exception)
                    and not isinstance(sync_failure, Exception)
                ):
                    raise

    def _verify_private_file(
            self, directory_descriptor, leaf_name, max_bytes,
            expected_identity, expected_inode):
        identity, details, _ = self._read_file_at(
            directory_descriptor, leaf_name, max_bytes, content=False,
        )
        if (
            identity != expected_identity
            or details.st_dev != expected_inode.st_dev
            or details.st_ino != expected_inode.st_ino
            or details.st_uid != os.geteuid()
            or stat.S_IMODE(details.st_mode) != 0o600
        ):
            raise ManagedFileAccessError('concurrent_change')
        return details

    @contextmanager
    def _opened_target(self, root, components, target_kind):
        target = self._open_relative(root, components, target_kind)
        try:
            yield target
        finally:
            target._close()

    def open_list_directory(self, root_id, relative_path=None):
        root = self._root_for_operation(root_id, ManagedFileOperation.LIST)
        components = (
            () if relative_path is None
            else parse_managed_relative_path(relative_path, root.limits)
        )
        self._require_relative_target(root.root_id, components, 'directory')
        return self._opened_target(root, components, 'directory')

    def open_read_file(self, root_id, relative_path):
        root = self._root_for_operation(root_id, ManagedFileOperation.READ)
        components = parse_managed_relative_path(relative_path, root.limits)
        self._require_relative_target(root.root_id, components, 'file')
        return self._opened_target(root, components, 'file')

    def list_directory(self, root_id, relative_path=None):
        root = self._root_for_operation(root_id, ManagedFileOperation.LIST)
        components = (
            () if relative_path is None
            else parse_managed_relative_path(relative_path, root.limits)
        )
        self._require_relative_target(root.root_id, components, 'directory')
        entries = []
        name_bytes = scanned = 0
        try:
            with self._opened_target(root, components, 'directory') as opened:
                try:
                    iterator = os.scandir(opened.fileno())
                    with iterator:
                        for item in iterator:
                            scanned += 1
                            if scanned > root.limits.max_listing_entries:
                                raise ManagedFileAccessError('limit_exceeded')
                            name = item.name
                            if type(name) is not str or name.startswith(
                                    _TEMPORARY_PREFIX):
                                continue
                            if not self._relative_target_allowed(
                                    root.root_id, (*components, name), 'file'):
                                continue
                            try:
                                parse_managed_relative_path(
                                    '/'.join((*components, name)), root.limits,
                                )
                                encoded = name.encode('utf-8')
                                details = os.stat(
                                    name, dir_fd=opened.fileno(),
                                    follow_symlinks=False,
                                )
                            except ManagedFileAccessError:
                                continue
                            except FileNotFoundError:
                                continue
                            except OSError:
                                raise ManagedFileAccessError(
                                    'filesystem_unavailable',
                                ) from None
                            if stat.S_ISDIR(details.st_mode):
                                kind = 'directory'
                                byte_count = None
                            elif (
                                _regular_single_link(details)
                                and details.st_size <= root.limits.max_file_bytes
                            ):
                                kind = 'file'
                                byte_count = details.st_size
                            else:
                                continue
                            if not self._relative_target_allowed(
                                    root.root_id, (*components, name), kind):
                                continue
                            name_bytes += len(encoded)
                            if name_bytes > root.limits.max_listing_bytes:
                                raise ManagedFileAccessError('limit_exceeded')
                            entries.append(ManagedFileDirectoryEntry(
                                name, kind, byte_count,
                            ))
                except OSError:
                    raise ManagedFileAccessError(
                        'filesystem_unavailable',
                    ) from None
            entries.sort(key=lambda entry: entry.name.encode('utf-8'))
            return ManagedFileListing(tuple(entries), name_bytes)
        except BaseException:
            entries.clear()
            raise

    def download_file(self, root_id, relative_path):
        root = self._root_for_operation(root_id, ManagedFileOperation.READ)
        components = parse_managed_relative_path(relative_path, root.limits)
        self._require_relative_target(root.root_id, components, 'file')
        if root.root_id == RUNTIME_RESULT_ROOT_ID:
            if not self._snapshot_gate.acquire(blocking=False):
                raise ManagedFileAccessError('download_busy')
            try:
                for attempt in range(2):
                    try:
                        with self._opened_target(root, components, 'file') as opened:
                            identity, snapshot = self._snapshot_descriptor(
                                root, opened.fileno(),
                            )
                        return ManagedFileDownload(
                            identity, snapshot=snapshot,
                        )
                    except ManagedFileAccessError as exc:
                        if attempt or exc.category not in (
                                'concurrent_change', 'unsafe_target'):
                            raise
            except BaseException:
                self._snapshot_gate.release()
                raise
            raise ManagedFileAccessError('concurrent_change')
        for attempt in range(2):
            try:
                with self._opened_target(root, components, 'file') as opened:
                    identity, _, content = self._read_descriptor(
                        opened.fileno(), root.limits.max_file_bytes,
                        include_content=True,
                    )
                return ManagedFileDownload(identity, content)
            except ManagedFileAccessError as exc:
                if attempt or exc.category not in ('concurrent_change', 'unsafe_target'):
                    raise
        raise ManagedFileAccessError('concurrent_change')

    def mutation_file_identity(
            self, root_id, relative_path, operation, *, require_private_sha256=None):
        if operation not in (
            ManagedFileOperation.CREATE_REPLACE, ManagedFileOperation.DELETE,
        ):
            raise ManagedFileAccessError('operation_not_allowed')
        root = self._root_for_operation(root_id, operation)
        components = parse_managed_relative_path(relative_path, root.limits)
        with self._opened_parent(root, components) as parent:
            directory_descriptor, leaf_name = parent
            try:
                try:
                    identity, details, _ = self._read_file_at(
                        directory_descriptor, leaf_name,
                        root.limits.max_file_bytes, content=False,
                    )
                except ManagedFileAccessError as exc:
                    if exc.category != 'not_found':
                        raise
                    self._fsync_directory(directory_descriptor)
                    try:
                        self._read_file_at(
                            directory_descriptor, leaf_name,
                            root.limits.max_file_bytes, content=False,
                        )
                    except ManagedFileAccessError as repeated:
                        if repeated.category == 'not_found':
                            raise exc
                        raise
                    raise ManagedFileAccessError('concurrent_change')
                # Replay may follow a crash after namespace publication. A
                # successful directory fsync turns the observed state into
                # durable evidence before terminal operation reconciliation.
                self._fsync_directory(directory_descriptor)
                self._revalidate_named_file(
                    directory_descriptor, leaf_name, details,
                )
                if (
                    require_private_sha256 is not None
                    and hmac.compare_digest(identity.sha256, require_private_sha256)
                    and (
                        details.st_uid != os.geteuid()
                        or stat.S_IMODE(details.st_mode) != 0o600
                    )
                ):
                    raise ManagedFileAccessError('unsafe_target')
            except ManagedFileAccessError:
                raise
        return identity

    def _create_replace_file_payload(
            self, root_id, relative_path, payload_box, *, expected_sha256):
        root = self._root_for_operation(
            root_id, ManagedFileOperation.CREATE_REPLACE,
        )
        components = parse_managed_relative_path(relative_path, root.limits)
        if len(payload_box) != 1 or type(payload_box[0]) is not bytes:
            raise ManagedFileAccessError('invalid_content')
        if len(payload_box[0]) > root.limits.max_file_bytes:
            raise ManagedFileAccessError('limit_exceeded')
        if expected_sha256 is not None:
            expected_sha256 = _valid_hash(expected_sha256)
        proposed = ManagedFileIdentity(
            hashlib.sha256(payload_box[0]).hexdigest(), len(payload_box[0]),
        )

        with self._mutation_lock, self._locked_parent(root, components) as parent:
            directory_descriptor, leaf_name = parent
            before = before_details = None
            if expected_sha256 is not None:
                before, before_details, _ = self._read_file_at(
                    directory_descriptor, leaf_name,
                    root.limits.max_file_bytes, content=False,
                )
                if not hmac.compare_digest(before.sha256, expected_sha256):
                    raise ManagedFileAccessError('hash_conflict')
                if before == proposed:
                    self._revalidate_named_file(
                        directory_descriptor, leaf_name, before_details,
                    )
                    return ManagedFileMutation(before, before, False)

            temporary_name = temporary_descriptor = None
            published = False
            failure = None
            try:
                (
                    temporary_name, temporary_descriptor,
                    staged_identity, staged_details,
                ) = self._stage_temporary_file(
                    directory_descriptor, payload_box[0],
                    root.limits.max_file_bytes,
                )
                if staged_identity != proposed:
                    raise ManagedFileAccessError('concurrent_change')
                if expected_sha256 is not None:
                    self._revalidate_named_file(
                        directory_descriptor, leaf_name, before_details,
                    )
                self._publish_temporary(
                    directory_descriptor, temporary_name, leaf_name,
                    create=expected_sha256 is None,
                )
                published = True
                try:
                    verified = self._verify_private_file(
                        directory_descriptor, leaf_name,
                        root.limits.max_file_bytes, proposed, staged_details,
                    )
                    self._revalidate_named_file(
                        directory_descriptor, leaf_name, verified,
                    )
                except ManagedFileAccessError as exc:
                    if exc.category == 'concurrent_change':
                        raise
                    raise ManagedFileAccessError(
                        'durability_uncertain',
                    ) from None
                return ManagedFileMutation(before, proposed, True)
            except BaseException as exc:
                failure = exc
                raise
            finally:
                if temporary_name is not None and not published:
                    self._release_unpublished_temporary(
                        directory_descriptor, temporary_name,
                        temporary_descriptor, failure,
                    )
                    temporary_descriptor = None
                try:
                    _close_descriptor(temporary_descriptor)
                except BaseException:
                    if failure is None:
                        raise

    def create_replace_file(
            self, root_id, relative_path, payload, *, expected_sha256):
        payload_box = [payload]
        payload = None
        try:
            return self._create_replace_file_payload(
                root_id, relative_path, payload_box,
                expected_sha256=expected_sha256,
            )
        finally:
            payload_box.clear()
            payload = payload_box = None

    def delete_file(self, root_id, relative_path, *, expected_sha256):
        root = self._root_for_operation(root_id, ManagedFileOperation.DELETE)
        components = parse_managed_relative_path(relative_path, root.limits)
        expected_sha256 = _valid_hash(expected_sha256)
        with self._mutation_lock, self._locked_parent(root, components) as parent:
            directory_descriptor, leaf_name = parent
            before, before_details, _ = self._read_file_at(
                directory_descriptor, leaf_name,
                root.limits.max_file_bytes, content=False,
            )
            if not hmac.compare_digest(before.sha256, expected_sha256):
                raise ManagedFileAccessError('hash_conflict')
            self._revalidate_named_file(
                directory_descriptor, leaf_name, before_details,
            )
            failure = None
            try:
                try:
                    os.unlink(leaf_name, dir_fd=directory_descriptor)
                except OSError:
                    raise ManagedFileAccessError(
                        'concurrent_change',
                    ) from None
            except BaseException as exc:
                failure = exc
                raise
            finally:
                try:
                    self._fsync_directory(directory_descriptor)
                except BaseException as sync_failure:
                    if (
                        failure is None
                        or isinstance(failure, Exception)
                        and not isinstance(sync_failure, Exception)
                    ):
                        raise
            return ManagedFileMutation(before, None, True)