#!/usr/bin/env python3 """Nexus Playbook Editor — two-column TUI mirroring Playbook.jsx.""" import curses import httpx import json import os import signal import subprocess import sys import tempfile import textwrap import time from dataclasses import dataclass NEXUS_ROOT = os.path.expanduser("~/nexus-core") API_BASE = "http://localhost:8000" PID_FILE = os.path.join(NEXUS_ROOT, "runtime/pids/backend.pid") LOG_FILE = os.path.join(NEXUS_ROOT, "runtime/backend.log") VENV_PY = os.path.join(NEXUS_ROOT, "Promethean/bin/python3") BACKEND_CMD = ["uvicorn", "synapse.main:sio_app", "--host", "0.0.0.0", "--port", "8000"] # ── color pairs ─────────────────────────────────────────────────────────────── CP_HEADER = 1 # top bar black on cyan CP_SHORTS = 2 # shortcut bar black on white CP_SEL = 3 # selected list item black on cyan CP_LABEL = 4 # field labels cyan on default CP_OK = 5 # success text green on default CP_ERR = 6 # error text red on default CP_HINT = 7 # dim hints yellow on default CP_MAIN = 8 # MAIN badge green on default CP_FOCUS = 9 # focused inline field default + underline (attr, not bg) CP_DIV = 10 # dividers white on default # ── field IDs (right-panel tab order) ───────────────────────────────────────── F_TITLE, F_TAGS, F_GOAL, F_INST = 0, 1, 2, 3 FIELD_LABELS = {F_TITLE: "Title", F_TAGS: "Tags", F_GOAL: "Goal", F_INST: "Instructions"} EDITOR_FIELDS = {F_GOAL, F_INST} # ── pane IDs ────────────────────────────────────────────────────────────────── PANE_LEFT = "left" PANE_RIGHT = "right" # ─── backend ────────────────────────────────────────────────────────────────── def backend_running() -> bool: try: return httpx.get(f"{API_BASE}/", timeout=2.0).status_code == 200 except Exception: return False def start_backend(): os.makedirs(os.path.dirname(PID_FILE), exist_ok=True) python = VENV_PY if os.path.exists(VENV_PY) else sys.executable with open(LOG_FILE, "a") as log: proc = subprocess.Popen( [python, "-m"] + BACKEND_CMD, cwd=NEXUS_ROOT, stdout=log, stderr=log, start_new_session=True, ) with open(PID_FILE, "w") as f: f.write(str(proc.pid)) for _ in range(30): time.sleep(1) if backend_running(): return proc.pid return None def stop_backend(pid: int): try: os.kill(pid, signal.SIGTERM) except ProcessLookupError: return for _ in range(10): time.sleep(0.5) try: os.kill(pid, 0) except ProcessLookupError: break else: try: os.kill(pid, signal.SIGKILL) except ProcessLookupError: pass try: os.remove(PID_FILE) except FileNotFoundError: pass # ─── API ────────────────────────────────────────────────────────────────────── def api_list(): r = httpx.get(f"{API_BASE}/playbooks", timeout=5.0) r.raise_for_status() return r.json().get("playbooks", []) def api_get(pb_id): r = httpx.get(f"{API_BASE}/playbooks/{pb_id}", timeout=5.0) r.raise_for_status() return r.json() def api_patch(pb_id, data): r = httpx.patch(f"{API_BASE}/playbooks/{pb_id}", json=data, timeout=5.0) r.raise_for_status() return r.json() def api_create(data): r = httpx.post(f"{API_BASE}/playbooks", json=data, timeout=5.0) r.raise_for_status() return r.json() def api_delete(pb_id): r = httpx.delete(f"{API_BASE}/playbooks/{pb_id}", timeout=5.0) r.raise_for_status() # ─── editor helper ──────────────────────────────────────────────────────────── def open_in_editor(text: str): """Write text to a tmp file, open $EDITOR, return new text or None if unchanged.""" editor = os.environ.get("VISUAL") or os.environ.get("EDITOR") or "nano" with tempfile.NamedTemporaryFile( mode="w", suffix=".txt", prefix="nexus-", delete=False ) as f: f.write(text) fname = f.name try: subprocess.call([editor, fname]) with open(fname) as f: edited = f.read() stripped_edit = edited.strip() stripped_orig = text.strip() return stripped_edit if stripped_edit != stripped_orig else None finally: try: os.unlink(fname) except OSError: pass # ─── line editor ────────────────────────────────────────────────────────────── @dataclass class LineEdit: value: str = "" cursor: int = 0 def set(self, v: str): self.value = v self.cursor = len(v) def insert(self, ch: str): self.value = self.value[:self.cursor] + ch + self.value[self.cursor:] self.cursor += 1 def backspace(self): if self.cursor > 0: self.value = self.value[:self.cursor - 1] + self.value[self.cursor:] self.cursor -= 1 def delete_fwd(self): if self.cursor < len(self.value): self.value = self.value[:self.cursor] + self.value[self.cursor + 1:] def left(self): self.cursor = max(0, self.cursor - 1) def right(self): self.cursor = min(len(self.value), self.cursor + 1) def home(self): self.cursor = 0 def end(self): self.cursor = len(self.value) def view(self, width: int): """Return (visible_str, cursor_x_in_visible) scrolled so cursor stays visible.""" if len(self.value) <= width: return self.value, self.cursor if self.cursor < width: return self.value[:width], self.cursor start = self.cursor - width + 1 return self.value[start:start + width], width - 1 # ─── layout constants (row offsets inside right pane, relative to row 1) ────── # row 0 = global header # row 1 = pane sub-headers # row 2 = separator / spacer # row 3 = Title label # row 4 = Title input # row 5 = Tags label # row 6 = Tags input # row 7 = Goal label # row 8 = Goal line 1 # row 9 = Goal line 2 # row 10 = Goal line 3 # row 11 = Instructions label # row 12…h-3 = Instructions lines # row h-2 = status / save hint # row h-1 = shortcut bar # ─── TUI ────────────────────────────────────────────────────────────────────── class PlaybookEditor: SHORTS_LEFT = [("^X","Exit"), ("^N","New"), ("^D","Delete"), ("^S","Save"), ("Tab","→ Right")] SHORTS_RIGHT = [("^X","Exit"), ("^S","Save"), ("Tab","Next"), ("Esc","← List"), ("^D","Delete")] def __init__(self, stdscr): self.scr = stdscr # list state self.playbooks = [] self.filtered = [] self.list_idx = 0 self.list_scroll = 0 # search self.search = LineEdit() # form state self.selected_id = None self.title = LineEdit() self.tags = LineEdit() self.goal = "" self.inst = "" # UI self.pane = PANE_LEFT self.field = F_TITLE self.status = "" self.status_ok = True self._setup_colors() curses.curs_set(1) self.scr.keypad(True) self.scr.timeout(100) def _setup_colors(self): curses.start_color() curses.use_default_colors() curses.init_pair(CP_HEADER, curses.COLOR_BLACK, curses.COLOR_CYAN) curses.init_pair(CP_SHORTS, curses.COLOR_BLACK, curses.COLOR_WHITE) curses.init_pair(CP_SEL, curses.COLOR_BLACK, curses.COLOR_CYAN) curses.init_pair(CP_LABEL, curses.COLOR_CYAN, -1) curses.init_pair(CP_OK, curses.COLOR_GREEN, -1) curses.init_pair(CP_ERR, curses.COLOR_RED, -1) curses.init_pair(CP_HINT, curses.COLOR_YELLOW,-1) curses.init_pair(CP_MAIN, curses.COLOR_GREEN, -1) curses.init_pair(CP_FOCUS, -1, -1) # just a placeholder; we use A_UNDERLINE curses.init_pair(CP_DIV, curses.COLOR_WHITE, -1) # ── geometry ────────────────────────────────────────────────────────────── def dims(self): return self.scr.getmaxyx() def split(self, w): lw = max(22, w * 35 // 100) return lw, w - lw - 1 # (left_width, right_width) # ── primitives ──────────────────────────────────────────────────────────── def _put(self, row, col, text, attr=0): h, w = self.dims() if row < 0 or row >= h or col < 0 or col >= w: return try: self.scr.addstr(row, col, text[:max(0, w - col)], attr) except curses.error: pass def _fill(self, row, col, width, attr=0): self._put(row, col, " " * width, attr) def _clear_right_row(self, row): h, w = self.dims() lw, rw = self.split(w) rx = lw + 1 self._fill(row, rx, rw) # ── header & footer ─────────────────────────────────────────────────────── def draw_header(self): h, w = self.dims() count = f" {len(self.playbooks)} playbook{'s' if len(self.playbooks) != 1 else ''} " title = " NEXUS PLAYBOOKS " bar = title + " " * max(0, w - len(title) - len(count)) + count self.scr.attron(curses.color_pair(CP_HEADER) | curses.A_BOLD) self._put(0, 0, bar.ljust(w)[:w]) self.scr.attroff(curses.color_pair(CP_HEADER) | curses.A_BOLD) def draw_shorts(self): h, w = self.dims() pairs = self.SHORTS_RIGHT if self.pane == PANE_RIGHT else self.SHORTS_LEFT self.scr.attron(curses.color_pair(CP_SHORTS)) self._put(h - 1, 0, " " * (w - 1)) self.scr.attroff(curses.color_pair(CP_SHORTS)) x = 1 for key, lbl in pairs: if x + len(key) + len(lbl) + 3 >= w: break self.scr.attron(curses.A_REVERSE) self._put(h - 1, x, key) self.scr.attroff(curses.A_REVERSE) x += len(key) self.scr.attron(curses.color_pair(CP_SHORTS)) self._put(h - 1, x, f" {lbl} ") self.scr.attroff(curses.color_pair(CP_SHORTS)) x += len(lbl) + 3 # ── vertical divider ────────────────────────────────────────────────────── def draw_divider(self): h, w = self.dims() lw, _ = self.split(w) for row in range(1, h - 1): try: self.scr.addch(row, lw, curses.ACS_VLINE, curses.color_pair(CP_DIV)) except curses.error: pass # ── left panel ──────────────────────────────────────────────────────────── def draw_left(self): h, w = self.dims() lw, _ = self.split(w) # Row 1: search SRCH_PREFIX = " / " sw = lw - len(SRCH_PREFIX) - 1 self._fill(1, 0, lw) self.scr.attron(curses.color_pair(CP_LABEL) | curses.A_BOLD) self._put(1, 0, SRCH_PREFIX) self.scr.attroff(curses.color_pair(CP_LABEL) | curses.A_BOLD) s_disp, _ = self.search.view(sw) attr = curses.A_UNDERLINE if self.pane == PANE_LEFT else 0 self._put(1, len(SRCH_PREFIX), s_disp.ljust(sw)[:sw], attr) # Row 2: separator self._fill(2, 0, lw) self.scr.attron(curses.color_pair(CP_LABEL)) self._put(2, 0, "─" * (lw - 1)) self.scr.attroff(curses.color_pair(CP_LABEL)) # Rows 3 … h-2: list ls, le = 3, h - 1 lh = le - ls if self.list_idx < self.list_scroll: self.list_scroll = self.list_idx elif self.list_idx >= self.list_scroll + lh: self.list_scroll = self.list_idx - lh + 1 for i in range(lh): row = ls + i idx = self.list_scroll + i self._fill(row, 0, lw) if idx >= len(self.filtered): continue pb = self.filtered[idx] t = pb.get("title", "Untitled") is_main = idx == 0 and not self.search.value.strip() is_cur = idx == self.list_idx badge = " MAIN" if is_main else "" avail = lw - 5 - len(badge) if len(t) > avail: t = t[:avail - 1] + "…" prefix = " ▶ " if is_cur else " " if is_cur and self.pane == PANE_LEFT: self.scr.attron(curses.color_pair(CP_SEL) | curses.A_BOLD) self._fill(row, 0, lw - 1) self._put(row, 0, prefix + t) self.scr.attroff(curses.color_pair(CP_SEL) | curses.A_BOLD) elif pb.get("id") == self.selected_id: self._put(row, 0, prefix + t, curses.A_BOLD) else: self._put(row, 0, prefix + t) if badge: self.scr.attron(curses.color_pair(CP_MAIN) | curses.A_BOLD) self._put(row, lw - len(badge) - 1, badge) self.scr.attroff(curses.color_pair(CP_MAIN) | curses.A_BOLD) # ── right panel ─────────────────────────────────────────────────────────── def draw_right(self): h, w = self.dims() lw, rw = self.split(w) rx = lw + 1 fw = rw - 4 # usable field width def clr(row): self._clear_right_row(row) def lbl(row, text): clr(row) self.scr.attron(curses.color_pair(CP_LABEL) | curses.A_BOLD) self._put(row, rx + 1, text) self.scr.attroff(curses.color_pair(CP_LABEL) | curses.A_BOLD) def inline_row(row, le: LineEdit, fid: int): clr(row) focused = self.pane == PANE_RIGHT and self.field == fid disp, _ = le.view(fw) attr = curses.A_UNDERLINE if focused else 0 self._put(row, rx + 2, disp.ljust(fw)[:fw], attr) def editor_rows(start_row, text: str, fid: int, n_rows: int): focused = self.pane == PANE_RIGHT and self.field == fid lines = _wrap(text, fw) if text.strip() else [""] for i in range(n_rows): r = start_row + i if r >= h - 2: break clr(r) line = lines[i] if i < len(lines) else "" if focused and i == 0: hint = " ↵ edit" self._put(r, rx + 2, line[:fw - len(hint)]) self.scr.attron(curses.color_pair(CP_HINT) | curses.A_BOLD) self._put(r, rx + 2 + min(len(line), fw - len(hint)), hint) self.scr.attroff(curses.color_pair(CP_HINT) | curses.A_BOLD) else: self._put(r, rx + 2, line[:fw]) # Row 1 — pane header clr(1) hdr = " New Playbook" if self.selected_id is None else " Edit Playbook" self.scr.attron(curses.color_pair(CP_LABEL) | curses.A_BOLD) self._put(1, rx + 1, hdr) self.scr.attroff(curses.color_pair(CP_LABEL) | curses.A_BOLD) if self.selected_id is not None: del_hint = "^D Delete" self.scr.attron(curses.color_pair(CP_ERR)) self._put(1, rx + rw - len(del_hint) - 2, del_hint) self.scr.attroff(curses.color_pair(CP_ERR)) # Row 2 — spacer clr(2) # Rows 3–4 — Title lbl(3, "Title") inline_row(4, self.title, F_TITLE) # Rows 5–6 — Tags lbl(5, "Tags (comma-separated)") inline_row(6, self.tags, F_TAGS) # Rows 7–10 — Goal lbl(7, "Goal") editor_rows(8, self.goal, F_GOAL, 3) # Rows 11–(h-3) — Instructions lbl(11, "Instructions") inst_rows = max(1, h - 3 - 12) editor_rows(12, self.inst, F_INST, inst_rows) # Row h-2 — status / save hint clr(h - 2) if self.status: cp = CP_OK if self.status_ok else CP_ERR self.scr.attron(curses.color_pair(cp) | curses.A_BOLD) self._put(h - 2, rx + 2, self.status[:rw - 4]) self.scr.attroff(curses.color_pair(cp) | curses.A_BOLD) else: self.scr.attron(curses.color_pair(CP_HINT)) self._put(h - 2, rx + 2, "^S to save") self.scr.attroff(curses.color_pair(CP_HINT)) # ── cursor ──────────────────────────────────────────────────────────────── def _place_cursor(self): h, w = self.dims() lw, rw = self.split(w) rx = lw + 1 fw = rw - 4 if self.pane == PANE_LEFT: _, cx = self.search.view(lw - 5) try: self.scr.move(1, 4 + cx) except curses.error: pass elif self.pane == PANE_RIGHT: if self.field == F_TITLE: _, cx = self.title.view(fw) try: self.scr.move(4, rx + 2 + cx) except curses.error: pass elif self.field == F_TAGS: _, cx = self.tags.view(fw) try: self.scr.move(6, rx + 2 + cx) except curses.error: pass elif self.field == F_GOAL: try: self.scr.move(8, rx + 2) except curses.error: pass elif self.field == F_INST: try: self.scr.move(12, rx + 2) except curses.error: pass # ── full redraw ─────────────────────────────────────────────────────────── def draw(self): self.draw_header() self.draw_divider() self.draw_left() self.draw_right() self.draw_shorts() self._place_cursor() self.scr.refresh() # ── data helpers ────────────────────────────────────────────────────────── def _set_status(self, msg, ok=True): self.status = msg self.status_ok = ok def load_list(self): try: self.playbooks = api_list() self._filter() except Exception as e: self._set_status(f"Load error: {e}", ok=False) def _filter(self): q = self.search.value.strip().lower() self.filtered = ( list(self.playbooks) if not q else [ p for p in self.playbooks if q in p.get("title","").lower() or q in p.get("goal","").lower() or any(q in t.lower() for t in p.get("tags", [])) ] ) self.list_idx = min(self.list_idx, max(0, len(self.filtered) - 1)) def _load_form(self, pb_summary): try: pb = api_get(str(pb_summary["id"])) except Exception: pb = pb_summary self.selected_id = pb.get("id") self.title.set(pb.get("title", "")) self.tags.set(", ".join(pb.get("tags", []))) self.goal = pb.get("goal", "") self.inst = pb.get("instructions", "") self.status = "" def _clear_form(self): self.selected_id = None self.title.set(""); self.tags.set("") self.goal = ""; self.inst = "" self.status = "" self.field = F_TITLE @property def _current(self): if self.filtered and self.list_idx < len(self.filtered): return self.filtered[self.list_idx] return None # ── actions ─────────────────────────────────────────────────────────────── def _resume(self): self.scr.refresh() curses.curs_set(1) def do_select(self): pb = self._current if pb: self._load_form(pb) self.pane = PANE_RIGHT self.field = F_TITLE def do_edit_field(self): text = self.goal if self.field == F_GOAL else self.inst curses.endwin() result = open_in_editor(text) self._resume() if result is not None: if self.field == F_GOAL: self.goal = result else: self.inst = result self._set_status("Field updated — ^S to save.") def do_save(self): tags_list = [t.strip() for t in self.tags.value.split(",") if t.strip()] payload = { "title": self.title.value.strip(), "goal": self.goal.strip(), "instructions": self.inst.strip(), "tags": tags_list, } if not all([payload["title"], payload["goal"], payload["instructions"]]): self._set_status("Title, goal, and instructions are all required.", ok=False) return try: if self.selected_id: api_patch(str(self.selected_id), payload) self._set_status(f"Updated: {payload['title']}") else: api_create(payload) self._set_status(f"Created: {payload['title']}") self.load_list() self._clear_form() self.pane = PANE_LEFT except Exception as e: self._set_status(f"Save failed: {e}", ok=False) def do_delete(self): if not self.selected_id: self._set_status("Nothing selected.", ok=False) return name = self.title.value or "this playbook" if not self._confirm(f'Delete "{name}"? [y/N] '): self._set_status("Cancelled.") return try: api_delete(str(self.selected_id)) self._set_status(f"Deleted: {name}") self._clear_form() self.load_list() self.pane = PANE_LEFT except Exception as e: self._set_status(f"Delete failed: {e}", ok=False) def _confirm(self, prompt: str) -> bool: h, w = self.dims() lw, rw = self.split(w) rx = lw + 1 self._clear_right_row(h - 2) self.scr.attron(curses.color_pair(CP_SHORTS) | curses.A_BOLD) self._put(h - 2, rx, " " * (rw - 1)) self._put(h - 2, rx + 2, prompt[:rw - 3]) self.scr.attroff(curses.color_pair(CP_SHORTS) | curses.A_BOLD) self.scr.refresh() self.scr.timeout(-1) key = self.scr.getch() self.scr.timeout(100) return key in (ord("y"), ord("Y")) # ── key handlers ────────────────────────────────────────────────────────── def _line_key(self, le: LineEdit, key: int): if key in (127, curses.KEY_BACKSPACE, 8): le.backspace() elif key == curses.KEY_DC: le.delete_fwd() elif key == curses.KEY_LEFT: le.left() elif key == curses.KEY_RIGHT: le.right() elif key == curses.KEY_HOME: le.home() elif key == curses.KEY_END: le.end() elif 32 <= key <= 126: le.insert(chr(key)) def key_left(self, key): if key == curses.KEY_UP: self.list_idx = max(0, self.list_idx - 1) self.status = "" elif key == curses.KEY_DOWN: self.list_idx = min(len(self.filtered) - 1, self.list_idx + 1) self.status = "" elif key == curses.KEY_PPAGE: h, _ = self.dims() self.list_idx = max(0, self.list_idx - (h - 5)) elif key == curses.KEY_NPAGE: h, _ = self.dims() self.list_idx = min(len(self.filtered) - 1, self.list_idx + (h - 5)) elif key in (10, 13): # Enter → select + right self.do_select() elif key == 9: # Tab → right pane self.do_select() elif key in (127, curses.KEY_BACKSPACE, 8): self.search.backspace(); self._filter() elif key == curses.KEY_LEFT: self.search.left() elif key == curses.KEY_RIGHT: self.search.right() elif 32 <= key <= 126: self.search.insert(chr(key)); self._filter() def key_right(self, key): if key == 27: # Esc → back to list self.pane = PANE_LEFT elif key == 9: # Tab → next field self.field = (self.field + 1) % 4 elif key in (10, 13): if self.field in EDITOR_FIELDS: self.do_edit_field() else: self.field = (self.field + 1) % 4 elif self.field == F_TITLE: self._line_key(self.title, key) elif self.field == F_TAGS: self._line_key(self.tags, key) # ── main loop ───────────────────────────────────────────────────────────── def run(self): self.load_list() while True: self.draw() try: key = self.scr.getch() except KeyboardInterrupt: break if key == -1: continue # global if key == 24: break # ^X elif key == 19: self.do_save() # ^S elif key == 4: self.do_delete() # ^D elif key == 14: # ^N self._clear_form() self.pane = PANE_RIGHT self.field = F_TITLE elif key == 18: self.load_list() # ^R elif self.pane == PANE_LEFT: self.key_left(key) elif self.pane == PANE_RIGHT: self.key_right(key) # ─── util ───────────────────────────────────────────────────────────────────── def _wrap(text: str, width: int) -> list: out = [] for line in text.splitlines(): out.extend(textwrap.wrap(line, width) or [""]) return out or [""] # ─── entry ──────────────────────────────────────────────────────────────────── def main(): print("Checking Nexus backend…", flush=True) started_pid = None if not backend_running(): print("Synapse not running — starting backend…", flush=True) started_pid = start_backend() if started_pid is None: print(f"ERROR: backend failed to start. Logs: {LOG_FILE}") sys.exit(1) print("Backend ready.", flush=True) else: print("Backend is running.", flush=True) time.sleep(0.2) try: curses.wrapper(lambda s: PlaybookEditor(s).run()) finally: if started_pid is not None: print("\nShutting down Synapse backend…", flush=True) stop_backend(started_pid) print("Done.", flush=True) if __name__ == "__main__": main()