box/bin/response-harvester.py

def is_title_noise(title):
    t = (title or "").lower().strip()
    return bool(re.search(r"generate.*(chat|session)?.*title", t))

#!/usr/bin/env python3
"""
response-harvester.py — Fleet agent readback and response harvesting daemon.

Monitors Chromebox agents (muse, pip, 646, opm), harvests incoming messages from
Main Chat and registered sidechats, maintains persistent watermarks, appends to
chat-history.jsonl, resolves pending follow-ups (verb-aware: [ACK|CLAIM] ->
acknowledged, [RESULT|DECLINE|NO-ACTION] -> resolved, outcome recorded), and
records [RESULT] completions in job-log.jsonl.

Features:
- Direct CDP over host veth interfaces (fast, no sudo needed).
- cdp_queue integration with PRIORITY_LOW (never blocks operator/DMs).
- URL state preservation (restores browser to initial thread/main via Ctrl+J).
- Bounded scroll-back for virtualized DOM (#hatch-chat-scroll).
- Per-node fault isolation (CDP errors on one node do not abort the cycle).
- Dual-mode execution (--once for systemd timers/CLI, --loop for daemon).

Usage:
    python3 response-harvester.py --once
    python3 response-harvester.py --loop --interval 30
    python3 response-harvester.py --agent 646 --once
"""

import argparse
import concurrent.futures
import hashlib
import json
import os
import re
import ssl
import subprocess
import sys
import time
import urllib.error
import urllib.request
import websocket
from datetime import datetime, timezone
from pathlib import Path

# Paths
NETVM_ROOT = Path("/home/super/Projects/NetVM")
BIN_DIR = NETVM_ROOT / "bin"
LOGS_DIR = NETVM_ROOT / "logs"
CHAT_HISTORY_LOG = LOGS_DIR / "chat-history.jsonl"
WATERMARKS_FILE = NETVM_ROOT / "siphon-watermarks.json"
JOB_LOG = NETVM_ROOT / "job-log.jsonl"
FOLLOWUPS_FILE = NETVM_ROOT / "followups.json"
JOB_SIDECHATS_FILE = NETVM_ROOT / "job-sidechats.json"
WAKE_SIDECHATS_FILE = Path("/home/super/sidechat-wake/wake-sidechats.json")
SWARM_FILE = NETVM_ROOT / "swarms.json"
NUDGE_TRACKER_FILE = NETVM_ROOT / "conversation-nudge-tracker.json"
JOBS_DIR = NETVM_ROOT / "jobs"
DISPATCH_PY = BIN_DIR / "job-dispatch.py"

# Ensure bin is in sys.path
sys.path.insert(0, str(BIN_DIR))
try:
    from cdp_queue import cdp_slot, PRIORITY_LOW
    HAS_CDP_QUEUE = True
except ImportError:
    HAS_CDP_QUEUE = False

try:
    import netvm_registry
    HAS_REGISTRY = True
except ImportError:
    HAS_REGISTRY = False

try:
    import pipeline_engine
    HAS_PIPELINE = True
except ImportError:
    HAS_PIPELINE = False

try:
    import muse_hybrid
    HAS_MUSE_HYBRID = True
except ImportError:
    HAS_MUSE_HYBRID = False

VALID_AGENTS = ["muse", "pip", "646", "opm", "dev", "def"]
DEFAULT_PORTS = {"muse": 9410, "pip": 9420, "646": 9430, "opm": 9440, "def": 9450, "dev": 9460}

# Matches EVERY [RESULT <job_id>] marker in a message (use with finditer, not
# search). The result text is lazy and stops before the next marker (or end of
# text), so a message closing two jobs records each with its own text instead
# of the first marker greedily swallowing the second.
RESULT_RE = re.compile(r"\[RESULT\s+([A-Za-z0-9_-]+)\]\s*(.*?)(?=\[RESULT\s|\Z)", re.S)

def iter_result_markers(text):
    """Yield (job_id, result_text) for every [RESULT <job_id>] marker in text."""
    for m in RESULT_RE.finditer(text or ""):
        yield m.group(1).strip(), m.group(2).strip()


# Verb markers for the digest response protocol:
# [ACK|CLAIM|RESULT|DECLINE|NO-ACTION <job_id>].
# ACK/CLAIM acknowledge a digest (nudge-suppressed, NOT closed);
# RESULT/DECLINE/NO-ACTION close the digest. Every verb match records
# outcome=<verb> on the followup record.
VERB_RE = re.compile(r"\[(ACK|CLAIM|RESULT|DECLINE|NO-ACTION)\s+([A-Za-z0-9_-]+)\]")


def iter_verb_markers(text):
    """Yield (verb, job_id) for every [VERB <job_id>] marker in text."""
    for m in VERB_RE.finditer(text or ""):
        yield m.group(1), m.group(2).strip()


REFUSAL_PATTERNS = [
    r"\bnot doing that\b",
    r"\brefuse to\b",
    r"\bdecline to\b",
    r"\bdeclining\b",
    r"\bwill not perform\b",
    r"\bnot emitting the \[?result\]?\b",
    r"\bnot sending result\b",
    r"\bproduction change with recursion risk\b",
]

def detect_explicit_refusal(text):
    """Detect plain-language refusal/decline from an assistant in a task thread."""
    t = (text or "").lower()
    return any(re.search(pat, t) for pat in REFUSAL_PATTERNS)


# Muse-native tool names (docs.muse-dev.online/capabilities.html) -> box exec ops.
NATIVE_ALIASES = {
    "cron.create": "followup.create",
    "cron.runonce": "followup.create",
    "cron.schedule": "followup.create",
    "subagent.spawn": "swarm.spawn",
    "subagents.spawn": "swarm.spawn",
    "subagent.list": "swarm.list",
    "subagent.status": "swarm.status",
}


def normalize_native_call(op, args):
    """Bridge native Muse tool names/args to the box ops the exec daemon allows."""
    op = NATIVE_ALIASES.get(op, op)
    if not isinstance(args, dict):
        return op, args
    args = dict(args)
    if op == "followup.create":
        for k in ("kind", "mode", "schedule"):  # native cron fields the box op does not take
            args.pop(k, None)
        if "in_m" not in args:
            for k in ("minutes", "in_minutes", "delay_m", "runonce_m"):
                if k in args:
                    args["in_m"] = args.pop(k)
                    break
        if "prompt" not in args:
            for k in ("body", "task", "message", "text"):
                if k in args:
                    args["prompt"] = args.pop(k)
                    break
    elif op == "swarm.spawn":
        if "task" not in args:
            for k in ("prompt", "body", "instructions"):
                if k in args:
                    args["task"] = args.pop(k)
                    break
        if "count" not in args and "n" in args:
            args["count"] = args.pop("n")
    return op, args


def parse_tool_calls(text):
    """
    Extract structured tool/exec calls from assistant messages.
    Supports:
      1. [TOOL <op> <json_args>] or [EXEC <op> <json_args>]
      2. ```box / ```tool / ```exec JSON blocks
    """
    calls = []
    for m in re.finditer(r"\[(?:TOOL|EXEC)\s+([a-zA-Z0-9_.-]+)(?:\s+(.*?))?\]", text or ""):
        op = m.group(1).strip()
        raw_args = (m.group(2) or "").strip()
        args = {}
        if raw_args:
            try:
                args = json.loads(raw_args)
            except Exception:
                args = {"raw": raw_args}
        calls.append((op, args))

    for m in re.finditer(r"```(?:box|tool|exec)\s*\n(.*?)```", text or "", re.DOTALL):
        block = m.group(1).strip()
        try:
            d = json.loads(block)
            if isinstance(d, dict) and "op" in d:
                calls.append((d["op"], d.get("args", {})))
        except Exception:
            pass

    # Extract curl command tool calls (e.g. curl ... -d '{"op": "...", "args": ...}')
    for m in re.finditer(r"curl\s+.*?https?://(?:box\.muse-dev\.online|100\.123\.153\.75:8444|exec\.muse-dev\.online)/exec.*?-[dD]\s+(['\"])(.*?)\1", text or "", re.DOTALL):
        raw_body = m.group(2)
        try:
            d = json.loads(raw_body)
            if isinstance(d, dict) and "op" in d:
                calls.append((d["op"], d.get("args", {})))
        except Exception:
            pass

    # De-duplicate identical calls within one message (envelope repeats the spawn
    # call at top and bottom; an agent echoing both must not run it twice).
    seen, unique = set(), []
    for op, args in [normalize_native_call(o, a) for o, a in calls]:
        key = (op, json.dumps(args, sort_keys=True, default=str))
        if key in seen:
            continue
        seen.add(key)
        unique.append((op, args))
    return unique


def execute_agent_tool(agent, op, args, timeout=30):
    """
    Execute tool call via local exec-constrained HTTPS daemon over Tailscale.
    Returns (ok, result_or_error_string).
    """
    token = None
    token_file = os.path.expanduser(f"~/.exec-tokens/{agent}")
    if not os.path.exists(token_file):
        token_file = os.path.expanduser("~/.exec-server-token")
    if os.path.exists(token_file):
        try:
            with open(token_file) as f:
                token = f.read().strip()
        except Exception:
            pass

    url = "https://100.123.153.75:8444/exec"
    payload = json.dumps({"op": op, "args": args}).encode("utf-8")
    headers = {
        "Content-Type": "application/json",
        "User-Agent": f"Box-Harvester/{agent}",
    }
    if token:
        headers["Authorization"] = f"Bearer {token}"

    ctx = ssl.create_default_context()
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE

    req = urllib.request.Request(url, data=payload, headers=headers, method="POST")
    try:
        with urllib.request.urlopen(req, context=ctx, timeout=timeout) as resp:
            data = resp.read().decode("utf-8")
            res = json.loads(data)
            if res.get("rc") == 0:
                return True, res.get("stdout", "").strip() or "OK"
            elif "stdout" in res:
                return False, (res.get("stdout") or res.get("stderr") or "failed").strip()
            elif "error" in res:
                return False, res["error"]
            return True, json.dumps(res)
    except urllib.error.HTTPError as he:
        try:
            err_body = he.read().decode("utf-8")
            return False, f"HTTP {he.code}: {err_body}"
        except Exception:
            return False, f"HTTP {he.code}"
    except Exception as e:
        return False, str(e)


def format_tool_result_for_chat(op, raw_output):
    """
    Format tool output into a clean, human-readable chat message
    instead of spewing a raw wall of unformatted JSON.
    """
    if not raw_output:
        return "OK"

    try:
        data = json.loads(raw_output)
    except Exception:
        s = str(raw_output).strip()
        return s[:500] if len(s) > 500 else s

    if op == "health.check" and isinstance(data, dict) and "fleet" in data:
        nodes = data.get("fleet", [])
        all_ok = all(n.get("cdp_ok") and n.get("proc_alive") for n in nodes)
        up_count = sum(1 for n in nodes if n.get("cdp_ok") and n.get("proc_alive"))
        lines = [f"Fleet Health: {'ALL GREEN' if all_ok else 'DEGRADED'} ({up_count}/{len(nodes)} nodes online)"]
        for n in nodes:
            st = "OK" if n.get("cdp_ok") and n.get("proc_alive") else "FAIL"
            lat = n.get("latency_ms", 0)
            lines.append(f"  • {n.get('node')}: {st} ({lat}ms)")
        return "\n".join(lines)

    if op == "cron.runs" and isinstance(data, dict) and "jobs" in data:
        jobs = data.get("jobs", [])
        return f"{len(jobs)} scheduled jobs configured (e.g. {', '.join(j.get('name') for j in jobs[:6])})"

    if op == "cron.status" and isinstance(data, dict):
        return f"Cron '{data.get('name')}': {'ACTIVE' if data.get('active') else 'INACTIVE'} (last result: {data.get('last_result', 'unknown')})"

    if op == "vars.list" and isinstance(data, dict) and "variables" in data:
        vars_dict = data.get("variables", {})
        sample = ", ".join(f"{k}={v}" for k, v in list(vars_dict.items())[:5])
        return f"{len(vars_dict)} variables: {sample}..."

    if op == "vars.get" and isinstance(data, dict):
        return f"{data.get('name')} = {data.get('value')}"

    if op == "files.read" and isinstance(data, dict):
        if not data.get("ok"):
            return f"Failed to read file: {data.get('error')}"
        path = data.get("path")
        lines = data.get("lines")
        trunc = " (first " + str(data.get("displayed_lines")) + " lines)" if data.get("truncated") else ""
        content = data.get("content", "").strip()
        return f"File `{path}` ({lines} lines total){trunc}:\n```\n{content}\n```"

    if op == "files.write" and isinstance(data, dict):
        if not data.get("ok"):
            return f"Failed to write file: {data.get('error')}"
        return f"File `{data.get('path')}` written successfully ({data.get('bytes_written')} bytes, {data.get('lines')} lines)."

    if op == "web.fetch" and isinstance(data, dict):
        if not data.get("ok"):
            return f"Failed to fetch {data.get('url')}: {data.get('error')}"
        status = data.get("status")
        url = data.get("url")
        trunc = " (truncated to 4KB)" if data.get("truncated") else ""
        return f"Web Fetch `{url}` (HTTP {status}){trunc}:\n```\n{data.get('body', '').strip()[:800]}\n```"

    if op == "service.status" and isinstance(data, dict):
        if not data.get("ok"):
            return f"Service check failed: {data.get('error')}"
        st = "ACTIVE" if data.get("active") else "INACTIVE"
        return f"Service `{data.get('service')}` is {st}.\nStatus: {data.get('status_line')}"

    if op in ("followup.create", "followup.schedule") and isinstance(data, dict):
        if not data.get("ok"):
            return f"Follow-up scheduling failed: {data.get('error')}"
        sec = data.get("in_seconds", 60)
        ag = data.get("agent", "")
        return f"Follow-up timer scheduled for {ag} in {sec}s ({round(sec/60, 1)}m)."

    if op == "swarm.spawn" and isinstance(data, dict):
        if not data.get("ok"):
            return f"Swarm spawn failed: {data.get('error')}"
        return f"Swarm `{data.get('swarm_id')}` spawned with {data.get('count')} worker slots (status: {data.get('status')})."

    if op == "swarm.status" and isinstance(data, dict):
        if not data.get("ok"):
            return f"Swarm status check failed: {data.get('error')}"
        sw = data.get("swarm", {})
        return f"Swarm `{sw.get('swarm_id')}`: {sw.get('status')} ({sw.get('done', 0)}/{sw.get('count', 0)} slots completed)."

    # General fallback: compact JSON capped to 400 chars
    s = json.dumps(data)
    return s[:400] + "..." if len(s) > 400 else s


def utcnow():
    return datetime.now(timezone.utc).isoformat()


def get_node_network(node):
    """Derive veth peer IP and CDP port from node identity."""
    tag = hashlib.sha256(node.encode()).hexdigest()[:8]
    idx = int(tag[:3], 16) % 200 + 10
    peer_ip = f"10.201.{idx}.2"

    port = None
    if HAS_REGISTRY:
        try:
            port = netvm_registry.port_for(node)
        except Exception:
            pass
    if not port:
        port = DEFAULT_PORTS.get(node, 9410)

    return peer_ip, port


def load_json_file(path, default=None):
    if default is None:
        default = {}
    if not os.path.exists(path):
        return default
    try:
        with open(path, "r", encoding="utf-8") as f:
            return json.load(f)
    except Exception:
        return default


def save_json_file(path, data):
    tmp_path = f"{path}.tmp.{os.getpid()}"
    with open(tmp_path, "w", encoding="utf-8") as f:
        json.dump(data, f, indent=2)
    os.replace(tmp_path, path)


def append_jsonl(path, record):
    os.makedirs(os.path.dirname(os.path.abspath(path)), exist_ok=True)
    with open(path, "a", encoding="utf-8") as f:
        f.write(json.dumps(record) + "\n")


# Case-insensitive failure/decline detection for [RESULT] text.
# "declined"/"decline"/"reject" count as failures so declines do NOT
# trigger onward pipeline chaining (previously only uppercase
# FAILED/UNABLE/FAIL matched, so a lowercase "declined" was logged as success).
FAIL_PREFIXES = ("failed", "unable", "fail", "declined", "decline", "reject", "error")


def is_fail_result(result_text):
    t = (result_text or "").lstrip().lower()
    return t.startswith(FAIL_PREFIXES)


def get_monitored_threads(target_agent=None):
    """
    Build dict of threads to monitor per agent:
    { agent: [ {"id": "<uuid>", "name": "<alias>"} ] }
    Filters to permanent channels, threads with pending followups, or recent threads (< 3h).
    """
    agents = [target_agent] if target_agent else VALID_AGENTS
    threads_by_agent = {a: [] for a in agents}

    pending_threads = set()
    followups = load_json_file(FOLLOWUPS_FILE)
    for f in followups.values():
        if f.get("status") in ("pending", "acknowledged"):
            tu = f.get("thread_uuid")
            if tu:
                pending_threads.add(tu)

    PERM_KEYWORDS = ("coord", "tasks", "task", "brain", "heartbeat", "sync", "audit", "main-loop")
    now = datetime.now(timezone.utc)

    state_files = [JOB_SIDECHATS_FILE, WAKE_SIDECHATS_FILE]
    for sf in state_files:
        if not sf.exists():
            continue
        try:
            data = json.loads(sf.read_text(encoding="utf-8"))
            for key, val in data.items():
                if key.startswith("_"):
                    continue
                created_at = None
                if isinstance(val, dict):
                    uuid = val.get("thread_uuid") or val.get("uuid")
                    agent = val.get("agent", "opm")
                    created_at = val.get("created_at")
                elif isinstance(val, str):
                    uuid = val
                    agent = "opm"
                else:
                    continue

                if not uuid or agent not in threads_by_agent:
                    continue

                is_perm = any(k in key.lower() for k in PERM_KEYWORDS)
                is_pending = uuid in pending_threads
                is_recent = False
                if created_at:
                    try:
                        cat = datetime.fromisoformat(created_at.replace("Z", "+00:00"))
                        if (now - cat).total_seconds() < 10800:
                            is_recent = True
                    except Exception:
                        pass
                else:
                    if not (key.startswith("pipe-") or key.startswith("test-") or key.startswith("onboarding-")):
                        is_recent = True

                # If already archived, exclude from active monitoring unless pending followup
                if isinstance(val, dict) and val.get("archived") and not is_pending:
                    continue

                if not (is_perm or is_pending or is_recent):
                    continue

                existing = [t["id"] for t in threads_by_agent[agent]]
                if uuid not in existing:
                    threads_by_agent[agent].append({"id": uuid, "name": key})
        except Exception:
            continue

    return threads_by_agent


class CDPClient:
    """Lightweight direct CDP client over host veth."""

    def __init__(self, node, peer_ip, port, timeout=10):
        self.node = node
        self.peer_ip = peer_ip
        self.port = port
        self.timeout = timeout
        self.ws = None
        self.msg_id = 0

    def connect(self):
        url = f"http://{self.peer_ip}:{self.port}/json/list"
        req = urllib.request.Request(url)
        with urllib.request.urlopen(req, timeout=self.timeout) as resp:
            targets = json.load(resp)
        pages = [t for t in targets if t.get("type") == "page"]
        if not pages:
            raise RuntimeError(f"No page target found on CDP for {self.node}")
        ws_url = pages[0]["webSocketDebuggerUrl"]
        self.ws = websocket.create_connection(ws_url, timeout=self.timeout)

    def send_cmd(self, method, params=None):
        self.msg_id += 1
        cid = self.msg_id
        payload = {"id": cid, "method": method, "params": params or {}}
        self.ws.send(json.dumps(payload))
        while True:
            raw = self.ws.recv()
            data = json.loads(raw)
            if data.get("id") == cid:
                return data

    def evaluate(self, expr, await_promise=False):
        res = self.send_cmd(
            "Runtime.evaluate",
            {"expression": expr, "returnByValue": True, "awaitPromise": await_promise},
        )
        result = res.get("result", {}).get("result", {})
        if res.get("result", {}).get("exceptionDetails"):
            desc = res["result"]["exceptionDetails"].get("text", "JS exception")
            raise RuntimeError(f"CDP eval error: {desc}")
        return result.get("value")

    def dispatch_key(self, key, code, modifiers=0):
        self.send_cmd(
            "Input.dispatchKeyEvent",
            {
                "type": "rawKeyDown",
                "key": key,
                "code": code,
                "modifiers": modifiers,
                "windowsVirtualKeyCode": 74 if code == "KeyJ" else 0,
            },
        )
        self.send_cmd(
            "Input.dispatchKeyEvent",
            {
                "type": "keyUp",
                "key": key,
                "code": code,
                "modifiers": modifiers,
                "windowsVirtualKeyCode": 74 if code == "KeyJ" else 0,
            },
        )

    def close(self):
        if self.ws:
            try:
                self.ws.close()
            except Exception:
                pass
            self.ws = None


DOM_EXTRACT_JS = """(() => {
    const els = [...document.querySelectorAll('[data-message-id]')];
    return els.map(m => {
        const id = m.getAttribute('data-message-id');
        const ps = [...m.querySelectorAll('p')].map(p => (p.innerText || '').trim()).filter(Boolean);
        let text = ps.join('\\n');
        if (!text) {
            text = (m.innerText || '').replace(/^(Assistant message:|User message:)\\s*/i, '').trim();
        }
        const t = m.querySelector('time');
        return {
            id: id,
            author: id.startsWith('assistant-msg') ? 'assistant' : 'user',
            text: text,
            ts: t ? (t.getAttribute('datetime') || t.innerText || null) : null
        };
    });
})()"""


def scrape_thread_messages(cdp, thread_id, watermark, max_scrollbacks=3):
    """Scrape messages with bounded scroll-back if watermark is out of view."""
    messages = cdp.evaluate(DOM_EXTRACT_JS) or []

    # If watermark exists and is already in view, or no watermark, no scroll-back needed
    seen_ids = {m["id"] for m in messages if m.get("id")}
    if watermark and watermark not in seen_ids and max_scrollbacks > 0:
        # Bounded scroll-back loop
        for _ in range(max_scrollbacks):
            cdp.evaluate("""(() => {
                const sc = document.getElementById('hatch-chat-scroll');
                if (sc) sc.scrollTop = 0;
            })()""")
            time.sleep(0.8)
            older = cdp.evaluate(DOM_EXTRACT_JS) or []
            for m in older:
                if m.get("id") and m["id"] not in seen_ids:
                    messages.insert(0, m)
                    seen_ids.add(m["id"])
            if watermark in seen_ids:
                break

    return messages


def process_messages(raw_messages, agent, thread_id, thread_name, last_wm, followups, dry_run=False, feed=None):
    """
    Standard message processor:
    - Filters for new messages based on watermark
    - Appends records to chat-history.jsonl
    - Detects [RESULT] and [VERB] markers, logging to job-log.jsonl and clearing followups
    - Returns (new_messages, new_watermark, job_results_count)
    """
    if not raw_messages:
        return [], last_wm, 0

    new_messages = []
    if not last_wm:
        # If thread has <= 10 messages (e.g. newly spawned job sidechat), process them!
        # Only fast-forward watermark if this is an established thread with a deep backlog.
        if len(raw_messages) <= 10:
            new_messages = raw_messages
        else:
            new_wm = raw_messages[-1]["id"]
            return [], new_wm, 0
    else:
        wm_idx = -1
        for i, m in enumerate(raw_messages):
            if m["id"] == last_wm:
                wm_idx = i
                break
        if wm_idx >= 0:
            new_messages = raw_messages[wm_idx + 1 :]
        else:
            # Watermark not found in loaded window
            new_messages = raw_messages

    if not new_messages:
        return [], last_wm, 0

    new_wm = new_messages[-1]["id"]
    job_results = 0

    for msg in new_messages:
        mid = msg.get("id", "")
        author = msg.get("author", "unknown")
        text = msg.get("text", "")
        msg_ts = msg.get("ts") or utcnow()

        record = {
            "ts": utcnow(),
            "agent": agent,
            "thread_id": thread_id,
            "thread_name": thread_name,
            "msg_id": mid,
            "author": author,
            "text": text,
            "source_ts": msg_ts,
        }
        if feed:
            record["feed"] = feed
        if not dry_run:
            append_jsonl(CHAT_HISTORY_LOG, record)

        if author == "assistant":
            markers = list(iter_result_markers(text))
            verbs = list(iter_verb_markers(text))

            # Synthesize [RESULT <job-id>] DECLINE if assistant explicitly refuses the task in plain text
            if not markers and not verbs and detect_explicit_refusal(text):
                matching_jid = None
                for f_id, f_rec in followups.items():
                    if f_rec.get("status") in ("pending", "acknowledged") and (f_rec.get("thread_uuid") == thread_id or f_rec.get("recipient") == agent):
                        matching_jid = f_rec.get("job_id") or f_id
                        break
                if not matching_jid and JOB_SIDECHATS_FILE.exists():
                    try:
                        sc_state = json.loads(JOB_SIDECHATS_FILE.read_text(encoding="utf-8"))
                        for k, v in sc_state.items():
                            if isinstance(v, dict) and v.get("thread_uuid") == thread_id:
                                matching_jid = k
                                break
                    except Exception:
                        pass
                if matching_jid:
                    refusal_snippet = f"DECLINE: {text.strip()[:180]}"
                    print(f"[{agent}] Synthesizing [RESULT {matching_jid}] DECLINE from explicit assistant refusal")
                    markers.append((matching_jid, refusal_snippet))

            # Check for structured tool / command calls from the agent
            tool_calls = parse_tool_calls(text)
            if tool_calls and not dry_run:
                for op, args in tool_calls:
                    if isinstance(args, dict) and "agent" not in args:
                        args["agent"] = agent
                    print(f"[{agent}] Executing tool '{op}' from message {mid[:8]} in thread {thread_name or thread_id[:8]}")
                    t_ok, t_res = execute_agent_tool(agent, op, args)
                    tool_record = {
                        "ts": utcnow(),
                        "type": "tool_exec",
                        "agent": agent,
                        "thread_id": thread_id,
                        "op": op,
                        "args": args,
                        "success": t_ok,
                        "result_snippet": str(t_res)[:300],
                        "msg_id": mid,
                    }
                    append_jsonl(JOB_LOG, tool_record)

                    # Post clean tool output back to the originating thread via fast headless gateway
                    if thread_id and re.fullmatch(r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}", thread_id.lower()):
                        try:
                            import muse_hybrid
                            thread_url = f"https://box.muse-dev.online/thread/{thread_id}"
                            tool_hint = (
                                f"[Runtime Context: {thread_url}]\n"
                                f"Tools available: [TOOL <op> <args>] or curl -sk -X POST https://exec.muse-dev.online/exec\n"
                                f"  • [TOOL followup.create {{\"in_m\": 5, \"prompt\": \"<reminder>\"}}]\n"
                                f"  • [TOOL swarm.spawn {{\"count\": 1, \"task\": \"<task>\"}}]\n"
                                f"  • [TOOL health.check {{}}]\n\n"
                                f"[Directive: Take next action or close with [RESULT <job_id>] <summary>]"
                            )
                            if t_ok:
                                clean_msg = format_tool_result_for_chat(op, t_res)
                                resp_text = f"Tool result (`{op}`):\n{clean_msg}\n\n{tool_hint}"
                            else:
                                resp_text = f"Tool error (`{op}`): {t_res}\n\n{tool_hint}"
                            muse_hybrid.send_message(agent, resp_text, thread_id=thread_id, wait=0)
                        except Exception as te:
                            sys.stderr.write(f"warning: failed to post tool response back to thread: {te}\n")

            if markers or verbs:
                for job_id, result_text in markers:
                    is_fail = is_fail_result(result_text)
                    job_results += 1

                    job_record = {
                        "ts": utcnow(),
                        "type": "job_result",
                        "job_id": job_id,
                        "agent": agent,
                        "success": not is_fail,
                        "result_snippet": result_text[:300],
                        "thread_id": thread_id,
                        "msg_id": mid,
                    }
                    if not dry_run:
                        append_jsonl(JOB_LOG, job_record)
                        # Check if this is a swarm slot result: sw-YYYYMMDD-HHMMSS-xxxx/<slot>
                        if "/" in job_id and job_id.startswith("sw-"):
                            try:
                                s_sid, s_slot = job_id.split("/", 1)
                                s_proc = subprocess.Popen(
                                    [sys.executable, str(BIN_DIR / "box-ctl.py"), "swarm-report", s_sid, s_slot],
                                    stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL
                                )
                                s_proc.communicate(input=json.dumps({"ok": not is_fail, "result": result_text}).encode("utf-8"))
                            except Exception as se:
                                sys.stderr.write(f"warning: failed to record swarm report: {se}\n")
                        else:
                            trigger_chain_next(job_id, result_text, success=not is_fail)
                        if not is_fail:
                            archive_ephemeral_thread(agent, thread_id, job_id=job_id)
                    clear_matching_followups(followups, agent, thread_id, mid, text,
                                             dry_run, job_id=job_id, verb="RESULT")
                for verb, job_id in verbs:
                    if verb == "RESULT":
                        continue
                    clear_matching_followups(followups, agent, thread_id, mid, text,
                                             dry_run, job_id=job_id, verb=verb)
            else:
                clear_matching_followups(followups, agent, thread_id, mid, text, dry_run)
                maybe_nudge_untagged_sidechat(agent, thread_id, thread_name, mid, text, dry_run=dry_run)

    return new_messages, new_wm, job_results


def harvest_thread_gateway(agent, thread_info, watermarks, followups, dry_run=False):
    """
    Harvests messages via fast headless gateway (muse_hybrid).
    Zero CDP connections, zero browser page disruption, zero URL hopping.
    """
    if not HAS_MUSE_HYBRID:
        raise RuntimeError("muse_hybrid module not available")

    thread_id = thread_info["id"]
    thread_name = thread_info["name"]
    wm_key = f"{agent}:{thread_id}"
    last_wm = watermarks.get(wm_key, "")

    hist, err = muse_hybrid.get_history(
        agent,
        thread_id=None if thread_id == "main" else thread_id,
        limit=30
    )
    if err or hist is None:
        if err and ("not_found" in err or "404" in err):
            # Session no longer exists; skip cleanly
            return [], last_wm, 0
        raise RuntimeError(f"muse_hybrid error for {agent}:{thread_id}: {err}")

    raw_messages = []
    for m in hist:
        mid = m.get("message_id") or m.get("id") or ""
        role = m.get("role") or m.get("author") or "unknown"
        text = m.get("text", "")
        raw_messages.append({
            "id": mid,
            "author": role,
            "text": text,
            "ts": m.get("ts") or utcnow(),
        })

    return process_messages(
        raw_messages, agent, thread_id, thread_name, last_wm, followups,
        dry_run=dry_run, feed="fast_gateway"
    )


def harvest_main_feed(cdp, agent, watermarks, followups, dry_run=False):
    """
    Dedicated Main Chat CDP harvester (fallback).
    Operates opportunistically: only scrapes if browser is ALREADY on Main Chat.
    """
    curr_url = cdp.evaluate("window.location.href") or ""
    if "/thread/" in curr_url:
        return [], None, 0

    wm_key = f"{agent}:main"
    last_wm = watermarks.get(wm_key, "")
    raw_messages = scrape_thread_messages(cdp, "main", last_wm, max_scrollbacks=0)
    return process_messages(
        raw_messages, agent, "main", "Main Chat", last_wm, followups,
        dry_run=dry_run, feed="main_chat_passive"
    )


def harvest_agent_thread(cdp, agent, thread_info, watermarks, followups, dry_run=False):
    """
    Harvests new messages for a single thread via CDP (fallback),
    preserves URL state, and returns (new_messages, new_watermark, job_results_count).
    """
    thread_id = thread_info["id"]
    thread_name = thread_info["name"]
    wm_key = f"{agent}:{thread_id}"
    last_wm = watermarks.get(wm_key, "")

    initial_url = cdp.evaluate("window.location.href") or "https://muse.ai/"
    is_init_main = "/thread/" not in initial_url

    try:
        if thread_id == "main":
            if not is_init_main:
                cdp.dispatch_key("j", "KeyJ", modifiers=2)
                time.sleep(2.0)
        else:
            target_url = f"https://muse.ai/thread/{thread_id}"
            if initial_url.strip() != target_url:
                cdp.evaluate(f"window.location.href = {json.dumps(target_url)}")
                time.sleep(2.5)

        raw_messages = scrape_thread_messages(cdp, thread_id, last_wm)
    finally:
        try:
            curr_url = cdp.evaluate("window.location.href") or ""
            if is_init_main:
                if "/thread/" in curr_url:
                    cdp.dispatch_key("j", "KeyJ", modifiers=2)
            else:
                if curr_url.strip() != initial_url.strip():
                    cdp.evaluate(f"window.location.href = {json.dumps(initial_url)}")
        except Exception:
            pass

    return process_messages(
        raw_messages, agent, thread_id, thread_name, last_wm, followups,
        dry_run=dry_run
    )


def archive_ephemeral_thread(agent, thread_id, job_id=None):
    """
    If thread_id belongs to an ephemeral job or one-off check,
    archive it via hybrid gateway and tag it as archived in job-sidechats.json.
    """
    if not thread_id or not re.fullmatch(r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}", thread_id.lower()):
        return

    # Check if thread is marked as persistent in job-sidechats.json
    try:
        if JOB_SIDECHATS_FILE.exists():
            sc_state = json.loads(JOB_SIDECHATS_FILE.read_text(encoding="utf-8"))
            for key, val in sc_state.items():
                if isinstance(val, dict) and val.get("thread_uuid") == thread_id:
                    if val.get("type") == "persistent":
                        return  # Do not archive persistent coordinator channels
                    val["archived"] = True
                    val["archived_at"] = utcnow()
                    val["archived_by_job"] = job_id
            JOB_SIDECHATS_FILE.write_text(json.dumps(sc_state, indent=2), encoding="utf-8")
    except Exception as e:
        sys.stderr.write(f"warning: failed to update job-sidechats state: {e}\n")

    # Call muse-threads.py archive via subprocess (runs in node's netns)
    try:
        helper = BIN_DIR / "muse-threads.py"
        subprocess.run(
            [sys.executable, str(helper), "archive", "--agent", agent, "--thread", thread_id],
            capture_output=True, text=True, timeout=15
        )
        print(f"[{agent}] Archived ephemeral thread {thread_id[:8]} upon completion of job {job_id or 'unknown'}")
    except Exception as ae:
        sys.stderr.write(f"warning: archive_ephemeral_thread failed: {ae}\n")


# NOTE 2026-10-05 (Fix Agent 2/5): dev/def removed from the pool. No worker
# agents exist on dev/def (no Meta sessions provisioned), so slots dispatched
# to them froze with null results. Re-add only after real dev/def workers exist.
SWARM_WORKER_POOL = ["muse"]

# Stuck-slot reaper: a slot that stays "running" with no result longer than
# this is treated as wedged (worker died / dispatch lost). Healthy slots
# complete in <5 min (observed p90 3.6 min over 26 done slots, 2026-10-05),
# so 60 min is conservative.
STUCK_SLOT_MINUTES = 60

# Verified dispatch: dm.py send runs synchronously and the slot is only
# marked running when the send is confirmed (rc 0 + "SENT" in stdout).
# Unverified sends stay pending for retry; after N attempts the slot fails
# loudly instead of freezing from birth.
DISPATCH_VERIFY_TIMEOUT = 120
DISPATCH_MAX_ATTEMPTS = 5


def _parse_ts(ts):
    try:
        return datetime.fromisoformat(str(ts).replace("Z", "+00:00"))
    except Exception:
        return None



def reconcile_and_dispatch_swarms(dry_run=False):
    """
    Autonomous Swarm Orchestrator:
    1. Scans swarms.json for pending slots.
    2. Dynamically allocates available auxiliary worker nodes (dev, def, muse).
    3. Provisions ephemeral sidechat per slot and dispatches the task with [RESULT <swarm_id>/<slot>].
    4. Upon completion of all slots, sends completion summary DM to originating coordinator.
    """
    if not SWARM_FILE.exists() or dry_run:
        return
    try:
        swarms = json.loads(SWARM_FILE.read_text(encoding="utf-8"))
    except Exception:
        return

    modified = False
    now = utcnow()

    # 0. Reap stuck slots BEFORE computing busy workers. A wedged worker
    # would otherwise pin itself "busy" forever and the pool stalls: with
    # all workers busy the fallback round-robin keeps feeding new slots to
    # the same dead workers.
    now_dt = _parse_ts(now)
    if now_dt is not None:
        for _sid, _swarm in swarms.items():
            if _swarm.get("status") not in ("pending", "running"):
                continue
            for _s in _swarm.get("slots", []):
                if _s.get("status") != "running" or _s.get("result") is not None:
                    continue
                _upd = _parse_ts(_s.get("updated_ts", ""))
                if _upd is None:
                    continue
                _age_min = (now_dt - _upd).total_seconds() / 60
                if _age_min > STUCK_SLOT_MINUTES:
                    _s["status"] = "failed"
                    _s["result"] = {
                        "ok": False,
                        "reaped": True,
                        "reason": "stuck: running with no result for %.0f min (limit %d)"
                        % (_age_min, STUCK_SLOT_MINUTES),
                        "worker": _s.get("agent_id"),
                    }
                    _s["updated_ts"] = now
                    modified = True
                    print("[swarm] Reaped stuck slot %d of %s (worker %s, silent %.0f min)"
                          % (_s["slot"], _sid, _s.get("agent_id"), _age_min))

    # Determine busy workers from running slots
    busy_workers = set()
    for sid, swarm in swarms.items():
        if swarm.get("status") in ("pending", "running"):
            for slot in swarm.get("slots", []):
                if slot.get("status") == "running" and slot.get("agent_id"):
                    busy_workers.add(slot["agent_id"])

    # Process each swarm
    for sid, swarm in swarms.items():
        st = swarm.get("status")
        creator = swarm.get("created_by") or "646"
        if creator not in ("646", "pip", "opm", "muse", "dev", "def"):
            creator = "646"

        # 1. Allocate & dispatch pending slots
        if st in ("pending", "running"):
            slots = swarm.get("slots", [])
            for s in slots:
                if s.get("status") == "pending":
                    # Find first available worker
                    worker = None
                    for w in SWARM_WORKER_POOL:
                        if w not in busy_workers:
                            worker = w
                            break
                    if not worker:
                        # Fallback round-robin across worker pool if all are busy
                        worker = SWARM_WORKER_POOL[s["slot"] % len(SWARM_WORKER_POOL)]

                    slot_idx = s["slot"]
                    task_text = swarm.get("task", "execute subagent task")
                    slot_target = f"{sid}-s{slot_idx}"

                    # Prepare task directive prompt
                    prompt = (
                        f"[JOB {sid}/{slot_idx}] Task for swarm slot {slot_idx}:\n"
                        f"{task_text}\n\n"
                        f"Reply with [RESULT {sid}/{slot_idx}] OK <summary> or FAIL <reason>."
                    )

                    # Verified dispatch: confirm the send landed BEFORE
                    # marking the slot running. Fire-and-forget used to
                    # freeze slots from birth -- the slot read "running"
                    # while no worker was ever notified (2026-10-05: dev's
                    # wedged WireGuard data path silently killed every
                    # gateway dispatch).
                    dispatch_ok = False
                    try:
                        dm_cmd = [
                            sys.executable, str(BIN_DIR / "dm.py"), "send",
                            "--agent", creator,
                            "--to", worker,
                            "--target", slot_target,
                            prompt,
                        ]
                        proc = subprocess.run(
                            dm_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                            text=True, timeout=DISPATCH_VERIFY_TIMEOUT)
                        _dout = proc.stdout or ""
                        dispatch_ok = (proc.returncode == 0 and "SENT" in _dout)
                        if not dispatch_ok:
                            sys.stderr.write(
                                "warning: swarm dispatch unverified %s slot %d -> %s "
                                "(rc=%d out=%.100s err=%.100s)\n"
                                % (sid, slot_idx, worker, proc.returncode,
                                   _dout, proc.stderr or ""))
                    except Exception as de:
                        sys.stderr.write(
                            "warning: swarm dispatch error %s slot %d -> %s: %s\n"
                            % (sid, slot_idx, worker, de))

                    if not dispatch_ok:
                        # Leave pending so the next harvester pass retries
                        # (possibly with a different worker). Fail loudly
                        # after N attempts instead of spinning forever.
                        attempts = s.get("dispatch_attempts", 0) + 1
                        s["dispatch_attempts"] = attempts
                        modified = True
                        if attempts >= DISPATCH_MAX_ATTEMPTS:
                            s["status"] = "failed"
                            s["result"] = {
                                "ok": False,
                                "reaped": True,
                                "reason": "dispatch failed %d times (send never verified)" % attempts,
                                "worker": worker,
                            }
                            s["updated_ts"] = now
                            print("[swarm] Dispatch failed %dx for slot %d of %s; marked failed"
                                  % (attempts, slot_idx, sid))
                        else:
                            print("[swarm] Dispatch unverified for slot %d of %s "
                                  "(attempt %d/%d); leaving pending for retry"
                                  % (slot_idx, sid, attempts, DISPATCH_MAX_ATTEMPTS))
                        continue

                    # Attach worker to slot (only after verified dispatch)
                    s["agent_id"] = worker
                    s["status"] = "running"
                    s["updated_ts"] = now
                    busy_workers.add(worker)
                    modified = True
                    print(f"[swarm] Dispatched slot {slot_idx} of {sid} to {worker} in sidechat {slot_target}")

            # Update swarm rollup status
            counts = {"pending": 0, "running": 0, "done": 0, "failed": 0, "killed": 0}
            for slot in slots:
                counts[slot.get("status", "pending")] = counts.get(slot.get("status", "pending"), 0) + 1
            if counts["pending"] + counts["running"] == 0:
                swarm["status"] = "completed" if counts["failed"] == 0 else "partial"
                swarm["updated_ts"] = now
                modified = True
            elif counts["running"] or counts["done"]:
                swarm["status"] = "running"
                swarm["updated_ts"] = now
                modified = True

        # 2. Check if newly completed/partial and notify coordinator
        if swarm.get("status") in ("completed", "partial") and not swarm.get("notified_coordinator"):
            swarm["notified_coordinator"] = True
            modified = True
            done_cnt = sum(1 for sl in swarm.get("slots", []) if sl.get("status") == "done")
            total_cnt = len(swarm.get("slots", []))
            summary_msg = (
                f"[Swarm Report] Swarm {sid} ({swarm.get('label') or 'task'}) finished: "
                f"{done_cnt}/{total_cnt} slots successful. "
                f"Console: https://box.muse-dev.online/#dashboard"
            )
            try:
                coord_dm = [
                    sys.executable, str(BIN_DIR / "dm.py"), "send",
                    "--agent", "box",
                    "--to", creator,
                    "--target", "main",
                    "--allow-main-chat",
                    summary_msg,
                ]
                subprocess.Popen(coord_dm, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
                print(f"[swarm] Delivered completion report for {sid} to coordinator {creator}")
            except Exception as ce:
                sys.stderr.write(f"warning: failed to notify swarm coordinator: {ce}\n")

    if modified:
        tmp_swarms = str(SWARM_FILE) + f".tmp.{os.getpid()}"
        with open(tmp_swarms, "w", encoding="utf-8") as f:
            json.dump(swarms, f, indent=2)
        os.replace(tmp_swarms, SWARM_FILE)


def check_and_archive_terminal_swarms():
    """Check swarms.json and auto-archive ephemeral sidechats for completed or terminal swarms."""
    if not SWARM_FILE.exists() or not JOB_SIDECHATS_FILE.exists():
        return
    try:
        swarms_data = json.loads(SWARM_FILE.read_text(encoding="utf-8"))
        sc_data = json.loads(JOB_SIDECHATS_FILE.read_text(encoding="utf-8"))
    except Exception:
        return

    for sid, swarm in swarms_data.items():
        st = swarm.get("status")
        if st in ("completed", "partial", "killed"):
            # Check slots or matching sidechats
            for key, val in sc_data.items():
                if isinstance(val, dict) and not val.get("archived") and val.get("type") != "persistent":
                    if sid in key or (swarm.get("label") and swarm.get("label") in key):
                        tu = val.get("thread_uuid")
                        ag = val.get("agent", "opm")
                        if tu:
                            archive_ephemeral_thread(ag, tu, job_id=sid)



def maybe_nudge_untagged_sidechat(agent, thread_id, thread_name, mid, text, dry_run=False):
    """
    If an agent replies conversationally in a sidechat backed by a job or follow-up
    without providing [RESULT <id>] or tool directives, deliver a terse 1-turn nudge footer.
    """
    if dry_run or not thread_id or thread_id == "main":
        return
    if not re.fullmatch(r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}", thread_id.lower()):
        return

    # Look up active follow-ups or job association for this thread
    followups = load_json_file(FOLLOWUPS_FILE)
    matching_job_id = None
    for f_id, f_rec in followups.items():
        if f_rec.get("status") in ("pending", "acknowledged") and f_rec.get("thread_uuid") == thread_id:
            matching_job_id = f_rec.get("job_id") or f_id
            break

    if not matching_job_id and JOB_SIDECHATS_FILE.exists():
        try:
            sc_state = json.loads(JOB_SIDECHATS_FILE.read_text(encoding="utf-8"))
            for k, v in sc_state.items():
                if isinstance(v, dict) and v.get("thread_uuid") == thread_id:
                    if k.startswith(("box-", "job-", "pipe-", "autonomy-pulse-")):
                        matching_job_id = k
                    break
        except Exception:
            pass

    if not matching_job_id:
        return

    tracker = load_json_file(NUDGE_TRACKER_FILE)
    rec = tracker.get(thread_id, {})
    if rec.get("nudged_for_mid") == mid:
        return

    n_count = rec.get("nudge_count", 0)

    # 1st turn: Strict rejection and tool invocation prompt
    if n_count == 0:
        thread_url = f"https://box.muse-dev.online/thread/{thread_id}"
        try:
            import prompt_envelope
            _spawn = prompt_envelope.spawn_call(
                matching_job_id, matching_job_id, prompt_envelope.pick_profile(matching_job_id))
        except Exception:
            _spawn = '[TOOL swarm.spawn {"count": 2, "task": "continue the job work"}]'
        nudge_msg = (
            f"{_spawn}\n"
            f"[STRICT ENFORCEMENT: Conversational commentary is rejected. Work requires active execution.]\n"
            f"Thread Console: {thread_url}\n"
            f"Emit executable tool calls now: [TOOL <op> <args>] or curl against https://exec.muse-dev.online/exec\n"
            f"When all operations are finished, close strictly with [RESULT {matching_job_id}] <outcome>.\n"
            f"{_spawn}"
        )
        try:
            import muse_hybrid
            print(f"[{agent}] Injecting 1-turn strict nudge into {thread_name or thread_id[:8]} for job {matching_job_id}")
            muse_hybrid.send_message(agent, nudge_msg, thread_id=thread_id, wait=0)
            tracker[thread_id] = {
                "ts": utcnow(),
                "job_id": matching_job_id,
                "nudged_for_mid": mid,
                "nudge_count": 1,
            }
            save_json_file(NUDGE_TRACKER_FILE, tracker)
        except Exception as ne:
            sys.stderr.write(f"warning: failed to deliver conversational nudge: {ne}\n")

    # 2nd turn: Trigger emergency 5m escalation follow-up to opm/coordinator
    elif n_count == 1 and not rec.get("escalated"):
        print(f"[{agent}] Conversational stall persisted in {thread_name or thread_id[:8]} - scheduling emergency escalation to opm")
        esc_prompt = f"[ESCALATION] Agent {agent} stalled in thread {thread_name or thread_id[:8]} on job {matching_job_id}. Review thread or reassign."
        try:
            cmd = [
                "systemd-run", "--user", "--on-active=300s",
                sys.executable, str(BIN_DIR / "box-ctl.py"), "notify", "opm", esc_prompt, "--sender", agent
            ]
            subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
            tracker[thread_id]["escalated"] = True
            tracker[thread_id]["escalated_ts"] = utcnow()
            save_json_file(NUDGE_TRACKER_FILE, tracker)
        except Exception as ee:
            sys.stderr.write(f"warning: failed to schedule emergency escalation: {ee}\n")



# Chain deduplication
CHAINED_JOBS_FILE = Path(__file__).parent / "chained-jobs.json"

def _has_chained(job_id):
    try:
        if CHAINED_JOBS_FILE.exists():
            import json as _j
            with open(CHAINED_JOBS_FILE) as f:
                return job_id in _j.load(f)
    except: pass
    return False

def _mark_chained(job_id):
    try:
        import json as _j
        c = []
        if CHAINED_JOBS_FILE.exists():
            with open(CHAINED_JOBS_FILE) as f: c = _j.load(f)
        if job_id not in c:
            c.append(job_id)
            c = c[-1000:]
            with open(CHAINED_JOBS_FILE, "w") as f: _j.dump(c, f)
    except: pass

def trigger_chain_next(job_id, result_text, success=True):
    """If the completed job has on_success, on_failure, or chain_next, dispatch downstream."""
    if _has_chained(job_id):
        return
    _mark_chained(job_id)

    m = re.match(r"^(.*)-(\d{8}-\d{6}-[a-f0-9]{8})$", job_id)
    if m:
        job_name = m.group(1)
    else:
        parts = job_id.split("-")
        if len(parts) < 3:
            return
        job_name = "-".join(parts[:-2])

    job_file = JOBS_DIR / f"{job_name}.json"
    if not job_file.exists():
        return

    # Update pipeline ledger if this job belongs to an active pipeline
    pipeline_run_id = None
    step_n = 1
    if HAS_PIPELINE:
        run_entry, step_entry = pipeline_engine.record_step_result(job_id, success, result_text)
        if run_entry:
            pipeline_run_id = run_entry.get("run_id")
            step_n = step_entry.get("step_n", 1) + 1

    try:
        with open(job_file, "r", encoding="utf-8") as f:
            cfg = json.load(f)

        next_job = None
        if success:
            next_job = cfg.get("on_success") or cfg.get("chain_next")
        else:
            next_job = cfg.get("on_failure")

        if next_job and (JOBS_DIR / f"{next_job}.json").exists():
            # Inter-step settle delay to prevent browser race conditions
            step_delay = int(cfg.get("step_delay", 5))
            if step_delay > 0:
                time.sleep(step_delay)

            env = os.environ.copy()
            env["CHAIN_PREV_JOB_ID"] = job_id
            env["CHAIN_PREV_RESULT"] = result_text[:1000]
            if pipeline_run_id:
                env["CHAIN_PIPELINE_RUN_ID"] = pipeline_run_id
                env["CHAIN_STEP_N"] = str(step_n)

            cmd = [sys.executable, str(DISPATCH_PY), next_job]
            if pipeline_run_id:
                cmd.extend(["--pipeline-run", pipeline_run_id, "--step-n", str(step_n)])

            subprocess.Popen(
                cmd,
                env=env,
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
            )
        else:
            # End of chain for this pipeline run
            if pipeline_run_id and HAS_PIPELINE:
                if success:
                    pipeline_engine.complete_pipeline(pipeline_run_id)
                else:
                    pipeline_engine.fail_pipeline(pipeline_run_id, "step_failed_without_fallback")
    except Exception:
        pass


def clear_matching_followups(followups, agent, thread_id, mid, text, dry_run=False,
                             job_id=None, verb=None):
    """Resolve follow-up records if an assistant message is detected in the thread.

    Matches on thread identity (thread_uuid or target='main') OR on job_id
    (from a [RESULT <job_id>] reply). The job_id path works regardless of
    thread_uuid or target, fixing ghost followups with null thread_uuid.
    Also matches a main-chat reply when the sweeper recorded
    final_nudge_target='main' (final nudge routed to main chat).

    When verb is given ([ACK|CLAIM|RESULT|DECLINE|NO-ACTION <job_id>]),
    matching is job_id-scoped for non-RESULT verbs (a verb marker names the
    digest it answers, so it must not touch unrelated pending followups that
    merely share the thread); RESULT keeps the historical thread-or-job
    matching. Every verb match records outcome=<verb>. ACK/CLAIM set status
    'acknowledged' (nudge-suppressed, NOT closed) instead of 'resolved', and
    also match already-'acknowledged' records so an ACK -> RESULT lifecycle
    closes correctly.
    """
    if not followups:
        return

    modified = False
    for f_id, f_rec in followups.items():
        # Verb replies can follow an ACK (ACK -> RESULT lifecycle), so verbs
        # also match 'acknowledged' records; plain replies only match pending.
        if verb:
            if f_rec.get("status") not in ("pending", "acknowledged"):
                continue
        elif f_rec.get("status") != "pending":
            continue
        if f_rec.get("recipient") != agent:
            continue

        # Match either exact thread_uuid, or target alias 'main'
        match_thread = False
        if f_rec.get("target") == "main" and thread_id == "main":
            match_thread = True
        elif f_rec.get("thread_uuid") and f_rec.get("thread_uuid") == thread_id:
            match_thread = True
        elif f_rec.get("final_nudge_target") == "main" and thread_id == "main":
            # C3: the sweeper routed the final nudge to main chat, so a
            # main-chat reply resolves even when the followup target is a
            # sidechat.
            match_thread = True

        # Match by job_id (from [RESULT <job_id>] or [VERB <job_id>]) --
        # works regardless of thread_uuid or target. This is an ADDITIONAL
        # path, not a replacement.
        match_job = False
        if job_id and f_rec.get("job_id") and f_rec.get("job_id") == job_id:
            match_job = True

        # Non-RESULT verbs are job-scoped: they must not acknowledge/resolve
        # unrelated pending followups that merely share the thread. RESULT
        # keeps the historical thread-or-job matching.
        if verb and verb != "RESULT" and not match_job:
            continue

        if match_thread or match_job:
            if verb:
                f_rec["outcome"] = verb
            if verb in ("ACK", "CLAIM"):
                # Acknowledged: sweeper nudges stop (status != pending), but
                # the digest is NOT closed until a closing verb arrives.
                f_rec["status"] = "acknowledged"
                f_rec["acknowledged_at"] = utcnow()
                f_rec["acknowledged_by_mid"] = mid
            else:
                f_rec["status"] = "resolved"
                f_rec["resolved_at"] = utcnow()
                f_rec["resolved_by_mid"] = mid
                f_rec["resolved_snippet"] = text[:150]
            modified = True

    if modified and not dry_run:
        save_json_file(FOLLOWUPS_FILE, followups)


def harvest_cycle(target_agent=None, dry_run=False, output_json=False):
    """Execute one full harvest cycle across agents and threads."""
    watermarks = load_json_file(WATERMARKS_FILE)
    followups = load_json_file(FOLLOWUPS_FILE)
    monitored = get_monitored_threads(target_agent)

    cycle_stats = {
        "timestamp": utcnow(),
        "agents": {},
        "total_new_messages": 0,
        "total_job_results": 0,
    }

    for agent, thread_list in monitored.items():
        agent_stats = {"status": "ok", "threads": {}, "new_messages": 0, "job_results": 0}

        # 1. Fast headless gateway (zero CDP locks, zero browser navigation / tab hopping)
        if HAS_MUSE_HYBRID and muse_hybrid.is_node_configured(agent):
            try:
                all_targets = list(thread_list) + [{"id": "main", "name": "Main Chat"}]

                def _harvest_one(target_info):
                    try:
                        return target_info, harvest_thread_gateway(
                            agent, target_info, watermarks, followups, dry_run
                        )
                    except Exception:
                        return target_info, ([], None, 0)

                with concurrent.futures.ThreadPoolExecutor(max_workers=min(len(all_targets) or 1, 5)) as executor:
                    harvest_results = list(executor.map(_harvest_one, all_targets))

                for t_info, (new_msgs, new_wm, j_res) in harvest_results:
                    wm_key = f"{agent}:{t_info['id']}"
                    if new_wm and not dry_run:
                        watermarks[wm_key] = new_wm
                    if new_msgs:
                        agent_stats["threads"][t_info["name"]] = len(new_msgs)
                    agent_stats["new_messages"] += len(new_msgs)
                    agent_stats["job_results"] += j_res

                cycle_stats["agents"][agent] = agent_stats
                cycle_stats["total_new_messages"] += agent_stats["new_messages"]
                cycle_stats["total_job_results"] += agent_stats["job_results"]
                continue
            except Exception:
                pass

        # 2. Fallback: Direct CDP over host veth
        peer_ip, port = get_node_network(agent)

        try:
            slot_ctx = (
                cdp_slot(agent, priority=PRIORITY_LOW, timeout=5)
                if HAS_CDP_QUEUE
                else None
            )
            if slot_ctx:
                with slot_ctx:
                    cdp = CDPClient(agent, peer_ip, port, timeout=8)
                    cdp.connect()
                    try:
                        # 1. Harvest registered sidechat threads
                        for t_info in thread_list:
                            new_msgs, new_wm, j_res = harvest_agent_thread(
                                cdp, agent, t_info, watermarks, followups, dry_run
                            )
                            wm_key = f"{agent}:{t_info['id']}"
                            if new_wm and not dry_run:
                                watermarks[wm_key] = new_wm
                            agent_stats["threads"][t_info["name"]] = len(new_msgs)
                            agent_stats["new_messages"] += len(new_msgs)
                            agent_stats["job_results"] += j_res

                        # 2. Opportunistic Main Chat feed harvest (zero navigation, only if already on main)
                        m_msgs, m_wm, m_res = harvest_main_feed(
                            cdp, agent, watermarks, followups, dry_run
                        )
                        if m_wm and not dry_run:
                            watermarks[f"{agent}:main"] = m_wm
                        if m_msgs:
                            agent_stats["threads"]["Main Chat (Passive)"] = len(m_msgs)
                            agent_stats["new_messages"] += len(m_msgs)
                            agent_stats["job_results"] += m_res
                    finally:
                        cdp.close()
            else:
                cdp = CDPClient(agent, peer_ip, port, timeout=8)
                cdp.connect()
                try:
                    for t_info in thread_list:
                        new_msgs, new_wm, j_res = harvest_agent_thread(
                            cdp, agent, t_info, watermarks, followups, dry_run
                        )
                        wm_key = f"{agent}:{t_info['id']}"
                        if new_wm and not dry_run:
                            watermarks[wm_key] = new_wm
                        agent_stats["threads"][t_info["name"]] = len(new_msgs)
                        agent_stats["new_messages"] += len(new_msgs)
                        agent_stats["job_results"] += j_res

                    m_msgs, m_wm, m_res = harvest_main_feed(
                        cdp, agent, watermarks, followups, dry_run
                    )
                    if m_wm and not dry_run:
                        watermarks[f"{agent}:main"] = m_wm
                    if m_msgs:
                        agent_stats["threads"]["Main Chat (Passive)"] = len(m_msgs)
                        agent_stats["new_messages"] += len(m_msgs)
                        agent_stats["job_results"] += m_res
                finally:
                    cdp.close()

        except Exception as e:
            agent_stats["status"] = "error"
            agent_stats["error"] = str(e)[:200]

        cycle_stats["agents"][agent] = agent_stats
        cycle_stats["total_new_messages"] += agent_stats["new_messages"]
        cycle_stats["total_job_results"] += agent_stats["job_results"]

    if not dry_run:
        save_json_file(WATERMARKS_FILE, watermarks)
        reconcile_and_dispatch_swarms()
        check_and_archive_terminal_swarms()

    # Output formatting
    if output_json:
        print(json.dumps(cycle_stats))
    else:
        ts_short = cycle_stats["timestamp"].split("T")[1][:8]
        summary_parts = []
        for ag, st in cycle_stats["agents"].items():
            if st["status"] == "ok":
                summary_parts.append(f"{ag}: {st['new_messages']} msgs ({st['job_results']} results)")
            else:
                summary_parts.append(f"{ag}: [UNREACHABLE: {st.get('error', 'err')[:40]}]")
        print(f"[{ts_short}Z] Harvest cycle: {', '.join(summary_parts)}")

    return cycle_stats


def main():
    parser = argparse.ArgumentParser(description="Fleet agent readback and response harvester")
    parser.add_argument("--once", action="store_true", help="Run once and exit (default)")
    parser.add_argument("--loop", action="store_true", help="Run continuously in a daemon loop")
    parser.add_argument("--interval", type=int, default=30, help="Interval in seconds for --loop (default 30)")
    parser.add_argument("--agent", choices=VALID_AGENTS, default=None, help="Harvest only specific agent")
    parser.add_argument("--dry-run", action="store_true", help="Scrape without persisting watermarks or logs")
    parser.add_argument("--json", action="store_true", help="Output summary as JSON")
    args = parser.parse_args()

    # Default to --once if --loop is not provided
    if not args.loop:
        harvest_cycle(target_agent=args.agent, dry_run=args.dry_run, output_json=args.json)
        return

    print(f"Starting response-harvester daemon (interval={args.interval}s, agent={args.agent or 'all'})...")
    while True:
        try:
            harvest_cycle(target_agent=args.agent, dry_run=args.dry_run, output_json=args.json)
        except Exception as e:
            print(f"ERROR in harvest loop: {e}", file=sys.stderr)
        time.sleep(args.interval)


if __name__ == "__main__":
    main()