Compare commits
2
Commits
2952b0c289
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
24c899c3f6 | ||
|
|
c0c5a9feb6 |
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user