diff --git a/bin/muse_choice_watcher.py b/bin/muse_choice_watcher.py index 4e7317d..28767da 100755 --- a/bin/muse_choice_watcher.py +++ b/bin/muse_choice_watcher.py @@ -28,6 +28,7 @@ Usage: """ import argparse +import fcntl import hashlib import json import os @@ -53,6 +54,7 @@ POLL_INTERVAL = 0.5 STABILITY_POLLS = 2 MAX_ANSWERS_PER_HOUR = 20 ANSWERED_TTL_SECONDS = 600 # identical prompt back after 10m => stuck, allow one recovery answer +CLAIM_TTL_SECONDS = 30 # concurrent-claim window: bounds wedge if winner dies pre-send RULES_FILE = os.path.join(REPO_ROOT, "muse-choices-rules.json") HOLD_WINDOW_SECONDS = 120 # D3: short hold window, then expire to approve NEGATIVE_KEYS = {"muse-approval": "2", "yn": "n"} # D2 deny keys @@ -163,6 +165,13 @@ def logfile_for(socket_path, pane_id): ) +def answered_file_for(socket_path, pane_id): + """On-disk answered-sig store: restarts must not re-answer prompts.""" + return os.path.join( + STATE_DIR, "%s-%s-%s.answered.json" % (FILE_PREFIX, slug_socket(socket_path), clean_pane(pane_id)) + ) + + class WatcherLog: """Never-raising JSON-lines logger with best-effort rotation.""" @@ -624,21 +633,49 @@ def _child_pids(pid): def muse_approval_flags(cmd_argv): - """Approval posture of a muse argv: {"auto_approve", "flags"}.""" + """Approval posture of a muse argv. + + Returns {"auto_approve", "flags", "profile", "mode", "bypass"}. + profile is the --permission-profile id (built-ins :read-only, + :standard, :unrestricted) or None; mode is "yolo" for --yolo, the + profile id when one was passed, else "default"; bypass is True + when tool-approval prompts are disabled at launch (yolo, + --disable-approval, or --approval-mode=never), in which case the + pane shows no approval dialogs (questions may still render, so the + watcher stays active). + """ flags = [] argv = cmd_argv or [] if "--yolo" in argv: flags.append("yolo") if "--disable-approval" in argv: flags.append("disable-approval") + if "--disable-sandbox" in argv: + flags.append("disable-sandbox") + if "--trust-workspace" in argv: + flags.append("trust-workspace") + profile = None for i, arg in enumerate(argv): if arg == "--approval-mode" and i + 1 < len(argv): flags.append("approval-mode=%s" % argv[i + 1]) elif arg.startswith("--approval-mode="): flags.append("approval-mode=%s" % arg.split("=", 1)[1]) + elif arg == "--permission-profile" and i + 1 < len(argv): + profile = argv[i + 1] + flags.append("permission-profile=%s" % profile) + elif arg.startswith("--permission-profile="): + profile = arg.split("=", 1)[1] + flags.append("permission-profile=%s" % profile) auto = ("yolo" in flags or "disable-approval" in flags or "approval-mode=never" in flags) - return {"auto_approve": auto, "flags": flags} + if "yolo" in flags: + mode = "yolo" + elif profile is not None: + mode = profile + else: + mode = "default" + return {"auto_approve": auto, "flags": flags, "profile": profile, + "mode": mode, "bypass": auto} def launch_opt_out(cmd_argv): @@ -688,9 +725,11 @@ def pane_muse_argv(socket_path, pane_id): # Minimum pane geometry for reliable approval rendering. Empirically # derived: a 35x7 tile drops approval text the matcher needs, while -# 35x35/36x35/71x27 panes answer cleanly. Below either bound the pane -# is flagged squeezed (see `box runtime layout` / `spread`). -MIN_APPROVAL_WIDTH = 40 +# 35x35/36x35/71x27 panes answer cleanly (width 35 works when tall +# enough; capture uses -J so wrapping is width-independent). Below +# either bound the pane is flagged squeezed (see `box runtime layout` / +# `spread`). +MIN_APPROVAL_WIDTH = 35 MIN_APPROVAL_HEIGHT = 12 # Session naming convention (see NODES.md): ---- @@ -749,7 +788,8 @@ def runtime_rows(socket_path): or pane_height < MIN_APPROVAL_HEIGHT)) node = node_from_session(session) is_muse = "muse-bin" in cmd or "muse-code" in cmd - posture = {"auto_approve": None, "flags": []} + posture = {"auto_approve": None, "flags": [], "profile": None, + "mode": None, "bypass": None} if is_muse and pane_pid: candidates = [pane_pid] + _child_pids(pane_pid) for cand in candidates: @@ -763,7 +803,7 @@ def runtime_rows(socket_path): continue st = runtime_state(text) match = st["match"] or {} - watcher_pid = is_running(socket_path, pane_id) + watcher_pid = watcher_alive(socket_path, pane_id) rows.append({ "socket": socket_path, "session": session, "window": window, "pane": pane_id, "cmd": cmd, @@ -773,6 +813,9 @@ def runtime_rows(socket_path): "squeezed": squeezed, "auto_approve": posture["auto_approve"], "approval_flags": posture["flags"], + "permission_mode": posture["mode"], + "permission_profile": posture["profile"], + "permission_bypass": posture["bypass"], "state": st["state"], "prompt_kind": match.get("kind"), "prompt_key": match.get("key"), @@ -813,14 +856,124 @@ def pane_state(socket_path, pane_id): class WatcherState: - """Tracks prompt stability and once-per-prompt answering.""" + """Tracks prompt stability and once-per-prompt answering. - def __init__(self): + When persist_path is set, answered sigs survive restarts (atomic + JSON store): a daemon that restarts while an answered prompt is + still visible must not answer it again (observed live: same sig + re-answered minutes later, stray "1" landing in the input box). + """ + + def __init__(self, persist_path=None): self.pending_sig = None self.stable_count = 0 self.answered_sigs = {} self.answer_times = [] self.last_capped_sig = None + self._persist_path = persist_path + if persist_path: + self._load_answered() + + def _load_answered(self): + try: + with open(self._persist_path) as f: + data = json.load(f) + except Exception: + return + if not isinstance(data, dict): + return + now = time.time() + for sig, ts in data.items(): + if (isinstance(sig, str) and isinstance(ts, (int, float)) + and now - ts < ANSWERED_TTL_SECONDS): + self.answered_sigs[sig] = ts + + def _save_answered(self): + if not self._persist_path: + return + try: + tmp = "%s.tmp.%d" % (self._persist_path, os.getpid()) + with open(tmp, "w") as f: + json.dump(self.answered_sigs, f) + os.replace(tmp, self._persist_path) + except Exception: + pass + + def refresh_answered(self): + """Merge on-disk answered sigs into memory. Never raises. + + Cross-process once-per-prompt: a peer watcher that answered + after our startup persisted its sig; re-reading before we type + suppresses the duplicate ('11' in the input box). + """ + if not self._persist_path: + return + try: + with open(self._persist_path) as f: + data = json.load(f) + except Exception: + return + if not isinstance(data, dict): + return + now = time.time() + for sig, ts in data.items(): + if (isinstance(sig, str) and isinstance(ts, (int, float)) + and now - ts < ANSWERED_TTL_SECONDS): + if sig not in self.answered_sigs: + self.answered_sigs[sig] = ts + + def try_claim(self, sig, now=None): + """Atomically claim a sig for this process. True iff we won. + + O_CREAT|O_EXCL makes the first claimer win even when two + watchers reach the same stable prompt in the same poll window; + the loser suppresses instead of double-typing. Claims expire + after CLAIM_TTL_SECONDS (concurrent window only; stuck-dialog + recovery is governed by the answered store's longer TTL), so a + winner that dies between claim and send wedges at most briefly. + Memory-only states (no persist path) always win. Never raises. + """ + if not self._persist_path: + return True + now = time.time() if now is None else now + base = os.path.basename(self._persist_path) + directory = os.path.dirname(self._persist_path) or STATE_DIR + # Prune expired claims for this pane (best-effort). + try: + for name in os.listdir(directory): + if not name.startswith(base + ".") or not name.endswith(".claim"): + continue + p = os.path.join(directory, name) + try: + with open(p) as f: + rec = json.load(f) + ts = float(rec.get("ts", 0)) + except Exception: + ts = 0 + try: + if now - ts >= CLAIM_TTL_SECONDS: + os.remove(p) + except OSError: + pass + except Exception: + pass + path = "%s.%s.claim" % (self._persist_path, sig) + try: + fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY) + except FileExistsError: + return False + except OSError: + return True # claim store unavailable: fail open, send once + try: + os.write(fd, json.dumps({"pid": os.getpid(), + "ts": now}).encode()) + except Exception: + pass + try: + os.close(fd) + except Exception: + pass + return True def _prune_times(self, now): cutoff = now - 3600 @@ -869,6 +1022,7 @@ class WatcherState: self.answer_times.append(now) self.pending_sig = None self.stable_count = 0 + self._save_answered() def _tmux(socket_path, *args, timeout=5): @@ -887,9 +1041,15 @@ def pane_exists(socket_path, pane_id): def capture_pane(socket_path, pane_id, history=80): + """Capture pane text with wrapped rows joined (-J). + + -J makes matching width-independent: narrow panes wrap the same + dialog onto more physical rows, which otherwise pushes cue/option + spans apart and breaks the matcher. + """ try: - r = _tmux(socket_path, "capture-pane", "-p", "-t", pane_id, "-S", "-%d" % history, - timeout=5) + r = _tmux(socket_path, "capture-pane", "-p", "-J", "-t", pane_id, + "-S", "-%d" % history, timeout=5) if r.returncode != 0: return None return r.stdout @@ -1065,10 +1225,53 @@ def list_holds(): return out +def _peer_suppressed(state, sig, now, log): + """True when a peer watcher already owns sig; suppress our send. + + Checks the shared on-disk answered store first (peer answered + earlier and persisted), then attempts an atomic claim (peer racing + us in the same window). On suppression the pending prompt is + reset so later polls re-observe cleanly; answered-store hits are + already merged into memory, claim-loss is not recorded (peer's + send owns it, and its claim expires quickly if it dies). Never + raises; memory-only states never suppress. + """ + try: + state.refresh_answered() + except Exception: + pass + try: + ts = state.answered_sigs.get(sig) + if ts is not None and now - ts < ANSWERED_TTL_SECONDS: + try: + log.log("info", "duplicate suppressed (peer answered)", + sig=sig) + except Exception: + pass + state.pending_sig = None + state.stable_count = 0 + return True + except Exception: + pass + try: + if not state.try_claim(sig, now): + try: + log.log("info", "duplicate suppressed (peer claimed)", + sig=sig) + except Exception: + pass + state.pending_sig = None + state.stable_count = 0 + return True + except Exception: + pass + return False + + def _check_hold(socket_path, pane_id, state, match, now, log): """Rule-hold gate. Returns None (fresh: evaluate rules), "released" (hold over: approve WITHOUT re-evaluating, else expiry would - re-hold forever), "held", or "denied". + re-hold forever), "held", "denied", or "duplicate-suppressed". The holdfile is the single source of truth (crash-safe, box-visible): present + fresh => suppress; directive deny => deny @@ -1093,6 +1296,8 @@ def _check_hold(socket_path, pane_id, state, match, now, log): log.log("warn", "resolve-deny refused: D2 never denies questions", sig=sig, kind=match["kind"]) return "held" + if _peer_suppressed(state, sig, now, log): + return "duplicate-suppressed" ok = send_answer(socket_path, pane_id, neg, enter=True) clear_hold(socket_path, pane_id) state.record_answer(sig, now) @@ -1132,8 +1337,8 @@ def _poll_once(socket_path, pane_id, state, log, dry_run=False): now = time.time() gate = _check_hold(socket_path, pane_id, state, match, now, log) \ if match is not None else None - if gate in ("held", "denied"): - if gate == "denied": + if gate in ("held", "denied", "duplicate-suppressed"): + if gate in ("denied", "duplicate-suppressed"): state.pending_sig = None state.stable_count = 0 return gate @@ -1175,6 +1380,8 @@ def _poll_once(socket_path, pane_id, state, log, dry_run=False): rule=rule["id"] if rule else None) state.record_answer(match["sig"], now) return "dry-denied" + if _peer_suppressed(state, match["sig"], now, log): + return "duplicate-suppressed" ok = send_answer(socket_path, pane_id, neg, enter=True) state.record_answer(match["sig"], now) log.log("info" if ok else "error", "denied %s" % neg, @@ -1216,6 +1423,8 @@ def _poll_once(socket_path, pane_id, state, log, dry_run=False): options=match["options"], cue=match["cue"]) state.record_answer(match["sig"], now) return "dry-answered" + if _peer_suppressed(state, match["sig"], now, log): + return "duplicate-suppressed" ok = send_answer(socket_path, pane_id, key, enter=match.get("enter", True)) state.record_answer(match["sig"], now) @@ -1242,12 +1451,45 @@ def _poll_once(socket_path, pane_id, state, log, dry_run=False): return "waiting" +def _log_posture(socket_path, pane_id, log): + """Log the pane's permission posture once at watcher start. + + The watcher answers with per-choice logging in every mode (that + trail is the default path and informs policy); a bypass posture + (yolo / approval disabled) additionally gets a box audit record, + since the session then makes choices outside the trail. Never + raises. + """ + try: + posture = muse_approval_flags(pane_muse_argv(socket_path, pane_id)) + except Exception: + return + try: + log.log("info", "pane posture", mode=posture["mode"], + bypass=posture["bypass"], flags=posture["flags"]) + except Exception: + pass + try: + if posture["bypass"] or posture["mode"] not in ("default", None): + audit("muse-choice-posture", + name="%s:%s" % (os.path.basename(socket_path), pane_id), + extra={"socket": socket_path, "pane": pane_id, + "mode": posture["mode"], + "profile": posture["profile"], + "bypass": posture["bypass"], + "flags": posture["flags"]}) + except Exception: + pass + + def watch_loop(socket_path, pane_id, dry_run=False): """Main daemon loop. Returns only when the pane is gone or signalled.""" log = WatcherLog(logfile_for(socket_path, pane_id)) - state = WatcherState() + state = WatcherState( + persist_path=answered_file_for(socket_path, pane_id)) log.log("info", "watcher started", socket=socket_path, pane=pane_id, dry_run=dry_run, pid=os.getpid()) + _log_posture(socket_path, pane_id, log) polls = 0 answers = 0 last_heartbeat = time.time() @@ -1346,6 +1588,30 @@ def is_running(socket_path, pane_id): return None +def watcher_alive(socket_path, pane_id): + """Pid of the live watcher for a pane, pidfile or orphan. + + is_running covers the normal pidfile case; _watch_procs catches an + identical watcher alive with a lost pidfile (/tmp cleaned under + it). Starters must consult this (not is_running alone) or they + spawn a second daemon onto the same pane (double answers). Never + raises; None means no live watcher. + """ + try: + pid = is_running(socket_path, pane_id) + if pid: + return pid + except Exception: + pass + try: + for proc in _watch_procs(): + if proc["socket"] == socket_path and proc["pane"] == pane_id: + return proc["pid"] + except Exception: + pass + return None + + def _daemonize(): """Double-fork away from the controlling terminal (survives shell exit).""" if os.fork() != 0: @@ -1361,27 +1627,85 @@ def _daemonize(): os.close(devnull) +_PIDFILE_LOCK_FH = None + + def _claim_pidfile(pidfile): - """Claim a pidfile for this process. Returns False if another live - watcher owns it (concurrent box + timer starts must not pile up).""" + """Claim a pidfile for this process. Returns False if another + starter holds it. Kernel-enforced: the winner holds an exclusive + flock for its whole lifetime, so concurrent box + timer starts can + never pile two daemons onto one pane (double answers, '11' in the + input box). Call _release_pidfile() on exit.""" + global _PIDFILE_LOCK_FH me = os.getpid() try: - with open(pidfile) as f: - other = int(f.read().strip()) + fh = open(pidfile, "a+") + except OSError: + return False + try: + fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB) + except (OSError, IOError): + fh.close() + return False + try: + fh.seek(0) + content = fh.read().strip() + other = int(content) if content else None except Exception: other = None if other and other != me and _pid_alive(other) and _pid_is_watcher(other): + # Live foreign owner (e.g. a daemon from before locking existed, + # or a pid recycled into a watcher): yield to it. + try: + fcntl.flock(fh, fcntl.LOCK_UN) + except Exception: + pass + fh.close() return False try: - with open(pidfile, "w") as f: - f.write(str(me)) + fh.seek(0) + fh.truncate() + fh.write(str(me)) + fh.flush() except Exception: pass + if _PIDFILE_LOCK_FH is not None: + try: + _PIDFILE_LOCK_FH.close() + except Exception: + pass + _PIDFILE_LOCK_FH = fh return True +def _release_pidfile(pidfile): + """Drop the claim taken by _claim_pidfile: remove the pidfile only + if we still own it, then release the lock. Never raises.""" + global _PIDFILE_LOCK_FH + try: + with open(pidfile) as f: + owner = f.read().strip() + except Exception: + owner = "" + if owner == str(os.getpid()): + try: + os.remove(pidfile) + except OSError: + pass + if _PIDFILE_LOCK_FH is not None: + try: + fcntl.flock(_PIDFILE_LOCK_FH, fcntl.LOCK_UN) + except Exception: + pass + try: + _PIDFILE_LOCK_FH.close() + except Exception: + pass + _PIDFILE_LOCK_FH = None + + def start_watcher(socket_path, pane_id, dry_run=False): - pid = is_running(socket_path, pane_id) + pid = watcher_alive(socket_path, pane_id) if pid: return {"ok": False, "status": "already_running", "pid": pid} if not pane_exists(socket_path, pane_id): @@ -1400,7 +1724,7 @@ def start_watcher(socket_path, pane_id, dry_run=False): def stop_watcher(socket_path, pane_id, timeout=5): - pid = is_running(socket_path, pane_id) + pid = watcher_alive(socket_path, pane_id) if not pid: return {"ok": True, "status": "not_running"} try: @@ -1448,7 +1772,7 @@ def _start_detached(socket_path, pane_id, dry_run=False): os._exit(0) os.waitpid(pid, 0) time.sleep(0.2) - return is_running(socket_path, pane_id) is not None + return watcher_alive(socket_path, pane_id) is not None def start_all(dry_run=False, sockets=None): @@ -1461,7 +1785,7 @@ def start_all(dry_run=False, sockets=None): if not panes: results.append({"socket": sock, "status": "no_muse_panes"}) for pane in panes: - if is_running(sock, pane): + if watcher_alive(sock, pane): results.append({"socket": sock, "pane": pane, "status": "already_running"}) continue @@ -1524,7 +1848,7 @@ def reconcile(sockets=None): if not os.path.exists(sock): continue for pane in muse_panes(sock): - if is_running(sock, pane): + if watcher_alive(sock, pane): already.append("%s:%s" % (sock, pane)) continue if _start_detached(sock, pane, dry_run=desired["dry_run"]): @@ -1660,7 +1984,7 @@ def main(argv=None): args = ap.parse_args(argv) if args.cmd == "start": - existing = is_running(args.socket, args.pane) + existing = watcher_alive(args.socket, args.pane) if existing: print(json.dumps({"ok": True, "status": "already_running", "pid": existing, @@ -1674,7 +1998,7 @@ def main(argv=None): os._exit(0) _, status = os.waitpid(pid, 0) time.sleep(0.3) - running = is_running(args.socket, args.pane) + running = watcher_alive(args.socket, args.pane) print(json.dumps({"ok": running is not None, "pid": running, "log": logfile_for(args.socket, args.pane)})) return 0 if running else 1 @@ -1716,16 +2040,7 @@ def main(argv=None): try: return watch_loop(args.socket, args.pane, dry_run=args.dry_run) finally: - try: - with open(pidfile) as f: - owner = f.read().strip() - except Exception: - owner = "" - if owner == str(os.getpid()): - try: - os.remove(pidfile) - except OSError: - pass + _release_pidfile(pidfile) if args.cmd == "match": text = sys.stdin.read() print(json.dumps(find_choice_prompt(text, tail_window=args.tail_window), indent=1)) diff --git a/bin/muse_resume_pool.py b/bin/muse_resume_pool.py new file mode 100755 index 0000000..3b68f47 --- /dev/null +++ b/bin/muse_resume_pool.py @@ -0,0 +1,545 @@ +#!/usr/bin/env python3 +"""muse_resume_pool: per-repo, profile-aware Muse Code resume pool. + +Why this exists +--------------- +``muse resume`` scopes its picker by workspace but is blind to muse-auth +profiles, and the TUI ``/resume`` reads session logs itself and lumps every +workspace into one heap. A session resumed under a different credential +than the one that created it fails server-side: the continuation is +cryptographically bound to the creating account, so the server rejects the +resume. This tool lists only the sessions that can actually resume here +and now, and guards ``resume`` calls before they fail. + +Data sources (read-only, no secrets) +------------------------------------ +- ``~/.local/share/muse/session-index.db`` ``sessions`` table (``mode=ro``). + Fallback when the index is missing: scan ``sessions/*/*/*/*/session.jsonl`` + for ``runtime.session.metadata`` (workspace_root) and + ``session.name.changed`` (session_name) records, mirroring ``/resume``. +- ``~/.config/muse/active_profile``, ``session_profiles.json``, + ``switch_history.jsonl``: profile *names* only. ``auth.json`` token bytes + are never read, logged, or compared. + +Profile resolution mirrors ``muse-auth``: cached ``session_profiles.json`` +mapping wins; otherwise the latest switch at or before session start; else +the earliest switch; else the active profile as fallback. When the auth +dir is unreadable (e.g. inside the Muse sandbox, which masks it), the +profile is ``unknown`` and only proven mismatches are hidden/blocked. +""" + +import argparse +import glob +import json +import os +import sqlite3 +import subprocess +import sys + +INDEX_COLUMNS = ( + "session_id", + "session_name", + "workspace_root", + "workspace_key", + "provider_id", + "model_id", + "git_branch", + "title", + "first_user_prompt", + "created_at_us", + "updated_at_us", + "prompt_count", + "status", +) + + +def default_paths(): + home = os.path.expanduser("~") + data_home = os.environ.get("XDG_DATA_HOME", os.path.join(home, ".local", "share")) + return { + "index_db": os.path.join(data_home, "muse", "session-index.db"), + "sessions_dir": os.path.join(data_home, "muse", "sessions"), + "config_dir": os.path.expanduser("~/.config/muse"), + } + + +def canonical_workspace(cwd=None): + """Repo root for the pool: git top-level, else real cwd.""" + cwd = cwd or os.getcwd() + try: + out = subprocess.run( + ["git", "-C", cwd, "rev-parse", "--show-toplevel"], + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + text=True, + timeout=10, + ) + if out.returncode == 0 and out.stdout.strip(): + return os.path.realpath(out.stdout.strip()) + except Exception: + pass + return os.path.realpath(cwd) + + +def load_index_rows(index_db): + """Read session rows from the index (read-only). None if unavailable.""" + if not os.path.exists(index_db): + return None + cols = ", ".join(INDEX_COLUMNS) + try: + uri = "file:{}?mode=ro".format(index_db.replace("?", "%3F")) + conn = sqlite3.connect(uri, uri=True, timeout=5) + try: + conn.row_factory = sqlite3.Row + cur = conn.execute( + "SELECT {} FROM sessions ORDER BY " + "updated_at_us DESC, created_at_us DESC, session_id ASC".format(cols) + ) + return [dict(r) for r in cur.fetchall()] + finally: + conn.close() + except sqlite3.Error: + return None + + +def _scan_log_for_session(session_log): + """Extract workspace/name/title from one session.jsonl (bounded read).""" + workspace = None + name = None + title = None + first_prompt = None + created_at_us = None + updated_at_us = None + prompt_count = 0 + try: + with open(session_log, "r", errors="replace") as fh: + for line in fh: + line = line.strip() + if not line: + continue + try: + rec = json.loads(line) + except ValueError: + continue + if "children" in rec: # retained permission frame wrapper + continue + rec_at = rec.get("recorded_at") + if isinstance(rec_at, int): + if created_at_us is None: + created_at_us = rec_at + updated_at_us = rec_at + ptype = rec.get("payload_type", "") + payload = rec.get("payload", {}) if isinstance(rec.get("payload"), dict) else {} + if ptype == "runtime.session.metadata": + record = payload.get("record", {}) + workspace = workspace or record.get("workspace_root") + elif ptype == "session.name.changed": + if payload.get("new_name"): + name = payload["new_name"] + elif ptype == "runtime.session": + event = payload.get("event", {}) + if event.get("kind") == "started" and not first_prompt: + prompt = event.get("prompt") or "" + first_prompt = prompt[:200] + title = title or prompt[:80] + prompt_count += 1 + except OSError: + return None + if workspace is None and name is None and created_at_us is None: + return None + session_id = os.path.basename(os.path.dirname(session_log)) + return { + "session_id": session_id, + "session_name": name, + "workspace_root": workspace, + "workspace_key": workspace, + "provider_id": None, + "model_id": None, + "git_branch": None, + "title": title or "New session", + "first_user_prompt": first_prompt, + "created_at_us": created_at_us, + "updated_at_us": updated_at_us or created_at_us, + "prompt_count": prompt_count, + "status": "valid", + } + + +def scan_session_logs(sessions_dir): + """Fallback pool source: scan top-level session.jsonl files directly.""" + pattern = os.path.join(sessions_dir, "*", "*", "*", "*", "session.jsonl") + rows = [] + for path in glob.glob(pattern): + row = _scan_log_for_session(path) + if row: + rows.append(row) + rows.sort( + key=lambda r: ( + r.get("updated_at_us") or 0, + r.get("created_at_us") or 0, + r.get("session_id") or "", + ), + reverse=True, + ) + return rows + + +def load_auth_state(config_dir): + """Load profile names only. Never touches auth.json token bytes.""" + state = { + "readable": False, + "active": None, + "session_profiles": {}, + "switch_history": [], + } + if not os.path.isdir(config_dir): + return state + if not (os.access(config_dir, os.R_OK) and os.access(config_dir, os.X_OK)): + return state + state["readable"] = True + try: + with open(os.path.join(config_dir, "active_profile"), "r") as fh: + state["active"] = fh.read().strip() or None + except OSError: + pass + try: + with open(os.path.join(config_dir, "session_profiles.json"), "r") as fh: + data = json.load(fh) + if isinstance(data, dict): + state["session_profiles"] = { + str(k): str(v) for k, v in data.items() + } + except (OSError, ValueError): + pass + try: + history = [] + with open(os.path.join(config_dir, "switch_history.jsonl"), "r") as fh: + for line in fh: + line = line.strip() + if not line: + continue + try: + entry = json.loads(line) + except ValueError: + continue + if entry.get("profile"): + history.append(entry) + history.sort(key=lambda e: e.get("epoch", 0)) + state["switch_history"] = history + except OSError: + pass + return state + + +def resolve_profile(session_id, created_at_us, auth_state): + """Return (profile_or_None, source). Mirrors muse-auth resolution order.""" + if not auth_state.get("readable"): + return None, "unknown" + cached = auth_state.get("session_profiles", {}) + if session_id in cached: + return cached[session_id], "cached" + history = auth_state.get("switch_history", []) + start_epoch = (created_at_us / 1e6) if created_at_us else None + if history and start_epoch: + for switch in reversed(history): + if switch.get("epoch", 0) <= start_epoch: + return switch.get("profile"), "history" + return history[0].get("profile"), "history" + if auth_state.get("active"): + return auth_state["active"], "fallback" + return None, "unknown" + + +def annotate_rows(rows, auth_state): + """Attach profile + source to each row (mutates and returns rows).""" + for row in rows: + profile, source = resolve_profile( + row.get("session_id"), row.get("created_at_us"), auth_state + ) + row["auth_profile"] = profile + row["auth_source"] = source + return rows + + +def pool_for_workspace(rows, workspace): + """Exact workspace_key match (workspace_root fallback), index order kept.""" + pool = [] + for row in rows: + key = row.get("workspace_key") or row.get("workspace_root") + if key and os.path.realpath(key) == workspace: + pool.append(row) + return pool + + +def split_resumable(pool, active_profile): + """(resumable, blocked): only proven profile mismatches are blocked. + + Unknown profiles (sandboxed auth dir, no mapping/history) stay resumable + but flagged, since blocking them would hide possibly valid sessions. + """ + resumable, blocked = [], [] + for row in pool: + profile = row.get("auth_profile") + if active_profile and profile and profile != active_profile: + blocked.append(row) + else: + resumable.append(row) + return resumable, blocked + + +def resolve_ref(rows, ref): + """Resolve UUID / UUID prefix / session name. Returns (matches, kind).""" + ref = (ref or "").strip() + if not ref: + return [], "empty" + exact = [r for r in rows if r.get("session_id") == ref] + if exact: + return exact, "uuid" + named = [r for r in rows if (r.get("session_name") or "") == ref] + if named: + return named, "name" + if len(ref) >= 8: + prefixed = [ + r for r in rows if (r.get("session_id") or "").startswith(ref) + ] + if prefixed: + return prefixed, "prefix" + return [], "none" + + +def check_resume(rows, ref, workspace, auth_state): + """Guard decision for resuming ``ref`` from ``workspace``. + + Returns dict(ok=bool, reason=str, detail=str, fix=str, row=row|None). + """ + matches, kind = resolve_ref(rows, ref) + if kind == "empty" or not matches: + return { + "ok": False, + "reason": "unknown-session", + "detail": "No session matches '{}'.".format(ref), + "fix": "List this repo's pool: muse_resume_pool.py pool", + "row": None, + } + if len(matches) > 1: + ids = ", ".join(m["session_id"][:12] for m in matches[:5]) + return { + "ok": False, + "reason": "ambiguous", + "detail": "'{}' matches {} sessions: {}".format(ref, len(matches), ids), + "fix": "Use a longer UUID prefix or the full session id.", + "row": None, + } + row = matches[0] + key = row.get("workspace_key") or row.get("workspace_root") + if not key or os.path.realpath(key) != workspace: + return { + "ok": False, + "reason": "wrong-workspace", + "detail": "Session '{}' belongs to workspace '{}', not '{}'.".format( + row.get("session_name") or row["session_id"][:12], key, workspace + ), + "fix": "cd '{}' first, or pick a session from this repo's pool.".format(key or "?"), + "row": row, + } + if row.get("status") and row["status"] != "valid": + return { + "ok": False, + "reason": "bad-status", + "detail": "Session '{}' has status '{}'.".format( + row.get("session_name") or row["session_id"][:12], row["status"] + ), + "fix": "Pick a session with status 'valid' from this repo's pool.", + "row": row, + } + active = auth_state.get("active") + profile = row.get("auth_profile") + if active and profile and profile != active: + return { + "ok": False, + "reason": "wrong-profile", + "detail": "Session '{}' was created under muse-auth profile '{}' " + "but the active profile is '{}'; the server would reject the " + "resume (continuation is bound to the creating credential).".format( + row.get("session_name") or row["session_id"][:12], profile, active + ), + "fix": "Run `muse-auth use {}` (outside the sandbox), then resume.".format(profile), + "row": row, + } + detail = "Session '{}' is resumable here.".format( + row.get("session_name") or row["session_id"][:12] + ) + if not auth_state.get("readable"): + detail += " (Profile unverified: auth dir unreadable from this shell.)" + elif not profile: + detail += " (Profile unknown: no mapping or switch history.)" + return { + "ok": True, + "reason": "ok", + "detail": detail, + "fix": "", + "row": row, + } + + +def load_rows(paths): + """Index first, log-scan fallback. Returns (rows, source).""" + rows = load_index_rows(paths["index_db"]) + if rows is not None: + return rows, "index" + return scan_session_logs(paths["sessions_dir"]), "log-scan" + + +def format_pool_table(resumable, blocked, show_all): + lines = [] + header = "{:<18} {:<12} {:<10} {:<5} {}".format( + "NAME", "SESSION", "PROFILE", "MSGS", "TITLE" + ) + lines.append(header) + for row in resumable: + flag = "?" if not row.get("auth_profile") else " " + lines.append( + "{:<18} {:<12} {:<10} {:<5} {}{}".format( + (row.get("session_name") or "-")[:18], + (row.get("session_id") or "")[:12], + (row.get("auth_profile") or "?")[:10], + row.get("prompt_count", 0), + flag, + (row.get("title") or "")[:60], + ) + ) + if show_all: + for row in blocked: + lines.append( + "{:<18} {:<12} {:<10} {:<5} {} [BLOCKED: profile mismatch]".format( + (row.get("session_name") or "-")[:18], + (row.get("session_id") or "")[:12], + (row.get("auth_profile") or "?")[:10], + row.get("prompt_count", 0), + (row.get("title") or "")[:60], + ) + ) + elif blocked: + lines.append( + "({} session(s) hidden: wrong muse-auth profile; use --all to show)".format( + len(blocked) + ) + ) + return "\n".join(lines) + + +def cmd_pool(args, paths): + workspace = os.path.realpath(args.workspace or canonical_workspace()) + rows, source = load_rows(paths) + auth_state = load_auth_state(paths["config_dir"]) + annotate_rows(rows, auth_state) + pool = pool_for_workspace(rows, workspace) + resumable, blocked = split_resumable(pool, auth_state.get("active")) + if args.json: + print(json.dumps({ + "workspace": workspace, + "source": source, + "active_profile": auth_state.get("active"), + "auth_readable": auth_state.get("readable"), + "resumable": resumable, + "blocked": blocked if args.all else [], + "blocked_count": len(blocked), + }, indent=2, default=str)) + return 0 + print("workspace: {} (source: {})".format(workspace, source)) + if auth_state.get("readable"): + print("active profile: {}".format(auth_state.get("active") or "(none)")) + else: + print("active profile: ? (auth dir unreadable from this shell)") + if not pool: + print("No sessions for this workspace.") + return 0 + print(format_pool_table(resumable, blocked, args.all)) + return 0 + + +def cmd_check(args, paths): + workspace = os.path.realpath(args.workspace or canonical_workspace()) + rows, _ = load_rows(paths) + auth_state = load_auth_state(paths["config_dir"]) + annotate_rows(rows, auth_state) + decision = check_resume(rows, args.ref, workspace, auth_state) + if args.json: + row = dict(decision["row"]) if decision["row"] else None + print(json.dumps({ + "ok": decision["ok"], + "reason": decision["reason"], + "detail": decision["detail"], + "fix": decision["fix"], + "row": row, + }, indent=2, default=str)) + else: + status = "OK" if decision["ok"] else "BLOCKED ({})".format(decision["reason"]) + print("{}: {}".format(status, decision["detail"])) + if decision["fix"]: + print("fix: {}".format(decision["fix"])) + return 0 if decision["ok"] else 1 + + +def cmd_resume(args, paths): + workspace = os.path.realpath(args.workspace or canonical_workspace()) + rows, _ = load_rows(paths) + auth_state = load_auth_state(paths["config_dir"]) + annotate_rows(rows, auth_state) + if args.ref == "--last": + pool = pool_for_workspace(rows, workspace) + resumable, _ = split_resumable(pool, auth_state.get("active")) + if not resumable: + print("No resumable sessions for this workspace.", file=sys.stderr) + return 1 + session_id = resumable[0]["session_id"] + else: + decision = check_resume(rows, args.ref, workspace, auth_state) + if not decision["ok"]: + print("refusing to resume: {}".format(decision["detail"]), file=sys.stderr) + if decision["fix"]: + print("fix: {}".format(decision["fix"]), file=sys.stderr) + return 1 + session_id = decision["row"]["session_id"] + cmd = ["muse-code", "resume", session_id] + if args.dry_run: + print("would exec: {}".format(" ".join(cmd))) + return 0 + os.execvp(cmd[0], cmd) + return 0 # unreachable + + +def main(argv=None): + parser = argparse.ArgumentParser( + prog="muse_resume_pool", + description="Per-repo, profile-aware Muse Code resume pool and guard.", + ) + parser.add_argument( + "--workspace", + help="Workspace root to scope to (default: git top-level or cwd).", + ) + sub = parser.add_subparsers(dest="command", required=True) + + p_pool = sub.add_parser("pool", help="List sessions resumable here and now.") + p_pool.add_argument("--all", action="store_true", + help="Also show profile-blocked sessions.") + p_pool.add_argument("--json", action="store_true", help="Machine-readable output.") + p_pool.set_defaults(func=cmd_pool) + + p_check = sub.add_parser("check", help="Explain whether a resume would succeed.") + p_check.add_argument("ref", help="Session UUID, UUID prefix, or session name.") + p_check.add_argument("--json", action="store_true", help="Machine-readable output.") + p_check.set_defaults(func=cmd_check) + + p_resume = sub.add_parser("resume", help="Guard then exec muse-code resume.") + p_resume.add_argument("ref", help="Session UUID, prefix, name, or --last.") + p_resume.add_argument("--dry-run", action="store_true", + help="Print the resume command instead of exec'ing.") + p_resume.set_defaults(func=cmd_resume) + + args = parser.parse_args(argv) + return args.func(args, default_paths()) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/bin/muse_session_bind.py b/bin/muse_session_bind.py new file mode 100755 index 0000000..bae9985 --- /dev/null +++ b/bin/muse_session_bind.py @@ -0,0 +1,401 @@ +#!/usr/bin/env python3 +"""muse_session_bind.py — Per-session credential isolation (P3). + +Each muse session gets its own config root:: + + /tmp/muse-session-/muse/ + + auth.json <- COPY of the bound profile's credentials (0600) + /tmp/muse-session-/bind.json <- {profile, pid, created, auth_src} + +``launch`` execs muse with XDG_CONFIG_HOME pointed at the session dir +(the binary resolves its config root as $XDG_CONFIG_HOME/muse, else +$HOME/.config/muse), so switching profiles never disturbs live +sessions: the fleet-wide 400 outage class disappears by construction. +Exec (not supervise) preserves the pane's ``muse-bin`` identity, so +watcher coverage and ``box runtime`` keep working unchanged. + +Token refreshes land in the session copy. ``save``/``reap`` copy newer +bytes back to the profile store (newest-wins across concurrent +sessions sharing a profile; nothing is ever written to the legacy +global auth.json). Dead sessions are reaped by scan, so kill -9 loses +nothing but promptness. + +Companion to the peer's muse_resume_pool (which reads +session_profiles.json): ``launch --session-id`` records the binding +there for future resume guards. +""" + +import argparse +import hashlib +import json +import os +import shutil +import sys +import time +from datetime import datetime, timezone + +SESSION_PREFIX = "muse-session-" +BIND_FILENAME = "bind.json" +AUTH_FILENAME = "auth.json" + + +def default_config_src(): + """Global config source (explicit env wins, else the real home).""" + return (os.environ.get("MUSE_CONFIG_SRC") + or os.path.join(os.path.expanduser("~"), ".config", "muse")) + + +def session_dir_for(parent, pid): + return os.path.join(parent, "%s%d" % (SESSION_PREFIX, pid)) + + +def _now(): + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def _pid_alive(pid): + try: + os.kill(pid, 0) + return True + except Exception: + return False + + +def _pid_is_muse(pid): + """True if pid's cmdline looks like a muse session (pid-reuse guard).""" + try: + with open("/proc/%d/cmdline" % pid, "rb") as f: + cmd = f.read().decode(errors="replace").lower() + return "muse-bin" in cmd or "muse-code" in cmd + except Exception: + return False + + +def _fingerprint(path): + """Short sha256 of a credential file for logs (never the bytes).""" + try: + h = hashlib.sha256() + with open(path, "rb") as f: + h.update(f.read()) + return h.hexdigest()[:12] + except OSError: + return "missing" + + +def _write_private_bytes(path, data): + """Write bytes with 0600 perms, atomically. Returns True on success.""" + try: + tmp = "%s.tmp.%d" % (path, os.getpid()) + fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + try: + os.write(fd, data) + os.fsync(fd) + finally: + os.close(fd) + os.replace(tmp, path) + return True + except OSError: + return False + + +def profile_auth_path(config_src, profile): + return os.path.join(config_src, "accounts", profile, AUTH_FILENAME) + + +def read_bind(sessdir): + try: + with open(os.path.join(sessdir, BIND_FILENAME)) as f: + data = json.load(f) + return data if isinstance(data, dict) else None + except (OSError, ValueError): + return None + + +def session_liveness(sessdir): + """live | dead | unknown (no/invalid bind record: never reap).""" + bind = read_bind(sessdir) + if not bind or not isinstance(bind.get("pid"), int): + return "unknown" + pid = bind["pid"] + if _pid_alive(pid) and _pid_is_muse(pid): + return "live" + return "dead" + + +def list_bound(parent="/tmp"): + """Session dirs carrying our bind record (foreign dirs ignored).""" + out = [] + try: + names = sorted(os.listdir(parent)) + except OSError: + return out + for name in names: + if not name.startswith(SESSION_PREFIX): + continue + sessdir = os.path.join(parent, name) + if not os.path.isdir(sessdir): + continue + if read_bind(sessdir) is None: + continue + out.append(sessdir) + return out + + +def build_session_dir(parent, pid, config_src, auth_src, profile): + """Create the isolated config root. Returns sessdir. + + Raises RuntimeError when the slot is held by a live session, or + OSError/ValueError for missing sources. + """ + auth_src = os.path.realpath(auth_src) + if not os.path.isfile(auth_src): + raise ValueError("no credentials at %s" % auth_src) + if not os.path.isdir(config_src): + raise ValueError("no config source at %s" % config_src) + sessdir = session_dir_for(parent, pid) + cfgdir = os.path.join(sessdir, "muse") + if os.path.exists(sessdir): + if session_liveness(sessdir) == "live": + raise RuntimeError("session slot %s is live" % sessdir) + shutil.rmtree(sessdir, ignore_errors=True) + os.makedirs(cfgdir) + for entry in sorted(os.listdir(config_src)): + if entry == AUTH_FILENAME: + continue + target = os.path.join(config_src, entry) + try: + os.symlink(target, os.path.join(cfgdir, entry)) + except OSError: + pass + with open(auth_src, "rb") as f: + creds = f.read() + if not _write_private_bytes(os.path.join(cfgdir, AUTH_FILENAME), creds): + raise OSError("cannot plant auth.json in %s" % cfgdir) + with open(os.path.join(sessdir, BIND_FILENAME), "w") as f: + json.dump({"profile": profile, "pid": pid, + "created": _now(), "auth_src": auth_src}, f, indent=1) + return sessdir + + +def record_session_profile(config_src, session_id, profile): + """Note session->profile for resume guards. Returns True on success.""" + path = os.path.join(config_src, "session_profiles.json") + try: + with open(path) as f: + data = json.load(f) + if not isinstance(data, dict): + data = {} + except (OSError, ValueError): + data = {} + data[str(session_id)] = str(profile) + try: + tmp = "%s.tmp.%d" % (path, os.getpid()) + with open(tmp, "w") as f: + json.dump(data, f, indent=1) + os.replace(tmp, path) + return True + except OSError: + return False + + +def save_session(sessdir, config_src=None): + """Sync a session copy back to its profile when newer. + + Returns {"status", ...}; statuses: synced | skipped-stale | + skipped-missing | no-bind. Never raises, never logs token bytes. + """ + config_src = config_src or default_config_src() + bind = read_bind(sessdir) + if not bind or not bind.get("profile"): + return {"status": "no-bind", "sessdir": sessdir} + profile = bind["profile"] + sess_auth = os.path.join(sessdir, "muse", AUTH_FILENAME) + dest = os.path.realpath(profile_auth_path(config_src, profile)) + if not os.path.isfile(sess_auth): + return {"status": "skipped-missing", "sessdir": sessdir, + "profile": profile} + try: + sess_mtime = os.path.getmtime(sess_auth) + except OSError: + return {"status": "skipped-missing", "sessdir": sessdir, + "profile": profile} + try: + dest_mtime = os.path.getmtime(dest) + except OSError: + dest_mtime = -1 + if dest_mtime >= sess_mtime: + return {"status": "skipped-stale", "sessdir": sessdir, + "profile": profile, "session_fp": _fingerprint(sess_auth), + "profile_fp": _fingerprint(dest)} + try: + with open(sess_auth, "rb") as f: + creds = f.read() + except OSError: + return {"status": "skipped-missing", "sessdir": sessdir, + "profile": profile} + try: + os.makedirs(os.path.dirname(dest), exist_ok=True) + except OSError: + pass + if not _write_private_bytes(dest, creds): + return {"status": "error", "sessdir": sessdir, "profile": profile} + return {"status": "synced", "sessdir": sessdir, "profile": profile, + "session_fp": _fingerprint(sess_auth), + "profile_fp": _fingerprint(dest)} + + +def reap(parent="/tmp", config_src=None): + """Sync + remove dead bound sessions. Returns {"reaped", "live"}.""" + config_src = config_src or default_config_src() + reaped, live = [], [] + for sessdir in list_bound(parent): + if session_liveness(sessdir) == "live": + live.append(sessdir) + continue + res = save_session(sessdir, config_src) + shutil.rmtree(sessdir, ignore_errors=True) + reaped.append({"sessdir": sessdir, "save": res["status"], + "profile": res.get("profile")}) + return {"reaped": reaped, "live": live} + + +def launch(profile=None, auth_file=None, session_id=None, config_src=None, + cmd=None, parent="/tmp", dry_run=False, _exec=os.execvpe): + """Bind then exec. With dry_run, return the plan without exec'ing.""" + config_src = config_src or default_config_src() + if auth_file: + auth_src = os.path.realpath(auth_file) + elif profile: + auth_src = profile_auth_path(config_src, profile) + else: + raise ValueError("need --profile or --auth-file") + if not os.path.isfile(auth_src): + raise ValueError("no credentials at %s" % auth_src) + pid = os.getpid() + if dry_run: + return {"sessdir": session_dir_for(parent, pid), + "xdg_config_home": session_dir_for(parent, pid), + "profile": profile, "auth_src": auth_src, + "cmd": cmd or []} + # Reap BEFORE building: our own fresh dir would read as dead (the + # launcher is python, not muse, until it execs) and eat itself. + reap(parent=parent, config_src=config_src) + sessdir = build_session_dir(parent, pid, config_src, auth_src, + profile or "explicit") + if session_id: + record_session_profile(config_src, session_id, + profile or "explicit") + env = dict(os.environ) + env["XDG_CONFIG_HOME"] = sessdir + env["MUSE_SESSION_BIND_DIR"] = sessdir + _exec(cmd[0], cmd, env) + return None # unreachable; exec replaces the image + + +def cmd_status(args): + config_src = args.config_src or default_config_src() + rows = [] + for sessdir in list_bound(args.parent): + bind = read_bind(sessdir) or {} + sess_auth = os.path.join(sessdir, "muse", AUTH_FILENAME) + prof_auth = profile_auth_path(config_src, bind.get("profile", "")) + rows.append({"sessdir": sessdir, "profile": bind.get("profile"), + "pid": bind.get("pid"), + "liveness": session_liveness(sessdir), + "session_fp": _fingerprint(sess_auth), + "profile_fp": _fingerprint(prof_auth)}) + if args.json: + print(json.dumps({"sessions": rows}, indent=1)) + else: + if not rows: + print("No bound sessions under %s." % args.parent) + return 0 + for r in rows: + print("%s profile=%s pid=%s %s session=%s profile=%s" % ( + r["sessdir"], r["profile"], r["pid"], r["liveness"], + r["session_fp"], r["profile_fp"])) + return 0 + + +def main(argv=None): + ap = argparse.ArgumentParser( + prog="muse_session_bind", + description="Per-session credential isolation for muse.") + ap.add_argument("--parent", default="/tmp", + help="Session dir parent (default /tmp).") + ap.add_argument("--config-src", default=None, + help="Global config source (default ~/.config/muse).") + sub = ap.add_subparsers(dest="command", required=True) + + p_l = sub.add_parser("launch", help="Bind a profile, then exec muse.") + p_l.add_argument("--profile", default=None) + p_l.add_argument("--auth-file", default=None) + p_l.add_argument("--session-id", default=None) + p_l.add_argument("--dry-run", action="store_true") + p_l.add_argument("cmd", nargs=argparse.REMAINDER, + help="Command after --, e.g. -- muse-code") + + p_s = sub.add_parser("save", help="Sync session tokens back to profile.") + g = p_s.add_mutually_exclusive_group(required=True) + g.add_argument("--pid", type=int) + g.add_argument("--dir") + g.add_argument("--all", action="store_true") + p_s.add_argument("--json", action="store_true") + + p_r = sub.add_parser("reap", help="Sync + remove dead sessions.") + p_r.add_argument("--json", action="store_true") + + p_st = sub.add_parser("status", help="List bound sessions.") + p_st.add_argument("--json", action="store_true") + + args = ap.parse_args(argv) + config_src = args.config_src or default_config_src() + if args.command == "launch": + cmd = [c for c in args.cmd if c != "--"] + if not cmd and not args.dry_run: + print("launch needs a command: launch ... -- muse-code [...]", + file=sys.stderr) + return 2 + try: + plan = launch(profile=args.profile, auth_file=args.auth_file, + session_id=args.session_id, config_src=config_src, + cmd=cmd, parent=args.parent, + dry_run=args.dry_run) + except (ValueError, RuntimeError, OSError) as e: + print("launch refused: %s" % (e,), file=sys.stderr) + return 1 + if args.dry_run: + print(json.dumps(plan, indent=1)) + return 0 + if args.command == "save": + if args.pid is not None: + targets = [session_dir_for(args.parent, args.pid)] + elif args.dir: + targets = [args.dir] + else: + targets = list_bound(args.parent) + results = [save_session(t, config_src) for t in targets] + if args.json: + print(json.dumps({"saved": results}, indent=1)) + else: + for r in results: + print("%s: %s" % (r["sessdir"], r["status"])) + return 0 + if args.command == "reap": + res = reap(parent=args.parent, config_src=config_src) + if args.json: + print(json.dumps(res, indent=1)) + else: + for r in res["reaped"]: + print("reaped %s (%s)" % (r["sessdir"], r["save"])) + if not res["reaped"]: + print("Nothing to reap.") + return 0 + if args.command == "status": + return cmd_status(args) + return 2 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/bin/super-cli.py b/bin/super-cli.py index eafa53d..55cb6e4 100755 --- a/bin/super-cli.py +++ b/bin/super-cli.py @@ -954,7 +954,7 @@ def cmd_runtime(args): print() return headers = ["SOCKET", "SESSION", "NODE", "PANE", "CMD", - "STATE", "APPROVE", "WATCHER"] + "STATE", "APPROVE", "MODE", "WATCHER"] table = [] for r in rows: state = r["state"] @@ -964,15 +964,20 @@ def cmd_runtime(args): if r["is_muse"]: approve = (badge_ok("YES") if r["auto_approve"] else badge_err("NO")) + mode = r.get("permission_mode") or "default" + if r.get("permission_bypass") and mode != "yolo": + mode += "!" + mode = mode[:14] else: approve = badge_dim("-") + mode = badge_dim("-") watcher = (badge_ok("ALIVE %s" % r["watcher_pid"]) if r["watcher_alive"] else badge_dim("-")) table.append([os.path.basename(r["socket"]), "%s:%s" % (r["session"], r["window"]), r["node"] or badge_dim("-"), r["pane"], (r["cmd"] or "")[:26], state, - approve, watcher]) + approve, mode, watcher]) print_table(headers, table) print() diff --git a/docs/MUSE-AUTH-CLI.md b/docs/MUSE-AUTH-CLI.md new file mode 100644 index 0000000..0dcc993 --- /dev/null +++ b/docs/MUSE-AUTH-CLI.md @@ -0,0 +1,88 @@ +# MUSE-AUTH-CLI Decision Record + +Status: **Draft** — taken over in this checkout 2026-10-07 per user choice. +Only explicit user acceptance moves this document (or any decision) to Final. + +Handoff note: a prior grill session settled D1–D11 and U1 and reportedly +marked its own record Final, but that file lives in another checkout (absent +here; this repo has no MUSE-AUTH-CLI.md, PI-AGENT-AUTH.md, OPERATORS.md, or +agy-auth-switch). D1–D11 details below are CARRIED, not verified — their full +text needs a paste or peer handoff before this record can go Final. + +## Goal + +Define the `muse-auth` CLI: per-profile credential switcher +(`~/.config/muse/accounts//auth.json`), tailnet push/pull of profiles +between nodes, and a session spend logger — without stranding live sessions +(the 2026-10-07 fleet-wide 400 outage was a mid-stream credential swap). + +## Non-goals (proposed) + +- Implementation of `muse-auth` (needs a separate explicit request). +- `agy-auth-switch` validation (Track B, separate lane; PI-AGENT-AUTH.md is Final). +- OPERATORS.md amendment for agent-invokable keys (U2 follow-on, own delta). + +## Settled (from prior-session transcript, unverified here) + +### U1. Allowance reset period — SETTLED (calendar month) + +Profile token allowances reset on the 1st of each month UTC, matching +standard billing cycles (not a rolling 30-day window). + +### D10/D11. Agent-invokable key handling — SETTLED in principle, amendment pending + +Decisions exist; codification as an OPERATORS.md amendment delta is the U2 +follow-on and is UNRESOLVED. + +### D1–D11 (remaining detail) — CARRIED, text unavailable + +Full decision text was settled in the prior session but is not present in +this checkout. CARRIED as-is; paste or peer handoff required to verify. +This record cannot go Final until they are quoted or re-settled here. + +## Scope contract (ACCEPTED 2026-10-07; user chose "accept the scope as written") + +- Artifact boundary: IN — this decision record only. OUT — runtime code, + tests, OPERATORS.md amendment, Track B validation. +- Done means: (1) push/pull file-set decision settled; (2) live-session + guard decision settled; (3) D1–D11 text verified or re-settled; + (4) user explicitly accepts this record as Final. +- Later stages (implementation, U2 amendment) each return for their own + interview; accepting this record never approves them. +- "Go"/"do it all" authorize only the boundary above. + +## Settled Decisions (New) + +### P1. Push/pull transfer file set — SETTLED (Credentials + Metadata) + +Transfer `auth.json` (cookies, tokens, session identity) and `metadata.json` (plan tier, spend watermarks, profile label). Ephemeral caches, runtime logs, and local locks are omitted from transfer. (`profile.json` in the earlier grill options was shorthand for this file and is superseded; confirmed 2026-10-07.) + +### P2. Live-session switch guard — SETTLED (Block with Force Override) + +Refuse to switch credentials if active `muse-bin` or worker processes are detected holding the old profile identity. Operators must either terminate active processes first or explicitly pass `--force` to override, preventing mid-stream 400 outages caused by stale in-memory tokens. (Confirmed in this interview 2026-10-07.) + +## Pending + +None. (All pending architectural decisions P1 and P2 are settled). + +## Session credential isolation (P3 — BUILT 2026-10-07, user-ordered) + +Each muse session runs with an isolated config dir +`/tmp/muse-session-/muse`: symlinks to `~/.config/muse/*` except +`auth.json`, which is replaced with the bound profile's credentials; +refreshed tokens sync back to the profile on session exit/save. + +Implications (unresolved): this largely obsoletes P2's block (switching +stops disturbing live sessions; the guard becomes a backstop for +legacy non-isolated sessions). Open risks: token sync-back races when +two sessions share a profile (solved: newest-wins by mtime), +sessions killed -9 never syncing (solved: reap-by-scan, no exit hook), +symlink fragility (accepted: rebuilt per launch). + +Implementation: bin/muse_session_bind.py (`launch` builds the dir and +execs with XDG_CONFIG_HOME; `save`/`reap` sync back; `status` lists). +Key integration choice: exec, not supervise, so panes keep their +muse-bin identity and watcher coverage is untouched. Tests: +tests/test_muse_session_bind.py (13). Follow-ups for the owning lanes: +wire `box runtime launch` / resume-pool `resume` through the binder, +and arm a reap timer once the profile store (P1) exists. diff --git a/job-sidechats.json b/job-sidechats.json index f89ac20..3d102b9 100644 --- a/job-sidechats.json +++ b/job-sidechats.json @@ -2473,5 +2473,16 @@ "agent": "opm", "title": "nonexistent-test", "created_at": "2026-10-06T23:15:46.206197+00:00" + }, + "61b59274": { + "thread_uuid": "ee8a7190-b137-43f0-840d-a632e48a1555", + "agent": "pip", + "title": "61b59274", + "created_at": "2026-10-07T01:11:47.250101+00:00" + }, + "pipe-234295": { + "thread_uuid": "dc9e6bad-94bd-4893-948c-1b35ec0523e0", + "agent": "646", + "created_at": "2026-10-07T01:14:57.101511+00:00" } } \ No newline at end of file diff --git a/muse-choices-rules.json b/muse-choices-rules.json new file mode 100644 index 0000000..3ab5c8a --- /dev/null +++ b/muse-choices-rules.json @@ -0,0 +1,38 @@ +{ + "_schema": "muse-choices-rules/v1", + "_notes": [ + "Deny/hold dictionary for the muse-choices daemon (decisions D0-D4).", + "First matching rule wins. Fields: id (required), kind (str|list, optional),", + "token (str|list, exact match on explicit-phrase key, optional), command", + "(regex on the approval $ block, approval kinds only, optional), text", + "(regex on cue+options, optional), decision (approve|deny|hold), reason.", + "Deny rules apply to permission kinds only (muse-approval, yn) and are", + "skipped for question kinds (D2). No match => approve top-choice (D4).", + "Seed policy: risky => hold (operator window, then approve). No active", + "deny rules yet -- add them from experience, one at a time, with tests." + ], + "version": 1, + "rules": [ + { + "id": "explicit-destructive-token", + "kind": "explicit-phrase", + "token": ["ABORT", "DELETE", "DESTROY", "WIPE", "KILL"], + "decision": "hold", + "reason": "destructive magic word; hold for operator eyes" + }, + { + "id": "cmd-destructive-shell", + "kind": ["muse-approval", "muse-approval-collapsed"], + "command": "\\brm\\s+-rf?\\b|\\bmkfs\\b|\\bdd\\s+[^\\n]*\\bof=|\\b(shutdown|reboot|halt|poweroff)\\b|git\\s+push\\S*\\s+--force\\b|git\\s+reset\\s+--hard\\b|\\bterraform\\s+(destroy|apply)\\b", + "decision": "hold", + "reason": "destructive shell command; hold for operator eyes" + }, + { + "id": "cmd-database-drop", + "kind": ["muse-approval", "muse-approval-collapsed"], + "command": "(?i)\\bdrop\\s+(table|database|schema)\\b|\\btruncate\\s+table\\b", + "decision": "hold", + "reason": "destructive database statement; hold for operator eyes" + } + ] +} diff --git a/tests/test_muse_choice_watcher.py b/tests/test_muse_choice_watcher.py index 279ee39..b579f44 100644 --- a/tests/test_muse_choice_watcher.py +++ b/tests/test_muse_choice_watcher.py @@ -505,6 +505,83 @@ class TestPidfileClaim(unittest.TestCase): self.assertTrue(w._claim_pidfile(path)) +class TestAnsweredPersistence(unittest.TestCase): + """Answered sigs survive restarts: a daemon that restarts while an + answered prompt is still visible must not answer it again (live + double-answer -> stray "1" in the input box).""" + + def test_restart_suppresses_answered_sig(self): + import tempfile + with tempfile.TemporaryDirectory() as td: + path = td + "/x.answered.json" + s1 = w.WatcherState(persist_path=path) + now = time.time() + s1.record_answer("sig-abc", now) + s2 = w.WatcherState(persist_path=path) + self.assertEqual(s2.observe({"sig": "sig-abc"}, now + 5), + "none") + + def test_expired_sig_not_loaded(self): + import tempfile + with tempfile.TemporaryDirectory() as td: + path = td + "/x.answered.json" + s1 = w.WatcherState(persist_path=path) + s1.record_answer("sig-old", + time.time() - w.ANSWERED_TTL_SECONDS - 10) + s2 = w.WatcherState(persist_path=path) + self.assertNotIn("sig-old", s2.answered_sigs) + + def test_corrupt_store_loads_empty(self): + import tempfile + with tempfile.TemporaryDirectory() as td: + path = td + "/x.answered.json" + with open(path, "w") as f: + f.write("not json{{{") + s = w.WatcherState(persist_path=path) + self.assertEqual(s.answered_sigs, {}) + + def test_memory_only_without_path(self): + s = w.WatcherState() + s.record_answer("sig-x", time.time()) + self.assertIn("sig-x", s.answered_sigs) + + +class TestPidfileFlock(unittest.TestCase): + """Single daemon per pane, kernel-enforced: concurrent starters + must not pile up (double answers, '11' in the input box).""" + + def _child_claim(self, path): + import subprocess + code = ("import sys; sys.path.insert(0, %r); " + "import muse_choice_watcher as w; " + "sys.exit(0 if w._claim_pidfile(%r) else 3)") % ( + str(BIN_DIR), path) + return subprocess.run([sys.executable, "-c", code], + capture_output=True, timeout=30) + + def test_second_process_refused_while_held(self): + import tempfile + with tempfile.TemporaryDirectory() as td: + path = td + "/x.pid" + self.assertTrue(w._claim_pidfile(path)) + try: + r = self._child_claim(path) + self.assertEqual(r.returncode, 3) + finally: + w._release_pidfile(path) + + def test_claim_succeeds_after_release(self): + import os + import tempfile + with tempfile.TemporaryDirectory() as td: + path = td + "/x.pid" + self.assertTrue(w._claim_pidfile(path)) + w._release_pidfile(path) + self.assertFalse(os.path.exists(path)) + r = self._child_claim(path) + self.assertEqual(r.returncode, 0) + + class TestReconcileFailed(unittest.TestCase): def test_failed_starts_recorded(self): with mock.patch.object(w, "get_desired", @@ -1451,5 +1528,124 @@ class TestBoxResolve(unittest.TestCase): self.assertIsNone(data["held"]) +class TestPeerAnsweredReload(unittest.TestCase): + """A peer watcher answering first must suppress this watcher. + + Concurrent duplicates share only the on-disk answered store: a + second watcher that loaded before the peer's answer must re-check + the store before typing, or both type '1' ('11' in the input box). + """ + + def test_disk_answer_suppresses_send(self): + import json + import tempfile + import time + with tempfile.TemporaryDirectory() as td: + path = td + "/x.answered.json" + state = w.WatcherState(persist_path=path) + log = mock.Mock() + sig = w.find_choice_prompt(PROMPT_YN)["sig"] + with mock.patch.object(w, "pane_exists", return_value=True), \ + mock.patch.object(w, "capture_pane", + side_effect=[PROMPT_YN] * 4), \ + mock.patch.object(w, "send_answer", + return_value=True) as send, \ + mock.patch.object(w, "audit"): + out1 = w._poll_once("/tmp/s", "%1", state, log, + dry_run=False) + self.assertEqual(out1, "seen") + # Peer answers the same prompt and persists first. + with open(path, "w") as f: + json.dump({sig: time.time()}, f) + out2 = w._poll_once("/tmp/s", "%1", state, log, + dry_run=False) + send.assert_not_called() + self.assertIn(out2, ("duplicate-suppressed", "none")) + self.assertIn(sig, state.answered_sigs) + + +class TestCaptureJoinWrapped(unittest.TestCase): + def test_capture_joins_wrapped_lines(self): + with mock.patch.object(w, "_tmux") as t: + t.return_value = mock.Mock(returncode=0, stdout="ok") + w.capture_pane("/tmp/s", "%1") + args = t.call_args[0] + self.assertIn("-J", args) + + +class TestLogPosture(unittest.TestCase): + """Watcher start records the pane's permission posture. + + The watcher answers with per-choice logging in every mode; bypass + postures (yolo / approval disabled) additionally audit, since the + session then acts outside the choice trail. + """ + + def test_bypass_audits(self): + log = mock.Mock() + with mock.patch.object(w, "pane_muse_argv", + return_value=["muse", "--yolo"]), \ + mock.patch.object(w, "audit") as audit: + w._log_posture("/tmp/s", "%1", log) + log.log.assert_called_once() + self.assertEqual(log.log.call_args[0][1], "pane posture") + self.assertEqual(log.log.call_args[1]["mode"], "yolo") + audit.assert_called_once() + self.assertEqual(audit.call_args[0][0], "muse-choice-posture") + self.assertTrue(audit.call_args[1]["extra"]["bypass"]) + + def test_default_logs_without_audit(self): + log = mock.Mock() + with mock.patch.object(w, "pane_muse_argv", + return_value=["muse"]), \ + mock.patch.object(w, "audit") as audit: + w._log_posture("/tmp/s", "%1", log) + log.log.assert_called_once() + audit.assert_not_called() + + def test_nondefault_profile_audits(self): + log = mock.Mock() + with mock.patch.object(w, "pane_muse_argv", + return_value=["muse", "--permission-profile", + ":unrestricted"]), \ + mock.patch.object(w, "audit") as audit: + w._log_posture("/tmp/s", "%1", log) + audit.assert_called_once() + extra = audit.call_args[1]["extra"] + self.assertEqual(extra["mode"], ":unrestricted") + self.assertFalse(extra["bypass"]) + + def test_argv_failure_never_raises(self): + log = mock.Mock() + with mock.patch.object(w, "pane_muse_argv", + side_effect=RuntimeError("tmux")), \ + mock.patch.object(w, "audit") as audit: + w._log_posture("/tmp/s", "%1", log) # must not raise + log.log.assert_not_called() + audit.assert_not_called() + + +class TestReconcileOrphan(unittest.TestCase): + """Reconcile must not start a second watcher when an orphan with a + lost pidfile is already alive for the pane.""" + + def test_orphan_counts_as_already(self): + orphan = [{"pid": 7777, "socket": "/tmp/sock", "pane": "%37"}] + with mock.patch.object(w, "get_desired", + return_value={"enabled": True, + "dry_run": False}), \ + mock.patch.object(w, "muse_panes", return_value=["%37"]), \ + mock.patch.object(w, "is_running", return_value=None), \ + mock.patch.object(w, "_watch_procs", return_value=orphan), \ + mock.patch.object(w, "_start_detached") as start, \ + mock.patch.object(w, "status_all", return_value=[]), \ + mock.patch.object(w, "audit"), \ + mock.patch.object(w.os.path, "exists", return_value=True): + res = w.reconcile(sockets=["/tmp/sock"]) + start.assert_not_called() + self.assertEqual(res["already"], ["/tmp/sock:%37"]) + self.assertEqual(res["failed"], []) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_muse_resume_pool.py b/tests/test_muse_resume_pool.py new file mode 100644 index 0000000..26e85c9 --- /dev/null +++ b/tests/test_muse_resume_pool.py @@ -0,0 +1,286 @@ +#!/usr/bin/env python3 +"""test_muse_resume_pool.py — focused tests for the per-repo, profile-aware +Muse Code resume pool and guard. Fixture-only: never touches the real +~/.config/muse or ~/.local/share/muse.""" + +import json +import os +import sqlite3 +import sys +import tempfile +import unittest + +REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +sys.path.insert(0, os.path.join(REPO_ROOT, "bin")) + +import muse_resume_pool as mrp + + +WS_NETVM = "/home/super/Projects/NetVM" +WS_HOME = "/home/super" + + +def make_row(session_id, workspace, name=None, created_us=1791000000000000, + updated_us=1791300000000000, status="valid", title="t"): + return { + "session_id": session_id, + "session_name": name, + "workspace_root": workspace, + "workspace_key": workspace, + "provider_id": "meta", + "model_id": "m", + "git_branch": None, + "title": title, + "first_user_prompt": title, + "created_at_us": created_us, + "updated_at_us": updated_us, + "prompt_count": 3, + "status": status, + } + + +def make_auth(active="muse", mapping=None, history=None, readable=True): + return { + "readable": readable, + "active": active, + "session_profiles": dict(mapping or {}), + "switch_history": list(history or []), + } + + +class TestProfileResolution(unittest.TestCase): + def test_cached_mapping_wins_over_history(self): + auth = make_auth( + mapping={"s1": "def"}, + history=[{"profile": "muse", "epoch": 1790000000.0}], + ) + profile, source = mrp.resolve_profile("s1", 1791000000000000, auth) + self.assertEqual((profile, source), ("def", "cached")) + + def test_history_picks_latest_switch_before_start(self): + auth = make_auth(history=[ + {"profile": "muse", "epoch": 1790000000.0}, + {"profile": "def", "epoch": 1791100000.0}, + {"profile": "opm", "epoch": 1791400000.0}, + ]) + profile, source = mrp.resolve_profile("sx", 1791200000000000, auth) + self.assertEqual((profile, source), ("def", "history")) + + def test_history_predating_all_switches_uses_earliest(self): + auth = make_auth(history=[ + {"profile": "def", "epoch": 1791100000.0}, + {"profile": "opm", "epoch": 1791400000.0}, + ]) + profile, source = mrp.resolve_profile("sx", 1790000000000000, auth) + self.assertEqual((profile, source), ("def", "history")) + + def test_no_history_falls_back_to_active(self): + auth = make_auth(active="pip", history=[]) + profile, source = mrp.resolve_profile("sx", 1791200000000000, auth) + self.assertEqual((profile, source), ("pip", "fallback")) + + def test_unreadable_auth_dir_is_unknown(self): + auth = make_auth(readable=False) + profile, source = mrp.resolve_profile("s1", 1791200000000000, auth) + self.assertEqual((profile, source), (None, "unknown")) + + +class TestWorkspacePool(unittest.TestCase): + def test_exact_workspace_match_only(self): + rows = [ + make_row("a" * 32, WS_NETVM, name="n1"), + make_row("b" * 32, WS_HOME, name="n2"), + make_row("c" * 32, WS_NETVM, name="n3"), + ] + pool = mrp.pool_for_workspace(rows, WS_NETVM) + self.assertEqual([r["session_id"] for r in pool], ["a" * 32, "c" * 32]) + + def test_subdirectory_does_not_match(self): + rows = [make_row("a" * 32, WS_NETVM + "/subdir")] + self.assertEqual(mrp.pool_for_workspace(rows, WS_NETVM), []) + + def test_split_hides_only_proven_mismatches(self): + rows = [ + make_row("a" * 32, WS_NETVM), + make_row("b" * 32, WS_NETVM), + make_row("c" * 32, WS_NETVM), + ] + rows[0]["auth_profile"] = "muse" + rows[1]["auth_profile"] = "def" + rows[2]["auth_profile"] = None # unknown stays visible, flagged + resumable, blocked = mrp.split_resumable(rows, "muse") + self.assertEqual([r["session_id"] for r in resumable], ["a" * 32, "c" * 32]) + self.assertEqual([r["session_id"] for r in blocked], ["b" * 32]) + + +class TestRefResolution(unittest.TestCase): + def setUp(self): + self.rows = [ + make_row("01a113f1" + "0" * 24, WS_NETVM, name="alder-halo"), + make_row("01a113d8" + "1" * 24, WS_NETVM, name="blue-merak"), + make_row("01a113f1" + "2" * 24, WS_NETVM, name="third-wheel"), + ] + + def test_full_uuid(self): + matches, kind = mrp.resolve_ref(self.rows, "01a113f1" + "0" * 24) + self.assertEqual((len(matches), kind), (1, "uuid")) + + def test_name(self): + matches, kind = mrp.resolve_ref(self.rows, "blue-merak") + self.assertEqual((len(matches), kind), (1, "name")) + + def test_prefix(self): + matches, kind = mrp.resolve_ref(self.rows, "01a113f100") + self.assertEqual((len(matches), kind), (1, "prefix")) + + def test_short_prefix_rejected(self): + matches, kind = mrp.resolve_ref(self.rows, "01a1") + self.assertEqual((matches, kind), ([], "none")) + + def test_unknown(self): + matches, kind = mrp.resolve_ref(self.rows, "nope-missing") + self.assertEqual((matches, kind), ([], "none")) + + def test_ambiguous_prefix(self): + matches, kind = mrp.resolve_ref(self.rows, "01a113f1") + self.assertEqual(len(matches), 2) + self.assertEqual(kind, "prefix") + + +class TestGuard(unittest.TestCase): + def setUp(self): + self.rows = mrp.annotate_rows([ + make_row("01a113f1" + "0" * 24, WS_NETVM, name="alder-halo"), + make_row("f" * 32, WS_HOME, name="far-away"), + ], make_auth(active="muse", mapping={ + "01a113f1" + "0" * 24: "def", + "f" * 32: "muse", + })) + + def test_wrong_profile_blocked_with_fix(self): + d = mrp.check_resume(self.rows, "alder-halo", WS_NETVM, + make_auth(active="muse", mapping={ + "01a113f1" + "0" * 24: "def"})) + self.assertFalse(d["ok"]) + self.assertEqual(d["reason"], "wrong-profile") + self.assertIn("muse-auth use def", d["fix"]) + + def test_matching_profile_ok(self): + auth = make_auth(active="muse", mapping={"01a113f1" + "0" * 24: "muse"}) + rows = mrp.annotate_rows( + [make_row("01a113f1" + "0" * 24, WS_NETVM, name="alder-halo")], auth) + d = mrp.check_resume(rows, "alder-halo", WS_NETVM, auth) + self.assertTrue(d["ok"]) + self.assertEqual(d["reason"], "ok") + + def test_wrong_workspace_blocked(self): + auth = make_auth(active="muse", mapping={"f" * 32: "muse"}) + rows = mrp.annotate_rows([make_row("f" * 32, WS_HOME, name="far-away")], auth) + d = mrp.check_resume(rows, "far-away", WS_NETVM, auth) + self.assertFalse(d["ok"]) + self.assertEqual(d["reason"], "wrong-workspace") + self.assertIn("cd", d["fix"]) + + def test_unknown_session(self): + d = mrp.check_resume(self.rows, "ghost-nope", WS_NETVM, make_auth()) + self.assertFalse(d["ok"]) + self.assertEqual(d["reason"], "unknown-session") + + def test_bad_status_blocked(self): + auth = make_auth(active="muse") + rows = mrp.annotate_rows( + [make_row("d" * 32, WS_NETVM, name="stale-one", status="deleted")], auth) + d = mrp.check_resume(rows, "stale-one", WS_NETVM, auth) + self.assertFalse(d["ok"]) + self.assertEqual(d["reason"], "bad-status") + + def test_unknown_profile_warns_not_blocks(self): + auth = make_auth(readable=False) + rows = mrp.annotate_rows( + [make_row("e" * 32, WS_NETVM, name="sandboxed-one")], auth) + d = mrp.check_resume(rows, "sandboxed-one", WS_NETVM, auth) + self.assertTrue(d["ok"]) + self.assertIn("unverified", d["detail"].lower()) + + +class TestIndexAndLogSources(unittest.TestCase): + def test_load_index_rows_read_only(self): + with tempfile.TemporaryDirectory() as tmp: + db_path = os.path.join(tmp, "session-index.db") + conn = sqlite3.connect(db_path) + conn.execute( + "CREATE TABLE sessions (session_id TEXT PRIMARY KEY," + " session_name TEXT, workspace_root TEXT, workspace_key TEXT," + " provider_id TEXT, model_id TEXT, git_branch TEXT, title TEXT," + " first_user_prompt TEXT, created_at_us INTEGER," + " updated_at_us INTEGER, prompt_count INTEGER, status TEXT)" + ) + older, newer = make_row("a" * 32, WS_NETVM, name="old", + updated_us=1791000000000000), \ + make_row("b" * 32, WS_NETVM, name="new", + updated_us=1791300000000000) + for r in (older, newer): + conn.execute( + "INSERT INTO sessions VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)", + tuple(r[c] for c in mrp.INDEX_COLUMNS), + ) + conn.commit() + conn.close() + rows = mrp.load_index_rows(db_path) + self.assertEqual([r["session_name"] for r in rows], ["new", "old"]) + + def test_missing_index_returns_none(self): + self.assertIsNone(mrp.load_index_rows("/nonexistent/x.db")) + + def test_log_scan_fallback_extracts_workspace_and_name(self): + with tempfile.TemporaryDirectory() as tmp: + session_id = "9" * 32 + sdir = os.path.join(tmp, "2026", "10", "07", session_id) + os.makedirs(sdir) + records = [ + {"payload_type": "runtime.session.metadata", + "recorded_at": 1791000000000000, + "payload": {"record": {"workspace_root": WS_NETVM}}}, + {"payload_type": "session.name.changed", + "recorded_at": 1791000001000000, + "payload": {"new_name": "scanned-one"}}, + {"payload_type": "runtime.session", + "recorded_at": 1791000002000000, + "payload": {"event": {"kind": "started", "prompt": "hello world"}}}, + ] + with open(os.path.join(sdir, "session.jsonl"), "w") as fh: + for rec in records: + fh.write(json.dumps(rec) + "\n") + rows = mrp.scan_session_logs(tmp) + self.assertEqual(len(rows), 1) + row = rows[0] + self.assertEqual(row["session_id"], session_id) + self.assertEqual(row["workspace_key"], WS_NETVM) + self.assertEqual(row["session_name"], "scanned-one") + self.assertEqual(row["prompt_count"], 1) + + def test_auth_state_never_reads_auth_json(self): + with tempfile.TemporaryDirectory() as tmp: + with open(os.path.join(tmp, "active_profile"), "w") as fh: + fh.write("muse\n") + with open(os.path.join(tmp, "session_profiles.json"), "w") as fh: + json.dump({"s1": "muse"}, fh) + with open(os.path.join(tmp, "switch_history.jsonl"), "w") as fh: + fh.write(json.dumps({"profile": "muse", "epoch": 1.0}) + "\n") + # A decoy auth.json with secret bytes must not be opened: make it + # unreadable and assert load still succeeds without error. + decoy = os.path.join(tmp, "auth.json") + with open(decoy, "w") as fh: + fh.write('{"secret": true}') + os.chmod(decoy, 0) + try: + state = mrp.load_auth_state(tmp) + finally: + os.chmod(decoy, 0o600) + self.assertTrue(state["readable"]) + self.assertEqual(state["active"], "muse") + self.assertEqual(state["session_profiles"], {"s1": "muse"}) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_muse_session_bind.py b/tests/test_muse_session_bind.py new file mode 100644 index 0000000..0777646 --- /dev/null +++ b/tests/test_muse_session_bind.py @@ -0,0 +1,241 @@ +#!/usr/bin/env python3 +"""test_muse_session_bind.py — Per-session credential isolation (P3). + +Covers: session dir layout (symlinks + private auth copy), newest-wins +save-back that never touches the global auth.json, dead-only reap, +exec env, and the resume-pool session binding record. +""" + +import json +import os +import stat +import sys +import time +import unittest +from pathlib import Path +from unittest import mock + +REPO_ROOT = Path("/home/super/Projects/NetVM") +BIN_DIR = REPO_ROOT / "bin" +sys.path.insert(0, str(BIN_DIR)) + +import muse_session_bind as b + + +def _mkconfig(root): + """Fake global config: profile creds + global files.""" + cfg = os.path.join(root, "config") + os.makedirs(os.path.join(cfg, "accounts", "alice")) + with open(os.path.join(cfg, "accounts", "alice", "auth.json"), "wb") as f: + f.write(b"TOKEN-ALICE") + with open(os.path.join(cfg, "auth.json"), "wb") as f: + f.write(b"GLOBAL-TOKEN") + with open(os.path.join(cfg, "settings.json"), "w") as f: + f.write("{}") + with open(os.path.join(cfg, "notes.txt"), "w") as f: + f.write("hi") + return cfg + + +class TestBuild(unittest.TestCase): + def test_layout_links_and_private_copy(self): + import tempfile + with tempfile.TemporaryDirectory() as td: + cfg = _mkconfig(td) + parent = os.path.join(td, "run") + os.makedirs(parent) + sess = b.build_session_dir( + parent, 4242, cfg, + os.path.join(cfg, "accounts", "alice", "auth.json"), + "alice") + muse = os.path.join(sess, "muse") + # Everything but auth.json is a symlink to global. + self.assertTrue(os.path.islink(os.path.join(muse, "settings.json"))) + self.assertTrue(os.path.islink(os.path.join(muse, "notes.txt"))) + self.assertTrue(os.path.islink(os.path.join(muse, "accounts"))) + # auth.json is a real file with the profile bytes, 0600. + auth = os.path.join(muse, "auth.json") + self.assertFalse(os.path.islink(auth)) + with open(auth, "rb") as f: + self.assertEqual(f.read(), b"TOKEN-ALICE") + self.assertEqual(stat.S_IMODE(os.stat(auth).st_mode), 0o600) + bind = json.load(open(os.path.join(sess, "bind.json"))) + self.assertEqual((bind["profile"], bind["pid"]), + ("alice", 4242)) + + def test_missing_creds_refused(self): + import tempfile + with tempfile.TemporaryDirectory() as td: + cfg = _mkconfig(td) + with self.assertRaises(ValueError): + b.build_session_dir(td, 1, cfg, + os.path.join(td, "nope.json"), "alice") + + def test_live_slot_refused_stale_slot_wiped(self): + import tempfile + with tempfile.TemporaryDirectory() as td: + cfg = _mkconfig(td) + auth = os.path.join(cfg, "accounts", "alice", "auth.json") + sess = b.build_session_dir(td, 99, cfg, auth, "alice") + with mock.patch.object(b, "session_liveness", + return_value="live"): + with self.assertRaises(RuntimeError): + b.build_session_dir(td, 99, cfg, auth, "alice") + with mock.patch.object(b, "session_liveness", + return_value="dead"): + sess2 = b.build_session_dir(td, 99, cfg, auth, "alice") + self.assertEqual(sess2, sess) + self.assertTrue(os.path.isfile( + os.path.join(sess2, "muse", "auth.json"))) + + +class TestSaveBack(unittest.TestCase): + def _bound(self, td, pid=777): + cfg = _mkconfig(td) + parent = os.path.join(td, "run") + os.makedirs(parent) + sess = b.build_session_dir( + parent, pid, cfg, + os.path.join(cfg, "accounts", "alice", "auth.json"), "alice") + return cfg, sess + + def test_newer_session_syncs_to_profile(self): + import tempfile + with tempfile.TemporaryDirectory() as td: + cfg, sess = self._bound(td) + prof = os.path.join(cfg, "accounts", "alice", "auth.json") + sauth = os.path.join(sess, "muse", "auth.json") + with open(sauth, "wb") as f: + f.write(b"TOKEN-REFRESHED") + now = time.time() + os.utime(prof, (now - 100, now - 100)) + os.utime(sauth, (now, now)) + res = b.save_session(sess, cfg) + self.assertEqual(res["status"], "synced") + with open(prof, "rb") as f: + self.assertEqual(f.read(), b"TOKEN-REFRESHED") + self.assertEqual(stat.S_IMODE(os.stat(prof).st_mode), 0o600) + # Global auth.json untouched. + with open(os.path.join(cfg, "auth.json"), "rb") as f: + self.assertEqual(f.read(), b"GLOBAL-TOKEN") + + def test_stale_session_skipped(self): + import tempfile + with tempfile.TemporaryDirectory() as td: + cfg, sess = self._bound(td) + prof = os.path.join(cfg, "accounts", "alice", "auth.json") + sauth = os.path.join(sess, "muse", "auth.json") + now = time.time() + os.utime(prof, (now, now)) + os.utime(sauth, (now - 100, now - 100)) + res = b.save_session(sess, cfg) + self.assertEqual(res["status"], "skipped-stale") + with open(prof, "rb") as f: + self.assertEqual(f.read(), b"TOKEN-ALICE") + + def test_unbound_dir_never_reaped(self): + import tempfile + with tempfile.TemporaryDirectory() as td: + foreign = os.path.join(td, "muse-session-1") + os.makedirs(foreign) + with open(os.path.join(foreign, "keep.txt"), "w") as f: + f.write("x") + self.assertEqual(b.list_bound(td), []) + res = b.reap(parent=td, config_src=os.path.join(td, "cfg")) + self.assertEqual(res["reaped"], []) + self.assertTrue(os.path.isfile( + os.path.join(foreign, "keep.txt"))) + + def test_reap_dead_keeps_live(self): + import tempfile + with tempfile.TemporaryDirectory() as td: + cfg = _mkconfig(td) + parent = os.path.join(td, "run") + os.makedirs(parent) + auth = os.path.join(cfg, "accounts", "alice", "auth.json") + dead = b.build_session_dir(parent, 11, cfg, auth, "alice") + live = b.build_session_dir(parent, 22, cfg, auth, "alice") + with mock.patch.object( + b, "session_liveness", + side_effect=lambda s: "live" if s == live else "dead"): + res = b.reap(parent=parent, config_src=cfg) + self.assertEqual([r["sessdir"] for r in res["reaped"]], [dead]) + self.assertEqual(res["live"], [live]) + self.assertFalse(os.path.exists(dead)) + self.assertTrue(os.path.isdir(live)) + + +class TestLaunch(unittest.TestCase): + def test_dry_run_plans_without_touching_disk(self): + import tempfile + with tempfile.TemporaryDirectory() as td: + cfg = _mkconfig(td) + plan = b.launch(profile="alice", config_src=cfg, + cmd=["muse-code", "--foo"], parent=td, + dry_run=True) + self.assertIn("muse-session-", plan["sessdir"]) + self.assertEqual(plan["xdg_config_home"], plan["sessdir"]) + self.assertTrue(plan["auth_src"].endswith( + "accounts/alice/auth.json")) + self.assertFalse(os.path.exists(plan["sessdir"])) + + def test_launch_execs_with_isolated_env(self): + import tempfile + with tempfile.TemporaryDirectory() as td: + cfg = _mkconfig(td) + seen = {} + + def fake_exec(path, argv, env): + seen.update(path=path, argv=argv, env=env) + + with mock.patch.object(b.os, "getpid", return_value=555): + b.launch(profile="alice", config_src=cfg, + cmd=["muse-code", "chat"], parent=td, + _exec=fake_exec) + sess = os.path.join(td, "muse-session-555") + self.assertEqual(seen["path"], "muse-code") + self.assertEqual(seen["env"]["XDG_CONFIG_HOME"], sess) + self.assertEqual(seen["env"]["MUSE_SESSION_BIND_DIR"], sess) + self.assertTrue(os.path.isfile( + os.path.join(sess, "muse", "auth.json"))) + + def test_launch_records_session_profile(self): + import tempfile + with tempfile.TemporaryDirectory() as td: + cfg = _mkconfig(td) + with mock.patch.object(b.os, "getpid", return_value=556): + b.launch(profile="alice", config_src=cfg, + session_id="sess-1", cmd=["muse-code"], + parent=td, _exec=lambda *a: None) + data = json.load(open(os.path.join(cfg, "session_profiles.json"))) + self.assertEqual(data, {"sess-1": "alice"}) + + def test_launch_without_profile_or_auth_refused(self): + with self.assertRaises(ValueError): + b.launch(cmd=["muse-code"], dry_run=True) + + +class TestLiveness(unittest.TestCase): + def test_recycled_pid_is_dead(self): + import tempfile + with tempfile.TemporaryDirectory() as td: + sess = os.path.join(td, "muse-session-9") + os.makedirs(sess) + with open(os.path.join(sess, "bind.json"), "w") as f: + json.dump({"profile": "alice", "pid": 9}, f) + with mock.patch.object(b, "_pid_alive", return_value=True), \ + mock.patch.object(b, "_pid_is_muse", return_value=False): + self.assertEqual(b.session_liveness(sess), "dead") + + def test_pid_is_muse_matches_binary_names(self): + import io + with mock.patch("builtins.open", + mock.mock_open(read_data=b"muse-bin-1.4\x00--x\x00")): + self.assertTrue(b._pid_is_muse(123)) + with mock.patch("builtins.open", + mock.mock_open(read_data=b"python3\x00foo\x00")): + self.assertFalse(b._pid_is_muse(123)) + + +if __name__ == "__main__": + unittest.main()