Capture this laptop's Omarchy customizations
Everything on top of a stock Omarchy 4.0.4-1 install, gathered so a second machine can be brought to the same state: the ~/.config files that actually differ from the shipped defaults, the three hand-built themes, the ten shell plugins pinned to commits, the two custom scripts, two user units, and the /etc drop-ins. install.sh applies it all idempotently, backing up anything it overwrites. capture.sh runs the other way so the repo stays the source of truth rather than a snapshot that rots. indie.herdr-hud's ten local-only commits ship as a git bundle plus a working-tree patch, since they stay off the upstream PR deliberately. Secrets, the dotfiles repo's territory (shell/git/mise), the remote-access stack and aether-managed themes are all scoped out; README says why for each. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5
+192
@@ -0,0 +1,192 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Start and close Herdr agents for the HUD. herdr-monitor stays read-only."""
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
TIMEOUT = 15
|
||||
|
||||
# Herdr's supported kinds (`herdr agent start --kind`), with display names.
|
||||
KINDS = {
|
||||
"claude": "Claude Code", "codex": "Codex", "gemini": "Gemini CLI", "cursor": "Cursor",
|
||||
"opencode": "opencode", "copilot": "Copilot CLI", "pi": "pi", "omp": "omp", "grok": "Grok",
|
||||
"hermes": "Hermes", "amp": "Amp", "droid": "Droid", "cline": "Cline", "kimi": "Kimi",
|
||||
"kiro": "Kiro", "qwen": "Qwen Code", "kilo": "Kilo", "devin": "Devin", "agy": "Antigravity",
|
||||
"mastracode": "Mastra Code", "qodercli": "Qoder", "maki": "Maki",
|
||||
}
|
||||
EXECUTABLE = {"cursor": "cursor-agent"}
|
||||
|
||||
|
||||
class AgentsError(RuntimeError):
|
||||
def __init__(self, message: str, code: str = ""):
|
||||
super().__init__(message)
|
||||
self.code = code
|
||||
|
||||
|
||||
def herdr(*args: str, timeout: float = TIMEOUT) -> dict:
|
||||
executable = shutil.which("herdr")
|
||||
if not executable:
|
||||
raise AgentsError("Herdr is not installed or not on PATH")
|
||||
try:
|
||||
done = subprocess.run([executable, *args], capture_output=True, text=True, timeout=timeout)
|
||||
except subprocess.TimeoutExpired as error:
|
||||
raise AgentsError("Herdr did not respond in time") from error
|
||||
if done.returncode != 0:
|
||||
# Herdr prints {"error": {"code", "message"}} on stderr.
|
||||
try:
|
||||
error = json.loads(done.stderr).get("error") or {}
|
||||
except (json.JSONDecodeError, AttributeError):
|
||||
error = {}
|
||||
raise AgentsError(error.get("message") or done.stderr.strip() or f"herdr {args[0]} failed",
|
||||
error.get("code", ""))
|
||||
return json.loads(done.stdout or "{}").get("result", {})
|
||||
|
||||
|
||||
def installed_kinds() -> list[str]:
|
||||
# Agents run in Herdr's login shell, which may have PATH entries (mise, npm)
|
||||
# that the desktop shell lacks, so ask that shell.
|
||||
executables = [EXECUTABLE.get(kind, kind) for kind in KINDS]
|
||||
script = "for e in " + " ".join(executables) + '; do command -v "$e" >/dev/null 2>&1 && echo "$e"; done'
|
||||
try:
|
||||
done = subprocess.run([os.environ.get("SHELL", "/bin/bash"), "-lic", script],
|
||||
capture_output=True, text=True, timeout=TIMEOUT, stdin=subprocess.DEVNULL)
|
||||
found = set(done.stdout.split())
|
||||
except subprocess.TimeoutExpired:
|
||||
found = {e for e in executables if shutil.which(e)}
|
||||
return [kind for kind in KINDS if EXECUTABLE.get(kind, kind) in found]
|
||||
|
||||
|
||||
def options() -> dict:
|
||||
agents = herdr("agent", "list").get("agents", [])
|
||||
spaces = herdr("workspace", "list").get("workspaces", [])
|
||||
folders = {str(a.get("cwd")) for a in agents if a.get("cwd")}
|
||||
work = Path.home() / "Work"
|
||||
if work.is_dir():
|
||||
folders |= {str(p) for p in work.iterdir() if p.is_dir() and not p.name.startswith(".")}
|
||||
# A workspace's folder is where its first agent runs; used to pick a default.
|
||||
space_cwd = {}
|
||||
for agent in agents:
|
||||
space_cwd.setdefault(str(agent.get("workspace_id", "")), agent.get("cwd"))
|
||||
return {
|
||||
"kinds": [{"kind": kind, "label": KINDS[kind]} for kind in installed_kinds()],
|
||||
"folders": sorted(folders),
|
||||
"workspaces": [{"id": s.get("workspace_id"), "label": s.get("label") or s.get("workspace_id"),
|
||||
"cwd": space_cwd.get(str(s.get("workspace_id")))} for s in spaces],
|
||||
}
|
||||
|
||||
|
||||
def slug(text: str, fallback: str) -> str:
|
||||
# Herdr agent names: [a-z][a-z0-9_-]{0,31}, unique among live agents.
|
||||
base = re.sub(r"[^a-z0-9_-]+", "-", text.lower())
|
||||
base = re.sub(r"^[^a-z]+|-+$", "", base)[:24]
|
||||
return base or fallback
|
||||
|
||||
|
||||
def start(request: dict) -> dict:
|
||||
kind = str(request.get("kind", ""))
|
||||
if kind not in installed_kinds():
|
||||
raise AgentsError(f"{KINDS.get(kind, kind)} isn't installed")
|
||||
home = Path.home().resolve()
|
||||
folder = Path(os.path.expanduser(str(request.get("cwd", "")).strip())).resolve()
|
||||
if not folder.is_dir():
|
||||
raise AgentsError("That folder doesn't exist")
|
||||
if folder != home and home not in folder.parents:
|
||||
raise AgentsError("Pick a folder in your home directory")
|
||||
name = str(request.get("name", "")).strip()[:60]
|
||||
prompt = str(request.get("prompt", "")).strip()
|
||||
workspace = str(request.get("workspace", "") or "")
|
||||
label = (name or folder.name)[:24]
|
||||
|
||||
if workspace:
|
||||
created = herdr("tab", "create", "--workspace", workspace, "--cwd", str(folder), "--label", label)
|
||||
else:
|
||||
created = herdr("workspace", "create", "--cwd", str(folder), "--label", folder.name, "--no-focus")
|
||||
herdr("tab", "rename", created["tab"]["tab_id"], label)
|
||||
pane = created["root_pane"]["pane_id"]
|
||||
tab = created["tab"]["tab_id"]
|
||||
|
||||
taken = {a.get("name") for a in herdr("agent", "list").get("agents", [])}
|
||||
base = slug(name or folder.name, kind)
|
||||
agent_name, n = base, 2
|
||||
while agent_name in taken:
|
||||
agent_name, n = f"{base[:28]}-{n}", n + 1
|
||||
|
||||
# Claude shows --name as its terminal title and session name.
|
||||
args = ["--", "--name", name] if kind == "claude" and name else []
|
||||
time.sleep(0.8) # let the shell reach its prompt
|
||||
try:
|
||||
herdr("agent", "start", agent_name, "--kind", kind, "--pane", pane, "--timeout", "45000", *args, timeout=50)
|
||||
except AgentsError as error:
|
||||
# Blocked at startup (a trust-this-folder question, say) still counts:
|
||||
# the HUD terminal can answer it. Otherwise close the empty tab.
|
||||
if error.code == "agent_not_ready":
|
||||
return {"pane_id": pane, "blocked": True}
|
||||
try:
|
||||
herdr("tab", "close", tab)
|
||||
except AgentsError:
|
||||
pass
|
||||
raise AgentsError(f"{KINDS[kind]} didn't start: {error}") from None
|
||||
if prompt:
|
||||
# Herdr reports the start a moment before it takes prompts for the name.
|
||||
for attempt in range(20):
|
||||
try:
|
||||
herdr("agent", "prompt", pane, prompt)
|
||||
break
|
||||
except AgentsError as error:
|
||||
if error.code != "agent_not_ready" or attempt == 19:
|
||||
return {"pane_id": pane, "blocked": False, "prompt_error": str(error)}
|
||||
time.sleep(0.5)
|
||||
return {"pane_id": pane, "blocked": False}
|
||||
|
||||
|
||||
def claude_attach_id(pane: str) -> str:
|
||||
info = herdr("pane", "process-info", "--pane", pane).get("process_info", {})
|
||||
for process in info.get("foreground_processes", []):
|
||||
argv = process.get("argv") or []
|
||||
if len(argv) >= 3 and os.path.basename(argv[0]) == "claude" and argv[1] == "attach":
|
||||
return argv[2]
|
||||
return ""
|
||||
|
||||
|
||||
def close(pane: str) -> dict:
|
||||
# A `claude attach` pane only views a background session; closing the pane
|
||||
# would leave the session running, so stop it (its conversation is kept).
|
||||
session = ""
|
||||
try:
|
||||
session = claude_attach_id(pane)
|
||||
except AgentsError:
|
||||
pass
|
||||
if session:
|
||||
claude = shutil.which("claude") or str(Path.home() / ".local/share/mise/shims/claude")
|
||||
done = subprocess.run([claude, "stop", session], capture_output=True, text=True, timeout=TIMEOUT)
|
||||
if done.returncode != 0:
|
||||
raise AgentsError((done.stderr or done.stdout).strip().splitlines()[0] if (done.stderr or done.stdout).strip()
|
||||
else "claude stop failed")
|
||||
herdr("pane", "close", pane)
|
||||
return {"closed": pane, "stopped_session": session}
|
||||
|
||||
|
||||
def main(argv: list[str]) -> int:
|
||||
if argv[1:] == ["options"]:
|
||||
result = options()
|
||||
elif len(argv) == 3 and argv[1] == "start":
|
||||
result = start(json.loads(argv[2]))
|
||||
elif len(argv) == 3 and argv[1] == "close":
|
||||
result = close(argv[2])
|
||||
else:
|
||||
raise AgentsError("usage: herdr-agents options|start JSON|close PANE")
|
||||
print(json.dumps(result, ensure_ascii=False))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
raise SystemExit(main(sys.argv))
|
||||
except (AgentsError, OSError, ValueError, KeyError, subprocess.TimeoutExpired) as error:
|
||||
print(str(error)[:4096], file=sys.stderr)
|
||||
raise SystemExit(1)
|
||||
Reference in New Issue
Block a user