pulse-bridge/bridge.py

#!/usr/bin/env python3
"""pulse-bridge v0 — mirror the estate's shared work ledger into Pulse posts.

Source: ~/.local/bin/shre-items (JSONL ledger, `feed --since` op stream).
Target: local mib007 (:5520) comms API — the backing store of the Pulse /
Activity feed. One ledger item = one post in the "activity" channel; stage
changes / failures / escalations = threaded comments; close = final comment.

Stdlib only. State (checkpoint + ledger-id -> post-id map) lives in
~/.shre/pulse-bridge/state.json, written atomically (tmp + rename).
"""

import json
import os
import re
import subprocess
import sys
import time
import urllib.error
import urllib.request
from datetime import datetime, timezone
from pathlib import Path

# ── Config (env file, then environment, then defaults) ───────────────────────

HOME = Path.home()
ENV_FILE = Path(os.environ.get("PULSE_BRIDGE_ENV",
                               Path(__file__).resolve().parent / "bridge.env"))


def _load_env_file(path: Path) -> dict:
    conf = {}
    if path.exists():
        for line in path.read_text(encoding="utf-8").splitlines():
            line = line.strip()
            if not line or line.startswith("#") or "=" not in line:
                continue
            k, v = line.split("=", 1)
            conf[k.strip()] = v.strip().strip('"').strip("'")
    return conf


_FILE = _load_env_file(ENV_FILE)


def cfg(key: str, default: str) -> str:
    return os.environ.get(key) or _FILE.get(key) or default


POLL_SEC = int(cfg("PULSE_BRIDGE_POLL_SEC", "30"))
MIB_BASE = cfg("PULSE_BRIDGE_MIB_BASE", "http://127.0.0.1:5520").rstrip("/")
WORKSPACE_ID = cfg("PULSE_BRIDGE_WORKSPACE_ID",
                   "293d29db-4978-4c54-a92e-b24d4c0a7115")  # Nirlab Command Center
CHANNEL_NAME = cfg("PULSE_BRIDGE_CHANNEL", "activity")
MENTION = cfg("PULSE_BRIDGE_MENTION", "nir")          # user handle to @mention
POSTER_NAME = cfg("PULSE_BRIDGE_POSTER_NAME", "Ledger")
SHRE_ITEMS = cfg("PULSE_BRIDGE_SHRE_ITEMS", str(HOME / ".local/bin/shre-items"))
STATE_FILE = Path(cfg("PULSE_BRIDGE_STATE",
                      str(HOME / ".shre/pulse-bridge/state.json")))
TOKEN_FILE = Path(cfg("PULSE_BRIDGE_TOKEN_FILE",
                      str(HOME / ".shre/service-tokens.json")))
TOKEN_KEY = cfg("PULSE_BRIDGE_TOKEN_KEY", "mib007")

STAGES = ["queued", "build", "review", "merge", "deploy", "verify"]


def log(msg: str) -> None:
    print(f"{datetime.now(timezone.utc).isoformat(timespec='seconds')} {msg}",
          flush=True)


# ── mib007 HTTP ──────────────────────────────────────────────────────────────

def _token() -> str:
    return json.loads(TOKEN_FILE.read_text(encoding="utf-8"))[TOKEN_KEY]


def api(method: str, path: str, body=None, timeout=15):
    req = urllib.request.Request(
        f"{MIB_BASE}{path}",
        data=json.dumps(body).encode() if body is not None else None,
        method=method,
        headers={
            "Authorization": f"Bearer {_token()}",
            "Content-Type": "application/json",
        },
    )
    with urllib.request.urlopen(req, timeout=timeout) as resp:
        raw = resp.read()
        return json.loads(raw) if raw else None


# ── State ────────────────────────────────────────────────────────────────────

def load_state() -> dict:
    if STATE_FILE.exists():
        try:
            return json.loads(STATE_FILE.read_text(encoding="utf-8"))
        except json.JSONDecodeError:
            # Starting fresh here would re-seed every open item -> a duplicate
            # post per item. Refuse to run until a human inspects the file.
            log(f"FATAL state file corrupt: {STATE_FILE} — fix or remove it "
                "deliberately; refusing to reseed automatically")
            sys.exit(1)
    return {"checkpoint": "", "seen": [], "map": {}, "seeded": False,
            "channel_id": ""}


def save_state(state: dict) -> None:
    STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
    tmp = STATE_FILE.with_suffix(".json.tmp")
    with open(tmp, "w", encoding="utf-8") as f:
        json.dump(state, f, indent=1)
        f.flush()
        os.fsync(f.fileno())
    os.replace(tmp, STATE_FILE)


# ── Content helpers ──────────────────────────────────────────────────────────

def neutralize(text: str) -> str:
    """Insert a zero-width space after '@' in ledger-derived text.

    mib007's comms route treats the FIRST '@handle' in a message as an agent
    mention: a matching agent fires an AI reply into the thread, and on this
    instance ANY mention 500s after the insert (the agent lookup references a
    nonexistent url_key column). So no bridge-emitted text may ever contain a
    bare '@handle' — including the bridge's own needs-you escalation, which is
    emitted zwsp-neutralised and renders identically in the UI.
    """
    return (text or "").replace("@", "@​")


def tag_line(item: dict) -> str:
    bits = [f"#{item.get('kind', 'item')}"]
    if item.get("stage"):
        bits.append(f"#{item['stage']}")
    for t in item.get("tags", []) or []:
        bits.append(f"#{t}")
    if item.get("surface"):
        bits.append(f"surface:{item['surface']}")
    if item.get("project"):
        bits.append(f"project:{item['project']}")
    bits.append(f"ledger:{item['id']}")
    return " · ".join(bits)


def post_body(item: dict) -> str:
    head = {"needs-you": "🙋", "failed": "⛔", "pipeline": "🔧",
            "recommended": "💡", "gap": "🕳"}.get(item.get("kind", ""), "📌")
    parts = [f"{head} {neutralize(item.get('title', '(untitled)'))}"]
    if item.get("detail"):
        parts.append(neutralize(item["detail"]))
    parts.append(tag_line(item))
    return "\n\n".join(parts)


def update_body(rec: dict) -> str:
    lines = []
    kind = rec.get("kind")
    stage = rec.get("stage")
    note = neutralize(rec.get("note", ""))
    if kind == "needs-you":
        lines.append(neutralize(f"@{MENTION}") + " NEEDS YOU: this item is now"
                     " waiting on a human." + (f" — {note}" if note else ""))
        note = ""  # already included
    elif kind == "failed":
        where = f" at {stage}" if stage else ""
        attempt = f", attempt {rec['attempt']}" if rec.get("attempt") else ""
        lines.append(f"⛔ failed{where}{attempt}" + (f" — {note}" if note else ""))
        note = ""
    else:
        if kind:
            lines.append(f"kind → {kind}")
        if stage:
            lines.append(f"stage → {stage}")
        if rec.get("attempt") and not kind:
            lines.append(f"attempt {rec['attempt']}")
    if rec.get("detail"):
        lines.append(f"detail updated: {neutralize(rec['detail'])}")
    if note:
        lines.append(note)
    return "\n".join(lines) or "(updated)"


def close_body(rec: dict) -> str:
    why = neutralize(rec.get("why", ""))
    if rec.get("status") == "dropped":
        return "🗑 dropped" + (f" — {why}" if why else "")
    return "✅ done" + (f" — {why}" if why else "")


# ── Ledger access ────────────────────────────────────────────────────────────

def shre_items(*args: str) -> str:
    out = subprocess.run([SHRE_ITEMS, *args], capture_output=True, text=True,
                        timeout=30)
    if out.returncode != 0:
        raise RuntimeError(f"shre-items {' '.join(args)} rc={out.returncode}: "
                           f"{out.stderr.strip()[:300]}")
    return out.stdout


def feed_since(checkpoint: str) -> list:
    args = ["feed"] + (["--since", checkpoint] if checkpoint else [])
    records = []
    for line in shre_items(*args).splitlines():
        line = line.strip()
        if not line:
            continue
        try:
            records.append(json.loads(line))
        except json.JSONDecodeError:
            continue
    records.sort(key=lambda r: r.get("at", ""))
    return records


def open_items() -> list:
    return json.loads(shre_items("list", "--json") or "[]")


# ── Pulse actions ────────────────────────────────────────────────────────────

def ensure_channel(state: dict) -> str:
    if state.get("channel_id"):
        return state["channel_id"]
    channels = api("GET", f"/api/workspaces/{WORKSPACE_ID}/comms/channels") or []
    for ch in channels:
        if (ch.get("name") or "").lower() == CHANNEL_NAME.lower():
            state["channel_id"] = ch["id"]
            return ch["id"]
    ch = api("POST", f"/api/workspaces/{WORKSPACE_ID}/comms/channels",
             {"name": CHANNEL_NAME, "description": "Workspace activity feed",
              "type": "public"})
    state["channel_id"] = ch["id"]
    return ch["id"]


def send_message(channel_id: str, content: str, thread_id=None) -> dict:
    body = {"content": content, "type": "text", "userName": POSTER_NAME}
    if thread_id:
        body["threadId"] = thread_id
    return api("POST",
               f"/api/workspaces/{WORKSPACE_ID}/comms/channels/{channel_id}/messages",
               body)


def find_existing_post(state: dict, ledger_id: str):
    """Recovery lookup: the target API can insert a row and then fail, so a
    retried create must first check whether a post tagged ledger:<id> already
    exists. Bounded to the most recent 200 channel messages."""
    try:
        msgs = api("GET", f"/api/workspaces/{WORKSPACE_ID}/comms/channels/"
                          f"{ensure_channel(state)}/messages?limit=200") or []
    except Exception:
        return None
    needle = f"ledger:{ledger_id}"
    for m in msgs:
        if not m.get("thread_id") and needle in (m.get("content") or ""):
            return m.get("id")
    return None


def create_post(state: dict, item: dict) -> None:
    if item["id"] in state["map"]:
        return  # already mirrored (seed/feed overlap)
    try:
        msg = send_message(ensure_channel(state), post_body(item))
    except urllib.error.HTTPError:
        # The insert may have landed before the error (proven insert-then-500
        # behaviour). Adopt an existing post instead of duplicating on retry.
        existing = find_existing_post(state, item["id"])
        if existing:
            state["map"][item["id"]] = existing
            log(f"post   {item['id'][:8]} -> adopted existing {existing}")
            return
        raise  # genuinely not created; retry next poll
    state["map"][item["id"]] = msg["id"]
    log(f"post   {item['id'][:8]} -> {msg['id']} ({item.get('title', '')[:60]})")


def comment(state: dict, ledger_id: str, content: str, what: str) -> None:
    post_id = state["map"].get(ledger_id)
    if not post_id:
        log(f"SKIP {what} {ledger_id[:8]}: no mapped post (predates bridge)")
        return
    try:
        msg = send_message(ensure_channel(state), content, thread_id=post_id)
    except urllib.error.HTTPError as e:
        # 500: mib007 comms can fail AFTER the insert (proven mention-path
        # crash) — retrying risks duplicates, so drop. 404/410: the post is
        # gone — retrying can never succeed. Everything else (401/403/429/
        # 502/503) is safe to retry next poll.
        if e.code in (500, 404, 410):
            log(f"DROP {what} {ledger_id[:8]}: HTTP {e.code} "
                "(comment not retried)")
            return
        raise
    log(f"{what:6} {ledger_id[:8]} -> comment {msg['id']}")


# ── Main loop ────────────────────────────────────────────────────────────────

def seed(state: dict) -> None:
    """First run: mirror currently-OPEN items only, then start the op stream
    from 'now'. Historic adds/closes are deliberately not replayed."""
    checkpoint = datetime.now(timezone.utc).isoformat(timespec="seconds")
    items = open_items()
    log(f"seeding {len(items)} open items")
    for item in items:
        try:
            create_post(state, item)
        except Exception as e:  # keep seeding the rest
            log(f"ERROR seeding {item.get('id', '?')[:8]}: {e}")
    state["seeded"] = True
    state["checkpoint"] = checkpoint
    save_state(state)


def process(state: dict, rec: dict) -> None:
    op, lid = rec.get("op"), rec.get("id")
    if not lid or op not in ("add", "update", "close"):
        return
    if op == "add":
        create_post(state, rec)
    elif op == "update":
        comment(state, lid, update_body(rec), "update")
    else:
        comment(state, lid, close_body(rec), "close")


def poll(state: dict) -> None:
    records = feed_since(state["checkpoint"])
    seen = {tuple(s) for s in state.get("seen", [])}
    new = [r for r in records
           if (r.get("id"), r.get("op"), r.get("at")) not in seen]
    for rec in new:
        try:
            process(state, rec)
        except Exception as e:
            log(f"ERROR {rec.get('op')} {str(rec.get('id'))[:8]}: {e}")
            # don't advance past a failed record's timestamp; retry next poll
            break
        seen.add((rec.get("id"), rec.get("op"), rec.get("at")))
        state["checkpoint"] = max(state["checkpoint"], rec.get("at", ""))
        # keep only boundary-second entries for the inclusive-window dedupe
        state["seen"] = [list(t) for t in seen if t[2] >= state["checkpoint"]]
        save_state(state)


def main() -> int:
    log(f"pulse-bridge v0 starting: mib={MIB_BASE} ws={WORKSPACE_ID} "
        f"channel={CHANNEL_NAME} poll={POLL_SEC}s state={STATE_FILE}")
    state = load_state()
    if not state.get("seeded"):
        try:
            seed(state)
        except Exception as e:
            log(f"FATAL seed failed: {e}")
            time.sleep(POLL_SEC)
            return 1  # launchd KeepAlive restarts us
    while True:
        try:
            poll(state)
        except Exception as e:
            log(f"ERROR poll: {e}")
        time.sleep(POLL_SEC)


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