commit 9a862bee2b6d73886c5f340d02df965d5332fd97 Author: Nirav Patel Date: Sat Aug 22 14:41:31 2026 -0400 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 Claude-Session: https://claude.ai/code/session_01YECpkAwQUwgu7NVy91R8fW diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..7a60b85 --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +__pycache__/ +*.pyc diff --git a/README.md b/README.md new file mode 100644 index 0000000..df30c74 --- /dev/null +++ b/README.md @@ -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: ` +- 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:`) | +| `update` | comment on the mapped post: `stage → review`, `⛔ failed at verify, attempt 2 — `, `kind → …` | +| `update` to `kind: needs-you` | comment `@nir NEEDS YOU: …` | +| `close` | comment `✅ done — ` or `🗑 dropped — ` | + +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 ` 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. diff --git a/ai.shre.pulse-bridge.plist b/ai.shre.pulse-bridge.plist new file mode 100644 index 0000000..4f4d367 --- /dev/null +++ b/ai.shre.pulse-bridge.plist @@ -0,0 +1,23 @@ + + + + + Label + ai.shre.pulse-bridge + ProgramArguments + + /usr/bin/python3 + /Users/aibot/Documents/Projects/pulse-bridge/bridge.py + + RunAtLoad + + KeepAlive + + ThrottleInterval + 30 + StandardOutPath + /Users/aibot/.shre/pulse-bridge/bridge.log + StandardErrorPath + /Users/aibot/.shre/pulse-bridge/bridge.log + + diff --git a/bridge.env b/bridge.env new file mode 100644 index 0000000..ffd520f --- /dev/null +++ b/bridge.env @@ -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 diff --git a/bridge.py b/bridge.py new file mode 100644 index 0000000..dc73fb9 --- /dev/null +++ b/bridge.py @@ -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())