bot.py

#!/usr/bin/env python3
"""ThisCodex reference bridge — Codex CLI app-server ↔ Discord (YOLO opt-in).

This is the **reference implementation** of the bridge daemon that `launch.sh`
runs as its `LAUNCH_CMD` infra process. It is what actually sends the
full-access sandbox to the Codex app-server. `launch.sh` alone is NOT enough —
without this bridge (or one that honors the same contract), a deployed user
cannot run a full-access Codex Discord bot. See docs/yolo-bridge-contract.md.

  Discord WebSocket Gateway
    ↓ inbound @mention / DM
  bot.py (discord.py + asyncio)
    ↓ format <channel source="discord" ...> text
    ↓ JSON-RPC over WebSocket → codex app-server
  codex app-server (ws://127.0.0.1:4222)
    ↓ thread/start (1x) → turn/start (per mention)
    ↓ codex calls the mcp__discord__reply tool (taught by BOT_WD/AGENTS.md —
      with no instruction file it writes text and calls NOTHING, measured
      2026-08-09; the bridge relays the text as a fail-safe in that case)
  Discord plugin (reused as a codex MCP server)
    ↓ Discord REST API
  → Discord message sent

╔═══════════════════════════════════════════════════════════════════════════╗
║  YOLO MODE — READ BEFORE DEPLOYING                                         ║
║                                                                           ║
║  This bridge sends sandbox="danger-full-access", approvalPolicy="never"   ║
║  on BOTH thread/start AND thread/resume. That gives the model unrestricted ║
║  shell + filesystem + network on the host. It is OPT-IN, not a default:    ║
║  set THISCODEX_YOLO=1 to enable. With THISCODEX_YOLO unset the bridge runs ║
║  the SAFE default sandbox ("workspace-write"). There is no interactive     ║
║  approval UI in a headless bridge, so risky ops are DENIED by default      ║
║  (the bridge answers approval requests with "cancel"), not prompted.       ║
║                                                                           ║
║  Untrusted Discord text → a YOLO model = arbitrary code execution on your  ║
║  machine. Only enable on a host you control with a TRUSTED private Discord ║
║  server. See docs/yolo-bridge-contract.md §Security.                       ║
╚═══════════════════════════════════════════════════════════════════════════╝
"""
from __future__ import annotations

import asyncio
import contextlib
import json
import os
import re
import sys
import time
import uuid
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from automation_observer import classify_automation_item, route_automation_record

# unbuffered stdout — immediate tmux pane log visibility
sys.stdout.reconfigure(line_buffering=True)
sys.stderr.reconfigure(line_buffering=True)

try:
    import discord
    import websockets
except ImportError as e:
    print(f"[FATAL] missing dependency: {e}. Run `pip install -r requirements.txt` first")
    sys.exit(1)

# The bridge's working dir is the BOT working dir (holds .codex-thread-id,
# dedup.json, bot-info.json, and the canonical auto-loaded AGENTS.md; a SOUL.md
# may remain only as the legacy fallback during migration) — NOT this
# examples/ folder. launch.sh starts the infra with cwd=$BOT_WD and exports
# BOT_WD; honor that. Never use __file__'s parent (that would scope state to
# ThisCodex/examples/ and mis-wire persona auto-load).
_bw = os.environ.get("BOT_WD")
WD = Path(_bw).resolve() if _bw else Path.cwd()

# Bot identity is NOT hardcoded — set BOT_NAME to your bot's channel-state slug.
# Token/state dir resolution (.env with DISCORD_BOT_TOKEN=..., yolo/QA sentinels):
#   1) $DISCORD_STATE_DIR — the guided installer's promise (confirm_state_dir
#      keeps token state OUTSIDE the bot working dir) and what the generated
#      run.sh / infra-launch.sh export. Claude Code plugin hosts set it too.
#   2) ~/.claude/channels/discord-<BOT_NAME>/ — Claude Code Discord plugin
#      layout, kept as fallback so one plugin install serves both bot families.
BOT_NAME = os.environ.get("BOT_NAME", "mybot")
_state_dir = os.environ.get("DISCORD_STATE_DIR", "").strip()
ENV_PATH = (Path(_state_dir) / ".env") if _state_dir else (
    Path.home() / ".claude" / "channels" / f"discord-{BOT_NAME}" / ".env"
)
DEDUP_PATH = WD / "dedup.json"
BOT_INFO_PATH = WD / "bot-info.json"
THREAD_ID_PATH = WD / ".codex-thread-id"
AUTOMATION_DIR = Path(os.environ.get(
    "THISCODEX_AUTOMATION_EVIDENCE_DIR",
    str(ENV_PATH.parent / ".thiscodex-automation"),
))
AUTOMATION_ACTIVE_TURN_PATH = AUTOMATION_DIR / "active-turn.json"
AUTOMATION_ACTIVE_FLOW_PATH = AUTOMATION_DIR / "active-flow.json"
AUTOMATION_ACTIVE_ATTEMPT_PATH = AUTOMATION_DIR / "active-attempt.json"
AUTOMATION_EVIDENCE_PATH = AUTOMATION_DIR / "browser-evidence.jsonl"
_BROWSER_PROVIDER_SERVERS = {
    value.strip()
    for value in os.environ.get(
        "THISCODEX_BROWSER_MCP_SERVERS", "playwright,claude-in-chrome"
    ).split(",")
    if value.strip()
}
_automation_seen_items: set[tuple[str, str]] = set()
_AUTOMATION_HANDOFF_RE = re.compile(
    r"thiscodex-manual-handoff|직접\s*(?:해|입력|설치|승인|로그인)|"
    r"수동으로\s*(?:진행|해|입력)|"
    r"(?:로그인|인증|계정|비밀번호|토큰|캡차|승인|입력|설치|설정|클릭|열기|완료)"
    r".{0,24}(?:해\s*주세요|해주세요|주셔야|해야\s*합니다|필요합니다)|"
    r"(?:please|you\s+(?:need|must|have)\s+to|operator\s+must|user\s+must)"
    r".{0,96}(?:sign\s*in|log\s*in|authenticate|enter|provide|approve|install|configure|click|complete)|"
    r"please\s+(?:do|complete|enter|install|approve).{0,40}manually|"
    r"manual\s+handoff", re.I | re.S,
)
_AUTOMATION_RECEIPT_RE = re.compile(
    r"<!--\s*thiscodex-automation-receipt:([a-f0-9]{48})\s*-->"
)
_THISCODEX_ROOT = Path(__file__).resolve().parents[1]
for _scripts_dir in (
    Path(os.environ.get("THISCODEX_ROOT", _THISCODEX_ROOT)) / "scripts",
    # VAULT_ROOT = your Obsidian vault path (default kept for the reference deployment)
    Path(os.environ.get("VAULT_ROOT", str(Path.home() / "obsidian-ai-vault"))) / ".claude/scripts",
):
    if _scripts_dir.exists():
        sys.path.insert(0, str(_scripts_dir))

try:
    import memory_s5
except Exception as e:
    memory_s5 = None
    print(f"[WARN] memory_s5 import failed: {e}")

CODEX_WS = os.environ.get("CODEX_WS", "ws://127.0.0.1:4222")
TURN_TIMEOUT_SEC = int(os.environ.get("CODEX_TURN_TIMEOUT", "300"))
MAX_DEDUP_ENTRIES = 5000
EXEC_INTERRUPTED_AFTER_REPLY = "interrupted_after_reply"


@dataclass
class ReplyAckState:
    """Confirmed delivery state for Discord reply tool calls in one turn."""

    message_ids: list[str] = field(default_factory=list)
    failures: list[str] = field(default_factory=list)


def _automation_private_dir() -> None:
    AUTOMATION_DIR.mkdir(parents=True, exist_ok=True, mode=0o700)
    AUTOMATION_DIR.chmod(0o700)


def _automation_write_active_turn(thread_id: str, turn_id: str) -> None:
    """Publish the one app-server turn whose MCP completions may be consumed.

    The file is bridge-owned, outside BOT_WD. The model may ask the shipped CLI
    to consume evidence, but it cannot turn a prior/cross-turn MCP call into the
    current attempt merely by passing flags.
    """
    try:
        _automation_private_dir()
        temp = AUTOMATION_ACTIVE_TURN_PATH.with_suffix(f".{os.getpid()}.tmp")
        temp.write_text(json.dumps({
            "schema_version": 1,
            "thread_id": thread_id,
            "turn_id": turn_id,
            "started_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        }))
        temp.chmod(0o600)
        temp.replace(AUTOMATION_ACTIVE_TURN_PATH)
        AUTOMATION_ACTIVE_TURN_PATH.chmod(0o600)
    except Exception as e:
        print(f"[AUTOMATION] active-turn write failed: {type(e).__name__}: {e}")


def _automation_clear_active_turn(turn_id: str | None = None) -> None:
    try:
        if turn_id and AUTOMATION_ACTIVE_TURN_PATH.exists():
            current = json.loads(AUTOMATION_ACTIVE_TURN_PATH.read_text())
            if current.get("turn_id") != turn_id:
                return
        AUTOMATION_ACTIVE_TURN_PATH.unlink(missing_ok=True)
    except Exception as e:
        print(f"[AUTOMATION] active-turn clear failed: {type(e).__name__}: {e}")


def _automation_flow_for_thread(thread_id: str) -> dict | None:
    try:
        flow = json.loads(AUTOMATION_ACTIVE_FLOW_PATH.read_text())
        started = datetime.fromisoformat(str(flow.get("started_at")).replace("Z", "+00:00"))
        fresh = (datetime.now(timezone.utc) - started).total_seconds() <= 2 * 60 * 60
        return flow if fresh and flow.get("thread_id") == thread_id and flow.get("flow") else None
    except Exception:
        return None


def _automation_attempt_for_flow(thread_id: str, turn_id: str, flow_name: str) -> dict | None:
    try:
        attempt = json.loads(AUTOMATION_ACTIVE_ATTEMPT_PATH.read_text())
        started = datetime.fromisoformat(str(attempt.get("started_at")).replace("Z", "+00:00"))
        fresh = (datetime.now(timezone.utc) - started).total_seconds() <= 15 * 60
        valid = (
            fresh
            and attempt.get("thread_id") == thread_id
            and attempt.get("turn_id") == turn_id
            and attempt.get("flow") == flow_name
            and attempt.get("attempt_id")
            and attempt.get("gate")
            and attempt.get("operation")
            and attempt.get("evidence_tool")
            and not attempt.get("observed_item_id")
        )
        return attempt if valid else None
    except Exception:
        return None


def _automation_mark_attempt_observed(attempt: dict, item_id: str) -> None:
    value = {
        **attempt,
        "observed_item_id": item_id,
        "observed_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
    }
    temp = AUTOMATION_ACTIVE_ATTEMPT_PATH.with_suffix(f".{os.getpid()}.tmp")
    temp.write_text(json.dumps(value, ensure_ascii=True))
    temp.chmod(0o600)
    temp.replace(AUTOMATION_ACTIVE_ATTEMPT_PATH)
    AUTOMATION_ACTIVE_ATTEMPT_PATH.chmod(0o600)


def _automation_bind_flow_provider(flow: dict, provider: str) -> bool:
    """Bind the first observed provider; app-server notifications are serial."""
    if flow.get("provider") and flow.get("provider") != provider:
        return False
    if flow.get("provider") == provider:
        return True
    flow = {**flow, "provider": provider,
            "updated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())}
    temp = AUTOMATION_ACTIVE_FLOW_PATH.with_suffix(f".{os.getpid()}.tmp")
    temp.write_text(json.dumps(flow, ensure_ascii=True))
    temp.chmod(0o600)
    temp.replace(AUTOMATION_ACTIVE_FLOW_PATH)
    AUTOMATION_ACTIVE_FLOW_PATH.chmod(0o600)
    return True


def observe_automation_item(item: dict, thread_id: str, turn_id: str) -> None:
    """Persist only an MCP/command completion envelope — never args or result.

    `arguments`, DOM/page text, URLs, screenshots, command strings, stdout and
    raw errors may contain credentials. They are deliberately neither copied
    nor summarized into this ledger.
    """
    if not thread_id or not turn_id or not isinstance(item, dict):
        return
    item_id = str(item.get("id") or "")
    key = (turn_id, item_id)
    if not item_id or key in _automation_seen_items:
        return

    record = classify_automation_item(item, _BROWSER_PROVIDER_SERVERS)
    if record is None:
        return

    try:
        _automation_private_dir()
        flow = _automation_flow_for_thread(str(thread_id))
        if not flow:
            return
        attempt = _automation_attempt_for_flow(str(thread_id), str(turn_id), str(flow.get("flow") if flow else ""))
        route = route_automation_record(
            record,
            str(flow.get("provider") or ""),
            str(attempt.get("evidence_tool") if attempt else ""),
        )
        auxiliary_provider = record["provider"] == "model-blind-clipboard"
        if not route["provider_allowed"]:
            return
        if not auxiliary_provider and not _automation_bind_flow_provider(flow, record["provider"]):
            return
        if not attempt or not route["evidence_eligible"]:
            return
        row = {
            "schema_version": 2,
            "thread_id": str(thread_id)[:96],
            "turn_id": str(turn_id)[:96],
            "item_id": item_id[:96],
            "attempt_id": str(attempt["attempt_id"])[:96],
            "gate": str(attempt["gate"])[:96],
            "flow": str(flow["flow"])[:96],
            "operation": str(attempt["operation"])[:96],
            **record,
            "observed_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        }
        with AUTOMATION_EVIDENCE_PATH.open("a", encoding="utf-8") as fh:
            fh.write(json.dumps(row, ensure_ascii=True) + "\n")
        AUTOMATION_EVIDENCE_PATH.chmod(0o600)
        _automation_mark_attempt_observed(attempt, item_id)
        _automation_seen_items.add(key)
    except Exception as e:
        print(f"[AUTOMATION] evidence write failed: {type(e).__name__}: {e}")


def automation_fallback_handoff_allowed(text: str) -> bool:
    """Fail closed when fallback relay would bypass the PreToolUse receipt gate."""
    try:
        active = json.loads(AUTOMATION_ACTIVE_TURN_PATH.read_text())
        active_flow = _automation_flow_for_thread(str(active.get("thread_id") or ""))
        if not active_flow and not _AUTOMATION_HANDOFF_RE.search(text or ""):
            return True
        match = _AUTOMATION_RECEIPT_RE.search(text or "")
        if not match:
            return False
        receipts_path = AUTOMATION_DIR / "handoff-receipts.jsonl"
        used_path = AUTOMATION_DIR / "used-handoff-receipts.jsonl"
        receipts = [json.loads(line) for line in receipts_path.read_text().splitlines() if line]
        token = match.group(1)
        used_marker = AUTOMATION_DIR / f"receipt-{token}.used"
        receipt = next((row for row in reversed(receipts)
                        if row.get("token") == token), None)
        expires = datetime.fromisoformat(
            str(receipt.get("expires_at")).replace("Z", "+00:00")
        )
        now = datetime.now(timezone.utc)
        valid = bool(
            receipt and not used_marker.exists() and expires >= now
            and receipt.get("thread_id") == active.get("thread_id")
            and receipt.get("turn_id") == active.get("turn_id")
            and (not active_flow or receipt.get("flow") == active_flow.get("flow"))
        )
        if not valid:
            return False
        _automation_private_dir()
        try:
            fd = os.open(used_marker, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
        except FileExistsError:
            return False
        with os.fdopen(fd, "w", encoding="utf-8") as marker:
            marker.write(json.dumps({"token": token, "used_at": now.isoformat()}))
        with used_path.open("a", encoding="utf-8") as fh:
            fh.write(json.dumps({
                "schema_version": 1, "token": token,
                "thread_id": active.get("thread_id"),
                "turn_id": active.get("turn_id"),
                "used_at": now.isoformat(),
            }, ensure_ascii=True) + "\n")
        used_path.chmod(0o600)
        if active_flow and receipt.get("resume_required") is False:
            AUTOMATION_ACTIVE_FLOW_PATH.unlink(missing_ok=True)
        return True
    except Exception as e:
        print(f"[AUTOMATION] fallback receipt validation failed: {type(e).__name__}: {e}")
        return False


def _extract_discord_message_ids(value) -> list[str]:
    found: list[str] = []
    sent_id_re = re.compile(r"\bsent\s*\(id:\s*(\d{15,25})\)")

    def visit(node) -> None:
        if isinstance(node, str):
            for match in sent_id_re.finditer(node):
                if match.group(1) not in found:
                    found.append(match.group(1))
        elif isinstance(node, dict):
            for child in node.values():
                visit(child)
        elif isinstance(node, (list, tuple)):
            for child in node:
                visit(child)

    visit(value)
    return found


def observe_discord_reply_item(item: dict, state: ReplyAckState) -> None:
    if item.get("type") != "mcpToolCall":
        return
    if item.get("server") != "discord" or item.get("tool") != "reply":
        return
    error = item.get("error")
    if item.get("status") != "completed" or error:
        state.failures.append(str(error or f"discord.reply status={item.get('status')}"))
        return
    ids = _extract_discord_message_ids(item.get("result"))
    if not ids:
        state.failures.append("discord.reply completed without message id")
        return
    for message_id in ids:
        if message_id not in state.message_ids:
            state.message_ids.append(message_id)


def _attach_reply_ack(result: dict, state: ReplyAckState | None) -> dict:
    if state is None:
        return result
    annotated = dict(result)
    if state.message_ids:
        annotated["_discord_reply_message_ids"] = list(state.message_ids)
    if state.failures:
        annotated["_discord_reply_failed"] = list(state.failures)
    if state.message_ids and not state.failures and result.get("status") != "completed":
        annotated["_bridge_status"] = EXEC_INTERRUPTED_AFTER_REPLY
    return annotated


def discord_reply_ack_marker(result: dict) -> str | None:
    if result.get("_discord_reply_failed"):
        return None
    ids = [
        str(message_id)
        for message_id in (result.get("_discord_reply_message_ids") or [])
        if str(message_id).isdigit() and 15 <= len(str(message_id)) <= 25
    ]
    return f"discord_reply:{','.join(ids)}" if ids else None

# YOLO is opt-in and selectable PER BOT. Opt-in signals — all OPERATOR-
# controlled and OUTSIDE the model's writable working dir (critical: a sentinel
# inside WD would let a model, fed untrusted Discord text in safe mode, write
# the file and self-upgrade safe→YOLO on the next bridge restart):
#   - env THISCODEX_YOLO=1                    (process-scoped, one-off launch)
#   - env THISCODEX_YOLO_FILE=/abs/path       (explicit operator path), else
#   - default sentinel  ~/.claude/channels/discord-<BOT_NAME>/.thiscodex-yolo
#     (the bridge/token state dir — per-bot, and NOT the model's cwd/WD, so the
#      model cannot create it). Neither present → safe default sandbox.
_yolo_file = os.environ.get("THISCODEX_YOLO_FILE")
_yolo_sentinel = Path(_yolo_file) if _yolo_file else (ENV_PATH.parent / ".thiscodex-yolo")
YOLO = (os.environ.get("THISCODEX_YOLO", "0") == "1") or _yolo_sentinel.is_file()
SANDBOX = "danger-full-access" if YOLO else "workspace-write"
APPROVAL_POLICY = "never" if YOLO else "on-request"

# B-fix: bridge-level progress heartbeat. A long turn (model working silently)
# emits a progress message to the originating channel every
# HEARTBEAT_INTERVAL_SEC, so a long/blocked task is never a silent gap even if
# the model forgets its proactive-report rule. 0 disables.
# Default MUST be < the turn timeout, or the timeout fires first and the
# heartbeat never sends. Clamp to guarantee at least one heartbeat per turn.
HEARTBEAT_INTERVAL_SEC = int(os.environ.get("THISCODEX_HEARTBEAT_SEC", "240"))
if 0 < TURN_TIMEOUT_SEC <= HEARTBEAT_INTERVAL_SEC:
    HEARTBEAT_INTERVAL_SEC = max(30, int(TURN_TIMEOUT_SEC * 0.8))


def load_token() -> str:
    if not ENV_PATH.exists():
        raise RuntimeError(f"DISCORD_BOT_TOKEN file not found: {ENV_PATH}")
    for line in ENV_PATH.read_text().splitlines():
        if line.startswith("DISCORD_BOT_TOKEN="):
            return line.split("=", 1)[1].strip().strip('"').strip("'")
    raise RuntimeError(f"no DISCORD_BOT_TOKEN= entry in {ENV_PATH}")


def load_dedup() -> dict:
    if not DEDUP_PATH.exists():
        return {}
    try:
        return json.loads(DEDUP_PATH.read_text() or "{}")
    except json.JSONDecodeError:
        return {}


def save_dedup(state: dict) -> None:
    if len(state) > MAX_DEDUP_ENTRIES:
        oldest = sorted(state.items(), key=lambda kv: kv[1].get("ts", 0))[: len(state) - MAX_DEDUP_ENTRIES]
        for k, _ in oldest:
            state.pop(k, None)
    DEDUP_PATH.write_text(json.dumps(state, ensure_ascii=False, indent=2))


class CodexRPC:
    """Minimal JSON-RPC client over WebSocket for the codex app-server v2 protocol."""

    def __init__(self, url: str):
        self.url = url
        self.ws = None
        self.next_id = 1
        self.pending: dict[int, asyncio.Future] = {}
        self.turn_done: dict[str, asyncio.Future] = {}
        self.reply_ack: dict[str, ReplyAckState] = {}
        self.reader_task = None
        self._lock = asyncio.Lock()

    async def connect(self) -> None:
        print(f"[CODEX-RPC] connecting to {self.url} ...")
        self.ws = await websockets.connect(self.url, max_size=None)
        self.reader_task = asyncio.create_task(self._reader())
        await self.call("initialize", {
            "clientInfo": {"name": "thiscodex_bridge", "version": "0.1.0"},
            "capabilities": {"experimentalApi": True},
        })
        # initialized notification (no id)
        await self.ws.send(json.dumps({"method": "initialized"}))
        print("[CODEX-RPC] initialized")

    async def _reader(self) -> None:
        try:
            async for raw in self.ws:
                # DEBUG: log every incoming frame (truncated)
                print(f"[WS-IN] {str(raw)[:400]}")
                try:
                    msg = json.loads(raw)
                except json.JSONDecodeError:
                    print(f"[CODEX-RPC] non-JSON frame: {str(raw)[:200]}")
                    continue

                if "id" in msg and ("result" in msg or "error" in msg):
                    fut = self.pending.pop(msg["id"], None)
                    if fut and not fut.done():
                        fut.set_result(msg)
                    continue

                method = msg.get("method", "")

                # Server-initiated JSON-RPC request — the client MUST respond, or
                # the codex turn hangs forever (turn/start then codex asks the
                # client for approval / elicitation / a tool call; ignoring it
                # never completes the turn).
                if "id" in msg and method:
                    print(f"[WS-SERVER-REQ] method={method} id={msg['id']}")
                    if method in ("item/commandExecution/requestApproval", "item/fileChange/requestApproval"):
                        # In YOLO the model should not be asked (approvalPolicy
                        # never); if it is, default-deny is the safe answer.
                        resp = {"id": msg["id"], "result": {"decision": "cancel"}}
                    elif method == "mcpServer/elicitation/request":
                        # MCP tool-call approval — codex asks the client right
                        # before calling e.g. mcp__discord__reply. accept +
                        # persist:"session" auto-approves for this session.
                        params = msg.get("params", {})
                        kind = (params.get("_meta") or {}).get("codex_approval_kind", "")
                        print(f"[ELICIT-ACCEPT] server={params.get('serverName')} kind={kind}")
                        resp = {"id": msg["id"], "result": {"action": "accept", "content": {}, "_meta": {"persist": "session"}}}
                    elif method == "item/tool/requestUserInput":
                        # Stateful Discord question shim — handled off the
                        # reader loop so a long human wait does not block
                        # other codex WS frames.
                        asyncio.create_task(self._handle_request_user_input(msg))
                        continue
                    elif method == "item/tool/call":
                        params = msg.get("params", {})
                        resp = {"id": msg["id"], "result": {
                            "success": False,
                            "contentItems": [{"type": "inputText", "text": f"Client-side tool not implemented: {params.get('tool', '?')}"}],
                        }}
                    elif method == "item/permissions/requestApproval":
                        resp = {"id": msg["id"], "result": {"decision": "cancel"}}
                    else:
                        resp = {"id": msg["id"], "error": {"code": -32601, "message": f"Unhandled server request: {method}"}}
                    await self.ws.send(json.dumps(resp))
                    continue

                # any "completed"-flavored turn notification resolves the future
                if method == "turn/completed" or method.endswith("/turn/completed") or "TurnCompleted" in method:
                    turn = msg.get("params", {}).get("turn") or msg.get("params", {})
                    tid = turn.get("id") if isinstance(turn, dict) else None
                    if tid:
                        turn = _attach_reply_ack(turn, self.reply_ack.pop(tid, None))
                        fut = self.turn_done.pop(tid, None)
                        if fut and not fut.done():
                            fut.set_result(turn)
                    else:
                        # fallback: resolve oldest pending turn
                        if self.turn_done:
                            key, fut = next(iter(self.turn_done.items()))
                            self.turn_done.pop(key, None)
                            if not fut.done():
                                fut.set_result(turn)
                elif method == "item/completed":
                    params = msg.get("params", {})
                    thread_id = params.get("threadId")
                    tid = params.get("turnId")
                    item = params.get("item")
                    if tid and isinstance(item, dict):
                        observe_discord_reply_item(
                            item,
                            self.reply_ack.setdefault(tid, ReplyAckState()),
                        )
                        observe_automation_item(item, thread_id, tid)
                elif method.startswith("item/agentMessage"):
                    # stream notification — log only
                    pass
            # async-for ended without an exception = server closed the WS
            # gracefully (ConnectionClosedOK is swallowed by the iterator).
            # Still a dead bridge — same fail-fast restart path.
            self._ws_fatal("server closed connection (graceful)")
        except websockets.ConnectionClosed as e:
            print(f"[CODEX-RPC] connection closed: {e}")
            self._ws_fatal(f"connection closed: {e}")
        except Exception as e:
            print(f"[CODEX-RPC] reader error: {type(e).__name__}: {e}")
            self._ws_fatal(f"reader error: {type(e).__name__}: {e}")

    def _ws_fatal(self, reason: str) -> None:
        """App-server WS is gone — exit fast so the supervised launcher restarts us.

        Without this, the bridge becomes a zombie: the reader is dead, every
        subsequent turn fails with ConnectionClosed, Discord messages pile up,
        and nothing restarts because the process itself never exits. The
        supervised loop in scripts/launch.sh (window 0 `while true; do
        $LAUNCH_CMD ...`) restarts the infra pair (app-server + bridge) within
        ~5s of our exit; thread continuity survives via .codex-thread-id +
        thread/resume. Exit code 17 marks a deliberate WS-loss restart."""
        print(f"[CODEX-RPC][FATAL] app-server websocket lost ({reason}) — "
              f"exiting so the supervised launcher restarts the bridge (exit 17)",
              flush=True)

        async def _notify_and_exit() -> None:
            chan = _active_origin.get("channel")
            if chan is not None:
                with contextlib.suppress(Exception):
                    await asyncio.wait_for(
                        chan.send("⚠️ codex app-server connection lost — bridge is "
                                  "restarting (supervised). If no reply lands in ~1 min, "
                                  "resend your last message.",
                                  allowed_mentions=_NO_MENTIONS),
                        timeout=5,
                    )
            os._exit(17)

        try:
            asyncio.get_running_loop().create_task(_notify_and_exit())
        except RuntimeError:
            os._exit(17)

    async def call(self, method: str, params: dict | None = None) -> dict:
        async with self._lock:
            rid = self.next_id
            self.next_id += 1
            fut = asyncio.get_running_loop().create_future()
            self.pending[rid] = fut
            req: dict = {"id": rid, "method": method}
            if params is not None:
                req["params"] = params
            await self.ws.send(json.dumps(req))
        try:
            msg = await asyncio.wait_for(fut, timeout=60)
        except asyncio.TimeoutError:
            self.pending.pop(rid, None)
            raise RuntimeError(f"timeout: {method}")
        if "error" in msg:
            raise RuntimeError(f"{method} failed: {msg['error']}")
        return msg.get("result", {})

    async def ensure_thread(self) -> str:
        if THREAD_ID_PATH.exists():
            tid = THREAD_ID_PATH.read_text().strip()
            try:
                # CONTRACT (do NOT drop): thread/resume MUST re-send sandbox +
                # approvalPolicy. If omitted, the resumed thread silently falls
                # back to the server default (workspace-write /
                # networkAccess:false) — so YOLO would apply on the very first
                # turn and then never again after the first resume. This is the
                # single nastiest bug; see docs/yolo-bridge-contract.md.
                await self.call("thread/resume", {
                    "threadId": tid,
                    "cwd": str(WD),
                    "sandbox": SANDBOX,
                    "approvalPolicy": APPROVAL_POLICY,
                })
                print(f"[CODEX-RPC] resumed thread {tid} (sandbox={SANDBOX})")
                return tid
            except Exception as e:
                print(f"[CODEX-RPC] resume failed ({e}) — starting new thread")

        res = await self.call("thread/start", {
            "cwd": str(WD),
            "sandbox": SANDBOX,
            "approvalPolicy": APPROVAL_POLICY,
            "threadSource": "user",
        })
        # spec uncertainty — try multiple shapes
        tid = None
        if isinstance(res, dict):
            tid = res.get("threadId") or (res.get("thread") or {}).get("id")
        if not tid:
            raise RuntimeError(f"thread/start did not return id: {res}")
        await self.materialize_thread_for_tui(tid)
        THREAD_ID_PATH.write_text(tid)
        print(f"[CODEX-RPC] started new thread {tid} (sandbox={SANDBOX})")
        return tid

    async def materialize_thread_for_tui(self, thread_id: str) -> None:
        """Force local rollout creation so `codex resume --remote` can attach immediately."""
        marker = {
            "type": "message",
            "role": "assistant",
            "content": [{
                "type": "output_text",
                "text": "[bridge:init] rollout materialized for local TUI attach. No action required.",
            }],
            "phase": None,
        }
        await self.call("thread/inject_items", {
            "threadId": thread_id,
            "items": [marker],
        })
        print(f"[CODEX-RPC] materialized thread rollout for TUI attach {thread_id}")

    async def _try_reconcile_turn(self, thread_id: str, turn_id: str, fut: asyncio.Future) -> None:
        """Best-effort poll for a turn that outlived the client soft timeout.

        The normal path is `turn/completed` notification. Polling is only a
        backup: if this app-server version lacks `thread/read` shape support,
        the preserved future still lets a late notification reconcile the turn.
        """
        if fut.done():
            return
        try:
            res = await self.call("thread/read", {"threadId": thread_id})
        except Exception as e:
            print(f"[CODEX-RPC] turn reconcile poll skipped: {type(e).__name__}: {e}")
            return
        payload = res.get("result", res) if isinstance(res, dict) else {}
        thread = payload.get("thread", payload) if isinstance(payload, dict) else {}
        turns = []
        if isinstance(thread, dict):
            turns = thread.get("turns") or thread.get("items") or []
        if isinstance(payload, dict) and not turns:
            turns = payload.get("turns") or []
        for turn in (turns if isinstance(turns, list) else []):
            if not isinstance(turn, dict) or turn.get("id") != turn_id:
                continue
            status = turn.get("status")
            if status and status not in ("inProgress", "running"):
                self.turn_done.pop(turn_id, None)
                if not fut.done():
                    # Reconciled turns bypass the item/completed notifications, so
                    # the ack state never saw this turn's reply calls — replay the
                    # fetched items into it, or downstream consumers read the same
                    # items for text but not for acks and double-send.
                    state = self.reply_ack.pop(turn_id, None) or ReplyAckState()
                    for item in (turn.get("items") or []):
                        if isinstance(item, dict):
                            observe_discord_reply_item(item, state)
                            observe_automation_item(item, thread_id, turn_id)
                    fut.set_result(_attach_reply_ack(turn, state))
                print(f"[CODEX-RPC] reconciled late turn via thread/read: {turn_id} status={status}")
            return

    async def send_turn(self, thread_id: str, text: str) -> dict:
        res = await self.call("turn/start", {
            "threadId": thread_id,
            "input": [{"type": "text", "text": text}],
        })
        print(f"[CODEX-RPC] turn/start response = {json.dumps(res)[:400]}")
        turn = res if isinstance(res, dict) else {}
        turn_id = turn.get("turnId") or (turn.get("turn") or {}).get("id")
        print(f"[CODEX-RPC] extracted turn_id = {turn_id}")
        if turn_id:
            # crash-recovery marker: record the live turn id so a mid-turn
            # bridge restart can reconcile this exact turn via thread/read.
            _inflight_update({"turn_id": turn_id})
            _automation_write_active_turn(thread_id, turn_id)
        if not turn_id:
            return turn
        fut = asyncio.get_running_loop().create_future()
        self.turn_done[turn_id] = fut
        started = time.monotonic()
        soft_reported = False
        while True:
            elapsed = time.monotonic() - started
            if TURN_HARD_TIMEOUT_SEC > 0 and elapsed >= TURN_HARD_TIMEOUT_SEC:
                self.turn_done.pop(turn_id, None)
                print(f"[CODEX-RPC] turn hard timeout: {turn_id}")
                return _attach_reply_ack(
                    {"status": "hard_timeout", "turnId": turn_id},
                    self.reply_ack.pop(turn_id, None),
                )
            wait_for = EFFECTIVE_TURN_TIMEOUT_SEC if not soft_reported else TURN_RECONCILE_INTERVAL_SEC
            if TURN_HARD_TIMEOUT_SEC > 0:
                wait_for = min(wait_for, max(1, int(TURN_HARD_TIMEOUT_SEC - elapsed)))
            try:
                # Shield is critical: wait_for timeout must not cancel the
                # future, because a late turn/completed notification is still
                # the canonical reconciliation path.
                return await asyncio.wait_for(asyncio.shield(fut), timeout=wait_for)
            except asyncio.TimeoutError:
                if not soft_reported:
                    soft_reported = True
                    print(f"[CODEX-RPC] turn exceeded soft timeout; preserving future for late completion: {turn_id}")
                await self._try_reconcile_turn(thread_id, turn_id, fut)
                if fut.done():
                    return fut.result()

    # ── AskUserQuestion shim for app-server requestUserInput ──
    async def _handle_request_user_input(self, msg: dict) -> None:
        params = msg.get("params", {}) or {}
        # Spec uncertainty: the public codex docs do not pin requestUserInput's
        # param shape. Parse defensively. KM `AskUserQuestion` sends a
        # `questions[]` batch (full/bootstrap multi-question); legacy single
        # {prompt,choices} is still accepted (→ 1-element, key "value").
        mid = msg.get("id")
        # BLOCKER: parse/send/wait may all raise; an unanswered JSON-RPC id
        # hangs the Codex request forever. Any failure → a guaranteed
        # {blocked:true, error:<type>} response (independent-review BLOCKER).
        try:
            specs = _normalize_questions(params)
            answers = await self._ask_via_discord(specs)
            if answers is not None:
                # answers is a map keyed by each question's stable key
                result = {"answers": answers}
            else:
                # condition 4: no answer + no default → blocked, never silently ok
                result = {"answers": {}, "blocked": True}
        except Exception as e:
            print(f"[QA] handler error: {type(e).__name__}: {e}")
            _audit("question_handler_error", {"id": mid, "error": type(e).__name__})
            result = {"answers": {}, "blocked": True, "error": type(e).__name__}
        # The JSON-RPC reply itself can fail (dead ws). Bounded best-effort
        # retry absorbs transient backpressure; a persistently dead ws is a
        # genuine transport failure no layer here can paper over — the
        # documented backstop is then EFFECTIVE_TURN_TIMEOUT_SEC on the Codex
        # side plus the bridge reconnect loop (independent-review BLOCKER:
        # honest limitation, not a silent hang). A request with no id is a
        # notification → nothing to answer; never emit an id:null reply.
        if mid is None:
            print("[QA] requestUserInput had no JSON-RPC id; nothing to answer")
            return
        for attempt in range(3):
            try:
                await self.ws.send(json.dumps({"id": mid, "result": result}))
                break
            except Exception as e:
                print(f"[QA] response send failed (attempt {attempt + 1}/3): "
                      f"{type(e).__name__}: {e}")
                if attempt < 2:
                    await asyncio.sleep(0.5 * (attempt + 1))
        else:
            _audit("question_response_undeliverable", {"id": mid})
            print("[QA] response undeliverable — ws transport dead; backstop = "
                  "turn timeout + bridge reconnect")

    async def _ask_via_discord(self, specs: list[dict]):
        # Snapshot the origin PER QUESTION at ask time — never read the shared
        # mutable _active_origin at reply time (worker.finally clears it; a
        # late reply would then pass the channel check with chat_id=None and
        # bypass condition 3 — independent-review BLOCKER). Each pending
        # question carries its own immutable origin.
        ch = _active_origin.get("channel")
        origin_user = _active_origin.get("user_id")
        origin_chat = _active_origin.get("chat_id")
        qid = uuid.uuid4().hex                       # condition 1: bridge UUID
        all_default = {s["key"]: s["default"] for s in specs
                       if s.get("default") is not None}
        defaults_map = (all_default
                        if len(all_default) == len(specs) and specs else None)
        if ch is None:
            _audit("question_no_origin", {"qid": qid, "n": len(specs)})
            return defaults_map
        # condition 2: bridge renders the fixed choice set; replies are matched
        # by number, free-form only where a question allows it.
        lines = [f"❓ please answer the following — question {qid[:8]}, "
                 f"one line per question (reply within {QUESTION_TIMEOUT_SEC}s):"]
        for i, s in enumerate(specs, 1):
            lines.append(f"Q{i} · {s['header']}: {s['question']}")
            for j, o in enumerate(s["options"], 1):
                lines.append(f"   {j}. {o}")
        lines.append("(one line per question in order — a number for a choice, "
                     "`1,3` for multi-select, or free text where allowed; you "
                     "may prefix `Q2: …`. defaults apply on timeout where set.)")
        fut = asyncio.get_running_loop().create_future()
        _pending_q[qid] = fut
        _q_meta[qid] = {"origin_user_id": origin_user, "origin_chat_id": origin_chat,
                        "specs": specs, "acc": {}, "defaults_map": defaults_map}
        _audit("question_asked", {"qid": qid, "origin_user_id": origin_user,
                                  "keys": [s["key"] for s in specs],
                                  "questions": [s["question"][:200] for s in specs]})
        rem = None
        try:
            try:
                await ch.send("\n".join(lines), allowed_mentions=_NO_MENTIONS)
            except Exception as e:
                # BLOCKER: the question never reached the user. Don't wait out
                # the full timeout on a dead question — fail fast so the
                # handler returns blocked+error immediately.
                print(f"[QA] question send failed: {type(e).__name__}: {e}")
                _audit("question_send_failed", {"qid": qid, "error": type(e).__name__})
                raise
            async def _remind():
                # MINOR: a long human wait should not look like a hung bot.
                await asyncio.sleep(max(30, QUESTION_TIMEOUT_SEC // 2))
                with contextlib.suppress(Exception):
                    await ch.send(f"⏳ question {qid[:8]} still open — "
                                  + ("defaults apply soon" if defaults_map is not None
                                     else "will block soon if unanswered"),
                                  allowed_mentions=_NO_MENTIONS)
            rem = asyncio.create_task(_remind())
            try:
                ans = await asyncio.wait_for(fut, timeout=QUESTION_TIMEOUT_SEC)
                _audit("question_answered", {"qid": qid, "answers": str(ans)[:500]})
                return ans
            except asyncio.TimeoutError:
                _audit("question_timeout", {"qid": qid,
                                            "had_defaults": defaults_map is not None})
                # condition 4: apply defaults only if EVERY question has one;
                # otherwise blocked, never a silent partial ok.
                if defaults_map is not None:
                    return defaults_map
                with contextlib.suppress(Exception):
                    await ch.send(f"⚠️ question {qid[:8]} timed out — not every "
                                  f"question had a default — task blocked "
                                  f"(bridge fail-safe).",
                                  allowed_mentions=_NO_MENTIONS)
                return None
        finally:
            if rem is not None:
                rem.cancel()
                with contextlib.suppress(asyncio.CancelledError):
                    await rem
            _pending_q.pop(qid, None)
            _q_meta.pop(qid, None)


TOKEN = load_token()
codex = CodexRPC(CODEX_WS)
queue: asyncio.Queue = asyncio.Queue()
thread_id: str | None = None

# ── AskUserQuestion shim state for app-server requestUserInput ──
# A skill calling the AskUserQuestion equivalent surfaces as a codex
# `item/tool/requestUserInput`. The bridge turns it into a stateful Discord
# question and returns the matched answer. The 6 security conditions are
# enforced here; an unenforced condition = a prompt-injection hole.
QUESTION_TIMEOUT_SEC = int(os.environ.get("THISCODEX_QUESTION_TIMEOUT", "300"))
# Multi-question policy (independent-review BLOCKER): KM `AskUserQuestion`
# sends `questions[]` (full Mode-R = 3, STEP-1 = 4, bootstrap = 2-call).
# The shim renders the WHOLE batch in ONE Discord message and accepts
# answers over one shared QUESTION_TIMEOUT budget (incrementally mergeable)
# — NOT N sequential per-question timeouts — so the turn-timeout invariant
# stays single-budget: turn ≥ max(turn, question + post-answer buffer).
QA_POST_ANSWER_BUFFER_SEC = int(os.environ.get("THISCODEX_QA_POST_ANSWER_BUFFER", "180"))
EFFECTIVE_TURN_TIMEOUT_SEC = max(TURN_TIMEOUT_SEC,
                                 QUESTION_TIMEOUT_SEC + QA_POST_ANSWER_BUFFER_SEC)
# Soft timeout no longer means "failed": it is the first reconciliation beat.
# The bridge keeps the active future so late `turn/completed` events still
# close the original Discord request and naturally backpressure later turns.
TURN_RECONCILE_INTERVAL_SEC = int(os.environ.get("THISCODEX_TURN_RECONCILE_SEC", "30"))
TURN_HARD_TIMEOUT_SEC = int(os.environ.get("THISCODEX_TURN_HARD_TIMEOUT", "0"))
# Question/reminder/timeout prompts are skill-supplied text echoed to Discord.
# Treat them as data — never let an embedded @everyone/@user become a real
# ping (independent-review P1). All shim sends pass this.
_NO_MENTIONS = discord.AllowedMentions.none()
# Forensic audit must NOT live in the model's writable WD: in safe mode the
# model cwd/WD is writable, so a model fed untrusted Discord text could
# tamper with its own audit trail. Same principle as the YOLO sentinel —
# keep it in the operator-controlled bridge state dir, outside the model's
# reach (independent-review P1). Overridable via env for ops.
AUDIT_PATH = Path(os.environ.get("THISCODEX_QA_AUDIT_FILE",
                                 str(ENV_PATH.parent / ".thiscodex-qa-audit.jsonl")))
# Mid-turn crash recovery marker: while a turn is in flight, its origin and
# turn id live here so a bridge restart (exit-17 WS-loss restart or a hard
# crash) can reconcile the orphaned turn via thread/read instead of dropping
# the originating Discord request silently. thread/resume restores codex-side
# context, but the bridge-side future dies with the process — this marker is
# the only record that a reply was still owed. Lives in the operator-
# controlled bridge state dir (same tamper-isolation principle as AUDIT_PATH).
INFLIGHT_PATH = Path(os.environ.get("THISCODEX_INFLIGHT_FILE",
                                    str(ENV_PATH.parent / ".thiscodex-inflight-turn.json")))


def _inflight_write(data: dict) -> None:
    try:
        INFLIGHT_PATH.write_text(json.dumps(data, ensure_ascii=False))
    except Exception as e:
        print(f"[INFLIGHT] write failed (fail-open): {type(e).__name__}: {e}")


def _inflight_update(extra: dict) -> None:
    try:
        cur = json.loads(INFLIGHT_PATH.read_text())
    except Exception:
        cur = {}
    cur.update(extra)
    _inflight_write(cur)


def _inflight_clear() -> None:
    try:
        INFLIGHT_PATH.unlink(missing_ok=True)
    except Exception as e:
        print(f"[INFLIGHT] clear failed: {type(e).__name__}: {e}")


async def _recover_inflight() -> None:
    """Reconcile a turn orphaned by a mid-turn bridge restart.

    Best-effort and fail-open. Marker is consumed up front so recovery can
    never loop a restart cycle. Outcomes: turn finished → recovered reply to
    origin; otherwise → explicit loss notice to origin (never a silent drop).
    """
    try:
        raw = INFLIGHT_PATH.read_text()
    except FileNotFoundError:
        return
    except Exception as e:
        print(f"[INFLIGHT] read failed: {type(e).__name__}: {e}")
        return
    _inflight_clear()
    try:
        mark = json.loads(raw)
    except Exception:
        print("[INFLIGHT] marker unparsable — dropped")
        return
    chat_id = mark.get("chat_id")
    t_id = mark.get("thread_id")
    turn_id = mark.get("turn_id")
    age_min = int((time.time() - float(mark.get("ts") or time.time())) // 60)
    channel = None
    if chat_id:
        channel = client.get_channel(int(chat_id))
        if channel is None:
            try:
                channel = await client.fetch_channel(int(chat_id))
            except Exception as e:
                print(f"[INFLIGHT] origin channel fetch failed: {type(e).__name__}: {e}")
    status, text = "unknown", None
    already_replied = False
    if t_id and turn_id:
        try:
            res = await codex.call("thread/read", {"threadId": t_id})
            payload = res.get("result", res) if isinstance(res, dict) else {}
            thread = payload.get("thread", payload) if isinstance(payload, dict) else {}
            turns = (thread.get("turns") or thread.get("items") or []) if isinstance(thread, dict) else []
            if isinstance(payload, dict) and not turns:
                turns = payload.get("turns") or []
            for turn in (turns if isinstance(turns, list) else []):
                if isinstance(turn, dict) and turn.get("id") == turn_id:
                    status = turn.get("status") or "unknown"
                    # Shape-defensive (same caveat as _try_reconcile_turn):
                    # find the last agent-authored text item, if any.
                    for item in reversed(turn.get("items") or []):
                        if isinstance(item, dict) and item.get("type") in (
                                "agentMessage", "agent_message", "assistantMessage"):
                            text = item.get("text") or item.get("content")
                            break
                    # Same items, both questions: "what did the model say" AND
                    # "did it already send" — reading only the first is how
                    # recovered turns double-send.
                    _ack = ReplyAckState()
                    for item in (turn.get("items") or []):
                        if isinstance(item, dict):
                            observe_discord_reply_item(item, _ack)
                            observe_automation_item(item, t_id, turn_id)
                    already_replied = bool(_ack.message_ids and not _ack.failures)
                    break
        except Exception as e:
            print(f"[INFLIGHT] thread/read reconcile failed: {type(e).__name__}: {e}")
    print(f"[INFLIGHT] recovered marker: turn={turn_id} status={status} "
          f"age={age_min}m text={'yes' if text else 'no'}")
    if channel is None:
        return
    try:
        if already_replied:
            print("[INFLIGHT] turn already replied via the discord tool — "
                  "skipping recovered-reply relay (would duplicate)")
        elif text and status not in ("inProgress", "running"):
            await channel.send("🔁 bridge restarted mid-turn; the codex turn had already "
                               f"finished — recovered reply:\n{str(text)[:1800]}",
                               allowed_mentions=_NO_MENTIONS)
        else:
            await channel.send(f"⚠️ bridge restarted mid-turn ({age_min}m ago) and the in-flight "
                               f"turn could not be recovered (status={status}). "
                               "Please re-send the request.",
                               allowed_mentions=_NO_MENTIONS)
    except Exception as e:
        print(f"[INFLIGHT] origin notify failed: {type(e).__name__}: {e}")


_pending_q: dict[str, asyncio.Future] = {}      # question_id → future(answers map)
_q_meta: dict[str, dict] = {}                   # question_id → {origin_*, specs[], acc{}, ...}
# origin of the turn currently being processed (worker sets this); the shim
# asks on THIS channel and only accepts a reply from THIS user (condition 3).
_active_origin: dict = {"channel": None, "chat_id": None, "user_id": None}
# condition 3: besides the original requester, only these explicitly approved
# operator user-ids may answer a question. Operator-controlled env, never
# derived from message content.
_QA_OPERATORS: set = {x for x in os.environ.get("THISCODEX_QA_OPERATORS", "").split(",") if x}


def _audit(event: str, rec: dict) -> None:
    """Condition 6: every question / response / timeout is logged. The asked
    prompt is recorded so forensic review can flag a hijacked question by a
    prompt-text ↔ answer mismatch (bridge logic itself is text-independent —
    choices/allow_free_text are params only, never driven by message text)."""
    try:
        line = json.dumps({"ts": int(time.time()), "event": event, **rec}, ensure_ascii=False)
        with open(AUDIT_PATH, "a", encoding="utf-8") as f:
            f.write(line + "\n")
    except Exception as e:
        print(f"[QA-AUDIT] write failed: {type(e).__name__}: {e}")


def _normalize_questions(params: dict) -> list[dict]:
    """KM `AskUserQuestion` sends {"questions":[{question,header,options:[{label,
    description}],multiSelect?}]}. Normalize to a stable internal spec list.
    Legacy single {prompt/question, choices/options} → a 1-element list with
    key "value" so the JSON-RPC result stays {"answers":{"value":...}}.
    Each spec key precedence: explicit id → header → q{index} (bridge-assigned,
    audited) so the answers map is stable even if the skill omits ids."""
    qs = params.get("questions")
    specs: list[dict] = []
    seen: set[str] = set()

    def _uniq(k: str, i: int) -> str:
        # independent-review P1: duplicate id/header would collapse in the
        # answers map (one reply resolving every copy). Force uniqueness
        # deterministically; the final key set is audited in question_asked.
        if k not in seen:
            seen.add(k)
            return k
        c = i
        nk = f"{k}#q{c}"
        while nk in seen:                            # loop until truly unique
            c += 1
            nk = f"{k}#q{c}"
        seen.add(nk)
        return nk

    if isinstance(qs, list) and qs:
        for i, q in enumerate(qs, 1):
            q = q if isinstance(q, dict) else {}
            header = q.get("header")
            key = _uniq(str(q.get("id") or header or f"q{i}"), i)
            raw = q.get("options") or q.get("choices") or []
            opts = [str(o.get("label") if isinstance(o, dict) else o)
                    for o in (raw if isinstance(raw, list) else [])]
            specs.append({
                "key": key,
                "header": str(header or key),
                "question": str(q.get("question") or q.get("prompt") or "(no prompt)"),
                "options": opts,
                # condition 2: free-form requires an EXPLICIT opt-in; an
                # optionless question without it is unanswerable → blocks
                # (never auto-enable free text — independent-review BLOCKER).
                "multiselect": bool(q.get("multiSelect") or q.get("multiselect")),
                "allow_free_text": bool(q.get("allow_free_text", False)),
                "default": q.get("default"),
            })
        return specs
    # legacy single-question shape
    raw = params.get("choices") or params.get("options") or []
    opts = [str(c) for c in raw] if isinstance(raw, list) else []
    specs.append({
        "key": "value",
        "header": "value",
        "question": str(params.get("prompt") or params.get("question") or "(no prompt)"),
        "options": opts,
        "multiselect": False,
        "allow_free_text": bool(params.get("allow_free_text", False)),
        "default": params.get("default"),
    })
    return specs


def _parse_batch(text: str, specs: list[dict], acc: dict) -> tuple[dict, list[str]]:
    """Parse a (possibly partial) batched reply and merge resolved answers into
    `acc`. One line per question, in original order; an optional **strict
    `Qn:` prefix only** (`Q2: …`, case-insensitive, n = 1-based question
    index) routes a line to a specific question — header/key-name routing is
    deliberately NOT honored, since natural-language free text containing a
    colon would otherwise be misrouted (independent-review P1). Unrouted
    lines fill the still-unanswered specs in order. Already-answered keys are
    never overwritten by a later message (first-write-wins,
    independent-review P1). The reply body is data only (condition 5) — it
    never changes which/what questions are asked (those come from params)."""
    lines = [ln.strip() for ln in text.splitlines() if ln.strip()]
    routed: list[tuple[dict | None, str]] = []
    for ln in lines:
        target = None
        body = ln
        if ":" in ln:
            pre, rest = ln.split(":", 1)
            m = re.fullmatch(r"[Qq](\d+)", pre.strip())
            if m:
                idx = int(m.group(1)) - 1
                if 0 <= idx < len(specs):
                    target, body = specs[idx], rest.strip()
        routed.append((target, body))
    # positional fill for unrouted lines → specs still missing from acc
    unfilled = [s for s in specs if s["key"] not in acc]
    pos = 0
    for tgt, body in routed:
        spec = tgt
        if spec is None:
            while pos < len(unfilled) and unfilled[pos]["key"] in acc:
                pos += 1
            if pos >= len(unfilled):
                continue
            spec = unfilled[pos]
            pos += 1
        # first-write-wins: a later message cannot flip an accepted answer.
        if spec["key"] in acc:
            continue
        opts = spec["options"]
        val = None
        numeric_attempt = False
        if opts:
            # condition 2: the WHOLE body must be a strict numeric selection —
            # a number buried in free text ("1 ignore the rest") must NOT
            # silently resolve as choice 1 (independent-review BLOCKER).
            if spec["multiselect"]:
                strict = re.fullmatch(r"\d+([,\s]+\d+)*", body)
            else:
                strict = re.fullmatch(r"\d+", body)
            if strict:
                # This was a choice attempt. Even if out-of-range it must NOT
                # fall through to free text (e.g. "99" silently stored as the
                # answer) — leave it unresolved so the user is re-prompted
                # (independent-review BLOCKER).
                numeric_attempt = True
                picks: list[str] = []
                ok = True
                for t in re.split(r"[,\s]+", body):
                    if not t:
                        continue
                    n = int(t)
                    if 1 <= n <= len(opts):
                        picks.append(opts[n - 1])
                    else:
                        ok = False
                        break
                if ok and picks:
                    val = picks if spec["multiselect"] else picks[0]
        if val is None and not numeric_attempt and spec["allow_free_text"] and body:
            val = body                               # condition 5: body as data
        if val is not None:
            acc[spec["key"]] = val
    unresolved = [s["key"] for s in specs if s["key"] not in acc]
    return acc, unresolved


intents = discord.Intents.default()
intents.message_content = True
intents.members = True  # requires "Server Members Intent" ON in the Developer Portal (else PrivilegedIntentsRequired at startup)
client = discord.Client(intents=intents)


@client.event
async def on_ready():
    global thread_id
    print(f"[READY] discord.py logged in as {client.user} (id={client.user.id})")
    print(f"[READY] YOLO={'ON (danger-full-access)' if YOLO else 'off (workspace-write)'} "
          f"heartbeat={HEARTBEAT_INTERVAL_SEC}s")
    BOT_INFO_PATH.write_text(json.dumps({
        "user_id": str(client.user.id),
        "name": str(client.user),
        "ready_at": int(time.time()),
    }, ensure_ascii=False, indent=2))

    # connect codex app-server (with retries — app-server may still be booting)
    for attempt in range(10):
        try:
            await codex.connect()
            break
        except Exception as e:
            print(f"[CODEX-RPC] connect attempt {attempt+1} failed: {e}")
            await asyncio.sleep(2)
    else:
        print("[FATAL] codex app-server connect failed after 10 attempts")
        await client.close()
        return

    thread_id = await codex.ensure_thread()
    print(f"[READY] codex thread = {thread_id}")
    # Reconcile a turn orphaned by a mid-turn restart BEFORE the worker can
    # start new turns (keeps the single-turn-per-thread invariant).
    await _recover_inflight()
    asyncio.create_task(worker())


async def _heartbeat(channel, started: float, stop: asyncio.Event) -> None:
    """B-fix: while a turn runs, emit a progress note every interval so a long
    or blocked task is never a silent gap (defense-in-depth alongside the
    model's proactive-report rule in canonical AGENTS.md)."""
    if HEARTBEAT_INTERVAL_SEC <= 0 or channel is None:
        return
    while not stop.is_set():
        try:
            await asyncio.wait_for(stop.wait(), timeout=HEARTBEAT_INTERVAL_SEC)
            return  # stop set → turn finished, no heartbeat needed
        except asyncio.TimeoutError:
            mins = int((time.time() - started) // 60)
            try:
                await channel.send(f"⏳ still working — {mins}m elapsed, no result yet "
                                    f"(bridge heartbeat; the bot will report when done or blocked)",
                                    allowed_mentions=_NO_MENTIONS)
            except Exception as e:
                print(f"[HEARTBEAT] send failed: {type(e).__name__}: {e}")


@client.event
async def on_message(msg: discord.Message):
    is_self = (msg.author == client.user)
    is_mentioned = (client.user in msg.mentions)
    is_dm = isinstance(msg.channel, discord.DMChannel)
    print(f"[RAW] ch={msg.channel.id} dm={is_dm} "
          f"author={msg.author} ({msg.author.id}) bot={msg.author.bot} "
          f"self={is_self} mentioned={is_mentioned} content='{msg.content[:80]}'")

    if is_self:
        return

    # ── AskUserQuestion shim: is this a reply to a pending bridge question? ──
    # Checked before the mention/dedup gates so the original requester can
    # answer without re-mentioning the bot. Enforces conditions 2,3,5.
    if _pending_q:
        for qid, fut in list(_pending_q.items()):
            meta = _q_meta.get(qid, {})
            ou = meta.get("origin_user_id")
            # condition 3: only the original requester or an approved operator
            ok_user = (str(msg.author.id) == str(ou)) or (str(msg.author.id) in _QA_OPERATORS)
            # answer must come from the SAME channel the question was asked on.
            # Use the per-question snapshot (never the shared _active_origin,
            # which worker.finally clears → would let chat_id=None match any
            # channel and bypass condition 3).
            same_ch = str(meta.get("origin_chat_id")) == str(msg.channel.id)
            if not (ok_user and same_ch):
                continue
            # strip a leading/instinctive bot mention from the answer too —
            # else "<@bot> 1" fails the digit test, falls through the mention
            # gate and enqueues a spurious NEW turn (independent-review P1).
            text = (msg.content.replace(f"<@{client.user.id}>", "")
                                .replace(f"<@!{client.user.id}>", "").strip())
            specs = meta.get("specs") or []
            acc = meta.setdefault("acc", {})
            # condition 5: the reply body is data only — parsed against the
            # fixed specs, never altering which questions are asked.
            acc, unresolved = _parse_batch(text, specs, acc)
            if unresolved:
                # From the question's origin user+channel but the batch is not
                # yet complete/valid. Consume with guidance and KEEP waiting
                # (one shared timeout) — do NOT leak a new codex turn (P1).
                _audit("question_partial_reply",
                       {"qid": qid, "resolved": list(acc.keys()),
                        "unresolved": unresolved, "raw": text[:200]})
                with contextlib.suppress(Exception):
                    todo = "; ".join(
                        f"Q{i} {s['header']}"
                        + (f" (1-{len(s['options'])})" if s["options"] else " (free text)")
                        for i, s in enumerate(specs, 1) if s["key"] in unresolved)
                    await msg.channel.send(
                        f"⚠️ question {qid[:8]}: still need — {todo}. "
                        f"One line per question, in order.",
                        allowed_mentions=_NO_MENTIONS)
                return                               # consumed, no new turn
            if not fut.done():
                fut.set_result(dict(acc))            # full answers map
            with contextlib.suppress(Exception):
                await msg.add_reaction("✅")
            return                                   # consumed; never goes to a codex turn

    # DM channel = no mention required (the bot receiving it IS the intent).
    # Guild channel/thread = mention required.
    if not is_dm and not is_mentioned:
        return

    state = load_dedup()
    if str(msg.id) in state:
        return
    state[str(msg.id)] = {"ts": int(time.time()), "ch": str(msg.channel.id)}
    save_dedup(state)

    print(f"[MSG] {msg.author} in {msg.channel} ({msg.id}): {msg.content[:120]}")
    try:
        await msg.add_reaction("🔍")
    except Exception:
        pass

    # format as a <channel> tag — codex reads it as conversation context
    clean = msg.content.replace(f"<@{client.user.id}>", "").replace(f"<@!{client.user.id}>", "").strip()
    parent_info = ""
    if hasattr(msg.channel, "parent") and msg.channel.parent is not None:
        parent_info = f' parent_chat_id="{msg.channel.parent.id}"'
    # No-reply marker (2026-07-29): a sender closing a bot-to-bot exchange can
    # mark the message "답신 불필요" / "회신 불필요" / "no reply needed" to
    # release the forced-reply directive. Without this, delayed queue delivery
    # x an unconditional reply directive yields endless mutual ack loops
    # ("acknowledged" ping-pong between two bridged bots).
    no_reply_requested = bool(
        re.search(r"답신\s*불필요|회신\s*불필요|no[- ]?reply needed", clean, re.I)
    )
    reply_directive = (
        "→ the sender marked this as needing no reply — do not reply "
        "(exception: it also contains a new question or task for you)."
        if no_reply_requested
        else "→ reply to the message above."
    )
    event = (
        f'<channel source="discord" chat_id="{msg.channel.id}"'
        f'{parent_info} message_id="{msg.id}"'
        f' user="{msg.author}" user_id="{msg.author.id}"'
        f' ts="{msg.created_at.isoformat()}">\n'
        f'{clean}\n'
        f'</channel>\n'
        f'{reply_directive}'
    )
    if memory_s5 is not None:
        try:
            reminder = memory_s5.bridge_reminder_text(clean)
            if reminder:
                print(f"[MEMORY-S5] trigger msg={msg.id} query={memory_s5.score_prompt(clean).get('query', '')[:80]}")
                event = f"{reminder}\n\n{event}"
        except Exception as e:
            print(f"[WARN] memory_s5 bridge injection failed: {type(e).__name__}: {e}")
    # Static reply rule lives in AGENTS.md (project-doc auto-loaded), NOT
    # re-injected per turn. Per-turn payload = the dynamic <channel> block +
    # one "→ reply" line only (no persona/SOUL re-announcement noise).
    await queue.put((event, msg.channel, msg.channel.id, msg.author.id))


async def _fallback_agent_text(turn: dict) -> str | None:
    """Last agent-authored text for a completed turn — inline items first, then
    a thread/read re-fetch (turn/completed payloads don't always carry items;
    same shape-defensive walk as the inflight recovery path)."""
    for item in reversed(turn.get("items") or []):
        if isinstance(item, dict) and item.get("type") in (
                "agentMessage", "agent_message", "assistantMessage"):
            return item.get("text") or item.get("content")
    t_id, turn_id = thread_id, turn.get("id")
    if not (t_id and turn_id):
        return None
    try:
        res = await codex.call("thread/read", {"threadId": t_id})
        payload = res.get("result", res) if isinstance(res, dict) else {}
        thread = payload.get("thread", payload) if isinstance(payload, dict) else {}
        turns = (thread.get("turns") or thread.get("items") or []) if isinstance(thread, dict) else []
        if isinstance(payload, dict) and not turns:
            turns = payload.get("turns") or []
        for t in (turns if isinstance(turns, list) else []):
            if isinstance(t, dict) and t.get("id") == turn_id:
                for item in reversed(t.get("items") or []):
                    if isinstance(item, dict) and item.get("type") in (
                            "agentMessage", "agent_message", "assistantMessage"):
                        return item.get("text") or item.get("content")
                break
    except Exception as e:
        print(f"[FALLBACK-RELAY] thread/read failed: {type(e).__name__}: {e}")
    return None


async def worker():
    """Serialize turns — the codex app-server is single-turn-per-thread here."""
    global thread_id
    print("[WORKER-START]")
    while True:
        event, channel, o_chat, o_user = await queue.get()
        print(f"[WORKER-GET] event size = {len(event)} chars")
        # The AskUserQuestion shim asks on, and only accepts a reply from,
        # this turn's origin (condition 3). Set before the turn can call it.
        _active_origin.update({"channel": channel, "chat_id": o_chat, "user_id": o_user})
        stop = asyncio.Event()
        hb = asyncio.create_task(_heartbeat(channel, time.time(), stop))
        blocked_reason = None
        reply_ack_marker = None
        # initialized here (not just inside the try) so the finally block
        # below can safely inspect it even if an exception fires before
        # send_turn() ever assigns it (e.g. ensure_thread()/_inflight_write()
        # raising) — an unguarded reference would otherwise be a NameError
        # that skips the rest of finally, including _inflight_clear().
        result = None
        try:
            if thread_id is None:
                thread_id = await codex.ensure_thread()
            print(f"[WORKER-SEND] turn/start to thread {thread_id} ...")
            # crash-recovery marker (turn_id filled in by send_turn): if the
            # bridge dies before the finally below, _recover_inflight() on the
            # next boot reconciles this turn instead of dropping it silently.
            _inflight_write({"chat_id": o_chat, "user_id": o_user,
                             "thread_id": thread_id, "ts": time.time(),
                             "preview": event[:200]})
            result = await codex.send_turn(thread_id, event)
            print(f"[TURN-DONE] result keys: {list(result.keys()) if isinstance(result, dict) else type(result).__name__}")
            reply_ack_marker = (
                discord_reply_ack_marker(result) if isinstance(result, dict) else None
            )
            if isinstance(result, dict) and result.get("status") == "hard_timeout":
                if reply_ack_marker:
                    print(
                        f"[TURN] {EXEC_INTERRUPTED_AFTER_REPLY}: "
                        f"suppressing hard-timeout fallback ack={reply_ack_marker}"
                    )
                else:
                    blocked_reason = f"turn hard-timed out after {TURN_HARD_TIMEOUT_SEC}s with no result"
                    # 2026-08-09 review (글재경): a hard-timeout can coincide
                    # with a discord reply tool refusal in the SAME turn — the
                    # reply was produced but rejected (e.g. allowlist), not
                    # simply "no result". Without this, the generic timeout
                    # text below discards the actionable refusal reason that
                    # the O-3 branch (below) would otherwise have surfaced.
                    _fails = result.get("_discord_reply_failed")
                    if _fails:
                        _reason = str(_fails[-1] if isinstance(_fails, list) and _fails
                                      else _fails)[:400]
                        blocked_reason += f"; discord reply tool refused: {_reason}"
            # 막힘 17 fail-safe (2026-08-09 WSL field): a turn can complete
            # "normally" with agent text but ZERO discord reply calls — e.g.
            # the bot-WD instruction file is missing, so the model never
            # learned that its text does not reach the user. That used to be
            # a silent gap (mute bot, no error anywhere). If nothing was sent
            # and nothing failed, the bridge relays the final text itself.
            elif (isinstance(result, dict)
                    and not reply_ack_marker
                    and not result.get("_discord_reply_failed")
                    and channel is not None):
                fb_text = await _fallback_agent_text(result)
                if fb_text:
                    try:
                        if automation_fallback_handoff_allowed(str(fb_text)):
                            await channel.send(
                                "📨 (bridge relay — the model wrote this reply but did not call "
                                "the discord reply tool; check BOT_WD/AGENTS.md instructions)\n"
                                + str(fb_text)[:1800],
                                allowed_mentions=_NO_MENTIONS)
                            print("[FALLBACK-RELAY] sent agent text via bridge "
                                  "(no reply tool call this turn)")
                        else:
                            await channel.send(
                                "⚠️ automatic handoff blocked — run the shipped automation "
                                "gate in this turn before asking for manual action.",
                                allowed_mentions=_NO_MENTIONS)
                            print("[FALLBACK-RELAY] blocked manual handoff without current-turn receipt")
                    except Exception as e:
                        print(f"[FALLBACK-RELAY] send failed: {type(e).__name__}: {e}")
            # O-3 (2026-08-09 field, condition table): the model DID call the
            # reply tool and the tool REFUSED (e.g. channel not allowlisted).
            # Not relaying the full text is right — the model already tried,
            # resending risks duplicates — but saying nothing is not: the
            # bridge knows the reason while the user sees dead silence.
            # Surface the tool's own error string; never invent one. The
            # bridge's direct send is discord.py, not the MCP tool, so the
            # allowlist that refused the reply does not gate this notice.
            elif (isinstance(result, dict)
                    and result.get("_discord_reply_failed")
                    and channel is not None):
                _fails = result.get("_discord_reply_failed")
                _reason = str(_fails[-1] if isinstance(_fails, list) and _fails
                              else _fails)[:400]
                _partial = (" (some replies were delivered before this)"
                            if result.get("_discord_reply_message_ids") else "")
                print(f"[REPLY-REJECTED] discord reply tool refused: {_reason}")
                try:
                    await channel.send(
                        "⚠️ (bridge notice — the model tried to reply but the "
                        f"discord tool refused: {_reason}{_partial} — check "
                        "access.json / allowlist; see README \"bot logs in but "
                        "never replies\")",
                        allowed_mentions=_NO_MENTIONS)
                except Exception as e:
                    print(f"[REPLY-REJECTED] notice send failed: {type(e).__name__}: {e}")
        except Exception as e:
            blocked_reason = f"turn dispatch error: {type(e).__name__}"
            print(f"[ERROR] turn dispatch: {type(e).__name__}: {e}")
        finally:
            stop.set()
            hb.cancel()
            with contextlib.suppress(asyncio.CancelledError):
                await hb
            # B-fix fail-safe: a timeout/exception otherwise produces NO Discord
            # message at all (the codex turn never called mcp__discord__reply).
            # Emit a generic blocked report so it is never a silent gap.
            if blocked_reason and not reply_ack_marker and channel is not None:
                # 2026-08-10 review (글재경 🔴, latent since 7b90193): the tail
                # states only what was checked. discord_reply_ack_marker()
                # returns None whenever ANY reply failed this turn, even if
                # OTHER replies in the same turn succeeded and delivered —
                # so "no reply was produced" was asserting a fact this branch
                # never verified. Check the delivered-id list directly.
                _delivered_ids = (
                    result.get("_discord_reply_message_ids")
                    if isinstance(result, dict) else None
                ) or []
                if _delivered_ids:
                    _n = len(_delivered_ids)
                    _tail = (f"{_n} repl{'y was' if _n == 1 else 'ies were'} "
                             f"delivered before the timeout")
                else:
                    _tail = "no reply was produced"
                try:
                    await channel.send(f"⚠️ blocked — {blocked_reason}; {_tail} "
                                       f"(bridge fail-safe report; check the bot host).",
                                       allowed_mentions=_NO_MENTIONS)
                except Exception as e:
                    print(f"[BLOCKED-REPORT] send failed: {type(e).__name__}: {e}")
            # turn over (normal return, timeout, or dispatch error — all
            # in-process outcomes) → the reply obligation was handled above,
            # so the crash-recovery marker must not survive into next boot.
            _inflight_clear()
            _automation_clear_active_turn()
            # turn over → invalidate origin so a late message can't match a
            # stale question against the wrong turn.
            _active_origin.update({"channel": None, "chat_id": None, "user_id": None})
            queue.task_done()


def main():
    print(f"[INFO] WD: {WD}")
    print(f"[INFO] CODEX_WS: {CODEX_WS}")
    print(f"[INFO] ENV: {ENV_PATH}")
    print(f"[INFO] YOLO: {'ON' if YOLO else 'off'}  SANDBOX: {SANDBOX}  APPROVAL: {APPROVAL_POLICY}")
    # Pre-flight for the two silent-mute causes measured 2026-08-09 (WSL):
    # both leave bot.py healthy while the discord MCP tool dies, so the
    # symptom is "bot logged in, never answers". Warn where the operator looks.
    try:
        if b"\r" in ENV_PATH.read_bytes():
            print(f"[WARN] {ENV_PATH} has CRLF line endings — bot.py tolerates them, "
                  f"but the discord MCP tool dies with 'DISCORD_BOT_TOKEN required'. "
                  f"Fix: sed -i 's/\\r$//' '{ENV_PATH}'")
    except OSError:
        pass
    _access = ENV_PATH.parent / "access.json"
    if not _access.exists():
        print(f"[WARN] {_access} missing — the discord MCP still starts (static "
              f"allowlist mode) and the bot HEARS you, but every reply is refused "
              f"('reply failed: … not allowlisted'). Copy access.json.example from "
              f"the same directory to access.json, fill in your channel/user ids, "
              f"then restart.")
    client.run(TOKEN, log_handler=None)


if __name__ == "__main__":
    main()