diff --git a/bridge.env b/bridge.env index 959eee7..c4a031c 100644 --- a/bridge.env +++ b/bridge.env @@ -22,6 +22,13 @@ 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 +# Outbound mirror: live ledger add events -> Dash Kanban triage cards. Seeded +# historic/open items still mirror only to Pulse posts. +PULSE_BRIDGE_KANBAN_MIRROR=1 +PULSE_BRIDGE_KANBAN_KINDS=needs-you,failed,pipeline,gap +PULSE_BRIDGE_KANBAN_TRIAGE=1 +PULSE_BRIDGE_KANBAN_INITIAL_STATUS= +PULSE_BRIDGE_DASH_AGENT=dash-agent # comma-separated user_names never ingested (bridge + agent accounts) PULSE_BRIDGE_INTAKE_IGNORE=Ledger,Ellie,AROS # chat-only prefixes (case-insensitive): posts starting with one of these are diff --git a/bridge.py b/bridge.py index ffa0056..20aecdf 100644 --- a/bridge.py +++ b/bridge.py @@ -30,6 +30,7 @@ Stdlib only. State (checkpoint + ledger-id -> post-id map) lives in import json import os import re +import shutil import subprocess import sys import time @@ -89,11 +90,45 @@ SELF_ACTOR_ID = cfg("PULSE_BRIDGE_SELF_ACTOR_ID", 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", @@ -143,8 +178,9 @@ def load_state() -> dict: log(f"FATAL state file corrupt: {STATE_FILE} — fix or remove it " "deliberately; refusing to reseed automatically") sys.exit(1) - return {"checkpoint": "", "seen": [], "map": {}, "seeded": False, - "channel_id": "", "intake_checkpoint": 0, "intake": {}} + return {"checkpoint": "", "seen": [], "map": {}, "kanban": {}, + "seeded": False, "channel_id": "", "intake_checkpoint": 0, + "intake": {}} def save_state(state: dict) -> None: @@ -174,10 +210,48 @@ def neutralize(text: str) -> str: 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"): @@ -194,6 +268,7 @@ def post_body(item: dict) -> str: 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) @@ -212,15 +287,15 @@ def update_body(rec: dict) -> str: " waiting on a human." + (f" — {note}" if note else "")) note = "" # already included elif kind == "failed": - where = f" at {stage}" if stage else "" + 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}") + lines.append(f"kind → {kind} ({KIND_LABELS.get(kind, kind)})") if stage: - lines.append(f"stage → {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"): @@ -342,19 +417,25 @@ def find_legacy_post(state: dict, ledger_id: str): return None -def create_post(state: dict, item: dict) -> 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: 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, @@ -381,17 +462,23 @@ def comment(state: dict, ledger_id: str, content: str, what: str, # ── Intake: Pulse posts -> ledger + kanban ─────────────────────────────────── def kanban_cmd() -> list[str]: - """How to invoke hermes' kanban CLI, resolved fresh on every call. + """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 `/venv/bin/python`). The `~/.local/bin/hermes` shim is + ENOENT on `/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)] @@ -411,6 +498,38 @@ def kanban(*args: str, timeout: int = 60) -> str: 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 @@ -640,7 +759,7 @@ def seed(state: dict) -> None: log(f"seeding {len(items)} open items") for item in items: try: - create_post(state, item) + 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