"""Close a request thread through a validated entry point (REQ-113).

A finished conversation had no way to be closed, so `message_check` reported
`unresolved_threads` forever — five in this repo, four of them on requests
already `implemented`. That degrades the one dashboard that catches transport
failures: a permanently-amber signal is one people stop reading, which is how
the REQ-125/130..134 transport bugs survived four rounds of half-fixes.

Written as a native entry point under the REQ-107 architecture rather than as
another hand-written MCP handler. It takes the same
`.prd_plugin/local/mcp-state.lock` directory lock as every other canonical
mutation, so it is safe to run concurrently with the MCP server and with
`prd_reflections`.
"""

import argparse
import json
import os
import shutil
import time
import uuid
from contextlib import contextmanager
from datetime import date
from pathlib import Path

REQUESTS_REL = ".prd_plugin/state/requests.json"
LOCK_WAIT_SECONDS = 10.0
LOCK_STALE_SECONDS = 60.0

# A thread is the conversation ABOUT a request. Closing it while the request is
# still live would hide work in progress, so only a settled request qualifies.
CLOSEABLE_STATUSES = {"implemented", "rejected", "deferred"}
CLOSED_THREAD_STATUSES = {"resolved", "closed"}


@contextmanager
def _state_lock(root):
    """The same directory lock as mcp/server.cjs and scripts/prd_reflections.py."""
    lock_dir = Path(root) / ".prd_plugin" / "local" / "mcp-state.lock"
    lock_dir.parent.mkdir(parents=True, exist_ok=True)
    token = f"{os.getpid()}-{uuid.uuid4().hex}"
    deadline = time.monotonic() + LOCK_WAIT_SECONDS
    acquired = False
    while not acquired:
        try:
            lock_dir.mkdir()
            (lock_dir / "owner").write_text(token, encoding="utf-8")
            acquired = True
        # PermissionError is the same condition on Windows: a directory another
        # process is deleting sits in a delete-pending state, and mkdir on it is
        # denied rather than reported as existing. FileNotFoundError means the
        # parent went with it. Treating any of these as fatal turns ordinary
        # contention into a crashed write.
        except (FileExistsError, PermissionError, FileNotFoundError):
            lock_dir.parent.mkdir(parents=True, exist_ok=True)
            try:
                if time.time() - lock_dir.stat().st_mtime > LOCK_STALE_SECONDS:
                    grave = lock_dir.with_name(lock_dir.name + f".stale-{token}")
                    os.replace(lock_dir, grave)
                    shutil.rmtree(grave, ignore_errors=True)
                    continue
            except OSError:
                # Fall through to the deadline check rather than looping
                # straight back: a stat that keeps failing must not spin.
                pass
            if time.monotonic() >= deadline:
                raise TimeoutError(
                    "timed out waiting for the PRD Plugin state lock "
                    "(.prd_plugin/local/mcp-state.lock)")
            time.sleep(0.02)
    try:
        yield
    finally:
        shutil.rmtree(lock_dir, ignore_errors=True)


def close(root, request_id, reason, today=None):
    """Close one request thread, recording why and when.

    Returns a result dict rather than raising for ordinary refusals: an
    unclosable thread is a reportable state, not a crash. Nothing is written
    unless the close is valid, so a refusal never mutates state.
    """
    root = Path(root)
    path = root / REQUESTS_REL

    if not str(reason or "").strip():
        return {"closed": False, "id": request_id,
                "error": "a reason is required: a thread closed without one is unauditable"}
    try:
        data = json.loads(path.read_text(encoding="utf-8-sig"))
    except (OSError, ValueError) as exc:
        return {"closed": False, "id": request_id,
                "error": f"could not read {path}: {type(exc).__name__}"}

    rows = data.get("requests")
    if not isinstance(rows, list):
        return {"closed": False, "id": request_id, "error": "requests state is malformed"}

    with _state_lock(root):
        # Re-read inside the lock: another writer may have moved it since.
        try:
            data = json.loads(path.read_text(encoding="utf-8-sig"))
        except (OSError, ValueError) as exc:
            return {"closed": False, "id": request_id,
                    "error": f"could not read {path}: {type(exc).__name__}"}
        request = next((r for r in data.get("requests", [])
                        if isinstance(r, dict) and r.get("id") == request_id), None)
        if request is None:
            return {"closed": False, "id": request_id,
                    "error": f"{request_id} is not in {REQUESTS_REL}"}

        status = request.get("status")
        thread = request.setdefault("thread", {})
        if thread.get("status") in CLOSED_THREAD_STATUSES:
            return {"closed": True, "id": request_id, "already_closed": True,
                    "thread_status": thread.get("status")}
        if status not in CLOSEABLE_STATUSES:
            return {"closed": False, "id": request_id,
                    "error": (f"{request_id} is {status!r}; a thread is closed when its request is "
                              f"settled ({', '.join(sorted(CLOSEABLE_STATUSES))}), so closing it now "
                              f"would hide live work")}

        thread["status"] = "resolved"
        thread["closed_reason"] = str(reason).strip()
        thread["closed_at"] = str(today or date.today())
        path.write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n",
                        encoding="utf-8", newline="\n")

    return {"closed": True, "id": request_id, "thread_status": "resolved",
            "closed_reason": str(reason).strip()}


def main(argv=None):
    parser = argparse.ArgumentParser(description="Close a request thread.")
    parser.add_argument("--repo-root", default=".")
    parser.add_argument("--request-id", required=True)
    parser.add_argument("--reason", required=True)
    args = parser.parse_args(argv)
    result = close(args.repo_root, args.request_id, args.reason)
    print(json.dumps(result, indent=2, ensure_ascii=False))
    return 0 if result.get("closed") else 2


if __name__ == "__main__":
    raise SystemExit(main())