electron-rare/Constant
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add Constant TUI and durable memory commands

Browse Source
This commit is contained in:
L'électron rare
2026-04-05 18:29:48 +02:00
parent a7a96ab057
commit a421ae189b
7 changed files with 729 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+41 -7
View File
@@ -39,11 +39,13 @@ Current build status in this repo:
- a 4-pane host-local Zellij session per machine
- a fleet cockpit with one tab per machine
- `Constant`, the orchestration CLI on top of the cockpit
- a terminal TUI with a central `hexapus` buddy rail
- local mission planning and verification
- host-local execution for `claude`, `codex`, and `vibe`
- `copilot` as a manual lane
- local message bus + cross-machine bridge
- MLX-ready model plumbing for a small local orchestrator stack on macOS
- workspace-first durable memory with lexical + local vector search
The historical repo name may still say `zellij-ai-triple`.
The product name is now `Constant`.
@@ -178,9 +180,12 @@ codex login --device-auth
```bash
./scripts/Constant
./scripts/Constant tui
./scripts/Constant doctor
./scripts/Constant mission create "audit the repo" --workspace "$PWD"
./scripts/Constant mission status
./scripts/Constant memory rebuild --workspace "$PWD"
./scripts/Constant memory search "buddy rail" --workspace "$PWD"
./scripts/Constant cockpit open --workspace "$PWD"
```
@@ -190,6 +195,25 @@ If `Constant` is on your `PATH`, you can also just run:
Constant
```
In interactive mode, `Constant` with no arguments now opens the TUI by default.
### TUI keys
`Constant tui` gives you:
- a mission deck
- a mission board
- a `hexapus` buddy rail
- a bottom timeline mixed with memory echoes
Useful keys:
- `j` / `k`: move between missions
- `e`: rebuild memory for the current workspace
- `s`: summarize the selected mission into durable memory
- `z`: open the full fleet cockpit
- `q`: quit the TUI
## Fleet Configuration
For a public setup, think in terms of roles, not personal machine names.
@@ -213,6 +237,15 @@ The command-center machine is where you run:
The workers are where execution happens.
Public example config:
```bash
mkdir -p ~/.config/constant
cp examples/fleet.example.json ~/.config/constant/fleet.json
```
Legacy `~/.config/constant/fleet.yaml` is still read for compatibility, but the public-facing format is now JSON.
## Messaging
Each machine exposes a local bus:
@@ -290,13 +323,14 @@ Planned layers for the public `Constant` repo:
## Repo Layout
```text
constant/ Python orchestration core
scripts/Constant canonical CLI entrypoint
scripts/constant-machine.sh canonical single-machine cockpit entrypoint
scripts/constant-fleet.sh canonical fleet cockpit entrypoint
scripts/constant-fleet-install.sh
scripts/ai-msg.sh local agent bus
scripts/ai-bridge.sh inter-machine bridge
constant/ Python orchestration core
examples/fleet.example.json public fleet template
scripts/Constant canonical CLI entrypoint
scripts/constant-machine.sh canonical single-machine cockpit entrypoint
scripts/constant-fleet.sh canonical fleet cockpit entrypoint
scripts/constant-fleet-install.sh canonical fleet installer/checker
scripts/ai-msg.sh local agent bus
scripts/ai-bridge.sh inter-machine bridge
```
## Current Caveats
constant/cli.py
+136 -1
View File
@@ -11,6 +11,16 @@ from typing import Any
from . import __version__
from .daemon import daemon_status, request as daemon_request, serve_foreground, start_background, stop_background
from .executors import bridge_sync, execute_step, fleet_check
from .memory import (
enroll_workspace,
list_decisions,
memory_status,
persona_markdown,
rebuild_workspace_memory,
search_memory,
summarize_mission,
sync_qdrant,
)
from .paths import cache_root, repo_root, scripts_dir
from .state import (
append_event,
@@ -25,6 +35,7 @@ from .state import (
save_mission,
write_artifact,
)
from .tui import run_tui
def _print(data: Any, as_json: bool = False) -> None:
@@ -44,6 +55,20 @@ def _command_exists(binary: str) -> bool:
return False
def _warm_memory(workspace: str) -> dict[str, Any] | None:
try:
return rebuild_workspace_memory(workspace, enroll=True)
except Exception as exc: # noqa: BLE001
return {"error": str(exc), "workspace": workspace}
def _finalize_mission_memory(mission_id: str) -> dict[str, Any] | None:
try:
return summarize_mission(mission_id)
except Exception: # noqa: BLE001
return None
def _ensure_planned(mission: dict[str, Any]) -> dict[str, Any]:
if mission["steps"]:
return mission
@@ -103,6 +128,7 @@ def cmd_doctor(args: argparse.Namespace) -> int:
"daemon": status,
"models": models,
"health": health,
"memory": memory_status(),
"fleet": [{"label": entry["label"], "target": entry["target"]} for entry in fleet["machines"]],
}
_print(report, args.json)
@@ -176,7 +202,22 @@ def cmd_cockpit_open(args: argparse.Namespace) -> int:
return 0
def cmd_tui(args: argparse.Namespace) -> int:
workspace = str(Path(args.workspace or os.getcwd()).expanduser().resolve())
action = run_tui(workspace)
if action and action.get("action") == "cockpit":
return cmd_cockpit_open(
argparse.Namespace(
workspace=action.get("workspace", workspace),
local_session=args.local_session,
session=args.session,
)
)
return 0
def cmd_mission_create(args: argparse.Namespace) -> int:
_warm_memory(args.workspace)
mission = create_mission(args.prompt, args.workspace)
mission = _ensure_planned(mission)
_print(_mission_summary(mission), args.json)
@@ -185,6 +226,7 @@ def cmd_mission_create(args: argparse.Namespace) -> int:
def cmd_mission_plan(args: argparse.Namespace) -> int:
mission = load_mission(args.mission_id)
_warm_memory(mission["workspace"])
payload = daemon_request("plan", {"mission": mission})
mission["steps"] = payload["plan"]["steps"]
mission["title"] = payload["plan"]["title"]
@@ -243,6 +285,7 @@ def cmd_mission_run(args: argparse.Namespace) -> int:
mission["status"] = "needs_human"
save_mission(mission)
append_event(mission["mission_id"], "step.needs_human", {"step_id": step["step_id"], "summary": verdict.get("summary", "")})
_finalize_mission_memory(mission["mission_id"])
_print({"summary": _mission_summary(mission), "verdict": verdict}, args.json)
return 1
else:
@@ -250,6 +293,7 @@ def cmd_mission_run(args: argparse.Namespace) -> int:
mission["status"] = "failed"
save_mission(mission)
append_event(mission["mission_id"], "step.failed", {"step_id": step["step_id"], "summary": verdict.get("summary", "")})
_finalize_mission_memory(mission["mission_id"])
_print({"summary": _mission_summary(mission), "verdict": verdict}, args.json)
return 1
@@ -260,6 +304,7 @@ def cmd_mission_run(args: argparse.Namespace) -> int:
mission["status"] = "done"
save_mission(mission)
append_event(mission["mission_id"], "mission.done", {"mission_id": mission["mission_id"]})
_finalize_mission_memory(mission["mission_id"])
_print(_mission_summary(mission), args.json)
return 0
@@ -350,6 +395,8 @@ def cmd_delegate(args: argparse.Namespace) -> int:
def cmd_buddy_ask(args: argparse.Namespace) -> int:
mission = load_mission(args.mission_id) if args.mission_id else None
if mission:
_warm_memory(mission["workspace"])
answer = daemon_request("buddy", {"mission": mission, "prompt": args.prompt})
if mission:
append_event(mission["mission_id"], "buddy.ask", {"prompt": args.prompt, "answer": answer})
@@ -357,6 +404,48 @@ def cmd_buddy_ask(args: argparse.Namespace) -> int:
return 0
def cmd_memory_status(args: argparse.Namespace) -> int:
_print(memory_status(args.workspace), args.json)
return 0
def cmd_memory_rebuild(args: argparse.Namespace) -> int:
_print(rebuild_workspace_memory(args.workspace, enroll=not args.no_enroll), args.json)
return 0
def cmd_memory_enroll(args: argparse.Namespace) -> int:
_print(enroll_workspace(args.path), args.json)
return 0
def cmd_memory_search(args: argparse.Namespace) -> int:
_print(search_memory(args.query, args.workspace, args.limit), args.json)
return 0
def cmd_memory_persona_show(args: argparse.Namespace) -> int:
payload = {"persona": persona_markdown()} if args.json else persona_markdown()
_print(payload, args.json)
return 0
def cmd_memory_decisions(args: argparse.Namespace) -> int:
_print(list_decisions(args.workspace, args.mission_id), args.json)
return 0
def cmd_memory_sync_qdrant(args: argparse.Namespace) -> int:
payload = sync_qdrant(args.workspace)
_print(payload, args.json)
return 0 if payload.get("ok") or payload.get("skipped") else 1
def cmd_mission_summarize(args: argparse.Namespace) -> int:
_print(summarize_mission(args.mission_id), args.json)
return 0
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(prog="Constant")
parser.add_argument("-V", "--version", action="version", version=__version__)
@@ -366,6 +455,12 @@ def build_parser() -> argparse.ArgumentParser:
doctor.add_argument("--json", action="store_true")
doctor.set_defaults(func=cmd_doctor)
tui = subparsers.add_parser("tui")
tui.add_argument("--workspace")
tui.add_argument("--local-session")
tui.add_argument("--session")
tui.set_defaults(func=cmd_tui)
daemon = subparsers.add_parser("daemon")
daemon_sub = daemon.add_subparsers(dest="daemon_command", required=True)
daemon_start = daemon_sub.add_parser("start")
@@ -435,6 +530,10 @@ def build_parser() -> argparse.ArgumentParser:
mission_retry.add_argument("--step-id")
mission_retry.add_argument("--json", action="store_true")
mission_retry.set_defaults(func=cmd_mission_retry)
mission_summarize = mission_sub.add_parser("summarize")
mission_summarize.add_argument("mission_id")
mission_summarize.add_argument("--json", action="store_true")
mission_summarize.set_defaults(func=cmd_mission_summarize)
delegate = subparsers.add_parser("delegate")
delegate.add_argument("mission_id")
@@ -454,6 +553,42 @@ def build_parser() -> argparse.ArgumentParser:
buddy_ask.add_argument("--json", action="store_true")
buddy_ask.set_defaults(func=cmd_buddy_ask)
memory = subparsers.add_parser("memory")
memory_sub = memory.add_subparsers(dest="memory_command", required=True)
memory_status_cmd = memory_sub.add_parser("status")
memory_status_cmd.add_argument("--workspace")
memory_status_cmd.add_argument("--json", action="store_true")
memory_status_cmd.set_defaults(func=cmd_memory_status)
memory_rebuild_cmd = memory_sub.add_parser("rebuild")
memory_rebuild_cmd.add_argument("--workspace", default=os.getcwd())
memory_rebuild_cmd.add_argument("--no-enroll", action="store_true")
memory_rebuild_cmd.add_argument("--json", action="store_true")
memory_rebuild_cmd.set_defaults(func=cmd_memory_rebuild)
memory_enroll_cmd = memory_sub.add_parser("enroll")
memory_enroll_cmd.add_argument("path")
memory_enroll_cmd.add_argument("--json", action="store_true")
memory_enroll_cmd.set_defaults(func=cmd_memory_enroll)
memory_search_cmd = memory_sub.add_parser("search")
memory_search_cmd.add_argument("query")
memory_search_cmd.add_argument("--workspace")
memory_search_cmd.add_argument("--limit", type=int)
memory_search_cmd.add_argument("--json", action="store_true")
memory_search_cmd.set_defaults(func=cmd_memory_search)
memory_persona_cmd = memory_sub.add_parser("persona")
memory_persona_sub = memory_persona_cmd.add_subparsers(dest="memory_persona_command", required=True)
memory_persona_show_cmd = memory_persona_sub.add_parser("show")
memory_persona_show_cmd.add_argument("--json", action="store_true")
memory_persona_show_cmd.set_defaults(func=cmd_memory_persona_show)
memory_decisions_cmd = memory_sub.add_parser("decisions")
memory_decisions_cmd.add_argument("--workspace")
memory_decisions_cmd.add_argument("--mission-id")
memory_decisions_cmd.add_argument("--json", action="store_true")
memory_decisions_cmd.set_defaults(func=cmd_memory_decisions)
memory_sync_cmd = memory_sub.add_parser("sync-qdrant")
memory_sync_cmd.add_argument("--workspace")
memory_sync_cmd.add_argument("--json", action="store_true")
memory_sync_cmd.set_defaults(func=cmd_memory_sync_qdrant)
hidden = subparsers.add_parser("__serve")
hidden.set_defaults(func=lambda _args: serve_foreground())
@@ -465,7 +600,7 @@ def main(argv: list[str] | None = None) -> int:
argv = sys.argv[1:]
if not argv:
if sys.stdin.isatty() and sys.stdout.isatty():
argv = ["cockpit", "open", "--workspace", os.getcwd()]
argv = ["tui", "--workspace", os.getcwd()]
else:
argv = ["doctor"]
parser = build_parser()
constant/memory.py
+18 -2
View File
@@ -79,15 +79,31 @@ INSTRUCTION_FILES = ("CLAUDE.md", "AGENTS.md")
INSTRUCTION_SUFFIXES = {".json", ".md", ".prompt", ".toml", ".txt", ".yaml", ".yml"}
MAX_TEXT_BYTES = 256 * 1024
MAX_INSTRUCTION_BYTES = 128 * 1024
SQLITE_TIMEOUT_SECONDS = 10.0
SQLITE_BUSY_TIMEOUT_MS = 10_000
def _enable_wal(connection: sqlite3.Connection) -> None:
try:
row = connection.execute("pragma journal_mode").fetchone()
current_mode = str(row[0]).lower() if row else ""
if current_mode == "wal":
return
connection.execute("pragma journal_mode = wal")
except sqlite3.OperationalError:
# Another process may already hold the database. Keep the connection usable
# instead of failing the whole command or TUI refresh.
return
def _connect() -> sqlite3.Connection:
path = memory_store_path()
path.parent.mkdir(parents=True, exist_ok=True)
connection = sqlite3.connect(path)
connection = sqlite3.connect(path, timeout=SQLITE_TIMEOUT_SECONDS)
connection.row_factory = sqlite3.Row
connection.execute("pragma journal_mode = wal")
connection.execute(f"pragma busy_timeout = {SQLITE_BUSY_TIMEOUT_MS}")
connection.execute("pragma foreign_keys = on")
_enable_wal(connection)
_ensure_schema(connection)
return connection
constant/paths.py
+3 -3
View File
@@ -32,15 +32,15 @@ def missions_dir() -> Path:
def fleet_config_path() -> Path:
return config_root() / "fleet.yaml"
return config_root() / "fleet.json"
def models_config_path() -> Path:
return config_root() / "models.yaml"
return config_root() / "models.json"
def memory_config_path() -> Path:
return config_root() / "memory.yaml"
return config_root() / "memory.json"
def daemon_pid_path() -> Path:
constant/state.py
+11
View File
@@ -112,8 +112,19 @@ def ensure_runtime_dirs() -> None:
path.mkdir(parents=True, exist_ok=True)
def _legacy_config_path(path: Path) -> Path:
if path.suffix == ".json":
return path.with_suffix(".yaml")
return path
def _read_json_yaml(path: Path, default: dict[str, Any]) -> dict[str, Any]:
ensure_runtime_dirs()
legacy_path = _legacy_config_path(path)
if not path.exists() and legacy_path != path and legacy_path.exists():
return json.loads(legacy_path.read_text(encoding="utf-8"))
if not path.exists():
_write_json_yaml(path, default)
return deepcopy(default)
constant/tui.py
+479
View File
@@ -0,0 +1,479 @@
from __future__ import annotations
import curses
import json
import textwrap
import time
from pathlib import Path
from typing import Any
from .daemon import request as daemon_request
from .memory import list_decisions, memory_status, persona_markdown, rebuild_workspace_memory, summarize_mission
from .state import list_missions, load_fleet_config, mission_events_file
def _error_line(prefix: str, exc: Exception) -> str:
return f"{prefix}: {str(exc).strip() or exc.__class__.__name__}"
def _safe_memory_status(workspace: str) -> dict[str, Any]:
try:
return memory_status(workspace)
except Exception as exc: # noqa: BLE001
return {
"workspace": workspace,
"counts": {},
"error": _error_line("memory", exc),
}
def _safe_persona_lines() -> list[str]:
try:
return [
line[2:]
for line in persona_markdown().splitlines()
if line.startswith("- ")
]
except Exception as exc: # noqa: BLE001
return [_error_line("persona", exc)]
def _safe_decision_lines(workspace: str) -> list[str]:
try:
decisions = list_decisions(workspace=workspace).get("decisions", [])[:3]
return [f"{item['decision_id'].split(':')[-1]} {item['status']} {item['title']}" for item in decisions]
except Exception as exc: # noqa: BLE001
return [_error_line("decisions", exc)]
def _clip(text: str, width: int) -> str:
if width <= 0:
return ""
if len(text) <= width:
return text
if width <= 3:
return text[:width]
return text[: width - 3] + "..."
def _safe_addstr(stdscr: Any, y: int, x: int, text: str, attr: int = 0) -> None:
height, width = stdscr.getmaxyx()
if y < 0 or y >= height or x >= width:
return
text = _clip(text, width - x)
if not text:
return
try:
stdscr.addstr(y, x, text, attr)
except curses.error:
pass
def _draw_box(stdscr: Any, y: int, x: int, height: int, width: int, title: str, title_attr: int = 0) -> None:
if height < 3 or width < 4:
return
_safe_addstr(stdscr, y, x, "+" + ("-" * max(0, width - 2)) + "+")
for row in range(y + 1, y + height - 1):
_safe_addstr(stdscr, row, x, "|" + (" " * max(0, width - 2)) + "|")
_safe_addstr(stdscr, y + height - 1, x, "+" + ("-" * max(0, width - 2)) + "+")
_safe_addstr(stdscr, y, x + 2, f"[ {title} ]", title_attr)
def _write_wrapped(stdscr: Any, y: int, x: int, width: int, lines: list[str], attr: int = 0) -> int:
row = y
for line in lines:
wrapped = textwrap.wrap(line, max(8, width)) or [""]
for item in wrapped:
_safe_addstr(stdscr, row, x, item, attr)
row += 1
return row
def _recent_events(mission_id: str | None, limit: int = 8) -> list[str]:
if mission_id:
path = mission_events_file(mission_id)
if not path.exists():
return []
lines = path.read_text(encoding="utf-8").splitlines()[-limit:]
events = []
for raw in lines:
try:
event = json.loads(raw)
except json.JSONDecodeError:
continue
event_type = event.get("type", "event")
timestamp = str(event.get("timestamp", ""))[-9:-1]
payload = event.get("payload", {})
detail = payload.get("summary") or payload.get("step_id") or payload.get("machine") or payload.get("mission_id") or ""
line = f"{timestamp} {event_type}"
if detail:
line += f" :: {detail}"
events.append(line)
return events[-limit:]
events: list[str] = []
for mission in sorted(list_missions(), key=lambda item: item.get("updated_at", ""), reverse=True)[:5]:
events.extend(_recent_events(mission["mission_id"], limit=3))
return events[-limit:]
def _normalize_buddy(review: dict[str, Any] | None) -> dict[str, str]:
if not review:
return {
"verdict": "idle",
"confidence": "--",
"why": "No buddy review yet.",
"change": "none",
"memory": "idle",
}
if "verdict" in review:
change = review.get("change") or {}
target = "/".join(
part for part in [change.get("machine"), change.get("cli"), change.get("backend")] if part
) or "none"
memory = review.get("memory", {})
memory_hint = "store" if memory.get("store") else "ignore"
return {
"verdict": str(review.get("verdict", "warn")),
"confidence": str(review.get("confidence", "medium")),
"why": str(review.get("why", review.get("summary", ""))) or "No reason provided.",
"change": target,
"memory": memory_hint,
}
agrees = bool(review.get("agrees", True))
verdict = "agree" if agrees else ("reroute" if review.get("suggested_cli") or review.get("suggested_backend") else "warn")
change = "/".join(part for part in [review.get("suggested_cli"), review.get("suggested_backend")] if part) or "none"
return {
"verdict": verdict,
"confidence": "medium" if agrees else "high",
"why": str(review.get("summary", "No review summary.")),
"change": change,
"memory": "consider",
}
def _hexapus_lines() -> list[str]:
return [
" .-====-.",
" .-' .--. `-.",
" / ( oo ) \\",
" | \\/\\/ |",
" | .-____-. |",
" \\__/_/ || \\_\\__/",
" /_ /||\\ _\\",
" .-' \\/ || \\/ `-.",
" <_____/ || \\_____>",
" /___/\\___\\",
]
def _init_colors() -> dict[str, int]:
palette = {"base": 0, "accent": 0, "muted": 0, "warn": 0, "good": 0, "bad": 0, "hot": 0}
if not curses.has_colors():
return palette
curses.start_color()
try:
curses.use_default_colors()
except curses.error:
pass
curses.init_pair(1, curses.COLOR_CYAN, -1)
curses.init_pair(2, curses.COLOR_BLUE, -1)
curses.init_pair(3, curses.COLOR_YELLOW, -1)
curses.init_pair(4, curses.COLOR_GREEN, -1)
curses.init_pair(5, curses.COLOR_RED, -1)
curses.init_pair(6, curses.COLOR_MAGENTA, -1)
palette = {
"base": curses.color_pair(1),
"accent": curses.color_pair(6) | curses.A_BOLD,
"muted": curses.color_pair(2),
"warn": curses.color_pair(3) | curses.A_BOLD,
"good": curses.color_pair(4) | curses.A_BOLD,
"bad": curses.color_pair(5) | curses.A_BOLD,
"hot": curses.color_pair(6) | curses.A_BOLD,
}
return palette
def _verdict_attr(review: dict[str, str], colors: dict[str, int]) -> int:
verdict = review["verdict"]
if verdict in {"agree", "done"}:
return colors["good"]
if verdict in {"reroute", "warn"}:
return colors["warn"]
if verdict in {"block", "failed"}:
return colors["bad"]
return colors["accent"]
def _status_tag(status: str) -> str:
return {
"done": "DONE",
"planned": "PLAN",
"running": "RUN ",
"failed": "FAIL",
"needs_human": "HELP",
"pending": "PEND",
"draft": "DRAF",
}.get(status, status[:4].upper())
def _draw_header(stdscr: Any, workspace: str, mission_count: int, fleet_count: int, memory: dict[str, Any], health: dict[str, Any], colors: dict[str, int]) -> None:
memory_counts = memory.get("counts", {})
planner_backend = health.get("models", {}).get("planner", {}).get("backend", "heuristic")
buddy_backend = health.get("models", {}).get("buddy", {}).get("backend", "heuristic")
_safe_addstr(stdscr, 0, 2, "########## CONSTANT::DEMOSCENE::TUI ##########", colors["accent"])
_safe_addstr(
stdscr,
1,
2,
_clip(
f"workspace={workspace} fleet={fleet_count} missions={mission_count} docs={memory_counts.get('documents', 0)} chunks={memory_counts.get('chunks', 0)} decisions={memory_counts.get('decisions', 0)}",
max(20, stdscr.getmaxyx()[1] - 4),
),
colors["base"],
)
_safe_addstr(
stdscr,
2,
2,
_clip(
f"planner={planner_backend} buddy={buddy_backend} keys: j/k move z cockpit e index s summarize q quit",
max(20, stdscr.getmaxyx()[1] - 4),
),
colors["muted"],
)
memory_error = memory.get("error")
if memory_error:
_safe_addstr(
stdscr,
3,
2,
_clip(memory_error, max(20, stdscr.getmaxyx()[1] - 4)),
colors["bad"],
)
def _draw_missions(stdscr: Any, y: int, x: int, height: int, width: int, missions: list[dict[str, Any]], selected: int, colors: dict[str, int]) -> None:
_draw_box(stdscr, y, x, height, width, "Mission Deck", colors["accent"])
row = y + 1
if not missions:
_safe_addstr(stdscr, row, x + 2, "No missions yet. Use `Constant mission create`.", colors["muted"])
return
for index, mission in enumerate(missions[: max(1, height - 2)]):
marker = ">" if index == selected else " "
status = _status_tag(mission.get("status", "unknown"))
line = f"{marker} {status} {mission['mission_id'][:6]} {mission['title']}"
attr = colors["accent"] if index == selected else colors["base"]
_safe_addstr(stdscr, row, x + 1, _clip(line, width - 3), attr)
row += 1
if row >= y + height - 1:
break
def _draw_board(stdscr: Any, y: int, x: int, height: int, width: int, mission: dict[str, Any] | None, colors: dict[str, int]) -> None:
_draw_box(stdscr, y, x, height, width, "Mission Board", colors["accent"])
row = y + 1
if not mission:
lines = [
"No active mission selected.",
"",
"Try:",
"Constant mission create \"audit this repo\" --workspace \"$PWD\"",
]
_write_wrapped(stdscr, row, x + 2, width - 4, lines, colors["muted"])
return
row = _write_wrapped(stdscr, row, x + 2, width - 4, [mission["title"]], colors["hot"])
row += 1
row = _write_wrapped(stdscr, row, x + 2, width - 4, [mission.get("goal", "")], colors["base"])
row += 1
_safe_addstr(stdscr, row, x + 2, f"workspace: {mission.get('workspace', '-')}", colors["muted"])
row += 2
_safe_addstr(stdscr, row, x + 2, "steps:", colors["accent"])
row += 1
for step in mission.get("steps", [])[: max(1, height - (row - y) - 2)]:
line = f"[{_status_tag(step['status'])}] {step['step_id']} {step['machine']}/{step['cli']}/{step['backend']}"
_safe_addstr(stdscr, row, x + 2, _clip(line, width - 4), colors["base"])
row += 1
details = step.get("result_summary") or step.get("qwen_review") or step.get("title", "")
if details and row < y + height - 1:
row = _write_wrapped(stdscr, row, x + 4, width - 6, [details], colors["muted"])
if row >= y + height - 1:
break
def _draw_buddy(stdscr: Any, y: int, x: int, height: int, width: int, review: dict[str, str], memory: dict[str, Any], colors: dict[str, int]) -> None:
_draw_box(stdscr, y, x, height, width, "Hexapus Buddy Rail", colors["accent"])
row = y + 1
verdict_attr = _verdict_attr(review, colors)
for line in _hexapus_lines():
if row >= y + height - 1:
return
_safe_addstr(stdscr, row, x + 2, _clip(line, width - 4), verdict_attr)
row += 1
row += 1
for line in [
f"verdict: {review['verdict']}",
f"confidence: {review['confidence']}",
f"change: {review['change']}",
f"memory: {review['memory']}",
]:
if row >= y + height - 1:
return
_safe_addstr(stdscr, row, x + 2, _clip(line, width - 4), colors["base"])
row += 1
row += 1
row = _write_wrapped(stdscr, row, x + 2, width - 4, [review["why"]], colors["muted"])
row += 1
counts = memory.get("counts", {})
for line in [
f"persona facts : {counts.get('persona_facts', 0)}",
f"decisions : {counts.get('decisions', 0)}",
f"mission notes : {counts.get('mission_summaries', 0)}",
]:
if row >= y + height - 1:
return
_safe_addstr(stdscr, row, x + 2, _clip(line, width - 4), colors["base"])
row += 1
def _draw_timeline(stdscr: Any, y: int, x: int, height: int, width: int, events: list[str], persona_lines: list[str], decision_lines: list[str], colors: dict[str, int]) -> None:
_draw_box(stdscr, y, x, height, width, "Timeline + Memory", colors["accent"])
half = max(20, width // 2)
_safe_addstr(stdscr, y + 1, x + 2, "events", colors["hot"])
_safe_addstr(stdscr, y + 1, x + half, "memory echoes", colors["hot"])
row_left = y + 2
for event in events[: max(1, height - 3)]:
_safe_addstr(stdscr, row_left, x + 2, _clip(event, half - 4), colors["base"])
row_left += 1
if row_left >= y + height - 1:
break
row_right = y + 2
memory_lines = [*persona_lines[:3], *decision_lines[:3]]
for line in memory_lines[: max(1, height - 3)]:
_safe_addstr(stdscr, row_right, x + half, _clip(line, width - half - 2), colors["muted"])
row_right += 1
if row_right >= y + height - 1:
break
def _collect_snapshot(default_workspace: str, selected_index: int) -> dict[str, Any]:
missions = sorted(list_missions(), key=lambda item: item.get("updated_at", ""), reverse=True)
if missions:
selected_index = max(0, min(selected_index, len(missions) - 1))
selected_mission = missions[selected_index]
workspace = str(selected_mission.get("workspace", default_workspace))
else:
selected_mission = None
workspace = default_workspace
health = daemon_request("health", auto_start=False)
memory = _safe_memory_status(workspace)
review = _normalize_buddy(selected_mission.get("buddy_review") if selected_mission else None)
persona_lines = _safe_persona_lines()
decision_lines = _safe_decision_lines(workspace)
events = _recent_events(selected_mission["mission_id"] if selected_mission else None)
fleet = load_fleet_config()
return {
"missions": missions,
"selected_index": selected_index,
"selected_mission": selected_mission,
"workspace": workspace,
"health": health,
"memory": memory,
"review": review,
"persona_lines": persona_lines,
"decision_lines": decision_lines,
"events": events,
"fleet_count": len(fleet["machines"]),
}
def _run(stdscr: Any, workspace: str) -> dict[str, Any] | None:
try:
curses.curs_set(0)
except curses.error:
pass
stdscr.nodelay(False)
stdscr.timeout(250)
colors = _init_colors()
selected_index = 0
flash = ""
flash_until = 0.0
while True:
snapshot = _collect_snapshot(workspace, selected_index)
selected_index = snapshot["selected_index"]
stdscr.erase()
height, width = stdscr.getmaxyx()
_draw_header(
stdscr,
snapshot["workspace"],
len(snapshot["missions"]),
snapshot["fleet_count"],
snapshot["memory"],
snapshot["health"],
colors,
)
top = 4
timeline_height = min(10, max(7, height // 4))
main_height = max(10, height - top - timeline_height - 1)
left_width = max(28, min(34, width // 4))
right_width = max(34, min(42, width // 3))
center_width = max(28, width - left_width - right_width - 4)
_draw_missions(stdscr, top, 0, main_height, left_width, snapshot["missions"], selected_index, colors)
_draw_board(stdscr, top, left_width + 1, main_height, center_width, snapshot["selected_mission"], colors)
_draw_buddy(stdscr, top, left_width + center_width + 2, main_height, right_width, snapshot["review"], snapshot["memory"], colors)
_draw_timeline(stdscr, top + main_height, 0, timeline_height, width, snapshot["events"], snapshot["persona_lines"], snapshot["decision_lines"], colors)
if flash and time.time() < flash_until:
_safe_addstr(stdscr, height - 1, 2, _clip(flash, width - 4), colors["warn"])
else:
flash = ""
_safe_addstr(stdscr, height - 1, 2, "q quit | j/k select mission | z open cockpit | e rebuild memory | s summarize mission", colors["muted"])
stdscr.refresh()
key = stdscr.getch()
if key == -1:
continue
if key in (ord("q"), 27):
return None
if key in (ord("j"), curses.KEY_DOWN):
if snapshot["missions"]:
selected_index = min(len(snapshot["missions"]) - 1, selected_index + 1)
continue
if key in (ord("k"), curses.KEY_UP):
if snapshot["missions"]:
selected_index = max(0, selected_index - 1)
continue
if key == ord("z"):
return {"action": "cockpit", "workspace": snapshot["workspace"]}
if key == ord("e"):
try:
rebuild_workspace_memory(snapshot["workspace"], enroll=True)
flash = f"memory rebuilt for {snapshot['workspace']}"
except Exception as exc: # noqa: BLE001
flash = _error_line("memory rebuild failed", exc)
flash_until = time.time() + 2.5
continue
if key == ord("s") and snapshot["selected_mission"]:
try:
summarize_mission(snapshot["selected_mission"]["mission_id"])
flash = f"mission summarized: {snapshot['selected_mission']['mission_id']}"
except Exception as exc: # noqa: BLE001
flash = _error_line("summary failed", exc)
flash_until = time.time() + 2.5
continue
def run_tui(workspace: str) -> dict[str, Any] | None:
try:
rebuild_workspace_memory(workspace, enroll=True)
except Exception:
pass
return curses.wrapper(lambda stdscr: _run(stdscr, str(Path(workspace).expanduser().resolve())))
examples/fleet.example.json
+41
View File
@@ -0,0 +1,41 @@
{
"version": 1,
"local_machine": "command-center",
"machines": [
{
"label": "command-center",
"target": "local",
"auto_clis": ["codex", "vibe", "claude"],
"manual_clis": ["copilot"],
"backends": ["omc", "cli-local", "zellij"]
},
{
"label": "builder-a",
"target": "dev@builder-a",
"auto_clis": ["codex", "vibe", "claude"],
"manual_clis": ["copilot"],
"backends": ["cli-ssh", "zellij"]
},
{
"label": "builder-b",
"target": "dev@builder-b",
"auto_clis": ["codex", "vibe", "claude"],
"manual_clis": ["copilot"],
"backends": ["cli-ssh", "zellij"]
},
{
"label": "edge-a",
"target": "dev@edge-a",
"auto_clis": ["codex", "vibe", "claude"],
"manual_clis": ["copilot"],
"backends": ["cli-ssh", "zellij"]
},
{
"label": "lab-a",
"target": "dev@lab-a",
"auto_clis": ["codex", "vibe", "claude"],
"manual_clis": ["copilot"],
"backends": ["cli-ssh", "zellij"]
}
]
}