pulse-bridge/bridge.py

#!/usr/bin/env python3
"""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) 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).
"""

import json
import os
import re
import shutil
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
# 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",
                      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")
# 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"]

KIND_LABELS = {
    "needs-you": "human needed",
    "failed": "gate failed",
    "pipeline": "machine owns it",
    "recommended": "idea",
    "gap": "machine owns it",
}

STAGE_LABELS = {
    "queued": "waiting for agent",
    "build": "agent working",
    "review": "ready for review",
    "merge": "ready to merge",
    "deploy": "ready to deploy",
    "verify": "prove it live",
}

STAGE_ACTIONS = {
    "queued": "wait",
    "build": "run",
    "review": "review",
    "merge": "merge",
    "deploy": "deploy",
    "verify": "verify",
}

# ── Intake (reverse direction: Pulse post -> ledger item + kanban task) ──────
INTAKE_ENABLED = cfg("PULSE_BRIDGE_INTAKE", "1") == "1"
INTAKE_MAX_CONCURRENT = int(cfg("PULSE_BRIDGE_INTAKE_MAX_CONCURRENT", "3"))
HERMES_DIR = cfg("PULSE_BRIDGE_HERMES_DIR", str(HOME / ".hermes/hermes-agent"))
HERMES_HOME_DIR = cfg("PULSE_BRIDGE_HERMES_HOME", str(HOME / ".hermes"))
KANBAN_MIRROR_ENABLED = cfg("PULSE_BRIDGE_KANBAN_MIRROR", "0") == "1"
KANBAN_MIRROR_KINDS = {k.strip() for k in
                       cfg("PULSE_BRIDGE_KANBAN_KINDS",
                           "needs-you,failed,pipeline,gap").split(",")
                       if k.strip()}
KANBAN_MIRROR_TRIAGE = cfg("PULSE_BRIDGE_KANBAN_TRIAGE", "1") == "1"
KANBAN_MIRROR_INITIAL_STATUS = cfg("PULSE_BRIDGE_KANBAN_INITIAL_STATUS",
                                   "").strip()
# user_names never ingested (service accounts / the bridge itself)
INTAKE_IGNORE_USERS = {u.strip().lower() for u in
                       cfg("PULSE_BRIDGE_INTAKE_IGNORE",
                           f"{POSTER_NAME},Ellie,AROS").split(",") if u.strip()}
# chat-only opt-out: posts starting with any of these prefixes (matched
# case-insensitively) are just talk — no ledger item, no task, no comments
INTAKE_OPTOUT_PREFIXES = tuple(
    p.strip().lower() for p in
    cfg("PULSE_BRIDGE_INTAKE_OPTOUT", "💬,chat:").split(",") if p.strip())


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": {}, "kanban": {},
            "seeded": False, "channel_id": "", "intake_checkpoint": 0,
            "intake": {}}


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 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("@", "@​")


def lifecycle(item: dict) -> dict:
    kind = item.get("kind", "item")
    stage = item.get("stage")
    out = {
        "kind": KIND_LABELS.get(kind, kind),
        "owner": "human" if kind == "needs-you" else "machine",
        "state": "open",
        "action": "none",
    }
    if kind == "recommended":
        out.update(owner="human", state="idea", action="optional")
    elif kind == "needs-you":
        out.update(state="waiting-on-human", action="answer")
    elif kind == "failed":
        out.update(state="failed", action="fix-or-retry")
    elif stage:
        out.update(state=stage, action=STAGE_ACTIONS.get(stage, "run"))
    if stage:
        out["stage"] = STAGE_LABELS.get(stage, stage)
    return out


def lifecycle_line(item: dict) -> str:
    facts = lifecycle(item)
    bits = [f"{key}:{value}" for key, value in facts.items()]
    if item.get("stage"):
        bits.append(f"stage_key:{item['stage']}")
    return " · ".join(bits)


def tag_line(item: dict) -> str:
    bits = [f"#{item.get('kind', 'item')}"]
    if item.get("stage"):
        bits.append(f"#{item['stage']}")
    facts = lifecycle(item)
    bits.extend([
        f"owner:{facts['owner']}",
        f"state:{facts['state']}",
        f"action:{facts['action']}",
    ])
    if item.get("stage"):
        bits.append(f"stage:{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(lifecycle_line(item))
    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":
        # Deliberately a LIVE mention (not neutralized): mib007 resolves
        # @<lower(user name)> against active workspace members and creates a
        # comment.mention notification, so the escalation actually pings the
        # human. Everything else in the line stays neutralized.
        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} ({STAGE_LABELS.get(stage, 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} ({KIND_LABELS.get(kind, kind)})")
        if stage:
            lines.append(f"stage → {stage} ({STAGE_LABELS.get(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 (unified message envelope) ─────────────────────────────────

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}/messages/channels") or []
    want = CHANNEL_NAME.lower()
    for ch in channels:
        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"]
    return None


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
    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_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:<id> 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}"
    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, *, mirror_kanban: bool = True) -> None:
    if item["id"] in state["map"]:
        if mirror_kanban:
            ensure_kanban_mirror(state, item)
        return  # already mirrored (seed/feed overlap)
    legacy = find_legacy_post(state, item["id"])
    if legacy:
        state["map"][item["id"]] = legacy
        log(f"post   {item['id'][:8]} -> adopted legacy {legacy}")
        if mirror_kanban:
            ensure_kanban_mirror(state, item)
        return
    # 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"]
    log(f"post   {item['id'][:8]} -> {msg['id']} ({item.get('title', '')[:60]})")
    if mirror_kanban:
        ensure_kanban_mirror(state, item)


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(state, content, thread_id=post_id, idem=idem)
    except urllib.error.HTTPError as e:
        # 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 retryable)")
            return
        raise
    log(f"{what:6} {ledger_id[:8]} -> comment {msg['id']}")


# ── Intake: Pulse posts -> ledger + kanban ───────────────────────────────────

def kanban_cmd() -> list[str]:
    """How to invoke Dash's kanban CLI, resolved fresh on every call.

    The shared checkout's venv is not stable: a release cutover can delete or
    replace it mid-run (observed 2026-08-23 — every intake tick failed with
    ENOENT on `<checkout>/venv/bin/python`). The `dash-agent` / shim entrypoint is
    repointed by the release recipe itself, so it is the durable entry point;
    the checkout venv stays as a fallback for installs that have no shim.
    """
    override = cfg("PULSE_BRIDGE_HERMES_PY", "")
    if override:
        return [override, "-m", "shiva_cli.main"]
    dash_agent = cfg("PULSE_BRIDGE_DASH_AGENT", "dash-agent")
    if dash_agent and shutil.which(dash_agent):
        return [dash_agent]
    dash_shim = HOME / ".local/bin/dash-agent"
    if os.access(dash_shim, os.X_OK):
        return [str(dash_shim)]
    shim = HOME / ".local/bin/hermes"
    if os.access(shim, os.X_OK):
        return [str(shim)]
    return [f"{HERMES_DIR}/venv/bin/python", "-m", "shiva_cli.main"]


def kanban(*args: str, timeout: int = 60) -> str:
    cmd = kanban_cmd()
    cwd = HERMES_DIR if os.path.isdir(HERMES_DIR) else str(HOME)
    out = subprocess.run(
        [*cmd, "kanban", *args],
        capture_output=True, text=True, timeout=timeout, cwd=cwd,
        env={**os.environ, "HERMES_HOME": HERMES_HOME_DIR})
    if out.returncode != 0:
        raise RuntimeError(f"kanban {args[0]} rc={out.returncode}: "
                           f"{out.stderr.strip()[:300]}")
    return out.stdout


def ensure_kanban_mirror(state: dict, item: dict) -> None:
    """Optionally mirror outbound ledger items into Dash Kanban."""
    if not KANBAN_MIRROR_ENABLED:
        return
    if item.get("kind") not in KANBAN_MIRROR_KINDS:
        return
    state.setdefault("kanban", {})
    if state["kanban"].get(item["id"]):
        return
    post_id = state.get("map", {}).get(item["id"], "")
    title = f"{item.get('kind', 'item')}: {item.get('title', '(untitled)')}"
    body = "\n\n".join(x for x in (
        neutralize(item.get("detail", "")),
        lifecycle_line(item),
        f"ledger:{item['id']}",
        f"pulse:{post_id}" if post_id else "",
    ) if x)
    args = ["create", title[:120], "--body", body,
            "--created-by", "pulse-bridge", "--assignee", "default",
            "--idempotency-key", f"shre-items-{item['id']}", "--json"]
    if KANBAN_MIRROR_TRIAGE:
        args.append("--triage")
    elif KANBAN_MIRROR_INITIAL_STATUS:
        args.extend(["--initial-status", KANBAN_MIRROR_INITIAL_STATUS])
    raw = kanban(*args)
    task = json.loads(raw)
    task_id = task["id"] if isinstance(task, dict) else task[0]["id"]
    state["kanban"][item["id"]] = task_id
    save_state(state)
    log(f"kanban {item['id'][:8]} -> {task_id}")


def ledger_add_for_post(post: dict) -> str:
    """Create (or adopt) a pipeline ledger item for a Pulse post; return its
    FULL id. Idempotent: a crash after `shre-items add` but before the state
    save must not create a second item on retry, so existing open items are
    searched for this post's marker first."""
    marker = f"post:{post['id']}"
    for item in open_items():
        if marker in (item.get("detail") or ""):
            return item["id"]
    content = (post.get("content") or "").strip()
    title = " ".join(content.split())[:80] or "(untitled pulse post)"
    detail = f"{content}\n\n[pulse-intake] {marker} " \
             f"author:{post.get('user_name') or post.get('user_id')}"
    out = shre_items("add", "--kind", "pipeline", "--stage", "queued",
                     "--title", title, "--detail", detail,
                     "--surface", "pulse", "--tag", "pulse-intake")
    short = out.split()[0]
    for item in open_items():
        if item["id"].startswith(short):
            return item["id"]
    raise RuntimeError(f"cannot resolve full ledger id for {short}")


def intake_active_count(state: dict) -> int:
    return sum(1 for e in state["intake"].values()
               if e.get("task_id") and not e.get("closed"))


def intake_start_task(state: dict, post_id: str, entry: dict) -> None:
    """Create the kanban task for an intake entry and acknowledge on the post."""
    content = entry.get("content", "")
    title = " ".join(content.split())[:80] or "(pulse task)"
    # Crash recovery leans on the CLI's documented idempotency contract:
    # "If a non-archived task with this key exists, its id is returned
    # instead of creating a duplicate" — so re-calling after a crash between
    # create and save_state adopts the existing task rather than forking one.
    raw = kanban("create", title,
                 "--body", f"{content}\n\n(from Pulse post {post_id}, "
                           f"ledger {entry['ledger_id'][:8]})",
                 "--created-by", "pulse-bridge", "--assignee", "default",
                 "--idempotency-key", f"pulse-{post_id}", "--json")
    task = json.loads(raw)
    task_id = task["id"] if isinstance(task, dict) else task[0]["id"]
    entry["task_id"] = task_id
    entry["last_status"] = task.get("status", "todo") if isinstance(task, dict) else "todo"
    entry.pop("queued", None)
    save_state(state)
    log(f"intake {post_id[:8]} -> kanban {task_id}")
    try:
        send_message(state,
                     f"🤖 picked up — ledger {entry['ledger_id'][:8]}, "
                     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:<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:
    """Turn new non-bridge top-level activity posts into ledger items + tasks."""
    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;
        # never ingest history. (+1 because the window below is INCLUSIVE.)
        state["intake_checkpoint"] = 1 + max(
            [int(m.get("created_at") or 0) for m in msgs] or [0])
        save_state(state)
        return
    # Inclusive window (ts >= cp) so two posts sharing one millisecond can't
    # lose the second to a crash between them; the intake map (keyed by post
    # id) is the dedupe, mirroring the outbound seen-set discipline.
    candidates = []
    for m in msgs:
        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
                or content.lower().startswith(INTAKE_OPTOUT_PREFIXES)):
            continue  # opt-out posts are chat-only: no item, no task, no reply
        candidates.append(m)
    for post in sorted(candidates, key=lambda m: int(m["created_at"])):
        ledger_id = ledger_add_for_post(post)
        # Dedupe with the outbound mirror: map the ledger item to the USER'S
        # post before its op:add reaches the feed, so create_post() skips it.
        state["map"][ledger_id] = post["id"]
        entry = {"ledger_id": ledger_id, "content": post.get("content") or "",
                 "author": post.get("user_name") or post.get("user_id")}
        state["intake"][post["id"]] = entry
        state["intake_checkpoint"] = max(int(state["intake_checkpoint"]),
                                         int(post["created_at"]))
        save_state(state)
        log(f"intake {post['id'][:8]} -> ledger {ledger_id[:8]}")
        backlog = any(e.get("queued") for e in state["intake"].values())
        if not backlog and intake_active_count(state) < INTAKE_MAX_CONCURRENT:
            intake_start_task(state, post["id"], entry)
        else:
            entry["queued"] = True
            save_state(state)
            n = intake_active_count(state)
            log(f"intake {post['id'][:8]} queued behind {n} tasks")
            try:
                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}")


STATUS_COMMENT = {"running": "▶️ running", "blocked": "⛔ blocked",
                  "todo": "📋 waiting for a worker"}


def intake_track(state: dict) -> None:
    """Advance ledger + post comments as kanban task statuses change."""
    open_entries = {pid: e for pid, e in state["intake"].items()
                    if e.get("task_id") and not e.get("closed")}
    if not open_entries:
        return
    tasks = {t["id"]: t for t in json.loads(kanban("list", "--json"))}
    for post_id, entry in open_entries.items():
        task = tasks.get(entry["task_id"])
        if task is None:
            continue  # archived/missing; leave for manual attention
        status = task.get("status")
        if status == entry.get("last_status"):
            continue
        entry["last_status"] = status
        lid = entry["ledger_id"]
        if status == "done":
            note = (task.get("result") or "").strip()
            body = "✅ completed" + (f" — {neutralize(note[:400])}" if note
                                    else "")
            entry["closed"] = True
            try:
                shre_items("done", lid[:8], "--why",
                           f"kanban {entry['task_id']} completed")
            except RuntimeError as e:
                log(f"WARN close {lid[:8]}: {e}")
        else:
            body = STATUS_COMMENT.get(status, f"status → {status}")
            if status == "running":
                try:
                    shre_items("update", lid[:8], "--stage", "build",
                               "--note", f"kanban {entry['task_id']} running")
                except RuntimeError as e:
                    log(f"WARN stage {lid[:8]}: {e}")
        save_state(state)
        log(f"track  {entry['task_id']} {status} (post {post_id[:8]})")
        try:
            # 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}")


def intake_promote(state: dict) -> None:
    """Start queued intake entries when concurrency slots free up."""
    for post_id, entry in state["intake"].items():
        if not entry.get("queued"):
            continue
        if intake_active_count(state) >= INTAKE_MAX_CONCURRENT:
            return
        intake_start_task(state, post_id, entry)


# ── 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, mirror_kanban=False)
        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)
        return
    # Intake-managed items already get status comments from intake_track();
    # 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", idem=idem)
    else:
        comment(state, lid, close_body(rec), "close", idem=idem)


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 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"):
        try:
            seed(state)
        except Exception as e:
            log(f"FATAL seed failed: {e}")
            time.sleep(POLL_SEC)
            return 1  # launchd KeepAlive restarts us
    # Older state files predate intake; give them the new keys.
    state.setdefault("intake_checkpoint", 0)
    state.setdefault("intake", {})
    while True:
        try:
            poll(state)
        except Exception as e:
            log(f"ERROR poll: {e}")
        if INTAKE_ENABLED:
            # track first (frees slots), then promote the queued backlog
            # (FIFO), then ingest new posts — so newcomers can't jump ahead.
            for step in (intake_track, intake_promote, intake_ingest):
                try:
                    step(state)
                except Exception as e:
                    log(f"ERROR {step.__name__}: {e}")
        time.sleep(POLL_SEC)


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