Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
24c899c3f6 | ||
|
|
c0c5a9feb6 |
@@ -22,6 +22,13 @@ PULSE_BRIDGE_INTAKE=1
|
|||||||
PULSE_BRIDGE_INTAKE_MAX_CONCURRENT=3
|
PULSE_BRIDGE_INTAKE_MAX_CONCURRENT=3
|
||||||
PULSE_BRIDGE_HERMES_DIR=/Users/aibot/.hermes/hermes-agent
|
PULSE_BRIDGE_HERMES_DIR=/Users/aibot/.hermes/hermes-agent
|
||||||
PULSE_BRIDGE_HERMES_HOME=/Users/aibot/.hermes
|
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)
|
# comma-separated user_names never ingested (bridge + agent accounts)
|
||||||
PULSE_BRIDGE_INTAKE_IGNORE=Ledger,Ellie,AROS
|
PULSE_BRIDGE_INTAKE_IGNORE=Ledger,Ellie,AROS
|
||||||
# chat-only prefixes (case-insensitive): posts starting with one of these are
|
# 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 json
|
||||||
import os
|
import os
|
||||||
import re
|
import re
|
||||||
|
import shutil
|
||||||
import subprocess
|
import subprocess
|
||||||
import sys
|
import sys
|
||||||
import time
|
import time
|
||||||
@@ -89,11 +90,45 @@ SELF_ACTOR_ID = cfg("PULSE_BRIDGE_SELF_ACTOR_ID",
|
|||||||
|
|
||||||
STAGES = ["queued", "build", "review", "merge", "deploy", "verify"]
|
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 (reverse direction: Pulse post -> ledger item + kanban task) ──────
|
||||||
INTAKE_ENABLED = cfg("PULSE_BRIDGE_INTAKE", "1") == "1"
|
INTAKE_ENABLED = cfg("PULSE_BRIDGE_INTAKE", "1") == "1"
|
||||||
INTAKE_MAX_CONCURRENT = int(cfg("PULSE_BRIDGE_INTAKE_MAX_CONCURRENT", "3"))
|
INTAKE_MAX_CONCURRENT = int(cfg("PULSE_BRIDGE_INTAKE_MAX_CONCURRENT", "3"))
|
||||||
HERMES_DIR = cfg("PULSE_BRIDGE_HERMES_DIR", str(HOME / ".hermes/hermes-agent"))
|
HERMES_DIR = cfg("PULSE_BRIDGE_HERMES_DIR", str(HOME / ".hermes/hermes-agent"))
|
||||||
HERMES_HOME_DIR = cfg("PULSE_BRIDGE_HERMES_HOME", str(HOME / ".hermes"))
|
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)
|
# user_names never ingested (service accounts / the bridge itself)
|
||||||
INTAKE_IGNORE_USERS = {u.strip().lower() for u in
|
INTAKE_IGNORE_USERS = {u.strip().lower() for u in
|
||||||
cfg("PULSE_BRIDGE_INTAKE_IGNORE",
|
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 "
|
log(f"FATAL state file corrupt: {STATE_FILE} — fix or remove it "
|
||||||
"deliberately; refusing to reseed automatically")
|
"deliberately; refusing to reseed automatically")
|
||||||
sys.exit(1)
|
sys.exit(1)
|
||||||
return {"checkpoint": "", "seen": [], "map": {}, "seeded": False,
|
return {"checkpoint": "", "seen": [], "map": {}, "kanban": {},
|
||||||
"channel_id": "", "intake_checkpoint": 0, "intake": {}}
|
"seeded": False, "channel_id": "", "intake_checkpoint": 0,
|
||||||
|
"intake": {}}
|
||||||
|
|
||||||
|
|
||||||
def save_state(state: dict) -> None:
|
def save_state(state: dict) -> None:
|
||||||
@@ -174,10 +210,48 @@ def neutralize(text: str) -> str:
|
|||||||
return (text or "").replace("@", "@​")
|
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:
|
def tag_line(item: dict) -> str:
|
||||||
bits = [f"#{item.get('kind', 'item')}"]
|
bits = [f"#{item.get('kind', 'item')}"]
|
||||||
if item.get("stage"):
|
if item.get("stage"):
|
||||||
bits.append(f"#{item['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 []:
|
for t in item.get("tags", []) or []:
|
||||||
bits.append(f"#{t}")
|
bits.append(f"#{t}")
|
||||||
if item.get("surface"):
|
if item.get("surface"):
|
||||||
@@ -194,6 +268,7 @@ def post_body(item: dict) -> str:
|
|||||||
parts = [f"{head} {neutralize(item.get('title', '(untitled)'))}"]
|
parts = [f"{head} {neutralize(item.get('title', '(untitled)'))}"]
|
||||||
if item.get("detail"):
|
if item.get("detail"):
|
||||||
parts.append(neutralize(item["detail"]))
|
parts.append(neutralize(item["detail"]))
|
||||||
|
parts.append(lifecycle_line(item))
|
||||||
parts.append(tag_line(item))
|
parts.append(tag_line(item))
|
||||||
return "\n\n".join(parts)
|
return "\n\n".join(parts)
|
||||||
|
|
||||||
@@ -212,15 +287,15 @@ def update_body(rec: dict) -> str:
|
|||||||
" waiting on a human." + (f" — {note}" if note else ""))
|
" waiting on a human." + (f" — {note}" if note else ""))
|
||||||
note = "" # already included
|
note = "" # already included
|
||||||
elif kind == "failed":
|
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 ""
|
attempt = f", attempt {rec['attempt']}" if rec.get("attempt") else ""
|
||||||
lines.append(f"⛔ failed{where}{attempt}" + (f" — {note}" if note else ""))
|
lines.append(f"⛔ failed{where}{attempt}" + (f" — {note}" if note else ""))
|
||||||
note = ""
|
note = ""
|
||||||
else:
|
else:
|
||||||
if kind:
|
if kind:
|
||||||
lines.append(f"kind → {kind}")
|
lines.append(f"kind → {kind} ({KIND_LABELS.get(kind, kind)})")
|
||||||
if stage:
|
if stage:
|
||||||
lines.append(f"stage → {stage}")
|
lines.append(f"stage → {stage} ({STAGE_LABELS.get(stage, stage)})")
|
||||||
if rec.get("attempt") and not kind:
|
if rec.get("attempt") and not kind:
|
||||||
lines.append(f"attempt {rec['attempt']}")
|
lines.append(f"attempt {rec['attempt']}")
|
||||||
if rec.get("detail"):
|
if rec.get("detail"):
|
||||||
@@ -342,19 +417,25 @@ def find_legacy_post(state: dict, ledger_id: str):
|
|||||||
return None
|
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 item["id"] in state["map"]:
|
||||||
|
if mirror_kanban:
|
||||||
|
ensure_kanban_mirror(state, item)
|
||||||
return # already mirrored (seed/feed overlap)
|
return # already mirrored (seed/feed overlap)
|
||||||
legacy = find_legacy_post(state, item["id"])
|
legacy = find_legacy_post(state, item["id"])
|
||||||
if legacy:
|
if legacy:
|
||||||
state["map"][item["id"]] = legacy
|
state["map"][item["id"]] = legacy
|
||||||
log(f"post {item['id'][:8]} -> adopted legacy {legacy}")
|
log(f"post {item['id'][:8]} -> adopted legacy {legacy}")
|
||||||
|
if mirror_kanban:
|
||||||
|
ensure_kanban_mirror(state, item)
|
||||||
return
|
return
|
||||||
# ledger:<id> as the idempotency key: a retry after insert-then-crash gets
|
# 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.
|
# 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']}")
|
msg = send_message(state, post_body(item), idem=f"ledger:{item['id']}")
|
||||||
state["map"][item["id"]] = msg["id"]
|
state["map"][item["id"]] = msg["id"]
|
||||||
log(f"post {item['id'][:8]} -> {msg['id']} ({item.get('title', '')[:60]})")
|
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,
|
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 ───────────────────────────────────
|
# ── Intake: Pulse posts -> ledger + kanban ───────────────────────────────────
|
||||||
|
|
||||||
def kanban_cmd() -> list[str]:
|
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
|
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
|
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;
|
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.
|
the checkout venv stays as a fallback for installs that have no shim.
|
||||||
"""
|
"""
|
||||||
override = cfg("PULSE_BRIDGE_HERMES_PY", "")
|
override = cfg("PULSE_BRIDGE_HERMES_PY", "")
|
||||||
if override:
|
if override:
|
||||||
return [override, "-m", "shiva_cli.main"]
|
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"
|
shim = HOME / ".local/bin/hermes"
|
||||||
if os.access(shim, os.X_OK):
|
if os.access(shim, os.X_OK):
|
||||||
return [str(shim)]
|
return [str(shim)]
|
||||||
@@ -411,6 +498,38 @@ def kanban(*args: str, timeout: int = 60) -> str:
|
|||||||
return out.stdout
|
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:
|
def ledger_add_for_post(post: dict) -> str:
|
||||||
"""Create (or adopt) a pipeline ledger item for a Pulse post; return its
|
"""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
|
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")
|
log(f"seeding {len(items)} open items")
|
||||||
for item in items:
|
for item in items:
|
||||||
try:
|
try:
|
||||||
create_post(state, item)
|
create_post(state, item, mirror_kanban=False)
|
||||||
except Exception as e: # keep seeding the rest
|
except Exception as e: # keep seeding the rest
|
||||||
log(f"ERROR seeding {item.get('id', '?')[:8]}: {e}")
|
log(f"ERROR seeding {item.get('id', '?')[:8]}: {e}")
|
||||||
state["seeded"] = True
|
state["seeded"] = True
|
||||||
|
|||||||
Reference in New Issue
Block a user