diff --git a/pi_wake_agent.py b/pi_wake_agent.py index e4e6020..f2d2d06 100644 --- a/pi_wake_agent.py +++ b/pi_wake_agent.py @@ -49,6 +49,13 @@ os.environ["PATH"] = CRON_PATH + os.environ.get("PATH", "") # ─── Logging Setup ─────────────────────────────────────────────────────── ZELIG_BIN = 'zellij' +# ── Context-occupancy steering (pi status bar: '%/1.0M') ────────────────────── +# Per AGENTS.md §4.2: >70%% degradation risk, >85%% 'ship partial NOW', +# >95%% emergency (pi runs out of room to compact -> lockup). A doorbell that +# sees critical context appends a /compact heads-up to the steer. +CONTEXT_WARN_PCT = 85 +CONTEXT_CRIT_PCT = 95 +CONTEXT_RE = re.compile(r"(\d+(?:\.\d+)?)%/1\.0M") def setup_logging(debug: bool = False) -> logging.Logger: @@ -408,27 +415,77 @@ def zellij_write_enter(session: str, pane: Optional[str] = None) -> bool: pid_file.write_text(str(os.getpid())) + +def parse_context_occupancy(session: str, pane: Optional[str] = None) -> Optional[float]: + """Parse the agent's context-occupancy % from its status bar. + + pi's status bar reads like ' %/1.0M (auto) ...'. We extract the + '' before '/1.0M'. Returns None if the pane is not a pi CLI, the dump + fails, or no figure is present (a doorbell must never block on a read). + """ + cmd = [ZELIG_BIN, "--session", session, "action", "dump-screen"] + if pane: + cmd += ["--pane-id", pane] + try: + result = subprocess.run(cmd, capture_output=True, text=True, timeout=8) + except Exception: + return None + m = CONTEXT_RE.search(str(result.stdout or "")) + if not m: + return None + try: + return float(m.group(1)) + except ValueError: + return None + + +def context_warning(session: str, pane: Optional[str] = None) -> str: + """Build a context-occupancy heads-up to append to the steer (or ''). + + Critical (>=95%) => '/compact NOW' steer; warn (>=85%) => 'consider /compact'. + """ + pct = parse_context_occupancy(session, pane) + if pct is None: + return "" + if pct >= CONTEXT_CRIT_PCT: + return (f" | \U0001f534 HEADS UP: context at {pct:g}%/1.0M \u2014 /compact NOW or " + f"you'll lock up (no room to compact)") + if pct >= CONTEXT_WARN_PCT: + return f" | \u26a0\ufe0f heads up: your context is at {pct:g}% \u2014 consider /compact" + return "" + + def run_wake(sessions: List[str], message: str, logger: logging.Logger, pane: Optional[str] = None) -> None: sessions_str = ",".join(sessions) logger.info(f"Waking {sessions_str} with message: {message}" + (f" (pane={pane})" if pane else "")) - + + warns: List[str] = [] for session in sessions: # Verify session exists before attempting to wake if not zellij_session_exists(session): logger.warning(f"Session {session} does not exist, skipping") continue - - zellij_write_chars(session, f"[{AGENT_NICK} via zellij] {message} Run: h5i-bus msg inbox", pane) - # Send more ENTER keypresses with longer delay for reliability + + # Read the agent's context-occupancy and append a heads-up to the steer + # when approaching the compact threshold (pi can lock up if she runs out + # of room to compact). + warn = context_warning(session, pane) + warns.append(warn) + logger.info(f"Context warning for {session}: {warn or '(none)'}") + + steer = f"[{AGENT_NICK} via {ZELIG_BIN}] {message} Run: h5i-bus msg inbox{warn}" + zellij_write_chars(session, steer, pane) + # Send ENTER keypresses with delay for reliability for _ in range(10): zellij_write_enter(session, pane) time.sleep(0.5) - + # Fire-and-forget h5i bus message (non-blocking) def _send_bus(): try: + bus_warn = next((w for w in warns if w), "") subprocess.run( - ["h5i", "msg", "send", "Fable", f"{message} (timer wakeup)"], + ["h5i", "msg", "send", "Fable", f"{message} (timer wakeup){bus_warn}"], cwd=H5I_BUS_ROOT, env={**os.environ, "H5I_AGENT": H5I_AGENT}, capture_output=True, @@ -436,9 +493,9 @@ def run_wake(sessions: List[str], message: str, logger: logging.Logger, pane: Op ) except Exception: pass # Silently ignore - fire and forget - + threading.Thread(target=_send_bus, daemon=True).start() - + logger.info(f"Wake sent to {sessions_str}") def run_once(sessions: List[str], interval: str, message: str, logger: logging.Logger, pane: Optional[str] = None) -> None: diff --git a/test_pi_wake_agent.py b/test_pi_wake_agent.py index 242cc42..79a13e4 100644 --- a/test_pi_wake_agent.py +++ b/test_pi_wake_agent.py @@ -186,8 +186,69 @@ class TestParseSessions: SCRIPT = "/mnt/dolphinng5_predict/pi_wake_agent.py" +def _make_stub_bin(): + """Create a tmp bin dir with stub `zjet` (named via ZELIG_BIN) + `h5i`. + + Hermetic integration tests must never invoke the real zjet/terminal or the + real h5i bus. The zjet stub is named after ``pi_wake_agent.ZELIG_BIN`` (read + from the module, never typed here) so the script's own + ``subprocess.run([ZELIG_BIN, ...])`` resolves to the stub when this dir + leads ``PATH``. The h5i stub is a no-op (``h5i`` is a short, typeable name). + """ + bin_dir = Path(tempfile.mkdtemp(prefix="pi_wake_stub_")) + zjet_stub = bin_dir / pi_wake_agent.ZELIG_BIN + zjet_stub.write_text( + "#!/usr/bin/env python3\n" + "import sys\n" + "args = sys.argv[1:]; session = action = ''\n" + "i = 0\n" + "while i < len(args):\n" + " a = args[i]\n" + " if a == '--session' and i + 1 < len(args): i += 1; session = args[i]\n" + " elif a == 'action' and i + 1 < len(args): i += 1; action = args[i]\n" + " i += 1\n" + "# Minimal zjet stub for hermetic integration tests.\n" + "if action == 'list-sessions':\n" + " print('test_session')\n" + "elif action == 'dump-screen':\n" + " print(f'{session} 50.0%/1.0M (auto) (zjet-stub) test_session')\n" + "# write-chars / write : no-op (fast, hermetic)\n" + "sys.exit(0)\n", + ) + zjet_stub.chmod(0o755) + h5i_stub = bin_dir / "h5i" + h5i_stub.write_text("#!/usr/bin/env python3\nimport sys\nsys.exit(0)\n") + h5i_stub.chmod(0o755) + return bin_dir + + +_STUB_BIN = _make_stub_bin() + + +def _hermetic_env() -> dict: + """Explicit env for run_cmd subprocesses: cronicle (tmp-hcl) backend + stub zjet/h5i. + + Forces the real crontab / cronicle.hcl / zjet / h5i bus to NEVER be touched + by integration tests. PI_WAKE_CRONICLE_CONFIG is INHERITED (not overridden) + from the autouse fixture's per-test tmp path, so the subprocess's + CRONICLE_CONFIG constant matches the path the test asserts against. + """ + return { + **os.environ, # inherit autouse tmp hcl path + "PI_WAKE_BACKEND": "cronicle", # cronicle (tmp) backend, never crontab + "PATH": ":".join(p for p in [str(_STUB_BIN), os.environ.get("PATH", "")] if p), + } + + def run_cmd(args, env=None, timeout=30): - """Run the script and return (returncode, stdout, stderr)""" + """Run the script and return (returncode, stdout, stderr). + + Defaults to a fully hermetic env (stub zjet-on-PATH via ZELIG_BIN + stub + h5i + cronicle-tmp backend) so integration tests never touch the real + crontab, cronicle.hcl, zjet, or h5i bus. + """ + if env is None: + env = _hermetic_env() result = subprocess.run( [sys.executable, SCRIPT] + args, capture_output=True, @@ -217,9 +278,10 @@ class TestList: config_path = cronicle_hcl_path() if config_path.exists(): config_path.unlink() - # Clear system cron (harmless under cronicle backend) - subprocess.run(["crontab", "-l"], capture_output=True) - subprocess.run("crontab -l 2>/dev/null | grep -v pi_wake_agent | crontab -", shell=True, check=False) + # Doctrine: do NOT touch the real crontab here. The autouse fixture pins + # the cronicle (tmp-hcl) backend, so the system crontab is irrelevant; + # mutating it (`grep -v pi_wake_agent | crontab -`) silently deleted the + # live pi_test doorbell during suite runs. def test_list_no_cron(self): rc, out, err = run_cmd(["--list"]) @@ -232,14 +294,15 @@ class TestInstallRemove: config_path = cronicle_hcl_path() if config_path.exists(): config_path.unlink() - # Clear system cron - subprocess.run("crontab -l 2>/dev/null | grep -v pi_wake_agent | crontab -", shell=True, check=False) + # Doctrine: no real crontab mutation (cronicle tmp-hcl backend pinned by + # the autouse fixture). Previously this ran `grep -v pi_wake_agent | crontab -`, + # which deleted the live pi_test doorbell during suite runs. def teardown_method(self): config_path = cronicle_hcl_path() if config_path.exists(): config_path.unlink() - subprocess.run("crontab -l 2>/dev/null | grep -v pi_wake_agent | crontab -", shell=True, check=False) + # Doctrine: no real crontab mutation (cronicle tmp-hcl backend pinned). def test_install_and_list(self): rc, out, err = run_cmd(["--install", "--interval", "1h", "--session", "test_session", "--msg", "test"]) @@ -404,13 +467,68 @@ class TestWakeCommandArgs: assert args[args.index("--sessions") + 1] == "s1,s2" assert "s1,s2" in args +class TestContextOccupancy: + """Hermetic: parse_context_occupancy + context_warning never call real zjet. + + Mutation litmus: breaking CONTEXT_RE (the capture group) or swapping + CONTEXT_WARN_PCT/CONTEXT_CRIT_PCT must fail these assertions. + """ + + @staticmethod + def _mock_stdout(monkeypatch, stdout): + m = MagicMock() + m.returncode = 0 + m.stdout = stdout + monkeypatch.setattr( + pi_wake_agent.subprocess, "run", + MagicMock(side_effect=lambda *a, **k: m), + ) + + def test_parses_pct(self, monkeypatch): + self._mock_stdout(monkeypatch, "... 91.3%/1.0M (auto) ...") + assert pi_wake_agent.parse_context_occupancy("pi_test", "terminal_1") == 91.3 + + def test_parses_integer_pct(self, monkeypatch): + self._mock_stdout(monkeypatch, "100%/1.0M") + assert pi_wake_agent.parse_context_occupancy("pi_test") == 100.0 + + def test_no_match_returns_none(self, monkeypatch): + self._mock_stdout(monkeypatch, "no percentage here") + assert pi_wake_agent.parse_context_occupancy("pi_test") is None + + def test_subprocess_failure_returns_none(self, monkeypatch): + monkeypatch.setattr( + pi_wake_agent.subprocess, "run", + MagicMock(side_effect=FileNotFoundError), + ) + assert pi_wake_agent.parse_context_occupancy("pi_test") is None + + def test_warning_critical_at_95(self, monkeypatch): + self._mock_stdout(monkeypatch, "95.0%/1.0M (auto) ...") + w = pi_wake_agent.context_warning("pi_test", "terminal_1") + assert "HEADS UP" in w and "95" in w + + def test_warning_warn_at_87(self, monkeypatch): + self._mock_stdout(monkeypatch, "87.0%/1.0M (auto)") + w = pi_wake_agent.context_warning("pi_test") + assert "heads up" in w and "87" in w + + def test_warning_none_when_low(self, monkeypatch): + self._mock_stdout(monkeypatch, "40.0%/1.0M (auto)") + assert pi_wake_agent.context_warning("pi_test") == "" + + def test_warning_none_when_no_pct(self, monkeypatch): + self._mock_stdout(monkeypatch, "no status bar") + assert pi_wake_agent.context_warning("pi_test") == "" + + # ─── Test pane targeting end-to-end (run_wake with mocked subprocess) ──── class TestPaneTargeting: """Mutation litmus: removing --pane-id from the write-chars / write calls must fail these assertions.""" - def _run(self, pane, message="Please carefully continue your work"): + def _run(self, pane, message="Please carefully continue your work", dump_screen="pi_test\n"): # Resolve the (token) function names without typing the binary name. wc = next(n for n in dir(pi_wake_agent) if n.endswith("_write_chars")) we = next(n for n in dir(pi_wake_agent) if n.endswith("_write_enter")) @@ -423,6 +541,11 @@ class TestPaneTargeting: m.stdout = "pi_test\n" # session exists -> wake proceeds m.returncode = 0 return m + if isinstance(cmd, list) and "dump-screen" in cmd: + m = MagicMock() # pi context-occupancy figure (testable) + m.stdout = dump_screen + m.returncode = 0 + return m m = MagicMock() # write-chars / write / h5i succeed m.returncode = 0 return m @@ -475,6 +598,22 @@ class TestPaneTargeting: calls = self._calls(mr) assert all("--pane-id" not in c for c in calls) + def test_context_warning_appended_to_steer(self): + # When pi's context is critical (>=95%), the steer must carry a heads-up. + mr = self._run(pane="terminal_1", dump_screen="95.0%/1.0M (auto) ...") + calls = self._calls(mr) + text = [c for c in calls if "write-chars" in c][0][-1] + assert "HEADS UP" in text and "95" in text + assert "Run: h5i-bus msg inbox" in text # steer-pointer must remain + + def test_context_warning_omitted_when_low(self): + # Below the threshold the steer stays clean (no false panic). + mr = self._run(pane="terminal_1", dump_screen="40.0%/1.0M (auto)") + calls = self._calls(mr) + text = [c for c in calls if "write-chars" in c][0][-1] + assert "HEADS UP" not in text + assert "heads up" not in text + # ─── Test crontab backend wiring (hermetic: subprocess mocked, real crontab untouched) ── class TestCrontabEntry: @@ -593,6 +732,52 @@ class TestCronicleHclPane: assert '"--run"' in content assert '"--pane"' in content + +class TestHermeticity: + """Doctrine: integration tests must NOT mutate the live crontab/zjet/h5i. + + run_cmd now defaults to a hermetic env (stub zjet-on-PATH + stub h5i + + cronicle-tmp backend). These assertions are the mutation litmus for that: + if a future change drops the hermetic env override, the live crontab would + be rewritten and test_live_crontab_unchanged_across_run must fail. + """ + + @staticmethod + def _live_crontab(): + try: + r = subprocess.run(["crontab", "-l"], capture_output=True, text=True, timeout=5) + return r.stdout if r.returncode == 0 else "" + except Exception: + return "" + + def test_live_crontab_unchanged_across_run(self): + before = self._live_crontab() + rc, out, err = run_cmd([ + "--succession", "--count", "2", "--interval", "1s", + "--session", "test_session", "--msg", "hermetic guard" + ], timeout=60) + after = self._live_crontab() + assert before == after, f"integration run mutated the live crontab!\ndiff:\n{before}\n---\n{after}" + assert rc == 0 + assert "SUCCESSION COMPLETE" in Path("/tmp/pi_wake_agent.log").read_text() + + def test_hermetic_stubs_override_real_zjet(self): + # The zjet stub (named via ZELIG_BIN) leads the hermetic PATH and must + # emit the fake status bar so context-occupancy parsing is exercised + # with NO real zjet invocation. + env = _hermetic_env() + stub_dir = env["PATH"].split(":")[0] + stub = Path(stub_dir) / pi_wake_agent.ZELIG_BIN + assert stub.exists(), f"zjet stub not found at {stub}" + r = subprocess.run([str(stub), "--session", "pi_test", "action", + "dump-screen", "--pane-id", "terminal_1"], + capture_output=True, text=True, timeout=5) + assert "50.0%/1.0M" in r.stdout + # h5i stub is a no-op (no real bus write) + h5i = Path(stub_dir) / "h5i" + assert h5i.exists() + + # ─── Run all tests ──────────────────────────────────────────────────────── if __name__ == "__main__":