migrate write+read path from mib007 comms to the unified message envelope
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:<id> 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/<old-id>) - 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:<id> 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 <[email protected]> Claude-Session: https://claude.ai/code/session_01YECpkAwQUwgu7NVy91R8fW
This commit is contained in:
co-authored by
Claude Fable 5
parent
f168ad6f4c
commit
eb0033b952
@@ -1,4 +1,4 @@
|
|||||||
# pulse-bridge v0
|
# pulse-bridge v1
|
||||||
|
|
||||||
Mirrors the estate's shared work ledger (`~/.local/bin/shre-items`,
|
Mirrors the estate's shared work ledger (`~/.local/bin/shre-items`,
|
||||||
JSONL store at `~/.shre/open-items/items.jsonl`) into **Pulse** — the
|
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
|
launchd `ai.shre.mib007`). Ledger items become posts, stage changes become
|
||||||
threaded comments, needs-you escalations mention the user.
|
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
|
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**
|
`ui/src/platform/lib/app-routes.ts`). Since the unified message envelope
|
||||||
tables, not the unified message envelope: the envelope work (`feed_items`
|
landed (migrations 0101-0111 applied; `feed_items` view live) the Activity
|
||||||
view, migrations 0101-0111) is in shreai PR #163 and is **not merged** into
|
feed **reads AND writes the envelope** (`useActivity.ts` stage 3), so the
|
||||||
the running mib007 (local checkout is at migration 0100). The
|
bridge does too:
|
||||||
`/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`
|
- create post: `POST /api/workspaces/:wid/messages`
|
||||||
`{content, type: "text", userName}` → returns the row (`id` = post id)
|
`{channelId | channelSlug, body, idempotencyKey}` → returns the message
|
||||||
|
(`id` = post id; a replayed idempotencyKey returns the ORIGINAL message)
|
||||||
- comment/reply: same endpoint with `threadId: <post message id>`
|
- comment/reply: same endpoint with `threadId: <post message id>`
|
||||||
- channel: the Activity app reads/writes the channel named `activity`
|
- read roots: `GET /api/workspaces/:wid/messages?channelId=&kind=post`
|
||||||
(creates it if missing) — same behaviour here.
|
(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`),
|
Auth: mib007 service token (`~/.shre/service-tokens.json`, key `mib007`),
|
||||||
board-level, valid from loopback. mib007 runs in `authenticated` mode, so
|
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
|
## 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
|
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
|
consistent — `shre-items done` on a closed item is a logged no-op and
|
||||||
`--idempotency-key pulse-<post id>` adopts the existing task on retry.
|
`--idempotency-key pulse-<post id>` adopts the existing task on retry.
|
||||||
- Bridge-output filtering is content-based (user name, `🤖` prefix,
|
- Bridge-output filtering is structural first (envelope-native bridge posts
|
||||||
`ledger:` tag line, threaded); renaming the poster or ignore list while
|
carry the service actor id `00000000-…-0001`, agents carry
|
||||||
old posts are inside the intake window could ingest bridge output. The
|
`actorType: "agent"`), then content-based for legacy posts (display name,
|
||||||
comms API has no structured message metadata to mark authorship.
|
`🤖` prefix, `ledger:` tag line, threaded).
|
||||||
|
|
||||||
## v0/v1 limitations
|
## Limitations
|
||||||
|
|
||||||
- **No real user-mention primitive.** mib007 comms has *agent* @mentions
|
- **Mentions dispatch server-side** — the envelope route runs the same
|
||||||
only (the first `@handle` in a message that matches a workspace agent
|
`dispatchMention` side-effect comms did: the first `@handle` matching a
|
||||||
triggers an AI reply). There is no user mention/notification hook, so
|
workspace agent fires an AI reply; a matching workspace member gets a
|
||||||
needs-you escalations are the literal text `@nir NEEDS YOU: …`.
|
`comment.mention` notification. The bridge zwsp-neutralises every `@` in
|
||||||
- **Mention path is broken on this instance (found during smoke):** ANY
|
ledger-derived text and deliberately keeps the needs-you `@rapidnir`
|
||||||
`@handle` in a message makes `POST …/messages` **500 after the row is
|
escalation live so the human is actually notified. The historical
|
||||||
inserted** — the mention agent-lookup queries a nonexistent `url_key`
|
insert-then-500 mention crash is fixed AND harmless now: writes carry
|
||||||
column on `agents` (`comms.ts` ~line 226; error in mib007 stderr log).
|
idempotency keys, so a retry returns the original message instead of
|
||||||
Retrying such a failure duplicates the message. The bridge therefore
|
duplicating it (`DROP` now only happens on 400/404/410 — thread target
|
||||||
(a) zwsp-neutralises **every** `@` it emits, including its own
|
gone).
|
||||||
`@nir` escalation (renders identically in the UI), and (b) treats an
|
- **No byline override.** See the attribution caveat above — posts render
|
||||||
HTTP error on a *comment* as non-retryable (`DROP … comment not
|
under the service actor, not "Ledger". Needs a mib007-side
|
||||||
retried` in the log) — comments are best-effort; posts still retry.
|
`actorDisplayName` passthrough for service tokens to restore.
|
||||||
- Updates for items that pre-date the bridge and were never seeded
|
- 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
|
(closed before first run) are skipped with a log line — there is no
|
||||||
post to comment on.
|
post to comment on.
|
||||||
- One workspace, one channel. Multi-workspace fan-out is v1.
|
- One workspace, one channel. Multi-workspace fan-out is future work.
|
||||||
- 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.
|
- 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.
|
|
||||||
|
|||||||
@@ -1,10 +1,27 @@
|
|||||||
#!/usr/bin/env python3
|
#!/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).
|
Source: ~/.local/bin/shre-items (JSONL ledger, `feed --since` op stream).
|
||||||
Target: local mib007 (:5520) comms API — the backing store of the Pulse /
|
Target: local mib007 (:5520) unified message ENVELOPE API (/api/workspaces/
|
||||||
Activity feed. One ledger item = one post in the "activity" channel; stage
|
:ws/messages) — the store the Pulse / Activity feed now reads AND writes
|
||||||
changes / failures / escalations = threaded comments; close = final comment.
|
(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
|
Stdlib only. State (checkpoint + ledger-id -> post-id map) lives in
|
||||||
~/.shre/pulse-bridge/state.json, written atomically (tmp + rename).
|
~/.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
|
"293d29db-4978-4c54-a92e-b24d4c0a7115") # Nirlab Command Center
|
||||||
CHANNEL_NAME = cfg("PULSE_BRIDGE_CHANNEL", "activity")
|
CHANNEL_NAME = cfg("PULSE_BRIDGE_CHANNEL", "activity")
|
||||||
MENTION = cfg("PULSE_BRIDGE_MENTION", "nir") # user handle to @mention
|
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")
|
POSTER_NAME = cfg("PULSE_BRIDGE_POSTER_NAME", "Ledger")
|
||||||
SHRE_ITEMS = cfg("PULSE_BRIDGE_SHRE_ITEMS", str(HOME / ".local/bin/shre-items"))
|
SHRE_ITEMS = cfg("PULSE_BRIDGE_SHRE_ITEMS", str(HOME / ".local/bin/shre-items"))
|
||||||
STATE_FILE = Path(cfg("PULSE_BRIDGE_STATE",
|
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",
|
TOKEN_FILE = Path(cfg("PULSE_BRIDGE_TOKEN_FILE",
|
||||||
str(HOME / ".shre/service-tokens.json")))
|
str(HOME / ".shre/service-tokens.json")))
|
||||||
TOKEN_KEY = cfg("PULSE_BRIDGE_TOKEN_KEY", "mib007")
|
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"]
|
STAGES = ["queued", "build", "review", "merge", "deploy", "verify"]
|
||||||
|
|
||||||
@@ -136,15 +162,14 @@ def save_state(state: dict) -> None:
|
|||||||
def neutralize(text: str) -> str:
|
def neutralize(text: str) -> str:
|
||||||
"""Insert a zero-width space after '@' in ledger-derived text.
|
"""Insert a zero-width space after '@' in ledger-derived text.
|
||||||
|
|
||||||
mib007's comms route treats the FIRST '@handle' in a message as a mention:
|
mib007 treats the FIRST '@handle' in a message as a mention — the envelope
|
||||||
a matching agent fires an AI reply into the thread, and a matching
|
route runs the SAME dispatchMention side-effect the comms route did: a
|
||||||
workspace-member user gets a notification. The historical crash (mention
|
matching agent fires an AI reply into the thread, and a matching
|
||||||
lookup on a nonexistent agents.url_key column 500'd AFTER the insert) is
|
workspace-member user gets a notification. We neutralize ledger-DERIVED
|
||||||
FIXED — Nirlabinc/mib007 PR #32 — so mentions are safe to send. We still
|
text (titles, details, notes) because arbitrary item text containing
|
||||||
neutralize ledger-DERIVED text (titles, details, notes) because arbitrary
|
"@aros"/"@ellie" must not accidentally fire an agent reply into the feed.
|
||||||
item text containing "@aros"/"@ellie" must not accidentally fire an agent
|
The bridge's own needs-you escalation line is the one place that
|
||||||
reply into the feed. The bridge's own needs-you escalation line is the one
|
deliberately carries a live @handle (see update_body).
|
||||||
place that deliberately carries a live @handle (see update_body).
|
|
||||||
"""
|
"""
|
||||||
return (text or "").replace("@", "@")
|
return (text or "").replace("@", "@")
|
||||||
|
|
||||||
@@ -242,81 +267,112 @@ def open_items() -> list:
|
|||||||
return json.loads(shre_items("list", "--json") or "[]")
|
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"):
|
if state.get("channel_id"):
|
||||||
return state["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:
|
for ch in channels:
|
||||||
if (ch.get("name") or "").lower() == CHANNEL_NAME.lower():
|
if (ch.get("slug") or "").lower() == want \
|
||||||
state["channel_id"] = ch["id"]
|
or (ch.get("name") or "").lower() == want:
|
||||||
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"]
|
state["channel_id"] = ch["id"]
|
||||||
|
save_state(state)
|
||||||
return ch["id"]
|
return ch["id"]
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
def send_message(channel_id: str, content: str, thread_id=None) -> dict:
|
def send_message(state: dict, content: str, thread_id=None, idem=None) -> dict:
|
||||||
body = {"content": content, "type": "text", "userName": POSTER_NAME}
|
"""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:
|
if thread_id:
|
||||||
body["threadId"] = thread_id
|
body["threadId"] = thread_id
|
||||||
return api("POST",
|
if idem:
|
||||||
f"/api/workspaces/{WORKSPACE_ID}/comms/channels/{channel_id}/messages",
|
body["idempotencyKey"] = idem
|
||||||
body)
|
# 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):
|
def find_legacy_post(state: dict, ledger_id: str):
|
||||||
"""Recovery lookup: the target API can insert a row and then fail, so a
|
"""Adopt an unmapped v0-era post before creating (codex-flagged).
|
||||||
retried create must first check whether a post tagged ledger:<id> already
|
|
||||||
exists. Bounded to the most recent 200 channel messages."""
|
v0 wrote via comms WITHOUT idempotency keys, so a v0 insert-then-error
|
||||||
try:
|
crash could have left a mirrored post that the map never recorded — the
|
||||||
msgs = api("GET", f"/api/workspaces/{WORKSPACE_ID}/comms/channels/"
|
ledger:<id> idempotency key cannot replay/adopt such a row, and blindly
|
||||||
f"{ensure_channel(state)}/messages?limit=200") or []
|
creating would duplicate it. Bounded to the latest 100 roots: a miss
|
||||||
except Exception:
|
cannot prove absence (documented trap), but a miss merely creates — and
|
||||||
return None
|
every v1 row carries the idempotency key, so v1-era retries never reach
|
||||||
|
this path with a duplicate risk.
|
||||||
|
"""
|
||||||
needle = f"ledger:{ledger_id}"
|
needle = f"ledger:{ledger_id}"
|
||||||
for m in msgs:
|
try:
|
||||||
if not m.get("thread_id") and needle in (m.get("content") or ""):
|
for m in channel_roots(state):
|
||||||
return m.get("id")
|
if needle in (m.get("content") or ""):
|
||||||
|
return m["id"]
|
||||||
|
except Exception:
|
||||||
|
return None # recovery is best-effort; creation still idempotent
|
||||||
return None
|
return None
|
||||||
|
|
||||||
|
|
||||||
def create_post(state: dict, item: dict) -> None:
|
def create_post(state: dict, item: dict) -> None:
|
||||||
if item["id"] in state["map"]:
|
if item["id"] in state["map"]:
|
||||||
return # already mirrored (seed/feed overlap)
|
return # already mirrored (seed/feed overlap)
|
||||||
try:
|
legacy = find_legacy_post(state, item["id"])
|
||||||
msg = send_message(ensure_channel(state), post_body(item))
|
if legacy:
|
||||||
except urllib.error.HTTPError:
|
state["map"][item["id"]] = legacy
|
||||||
# The insert may have landed before the error (proven insert-then-500
|
log(f"post {item['id'][:8]} -> adopted legacy {legacy}")
|
||||||
# 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
|
return
|
||||||
raise # genuinely not created; retry next poll
|
# ledger:<id> 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"]
|
state["map"][item["id"]] = msg["id"]
|
||||||
log(f"post {item['id'][:8]} -> {msg['id']} ({item.get('title', '')[:60]})")
|
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)
|
post_id = state["map"].get(ledger_id)
|
||||||
if not post_id:
|
if not post_id:
|
||||||
log(f"SKIP {what} {ledger_id[:8]}: no mapped post (predates bridge)")
|
log(f"SKIP {what} {ledger_id[:8]}: no mapped post (predates bridge)")
|
||||||
return
|
return
|
||||||
try:
|
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:
|
except urllib.error.HTTPError as e:
|
||||||
# 500: mib007 comms can fail AFTER the insert (proven mention-path
|
# 400 here means the thread target is invalid (deleted/foreign post) —
|
||||||
# crash) — retrying risks duplicates, so drop. 404/410: the post is
|
# retrying can never succeed, same as 404/410. 5xx IS retried now:
|
||||||
# gone — retrying can never succeed. Everything else (401/403/429/
|
# the idempotency key makes a replay return the original comment, so
|
||||||
# 502/503) is safe to retry next poll.
|
# the v0 drop-on-500 duplicate guard is no longer needed.
|
||||||
if e.code in (500, 404, 410):
|
if e.code in (400, 404, 410):
|
||||||
log(f"DROP {what} {ledger_id[:8]}: HTTP {e.code} "
|
log(f"DROP {what} {ledger_id[:8]}: HTTP {e.code} "
|
||||||
"(comment not retried)")
|
"(comment not retryable)")
|
||||||
return
|
return
|
||||||
raise
|
raise
|
||||||
log(f"{what:6} {ledger_id[:8]} -> comment {msg['id']}")
|
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)
|
save_state(state)
|
||||||
log(f"intake {post_id[:8]} -> kanban {task_id}")
|
log(f"intake {post_id[:8]} -> kanban {task_id}")
|
||||||
try:
|
try:
|
||||||
send_message(ensure_channel(state),
|
send_message(state,
|
||||||
f"🤖 picked up — ledger {entry['ledger_id'][:8]}, "
|
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:
|
except urllib.error.HTTPError as e:
|
||||||
log(f"DROP ack {post_id[:8]}: HTTP {e.code}")
|
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:<id>"; 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:
|
def intake_ingest(state: dict) -> None:
|
||||||
"""Turn new non-bridge top-level activity posts into ledger items + tasks."""
|
"""Turn new non-bridge top-level activity posts into ledger items + tasks."""
|
||||||
msgs = api("GET", f"/api/workspaces/{WORKSPACE_ID}/comms/channels/"
|
msgs = channel_roots(state)
|
||||||
f"{ensure_channel(state)}/messages?limit=100") or []
|
|
||||||
cp = int(state.get("intake_checkpoint") or 0)
|
cp = int(state.get("intake_checkpoint") or 0)
|
||||||
if cp == 0:
|
if cp == 0:
|
||||||
# First intake run: start strictly after the newest existing post;
|
# 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)
|
ts = int(m.get("created_at") or 0)
|
||||||
content = (m.get("content") or "").strip()
|
content = (m.get("content") or "").strip()
|
||||||
if (ts < cp or m.get("thread_id") or m["id"] in state["intake"]
|
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_name") or "").lower() in INTAKE_IGNORE_USERS
|
||||||
or (m.get("user_id") or "").startswith("agent:")
|
or (m.get("user_id") or "").startswith("agent:")
|
||||||
or content.startswith("🤖") or "ledger:" in content
|
or content.startswith("🤖") or "ledger:" in content
|
||||||
@@ -439,9 +544,9 @@ def intake_ingest(state: dict) -> None:
|
|||||||
n = intake_active_count(state)
|
n = intake_active_count(state)
|
||||||
log(f"intake {post['id'][:8]} queued behind {n} tasks")
|
log(f"intake {post['id'][:8]} queued behind {n} tasks")
|
||||||
try:
|
try:
|
||||||
send_message(ensure_channel(state),
|
send_message(state, f"⏳ queued behind {n} tasks",
|
||||||
f"⏳ queued behind {n} tasks",
|
thread_id=post["id"],
|
||||||
thread_id=post["id"])
|
idem=f"pulse-queued:{post['id']}")
|
||||||
except urllib.error.HTTPError as e:
|
except urllib.error.HTTPError as e:
|
||||||
log(f"DROP queue-note {post['id'][:8]}: HTTP {e.code}")
|
log(f"DROP queue-note {post['id'][:8]}: HTTP {e.code}")
|
||||||
|
|
||||||
@@ -487,7 +592,11 @@ def intake_track(state: dict) -> None:
|
|||||||
save_state(state)
|
save_state(state)
|
||||||
log(f"track {entry['task_id']} {status} (post {post_id[:8]})")
|
log(f"track {entry['task_id']} {status} (post {post_id[:8]})")
|
||||||
try:
|
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:
|
except urllib.error.HTTPError as e:
|
||||||
log(f"DROP track {post_id[:8]}: HTTP {e.code}")
|
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.
|
# 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()):
|
if any(e.get("ledger_id") == lid for e in state.get("intake", {}).values()):
|
||||||
return
|
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":
|
if op == "update":
|
||||||
comment(state, lid, update_body(rec), "update")
|
comment(state, lid, update_body(rec), "update", idem=idem)
|
||||||
else:
|
else:
|
||||||
comment(state, lid, close_body(rec), "close")
|
comment(state, lid, close_body(rec), "close", idem=idem)
|
||||||
|
|
||||||
|
|
||||||
def poll(state: dict) -> None:
|
def poll(state: dict) -> None:
|
||||||
@@ -557,7 +669,7 @@ def poll(state: dict) -> None:
|
|||||||
|
|
||||||
|
|
||||||
def main() -> int:
|
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}")
|
f"channel={CHANNEL_NAME} poll={POLL_SEC}s state={STATE_FILE}")
|
||||||
state = load_state()
|
state = load_state()
|
||||||
if not state.get("seeded"):
|
if not state.get("seeded"):
|
||||||
|
|||||||
Reference in New Issue
Block a user