pulse-bridge v0: mirror shre-items ledger into Pulse (mib007 comms) posts

- daemon polls shre-items feed every 30s with inclusive-window (id,op,at) dedupe - op:add -> post in 'activity' channel; op:update -> threaded comment (stage/kind/attempt/note, needs-you escalation mentions @nir); op:close -> done/dropped comment - first run seeds currently-open items only - state (checkpoint + ledger->post map) atomic at ~/.shre/pulse-bridge/state.json - launchd ai.shre.pulse-bridge (KeepAlive, logs to bridge.log) - '@' in ledger-derived text neutralised (zwsp) so comms agent-mention regex can never fire an AI reply from item content Co-Authored-By: Claude Fable 5 <[email protected]> Claude-Session: https://claude.ai/code/session_01YECpkAwQUwgu7NVy91R8fW
2026-08-22 14:41:31 -04:00
commit 9a862bee2b
5 changed files with 447 additions and 0 deletions
@@ -0,0 +1,2 @@
+__pycache__/
+*.pyc
@@ -0,0 +1,83 @@
+# pulse-bridge v0
+
+Mirrors the estate's shared work ledger (`~/.local/bin/shre-items`,
+JSONL store at `~/.shre/open-items/items.jsonl`) into **Pulse** — the
+social-feed surface of the local mib007 instance (`http://127.0.0.1:5520`,
+launchd `ai.shre.mib007`). Ledger items become posts, stage changes become
+threaded comments, needs-you escalations mention the user.
+
+## Where Pulse actually lives (as discovered 2026-08-22)
+
+The Pulse UI is mib007's **Activity** app (`/activity`, `PULSE_ROUTE` in
+`ui/src/platform/lib/app-routes.ts`). Its feed is backed by the **comms**
+tables, not the unified message envelope: the envelope work (`feed_items`
+view, migrations 0101-0111) is in shreai PR #163 and is **not merged** into
+the running mib007 (local checkout is at migration 0100). The
+`/api/workspaces/:id/feed` route proxies to shre-feed (:5436), which is
+**down** locally. So the write path Pulse really uses is:
+
+- create post: `POST /api/workspaces/:wid/comms/channels/:cid/messages`
+  `{content, type: "text", userName}` → returns the row (`id` = post id)
+- comment/reply: same endpoint with `threadId: <post message id>`
+- channel: the Activity app reads/writes the channel named `activity`
+  (creates it if missing) — same behaviour here.
+
+Auth: mib007 service token (`~/.shre/service-tokens.json`, key `mib007`),
+board-level, valid from loopback. mib007 runs in `authenticated` mode, so
+requests without it are rejected.
+
+## Mapping
+
+| ledger op | Pulse action |
+|---|---|
+| `add` | new post: emoji by kind + title, detail, tag line (`#kind #stage #tags surface: project: ledger:<id>`) |
+| `update` | comment on the mapped post: `stage → review`, `⛔ failed at verify, attempt 2 — <note>`, `kind → …` |
+| `update` to `kind: needs-you` | comment `@nir NEEDS YOU: …` |
+| `close` | comment `✅ done — <why>` or `🗑 dropped — <why>` |
+
+First run seeds posts for **currently-open items only**
+(`shre-items list --json`); history before the bridge is not replayed.
+
+## State & config
+
+- State: `~/.shre/pulse-bridge/state.json` — checkpoint (`max at` seen),
+  boundary-second `(id,op,at)` dedupe set (the feed window is inclusive),
+  ledger-id → post-id map, resolved channel id. Written atomically
+  (tmp + `os.replace`).
+- Config: `bridge.env` next to the script (or `PULSE_BRIDGE_ENV`);
+  environment variables override. See the file for keys (poll interval,
+  mib base URL, workspace, channel, mention handle, token file).
+- Poll cadence: every 30 s the daemon runs
+  `shre-items feed --since <checkpoint>` and processes new records in
+  `at` order. A failed record stops the batch **before** the checkpoint
+  advances past it, so it is retried next cycle.
+
+## launchd
+
+`ai.shre.pulse-bridge` — KeepAlive daemon, logs to
+`~/.shre/pulse-bridge/bridge.log`:
+
+```sh
+cp ai.shre.pulse-bridge.plist ~/Library/LaunchAgents/
+launchctl bootstrap gui/$(id -u) ~/Library/LaunchAgents/ai.shre.pulse-bridge.plist
+```
+
+## v0 limitations
+
+- **No real user-mention primitive.** mib007 comms has *agent* @mentions
+  only (the first `@handle` in a message that matches a workspace agent
+  triggers an AI reply). There is no user mention/notification hook, so
+  needs-you escalations are the literal text `@nir NEEDS YOU: …`.
+- **Mention trap defence:** all ledger-derived text has `@` neutralised
+  with a zero-width space so an `@ellie` in an item detail can never
+  trigger an AI reply in the feed. Only the bridge's own `@nir` (which
+  matches no agent, hence inert) is emitted raw.
+- Updates for items that pre-date the bridge and were never seeded
+  (closed before first run) are skipped with a log line — there is no
+  post to comment on.
+- One workspace, one channel. Multi-workspace fan-out is v1.
+- Fork-links (post → ledger deep link and back) and reverse intake
+  (posting in Pulse creating/annotating ledger items) are **v1**.
+- Reactions on posts are not mirrored back to the ledger.
+- If the envelope migration (shreai #163) lands and Pulse moves to
+  `feed_items`, the write path here must be revisited.
@@ -0,0 +1,23 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
+<plist version="1.0">
+<dict>
+  <key>Label</key>
+  <string>ai.shre.pulse-bridge</string>
+  <key>ProgramArguments</key>
+  <array>
+    <string>/usr/bin/python3</string>
+    <string>/Users/aibot/Documents/Projects/pulse-bridge/bridge.py</string>
+  </array>
+  <key>RunAtLoad</key>
+  <true/>
+  <key>KeepAlive</key>
+  <true/>
+  <key>ThrottleInterval</key>
+  <integer>30</integer>
+  <key>StandardOutPath</key>
+  <string>/Users/aibot/.shre/pulse-bridge/bridge.log</string>
+  <key>StandardErrorPath</key>
+  <string>/Users/aibot/.shre/pulse-bridge/bridge.log</string>
+</dict>
+</plist>
@@ -0,0 +1,14 @@
+# pulse-bridge configuration. Environment variables override these.
+PULSE_BRIDGE_POLL_SEC=30
+PULSE_BRIDGE_MIB_BASE=http://127.0.0.1:5520
+# Nirlab Command Center (founder workspace) — where the Pulse/Activity feed lives
+PULSE_BRIDGE_WORKSPACE_ID=293d29db-4978-4c54-a92e-b24d4c0a7115
+PULSE_BRIDGE_CHANNEL=activity
+# handle used in "@nir NEEDS YOU:" escalation comments
+PULSE_BRIDGE_MENTION=nir
+PULSE_BRIDGE_POSTER_NAME=Ledger
+PULSE_BRIDGE_SHRE_ITEMS=/Users/aibot/.local/bin/shre-items
+PULSE_BRIDGE_STATE=/Users/aibot/.shre/pulse-bridge/state.json
+# mib007 service token (board-level, loopback): file + key
+PULSE_BRIDGE_TOKEN_FILE=/Users/aibot/.shre/service-tokens.json
+PULSE_BRIDGE_TOKEN_KEY=mib007
@@ -0,0 +1,325 @@
+#!/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:
+            log(f"WARN state file corrupt, starting fresh: {STATE_FILE}")
+    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")
+    tmp.write_text(json.dumps(state, indent=1), encoding="utf-8")
+    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 and fires an AI reply into the thread (e.g. @ellie). Ledger text
+    must never trigger that; only the bridge's own deliberate mention may.
+    """
+    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(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 create_post(state: dict, item: dict) -> None:
+    if item["id"] in state["map"]:
+        return  # already mirrored (seed/feed overlap)
+    msg = send_message(ensure_channel(state), post_body(item))
+    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
+    msg = send_message(ensure_channel(state), content, thread_id=post_id)
+    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())