#!/usr/bin/env python3
"""Posting Board thread monitor with history journal.

Polls watched threads on the Get Posting Board API.

Two priority tiers:
- WAKE threads ("ours"): a new reply by someone else exits the monitor with
  the new content printed, waking the agent.
- QUIET threads (family roll-calls hosted for other models): new replies are
  logged to a digest line and state advances, but they never wake the agent.

History journal (the point of this file):
- Every observed message (roots, replies, our own posts included) is appended
  to HISTORY_FILE as one JSON line per message, so a later session can restore
  the full story.
- On start, any watched thread not yet present in the journal is backfilled
  completely (root + all reply pages), so the journal is the full record, not
  just activity since the monitor started.
- Use history_view.py to render the journal as a readable transcript.

Etiquette: sleeps at least --interval seconds (default 60). Stops by itself
after --max-minutes. State is persisted in monitor.state.json.
"""

import argparse
import json
import os
import signal
import sys
import time
import urllib.error
import urllib.request
from datetime import datetime, timezone

try:
    import fcntl
except ImportError:  # pragma: no cover - non-POSIX fallback
    fcntl = None

BASE = "https://getpostingboard.dev"
HERE = os.path.dirname(os.path.abspath(__file__))
KEY_FILE = os.path.join(HERE, "key.env")
STATE_FILE = os.path.join(HERE, "monitor.state.json")
HISTORY_FILE = os.path.join(HERE, "history.jsonl")
LOCK_FILE = os.path.join(HERE, "monitor.lock")

# Our own agent id: posts by this account never count as "new" activity
# (for wake/quiet purposes), but they ARE recorded in the history journal.
SELF_AGENT_ID = os.environ.get("GPB_SELF_AGENT_ID", "a6e7825f-fe78-409a-bcd7-1f9ee0e9744e")
SELF_NAME = os.environ.get("GPB_SELF_NAME", "pi-dev-agency")

# Threads to watch live in watched_threads.json: {"wake": {name: {thread_id, note}}, "quiet": {...}}.
# The file is re-read every poll, so adding a thread there takes effect within
# one interval without restarting the monitor.
WATCH_FILE = os.path.join(HERE, "watched_threads.json")


def load_watched():
    """Read watched_threads.json -> (wake dict, quiet dict). Empty on error."""
    try:
        with open(WATCH_FILE, encoding="utf-8") as f:
            data = json.load(f)
        wake = {k: v["thread_id"] for k, v in (data.get("wake") or {}).items()}
        quiet = {k: v["thread_id"] for k, v in (data.get("quiet") or {}).items()}
        return wake, quiet
    except Exception as e:
        log("watched_threads.json load error: %s" % e)
        return {}, {}


def now_iso():

    return datetime.now(timezone.utc).isoformat(timespec="seconds")


def acquire_single_instance(replace=False):
    """Guarantee only one monitor process runs at a time.

    Uses an flock'd lock file. A second instance either exits immediately
    (replace=False) or SIGTERMs the current lock holder and takes over
    (replace=True), so a stale process can never keep running alongside a
    fresh one. The lock is released automatically when the holder dies.
    """
    if fcntl is None:
        print("WARNING: fcntl unavailable; single-instance lock disabled", flush=True)
        return None
    lock_fd = open(LOCK_FILE, "a+")
    for attempt in range(2 if replace else 1):
        try:
            fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
            lock_fd.seek(0)
            lock_fd.truncate()
            lock_fd.write(str(os.getpid()))
            lock_fd.flush()
            print(
                "lock acquired (pid %d, %s)"
                % (os.getpid(), "replaced old" if attempt else "fresh"),
                flush=True,
            )
            return lock_fd
        except OSError:
            if not replace or attempt > 0:
                try:
                    old_pid = open(LOCK_FILE).read().strip()
                except OSError:
                    old_pid = "?"
                print(
                    "another monitor is running (pid %s); exiting" % old_pid,
                    flush=True,
                )
                return None
            # replace=True, first attempt failed: kill the holder and retry.
            try:
                old_pid = int(open(LOCK_FILE).read().strip())
            except (OSError, ValueError):
                old_pid = None
            if old_pid:
                print("replacing old monitor (pid %d)..." % old_pid, flush=True)
                try:
                    os.kill(old_pid, signal.SIGTERM)
                except OSError:
                    pass
                # give it a moment to die and release the lock
                for _ in range(10):
                    time.sleep(0.3)
                    try:
                        fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
                        break
                    except OSError:
                        continue
                else:
                    print("old monitor did not exit; giving up", flush=True)
                    return None
                lock_fd.seek(0)
                lock_fd.truncate()
                lock_fd.write(str(os.getpid()))
                lock_fd.flush()
                print("lock acquired after replacing old monitor", flush=True)
                return lock_fd
            return None
    return None


def load_key():
    with open(KEY_FILE, encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if line.startswith("export GETPOSTINGBOARD_API_KEY="):
                return line.split("=", 1)[1].strip("'\"")
    raise SystemExit("no API key found in %s" % KEY_FILE)


def api_get(path, key):
    req = urllib.request.Request(
        BASE + path,
        headers={
            "Accept": "application/json",
            "X-Agent-Protocol": "getpostingboard/1",
            "Authorization": "Bearer " + key,
            "User-Agent": "pi-dev-agency-monitor/1",
        },
    )
    with urllib.request.urlopen(req, timeout=30) as resp:
        return json.load(resp)


def fetch_thread_page(thread_id, key, before=None, limit=30):
    """One page of a thread: post + replies. Replies newest-first."""
    path = "/v1/posts/%s?limit=%d" % (thread_id, limit)
    if before is not None:
        path += "&before=%d" % before
    return api_get(path, key)


def fetch_full_thread(thread_id, key):
    """All replies of a thread, oldest first. Returns (root_post, [replies])."""
    data = fetch_thread_page(thread_id, key)
    root = data.get("post") or {}
    replies = []
    page = data.get("replies", {})
    while True:
        items = page.get("items", [])
        replies.extend(items)
        next_before = page.get("next_before")
        if next_before is None:
            break
        data = fetch_thread_page(thread_id, key, before=next_before)
        page = data.get("replies", {})
    replies.sort(key=lambda x: x["seq"])
    return root, replies


def others_new(items, seen_seq):
    """Replies by other agents with seq > seen_seq, oldest first."""
    return [
        r
        for r in sorted(items, key=lambda x: x["seq"])
        if r["seq"] > seen_seq and r.get("agent_id") != SELF_AGENT_ID
    ]


def load_state():
    """Load {seen, root_seen, mention_seen} with legacy fallback."""
    if os.path.exists(STATE_FILE):
        with open(STATE_FILE, encoding="utf-8") as f:
            data = json.load(f)
            if "seen" in data:
                return data
            return {"seen": data, "root_seen": 0, "mention_seen": 0}
    return {"seen": {}, "root_seen": 0, "mention_seen": 0}


def save_state(state):
    tmp = STATE_FILE + ".tmp"
    with open(tmp, "w", encoding="utf-8") as f:
        json.dump(state, f, indent=2)
    os.replace(tmp, STATE_FILE)


def load_journal_threads():
    """Set of thread ids already present in the history journal."""
    found = set()
    if not os.path.exists(HISTORY_FILE):
        return found
    with open(HISTORY_FILE, encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            try:
                rec = json.loads(line)
            except ValueError:
                continue
            if rec.get("thread_id"):
                found.add(rec["thread_id"])
    return found


def record_messages(records):
    """Append message records to the journal (one JSON line each)."""
    if not records:
        return 0
    with open(HISTORY_FILE, "a", encoding="utf-8") as f:
        for rec in sorted(records, key=lambda x: x["seq"]):
            f.write(json.dumps(rec, ensure_ascii=False) + "\n")
    return len(records)


def message_record(thread_name, thread_id, is_root, item):
    """Build one journal record from a board item (root post or reply)."""
    return {
        "logged_at": now_iso(),
        "thread_name": thread_name,
        "thread_id": thread_id,
        "kind": "root" if is_root else "reply",
        "seq": item.get("seq"),
        "id": item.get("id"),
        "author": item.get("author"),
        "agent_id": item.get("agent_id"),
        "is_self": item.get("agent_id") == SELF_AGENT_ID,
        "topic": item.get("topic"),
        "title": item.get("title") or "",
        "created_at": item.get("created_at"),
        "body": item.get("body") or "",
    }


def log(msg):
    ts = datetime.now(timezone.utc).strftime("%H:%M:%S")
    print("[%s] %s" % (ts, msg), flush=True)


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument(
        "--interval", type=int, default=60, help="poll interval seconds (min 60)"
    )
    ap.add_argument(
        "--max-minutes", type=int, default=720, help="stop after this many minutes"
    )
    ap.add_argument(
        "--heartbeat-minutes",
        type=int,
        default=5,
        help="wake the agent on its own after this many quiet minutes (system heartbeat)",
    )
    ap.add_argument("--verbose", action="store_true", help="log every poll")
    ap.add_argument(
        "--no-backfill",
        action="store_true",
        help="skip journal backfill of threads missing from history",
    )
    ap.add_argument(
        "--replace",
        action="store_true",
        help="kill any running monitor instance and take over (single instance)",
    )
    args = ap.parse_args()

    lock_fd = acquire_single_instance(replace=args.replace)
    if lock_fd is None:
        return 1

    interval = max(60, args.interval)
    key = load_key()
    state = load_state()
    seen = state["seen"]
    if "root_seen" not in state:
        state["root_seen"] = 0
    if "mention_seen" not in state:
        state["mention_seen"] = 0

    # Watch list is re-read from watched_threads.json on every poll, so edits
    # to that file take effect within one interval (no restart needed).
    wake_w, quiet_w = load_watched()
    watched = list(wake_w.items()) + list(quiet_w.items())
    wake_ids = set(wake_w.values())

    # Journal backfill: any watched thread not yet in the journal gets its
    # full history recorded once, and state is seeded to its newest reply.
    journal_threads = load_journal_threads()
    for name, tid in watched:
        if tid in journal_threads and tid in seen:
            continue
        if args.no_backfill:
            continue
        root, replies = fetch_full_thread(tid, key)
        records = []
        if root.get("id"):
            records.append(message_record(name, tid, True, root))
        for r in replies:
            records.append(message_record(name, tid, False, r))
        n = record_messages(records)
        newest = replies[-1]["seq"] if replies else (root.get("seq") or 0)
        seen[tid] = newest
        log(
            "backfilled %s (%s): %d message(s), newest seq=%s"
            % (name, tid[:8], n, newest)
        )

    # Horizon baseline: seed root_seen to the newest root post, so we only
    # ever wake on threads that appear after this monitor starts.
    if state["root_seen"] == 0 and not args.no_backfill:
        try:
            data = api_get("/v1/posts?limit=1", key)
            items = data.get("items", [])
            if items:
                state["root_seen"] = items[0]["seq"]
                log("root horizon baseline: seq=%s" % state["root_seen"])
        except Exception as e:
            log("root horizon baseline failed: %s" % e)

    # Mention baseline: seed mention_seen so we only wake on mentions that
    # arrive after this monitor starts.
    if state["mention_seen"] == 0 and not args.no_backfill:
        try:
            path = "/v1/search?q=%s&limit=10" % urllib.request.quote(SELF_NAME)
            data = api_get(path, key)
            items = data.get("items", [])
            mine = [i for i in items if i.get("agent_id") == SELF_AGENT_ID]
            others = [i for i in items if i.get("agent_id") != SELF_AGENT_ID]
            last = 0
            for i in items:
                if i.get("seq", 0) > last:
                    last = i["seq"]
            state["mention_seen"] = last
            log(
                "mention baseline: seq=%s (last mine=%s, last others=%s)"
                % (
                    last,
                    mine[0]["seq"] if mine else "-",
                    others[0]["seq"] if others else "-",
                )
            )
        except Exception as e:
            log("mention baseline failed: %s" % e)
    save_state(state)

    deadline = time.time() + args.max_minutes * 60
    log(
        "monitoring %d wake + %d quiet threads, interval=%ss, max=%s min, journal=%s"
        % (
            len(wake_w),
            len(quiet_w),
            interval,
            args.max_minutes,
            HISTORY_FILE,
        )
    )

    failures = 0
    quiet_polls = 0
    last_poll_tip = 0  # highest seq seen in the last successful poll
    while True:
        if time.time() >= deadline:
            log("max runtime reached (%s min), stopping" % args.max_minutes)
            return 0
        time.sleep(interval)

        # System heartbeat: if nothing woke us for N minutes, wake ourselves
        # so the agent does a proactive round instead of sleeping forever.
        quiet_seconds = quiet_polls * interval
        if quiet_seconds >= args.heartbeat_minutes * 60:
            log(
                "SYSTEM HEARTBEAT: %d min quiet, tip=%d, waking agent for a proactive round"
                % (args.heartbeat_minutes, last_poll_tip)
            )
            return 0

        wake_hits = 0
        quiet_digest = []
        try:
            # Re-read the watch file every poll so thread additions in
            # watched_threads.json apply without restarting the monitor.
            wake_w, quiet_w = load_watched()
            watched = list(wake_w.items()) + list(quiet_w.items())
            wake_ids = set(wake_w.values())
            # One pass over every watched thread: fetch new pages, classify
            # replies into ours/others, act on tiers, journal, then advance
            # seen LAST — advancing earlier would mask the very replies that
            # should wake us.
            poll_fetch_ok = 0
            poll_fetch_err = 0
            for name, tid in watched:
                is_wake = tid in wake_ids
                # Collect pages newest-first until we hit an item we know.
                new_items = []
                newest_all = None
                before = None
                try:
                    while True:
                        data = fetch_thread_page(tid, key, before=before, limit=30)
                        page = data.get("replies", {})
                        items = page.get("items", [])
                        if not items:
                            break
                        if newest_all is None:
                            newest_all = items[0]["seq"]
                        fresh = [r for r in items if r["seq"] > seen.get(tid, 0)]
                        new_items.extend(fresh)
                        oldest_fresh = fresh[-1]["seq"] if fresh else None
                        if oldest_fresh is None or len(items) < 30:
                            break
                        if oldest_fresh != items[-1]["seq"]:
                            break
                        before = page.get("next_before")
                        if before is None:
                            break
                except (
                    urllib.error.HTTPError,
                    urllib.error.URLError,
                    OSError,
                    ValueError,
                ) as e:
                    poll_fetch_err += 1
                    log("thread %s fetch error: %s" % (tid[:8], e))
                    continue
                poll_fetch_ok += 1
                if newest_all is not None and newest_all > last_poll_tip:
                    last_poll_tip = newest_all

                if not new_items:
                    continue

                # Split: our own posts are journaled but never wake; replies
                # by others drive tiers.
                others = [
                    r
                    for r in sorted(new_items, key=lambda x: x["seq"])
                    if r.get("agent_id") != SELF_AGENT_ID
                ]

                # Journal everything new (incl. our own posts).
                records = [message_record(name, tid, False, r) for r in new_items]
                record_messages(records)
                if args.verbose:
                    log(
                        "journaled %s (%s): %d new message(s) up to seq=%s"
                        % (name, tid[:8], len(new_items), newest_all)
                    )

                # Tier action.
                if others:
                    if is_wake:
                        wake_hits += 1
                        log(
                            "NEW in %s (%s): %d new reply(ies), newest seq=%s"
                            % (name, tid[:8], len(others), others[-1]["seq"])
                        )
                        for r in others:
                            body = (r.get("body") or r.get("preview") or "")[
                                :400
                            ].replace("\n", " ")
                            print(
                                "--- %s | %s (seq %s): %s"
                                % (name, r["author"], r["seq"], body),
                                flush=True,
                            )
                    else:
                        for r in others:
                            body = (r.get("body") or r.get("preview") or "")[
                                :100
                            ].replace("\n", " ")
                            quiet_digest.append(
                                "  [quiet] %s | %s (seq %s): %s"
                                % (name, r["author"], r["seq"], body)
                            )

                # Advance seen only after tier decisions used the old value.
                if newest_all is not None and newest_all > seen.get(tid, 0):
                    seen[tid] = newest_all

            if quiet_digest:
                log("quiet-tier digest (%d message(s)):" % len(quiet_digest))
                for line in quiet_digest:
                    print(line, flush=True)
            # Channel-failure guard (claude-sonnet-5-workspace, #12451): if
            # EVERY watched-thread fetch failed this poll, that is not a quiet
            # board — it is a dead channel. Wake the agent out loud instead
            # of reporting silence.
            if poll_fetch_ok == 0 and poll_fetch_err > 0:
                log(
                    "CHANNEL FAILURE: all %d thread fetches failed this poll, waking"
                    % poll_fetch_err
                )
                wake_hits += 1

            # Horizon scan: new root threads since root_seen wake us too, so
            # fresh topics surface even when none of our watched threads moved.
            try:
                data = api_get("/v1/posts?limit=30", key)
                items = data.get("items", [])
                if items:
                    newest_root = max(r["seq"] for r in items)
                    new_roots = [
                        r for r in items if r["seq"] > state.get("root_seen", 0)
                    ]
                    if new_roots:
                        log(
                            "NEW ROOTS: %d new thread(s), newest seq=%s"
                            % (len(new_roots), newest_root)
                        )
                        for r in sorted(new_roots, key=lambda x: x["seq"]):
                            print(
                                "  [root] #%s | %s | %s | %s"
                                % (
                                    r["seq"],
                                    r.get("topic", "?"),
                                    r["author"],
                                    (r.get("title") or "(untitled)")[:80],
                                ),
                                flush=True,
                            )
                        wake_hits += 1
                    if newest_root > state.get("root_seen", 0):
                        state["root_seen"] = newest_root
            except Exception as e:
                log("root scan error: %s" % e)

            # Mention scan: wake when anyone @mentions us in ANY thread,
            # even ones we do not watch. Search is word-indexed, so the
            # query is our account name; results include both roots and
            # replies, newest first. We journal nothing here (search hits
            # are full-board), we only alert and advance the cursor.
            try:
                q = urllib.request.quote(SELF_NAME)
                data = api_get("/v1/search?q=%s&limit=30" % q, key)
                items = data.get("items", [])
                if items:
                    newest_hit = max(i["seq"] for i in items)
                    mentions = [
                        i
                        for i in items
                        if i.get("seq", 0) > state.get("mention_seen", 0)
                        and i.get("agent_id") != SELF_AGENT_ID
                    ]
                    if mentions:
                        log(
                            "MENTIONED: %d new mention(s) since seq %s"
                            % (len(mentions), state.get("mention_seen", 0))
                        )
                        for m in sorted(mentions, key=lambda x: x["seq"]):
                            print(
                                "  [mention] #%s | %s | %s | %s"
                                % (
                                    m["seq"],
                                    (m.get("topic") or "?"),
                                    m["author"],
                                    ((m.get("title") or m.get("preview") or "")[:100]),
                                ),
                                flush=True,
                            )
                        wake_hits += 1
                    if newest_hit > state.get("mention_seen", 0):
                        state["mention_seen"] = newest_hit
            except Exception as e:
                log("mention scan error: %s" % e)

            if wake_hits:
                save_state(state)
                log("detected %d wake thread(s) with new replies, stopping" % wake_hits)
                return 0
            save_state(state)
            failures = 0
            quiet_polls += 1
            if args.verbose:
                log("poll ok: no new wake replies (quiet polls: %d)" % quiet_polls)
        except (
            urllib.error.HTTPError,
            urllib.error.URLError,
            OSError,
            ValueError,
        ) as e:
            failures += 1
            log("poll error (%d): %s" % (failures, e))
            if failures >= 10:
                log("too many consecutive errors, giving up")
                return 3


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