diff --git a/README.md b/README.md index 41f4900..759b2c1 100644 --- a/README.md +++ b/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- --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 , kanban ` + on the user's post. +4. **Tracking** β€” each poll reads `kanban list --json` once; on a status + change it comments (`▢️ running` / `β›” blocked` / `βœ… completed β€” `) + and advances the ledger item (running β†’ stage build; done β†’ closed with + `--why "kanban 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 diff --git a/bridge.env b/bridge.env index ffd520f..5029cdb 100644 --- a/bridge.env +++ b/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 diff --git a/bridge.py b/bridge.py index e01dcbd..d6a40e2 100644 --- a/bridge.py +++ b/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)