#!/usr/bin/env python3
"""capital-region-meeting-asr joiner agent.

One host in the distributed meeting-ASR team. Claims work from the Worker
(POST /work/claim), downloads audio per venue rules, transcribes locally with
mlx-whisper large-v3-turbo (greedy, word timestamps, en), uploads the
timestamped transcript (PUT /telemetry/transcript/...), and releases leases on
download errors. Heartbeat (with lanes[]) + log poll every 10 minutes.

Remote control (v1.2.0): every heartbeat response carries this host's desired config
(also GET /hosts/<host_label>/config): asr_lanes, home_lanes, ocr_lanes, fetch_lanes, agent_versions.
The agent reconciles to it with no login on the Mac:
  - self-update: when the Worker advertises a newer agent.py, download it, verify sha256, compile,
    swap it in, let the decode lanes finish their current meeting, release the rest, and re-exec
    with the same arguments;
  - OCR lane: when ocr_lanes/fetch_lanes > 0, install ~/meeting-docs-ocr (venv + sha256-verified
    ocr_agent.py) and start ONE ocr_agent.py child at nice 10 unless one is already running
    (launchd or by hand; ocr_agent.py holds a lock). ocr_agent.py reads the lane counts itself and
    scales its claim loops up/down; it pauses claims under memory/thermal pressure (ASR first).
  --no-self-update / --no-ocr opt out.

Telemetry is Mac-pushed on events: every lane transition POSTs /telemetry/event
{host_label, lane, event: claimed|download_start|download_done|transcribe_start|
transcribe_done|upload_done|blocked|idle, municipality_id, meeting_id, ts,
audio_minutes?, word_count?, detail?}. Nothing polls this Mac.

YouTube bot checks (v1.3.0): a "Sign in to confirm you're not a bot" wall is a session signal, not a bad
meeting. The agent refreshes cookies from this Mac's own signed-in browser (--cookies-from-browser, or
--cookie-refresh-cmd e.g. a headless Playwright export from a dedicated Chrome profile), retries ONCE,
and if it still bot-checks releases the row as 'retry' with the reason. The Worker then puts this host in a
short YouTube cooldown (10/20/40/80/120 min) and serves non-YouTube rows only; the row is never marked
failed. If non-YouTube sites also fail (true IP block) it reports kind=ip_block, posts a blocker and stops
YouTube. Heartbeats carry bot_checks_1h, last_cookie_refresh and public_ip. No proxies, no VPNs, no
parallel YouTube across profiles.

Stdlib only (mlx_whisper is imported in the decode subprocess).
The telemetry token is read from a file and never printed or logged.
"""
import argparse
import json
import os
import queue
import random
import shutil
import signal
import subprocess
import sys
import threading
import time
import urllib.error
import urllib.request
import wave
from datetime import datetime, timezone

BASE_DEFAULT = "https://capital-region-meeting-asr.canalandcountry.workers.dev"
MODEL_DEFAULT = "mlx-community/whisper-large-v3-turbo"
BROWSER_UA = (
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/129.0 Safari/537.36"
)
AGENT_VERSION = "1.3.0"
DOCS_BASE = "https://meeting-docs.canalandcountry.workers.dev"
OCR_DIR = os.path.expanduser("~/meeting-docs-ocr")
TEN_MIN = 600
STOP = threading.Event()
DRAIN = threading.Event()  # finish the current meeting(s), claim nothing new (self-update)


def utcnow():
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


class Log:
    def __init__(self, path):
        self.path = path
        self.lock = threading.Lock()

    def __call__(self, *parts):
        line = "%s %s" % (datetime.now().strftime("%Y-%m-%d %H:%M:%S"), " ".join(str(p) for p in parts))
        with self.lock:
            print(line, flush=True)
            if self.path:
                with open(self.path, "a") as f:
                    f.write(line + "\n")


LOG = Log(None)


# ---------------------------------------------------------------- HTTP client
class Api:
    def __init__(self, base, token, host_label):
        self.base = base.rstrip("/")
        self.token = token
        self.host = host_label

    def _req(self, method, path, body=None, auth=True, content_type="application/json", tries=5):
        url = self.base + path
        data = None
        if body is not None:
            data = body if isinstance(body, (bytes, bytearray)) else json.dumps(body).encode()
        delay = 2.0
        for attempt in range(tries):
            req = urllib.request.Request(url, data=data, method=method)
            req.add_header("User-Agent", "capreg-joiner/%s (%s)" % (AGENT_VERSION, self.host))
            if data is not None:
                req.add_header("Content-Type", content_type)
            if auth and self.token:
                req.add_header("Authorization", "Bearer " + self.token)
            try:
                with urllib.request.urlopen(req, timeout=120) as r:
                    raw = r.read()
                    return r.status, (json.loads(raw) if raw[:1] in (b"{", b"[") else raw.decode("utf-8", "replace"))
            except urllib.error.HTTPError as e:
                raw = e.read()
                try:
                    payload = json.loads(raw)
                except Exception:
                    payload = raw.decode("utf-8", "replace")[:500]
                if e.code == 429:
                    wait = int(e.headers.get("retry-after") or 60)
                    LOG("rate limited; sleeping", wait, "s")
                    time.sleep(wait)
                    continue
                if e.code >= 500 and attempt < tries - 1:
                    time.sleep(delay)
                    delay *= 2
                    continue
                return e.code, payload
            except Exception as e:  # network error
                if attempt < tries - 1:
                    time.sleep(delay)
                    delay *= 2
                    continue
                return 0, {"error": "network", "detail": str(e)[:300]}
        return 0, {"error": "retries_exhausted"}

    def claim(self, region, venues, exclude, dry_run, scope=None, lane=None):
        body = {"host_label": self.host, "body_priority": "governing-first", "dry_run": bool(dry_run)}
        if scope:
            body["scope"] = scope  # "home": Niskayuna -> Schenectady Co. -> neighbours until exhausted, then general
        if lane:
            body["lane"] = lane
        if region:
            body["region"] = region
        if venues:
            body["venues"] = venues
        if exclude:
            body["exclude_venues"] = exclude
        return self._req("POST", "/work/claim", body)

    def release(self, row, status, reason, cookie_refreshed=None):
        body = {
            "municipality_id": row["municipality_id"], "meeting_id": row["meeting_id"],
            "host_label": self.host, "status": status, "reason": reason[:480],
        }
        if cookie_refreshed is not None:
            body["cookie_refreshed"] = bool(cookie_refreshed)
        return self._req("POST", "/work/release", body)

    def botcheck(self, reason, kind="botcheck", row=None, cookie_refreshed=False):
        body = {"host_label": self.host, "reason": reason[:480], "kind": kind, "cookie_refreshed": bool(cookie_refreshed)}
        if row is not None:
            body["municipality_id"] = row["municipality_id"]
            body["meeting_id"] = row["meeting_id"]
        return self._req("POST", "/work/botcheck", body, tries=3)

    def heartbeat(self, queue_depth, lanes=None, yt=None):
        body = {"host_label": self.host, "queue_depth": queue_depth, "ts": utcnow()}
        if yt:
            body.update({k: v for k, v in yt.items() if v is not None})  # bot_checks_1h, last_cookie_refresh, public_ip
        if lanes is not None:
            body["lanes"] = lanes  # optional, backward compatible: [{lane, municipality_id, meeting_id, state, since}]
        return self._req("POST", "/telemetry/heartbeat", body)

    def event(self, lane, event, row=None, **extra):
        """Event-driven lane telemetry: one POST per lane transition (Mac-pushed, nobody polls)."""
        body = {"host_label": self.host, "lane": str(lane), "event": event, "ts": utcnow()}
        if row is not None:
            body["municipality_id"] = row["municipality_id"]
            body["meeting_id"] = row["meeting_id"]
        body.update({k: v for k, v in extra.items() if v is not None})
        return self._req("POST", "/telemetry/event", body, tries=3)

    def telemetry(self, row, status, **extra):
        ev = {"municipality_id": row["municipality_id"], "meeting_id": row["meeting_id"],
              "meeting_date": row.get("meeting_date"), "status": status,
              "source": row.get("telemetry_source", "town_mp4"), "ts": utcnow()}
        ev.update({k: v for k, v in extra.items() if v is not None})
        return self._req("POST", "/telemetry", ev)

    def put_transcript(self, row, payload):
        return self._req("PUT", row["transcript_put_path"] + "?host_label=" + self.host, json.dumps(payload).encode())

    def host_config(self):
        return self._req("GET", "/hosts/%s/config" % self.host, auth=False, tries=2)

    def log_since(self, since):
        return self._req("GET", "/log.json?since=%d" % since, auth=False)

    def post_log(self, kind, message, municipality_id=None, meeting_id=None):
        body = {"kind": kind, "message": ("[%s] " % self.host) + message[:1900]}
        if municipality_id:
            body["municipality_id"] = municipality_id
        if meeting_id:
            body["meeting_id"] = meeting_id
        return self._req("POST", "/log", body)


# ---------------------------------------------------------------- download
def wav_seconds(path):
    with wave.open(path, "rb") as w:
        return w.getnframes() / float(w.getframerate())


def run(cmd, timeout=None):
    p = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, timeout=timeout)
    return p.returncode, p.stdout[-4000:]


def to_wav_cmd(src, out, fetch=None, max_seconds=None):
    cmd = ["ffmpeg", "-nostdin", "-hide_banner", "-loglevel", "error", "-y"]
    if fetch is not None:
        hdr = ""
        if fetch.get("referer"):
            hdr += "Referer: %s\r\n" % fetch["referer"]
        if hdr:
            cmd += ["-headers", hdr]
        if fetch.get("browser_ua"):
            cmd += ["-user_agent", BROWSER_UA]
    cmd += ["-i", src, "-vn", "-ac", "1", "-ar", "16000", "-c:a", "pcm_s16le"]
    if max_seconds:
        cmd += ["-t", str(int(max_seconds))]
    return cmd + [out]


def is_bot_check(err):
    e = (err or "").lower()
    return ("not a bot" in e or "confirm you\u2019re not" in e or "confirm you're not" in e) and "confirm your age" not in e


def download_wav(row, buf_dir, cookies=None, max_seconds=None, yt_opts=None):
    """Return (wav_path, None) or (None, reason). yt_opts: {cookies_from_browser, player_client}."""
    yt_opts = yt_opts or {}
    fetch = row["fetch"]
    base = "%s__%s" % (row["municipality_id"].replace("/", "__"), row["meeting_id"])
    out = os.path.join(buf_dir, base + ".wav")
    part = out + ".part.wav"
    if fetch["method"] == "ffmpeg_direct":
        code, txt = run(to_wav_cmd(fetch["url"], part, fetch, max_seconds), timeout=4 * 3600)
        if code != 0:
            _rm(part)
            return None, "ffmpeg_direct: " + _last_line(txt)
    else:
        tmpl = os.path.join(buf_dir, base + ".src.%(ext)s")
        cmd = ["yt-dlp", "--no-playlist", "--no-progress", "-f", "bestaudio/best", "-o", tmpl,
               "--retries", "3", "--socket-timeout", "30"]
        if fetch.get("referer"):
            cmd += ["--referer", fetch["referer"]]
        if fetch.get("browser_ua"):
            cmd += ["--user-agent", BROWSER_UA]
        if row.get("venue") == "youtube":
            # Prefer a freshly exported jar from this Mac's own signed-in browser; else read the browser directly.
            if cookies and os.path.exists(cookies):
                cmd += ["--cookies", cookies]
            elif yt_opts.get("cookies_from_browser"):
                cmd += ["--cookies-from-browser", yt_opts["cookies_from_browser"]]
            if yt_opts.get("player_client"):
                cmd += ["--extractor-args", "youtube:player_client=%s" % yt_opts["player_client"]]
        if max_seconds:
            cmd += ["--download-sections", "*0-%d" % int(max_seconds)]
        code, txt = run(cmd + [fetch["url"]], timeout=4 * 3600)
        srcs = [os.path.join(buf_dir, f) for f in os.listdir(buf_dir) if f.startswith(base + ".src.")]
        if code != 0 or not srcs:
            for s in srcs:
                _rm(s)
            return None, "yt-dlp: " + _last_line(txt)
        code, txt = run(to_wav_cmd(srcs[0], part, None, max_seconds), timeout=3600)
        for s in srcs:
            _rm(s)
        if code != 0:
            _rm(part)
            return None, "ffmpeg: " + _last_line(txt)
    try:
        secs = wav_seconds(part)
    except Exception as e:
        _rm(part)
        return None, "bad wav: %s" % e
    if secs < 5:
        _rm(part)
        return None, "audio too short (%.1fs)" % secs
    os.replace(part, out)
    return out, None


def _rm(p):
    try:
        os.remove(p)
    except OSError:
        pass


def _last_line(txt):
    lines = [l for l in (txt or "").strip().splitlines() if l.strip()]
    return (lines[-1] if lines else "unknown error")[:400]


# ---------------------------------------------------------------- transcribe
def decode_one(wav, out_json, model):
    """Runs in a fresh subprocess so each meeting gets clean GPU memory."""
    import mlx_whisper  # noqa: imported only here

    res = mlx_whisper.transcribe(
        wav,
        path_or_hf_repo=model,
        language="en",
        word_timestamps=True,
        temperature=(0.0, 0.2, 0.4),  # greedy at 0.0; fallback only on failure thresholds
        condition_on_previous_text=False,
        no_speech_threshold=0.6,
        compression_ratio_threshold=2.4,
        verbose=None,
    )
    with open(out_json, "w") as f:
        json.dump(res, f)


def mock_decode(wav, out_json, model):
    secs = wav_seconds(wav) if wav and os.path.exists(wav) else 60.0
    segs = [{"start": 0.0, "end": min(secs, 5.0), "text": " simulated segment",
             "words": [{"word": " simulated", "start": 0.0, "end": 1.0, "probability": 0.9},
                       {"word": " segment", "start": 1.0, "end": 2.0, "probability": 0.9}]}]
    with open(out_json, "w") as f:
        json.dump({"text": " simulated segment", "segments": segs, "language": "en"}, f)


def build_payload(res, audio_seconds, model, versions):
    segments = []
    speech = 0.0
    for s in res.get("segments", []):
        start = max(0.0, float(s.get("start", 0.0)))
        end = max(start, float(s.get("end", start)))
        text = (s.get("text") or "").strip()
        if not text:
            continue
        seg = {"start": round(start, 3), "end": round(end, 3), "text": text}
        words = []
        for w in s.get("words") or []:
            ws = max(0.0, float(w.get("start", start)))
            we = max(ws, float(w.get("end", ws)))
            word = {"w": (w.get("word") or "").strip()[:100], "start": round(ws, 3), "end": round(we, 3)}
            p = w.get("probability")
            if isinstance(p, (int, float)) and 0 <= p <= 1:
                word["p"] = round(float(p), 4)
            if word["w"]:
                words.append(word)
        if words:
            seg["words"] = words
        for k in ("avg_logprob", "no_speech_prob", "compression_ratio"):
            v = s.get(k)
            if isinstance(v, (int, float)):
                seg[k] = round(float(v), 4)
        speech += end - start
        segments.append(seg)
    text = " ".join(s["text"] for s in segments)
    word_count = len(text.split())
    return {
        "text": text,
        "segments": segments,
        "asr_engine": "mlx-whisper large-v3-turbo (greedy, word_timestamps)",
        "audio_minutes": round(audio_seconds / 60.0, 2),
        "word_count": word_count,
        "qa": {"audio_seconds": round(audio_seconds, 3), "transcribed_seconds": round(min(speech, audio_seconds), 3)},
        "pipeline": {"asr_model": model, "aligner": "whisper word_timestamps", "diarizer": None, "vad": None,
                     "versions": versions},
    }


def tool_versions():
    v = {"agent": AGENT_VERSION, "python": sys.version.split()[0]}
    try:
        import importlib.metadata as md
        v["mlx_whisper"] = md.version("mlx-whisper")
    except Exception:
        pass
    return v


# ---------------------------------------------------------------- team loop
class Agent:
    def __init__(self, a, api):
        self.a = a
        self.api = api
        self.buf = os.path.join(a.workdir, "buffer")
        self.out = os.path.join(a.workdir, "out")
        os.makedirs(self.buf, exist_ok=True)
        os.makedirs(self.out, exist_ok=True)
        self.ready = queue.Queue()
        self.inflight = 0
        self.lock = threading.Lock()
        self.remaining = None
        self.youtube_lock = threading.Lock()
        self.held = {}  # meeting key -> row (leased by us, not yet uploaded)
        self.done = 0
        self.claims = 0
        self.state_path = os.path.join(a.workdir, "state.json")
        self.versions = tool_versions()
        self.lanes = {}  # lane -> {lane, municipality_id, meeting_id, state, since}
        self.cfg = {}  # desired config from the Worker (heartbeat response)
        self.reexec = False
        self.ocr_proc = None
        self.ocr_last_spawn = 0.0
        self.yt_pause_until = 0.0  # local mirror of the Worker's YouTube cooldown / ip_suspect
        self.bot_checks = []  # epoch seconds of bot checks (for bot_checks_1h)
        self.last_cookie_refresh = None
        self.public_ip = None
        self.public_ip_at = 0.0

    EVENT_STATE = {"claimed": "claimed", "download_start": "downloading", "download_done": "downloaded",
                   "transcribe_start": "transcribing", "transcribe_done": "uploading", "blocked": "blocked",
                   "upload_done": "idle", "idle": "idle"}

    def emit(self, lane, event, row=None, **extra):
        """Record lane state locally (for heartbeat lanes[]) and push the transition to the Worker."""
        st = self.EVENT_STATE[event]
        self.lanes[str(lane)] = {"lane": str(lane), "state": st, "since": utcnow(),
                                 "municipality_id": row["municipality_id"] if row and st != "idle" else None,
                                 "meeting_id": row["meeting_id"] if row and st != "idle" else None}
        if self.a.simulate:
            LOG("simulate: would POST /telemetry/event lane=%s event=%s %s" % (
                lane, event, self.key(row) if row else ""))
            return
        code, r = self.api.event(lane, event, row, **extra)
        if code != 200:
            LOG("event post failed:", lane, event, code, r)

    def lanes_payload(self):
        return [dict(v) for _, v in sorted(self.lanes.items())]

    def key(self, row):
        return row["municipality_id"] + "|" + row["meeting_id"]

    def exclude(self):
        ex = list(self.a.exclude_venues)
        if self.a.allow_youtube and time.time() >= self.yt_pause_until:
            ex = [v for v in ex if v != "youtube"]
        elif "youtube" not in ex:
            ex.append("youtube")
        return ex

    # ------------------------------------------------------------ YouTube resilience (v1.3.0)
    def yt_opts(self):
        return {"cookies_from_browser": self.a.cookies_from_browser, "player_client": self.a.yt_player_client}

    def note_yt_gate(self, gate):
        """Mirror the Worker's YouTube gate ({youtube_allowed, cooldown_until, ip_suspect_until})."""
        if not isinstance(gate, dict) or gate.get("youtube_allowed", True):
            return
        until = gate.get("ip_suspect_until") or gate.get("cooldown_until")
        try:
            ts = datetime.strptime(until[:19], "%Y-%m-%dT%H:%M:%S").replace(tzinfo=timezone.utc).timestamp()
        except Exception:
            ts = time.time() + 600
        if ts > self.yt_pause_until:
            self.yt_pause_until = ts
            LOG("YouTube paused for this host until %s (%s); non-YouTube work and OCR continue" % (until, gate.get("reason")))

    def refresh_cookies(self, url):
        """Re-export cookies from this Mac's own signed-in browser profile. Returns True on success."""
        ok = False
        if self.a.cookie_refresh_cmd:
            code, txt = run(["/bin/sh", "-c", self.a.cookie_refresh_cmd], timeout=300)
            ok = code == 0 and os.path.exists(self.a.cookies)
            LOG("cookie refresh (cmd):", "ok" if ok else "failed: " + _last_line(txt))
        elif self.a.cookies_from_browser:
            tmp = self.a.cookies + ".new"
            code, txt = run(["yt-dlp", "--cookies-from-browser", self.a.cookies_from_browser, "--cookies", tmp,
                             "--skip-download", "--no-playlist", "--quiet", "--no-warnings", url], timeout=300)
            if os.path.exists(tmp) and os.path.getsize(tmp) > 200:
                os.makedirs(os.path.dirname(self.a.cookies) or ".", exist_ok=True)
                os.replace(tmp, self.a.cookies)
                os.chmod(self.a.cookies, 0o600)
                ok = True
            else:
                _rm(tmp)
            LOG("cookie refresh (from %s):" % self.a.cookies_from_browser.split(":")[0], "ok" if ok else "failed: " + _last_line(txt))
        else:
            LOG("cookie refresh: no --cookies-from-browser or --cookie-refresh-cmd configured")
        if ok:
            self.last_cookie_refresh = utcnow()
        return ok

    def probe_public(self):
        """Non-YouTube reachability + public IP (Cloudflare trace). Returns (ok, ip)."""
        ok_any, ip = False, None
        for u in ("https://www.cloudflare.com/cdn-cgi/trace", "https://www.google.com/generate_204", "https://en.wikipedia.org/robots.txt"):
            try:
                with urllib.request.urlopen(urllib.request.Request(u, headers={"User-Agent": BROWSER_UA}), timeout=15) as r:
                    body = r.read(4096).decode("utf-8", "replace")
                    ok_any = ok_any or r.status < 400
                    if "cdn-cgi/trace" in u:
                        for line in body.splitlines():
                            if line.startswith("ip="):
                                ip = line[3:].strip()
            except Exception:
                pass
        if ip:
            self.public_ip, self.public_ip_at = ip, time.time()
        return ok_any, ip

    def yt_heartbeat_fields(self):
        now = time.time()
        self.bot_checks = [t for t in self.bot_checks if now - t < 3600]
        if now - self.public_ip_at > 3600:
            self.probe_public()
        return {"bot_checks_1h": len(self.bot_checks), "last_cookie_refresh": self.last_cookie_refresh, "public_ip": self.public_ip}

    def youtube_download(self, row):
        """1 at a time + 20-40 s jitter. Bot check -> refresh cookies, retry once -> report (cooldown), never fail the row."""
        with self.youtube_lock:
            wav, err = download_wav(row, self.buf, self.a.cookies, yt_opts=self.yt_opts())
            refreshed = False
            if err and is_bot_check(err):
                self.bot_checks.append(time.time())
                LOG("bot check on", self.key(row), "- refreshing cookies and retrying once")
                refreshed = self.refresh_cookies(row["fetch"]["url"])
                time.sleep(random.uniform(20, 40))
                wav, err = download_wav(row, self.buf, self.a.cookies, yt_opts=self.yt_opts())
                if err and is_bot_check(err):
                    self.bot_checks.append(time.time())
            time.sleep(random.uniform(20, 40))
        return wav, err, refreshed

    def prefetch_loop(self, idx):
        while not STOP.is_set() and not DRAIN.is_set():
            if self.a.max_items and self.claims >= self.a.max_items:
                return
            with self.lock:
                full = self.ready.qsize() + self.inflight >= self.a.buffer_ahead
                if not full:
                    self.inflight += 1
            if full:
                time.sleep(5)
                continue
            try:
                self._prefetch_one(idx)
            finally:
                with self.lock:
                    self.inflight -= 1

    def _prefetch_one(self, idx):
        home_lanes = self.a.home_lanes
        if self.a.home_lanes_from_config and isinstance(self.cfg.get("home_lanes"), int):
            home_lanes = self.cfg["home_lanes"] if self.cfg.get("home_scope", "home") == "home" else 0
        scope = "home" if idx < max(0, home_lanes) else None
        code, r = self.api.claim(self.a.region, self.a.venues, self.exclude(), dry_run=self.a.simulate,
                                 scope=scope, lane="dl%d" % (idx + 1))
        if code != 200 or not isinstance(r, dict):
            LOG("claim failed:", code, r)
            time.sleep(60)
            return
        self.remaining = r.get("remaining_eligible")
        row = r.get("row")
        lane = "dl%d" % (idx + 1)
        if not row:
            self.emit(lane, "idle")
            LOG("no eligible rows for this host; sleeping 10 min")
            t_end = time.time() + TEN_MIN
            while time.time() < t_end and not STOP.is_set() and not DRAIN.is_set():
                STOP.wait(5)
            return
        self.claims += 1
        k = self.key(row)
        LOG("claimed%s" % (" (dry_run)" if self.a.simulate else ""), k, row["venue"], row.get("body_type"),
            "home_tier=%s" % row.get("home_tier"), "scope=%s%s" % (scope or "all", " (fallback)" if r.get("fallback") else ""),
            "lease_until=%s" % ((r.get("lease") or {}).get("expires_at")))
        if self.a.simulate:
            # Never download or upload in simulate mode; synthesize a 60 s silent WAV.
            wav = os.path.join(self.buf, k.replace("/", "__").replace("|", "__") + ".wav")
            code2, txt = run(["ffmpeg", "-nostdin", "-loglevel", "error", "-y", "-f", "lavfi", "-i",
                              "anullsrc=r=16000:cl=mono", "-t", "60", "-c:a", "pcm_s16le", wav])
            if code2 != 0:
                LOG("simulate: ffmpeg missing?", _last_line(txt))
                return
            self.emit(lane, "claimed", row)
            self.emit(lane, "download_done", row, audio_minutes=1.0)
            self.ready.put((row, wav))
            if self.a.max_items:
                return
            time.sleep(5)
            return
        self.held[k] = row
        self.emit(lane, "claimed", row)
        self.emit(lane, "download_start", row)
        gate = r.get("youtube")
        self.note_yt_gate(gate)
        if row["venue"] == "youtube":
            wav, err, refreshed = self.youtube_download(row)
            if err and is_bot_check(err):
                # Not the meeting's fault: release as retry (the Worker starts a host cooldown, no per-row block).
                LOG("bot check persisted after %s; releasing as retry, switching to non-YouTube/OCR" % (
                    "cookie refresh" if refreshed else "retry"), k)
                self.emit(lane, "blocked", row, detail=("[youtube] " + err)[:400])
                code_r, resp = self.api.release(row, "retry", err, cookie_refreshed=refreshed)
                bc = (resp or {}).get("botcheck") if isinstance(resp, dict) else None
                if bc:
                    self.note_yt_gate({"youtube_allowed": False, "reason": "cooldown", "cooldown_until": bc.get("cooldown_until"),
                                       "ip_suspect_until": None})
                else:
                    self.yt_pause_until = max(self.yt_pause_until, time.time() + 600)
                ok, ip = self.probe_public()
                if not ok:
                    LOG("non-YouTube probes failed too: possible IP block; reporting and stopping YouTube for 1 h")
                    self.api.botcheck("non-YouTube probes failed after a YouTube bot check: " + err[:300], kind="ip_block", row=row,
                                      cookie_refreshed=refreshed)
                    self.api.post_log("blocker", "Possible IP block: YouTube bot check AND non-YouTube probes (cloudflare trace, google 204, wikipedia) failed. YouTube stopped on this host for 1 h; non-YouTube/OCR continue if reachable.")
                    self.yt_pause_until = max(self.yt_pause_until, time.time() + 3600)
                self.held.pop(k, None)
                return
        else:
            wav, err = download_wav(row, self.buf)
        if err:
            LOG("download blocked:", k, err)
            self.emit(lane, "blocked", row, detail=err[:400])
            self.api.release(row, "blocked", err)
            self.held.pop(k, None)
            return
        LOG("ready:", k, "%.1f min" % (wav_seconds(wav) / 60))
        self.emit(lane, "download_done", row, audio_minutes=round(wav_seconds(wav) / 60.0, 2))
        self.ready.put((row, wav))

    def decode_loop(self, lane):
        while not STOP.is_set() and not DRAIN.is_set():
            try:
                row, wav = self.ready.get(timeout=5)
            except queue.Empty:
                if self.a.max_items and self.claims >= self.a.max_items and self.inflight == 0:
                    return
                continue
            k = self.key(row)
            try:
                self._decode_and_upload(lane, row, wav)
            except Exception as e:
                LOG("lane%d error on %s: %s" % (lane, k, e))
                self.emit(lane, "blocked", row, detail=("decode error: %s" % e)[:400])
                if not self.a.simulate:
                    self.api.release(row, "retry", "decode error: %s" % str(e)[:300])
            finally:
                _rm(wav)
                self.held.pop(k, None)

    def _decode_and_upload(self, lane, row, wav):
        k = self.key(row)
        secs = wav_seconds(wav)
        out_json = os.path.join(self.out, k.replace("/", "__").replace("|", "__") + ".mlx.json")
        if not self.a.simulate:
            self.api.telemetry(row, "transcribing")
        self.emit(lane, "transcribe_start", row, audio_minutes=round(secs / 60.0, 2))
        t0 = time.time()
        if self.a.simulate:
            mock_decode(wav, out_json, self.a.model)
        else:
            code, txt = run([sys.executable, os.path.abspath(__file__), "--decode-one", wav, out_json,
                             "--model", self.a.model], timeout=6 * 3600)
            if code != 0:
                raise RuntimeError("mlx decode failed: " + _last_line(txt))
        with open(out_json) as f:
            res = json.load(f)
        payload = build_payload(res, secs, self.a.model, self.versions)
        LOG("lane%d decoded %s: %.1f audio min in %.1f min, %d words, %d segments" % (
            lane, k, secs / 60, (time.time() - t0) / 60, payload["word_count"], len(payload["segments"])))
        self.emit(lane, "transcribe_done", row, audio_minutes=payload["audio_minutes"], word_count=payload["word_count"])
        if self.a.simulate:
            LOG("simulate: NOT uploading transcript for", k, "(payload keys: %s)" % ",".join(sorted(payload)))
            self.emit(lane, "upload_done", row, audio_minutes=payload["audio_minutes"], word_count=payload["word_count"])
            return
        code, r = self.api.put_transcript(row, payload)
        if code != 200:
            raise RuntimeError("transcript PUT %s: %s" % (code, str(r)[:300]))
        self.api.telemetry(row, "transcribed", audio_minutes=payload["audio_minutes"],
                           word_count=payload["word_count"], asr_engine=payload["asr_engine"])
        self.emit(lane, "upload_done", row, audio_minutes=payload["audio_minutes"], word_count=payload["word_count"])
        self.done += 1
        qa = payload["qa"]
        cov = round(100.0 * qa["transcribed_seconds"] / qa["audio_seconds"], 1) if qa["audio_seconds"] else 0
        self.api.post_log("note", "%s local mlx-whisper turbo. coverage_pct=%s words=%d audio_minutes=%s." % (
            row["meeting_id"], cov, payload["word_count"], payload["audio_minutes"]),
            row["municipality_id"], row["meeting_id"])
        _rm(out_json) if not self.a.keep_json else None

    def heartbeat_loop(self):
        st = self._load_state()
        last_id = st.get("log_last_id")
        while not STOP.is_set():
            depth = self.remaining if isinstance(self.remaining, int) else 0
            if self.a.simulate:
                code, resp = "skipped (simulate)", None
            else:
                code, resp = self.api.heartbeat(depth, self.lanes_payload(), self.yt_heartbeat_fields())
                if code == 200 and isinstance(resp, dict):
                    self.note_yt_gate(resp.get("youtube"))
            LOG("heartbeat", "ok" if code == 200 else code, "queue_depth=%s ready=%d inflight=%d done=%d" % (
                depth, self.ready.qsize(), self.inflight, self.done))
            if code == 200 and isinstance(resp, dict) and not DRAIN.is_set():
                try:
                    self.apply_config(resp)
                except Exception as e:  # never let remote config break the ASR lanes
                    LOG("config reconcile error:", str(e)[:300])
            last_id = self.poll_log(last_id)
            st["log_last_id"] = last_id
            self._save_state(st)
            STOP.wait(TEN_MIN)

    # ------------------------------------------------------------ remote config
    def apply_config(self, resp):
        cfg = resp.get("config") if isinstance(resp.get("config"), dict) else None
        if cfg is None:
            return
        if cfg != self.cfg:
            LOG("server config:", json.dumps({k: cfg.get(k) for k in ("asr_lanes", "home_lanes", "home_scope", "ocr_lanes", "fetch_lanes", "paused")}))
        self.cfg = cfg
        av = ((resp.get("agent_versions") or {}).get("asr_agent") or {})
        if not self.a.no_self_update and _vt(av.get("version")) > _vt(AGENT_VERSION):
            if self.self_update(av):
                self.reexec = True
                DRAIN.set()
                return
        if not self.a.no_ocr and (int(cfg.get("ocr_lanes") or 0) > 0 or int(cfg.get("fetch_lanes") or 0) > 0) and not cfg.get("paused"):
            self.ensure_ocr()

    def self_update(self, av):
        import hashlib
        import py_compile
        code, body = self.api._req("GET", "/join/agent.py", auth=False, tries=3)
        if code != 200 or not isinstance(body, str):
            LOG("self-update: download failed", code)
            return False
        raw = body.encode("utf-8")
        if hashlib.sha256(raw).hexdigest() != av.get("sha256"):
            LOG("self-update: sha256 mismatch for %s; keeping %s" % (av.get("version"), AGENT_VERSION))
            return False
        me = os.path.abspath(__file__)
        tmp = me + ".new"
        with open(tmp, "wb") as f:
            f.write(raw)
        try:
            py_compile.compile(tmp, doraise=True)
        except Exception as e:
            LOG("self-update: new agent does not compile (%s); keeping %s" % (str(e)[:200], AGENT_VERSION))
            _rm(tmp)
            return False
        shutil.copy2(me, me + ".prev")
        os.replace(tmp, me)
        LOG("self-update: %s -> %s (sha256 verified). Draining: current meeting(s) finish, then re-exec." % (AGENT_VERSION, av.get("version")))
        return True

    def ocr_running(self):
        """True if any ocr_agent.py (launchd, by hand, or our child) holds ~/meeting-docs-ocr/ocr_agent.lock."""
        import fcntl
        path = os.path.join(OCR_DIR, "ocr_agent.lock")
        if not os.path.exists(path):
            return False
        with open(path, "a+") as fh:
            try:
                fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB)
            except OSError:
                return True
            fcntl.flock(fh, fcntl.LOCK_UN)
        return False

    def ensure_ocr(self):
        if self.ocr_proc is not None and self.ocr_proc.poll() is None:
            return
        if self.ocr_running():
            return
        if time.time() - self.ocr_last_spawn < 900:  # back off 15 min after a failed start
            return
        self.ocr_last_spawn = time.time()
        os.makedirs(OCR_DIR, exist_ok=True)
        py = os.path.join(OCR_DIR, "venv", "bin", "python")
        if not os.path.exists(py):
            LOG("ocr: creating %s/venv (ocrmac, pymupdf, pillow)" % OCR_DIR)
            code, txt = run([sys.executable, "-m", "venv", os.path.join(OCR_DIR, "venv")], timeout=300)
            if code != 0:
                LOG("ocr: venv failed:", _last_line(txt))
                return
        code, txt = run([py, "-c", "import ocrmac, pymupdf, PIL"], timeout=120)
        if code != 0:
            code, txt = run([py, "-m", "pip", "install", "-q", "--upgrade", "pip", "ocrmac", "pymupdf", "pillow"], timeout=1800)
            if code != 0:
                LOG("ocr: pip install failed:", _last_line(txt))
                return
        if not self.fetch_ocr_agent(py):
            return
        out = open(os.path.join(OCR_DIR, "ocr_agent.out"), "a")
        self.ocr_proc = subprocess.Popen(
            [py, os.path.join(OCR_DIR, "ocr_agent.py"), "--host-label", self.api.host],
            stdout=out, stderr=subprocess.STDOUT, stdin=subprocess.DEVNULL, cwd=OCR_DIR,
            preexec_fn=lambda: os.nice(10), start_new_session=True)
        st = self._load_state()
        st["ocr_child_pid"] = self.ocr_proc.pid
        self._save_state(st)
        LOG("ocr: started ocr_agent.py pid=%d at nice 10 (lanes follow the server config: ocr=%s fetch=%s)" % (
            self.ocr_proc.pid, self.cfg.get("ocr_lanes"), self.cfg.get("fetch_lanes")))

    def fetch_ocr_agent(self, py):
        import hashlib
        dest = os.path.join(OCR_DIR, "ocr_agent.py")
        try:
            req = urllib.request.Request(DOCS_BASE + "/join/ocr_agent.json", headers={"User-Agent": "capreg-joiner/" + AGENT_VERSION})
            with urllib.request.urlopen(req, timeout=60) as r:
                m = json.loads(r.read())
            if os.path.exists(dest):
                with open(dest, "rb") as f:
                    if hashlib.sha256(f.read()).hexdigest() == m.get("sha256"):
                        return True
            req = urllib.request.Request(DOCS_BASE + "/join/ocr_agent.py", headers={"User-Agent": "capreg-joiner/" + AGENT_VERSION})
            with urllib.request.urlopen(req, timeout=120) as r:
                body = r.read()
        except Exception as e:
            LOG("ocr: could not fetch ocr_agent.py:", str(e)[:200])
            return os.path.exists(dest)
        if hashlib.sha256(body).hexdigest() != m.get("sha256"):
            LOG("ocr: ocr_agent.py sha256 mismatch; not installing")
            return os.path.exists(dest)
        with open(dest + ".new", "wb") as f:
            f.write(body)
        code, txt = run([py, "-m", "py_compile", dest + ".new"], timeout=60)
        if code != 0:
            LOG("ocr: ocr_agent.py does not compile:", _last_line(txt))
            return os.path.exists(dest)
        os.replace(dest + ".new", dest)
        LOG("ocr: installed ocr_agent.py %s (sha256 verified)" % m.get("version"))
        return True

    def stop_ocr_child(self):
        """On a normal stop, end the ocr_agent we started (not a launchd-managed one)."""
        pid = self._load_state().get("ocr_child_pid")
        if not pid:
            return
        try:
            with open(os.path.join(OCR_DIR, "ocr_agent.lock")) as f:
                holder = int((f.read() or "0").strip() or 0)
        except Exception:
            holder = 0
        if holder == pid:
            try:
                os.kill(pid, signal.SIGTERM)
                LOG("ocr: sent SIGTERM to ocr_agent pid", pid)
            except OSError:
                pass

    def poll_log(self, last_id):
        first = last_id is None
        since = last_id or 0
        while True:
            code, r = self.api.log_since(since)
            if code != 200 or not isinstance(r, dict):
                return last_id
            entries = r.get("entries") or []
            for e in entries:
                if first:
                    continue
                if e.get("author") != "progops":
                    continue
                msg = e.get("message", "")
                mine = self.api.host in msg or "all hosts" in msg.lower()
                flag = "  <== ACTION FOR THIS HOST" if mine and e.get("kind") in ("blocker", "question") else ""
                LOG("LOG #%s progops %s: %s%s" % (e.get("id"), e.get("kind"), msg[:600], flag))
            if not entries:
                break
            since = entries[-1]["id"]
            last_id = since
            if len(entries) < 100:
                break
        if first:
            LOG("log cursor initialised at #%s (older entries skipped; read /join for the current recipe)" % last_id)
        return last_id

    def _load_state(self):
        try:
            with open(self.state_path) as f:
                return json.load(f)
        except Exception:
            return {}

    def _save_state(self, st):
        with open(self.state_path, "w") as f:
            json.dump(st, f)

    def release_all(self, reason="host stopping"):
        for k, row in list(self.held.items()):
            code, _ = self.api.release(row, "retry", reason)
            LOG("released", k, code)
            self.held.pop(k, None)

    def run(self):
        threads = [threading.Thread(target=self.heartbeat_loop, daemon=True)]
        threads += [threading.Thread(target=self.prefetch_loop, args=(i,), daemon=True) for i in range(self.a.prefetch)]
        lanes = [threading.Thread(target=self.decode_loop, args=(i + 1,), daemon=True) for i in range(self.a.decode_lanes)]
        for i in range(self.a.decode_lanes):
            self.emit(i + 1, "idle")
        for t in threads + lanes:
            t.start()
        try:
            while any(t.is_alive() for t in lanes) and not STOP.is_set():
                time.sleep(2)
            if DRAIN.is_set():  # decode lanes finished their current meeting; let in-flight downloads land (max 30 min)
                t_end = time.time() + 1800
                while self.inflight > 0 and time.time() < t_end and not STOP.is_set():
                    time.sleep(5)
        finally:
            STOP.set()
            if not self.a.simulate:
                self.release_all("host restarting (self-update)" if self.reexec else "host stopping")
            if not self.reexec and not self.a.simulate:
                self.stop_ocr_child()
            for ln in list(self.lanes):
                self.emit(ln, "idle")
            for f in os.listdir(self.buf):
                if f.endswith(".part.wav") or ".src." in f:
                    _rm(os.path.join(self.buf, f))
        LOG("stopped. transcripts uploaded this run:", self.done)
        if self.reexec:
            LOG("re-exec:", " ".join([sys.executable, os.path.abspath(__file__)] + sys.argv[1:]))
            os.execv(sys.executable, [sys.executable, os.path.abspath(__file__)] + sys.argv[1:])


# ---------------------------------------------------------------- self-test
def self_test(a, api):
    ok = True
    LOG("self-test: host_label=%s base=%s" % (a.host_label, a.base))
    for tool in ("ffmpeg", "yt-dlp"):
        LOG(" ", tool, "OK" if shutil.which(tool) else "MISSING")
        ok = ok and bool(shutil.which(tool))
    code, r = api.claim(a.region, a.venues, ["youtube"] if not a.allow_youtube else [], dry_run=True,
                        scope="home" if a.home_lanes > 0 else None, lane="self-test")
    if code == 200 and isinstance(r, dict):
        row = r.get("row")
        LOG("  home lane: scope=%s home_remaining=%s fallback=%s home_tier=%s" % (
            r.get("scope"), r.get("home_remaining"), r.get("fallback"), (row or {}).get("home_tier")))
        LOG("  dry-run claim OK (no lease taken): %s" % (
            "%s|%s venue=%s body=%s eligible=%s" % (row["municipality_id"], row["meeting_id"], row["venue"],
                                                    row.get("body_type"), r.get("remaining_eligible")) if row else "no eligible rows"))
    else:
        LOG("  dry-run claim FAILED:", code, r if code != 401 else "unauthorized (check the token file)")
        return False
    if a.simulate:
        LOG("  simulate: skipping download + mlx decode")
        return ok
    if not row:
        return ok
    tdir = os.path.join(a.workdir, "selftest")
    os.makedirs(tdir, exist_ok=True)
    wav, err = download_wav(row, tdir, max_seconds=60)
    if err:
        LOG("  60 s download test failed (%s). Not fatal: that row may be blocked for this host." % err)
        return ok
    out_json = os.path.join(tdir, "selftest.mlx.json")
    t0 = time.time()
    code, txt = run([sys.executable, os.path.abspath(__file__), "--decode-one", wav, out_json, "--model", a.model], timeout=1800)
    if code != 0:
        LOG("  mlx-whisper decode FAILED:", _last_line(txt))
        _rm(wav)
        return False
    with open(out_json) as f:
        res = json.load(f)
    payload = build_payload(res, wav_seconds(wav), a.model, tool_versions())
    LOG("  mlx-whisper OK: %d words, %d segments from 60 s in %.1fs (NOT uploaded)" % (
        payload["word_count"], len(payload["segments"]), time.time() - t0))
    LOG("  first words:", payload["text"][:160])
    _rm(wav)
    _rm(out_json)
    return ok


def _vt(v):
    try:
        return tuple(int(x) for x in str(v).split("."))
    except Exception:
        return (0,)


def read_token(path):
    p = os.path.expanduser(path)
    if not os.path.exists(p):
        return None
    with open(p) as f:
        return f.read().strip() or None


def main():
    ap = argparse.ArgumentParser(description="capital-region-meeting-asr joiner agent")
    ap.add_argument("--base", default=os.environ.get("CAPREG_BASE", BASE_DEFAULT))
    ap.add_argument("--host-label", default=os.environ.get("CAPREG_HOST_LABEL", "m1-mbp"))
    ap.add_argument("--token-file", default="~/.secrets/capreg-telemetry-token")
    ap.add_argument("--workdir", default="~/capital-region-asr/joiner")
    ap.add_argument("--region", choices=["capital-region", "rochester", "mid-hudson"], default=None)
    ap.add_argument("--venues", default="", help="comma list to include, e.g. granicus,telvue,vimeo")
    ap.add_argument("--exclude-venues", default="youtube")
    ap.add_argument("--allow-youtube", action="store_true", help="only if the operator enables YouTube for this host")
    ap.add_argument("--cookies", default="~/capital-region-asr/youtube-cookies.txt",
                    help="Netscape cookies file exported from this Mac's own signed-in browser (refreshed on a bot check)")
    ap.add_argument("--cookies-from-browser", default=os.environ.get("CAPREG_COOKIES_FROM_BROWSER"),
                    help="e.g. chrome, safari, or 'chrome:Profile 2' (a dedicated profile signed in to the team Google account)")
    ap.add_argument("--cookie-refresh-cmd", default=os.environ.get("CAPREG_COOKIE_REFRESH_CMD"),
                    help="shell command that rewrites --cookies (e.g. headless Playwright export from a dedicated Chrome profile)")
    ap.add_argument("--yt-player-client", default=None,
                    help="optional yt-dlp youtube player_client override; default = yt-dlp's own recommended clients")
    ap.add_argument("--update-yt-dlp", action="store_true", help="run 'yt-dlp --update-to nightly' at start (pip/brew installs: upgrade there)")
    ap.add_argument("--decode-lanes", type=int, default=None, help="default: server config asr_lanes (2)")
    ap.add_argument("--home-lanes", type=int, default=None,
                    help="first N download slots claim with scope=home (Niskayuna/Schenectady/neighbours first; falls back when exhausted). 0 = off")
    ap.add_argument("--prefetch", type=int, default=3)
    ap.add_argument("--buffer-ahead", type=int, default=6)
    ap.add_argument("--model", default=MODEL_DEFAULT)
    ap.add_argument("--max-items", type=int, default=0, help="stop after N claims (0 = run forever)")
    ap.add_argument("--keep-json", action="store_true")
    ap.add_argument("--self-test", action="store_true")
    ap.add_argument("--simulate", action="store_true", help="dry-run claims, mock decode, never upload")
    ap.add_argument("--no-self-update", action="store_true", help="ignore newer agent.py advertised by the Worker")
    ap.add_argument("--no-ocr", action="store_true", help="never start the meeting-docs OCR lane from this agent")
    ap.add_argument("--decode-one", nargs=2, metavar=("WAV", "OUT_JSON"), help=argparse.SUPPRESS)
    a = ap.parse_args()

    if a.decode_one:
        decode_one(a.decode_one[0], a.decode_one[1], a.model)
        return 0

    a.workdir = os.path.expanduser(a.workdir)
    a.cookies = os.path.expanduser(a.cookies)
    a.venues = [v for v in a.venues.split(",") if v]
    a.exclude_venues = [v for v in a.exclude_venues.split(",") if v]
    a.home_lanes_from_config = a.home_lanes is None
    if a.home_lanes is None:
        a.home_lanes = 1
    os.makedirs(a.workdir, exist_ok=True)
    global LOG
    LOG = Log(os.path.join(a.workdir, "agent.log"))

    token = read_token(a.token_file)
    if not token:
        LOG("No telemetry token at %s. Copy ~/.secrets/capreg-telemetry-token from the existing team host "
            "(AirDrop or scp on the LAN), chmod 600 it, and re-run. See %s/join" % (a.token_file, a.base))
        return 2
    api = Api(a.base, token, a.host_label)
    if a.decode_lanes is None:
        code, r = api.host_config()
        cfg = (r or {}).get("config") if code == 200 and isinstance(r, dict) else None
        a.decode_lanes = int((cfg or {}).get("asr_lanes") or 2)
    a.decode_lanes = max(1, min(a.decode_lanes, 2)) if not a.simulate else a.decode_lanes

    def _stop(*_):
        LOG("stop requested; releasing leases and exiting")
        STOP.set()

    signal.signal(signal.SIGINT, _stop)
    signal.signal(signal.SIGTERM, _stop)

    if a.update_yt_dlp and shutil.which("yt-dlp"):
        code, txt = run(["yt-dlp", "--update-to", "nightly"], timeout=300)
        LOG("yt-dlp update:", _last_line(txt))
    if a.self_test:
        return 0 if self_test(a, api) else 1
    LOG("starting host=%s lanes=%d prefetch=%d buffer_ahead=%d exclude=%s simulate=%s" % (
        a.host_label, a.decode_lanes, a.prefetch, a.buffer_ahead, ",".join(Agent(a, api).exclude()), a.simulate))
    Agent(a, api).run()
    return 0


if __name__ == "__main__":
    sys.exit(main())