v1 intake: Pulse posts become ledger items + dispatched kanban tasks

- new top-level non-bridge posts in the activity channel -> shre-items
  pipeline item (tag pulse-intake) + hermes kanban task (idempotency-key
  pulse-<post id>, created-by pulse-bridge) + ack comment
- outbound dedupe: ledger->post map seeded with the user's post id before
  the op:add hits the feed, so the mirror adopts the post
- status tracking via one 'kanban list --json' per poll: running/blocked/
  done comments on the post, ledger stage build -> done
- cap 3 concurrent intake tasks; overflow queued with a comment and
  promoted when slots free; intake state in the same atomic state.json
- first intake run only sets the checkpoint (no history ingestion)

Co-Authored-By: Claude Fable 5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01YECpkAwQUwgu7NVy91R8fW

README.md

@@ -62,7 +62,34 @@ cp ai.shre.pulse-bridge.plist ~/Library/LaunchAgents/
 launchctl bootstrap gui/$(id -u) ~/Library/LaunchAgents/ai.shre.pulse-bridge.plist
 ```
 
-## v0 limitations
+## v1 intake (reverse direction)
+
+New top-level posts in the `activity` channel written by a human (not the
+bridge, not agent accounts, not `🤖`-prefixed, not mirror posts carrying a
+`ledger:` tag line) become executed, tracked background work:
+
+1. **Ledger item** — `shre-items add --kind pipeline --stage queued
+   --tag pulse-intake`, title = first 80 chars of the post, detail = full
+   post + post id. The ledger→post map is seeded with the USER'S post id
+   *before* the item's `op:add` reaches the feed, so the outbound mirror
+   adopts the user's post instead of creating a duplicate.
+2. **Kanban task** — `hermes kanban create … --created-by pulse-bridge
+   --idempotency-key pulse-<post id> --json` on the default board; the
+   Hermes embedded-gateway dispatcher picks it up (~60 s). Cap: max
+   **3** concurrent intake-spawned tasks; overflow posts get
+   `⏳ queued behind N tasks` and start when a slot frees.
+3. **Acknowledgment** — comment `🤖 picked up — ledger <id>, kanban <task>`
+   on the user's post.
+4. **Tracking** — each poll reads `kanban list --json` once; on a status
+   change it comments (`▶️ running` / `⛔ blocked` / `✅ completed — <result>`)
+   and advances the ledger item (running → stage build; done → closed with
+   `--why "kanban <id> completed"`).
+
+Intake state (`intake_checkpoint` epoch-ms + post↔ledger↔task map) lives in
+the same atomically-written state.json. First intake run only sets the
+checkpoint to the newest existing post — history is never ingested.
+
+## v0/v1 limitations
 
 - **No real user-mention primitive.** mib007 comms has *agent* @mentions
   only (the first `@handle` in a message that matches a workspace agent

bridge.env

@@ -12,3 +12,11 @@ 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
+# Reverse intake: user posts in the activity channel become ledger items +
+# kanban tasks executed by the Hermes dispatcher
+PULSE_BRIDGE_INTAKE=1
+PULSE_BRIDGE_INTAKE_MAX_CONCURRENT=3
+PULSE_BRIDGE_HERMES_DIR=/Users/aibot/.hermes/hermes-agent
+PULSE_BRIDGE_HERMES_HOME=/Users/aibot/.hermes
+# comma-separated user_names never ingested (bridge + agent accounts)
+PULSE_BRIDGE_INTAKE_IGNORE=Ledger,Ellie,AROS

bridge.py

@@ -63,6 +63,16 @@ TOKEN_KEY = cfg("PULSE_BRIDGE_TOKEN_KEY", "mib007")
 
 STAGES = ["queued", "build", "review", "merge", "deploy", "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"))
+# 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()}
+
 
 def log(msg: str) -> None:
     print(f"{datetime.now(timezone.utc).isoformat(timespec='seconds')} {msg}",
@@ -103,7 +113,7 @@ def load_state() -> dict:
                 "deliberately; refusing to reseed automatically")
             sys.exit(1)
     return {"checkpoint": "", "seen": [], "map": {}, "seeded": False,
-            "channel_id": ""}
+            "channel_id": "", "intake_checkpoint": 0, "intake": {}}
 
 
 def save_state(state: dict) -> None:
@@ -300,6 +310,170 @@ def comment(state: dict, ledger_id: str, content: str, what: str) -> None:
     log(f"{what:6} {ledger_id[:8]} -> comment {msg['id']}")
 
 
+# ── Intake: Pulse posts -> ledger + kanban ───────────────────────────────────
+
+def kanban(*args: str, timeout: int = 60) -> str:
+    out = subprocess.run(
+        [f"{HERMES_DIR}/venv/bin/python", "-m", "shiva_cli.main",
+         "kanban", *args],
+        capture_output=True, text=True, timeout=timeout, cwd=HERMES_DIR,
+        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 ledger_add_for_post(post: dict) -> str:
+    """Create a pipeline ledger item for a Pulse post; return its FULL id."""
+    content = (post.get("content") or "").strip()
+    title = " ".join(content.split())[:80] or "(untitled pulse post)"
+    detail = f"{content}\n\n[pulse-intake] post:{post['id']} " \
+             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)"
+    raw = kanban("create", title,
+                 "--body", f"{content}\n\n(from Pulse post {post_id}, "
+                           f"ledger {entry['ledger_id'][:8]})",
+                 "--created-by", "pulse-bridge",
+                 "--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(ensure_channel(state),
+                     f"🤖 picked up — ledger {entry['ledger_id'][:8]}, "
+                     f"kanban {task_id}", thread_id=post_id)
+    except urllib.error.HTTPError as e:
+        log(f"DROP ack {post_id[:8]}: HTTP {e.code}")
+
+
+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 []
+    cp = int(state.get("intake_checkpoint") or 0)
+    if cp == 0:
+        # First intake run: start from the newest existing post; never ingest
+        # history.
+        state["intake_checkpoint"] = max(
+            [int(m.get("created_at") or 0) for m in msgs] or [0])
+        save_state(state)
+        return
+    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"]
+                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):
+            continue
+        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]}")
+        if 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(ensure_channel(state),
+                             f"⏳ queued behind {n} tasks",
+                             thread_id=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:
+            send_message(ensure_channel(state), body, thread_id=post_id)
+        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:
@@ -360,11 +534,20 @@ def main() -> int:
             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:
+            for step in (intake_ingest, intake_track, intake_promote):
+                try:
+                    step(state)
+                except Exception as e:
+                    log(f"ERROR {step.__name__}: {e}")
         time.sleep(POLL_SEC)