Add ledger lifecycle tags and Dash Kanban mirror

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:<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,
@@ -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 `<checkout>/venv/bin/python`). The `~/.local/bin/hermes` shim is
+    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)]
@@ -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