"""Shared watcher for both sides.

Each side runs this with --side <node-id>. Watches the inbox addressed to

that side, verifies HMAC, optionally gates on human ack, then spawns

dispatch-exec with a per-side concurrency cap.

Honors the killswitch file. Concurrency cap = DISPATCH_MAX_PARALLEL (default 2).

"""

from __future__ import annotations

import argparse

import json

import os

import pathlib

import subprocess

import sys

import time

sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent))

import dispatch_lib as d

POLL = int(os.environ.get("DISPATCH_POLL_SEC", "3"))

MAX_PARALLEL = int(os.environ.get("DISPATCH_MAX_PARALLEL", "2"))

EXEC_BIN = os.environ.get(

    "DISPATCH_EXEC_BIN",

    str(pathlib.Path(__file__).resolve().parent / "dispatch-exec"),

)

STATE_FILE = pathlib.Path(os.environ.get("DISPATCH_STATE_FILE", "/tmp/dispatch-watcher-state.json"))

def lanes(side: str) -> dict[str, pathlib.Path]:

    other = d.NODE_B if side == d.NODE_A else d.NODE_A

    tree = d.ROOT / f"{other}-to-{side}"

    return {

        "inbox":       tree / "inbox",

        "pending_ack": tree / "pending-ack",

        "processing":  tree / "processing",

        "done":        tree / "done",

        "rejected":    tree / "rejected",

        "pending_json": tree / "pending.json",

    }

def active_exec_count() -> int:

    n = 0

    proc = pathlib.Path("/proc")

    if not proc.exists():

        return 0

    for p in proc.iterdir():

        if not p.name.isdigit():

            continue

        try:

            cmdline = (p / "cmdline").read_bytes().replace(b"\x00", b" ").decode(errors="replace")

        except Exception:

            continue

        if "dispatch-exec" in cmdline:

            n += 1

    return n

def process_inbox(side: str, L: dict) -> None:

    for path in sorted(L["inbox"].glob("*.json")):

        try:

            task = json.loads(path.read_text())

        except Exception as e:

            target = L["rejected"] / path.name

            try: path.rename(target)

            except Exception: pass

            d.log_event(side, "task_reject", f"parse err: {e}")

            continue

        if not d.verify(task):

            path.rename(L["rejected"] / path.name)

            d.log_event(side, "task_reject", f"bad HMAC id={task.get('id', '?')[:8]}")

            continue

        d.fill_defaults(task)

        path.write_text(json.dumps(task, indent=2))

        if task.get("require_ack"):

            target = L["pending_ack"] / path.name

            path.rename(target)

            d.log_event(side, "task_pending_ack", task["request"][:80], task["id"])

            write_pending_snapshot(L)

        else:

            target = L["processing"] / path.name

            path.rename(target)

            d.log_event(side, "task_auto", task["request"][:80], task["id"])

def process_acked(side: str, L: dict) -> None:

    for ack in sorted(L["pending_ack"].glob("*.ack")):

        stem = ack.name[:-len(".ack")]

        candidates = [L["pending_ack"] / f"{stem}.json", L["pending_ack"] / stem]

        task_path = next((p for p in candidates if p.exists()), None)

        if task_path is None:

            ack.unlink()

            continue

        target = L["processing"] / task_path.name

        task_path.rename(target)

        ack.unlink()

        write_pending_snapshot(L)

def dispatch_processing(side: str, L: dict) -> None:

    tripped, _ = d.killswitch_tripped()

    if tripped:

        return

    queued = []

    for p in sorted(L["processing"].glob("*.json")):

        marker = p.with_suffix(".running")

        if marker.exists():

            continue

        queued.append(p)

    if not queued:

        return

    slots = MAX_PARALLEL - active_exec_count()

    if slots <= 0:

        return

    for p in queued[:slots]:

        marker = p.with_suffix(".running")

        marker.write_text(str(int(time.time())))

        try:

            subprocess.Popen(

                [EXEC_BIN, str(p)],

                stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,

                start_new_session=True,

            )

        except Exception as e:

            marker.unlink(missing_ok=True)

            d.log_event(side, "exec_spawn_err", str(e)[:120])

def cleanup_markers(L: dict) -> None:

    for m in L["processing"].glob("*.running"):

        task_json = m.with_suffix(".json")

        if not task_json.exists():

            m.unlink(missing_ok=True)

def write_pending_snapshot(L: dict) -> None:

    items = []

    for p in sorted(L["pending_ack"].glob("*.json")):

        try:

            t = json.loads(p.read_text())

            items.append({

                "id": t["id"], "from": t.get("from"), "priority": t.get("priority"),

                "request": t.get("request", "")[:200], "created": t.get("created"),

            })

        except Exception:

            pass

    L["pending_json"].write_text(json.dumps({"count": len(items), "items": items}, indent=2))

def boot_once_log(side: str) -> None:

    try:

        prev = json.loads(STATE_FILE.read_text()) if STATE_FILE.exists() else {}

    except Exception:

        prev = {}

    now = int(time.time())

    last = prev.get("last_start", 0)

    if now - last > 120:

        d.log_event(side, "watcher_start", f"poll={POLL}s max_parallel={MAX_PARALLEL}")

    prev["last_start"] = now

    STATE_FILE.write_text(json.dumps(prev))

def main() -> int:

    ap = argparse.ArgumentParser()

    ap.add_argument("--side", required=True, help="this node's identifier")

    args = ap.parse_args()

    side = args.side

    L = lanes(side)

    for k, p in L.items():

        if k == "pending_json":

            continue

        p.mkdir(parents=True, exist_ok=True)

    boot_once_log(side)

    while True:

        try:

            process_inbox(side, L)

            process_acked(side, L)

            dispatch_processing(side, L)

            cleanup_markers(L)

        except Exception as e:

            d.log_event(side, "watcher_err", str(e)[:120])

        time.sleep(POLL)

if __name__ == "__main__":

    raise SystemExit(main())