"""Private, fail-closed routing for Activity Monitor's Dispatch surface.

Routing inspects local task/session metadata, current Agent Board ownership, and
file-name metadata from a connected Brain. It never opens Brain note bodies,
credential files, or stores dispatched message bodies.
"""

from __future__ import annotations

from datetime import datetime, timezone
import hashlib
import json
import os
from pathlib import Path
import queue
import re
import secrets
import shutil
import socket
import stat
import struct
import subprocess
import threading
import time
import uuid


SCHEMA_VERSION = "ke.activity-monitor-dispatch.v1"
HISTORY_SCHEMA_VERSION = "ke.activity-monitor-dispatch-history.v1"
MAX_MESSAGE_BYTES = 900_000
MAX_HISTORY_ENTRIES = 100
MAX_IPC_FRAME_BYTES = 4 * 1024 * 1024
PERMISSION_AUDIT_CACHE_SECONDS = 60.0
DISPATCH_RECONCILIATION_WINDOW_SECONDS = 24 * 60 * 60
UUID_RE = re.compile(
    r"(?<![0-9a-f])[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}(?![0-9a-f])",
    re.IGNORECASE,
)
STOP_WORDS = {
    "about", "after", "also", "and", "are", "been", "before", "being", "build",
    "can", "could", "for", "from", "have", "here", "into", "just", "make", "message",
    "please", "should", "that", "the", "their", "then", "this", "through", "user", "want",
    "what", "when", "where", "which", "will", "with", "would", "you", "your",
}
ALLOWED_STATES = {
    "resolving",
    "ambiguous",
    "no target",
    "queued",
    "transcript observed",
    "acknowledged",
    "working",
    "completed",
    "failed",
    "owner preflight",
    "exact-task resume/steer",
    "exact-session delivery",
    "accepted",
    "failed before send",
    "uncertain after send",
}
OWNER_DISCOVERY_FALLBACK_CODES = {
    "desktop_ipc_unavailable",
    "desktop_ipc_timeout",
    "desktop_ipc_read_failed",
    "desktop_ipc_closed",
}
WILLIAM_PREFACE = (
    "Forwarded verbatim from William via Dispatch. Treat this as William's current\n"
    "instruction for this task, without adding authority beyond his words:\n\n"
)
DIAGNOSTIC_PREFACE = (
    "Diagnostic generated by the Dispatch implementation task. William requested\n"
    "the router, but did not author this canary. Do not treat it as new authority:\n\n"
)


def _utc_now() -> str:
    return datetime.now(tz=timezone.utc).isoformat().replace("+00:00", "Z")


def _sha256(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


def _tokens(text: str) -> set[str]:
    return {
        value
        for value in re.findall(r"[a-z0-9][a-z0-9._+-]{2,}", str(text or "").lower())
        if value not in STOP_WORDS and not UUID_RE.fullmatch(value)
    }


def _safe_error(value: object, limit: int = 280) -> str:
    text = re.sub(r"\s+", " ", str(value or "")).strip()
    return text[:limit] or "Dispatch operation failed"


def attributed_message(original: str, diagnostic: bool = False) -> str:
    """Preserve the original text exactly after a fixed attribution."""
    return (DIAGNOSTIC_PREFACE if diagnostic else WILLIAM_PREFACE) + original


class DispatchError(RuntimeError):
    def __init__(
        self,
        code: str,
        message: str,
        *,
        phase: str | None = None,
        delivery_attempted: bool = False,
        receipt: dict | None = None,
    ):
        super().__init__(message)
        self.code = code
        self.phase = phase
        self.delivery_attempted = bool(delivery_attempted)
        self.receipt = dict(receipt or {})


class DesktopIpcClient:
    """Authenticated client for the Codex desktop owner's local IPC router."""

    def __init__(self, socket_path: str | Path, request_timeout: float = 6.0):
        self.socket_path = Path(socket_path).expanduser()
        self.request_timeout = max(0.5, float(request_timeout))
        self.socket: socket.socket | None = None
        self.client_id: str | None = None

    def _validate_socket(self) -> None:
        try:
            metadata = self.socket_path.lstat()
        except FileNotFoundError as error:
            raise DispatchError("desktop_ipc_unavailable", "The Codex desktop IPC socket is unavailable") from error
        except OSError as error:
            raise DispatchError("desktop_ipc_unavailable", _safe_error(error)) from error
        if not stat.S_ISSOCK(metadata.st_mode):
            raise DispatchError("desktop_ipc_insecure", "The Codex desktop IPC path is not a socket")
        if metadata.st_uid != os.getuid() or metadata.st_mode & 0o077:
            raise DispatchError(
                "desktop_ipc_insecure",
                "The Codex desktop IPC socket is not private to the current user",
            )

    def connect(self) -> None:
        if self.socket is not None:
            return
        self._validate_socket()
        client = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        client.settimeout(self.request_timeout)
        try:
            client.connect(str(self.socket_path))
        except OSError as error:
            client.close()
            raise DispatchError("desktop_ipc_unavailable", _safe_error(error)) from error
        self.socket = client
        response = self._request(
            "initialize",
            {"clientType": "activity-monitor-dispatch"},
            version=0,
            include_source=False,
        )
        if response.get("resultType") != "success":
            self.close()
            raise DispatchError("desktop_ipc_initialize_failed", self.error_text(response))
        result = response.get("result") or {}
        client_id = str(result.get("clientId") or "") if isinstance(result, dict) else ""
        if not UUID_RE.fullmatch(client_id):
            self.close()
            raise DispatchError("desktop_ipc_protocol_error", "Codex desktop returned no exact client identity")
        self.client_id = client_id

    def _read_exact(self, size: int) -> bytes:
        client = self.socket
        if client is None:
            raise DispatchError("desktop_ipc_closed", "Codex desktop IPC is not connected")
        chunks: list[bytes] = []
        remaining = size
        while remaining:
            try:
                chunk = client.recv(remaining)
            except (OSError, socket.timeout) as error:
                raise DispatchError("desktop_ipc_read_failed", _safe_error(error)) from error
            if not chunk:
                raise DispatchError("desktop_ipc_closed", "Codex desktop IPC closed before responding")
            chunks.append(chunk)
            remaining -= len(chunk)
        return b"".join(chunks)

    def _read_message(self) -> dict:
        frame_size = struct.unpack("<I", self._read_exact(4))[0]
        if frame_size <= 0 or frame_size > MAX_IPC_FRAME_BYTES:
            raise DispatchError("desktop_ipc_protocol_error", "Codex desktop returned an invalid IPC frame")
        try:
            value = json.loads(self._read_exact(frame_size).decode("utf-8"))
        except (UnicodeError, ValueError, TypeError) as error:
            raise DispatchError("desktop_ipc_protocol_error", "Codex desktop returned invalid JSON") from error
        if not isinstance(value, dict):
            raise DispatchError("desktop_ipc_protocol_error", "Codex desktop returned a non-object response")
        return value

    def _write_message(self, value: dict) -> None:
        client = self.socket
        if client is None:
            raise DispatchError("desktop_ipc_closed", "Codex desktop IPC is not connected")
        encoded = json.dumps(value, separators=(",", ":")).encode("utf-8")
        if len(encoded) > MAX_IPC_FRAME_BYTES:
            raise DispatchError("desktop_ipc_frame_too_large", "The Codex desktop IPC request is too large")
        try:
            client.sendall(struct.pack("<I", len(encoded)) + encoded)
        except OSError as error:
            raise DispatchError("desktop_ipc_write_failed", _safe_error(error)) from error

    def _request(
        self,
        method: str,
        params: dict,
        *,
        version: int,
        target_client_id: str | None = None,
        timeout_ms: int = 6000,
        include_source: bool = True,
    ) -> dict:
        request_id = str(uuid.uuid4())
        request = {
            "type": "request",
            "requestId": request_id,
            "version": int(version),
            "method": method,
            "params": params,
            "timeoutMs": int(timeout_ms),
        }
        if include_source:
            if not self.client_id:
                raise DispatchError("desktop_ipc_protocol_error", "Codex desktop client is not initialized")
            request["sourceClientId"] = self.client_id
        if target_client_id:
            request["targetClientId"] = target_client_id
        if self.socket is not None:
            self.socket.settimeout(max(self.request_timeout, timeout_ms / 1000 + 1.0))
        self._write_message(request)
        deadline = time.monotonic() + max(0.5, timeout_ms / 1000 + 1.0)
        while time.monotonic() < deadline:
            response = self._read_message()
            if response.get("type") == "response" and response.get("requestId") == request_id:
                return response
        raise DispatchError("desktop_ipc_timeout", f"Timed out during {method}")

    def request(
        self,
        method: str,
        params: dict,
        *,
        version: int = 1,
        target_client_id: str | None = None,
        timeout_ms: int = 6000,
    ) -> dict:
        self.connect()
        return self._request(
            method,
            params,
            version=version,
            target_client_id=target_client_id,
            timeout_ms=timeout_ms,
        )

    @staticmethod
    def error_text(response: dict) -> str:
        for key in ("error", "message", "result"):
            value = response.get(key)
            if value:
                return _safe_error(json.dumps(value, sort_keys=True) if isinstance(value, (dict, list)) else value)
        return "Codex desktop IPC request failed"

    def close(self) -> None:
        client, self.socket = self.socket, None
        self.client_id = None
        if client is not None:
            try:
                client.close()
            except OSError:
                pass


class CodexAppServerClient:
    """Small JSONL client for the supported experimental Codex app-server API."""

    def __init__(self, command: str, request_timeout: float = 6.0):
        self.command = command
        self.request_timeout = max(0.5, float(request_timeout))
        self.process: subprocess.Popen | None = None
        self._waiters: dict[int, queue.Queue] = {}
        self._waiters_lock = threading.Lock()
        self._write_lock = threading.Lock()
        self._notifications: queue.Queue = queue.Queue()
        self._next_id = 1
        self._stderr_tail: list[str] = []

    def start(self) -> None:
        if self.process and self.process.poll() is None:
            return
        try:
            self.process = subprocess.Popen(
                [self.command, "app-server", "--stdio"],
                stdin=subprocess.PIPE,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True,
                bufsize=1,
            )
        except OSError as error:
            raise DispatchError("codex_app_server_unavailable", _safe_error(error)) from error
        threading.Thread(target=self._read_stdout, daemon=True).start()
        threading.Thread(target=self._read_stderr, daemon=True).start()
        self.request(
            "initialize",
            {
                "clientInfo": {
                    "name": "ke_activity_monitor_dispatch",
                    "title": "Activity Monitor Dispatch",
                    "version": "2.7.0",
                },
                "capabilities": {"experimentalApi": True},
            },
        )
        self.notify("initialized", {})

    def _read_stdout(self) -> None:
        process = self.process
        if not process or not process.stdout:
            return
        for line in process.stdout:
            try:
                message = json.loads(line)
            except (ValueError, TypeError):
                continue
            response_id = message.get("id")
            if response_id is not None:
                with self._waiters_lock:
                    waiter = self._waiters.get(response_id)
                if waiter:
                    waiter.put(message)
                    continue
            self._notifications.put(message)
        self._notifications.put({"method": "_process/closed", "params": {}})

    def _read_stderr(self) -> None:
        process = self.process
        if not process or not process.stderr:
            return
        for line in process.stderr:
            self._stderr_tail.append(_safe_error(line, 400))
            del self._stderr_tail[:-8]

    def request(self, method: str, params: dict | None = None, timeout: float | None = None) -> dict:
        process = self.process
        if not process or process.poll() is not None or not process.stdin:
            raise DispatchError("codex_app_server_closed", "Codex app-server is not running")
        with self._write_lock:
            request_id = self._next_id
            self._next_id += 1
            waiter: queue.Queue = queue.Queue(maxsize=1)
            with self._waiters_lock:
                self._waiters[request_id] = waiter
            try:
                process.stdin.write(
                    json.dumps(
                        {"jsonrpc": "2.0", "id": request_id, "method": method, "params": params or {}},
                        separators=(",", ":"),
                    )
                    + "\n"
                )
                process.stdin.flush()
            except OSError as error:
                with self._waiters_lock:
                    self._waiters.pop(request_id, None)
                raise DispatchError("codex_app_server_write_failed", _safe_error(error)) from error
        try:
            response = waiter.get(timeout=timeout or self.request_timeout)
        except queue.Empty as error:
            detail = self._stderr_tail[-1] if self._stderr_tail else f"Timed out during {method}"
            raise DispatchError("codex_app_server_timeout", detail) from error
        finally:
            with self._waiters_lock:
                self._waiters.pop(request_id, None)
        if response.get("error"):
            error = response["error"]
            message = _safe_error(error.get("message") if isinstance(error, dict) else error)
            lowered = message.lower()
            if "active writer" in lowered or "already running" in lowered or "in use" in lowered:
                raise DispatchError(
                    "busy_exact_task_unsupported",
                    "That exact Codex task is held by another app writer; Dispatch refused to create a companion.",
                )
            raise DispatchError("codex_request_failed", message)
        result = response.get("result")
        if not isinstance(result, dict):
            raise DispatchError("codex_protocol_error", f"{method} returned no object result")
        return result

    def notify(self, method: str, params: dict | None = None) -> None:
        process = self.process
        if not process or process.poll() is not None or not process.stdin:
            raise DispatchError("codex_app_server_closed", "Codex app-server is not running")
        with self._write_lock:
            process.stdin.write(
                json.dumps(
                    {"jsonrpc": "2.0", "method": method, "params": params or {}},
                    separators=(",", ":"),
                )
                + "\n"
            )
            process.stdin.flush()

    def next_notification(self, timeout: float = 1.0) -> dict | None:
        try:
            return self._notifications.get(timeout=max(0.05, timeout))
        except queue.Empty:
            return None

    def close(self) -> None:
        process = self.process
        if not process:
            return
        if process.poll() is None:
            try:
                process.terminate()
                process.wait(timeout=1.5)
            except (OSError, subprocess.TimeoutExpired):
                try:
                    process.kill()
                except OSError:
                    pass
        self.process = None


class DispatchService:
    """Resolve exactly one existing owner and deliver without provider substitution."""

    def __init__(
        self,
        *,
        home: str | Path | None = None,
        history_path: str | Path | None = None,
        brain_service=None,
        workspace_service=None,
        codex_command: str | None = None,
        board_command: list[str] | None = None,
        claude_sender: str | Path | None = None,
        desktop_ipc_path: str | Path | None = None,
        desktop_ipc_factory=DesktopIpcClient,
        source_task_id: str = "01a020e9-efef-70c0-8aee-5370b9867bb0",
        monitor_seconds: float = 1800.0,
    ):
        self.home = Path(home or Path.home()).expanduser().resolve()
        support = self.home / "Library" / "Application Support" / "KE Studios" / "Activity Monitor"
        self.history_path = Path(history_path or support / "dispatch-history.json").expanduser()
        self.brain_service = brain_service
        self.workspace_service = workspace_service
        self.codex_command = self._validated_executable(codex_command or self._find_codex_command())
        self.board_command = board_command or [
            "/usr/bin/python3",
            str(self.home / "ke-agent-rooms" / "board" / "boardctl.py"),
            "status",
            "--event-limit",
            "0",
        ]
        self.claude_sender = Path(
            claude_sender
            or self.home / ".codex" / "skills" / "dispatch-chat" / "scripts" / "send_claude_session.mjs"
        )
        self.desktop_ipc_path = Path(desktop_ipc_path or self.home / ".codex" / "ipc" / "ipc.sock")
        self.desktop_ipc_factory = desktop_ipc_factory
        self.source_task_id = source_task_id
        self.monitor_seconds = max(5.0, float(monitor_seconds))
        self._lock = threading.RLock()
        self._resolutions: dict[str, dict] = {}
        self._clients: dict[str, CodexAppServerClient] = {}
        # None means no permission audit has completed. A numeric sentinel such
        # as 0.0 is unsafe because time.monotonic() can also be near zero in a
        # newly started interpreter, which would incorrectly cache a failure.
        self._permission_audit: tuple[float | None, bool] = (None, False)

    def _find_codex_command(self) -> str | None:
        found = shutil.which("codex")
        if found:
            return found
        candidates = [
            self.home / ".local" / "bin" / "codex",
            self.home / ".codex" / "packages" / "standalone" / "current" / "bin" / "codex",
        ]
        return next((str(path) for path in candidates if path.is_file()), None)

    @staticmethod
    def _validated_executable(command: str | None) -> str | None:
        """Resolve only a fixed root/current-user executable for app-server launch."""
        if not command:
            return None
        try:
            path = Path(command).expanduser().resolve(strict=True)
            metadata = path.stat()
        except OSError:
            return None
        if (
            not stat.S_ISREG(metadata.st_mode)
            or metadata.st_uid not in {0, os.getuid()}
            or metadata.st_mode & 0o022
            or not os.access(path, os.X_OK)
        ):
            return None
        return str(path)

    @staticmethod
    def _open_private_directory(path: str | Path, *, dir_fd: int | None = None) -> int:
        flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0)
        descriptor = os.open(path, flags, dir_fd=dir_fd)
        metadata = os.fstat(descriptor)
        if (
            not stat.S_ISDIR(metadata.st_mode)
            or metadata.st_uid != os.getuid()
            or stat.S_IMODE(metadata.st_mode) & 0o077
        ):
            os.close(descriptor)
            raise OSError("directory is not private current-user metadata")
        return descriptor

    @staticmethod
    def _open_history_directory(path: str | Path) -> int:
        """Open an existing support directory without hiding safe 0755 upgrades."""
        flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0)
        descriptor = os.open(path, flags)
        metadata = os.fstat(descriptor)
        if (
            not stat.S_ISDIR(metadata.st_mode)
            or metadata.st_uid != os.getuid()
            or stat.S_IMODE(metadata.st_mode) & 0o022
        ):
            os.close(descriptor)
            raise OSError("history directory is not a safe current-user support directory")
        return descriptor

    def _read_history_checked(self) -> tuple[dict, str | None]:
        """Return healthy history or a fail-closed status for an existing bad store.

        A genuinely missing file is the normal first-launch state. Once a
        history path exists, any trust, permission, size, schema, or decoding
        failure may conceal a prior attempted receipt and must disable sends.
        """
        defaults = {"schemaVersion": HISTORY_SCHEMA_VERSION, "entries": []}
        descriptors: list[int] = []
        try:
            try:
                parent_fd = self._open_history_directory(self.history_path.parent)
            except FileNotFoundError:
                return defaults, None
            except OSError:
                return defaults, "dispatch_history_unavailable"
            descriptors.append(parent_fd)
            flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
            try:
                history_fd = os.open(self.history_path.name, flags, dir_fd=parent_fd)
            except FileNotFoundError:
                return defaults, None
            except OSError:
                return defaults, "dispatch_history_unavailable"
            descriptors.append(history_fd)
            metadata = os.fstat(history_fd)
            if (
                not stat.S_ISREG(metadata.st_mode)
                or metadata.st_uid != os.getuid()
                or stat.S_IMODE(metadata.st_mode) != 0o600
                or metadata.st_size < 2
                or metadata.st_size > 2 * 1024 * 1024
            ):
                return defaults, "dispatch_history_unavailable"
            chunks: list[bytes] = []
            remaining = metadata.st_size
            while remaining > 0:
                chunk = os.read(history_fd, min(16_384, remaining))
                if not chunk:
                    break
                chunks.append(chunk)
                remaining -= len(chunk)
            value = json.loads(b"".join(chunks).decode("utf-8"))
            entries = value.get("entries") if isinstance(value, dict) else None
            if (
                value.get("schemaVersion") != HISTORY_SCHEMA_VERSION
                or not isinstance(entries, list)
                or len(entries) > MAX_HISTORY_ENTRIES
                or any(not isinstance(entry, dict) for entry in entries)
            ):
                return defaults, "dispatch_history_unavailable"
            return value, None
        except (OSError, ValueError, TypeError, AttributeError, UnicodeError):
            return defaults, "dispatch_history_unavailable"
        finally:
            for descriptor in reversed(descriptors):
                try:
                    os.close(descriptor)
                except OSError:
                    pass

    def _read_history(self) -> dict:
        return self._read_history_checked()[0]

    @staticmethod
    def _history_error() -> DispatchError:
        return DispatchError(
            "dispatch_history_unavailable",
            "Existing Dispatch receipt history is unavailable or untrusted; no new delivery started, and sending is disabled until the local metadata-only history is repaired or reconciled",
            phase="history reconciliation",
            delivery_attempted=False,
            receipt={
                "phase": "history reconciliation",
                "deliveryAttempted": False,
                "retrySafe": False,
                "reconciliationRequired": True,
            },
        )

    def _write_history(self, value: dict) -> None:
        self.history_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
        _current, issue = self._read_history_checked()
        if issue:
            raise OSError("existing dispatch history is unavailable or untrusted")
        parent_fd = self._open_history_directory(self.history_path.parent)
        temporary_name = self.history_path.name + ".tmp-" + secrets.token_hex(4)
        descriptor = -1
        try:
            flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0)
            descriptor = os.open(temporary_name, flags, 0o600, dir_fd=parent_fd)
            with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
                descriptor = -1
                json.dump(value, handle, indent=2, sort_keys=True)
                handle.write("\n")
                handle.flush()
                os.fsync(handle.fileno())
            os.replace(
                temporary_name,
                self.history_path.name,
                src_dir_fd=parent_fd,
                dst_dir_fd=parent_fd,
            )
            os.chmod(self.history_path.name, 0o600, dir_fd=parent_fd, follow_symlinks=False)
            os.fsync(parent_fd)
        finally:
            if descriptor >= 0:
                os.close(descriptor)
            try:
                os.unlink(temporary_name, dir_fd=parent_fd)
            except FileNotFoundError:
                pass
            os.close(parent_fd)

    def _history_entry(
        self,
        target: dict | None,
        sha256: str,
        reason: str,
        state: str,
        diagnostic: bool,
        receipt: dict | None = None,
    ) -> dict:
        if state not in ALLOWED_STATES:
            raise ValueError(f"Unsupported dispatch state: {state}")
        now = _utc_now()
        entry = {
            "id": "dispatch_" + secrets.token_hex(10),
            "timestamp": now,
            "provider": target.get("providerLabel") if target else None,
            "destinationTitle": target.get("title") if target else None,
            "destinationId": target.get("id") if target else None,
            "routingReason": reason,
            "state": state,
            "sha256": sha256,
            "diagnostic": bool(diagnostic),
            "phases": [{"state": state, "at": now}],
            "receipt": self._receipt_metadata(receipt),
        }
        with self._lock:
            history, issue = self._read_history_checked()
            if issue:
                raise self._history_error()
            history["entries"] = [entry, *history["entries"]][:MAX_HISTORY_ENTRIES]
            try:
                self._write_history(history)
            except OSError as error:
                raise self._history_error() from error
        return entry

    def _update_history(self, entry_id: str, state: str, receipt: dict | None = None) -> bool:
        if state not in ALLOWED_STATES:
            return False
        with self._lock:
            history, issue = self._read_history_checked()
            if issue:
                return False
            updated = False
            for entry in history["entries"]:
                if entry.get("id") != entry_id:
                    continue
                if entry.get("state") != state:
                    entry["state"] = state
                    entry.setdefault("phases", []).append({"state": state, "at": _utc_now()})
                if receipt:
                    entry.setdefault("receipt", {}).update(self._receipt_metadata(receipt))
                updated = True
                break
            if not updated:
                return False
            try:
                self._write_history(history)
            except OSError:
                return False
            return True

    @staticmethod
    def _receipt_metadata(receipt: dict | None) -> dict:
        """Retain delivery facts only; never persist provider error/body echoes."""
        allowed = {
            "agentMessageObserved",
            "appServerAccepted",
            "clientUserMessageId",
            "code",
            "completedAt",
            "contract",
            "deliveryMode",
            "deliveryAttempted",
            "desktopOwnerAccepted",
            "msgId",
            "ownerClientId",
            "providerPreserved",
            "phase",
            "reconciliationRequired",
            "retrySafe",
            "socketWritten",
            "transcriptMetadataObserved",
            "transcriptObservationTimedOut",
            "transcriptObserved",
            "transport",
            "turnId",
            "turnStarted",
            "turnStatus",
        }
        if not isinstance(receipt, dict):
            return {}
        return {
            key: value
            for key, value in receipt.items()
            if key in allowed and isinstance(value, (str, int, float, bool, type(None)))
        }

    def _board_agents(self) -> list[dict]:
        if not self.board_command or not Path(self.board_command[0]).exists():
            return []
        try:
            result = subprocess.run(
                self.board_command,
                stdin=subprocess.DEVNULL,
                capture_output=True,
                text=True,
                timeout=3.0,
                check=False,
            )
            if result.returncode != 0 or len(result.stdout.encode("utf-8")) > 4 * 1024 * 1024:
                return []
            value = json.loads(result.stdout)
            return value.get("agents", []) if isinstance(value.get("agents"), list) else []
        except (OSError, subprocess.SubprocessError, ValueError, TypeError):
            return []

    @staticmethod
    def _normalized_endpoint(provider: str, endpoint: object) -> str:
        value = str(endpoint or "")
        prefix = provider.lower() + ":"
        return value[len(prefix) :] if value.lower().startswith(prefix) else value

    @staticmethod
    def _board_state(agent: dict | None) -> str:
        if not agent:
            return "not registered"
        status = str(agent.get("status") or "unknown").lower()
        try:
            observed = datetime.fromisoformat(str(agent.get("last_seen_at") or "").replace("Z", "+00:00"))
            if (datetime.now(tz=timezone.utc) - observed).total_seconds() > 900:
                return "stale"
        except (ValueError, TypeError):
            pass
        return status

    def _list_codex_threads(self) -> list[dict]:
        if not self.codex_command:
            return []
        client = CodexAppServerClient(self.codex_command)
        data: list[dict] = []
        try:
            client.start()
            cursor = None
            for _ in range(3):
                params = {
                    "archived": False,
                    "limit": 200,
                    "sortKey": "recency_at",
                    "sortDirection": "desc",
                    "useStateDbOnly": True,
                }
                if cursor:
                    params["cursor"] = cursor
                response = client.request("thread/list", params, timeout=8.0)
                page = response.get("data")
                if not isinstance(page, list):
                    break
                data.extend(item for item in page if isinstance(item, dict))
                cursor = response.get("nextCursor")
                if not cursor:
                    break
        except DispatchError:
            return []
        finally:
            client.close()
        return data

    def _live_claude_sessions(self) -> list[dict]:
        registry = self.home / ".claude" / "sessions"
        sessions: dict[str, dict] = {}
        root_fd = -1
        try:
            root_fd = self._open_private_directory(registry)
            with os.scandir(root_fd) as iterator:
                names: list[str] = []
                for entry in iterator:
                    if len(names) >= 1_000:
                        break
                    if re.fullmatch(r"\d+\.json", entry.name) and entry.is_file(follow_symlinks=False):
                        names.append(entry.name)
            names.sort()
        except OSError:
            if root_fd >= 0:
                os.close(root_fd)
            return []
        try:
            for name in names:
                descriptor = -1
                try:
                    flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
                    descriptor = os.open(name, flags, dir_fd=root_fd)
                    metadata = os.fstat(descriptor)
                    if (
                        not stat.S_ISREG(metadata.st_mode)
                        or metadata.st_uid != os.getuid()
                        or stat.S_IMODE(metadata.st_mode) & 0o022
                        or metadata.st_size < 2
                        or metadata.st_size > 262_144
                    ):
                        continue
                    chunks: list[bytes] = []
                    remaining = metadata.st_size
                    while remaining > 0:
                        chunk = os.read(descriptor, min(16_384, remaining))
                        if not chunk:
                            break
                        chunks.append(chunk)
                        remaining -= len(chunk)
                    record = json.loads(b"".join(chunks).decode("utf-8"))
                    session_id = str(record.get("sessionId") or "")
                    socket_path = Path(str(record.get("messagingSocketPath") or ""))
                    pid = int(record.get("pid") or 0)
                    if not UUID_RE.fullmatch(session_id) or record.get("kind") != "interactive" or record.get("peerProtocol") != 1:
                        continue
                    if pid <= 1 or not ps_process_alive(pid) or not socket_path.is_absolute():
                        continue
                    socket_meta = socket_path.lstat()
                    if (
                        not stat.S_ISSOCK(socket_meta.st_mode)
                        or socket_meta.st_uid != os.getuid()
                        or stat.S_IMODE(socket_meta.st_mode) & 0o077
                    ):
                        continue
                    sessions[session_id] = record
                except (OSError, ValueError, TypeError, AttributeError, UnicodeError):
                    continue
                finally:
                    if descriptor >= 0:
                        os.close(descriptor)
        finally:
            if root_fd >= 0:
                os.close(root_fd)
        return list(sessions.values())

    def _brain_metadata_terms(self, query_terms: set[str]) -> set[str]:
        if not query_terms or not self.brain_service:
            return set()
        try:
            inventory = self.brain_service.scan(force=False)
        except Exception:
            return set()
        roots = [
            Path(item["path"])
            for item in inventory.get("brains", [])
            if item.get("status") == "connected" and not item.get("indexedOnly")
        ]
        deadline = time.monotonic() + 0.8
        seen = 0
        matched: set[str] = set()
        pending = list(roots)
        while pending and seen < 10_000 and time.monotonic() < deadline:
            current = pending.pop()
            try:
                with os.scandir(current) as entries:
                    for entry in entries:
                        seen += 1
                        name_terms = _tokens(Path(entry.name).stem)
                        matched.update(query_terms & name_terms)
                        if entry.is_dir(follow_symlinks=False) and not entry.name.startswith(".") and entry.name not in {
                            "node_modules", ".git", "build", "dist", "Caches"
                        }:
                            pending.append(Path(entry.path))
                        if seen >= 10_000 or time.monotonic() >= deadline:
                            break
            except OSError:
                continue
        return matched

    def _workspace_associations(self) -> dict[str, dict]:
        if self.workspace_service is None:
            return {}
        try:
            try:
                value = self.workspace_service.routing_index(force=False, cached_only=True)
            except TypeError:
                value = self.workspace_service.routing_index(force=False)
            return value if isinstance(value, dict) else {}
        except Exception:
            return {}

    def _collect_targets(self, message: str) -> tuple[list[dict], set[str], bool]:
        board_agents = self._board_agents()
        workspace_associations = self._workspace_associations()
        board_by_key: dict[tuple[str, str], dict] = {}
        for agent in board_agents:
            provider = str(agent.get("provider") or "").lower()
            endpoint = self._normalized_endpoint(provider, agent.get("endpoint"))
            if provider in {"codex", "claude"} and UUID_RE.fullmatch(endpoint):
                board_by_key[(provider, endpoint)] = agent

        targets: list[dict] = []
        for thread in self._list_codex_threads():
            thread_id = str(thread.get("id") or "")
            if not UUID_RE.fullmatch(thread_id) or thread.get("parentThreadId"):
                continue
            agent = board_by_key.get(("codex", thread_id))
            association = workspace_associations.get(thread_id) or {}
            title = str(thread.get("name") or "").strip() or f"Untitled Codex task · {thread_id[:8]}"
            runtime_status = thread.get("status") or {}
            if isinstance(runtime_status, dict):
                runtime_state = str(runtime_status.get("type") or "not loaded")
            else:
                runtime_state = str(runtime_status or "not loaded")
            routing_text = " ".join(
                [
                    title,
                    str(thread.get("preview") or "")[:500],
                    str((agent or {}).get("display_name") or ""),
                    " ".join(map(str, (agent or {}).get("capabilities") or [])),
                    " ".join(map(str, (agent or {}).get("writable_scopes") or [])),
                    str(association.get("projectName") or ""),
                    str(association.get("brainId") or ""),
                ]
            )
            targets.append(
                {
                    "provider": "codex",
                    "providerLabel": "Codex",
                    "id": thread_id,
                    "title": title,
                    "state": self._board_state(agent) if agent else runtime_state,
                    "busy": self._board_state(agent) == "active",
                    "modelProvider": str(thread.get("modelProvider") or ""),
                    "projectId": association.get("projectId"),
                    "projectName": association.get("projectName"),
                    "brainId": association.get("brainId"),
                    "parentBrainId": association.get("parentBrainId"),
                    "_routingText": routing_text,
                    "_board": bool(agent),
                    "_cwd": str(thread.get("cwd") or self.home),
                    "_path": str(thread.get("path") or ""),
                }
            )

        for session in self._live_claude_sessions():
            session_id = str(session["sessionId"])
            agent = board_by_key.get(("claude", session_id))
            association = workspace_associations.get(session_id) or {}
            title = str((agent or {}).get("display_name") or session.get("name") or "Claude session")
            routing_text = " ".join(
                [
                    title,
                    str(session.get("cwd") or ""),
                    " ".join(map(str, (agent or {}).get("capabilities") or [])),
                    " ".join(map(str, (agent or {}).get("writable_scopes") or [])),
                    str(association.get("projectName") or ""),
                    str(association.get("brainId") or ""),
                ]
            )
            targets.append(
                {
                    "provider": "claude",
                    "providerLabel": "Claude",
                    "id": session_id,
                    "title": title,
                    "state": self._board_state(agent) if agent else "live session",
                    "busy": self._board_state(agent) == "active",
                    "livePid": int(session["pid"]),
                    "projectId": association.get("projectId"),
                    "projectName": association.get("projectName"),
                    "brainId": association.get("brainId"),
                    "parentBrainId": association.get("parentBrainId"),
                    "_routingText": routing_text,
                    "_board": bool(agent),
                    "_projectKey": re.sub(r"[^A-Za-z0-9]", "-", str(session.get("cwd") or "")),
                }
            )
        query_terms = _tokens(message)
        return targets, self._brain_metadata_terms(query_terms), bool(board_agents)

    @staticmethod
    def _public_target(target: dict) -> dict:
        return {key: value for key, value in target.items() if not key.startswith("_") and key != "modelProvider"}

    def resolve(self, message: str) -> dict:
        original = str(message or "")
        size = len(original.encode("utf-8"))
        if not original.strip():
            raise DispatchError("empty_message", "Type or paste a message before resolving it")
        if size > MAX_MESSAGE_BYTES:
            raise DispatchError("message_too_large", f"Dispatch messages are limited to {MAX_MESSAGE_BYTES} bytes")
        sha256 = _sha256(original)
        targets, brain_terms, board_available = self._collect_targets(original)
        query_terms = _tokens(original)
        exact_ids = {value.lower() for value in UUID_RE.findall(original)}
        asks_claude = "claude" in query_terms or "anthropic" in query_terms
        asks_codex = bool({"codex", "chatgpt", "openai"} & query_terms)

        scored: list[dict] = []
        for target in targets:
            title_terms = _tokens(target["title"])
            routing_terms = _tokens(target.get("_routingText", ""))
            score = 0
            reasons: list[str] = []
            if target["id"].lower() in exact_ids:
                score += 1000
                reasons.append("exact ID")
            provider_requested = asks_claude or asks_codex
            provider_matches = (asks_claude and target["provider"] == "claude") or (
                asks_codex and target["provider"] == "codex"
            )
            if provider_matches:
                score += 18
                reasons.append("provider named")
            elif provider_requested:
                score -= 18
            title_overlap = query_terms & title_terms
            evidence_overlap = query_terms & routing_terms
            brain_overlap = title_terms & brain_terms
            project_overlap = query_terms & _tokens(target.get("projectName") or "")
            score += len(title_overlap) * 10
            score += len(evidence_overlap - title_overlap) * 3
            score += len(brain_overlap) * 4
            score += len(project_overlap - title_overlap) * 8
            if title_overlap:
                reasons.append("task title match")
            if evidence_overlap - title_overlap:
                reasons.append("owner/session evidence")
            if brain_overlap:
                reasons.append("connected Brain metadata")
            if project_overlap:
                reasons.append("exact project Brain association")
            if target.get("_board"):
                score += 2
            candidate = dict(target)
            candidate["score"] = score
            candidate["reason"] = ", ".join(reasons) or "No material routing evidence"
            scored.append(candidate)
        scored.sort(key=lambda item: (-item["score"], item["provider"], item["title"].lower(), item["id"]))

        resolution_id = "resolution_" + secrets.token_hex(12)
        expires = time.time() + 300
        state = "no target"
        choices: list[dict] = []
        target = None
        reason = "No existing owner or live session matched strongly enough; Dispatch will not create one."
        if scored and scored[0]["score"] >= 8:
            top = scored[0]
            second = scored[1] if len(scored) > 1 else None
            ambiguous = bool(
                top["score"] < 1000
                and second
                and second["score"] >= 8
                and second["score"] >= max(top["score"] - 4, top["score"] * 0.82)
            )
            if ambiguous:
                state = "ambiguous"
                choices = [item for item in scored[:3] if item["score"] >= 8]
                reason = "Several existing destinations have materially similar evidence; choose exactly one."
            else:
                state = "resolved"
                target = top
                choices = [top]
                brain_phrase = (
                    "Exact project/Brain association and "
                    if top.get("brainId")
                    else "Connected Brain metadata and "
                    if brain_terms
                    else ""
                )
                owner_phrase = "current Agent Board/session evidence" if board_available else "existing session metadata"
                reason = f"{brain_phrase}{owner_phrase} identify this as the materially strongest existing destination."

        with self._lock:
            self._resolutions[resolution_id] = {
                "createdAt": time.time(),
                "expiresAt": expires,
                "sha256": sha256,
                "state": state,
                "candidates": {item["id"]: item for item in choices},
                "reason": reason,
            }
            self._resolutions = {
                key: value for key, value in self._resolutions.items() if value["expiresAt"] >= time.time()
            }
        if state in {"ambiguous", "no target"}:
            self._history_entry(None, sha256, reason, state, False)
        return {
            "ok": state == "resolved",
            "schemaVersion": SCHEMA_VERSION,
            "state": state,
            "resolutionId": resolution_id,
            "messageSha256": sha256,
            "reason": reason,
            "target": self._public_target(target) if target else None,
            "choices": [self._public_target(item) for item in choices],
            "expiresAtEpoch": expires,
            "privacy": self._privacy(),
        }

    def _permission_bypass_verified(self) -> bool:
        observed, cached = self._permission_audit
        now = time.monotonic()
        if observed is not None:
            age = now - observed
            if 0.0 <= age < PERMISSION_AUDIT_CACHE_SECONDS:
                return cached
        audit = self.home / ".codex" / "skills" / "maintain-full-access-plus" / "scripts" / "audit.zsh"
        if not audit.is_file():
            self._permission_audit = (time.monotonic(), False)
            return False
        env = dict(os.environ)
        env["CODEX_THREAD_ID"] = self.source_task_id
        try:
            result = subprocess.run(
                [str(audit)],
                stdin=subprocess.DEVNULL,
                capture_output=True,
                text=True,
                timeout=15.0,
                check=False,
                env=env,
            )
            healthy = result.returncode == 0 and "FULL_ACCESS_PLUS=healthy" in (result.stdout or "")
        except (OSError, subprocess.SubprocessError):
            healthy = False
        self._permission_audit = (time.monotonic(), healthy)
        return healthy

    def _send_claude(self, target: dict, original: str, diagnostic: bool) -> tuple[str, dict]:
        node = shutil.which("node")
        if not node or not self.claude_sender.is_file():
            raise DispatchError("claude_bridge_unavailable", "The private local Claude session bridge is unavailable")
        if not self._permission_bypass_verified():
            raise DispatchError(
                "source_permission_mode_unverified",
                "The Dispatch source permission mode could not be verified; Claude delivery was refused.",
            )
        args = [
            node,
            str(self.claude_sender),
            "--session-id",
            target["id"],
            "--source-session-id",
            self.source_task_id,
            "--source-name",
            "Activity Monitor Dispatch",
            "--source-mode",
            "bypass",
            "--verify-ms",
            "8000",
        ]
        if diagnostic:
            args.append("--diagnostic")
        receipt: dict = {}
        try:
            result = subprocess.run(
                args,
                input=original,
                capture_output=True,
                text=True,
                timeout=15.0,
                check=False,
            )
        except OSError as error:
            raise DispatchError(
                "claude_delivery_failed",
                _safe_error(error),
                phase="exact-session delivery",
                delivery_attempted=False,
            ) from error
        except subprocess.SubprocessError as error:
            raise DispatchError(
                "claude_delivery_failed",
                _safe_error(error),
                phase="exact-session delivery",
                delivery_attempted=True,
                receipt={
                    "deliveryAttempted": True,
                    "retrySafe": False,
                    "reconciliationRequired": True,
                    "phase": "exact-session delivery",
                    "transport": "claude-private-session-socket",
                },
            ) from error
        raw_receipt = result.stdout or ""
        try:
            if len(raw_receipt.encode("utf-8")) > 65_536:
                raise ValueError("Claude receipt exceeds the bounded receipt limit")
            receipt = json.loads(raw_receipt)
            if not isinstance(receipt, dict):
                raise TypeError("Claude receipt is not an object")
        except (ValueError, TypeError, UnicodeError) as error:
            # The bridge process ran but returned no trustworthy pre-write
            # proof. Conservatively prevent a second send.
            raise DispatchError(
                "claude_receipt_invalid",
                _safe_error(error),
                phase="exact-session delivery",
                delivery_attempted=True,
                receipt={
                    "deliveryAttempted": True,
                    "retrySafe": False,
                    "reconciliationRequired": True,
                    "phase": "exact-session delivery",
                    "transport": "claude-private-session-socket",
                },
            ) from error
        socket_written = receipt.get("socketWritten") is True
        receipt_metadata = {
            "msgId": receipt.get("msgId"),
            "socketWritten": socket_written,
            "transcriptObserved": bool(receipt.get("transcriptObserved")),
            "contract": receipt.get("contract"),
            "deliveryAttempted": socket_written,
            "retrySafe": not socket_written,
            "reconciliationRequired": socket_written and not bool(receipt.get("transcriptObserved")),
            "phase": "exact-session delivery",
            "transport": "claude-private-session-socket",
        }
        if receipt.get("state") == "queued" and not socket_written:
            raise DispatchError(
                "claude_receipt_protocol_error",
                "Claude reported a queued delivery without confirming the socket write",
                phase="exact-session delivery",
                delivery_attempted=True,
                receipt={
                    **receipt_metadata,
                    "deliveryAttempted": True,
                    "retrySafe": False,
                    "reconciliationRequired": True,
                },
            )
        if result.returncode != 0 or receipt.get("state") != "queued":
            raise DispatchError(
                str(receipt.get("code") or "claude_delivery_failed"),
                _safe_error(receipt.get("message")),
                phase="exact-session delivery",
                delivery_attempted=socket_written,
                receipt=receipt_metadata,
            )
        if receipt.get("sessionId") != target["id"] or receipt.get("sourceSha256") != _sha256(original):
            raise DispatchError(
                "claude_receipt_mismatch",
                "Claude delivery receipt did not preserve exact identity",
                phase="exact-session delivery",
                delivery_attempted=socket_written,
                receipt=receipt_metadata,
            )
        state = "transcript observed" if receipt.get("transcriptObserved") else "queued"
        return state, receipt_metadata

    @staticmethod
    def _ipc_result(response: dict, method: str, owner_client_id: str) -> dict:
        if response.get("resultType") != "success":
            raise DispatchError("codex_owner_delivery_failed", DesktopIpcClient.error_text(response))
        if response.get("method") not in {None, method}:
            raise DispatchError("desktop_ipc_protocol_error", "Codex desktop returned the wrong method receipt")
        if response.get("handledByClientId") != owner_client_id:
            raise DispatchError("codex_owner_mismatch", "Codex desktop did not use the discovered task owner")
        value = response.get("result")
        if not isinstance(value, dict):
            raise DispatchError("desktop_ipc_protocol_error", "Codex desktop returned no owner result")
        nested = value.get("result")
        return nested if isinstance(nested, dict) else value

    @staticmethod
    def _inactive_steer_response(response: dict) -> bool:
        text = DesktopIpcClient.error_text(response)
        return bool(
            "SteerTurnInactiveError" in text
            or re.fullmatch(
                r"Cannot steer conversation [0-9a-f-]+ because its active turn already ended",
                text,
                flags=re.IGNORECASE,
            )
            or text.strip().lower() == "no active turn to steer"
        )

    def _try_desktop_codex(
        self,
        target: dict,
        original: str,
        diagnostic: bool,
    ) -> tuple[str, dict, None] | None:
        client = self.desktop_ipc_factory(self.desktop_ipc_path)
        try:
            try:
                discovery = client.request(
                    "thread-owner-discovery",
                    {"hostId": "local", "conversationId": target["id"]},
                    version=1,
                    timeout_ms=5000,
                )
            except DispatchError as error:
                # Owner discovery carries no message. A stale/not-loaded task
                # may therefore fall through once to exact app-server resume.
                # This exception is deliberately scoped to discovery; after a
                # steer/start request contains text, no fallback is allowed.
                if error.code in OWNER_DISCOVERY_FALLBACK_CODES:
                    return None
                raise
            if discovery.get("resultType") != "success":
                error_text = DesktopIpcClient.error_text(discovery).lower()
                if "no-client-found" in error_text:
                    return None
                raise DispatchError("codex_owner_discovery_failed", DesktopIpcClient.error_text(discovery))
            owner_client_id = str(discovery.get("handledByClientId") or "")
            if not UUID_RE.fullmatch(owner_client_id):
                raise DispatchError("codex_owner_discovery_failed", "Codex desktop returned no exact live owner")

            text = attributed_message(original, diagnostic)
            client_user_message_id = str(uuid.uuid4())
            restore_message_id = str(uuid.uuid4())
            cwd = str(target.get("_cwd") or self.home)
            if not Path(cwd).is_absolute():
                cwd = str(self.home)
            input_items = [{"type": "text", "text": text, "text_elements": []}]
            steer_params = {
                "conversationId": target["id"],
                "clientUserMessageId": client_user_message_id,
                "input": input_items,
                "attachments": [],
                "restoreMessage": {
                    "id": restore_message_id,
                    "text": text,
                    "context": {
                        "prompt": text,
                        "addedFiles": [],
                        "fileAttachments": [],
                        "ideContext": None,
                        "imageAttachments": [],
                        "workspaceRoots": [cwd],
                    },
                    "cwd": cwd,
                    "createdAt": int(time.time() * 1000),
                },
            }
            uncertain_receipt = {
                "clientUserMessageId": client_user_message_id,
                "ownerClientId": owner_client_id,
                "deliveryAttempted": True,
                "retrySafe": False,
                "reconciliationRequired": True,
                "phase": "exact-task resume/steer",
                "transport": "codex-desktop-owner-ipc",
                "providerPreserved": True,
            }
            try:
                steer = client.request(
                    "thread-follower-steer-turn",
                    steer_params,
                    version=1,
                    target_client_id=owner_client_id,
                    timeout_ms=12_000,
                )
            except DispatchError as error:
                raise DispatchError(
                    error.code,
                    str(error),
                    phase="exact-task resume/steer",
                    delivery_attempted=True,
                    receipt={**uncertain_receipt, "deliveryMode": "steer-uncertain"},
                ) from error
            delivery_mode = "steer"
            if steer.get("resultType") == "success":
                try:
                    result = self._ipc_result(steer, "thread-follower-steer-turn", owner_client_id)
                except DispatchError as error:
                    raise DispatchError(
                        error.code,
                        str(error),
                        phase="exact-task resume/steer",
                        delivery_attempted=True,
                        receipt={**uncertain_receipt, "deliveryMode": "steer-uncertain"},
                    ) from error
            elif self._inactive_steer_response(steer):
                try:
                    start = client.request(
                        "thread-follower-start-turn",
                        {
                            "conversationId": target["id"],
                            "turnStartParams": {
                                "input": input_items,
                                "clientUserMessageId": client_user_message_id,
                            },
                        },
                        version=1,
                        target_client_id=owner_client_id,
                        timeout_ms=12_000,
                    )
                    result = self._ipc_result(start, "thread-follower-start-turn", owner_client_id)
                except DispatchError as error:
                    raise DispatchError(
                        error.code,
                        str(error),
                        phase="exact-task resume/steer",
                        delivery_attempted=True,
                        receipt={**uncertain_receipt, "deliveryMode": "start-after-inactive-steer-uncertain"},
                    ) from error
                delivery_mode = "start-after-inactive-steer"
            else:
                raise DispatchError(
                    "codex_owner_delivery_failed",
                    DesktopIpcClient.error_text(steer),
                    phase="exact-task resume/steer",
                    delivery_attempted=True,
                    receipt={**uncertain_receipt, "deliveryMode": "steer-uncertain"},
                )

            turn = result.get("turn") if isinstance(result.get("turn"), dict) else result
            turn_id = str(turn.get("id") or turn.get("turnId") or result.get("turnId") or "")
            if not UUID_RE.fullmatch(turn_id):
                raise DispatchError(
                    "codex_turn_receipt_missing",
                    "Codex desktop returned no exact turn receipt",
                    phase="exact-task resume/steer",
                    delivery_attempted=True,
                    receipt={**uncertain_receipt, "deliveryMode": f"{delivery_mode}-receipt-uncertain"},
                )
            return "queued", {
                "turnId": turn_id,
                "clientUserMessageId": client_user_message_id,
                "desktopOwnerAccepted": True,
                "ownerClientId": owner_client_id,
                "deliveryMode": delivery_mode,
                "transport": "codex-desktop-owner-ipc",
                "providerPreserved": True,
                "deliveryAttempted": True,
                "retrySafe": False,
                "phase": "accepted",
            }, None
        finally:
            client.close()

    def _start_codex(
        self,
        target: dict,
        original: str,
        diagnostic: bool,
    ) -> tuple[str, dict, CodexAppServerClient | None]:
        desktop_result = self._try_desktop_codex(target, original, diagnostic)
        if desktop_result is not None:
            return desktop_result
        if not self.codex_command:
            raise DispatchError("codex_app_server_unavailable", "Codex app-server is unavailable")
        client = CodexAppServerClient(self.codex_command)
        try:
            client.start()
            resumed = client.request(
                "thread/resume",
                {"threadId": target["id"], "excludeTurns": True},
                timeout=8.0,
            )
            thread = resumed.get("thread") or {}
            if thread.get("id") != target["id"]:
                raise DispatchError("codex_identity_mismatch", "Codex resumed a different task")
            expected_provider = target.get("modelProvider")
            actual_provider = str(resumed.get("modelProvider") or thread.get("modelProvider") or "")
            if expected_provider and actual_provider and actual_provider != expected_provider:
                raise DispatchError("codex_provider_mismatch", "Codex provider identity changed during resume")
            client_message_id = str(uuid.uuid4())
            try:
                response = client.request(
                    "turn/start",
                    {
                        "threadId": target["id"],
                        "input": [
                            {
                                "type": "text",
                                "text": attributed_message(original, diagnostic),
                                "text_elements": [],
                            }
                        ],
                        "clientUserMessageId": client_message_id,
                    },
                    timeout=12.0,
                )
            except DispatchError as error:
                raise DispatchError(
                    error.code,
                    str(error),
                    phase="exact-task resume/steer",
                    delivery_attempted=True,
                    receipt={
                        "clientUserMessageId": client_message_id,
                        "deliveryAttempted": True,
                        "retrySafe": False,
                        "reconciliationRequired": True,
                        "phase": "exact-task resume/steer",
                        "deliveryMode": "app-server-start-uncertain",
                        "transport": "codex-exact-thread-app-server",
                        "providerPreserved": True,
                    },
                ) from error
            turn = response.get("turn") or {}
            turn_id = str(turn.get("id") or "")
            if not UUID_RE.fullmatch(turn_id):
                raise DispatchError(
                    "codex_turn_receipt_missing",
                    "Codex did not return an exact turn receipt",
                    phase="exact-task resume/steer",
                    delivery_attempted=True,
                    receipt={
                        "clientUserMessageId": client_message_id,
                        "deliveryAttempted": True,
                        "retrySafe": False,
                        "reconciliationRequired": True,
                        "phase": "exact-task resume/steer",
                        "deliveryMode": "app-server-start-receipt-uncertain",
                        "transport": "codex-exact-thread-app-server",
                        "providerPreserved": True,
                    },
                )
            return "queued", {
                "turnId": turn_id,
                "clientUserMessageId": client_message_id,
                "appServerAccepted": True,
                "transport": "codex-exact-thread-app-server",
                "providerPreserved": True,
                "deliveryAttempted": True,
                "retrySafe": False,
                "phase": "accepted",
            }, client
        except Exception:
            client.close()
            raise

    def _rollout_contains_id(self, path_value: str, client_user_message_id: str) -> bool:
        try:
            path = Path(path_value).expanduser().resolve(strict=True)
            root = (self.home / ".codex" / "sessions").resolve(strict=True)
            path.relative_to(root)
            metadata = path.stat()
            if not path.is_file() or metadata.st_uid != os.getuid():
                return False
            with path.open("rb") as handle:
                handle.seek(max(0, metadata.st_size - 2 * 1024 * 1024))
                return client_user_message_id.encode("ascii") in handle.read(2 * 1024 * 1024)
        except (OSError, ValueError, UnicodeError):
            return False

    def _claude_transcript_contains_id(
        self,
        project_key: str,
        session_id: str,
        message_id: str,
    ) -> bool:
        if (
            not re.fullmatch(r"[A-Za-z0-9-]{1,4096}", str(project_key or ""))
            or not UUID_RE.fullmatch(str(session_id or ""))
            or not UUID_RE.fullmatch(str(message_id or ""))
        ):
            return False
        descriptors: list[int] = []
        flags_directory = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0)
        try:
            root = self.home / ".claude" / "projects"
            root_fd = os.open(root, flags_directory)
            descriptors.append(root_fd)
            root_metadata = os.fstat(root_fd)
            if (
                not stat.S_ISDIR(root_metadata.st_mode)
                or root_metadata.st_uid != os.getuid()
                or stat.S_IMODE(root_metadata.st_mode) & 0o077
            ):
                return False
            project_fd = os.open(project_key, flags_directory, dir_fd=root_fd)
            descriptors.append(project_fd)
            project_metadata = os.fstat(project_fd)
            if (
                not stat.S_ISDIR(project_metadata.st_mode)
                or project_metadata.st_uid != os.getuid()
                or stat.S_IMODE(project_metadata.st_mode) & 0o077
            ):
                return False
            file_flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
            transcript_fd = os.open(f"{session_id}.jsonl", file_flags, dir_fd=project_fd)
            descriptors.append(transcript_fd)
            transcript_metadata = os.fstat(transcript_fd)
            if not stat.S_ISREG(transcript_metadata.st_mode) or transcript_metadata.st_uid != os.getuid():
                return False
            tail_bytes = min(max(0, transcript_metadata.st_size), 2 * 1024 * 1024)
            os.lseek(transcript_fd, max(0, transcript_metadata.st_size - tail_bytes), os.SEEK_SET)
            needle = message_id.encode("ascii")
            overlap = b""
            remaining = tail_bytes
            while remaining > 0:
                chunk = os.read(transcript_fd, min(16_384, remaining))
                if not chunk:
                    break
                if needle in overlap + chunk:
                    return True
                overlap = (overlap + chunk)[-max(0, len(needle) - 1):]
                remaining -= len(chunk)
            return False
        except (OSError, ValueError, UnicodeError):
            return False
        finally:
            for descriptor in reversed(descriptors):
                try:
                    os.close(descriptor)
                except OSError:
                    pass

    def _monitor_ipc_transcript(
        self,
        entry_id: str,
        rollout_path: str,
        client_user_message_id: str,
    ) -> None:
        deadline = time.monotonic() + min(self.monitor_seconds, 60.0)
        while time.monotonic() < deadline:
            if self._rollout_contains_id(rollout_path, client_user_message_id):
                self._update_history(
                    entry_id,
                    "transcript observed",
                    {"transcriptMetadataObserved": True},
                )
                return
            time.sleep(0.25)
        self._update_history(entry_id, "accepted", {"transcriptObservationTimedOut": True})

    def _monitor_uncertain_transcript(
        self,
        entry_id: str,
        rollout_path: str,
        client_user_message_id: str,
    ) -> None:
        deadline = time.monotonic() + min(self.monitor_seconds, 60.0)
        while time.monotonic() < deadline:
            if self._rollout_contains_id(rollout_path, client_user_message_id):
                self._update_history(
                    entry_id,
                    "transcript observed",
                    {"transcriptMetadataObserved": True, "reconciliationRequired": False},
                )
                return
            time.sleep(0.25)
        self._update_history(
            entry_id,
            "uncertain after send",
            {"transcriptObservationTimedOut": True, "reconciliationRequired": True},
        )

    def _monitor_uncertain_claude_transcript(
        self,
        entry_id: str,
        project_key: str,
        session_id: str,
        message_id: str,
    ) -> None:
        deadline = time.monotonic() + min(self.monitor_seconds, 60.0)
        while time.monotonic() < deadline:
            if self._claude_transcript_contains_id(project_key, session_id, message_id):
                self._update_history(
                    entry_id,
                    "transcript observed",
                    {
                        "transcriptMetadataObserved": True,
                        "reconciliationRequired": False,
                        "msgId": message_id,
                    },
                )
                return
            time.sleep(0.25)
        self._update_history(
            entry_id,
            "uncertain after send",
            {"transcriptObservationTimedOut": True, "reconciliationRequired": True, "msgId": message_id},
        )

    def _monitor_codex(self, delivery_id: str, entry_id: str, turn_id: str, client: CodexAppServerClient) -> None:
        deadline = time.monotonic() + self.monitor_seconds
        try:
            while time.monotonic() < deadline:
                message = client.next_notification(timeout=1.0)
                if not message:
                    continue
                method = str(message.get("method") or "")
                params = message.get("params") or {}
                event_turn = params.get("turn") or {}
                if method == "turn/started" and event_turn.get("id") == turn_id:
                    self._update_history(entry_id, "working", {"turnStarted": True})
                elif method == "item/completed" and params.get("turnId") == turn_id:
                    item = params.get("item") or {}
                    if str(item.get("type") or "").lower() in {"agentmessage", "agent_message"}:
                        self._update_history(entry_id, "acknowledged", {"agentMessageObserved": True})
                elif method == "turn/completed" and event_turn.get("id") == turn_id:
                    turn_state = str(event_turn.get("status") or "")
                    final_state = "completed" if turn_state == "completed" else "failed"
                    self._update_history(
                        entry_id,
                        final_state,
                        {"turnStatus": turn_state, "completedAt": event_turn.get("completedAt")},
                    )
                    return
                elif method == "_process/closed":
                    self._update_history(entry_id, "failed", {"code": "codex_observer_closed"})
                    return
            self._update_history(entry_id, "failed", {"code": "codex_completion_timeout"})
        finally:
            client.close()
            with self._lock:
                self._clients.pop(delivery_id, None)

    def _recent_delivery_conflict(
        self,
        target_id: str,
        sha256: str,
        history: dict | None = None,
    ) -> dict | None:
        blocking_states = {
            "attempting",
            "owner preflight",
            "exact-task resume/steer",
            "exact-session delivery",
            "accepted",
            "queued",
            "working",
            "acknowledged",
            "completed",
            "uncertain after send",
            "transcript observed",
        }
        now = datetime.now(tz=timezone.utc)
        source = history if isinstance(history, dict) else self._read_history()
        for entry in source.get("entries", [])[:MAX_HISTORY_ENTRIES]:
            if not isinstance(entry, dict):
                continue
            if entry.get("destinationId") != target_id or entry.get("sha256") != sha256:
                continue
            try:
                observed = datetime.fromisoformat(str(entry.get("timestamp") or "").replace("Z", "+00:00"))
                if observed.tzinfo is None:
                    observed = observed.replace(tzinfo=timezone.utc)
                age = (now - observed.astimezone(timezone.utc)).total_seconds()
                if age < 0 or age > DISPATCH_RECONCILIATION_WINDOW_SECONDS:
                    continue
            except (ValueError, TypeError, OverflowError):
                continue
            state = str(entry.get("state") or "")
            receipt = entry.get("receipt") if isinstance(entry.get("receipt"), dict) else {}
            if (
                state == "failed before send"
                and receipt.get("deliveryAttempted") is False
                and receipt.get("retrySafe") is True
            ):
                continue
            if state in blocking_states or receipt.get("deliveryAttempted") is True:
                return entry
        return None

    def send(self, resolution_id: str, target_id: str, message: str, diagnostic: bool = False) -> dict:
        original = str(message or "")
        sha256 = _sha256(original)
        with self._lock:
            resolution = self._resolutions.get(str(resolution_id))
            if not resolution or resolution["expiresAt"] < time.time():
                raise DispatchError("resolution_expired", "Resolve the message again before sending")
            if resolution["sha256"] != sha256:
                raise DispatchError("message_changed", "The message changed after resolution; resolve it again")
            target = resolution["candidates"].get(str(target_id))
            if not target:
                raise DispatchError("target_not_allowed", "Choose one destination from the current resolution")
            history, history_issue = self._read_history_checked()
            if history_issue:
                raise self._history_error()
            conflict = self._recent_delivery_conflict(target["id"], sha256, history)
            if conflict:
                raise DispatchError(
                    "dispatch_reconciliation_required",
                    "A recent metadata-only receipt says this exact message may already have reached this target; reconcile it instead of sending again",
                    phase=str(conflict.get("state") or "reconciliation"),
                    delivery_attempted=True,
                    receipt={
                        "deliveryAttempted": True,
                        "retrySafe": False,
                        "reconciliationRequired": True,
                        "phase": str(conflict.get("state") or "reconciliation"),
                    },
                )
            send_state = str(resolution.get("sendState") or "ready")
            if send_state in {"attempting", "accepted", "uncertain"}:
                raise DispatchError(
                    "dispatch_already_attempted",
                    "This exact resolved message has already been attempted; reconcile it instead of sending again",
                )
            resolution["sendState"] = "attempting"
            reason = resolution["reason"]
        entry = self._history_entry(
            target,
            sha256,
            reason,
            "owner preflight",
            bool(diagnostic),
            {"phase": "owner preflight", "deliveryAttempted": False, "retrySafe": True},
        )
        try:
            if target["provider"] == "claude":
                if not self._update_history(
                    entry["id"], "exact-session delivery", {"phase": "exact-session delivery"}
                ):
                    raise self._history_error()
                state, receipt = self._send_claude(target, original, bool(diagnostic))
                final_state = "transcript observed" if state == "transcript observed" else "accepted"
                if not self._update_history(
                    entry["id"],
                    final_state,
                    {**receipt, "phase": final_state, "deliveryAttempted": True, "retrySafe": False},
                ):
                    raise DispatchError(
                        "dispatch_history_update_failed",
                        "Delivery crossed the Claude transport boundary but its metadata-only receipt could not be persisted",
                        phase=final_state,
                        delivery_attempted=True,
                        receipt={
                            **receipt,
                            "phase": final_state,
                            "deliveryAttempted": True,
                            "retrySafe": False,
                            "reconciliationRequired": True,
                        },
                    )
                message_id = str(receipt.get("msgId") or "")
                project_key = str(target.get("_projectKey") or "")
                if (
                    final_state == "accepted"
                    and UUID_RE.fullmatch(message_id)
                    and re.fullmatch(r"[A-Za-z0-9-]{1,4096}", project_key)
                ):
                    threading.Thread(
                        target=self._monitor_uncertain_claude_transcript,
                        args=(entry["id"], project_key, target["id"], message_id),
                        daemon=True,
                    ).start()
            elif target["provider"] == "codex":
                if not self._update_history(
                    entry["id"], "exact-task resume/steer", {"phase": "exact-task resume/steer"}
                ):
                    raise self._history_error()
                state, receipt, client = self._start_codex(target, original, bool(diagnostic))
                final_state = "accepted"
                if not self._update_history(
                    entry["id"],
                    final_state,
                    {**receipt, "phase": "accepted", "deliveryAttempted": True, "retrySafe": False},
                ):
                    if client is not None:
                        client.close()
                    raise DispatchError(
                        "dispatch_history_update_failed",
                        "Delivery crossed the Codex transport boundary but its metadata-only receipt could not be persisted",
                        phase="accepted",
                        delivery_attempted=True,
                        receipt={
                            **receipt,
                            "phase": "accepted",
                            "deliveryAttempted": True,
                            "retrySafe": False,
                            "reconciliationRequired": True,
                        },
                    )
                delivery_id = entry["id"]
                if client is not None:
                    with self._lock:
                        self._clients[delivery_id] = client
                    threading.Thread(
                        target=self._monitor_codex,
                        args=(delivery_id, entry["id"], receipt["turnId"], client),
                        daemon=True,
                    ).start()
                else:
                    threading.Thread(
                        target=self._monitor_ipc_transcript,
                        args=(entry["id"], str(target.get("_path") or ""), receipt["clientUserMessageId"]),
                        daemon=True,
                    ).start()
            else:
                raise DispatchError("unsupported_provider", "Dispatch will not substitute providers")
        except DispatchError as error:
            uncertain = bool(error.delivery_attempted)
            receipt = {
                "code": error.code,
                "phase": error.phase or ("exact-task resume/steer" if target["provider"] == "codex" else "exact-session delivery"),
                "deliveryAttempted": uncertain,
                "retrySafe": not uncertain,
                "reconciliationRequired": uncertain,
                **error.receipt,
            }
            delivery_attempted = bool(uncertain or receipt.get("deliveryAttempted") is True)
            reconciliation_required = bool(
                delivery_attempted or receipt.get("reconciliationRequired") is True
            )
            retry_safe = bool(
                not delivery_attempted
                and receipt.get("retrySafe") is True
                and not reconciliation_required
            )
            final_state = "uncertain after send" if delivery_attempted else "failed before send"
            self._update_history(entry["id"], final_state, receipt)
            with self._lock:
                resolution["sendState"] = "uncertain" if reconciliation_required else "retryable"
            client_message_id = str(receipt.get("clientUserMessageId") or "")
            if delivery_attempted and UUID_RE.fullmatch(client_message_id):
                threading.Thread(
                    target=self._monitor_uncertain_transcript,
                    args=(entry["id"], str(target.get("_path") or ""), client_message_id),
                    daemon=True,
                ).start()
            claude_message_id = str(receipt.get("msgId") or "")
            claude_project_key = str(target.get("_projectKey") or "")
            if (
                delivery_attempted
                and target["provider"] == "claude"
                and UUID_RE.fullmatch(claude_message_id)
                and re.fullmatch(r"[A-Za-z0-9-]{1,4096}", claude_project_key)
            ):
                threading.Thread(
                    target=self._monitor_uncertain_claude_transcript,
                    args=(entry["id"], claude_project_key, target["id"], claude_message_id),
                    daemon=True,
                ).start()
            return {
                "ok": False,
                "schemaVersion": SCHEMA_VERSION,
                "state": final_state,
                "error": str(error),
                "code": error.code,
                "phase": receipt.get("phase"),
                "deliveryAttempted": delivery_attempted,
                "historyId": entry["id"],
                "messageSha256": sha256,
                "target": self._public_target(target),
                "retrySafe": retry_safe,
                "reconciliationRequired": reconciliation_required,
                "receipt": self._receipt_metadata(receipt),
                "privacy": self._privacy(),
            }
        with self._lock:
            resolution["sendState"] = "accepted"
        return {
            "ok": True,
            "schemaVersion": SCHEMA_VERSION,
            "state": final_state,
            "phase": final_state,
            "deliveryAttempted": True,
            "retrySafe": False,
            "reconciliationRequired": bool(receipt.get("reconciliationRequired")),
            "historyId": entry["id"],
            "messageSha256": sha256,
            "target": self._public_target(target),
            "receipt": receipt,
            "privacy": self._privacy(),
        }

    @staticmethod
    def _privacy() -> dict:
        return {
            "localOnly": True,
            "messageBodiesPersisted": False,
            "brainNoteBodiesRead": False,
            "credentialFilesRead": False,
            "providerSubstitution": False,
            "fanout": False,
            "autoCreation": False,
        }

    def state(self) -> dict:
        history, history_issue = self._read_history_checked()
        history_available = history_issue is None
        return {
            "ok": True,
            "schemaVersion": SCHEMA_VERSION,
            "history": history["entries"],
            "readiness": {
                "codexExactTask": bool(self.codex_command),
                "codexDesktopOwnerIpc": self.desktop_ipc_path.exists(),
                "claudeExactSession": bool(shutil.which("node") and self.claude_sender.is_file()),
                "historyAvailable": history_available,
                "dispatchSendAllowed": history_available,
                "historyState": "ready" if history_available else "degraded",
                "busyCodexBehavior": "exact owner steer; inactive-turn start only; no companion task",
            },
            "warnings": ([] if history_available else [
                "Existing metadata-only Dispatch history is unavailable or untrusted; sending is disabled until it is repaired or reconciled."
            ]),
            "privacy": self._privacy(),
        }


def ps_process_alive(pid: int) -> bool:
    try:
        os.kill(pid, 0)
        return True
    except OSError:
        return False
