"""Apply hub resolutions from scoped mailboxes into local requests (REQ-125).

The hub answers an imported request by publishing a scoped mailbox into the
origin repo. Two defects kept that last mile from closing:

1. Rows were matched on the mailbox row's `id` — which is the HUB's REQ id, not
   the local one. A real hub mailbox therefore never resolved the local record
   and appended a FOREIGN request into downstream state instead. Rows now
   resolve by `source_request_id` (the downstream id the hub recorded) and
   stamp `upstream_request_id` back onto the local record.
2. Nothing pulled the mailbox, so a delivered resolution sat unconsumed and the
   repo kept warning "submitted upstream, no answer". `apply_pending_mailboxes`
   scans the transport directory and is wired into session start, so resolution
   lands without a human relay.

Both operations are idempotent: re-applying a mailbox changes nothing.
"""

import argparse
import json
from pathlib import Path

MAILBOXES_DIR = ".prd_plugin/mailboxes"
OUTBOX_DIR = ".prd_plugin/outbox"
TERMINAL_STATUSES = {"implemented", "rejected", "deferred"}
REQUESTS_FILE = ".prd_plugin/state/requests.json"


def _read_json(path):
    return json.loads(Path(path).read_text(encoding="utf-8-sig"))


def _write_json(path, data):
    Path(path).parent.mkdir(parents=True, exist_ok=True)
    Path(path).write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8")


def _merge_messages(local_request, incoming_request):
    """Append incoming messages that are not already present. Returns True when
    the local thread changed."""
    changed = False
    local_thread = local_request.setdefault("thread", {})
    incoming_thread = incoming_request.get("thread", {})
    local_messages = local_thread.setdefault("messages", [])
    existing_ids = {message.get("id") for message in local_messages if isinstance(message, dict)}
    existing_signatures = {
        (message.get("id"), message.get("body"), message.get("author_agent"))
        for message in local_messages
        if isinstance(message, dict)
    }
    # Body+author identifies a message even when the hub renumbered its id, so
    # a second pull never duplicates an already-applied reply.
    existing_bodies = {
        (message.get("body"), message.get("author_agent"))
        for message in local_messages
        if isinstance(message, dict)
    }
    for message in incoming_thread.get("messages", []):
        if not isinstance(message, dict):
            continue
        signature = (message.get("id"), message.get("body"), message.get("author_agent"))
        if signature in existing_signatures:
            continue
        if (message.get("body"), message.get("author_agent")) in existing_bodies:
            continue
        incoming = dict(message)
        if incoming.get("id") in existing_ids:
            incoming["id"] = f"{incoming.get('id')}-hub"
        local_messages.append(incoming)
        existing_ids.add(incoming.get("id"))
        existing_signatures.add((incoming.get("id"), incoming.get("body"), incoming.get("author_agent")))
        existing_bodies.add((incoming.get("body"), incoming.get("author_agent")))
        changed = True
    status = incoming_thread.get("status")
    if status and local_thread.get("status") != status:
        local_thread["status"] = status
        changed = True
    return changed


def _resolve_local(incoming, by_id):
    """The local record a mailbox row refers to: its source_request_id (the
    downstream id the hub recorded) first, else the row id for same-id flows."""
    source_id = incoming.get("source_request_id")
    if source_id and source_id in by_id:
        return by_id[source_id]
    row_id = incoming.get("id")
    if row_id in by_id:
        return by_id[row_id]
    return None


def apply_mailbox(requests_path, mailbox):
    """Apply one scoped mailbox to a local requests file.

    Returns {"changed": [ids], "unmatched": [hub ids]}. Rows naming an unknown
    local record are reported, never appended: another repo's canonical record
    must not be planted in this repo's state.
    """
    data = _read_json(requests_path)
    requests = data.setdefault("requests", [])
    by_id = {request.get("id"): request for request in requests if isinstance(request, dict)}
    changed, unmatched = [], []

    for incoming in mailbox.get("requests", []):
        if not isinstance(incoming, dict):
            continue
        local = _resolve_local(incoming, by_id)
        if local is None:
            unmatched.append(incoming.get("id", "<missing-id>"))
            continue
        record_changed = False
        status = incoming.get("status")
        if status and local.get("status") != status:
            local["status"] = status
            record_changed = True
        hub_id = incoming.get("id")
        if hub_id and hub_id != local.get("id") and local.get("upstream_request_id") != hub_id:
            local["upstream_request_id"] = hub_id
            record_changed = True
        if _merge_messages(local, incoming):
            record_changed = True
        if record_changed:
            changed.append(local.get("id"))

    if changed:
        _write_json(requests_path, data)
    return {"changed": changed, "unmatched": unmatched}


def prune_resolved_outbox(repo_root=".", requests_path=None):
    """Delete outbox packages whose local request reached a terminal status
    (REQ-126).

    A sent submission that has been answered is transport residue; leaving it
    kept the origin repo's message check at "attention" forever. Packages for
    requests still awaiting an answer are never touched — an unsent or
    unresolved submission must survive.
    """
    root = Path(repo_root)
    target = Path(requests_path) if requests_path else root / REQUESTS_FILE
    report = {"pruned": [], "kept": []}
    outbox = root / OUTBOX_DIR
    if not outbox.is_dir() or not target.is_file():
        return report
    try:
        data = _read_json(target)
    except (OSError, ValueError):
        return report
    resolved = {
        request.get("id")
        for request in data.get("requests", [])
        if isinstance(request, dict) and request.get("status") in TERMINAL_STATUSES
    }
    for path in sorted(outbox.glob("*.json")):
        try:
            package = _read_json(path)
        except (OSError, ValueError):
            continue
        request_id = package.get("id") if isinstance(package, dict) else None
        if request_id in resolved:
            try:
                path.unlink()
            except OSError:
                continue
            report["pruned"].append(request_id)
        elif request_id:
            report["kept"].append(request_id)
    return report


def apply_pending_mailboxes(repo_root=".", requests_path=None):
    """Apply every scoped mailbox delivered under .prd_plugin/mailboxes/.

    Wired into session start so a hub resolution reaches local state without a
    human relay. Idempotent: applied is 0 once everything has landed.
    """
    root = Path(repo_root)
    target = Path(requests_path) if requests_path else root / REQUESTS_FILE
    report = {"applied": 0, "resolved_ids": [], "unmatched": [], "mailboxes": [],
              "pruned_outbox": [], "consumed_mailboxes": []}
    if not target.is_file():
        return report
    mailboxes_dir = root / MAILBOXES_DIR
    if not mailboxes_dir.is_dir():
        report["pruned_outbox"] = prune_resolved_outbox(root, target)["pruned"]
        return report
    for path in sorted(mailboxes_dir.rglob("*.json")):
        try:
            mailbox = _read_json(path)
        except (OSError, ValueError):
            continue
        if not isinstance(mailbox, dict) or "requests" not in mailbox:
            continue
        result = apply_mailbox(target, mailbox)
        relative = str(path.relative_to(root).as_posix())
        report["mailboxes"].append(relative)
        report["applied"] += len(result["changed"])
        report["resolved_ids"].extend(result["changed"])
        report["unmatched"].extend(result["unmatched"])
        # A fully delivered envelope is transport residue: its content now
        # lives in canonical records, and leaving it holds message_check at
        # "attention" forever. Keep any mailbox still carrying rows this repo
        # could not match — that information has not landed anywhere.
        if not result["unmatched"] and not apply_mailbox(target, mailbox)["changed"]:
            try:
                path.unlink()
                report["consumed_mailboxes"].append(relative)
            except OSError:
                pass
    # Applying a resolution makes the matching sent package stale.
    report["pruned_outbox"] = prune_resolved_outbox(root, target)["pruned"]
    return report


def main(argv=None):
    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    parser.add_argument("--requests", default=REQUESTS_FILE)
    parser.add_argument("--mailbox", help="Apply one mailbox file.")
    parser.add_argument("--repo-root", default=".")
    parser.add_argument("--all", action="store_true",
                        help="Apply every mailbox under .prd_plugin/mailboxes/ (session-start default).")
    args = parser.parse_args(argv)

    if args.all or not args.mailbox:
        report = apply_pending_mailboxes(args.repo_root, args.requests)
        print(json.dumps(report, indent=2))
        return 0
    result = apply_mailbox(args.requests, _read_json(args.mailbox))
    print(json.dumps(result, indent=2))
    return 0


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