From eb0033b952f30fe5aaf5d42a5911dfccc9552518 Mon Sep 17 00:00:00 2001 From: Nirav Patel Date: Sat, 22 Aug 2026 22:21:34 -0400 Subject: [PATCH] migrate write+read path from mib007 comms to the unified message envelope MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pulse's Activity UI (verified live on :5520, Vite dev serving the checkout at 7293d857) now reads AND writes /api/workspaces/:ws/messages, so the bridge follows: - writes: POST /messages with server-side idempotencyKey (ledger: for roots, (id,op,at) for feed comments, pulse-* keys for intake replies); replay returns the ORIGINAL message, so the v0 find-existing recovery scan and drop-on-500 comment policy are gone (5xx now retries safely, only 400/404/410 drop) - reads: intake ingests envelope roots (kind=post), normalized to the old comms row shape at the boundary; comms GET would miss envelope-native user posts entirely - state carries over unchanged: backfill 0103 + mirror trigger 0104 preserve comms ids, so the ledger→post map and channel id stay valid and old threads accept replies (verified via /messages/threads/) - self-recognition: envelope posts carry no userName override, so intake now skips the service actor id (00000000-…-0001) structurally - codex P2 fixed: adopt unmapped v0-era posts (no idempotency key) by ledger: body scan before creating, so an old insert-then-error survivor cannot be duplicated Known cosmetic tradeoff (documented in README): no byline override on the envelope route — new posts render under the service actor, not "Ledger". E2E on the live feed: post 715f52ec, stage comment 74d19219, needs-you @rapidnir comment 2f898b75 + comment.mention notification row for rapidnir-admin, close comment 9f0a5097; timed-out create retried into the same row (idempotency proven, count=1). Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01YECpkAwQUwgu7NVy91R8fW --- README.md | 76 +++++++++-------- bridge.py | 250 +++++++++++++++++++++++++++++++++++++++--------------- 2 files changed, 221 insertions(+), 105 deletions(-) diff --git a/README.md b/README.md index aa28863..e3856cc 100644 --- a/README.md +++ b/README.md @@ -1,4 +1,4 @@ -# pulse-bridge v0 +# pulse-bridge v1 Mirrors the estate's shared work ledger (`~/.local/bin/shre-items`, JSONL store at `~/.shre/open-items/items.jsonl`) into **Pulse** — the @@ -6,25 +6,33 @@ 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) +## Where Pulse actually lives (re-verified 2026-08-23) 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: +`ui/src/platform/lib/app-routes.ts`). Since the unified message envelope +landed (migrations 0101-0111 applied; `feed_items` view live) the Activity +feed **reads AND writes the envelope** (`useActivity.ts` stage 3), so the +bridge does too: -- create post: `POST /api/workspaces/:wid/comms/channels/:cid/messages` - `{content, type: "text", userName}` → returns the row (`id` = post id) +- create post: `POST /api/workspaces/:wid/messages` + `{channelId | channelSlug, body, idempotencyKey}` → returns the message + (`id` = post id; a replayed idempotencyKey returns the ORIGINAL message) - comment/reply: same endpoint with `threadId: ` -- channel: the Activity app reads/writes the channel named `activity` - (creates it if missing) — same behaviour here. +- read roots: `GET /api/workspaces/:wid/messages?channelId=&kind=post` + (root posts only; replies are nested in `blocks.replies`) — the same + query the Activity UI issues +- channel: the envelope channel with slug `activity`. Comms-era channel and + message **ids are preserved** in the envelope (backfill 0103 + mirror + trigger 0104), so v0 state (channel id, ledger→post map) carried over + unchanged and old threads still accept replies. 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. +requests without it are rejected. **Attribution caveat:** the envelope +attributes writes to the authenticated actor and has no `userName` override, +so bridge posts no longer carry the "Ledger" byline — they render under the +service actor (UI fallback "User"). Legacy posts keep their mirrored +`actorDisplayName: "Ledger"`. ## Mapping @@ -100,32 +108,28 @@ checkpoint to the newest existing post — history is never ingested. comment can drop or (once) repeat a comment. Ledger and kanban stay consistent — `shre-items done` on a closed item is a logged no-op and `--idempotency-key pulse-` adopts the existing task on retry. -- Bridge-output filtering is content-based (user name, `🤖` prefix, - `ledger:` tag line, threaded); renaming the poster or ignore list while - old posts are inside the intake window could ingest bridge output. The - comms API has no structured message metadata to mark authorship. +- Bridge-output filtering is structural first (envelope-native bridge posts + carry the service actor id `00000000-…-0001`, agents carry + `actorType: "agent"`), then content-based for legacy posts (display name, + `🤖` prefix, `ledger:` tag line, threaded). -## v0/v1 limitations +## 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 path is broken on this instance (found during smoke):** ANY - `@handle` in a message makes `POST …/messages` **500 after the row is - inserted** — the mention agent-lookup queries a nonexistent `url_key` - column on `agents` (`comms.ts` ~line 226; error in mib007 stderr log). - Retrying such a failure duplicates the message. The bridge therefore - (a) zwsp-neutralises **every** `@` it emits, including its own - `@nir` escalation (renders identically in the UI), and (b) treats an - HTTP error on a *comment* as non-retryable (`DROP … comment not - retried` in the log) — comments are best-effort; posts still retry. +- **Mentions dispatch server-side** — the envelope route runs the same + `dispatchMention` side-effect comms did: the first `@handle` matching a + workspace agent fires an AI reply; a matching workspace member gets a + `comment.mention` notification. The bridge zwsp-neutralises every `@` in + ledger-derived text and deliberately keeps the needs-you `@rapidnir` + escalation live so the human is actually notified. The historical + insert-then-500 mention crash is fixed AND harmless now: writes carry + idempotency keys, so a retry returns the original message instead of + duplicating it (`DROP` now only happens on 400/404/410 — thread target + gone). +- **No byline override.** See the attribution caveat above — posts render + under the service actor, not "Ledger". Needs a mib007-side + `actorDisplayName` passthrough for service tokens to restore. - 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**. +- One workspace, one channel. Multi-workspace fan-out is future work. - 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/bridge.py b/bridge.py index e325093..c1fd237 100644 --- a/bridge.py +++ b/bridge.py @@ -1,10 +1,27 @@ #!/usr/bin/env python3 -"""pulse-bridge v0 — mirror the estate's shared work ledger into Pulse posts. +"""pulse-bridge v1 — 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. +Target: local mib007 (:5520) unified message ENVELOPE API (/api/workspaces/ +:ws/messages) — the store the Pulse / Activity feed now reads AND writes +(useActivity.ts stage 3). One ledger item = one post in the "activity" +channel; stage changes / failures / escalations = threaded comments; close = +a final comment. + +v0 wrote to the legacy comms channels API. The migration was seamless for +state because mib007's backfill (0103) and mirror trigger (0104) PRESERVE +message ids comms -> envelope, so the existing ledger-id -> post-id map keeps +pointing at valid envelope messages and old threads keep accepting replies. + +Two envelope-specific notes: +- Idempotency is server-side now: every write carries an idempotencyKey and a + replay returns the ORIGINAL message (201, same id), so retries after + insert-then-crash can neither duplicate posts nor comments. The v0 + find-existing recovery scan (bounded page, can't prove absence) is gone. +- Attribution comes from the AUTHENTICATED actor, never the body. The comms + `userName: "Ledger"` byline override does not exist on the envelope route, + so envelope-native bridge posts render under the service actor (the UI + falls back to "User"). Known cosmetic tradeoff, documented in README. Stdlib only. State (checkpoint + ledger-id -> post-id map) lives in ~/.shre/pulse-bridge/state.json, written atomically (tmp + rename). @@ -53,6 +70,10 @@ 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 +# Historical byline. The envelope route has no userName override (attribution +# is the authenticated actor), so this no longer names the posts; it is kept +# because the intake-ignore default below references it and legacy posts +# carry it as actorDisplayName. 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", @@ -60,6 +81,11 @@ STATE_FILE = Path(cfg("PULSE_BRIDGE_STATE", TOKEN_FILE = Path(cfg("PULSE_BRIDGE_TOKEN_FILE", str(HOME / ".shre/service-tokens.json"))) TOKEN_KEY = cfg("PULSE_BRIDGE_TOKEN_KEY", "mib007") +# The envelope attributes writes to the AUTHENTICATED actor. The mib007 +# service token authenticates as this fixed board actor (middleware/auth.ts), +# which is how the bridge recognises (and never re-ingests) its own posts. +SELF_ACTOR_ID = cfg("PULSE_BRIDGE_SELF_ACTOR_ID", + "00000000-0000-0000-0000-000000000001") STAGES = ["queued", "build", "review", "merge", "deploy", "verify"] @@ -136,15 +162,14 @@ def save_state(state: dict) -> None: 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 a mention: - a matching agent fires an AI reply into the thread, and a matching - workspace-member user gets a notification. The historical crash (mention - lookup on a nonexistent agents.url_key column 500'd AFTER the insert) is - FIXED — Nirlabinc/mib007 PR #32 — so mentions are safe to send. We still - neutralize ledger-DERIVED text (titles, details, notes) because arbitrary - item text containing "@aros"/"@ellie" must not accidentally fire an agent - reply into the feed. The bridge's own needs-you escalation line is the one - place that deliberately carries a live @handle (see update_body). + mib007 treats the FIRST '@handle' in a message as a mention — the envelope + route runs the SAME dispatchMention side-effect the comms route did: a + matching agent fires an AI reply into the thread, and a matching + workspace-member user gets a notification. We neutralize ledger-DERIVED + text (titles, details, notes) because arbitrary item text containing + "@aros"/"@ellie" must not accidentally fire an agent reply into the feed. + The bridge's own needs-you escalation line is the one place that + deliberately carries a live @handle (see update_body). """ return (text or "").replace("@", "@​") @@ -242,81 +267,112 @@ def open_items() -> list: return json.loads(shre_items("list", "--json") or "[]") -# ── Pulse actions ──────────────────────────────────────────────────────────── +# ── Pulse actions (unified message envelope) ───────────────────────────────── -def ensure_channel(state: dict) -> str: +def ensure_channel(state: dict): + """Resolve the activity channel's envelope id (cached in state). + + Comms and envelope channel ids are IDENTICAL for pre-envelope channels + (backfill 0103 preserves ids), so a v0 state file's cached channel_id is + already the envelope id. A missing channel is not created here — + send_message() falls back to channelSlug, which the envelope route + find-or-creates, and the id is adopted from the response. + """ if state.get("channel_id"): return state["channel_id"] - channels = api("GET", f"/api/workspaces/{WORKSPACE_ID}/comms/channels") or [] + channels = api("GET", + f"/api/workspaces/{WORKSPACE_ID}/messages/channels") or [] + want = CHANNEL_NAME.lower() for ch in channels: - if (ch.get("name") or "").lower() == CHANNEL_NAME.lower(): + if (ch.get("slug") or "").lower() == want \ + or (ch.get("name") or "").lower() == want: state["channel_id"] = ch["id"] + save_state(state) 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"] + return None -def send_message(channel_id: str, content: str, thread_id=None) -> dict: - body = {"content": content, "type": "text", "userName": POSTER_NAME} +def send_message(state: dict, content: str, thread_id=None, idem=None) -> dict: + """POST one message into the envelope. + + idem is the server-side idempotency key: a replay after a crash or an + insert-then-error returns the ORIGINAL message with its original id, so + every caller can safely retry. Attribution is the authenticated actor; + there is no userName override on this route. + """ + body = {"body": content} + channel_id = ensure_channel(state) + if channel_id: + body["channelId"] = channel_id + else: + body["channelSlug"] = CHANNEL_NAME # find-or-create, id adopted below if thread_id: body["threadId"] = thread_id - return api("POST", - f"/api/workspaces/{WORKSPACE_ID}/comms/channels/{channel_id}/messages", - body) + if idem: + body["idempotencyKey"] = idem + # The envelope route runs side-effects (audit, mention dispatch) before + # responding and can exceed the default 15s under load; a timeout here is + # retried next poll and the idempotency key de-duplicates the replay. + msg = api("POST", f"/api/workspaces/{WORKSPACE_ID}/messages", body, + timeout=30) + if not state.get("channel_id") and msg.get("channelId"): + state["channel_id"] = msg["channelId"] + save_state(state) + return msg -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: 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 +def find_legacy_post(state: dict, ledger_id: str): + """Adopt an unmapped v0-era post before creating (codex-flagged). + + v0 wrote via comms WITHOUT idempotency keys, so a v0 insert-then-error + crash could have left a mirrored post that the map never recorded — the + ledger: idempotency key cannot replay/adopt such a row, and blindly + creating would duplicate it. Bounded to the latest 100 roots: a miss + cannot prove absence (documented trap), but a miss merely creates — and + every v1 row carries the idempotency key, so v1-era retries never reach + this path with a duplicate risk. + """ 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") + try: + for m in channel_roots(state): + if needle in (m.get("content") or ""): + return m["id"] + except Exception: + return None # recovery is best-effort; creation still idempotent 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 + legacy = find_legacy_post(state, item["id"]) + if legacy: + state["map"][item["id"]] = legacy + log(f"post {item['id'][:8]} -> adopted legacy {legacy}") + return + # ledger: as the idempotency key: a retry after insert-then-crash gets + # the original post back (same id), so no duplicate and no recovery scan. + msg = send_message(state, post_body(item), idem=f"ledger:{item['id']}") 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: +def comment(state: dict, ledger_id: str, content: str, what: str, + idem=None) -> 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) + msg = send_message(state, content, thread_id=post_id, idem=idem) 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): + # 400 here means the thread target is invalid (deleted/foreign post) — + # retrying can never succeed, same as 404/410. 5xx IS retried now: + # the idempotency key makes a replay return the original comment, so + # the v0 drop-on-500 duplicate guard is no longer needed. + if e.code in (400, 404, 410): log(f"DROP {what} {ledger_id[:8]}: HTTP {e.code} " - "(comment not retried)") + "(comment not retryable)") return raise log(f"{what:6} {ledger_id[:8]} -> comment {msg['id']}") @@ -385,17 +441,62 @@ def intake_start_task(state: dict, post_id: str, entry: dict) -> None: save_state(state) log(f"intake {post_id[:8]} -> kanban {task_id}") try: - send_message(ensure_channel(state), + send_message(state, f"🤖 picked up — ledger {entry['ledger_id'][:8]}, " - f"kanban {task_id}", thread_id=post_id) + f"kanban {task_id}", thread_id=post_id, + idem=f"pulse-ack:{post_id}") except urllib.error.HTTPError as e: log(f"DROP ack {post_id[:8]}: HTTP {e.code}") +def _created_ms(iso: str) -> int: + """Envelope timestamps are ISO-8601; the intake checkpoint stays epoch-ms + so a v0 state file's checkpoint keeps meaning the same instant.""" + if not iso: + return 0 + try: + return int(datetime.fromisoformat( + iso.replace("Z", "+00:00")).timestamp() * 1000) + except ValueError: + return 0 + + +def channel_roots(state: dict) -> list: + """Read the activity channel's root posts from the envelope, normalized to + the comms row shape the intake logic was written against (id / content / + user_id / user_name / created_at-ms / thread_id=None). + + This read HAD to move with the writes: the Pulse UI now posts envelope- + native messages, which never appear in the legacy comms GET — reading + comms would silently stop ingesting exactly the posts users make. + kind=post returns ROOT posts only (replies are nested), which is what the + intake wants anyway. + """ + channel_id = ensure_channel(state) + if not channel_id: + return [] # no activity channel yet -> nothing to ingest + data = api("GET", f"/api/workspaces/{WORKSPACE_ID}/messages" + f"?channelId={channel_id}&kind=post&limit=100") or {} + rows = [] + for m in data.get("items") or []: + rows.append({ + "id": m.get("sourceId"), + "content": m.get("body") or "", + # comms encoded agents as user_id "agent:"; the envelope keeps + # actorType separately — reconstruct the prefix the filters test. + "user_id": (f"agent:{m.get('actorId')}" + if m.get("actorType") == "agent" else m.get("actorId")), + "user_name": m.get("actorDisplayName"), + "created_at": _created_ms(m.get("createdAt") or ""), + "thread_id": None, + "actor_id": m.get("actorId"), + }) + return rows + + def intake_ingest(state: dict) -> None: """Turn new non-bridge top-level activity posts into ledger items + tasks.""" - msgs = api("GET", f"/api/workspaces/{WORKSPACE_ID}/comms/channels/" - f"{ensure_channel(state)}/messages?limit=100") or [] + msgs = channel_roots(state) cp = int(state.get("intake_checkpoint") or 0) if cp == 0: # First intake run: start strictly after the newest existing post; @@ -412,6 +513,10 @@ def intake_ingest(state: dict) -> None: ts = int(m.get("created_at") or 0) content = (m.get("content") or "").strip() if (ts < cp or m.get("thread_id") or m["id"] in state["intake"] + # Envelope-native bridge posts carry NO display name (no + # userName override on the envelope route) — the service + # actor id is how the bridge recognises its own posts now. + or m.get("actor_id") == SELF_ACTOR_ID or (m.get("user_name") or "").lower() in INTAKE_IGNORE_USERS or (m.get("user_id") or "").startswith("agent:") or content.startswith("🤖") or "ledger:" in content @@ -439,9 +544,9 @@ def intake_ingest(state: dict) -> None: n = intake_active_count(state) log(f"intake {post['id'][:8]} queued behind {n} tasks") try: - send_message(ensure_channel(state), - f"⏳ queued behind {n} tasks", - thread_id=post["id"]) + send_message(state, f"⏳ queued behind {n} tasks", + thread_id=post["id"], + idem=f"pulse-queued:{post['id']}") except urllib.error.HTTPError as e: log(f"DROP queue-note {post['id'][:8]}: HTTP {e.code}") @@ -487,7 +592,11 @@ def intake_track(state: dict) -> None: save_state(state) log(f"track {entry['task_id']} {status} (post {post_id[:8]})") try: - send_message(ensure_channel(state), body, thread_id=post_id) + # Keyed per task+status: re-entering a status (blocked -> running + # -> blocked) dedupes to the first comment, which is the right + # noise level for a status thread. + send_message(state, body, thread_id=post_id, + idem=f"pulse-status:{entry['task_id']}:{status}") except urllib.error.HTTPError as e: log(f"DROP track {post_id[:8]}: HTTP {e.code}") @@ -531,10 +640,13 @@ def process(state: dict, rec: dict) -> None: # mirroring their ledger ops too would double-comment the user's post. if any(e.get("ledger_id") == lid for e in state.get("intake", {}).values()): return + # (id, op, at) is exactly the tuple the seen-set dedupes on, so making it + # the idempotency key gives server-side retries the same identity rule. + idem = f"ledger:{lid}:{op}:{rec.get('at', '')}" if op == "update": - comment(state, lid, update_body(rec), "update") + comment(state, lid, update_body(rec), "update", idem=idem) else: - comment(state, lid, close_body(rec), "close") + comment(state, lid, close_body(rec), "close", idem=idem) def poll(state: dict) -> None: @@ -557,7 +669,7 @@ def poll(state: dict) -> None: def main() -> int: - log(f"pulse-bridge v0 starting: mib={MIB_BASE} ws={WORKSPACE_ID} " + log(f"pulse-bridge v1 (envelope) 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"):