box/bin/followup-sweeper.py

#!/usr/bin/env python3
"""
followup-sweeper.py — Autonomous follow-up deadline tracking and nudge sweeper.

Monitors pending follow-ups in followups.json, delivers progressive nudges to
recipients when deadlines expire (in-thread first, Main Chat on final nudge),
and executes terminal escalations to opm when all nudges are exhausted.

Usage:
    python3 followup-sweeper.py --once
    python3 followup-sweeper.py --loop --interval 30
"""

import argparse
import json
import os
import subprocess
import sys
import time
from datetime import datetime, timezone, timedelta
from pathlib import Path

# Paths
NETVM_ROOT = Path("/home/super/Projects/NetVM")
BIN_DIR = NETVM_ROOT / "bin"
FOLLOWUPS_FILE = NETVM_ROOT / "followups.json"
JOB_LOG = NETVM_ROOT / "job-log.jsonl"
DM_PY = BIN_DIR / "dm.py"


def utcnow_dt():
    return datetime.now(timezone.utc)


def utcnow_str():
    return utcnow_dt().isoformat()


def parse_iso(ts_str):
    if not ts_str:
        return None
    try:
        ts_clean = ts_str.replace("Z", "+00:00")
        dt = datetime.fromisoformat(ts_clean)
        if dt.tzinfo is None:
            dt = dt.replace(tzinfo=timezone.utc)
        return dt
    except Exception:
        return None


def load_followups():
    if not FOLLOWUPS_FILE.exists():
        return {}
    try:
        with open(FOLLOWUPS_FILE, "r", encoding="utf-8") as f:
            return json.load(f)
    except Exception:
        return {}


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


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


def send_dm(sender, recipient, target, text):
    """Dispatch a DM via dm.py."""
    cmd = [
        sys.executable,
        str(DM_PY),
        "send",
        "--agent", sender,
        "--to", recipient,
        "--target", target,
        text,
    ]
    try:
        res = subprocess.run(cmd, capture_output=True, text=True, timeout=90)
        return res.returncode == 0, res.stdout.strip() or res.stderr.strip()
    except Exception as e:
        return False, str(e)


def sweep_cycle(dry_run=False):
    followups = load_followups()
    if not followups:
        return {"status": "ok", "pending": 0, "nudges_sent": 0, "escalations": 0}

    now = utcnow_dt()
    nudges_count = 0
    escalations_count = 0
    modified = False

    for dm_id, rec in list(followups.items()):
        if rec.get("status") != "pending":
            continue

        deadline_dt = parse_iso(rec.get("deadline"))
        if not deadline_dt or now < deadline_dt:
            continue

        # Deadline has expired!
        nudges_sent = rec.get("nudges_sent", 0)
        nudges_allowed = rec.get("nudges_allowed", 2)
        sender = rec.get("sender", "opm")
        recipient = rec.get("recipient")
        orig_target = rec.get("target", "main")
        thread_uuid = rec.get("thread_uuid")

        if nudges_sent < nudges_allowed:
            # Deliver next nudge
            nudge_num = nudges_sent + 1
            is_final = (nudge_num == nudges_allowed)
            # Routing: In-thread first, Main on final nudge
            delivery_target = "main" if is_final else orig_target

            nudge_text = (
                f"[nudge {nudge_num}/{nudges_allowed}] [ref:{dm_id}] "
                f"Reminder: awaiting reply to request sent at {rec.get('sent_at', 'earlier')}."
            )
            if is_final and orig_target != "main":
                nudge_text += f" (Origin thread: {orig_target})"

            print(f"Sweeper: Sending nudge {nudge_num}/{nudges_allowed} to {recipient}/{delivery_target}...")
            if not dry_run:
                ok, out = send_dm(sender, recipient, delivery_target, nudge_text)
                if ok:
                    nudges_count += 1
                    rec["nudges_sent"] = nudge_num
                    rec["last_nudge_at"] = utcnow_str()
                    # Calculate interval for next nudge: proportional to timeout or default 10m
                    timeout_s = rec.get("timeout_s", 1800)
                    step_s = max(300, timeout_s // (nudges_allowed + 1))
                    rec["deadline"] = (now + timedelta(seconds=step_s)).isoformat()
                    modified = True
                    append_job_log({
                        "ts": utcnow_str(),
                        "type": "followup_nudged",
                        "dm_id": dm_id,
                        "nudge_num": nudge_num,
                        "recipient": recipient,
                        "target": delivery_target,
                    })
                else:
                    print(f"Sweeper WARNING: nudge send failed: {out}", file=sys.stderr)
            else:
                nudges_count += 1

        else:
            # All nudges exhausted: Terminal escalation
            escalate_to = rec.get("escalate_to", "opm")
            esc_text = (
                f"[ESCALATION] Agent {recipient} failed to reply to DM {dm_id} "
                f"after {nudges_allowed} nudges. Target was: {orig_target} "
                f"(thread: {thread_uuid or 'n/a'}). Request sent: {rec.get('sent_at')}."
            )
            print(f"Sweeper: Escalating expired follow-up {dm_id} to {escalate_to}...")
            if not dry_run:
                ok, out = send_dm("bl", escalate_to, "main", esc_text)
                rec["status"] = "escalated"
                rec["escalated_at"] = utcnow_str()
                escalations_count += 1
                modified = True
                append_job_log({
                    "ts": utcnow_str(),
                    "type": "followup_escalated",
                    "dm_id": dm_id,
                    "recipient": recipient,
                    "escalated_to": escalate_to,
                })
            else:
                escalations_count += 1

    if modified and not dry_run:
        save_followups(followups)

    pending_count = sum(1 for r in followups.values() if r.get("status") == "pending")
    return {
        "status": "ok",
        "pending": pending_count,
        "nudges_sent": nudges_count,
        "escalations": escalations_count,
    }


def main():
    parser = argparse.ArgumentParser(description="Autonomous follow-up deadline tracker and sweeper")
    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=60, help="Interval in seconds for loop (default 60)")
    parser.add_argument("--dry-run", action="store_true", help="Inspect without sending nudges or updating records")
    args = parser.parse_args()

    if not args.loop:
        stats = sweep_cycle(dry_run=args.dry_run)
        print(f"[{datetime.now(timezone.utc).strftime('%H:%M:%SZ')}] Sweep cycle: {stats['pending']} pending, {stats['nudges_sent']} nudges, {stats['escalations']} escalations.")
        return

    print(f"Starting follow-up sweeper loop (interval={args.interval}s)...")
    while True:
        try:
            stats = sweep_cycle(dry_run=args.dry_run)
            print(f"[{datetime.now(timezone.utc).strftime('%H:%M:%SZ')}] Sweep cycle: {stats['pending']} pending, {stats['nudges_sent']} nudges, {stats['escalations']} escalations.")
        except Exception as e:
            print(f"ERROR in sweeper loop: {e}", file=sys.stderr)
        time.sleep(args.interval)


if __name__ == "__main__":
    main()