From e00717ebed6dd547a3112a6964836f88cce28a74 Mon Sep 17 00:00:00 2001
From: Mukul Sharma <elitecoder.mukul@gmail.com>
Date: Sun, 27 Sep 2026 00:48:18 -0700
Subject: [PATCH 1/8] Revert self-update pulls that bring unparseable code

After a git pull, compile the core pulse files and scan them for leftover
merge conflict markers. If either check fails, reset back to the pre-pull
commit, surface a self-update-syntax-fail ledger entry, and skip the update
so the next launchd restart can't crash-loop on broken code.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 CHANGELOG.md              |  4 ++
 bin/pulse.py              | 15 ++++++-
 bin/self_update.py        | 83 ++++++++++++++++++++++++++++++++++++++
 tests/test_pulse_topup.py | 38 +++++++++++++++++
 tests/test_self_update.py | 85 +++++++++++++++++++++++++++++++++++++++
 5 files changed, 224 insertions(+), 1 deletion(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index d3f2c4c..906af35 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -10,6 +10,10 @@ The version is carried in `pyproject.toml` and `src/assistant/__init__.py`
 ## [Unreleased]
 
 ### Added
+- Gate every self-update pull: after a pull, check that the core pulse files
+  parse and carry no leftover merge conflict markers. If a pull brings broken
+  code, revert it, log the failure, and record it on the dashboard instead of
+  letting the next restart crash on unparseable code.
 - Enforce 100% changed-code coverage with separate Python and real-browser reports.
   Add missing failure-path tests and repeatable mutation checks for key protections.
   Bind reports to measured sources and correctly map multiline Python and JavaScript changes.
diff --git a/bin/pulse.py b/bin/pulse.py
index 612eb21..3558a4c 100755
--- a/bin/pulse.py
+++ b/bin/pulse.py
@@ -461,6 +461,7 @@ def self_update_pulse(pulse_idx: int) -> None:
     if not changed and reason is None and not result.get("error"):
         return
 
+    kind = "self-update"
     if changed:
         files = result.get("files_changed", [])
         installed = result.get("installed")
@@ -495,6 +496,18 @@ def self_update_pulse(pulse_idx: int) -> None:
         outcome = "failed"
         evidence = f"self-update auto-stash failed: {result.get('error', '')}"[:300]
         key = f"self-update-stash-failed-p{pulse_idx}"
+    elif reason == "syntax-fail":
+        # The pull brought unparseable code (conflict markers or a SyntaxError);
+        # self_update reverted it so the pulse won't crash-loop. Surface loudly.
+        outcome = "failed"
+        kind = "self-update-syntax-fail"
+        revert = "reverted" if result.get("revert_ok") else "REVERT FAILED"
+        evidence = (
+            f"blocked broken self-update {result.get('from_sha')}.."
+            f"{result.get('to_sha')} ({revert} to {result.get('reverted_to')}): "
+            f"{result.get('syntax_error', '')}"
+        )[:300]
+        key = f"self-update-syntax-fail-p{pulse_idx}"
     else:
         outcome = "failed"
         evidence = f"self-update {reason or 'error'}: {result.get('error', '')}"[:300]
@@ -505,7 +518,7 @@ def self_update_pulse(pulse_idx: int) -> None:
         "epoch": utc_ts(),
         "pulse_idx": pulse_idx,
         "key": key,
-        "kind": "self-update",
+        "kind": kind,
         "ws_ref": "(launchd)",
         "outcome": outcome,
         "evidence": evidence,
diff --git a/bin/self_update.py b/bin/self_update.py
index 89ac10a..3336b42 100644
--- a/bin/self_update.py
+++ b/bin/self_update.py
@@ -37,6 +37,7 @@
 
 import json
 import subprocess
+import sys
 import time
 from pathlib import Path
 
@@ -61,6 +62,22 @@
 # (`git stash list` / `git stash pop`); it is never dropped or discarded.
 DEFAULT_DIRTY_STASH_AFTER_SEC = 86400  # 1 day
 
+# ── Post-pull syntax gate ────────────────────────────────────────────────────
+# The files a bad pull most commonly corrupts and which MUST parse for the
+# pulse to restart at all. After every pull that changed something, each of
+# these is compiled out-of-process and scanned for git conflict markers; a
+# failure reverts the pull rather than letting the next launchd restart
+# crash-loop on unparseable code (the July 2026 incident: commit 17f3862
+# pulled unresolved conflict markers into bin/pulse.py and the pulse silently
+# throttled for months).
+SYNTAX_GATE_FILES = ("bin/pulse.py", "bin/agent_session.py", "bin/todo-server.py")
+
+# Git writes these at column 0 in an unresolved merge. A committed marker makes
+# a .py file a SyntaxError, so py_compile already catches most of them — but we
+# scan explicitly too, since a marker inside a string or docstring can slip past
+# the compiler yet still corrupt behavior.
+_CONFLICT_MARKERS = ("<<<<<<<", "=======", ">>>>>>>")
+
 
 def _git(repo: Path, *args: str, timeout: int = 90) -> tuple[int, str, str]:
     """Run a git command in `repo`. Returns (rc, stdout, stderr); never raises."""
@@ -178,6 +195,52 @@ def should_attempt(marker: dict, now: float, interval_sec: int) -> bool:
     return (now - last) >= interval_sec
 
 
+def _scan_conflict_markers(paths: list[Path]) -> str | None:
+    """Return "path:line" of the first git conflict marker found, else None.
+
+    Only lines that START with a marker count — a git conflict marker always
+    sits at column 0, so this avoids flagging an incidental `=======` mid-line."""
+    for p in paths:
+        try:
+            text = p.read_text(errors="replace")
+        except OSError:
+            continue
+        for lineno, line in enumerate(text.splitlines(), 1):
+            if any(line.startswith(m) for m in _CONFLICT_MARKERS):
+                return f"{p}:{lineno}"
+    return None
+
+
+def syntax_gate(repo: Path, rel_files: tuple[str, ...] = SYNTAX_GATE_FILES) -> tuple[bool, str]:
+    """Verify the given repo-relative files parse and carry no conflict markers.
+
+    Returns (ok, detail). Scans for conflict markers first (cheap, and catches
+    a marker even inside a string that would still compile), then compiles the
+    existing targets out-of-process with `python -m py_compile` — a true parse
+    by a fresh interpreter, the same way a launchd restart loads them. Missing
+    targets are skipped (a checkout may not ship every file). `detail` names
+    the first failure, or "ok"."""
+    paths = [repo / rel for rel in rel_files if (repo / rel).is_file()]
+
+    marker_hit = _scan_conflict_markers(paths)
+    if marker_hit:
+        return False, f"conflict marker at {marker_hit}"
+
+    if not paths:
+        return True, "ok (no target files present)"
+
+    try:
+        p = subprocess.run(
+            [sys.executable, "-m", "py_compile", *[str(x) for x in paths]],
+            capture_output=True, text=True, timeout=60,
+        )
+    except subprocess.TimeoutExpired:
+        return False, "py_compile timed out after 60s"
+    if p.returncode != 0:
+        return False, (p.stderr or p.stdout).strip()[:500]
+    return True, "ok"
+
+
 def maybe_update(
     repo: Path,
     *,
@@ -299,6 +362,26 @@ def _log(msg: str) -> None:
     if not result["changed"]:
         return result
 
+    # Post-pull syntax gate. If the pull brought code that won't parse (conflict
+    # markers or a SyntaxError), revert it now rather than let the next launchd
+    # restart crash-loop on it. Resetting to the pre-pull SHA is safe here: the
+    # pull was fast-forward-only over a tree we verified clean and zero commits
+    # ahead, so every reverted commit is a REMOTE commit — never operator work.
+    ok, detail = syntax_gate(repo)
+    if not ok:
+        rc, _, rerr = _git(repo, "reset", "--hard", old_head)
+        result["changed"] = False
+        result["skipped_reason"] = "syntax-fail"
+        result["syntax_error"] = detail[:500]
+        result["reverted_to"] = old_head[:12]
+        result["revert_ok"] = rc == 0
+        if rc != 0:
+            result["error"] = f"revert failed after syntax gate: {rerr}"[:300]
+        _log(f"post-pull syntax gate FAILED — reverted {new_head[:12]} → "
+             f"{old_head[:12]} ({'ok' if rc == 0 else 'REVERT FAILED'}): "
+             f"{detail[:200]}")
+        return result
+
     rc, files_out, _ = _git(repo, "diff", "--name-only", f"{old_head}..{new_head}")
     files = [f for f in files_out.splitlines() if f.strip()]
     result["files_changed"] = files
diff --git a/tests/test_pulse_topup.py b/tests/test_pulse_topup.py
index 8ee4f69..c436892 100644
--- a/tests/test_pulse_topup.py
+++ b/tests/test_pulse_topup.py
@@ -216,6 +216,44 @@ def test_self_update_reason_stash_failed(mod, home):
     assert e["key"] == "self-update-stash-failed-p9"
 
 
+def test_self_update_reason_syntax_fail(mod, home):
+    _inject_self_update(mod, {
+        "changed": False, "skipped_reason": "syntax-fail",
+        "from_sha": "aaaaaaaaaaaa", "to_sha": "bbbbbbbbbbbb",
+        "reverted_to": "aaaaaaaaaaaa", "revert_ok": True,
+        "syntax_error": "conflict marker at bin/pulse.py:42",
+    })
+    try:
+        mod.self_update_pulse(13)
+    finally:
+        sys.modules.pop("self_update", None)
+    e = _read_ledger(home)[0]
+    assert e["outcome"] == "failed"
+    assert e["kind"] == "self-update-syntax-fail"
+    assert e["key"] == "self-update-syntax-fail-p13"
+    assert "blocked broken self-update aaaaaaaaaaaa..bbbbbbbbbbbb" in e["evidence"]
+    assert "reverted to aaaaaaaaaaaa" in e["evidence"]
+    assert "conflict marker at bin/pulse.py:42" in e["evidence"]
+
+
+def test_self_update_syntax_fail_revert_failed_note(mod, home):
+    _inject_self_update(mod, {
+        "changed": False, "skipped_reason": "syntax-fail",
+        "from_sha": "aaaa", "to_sha": "bbbb",
+        "reverted_to": "aaaa", "revert_ok": False,
+        "syntax_error": "SyntaxError: invalid syntax",
+        "error": "revert failed after syntax gate: detached",
+    })
+    try:
+        mod.self_update_pulse(14)
+    finally:
+        sys.modules.pop("self_update", None)
+    e = _read_ledger(home)[0]
+    assert e["outcome"] == "failed"
+    assert e["kind"] == "self-update-syntax-fail"
+    assert "REVERT FAILED" in e["evidence"]
+
+
 def test_self_update_other_reason_failed(mod, home):
     _inject_self_update(mod, {
         "changed": False, "skipped_reason": "pull-failed",
diff --git a/tests/test_self_update.py b/tests/test_self_update.py
index dce96c9..1125968 100644
--- a/tests/test_self_update.py
+++ b/tests/test_self_update.py
@@ -480,6 +480,91 @@ def test_git_os_error_returns_minus1(self):
         self.assertIn("no git binary", err)
 
 
+class SyntaxGateTests(unittest.TestCase):
+    """The post-pull gate: a pull that brings unparseable code (conflict
+    markers or a SyntaxError) is reverted, not applied — so the pulse can never
+    crash-loop on it the way it did in July 2026."""
+
+    def test_gate_passes_clean_file(self):
+        with TemporaryDirectory() as t:
+            tmp = Path(t)
+            (tmp / "bin").mkdir()
+            (tmp / "bin/pulse.py").write_text("x = 1\n")
+            ok, detail = su.syntax_gate(tmp, ("bin/pulse.py",))
+            self.assertTrue(ok, detail)
+
+    def test_gate_flags_conflict_marker(self):
+        with TemporaryDirectory() as t:
+            tmp = Path(t)
+            (tmp / "bin").mkdir()
+            (tmp / "bin/pulse.py").write_text(
+                "x = 1\n<<<<<<< HEAD\ny = 2\n=======\ny = 3\n>>>>>>> other\n")
+            ok, detail = su.syntax_gate(tmp, ("bin/pulse.py",))
+            self.assertFalse(ok)
+            self.assertIn("conflict marker at", detail)
+            self.assertIn("bin/pulse.py:2", detail)
+
+    def test_gate_flags_syntax_error_without_markers(self):
+        with TemporaryDirectory() as t:
+            tmp = Path(t)
+            (tmp / "bin").mkdir()
+            (tmp / "bin/pulse.py").write_text("def broken(:\n    pass\n")
+            ok, detail = su.syntax_gate(tmp, ("bin/pulse.py",))
+            self.assertFalse(ok)
+            self.assertNotIn("conflict marker", detail)
+
+    def test_gate_skips_missing_targets(self):
+        with TemporaryDirectory() as t:
+            tmp = Path(t)
+            ok, detail = su.syntax_gate(tmp, ("bin/pulse.py", "bin/absent.py"))
+            self.assertTrue(ok)
+
+    def test_pull_with_conflict_marker_is_reverted(self):
+        with TemporaryDirectory() as t:
+            tmp = Path(t)
+            clone, remote = make_repos(tmp)
+            old_head = git(clone, "rev-parse", "HEAD")
+            advance_remote(
+                tmp, remote,
+                {"bin/pulse.py": "# pulse\n<<<<<<< HEAD\na = 1\n=======\na = 2\n>>>>>>> x\n"},
+                "bad merge with markers")
+            r = su.maybe_update(clone, interval_sec=0, marker_path=tmp / "m.json")
+            self.assertEqual(r["skipped_reason"], "syntax-fail")
+            self.assertFalse(r["changed"])
+            self.assertTrue(r["revert_ok"])
+            self.assertIn("conflict marker", r["syntax_error"])
+            # The bad commit was reverted: HEAD is back where it started and the
+            # working file is the pre-pull content, with no markers.
+            self.assertEqual(git(clone, "rev-parse", "HEAD"), old_head)
+            self.assertNotIn("<<<<<<<", (clone / "bin/pulse.py").read_text())
+            self.assertEqual(git(clone, "status", "--porcelain"), "")
+
+    def test_pull_with_syntax_error_is_reverted(self):
+        with TemporaryDirectory() as t:
+            tmp = Path(t)
+            clone, remote = make_repos(tmp)
+            old_head = git(clone, "rev-parse", "HEAD")
+            advance_remote(tmp, remote, {"bin/pulse.py": "def broken(:\n"},
+                           "syntax error, no markers")
+            r = su.maybe_update(clone, interval_sec=0, marker_path=tmp / "m.json")
+            self.assertEqual(r["skipped_reason"], "syntax-fail")
+            self.assertFalse(r["changed"])
+            self.assertTrue(r["revert_ok"])
+            self.assertEqual(git(clone, "rev-parse", "HEAD"), old_head)
+
+    def test_pull_with_valid_code_still_applies(self):
+        # The gate is transparent to a healthy pull.
+        with TemporaryDirectory() as t:
+            tmp = Path(t)
+            clone, remote = make_repos(tmp)
+            advance_remote(tmp, remote, {"bin/pulse.py": "# pulse v2\nok = True\n"},
+                           "clean change")
+            r = su.maybe_update(clone, interval_sec=0, marker_path=tmp / "m.json")
+            self.assertTrue(r["changed"])
+            self.assertIsNone(r["skipped_reason"])
+            self.assertIn("ok = True", (clone / "bin/pulse.py").read_text())
+
+
 class ResolveRemoteBranchTests(unittest.TestCase):
     def test_detached_head_returns_none(self):
         """When HEAD is detached, resolve_remote_branch returns None."""

From 32b3f31ceac4c84d7ecb6e167e43ef030aecbdca Mon Sep 17 00:00:00 2001
From: Mukul Sharma <elitecoder.mukul@gmail.com>
Date: Sun, 27 Sep 2026 00:51:32 -0700
Subject: [PATCH 2/8] Add launchd pre-flight wrapper that compile-checks
 pulse.py

Route the pulse LaunchAgent through bin/run-pulse.sh, which runs py_compile on
bin/pulse.py before exec'ing it. On failure it logs to the launchd err file and
exits 0, so a broken pulse can't crash-loop launchd into silent throttling.
The plist template now runs the wrapper and passes the arch-resolved python3
as its argument.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 CHANGELOG.md                                  |  4 +
 bin/run-pulse.sh                              | 37 ++++++++++
 .../com.assistant.assistant-pulse.plist       |  2 +-
 tests/test_run_pulse_wrapper.py               | 73 +++++++++++++++++++
 4 files changed, 115 insertions(+), 1 deletion(-)
 create mode 100755 bin/run-pulse.sh
 create mode 100644 tests/test_run_pulse_wrapper.py

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 906af35..c20cc0b 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -14,6 +14,10 @@ The version is carried in `pyproject.toml` and `src/assistant/__init__.py`
   parse and carry no leftover merge conflict markers. If a pull brings broken
   code, revert it, log the failure, and record it on the dashboard instead of
   letting the next restart crash on unparseable code.
+- Run the pulse through a pre-flight wrapper (`bin/run-pulse.sh`) that
+  compile-checks `pulse.py` before each launchd start. If the check fails, it
+  logs the reason and exits cleanly so launchd keeps its schedule instead of
+  throttling on a crash-loop.
 - Enforce 100% changed-code coverage with separate Python and real-browser reports.
   Add missing failure-path tests and repeatable mutation checks for key protections.
   Bind reports to measured sources and correctly map multiline Python and JavaScript changes.
diff --git a/bin/run-pulse.sh b/bin/run-pulse.sh
new file mode 100755
index 0000000..27aeb6d
--- /dev/null
+++ b/bin/run-pulse.sh
@@ -0,0 +1,37 @@
+#!/bin/bash
+# Launchd pre-flight wrapper for the Assistant pulse.
+#
+# Compile-checks bin/pulse.py before running it. If pulse.py won't parse (a
+# self-update that slipped broken code past its own gate, or any local
+# corruption), this logs the failure and exits 0 so launchd keeps the
+# StartInterval schedule alive. A non-zero exit here would make launchd
+# throttle and eventually stop retrying — exactly the silent months-long
+# outage this guards against. The next self-update (or a manual fix) heals the
+# tree and the following run succeeds.
+#
+# The plist passes the arch-resolved python3 as $1 (install.sh's __PYTHON__
+# substitution); the repo is derived from this script's own location.
+set -u
+
+if [ "$#" -gt 0 ]; then
+    PYTHON="$1"
+    shift
+else
+    PYTHON="python3"
+fi
+
+REPO_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+PULSE="$REPO_DIR/bin/pulse.py"
+ERR_LOG="${HOME}/.assistant/logs/assistant-pulse.launchd.err"
+
+if ! compile_out="$("$PYTHON" -m py_compile "$PULSE" 2>&1)"; then
+    mkdir -p "$(dirname "$ERR_LOG")"
+    {
+        printf '[%s] pulse.py pre-flight py_compile FAILED — skipping this run\n' \
+            "$(date -u +%Y-%m-%dT%H:%M:%SZ)"
+        printf '%s\n' "$compile_out"
+    } >> "$ERR_LOG"
+    exit 0
+fi
+
+exec "$PYTHON" "$PULSE" "$@"
diff --git a/launchagents/com.assistant.assistant-pulse.plist b/launchagents/com.assistant.assistant-pulse.plist
index 769cd96..be3144c 100644
--- a/launchagents/com.assistant.assistant-pulse.plist
+++ b/launchagents/com.assistant.assistant-pulse.plist
@@ -6,8 +6,8 @@
     <string>com.assistant.assistant-pulse</string>
     <key>ProgramArguments</key>
     <array>
+        <string>__REPO__/bin/run-pulse.sh</string>
         <string>__PYTHON__</string>
-        <string>__REPO__/bin/pulse.py</string>
     </array>
     <key>StartInterval</key>
     <integer>300</integer>
diff --git a/tests/test_run_pulse_wrapper.py b/tests/test_run_pulse_wrapper.py
new file mode 100644
index 0000000..b503cfe
--- /dev/null
+++ b/tests/test_run_pulse_wrapper.py
@@ -0,0 +1,73 @@
+"""Integration tests for bin/run-pulse.sh — the launchd pre-flight wrapper.
+
+Drives the REAL shell script (byte-for-byte copied into a throwaway repo layout
+with a stub pulse.py) so the compile-check / exit-0-on-failure / arg-passthrough
+behavior is exercised end to end, not asserted from reading the source. This is
+the guard that keeps a bad pulse.py from crash-looping launchd into a silent,
+throttled outage.
+"""
+from __future__ import annotations
+
+import os
+import shutil
+import stat
+import subprocess
+from pathlib import Path
+
+REPO = Path(__file__).resolve().parent.parent
+WRAPPER = REPO / "bin/run-pulse.sh"
+
+
+def _make_layout(tmp: Path, pulse_body: str) -> tuple[Path, Path]:
+    """Copy the real wrapper into tmp/bin next to a stub pulse.py. Returns
+    (wrapper_path, fake_home)."""
+    (tmp / "bin").mkdir()
+    dst = tmp / "bin/run-pulse.sh"
+    shutil.copy2(WRAPPER, dst)
+    dst.chmod(dst.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
+    (tmp / "bin/pulse.py").write_text(pulse_body)
+    home = tmp / "home"
+    home.mkdir()
+    return dst, home
+
+
+def _run(wrapper: Path, home: Path, *args: str) -> subprocess.CompletedProcess:
+    env = dict(os.environ, HOME=str(home))
+    return subprocess.run([str(wrapper), *args], capture_output=True, text=True,
+                          env=env, timeout=60)
+
+
+def test_committed_wrapper_is_executable_with_shebang():
+    assert WRAPPER.exists()
+    assert os.access(WRAPPER, os.X_OK), "run-pulse.sh must be executable for launchd"
+    assert WRAPPER.read_text().startswith("#!/bin/bash")
+
+
+def test_good_pulse_runs_and_passes_args(tmp_path):
+    wrapper, home = _make_layout(
+        tmp_path, 'import sys\nprint("RAN", " ".join(sys.argv[1:]))\n')
+    r = _run(wrapper, home, "python3", "--pulse-idx", "5")
+    assert r.returncode == 0
+    assert "RAN --pulse-idx 5" in r.stdout
+    # A healthy run leaves no failure log behind.
+    assert not (home / ".assistant/logs/assistant-pulse.launchd.err").exists()
+
+
+def test_broken_pulse_exits_zero_and_logs(tmp_path):
+    wrapper, home = _make_layout(tmp_path, "def broken(:\n    pass\n")
+    r = _run(wrapper, home, "python3")
+    # Exit 0 is the whole point: a non-zero exit would make launchd throttle.
+    assert r.returncode == 0, f"wrapper must exit 0 on compile failure, got {r.returncode}"
+    err_log = home / ".assistant/logs/assistant-pulse.launchd.err"
+    assert err_log.exists(), "compile failure must be logged for the operator"
+    text = err_log.read_text()
+    assert "pre-flight py_compile FAILED" in text
+    assert "SyntaxError" in text
+
+
+def test_conflict_markers_block_the_run(tmp_path):
+    wrapper, home = _make_layout(
+        tmp_path, "x = 1\n<<<<<<< HEAD\ny = 2\n=======\ny = 3\n>>>>>>> other\n")
+    r = _run(wrapper, home, "python3")
+    assert r.returncode == 0
+    assert (home / ".assistant/logs/assistant-pulse.launchd.err").exists()

From dc38a662c055b3e1524f7a3b53e5d1fc2444607a Mon Sep 17 00:00:00 2001
From: Mukul Sharma <elitecoder.mukul@gmail.com>
Date: Sun, 27 Sep 2026 01:18:49 -0700
Subject: [PATCH 3/8] Harden syntax gate and correct wrapper docs after review

- Only flag a bare === line as a conflict marker when an arrow marker is also
  present, so an RST doc underline in a gate file can't false-trip the gate and
  revert a healthy pull.
- Never let the gate's py_compile subprocess raise into the pulse.
- Carry the auto-stash recovery hint into the syntax-fail ledger entry.
- Fix run-pulse.sh and CHANGELOG wording: this LaunchAgent has no KeepAlive, so
  it fires every 300s regardless of exit code; a broken pulse.py can't self-heal
  from inside pulse.py; the wrapper reaches existing machines only after reload.
- Add regression tests: RST underline doesn't trip the gate; wrapper good path
  with no extra args.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 CHANGELOG.md                    |  8 +++++---
 bin/pulse.py                    |  7 ++++++-
 bin/run-pulse.sh                | 16 +++++++++++-----
 bin/self_update.py              | 23 +++++++++++++++++------
 tests/test_run_pulse_wrapper.py | 10 ++++++++++
 tests/test_self_update.py       | 11 +++++++++++
 6 files changed, 60 insertions(+), 15 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index c20cc0b..9ee1ecd 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -15,9 +15,11 @@ The version is carried in `pyproject.toml` and `src/assistant/__init__.py`
   code, revert it, log the failure, and record it on the dashboard instead of
   letting the next restart crash on unparseable code.
 - Run the pulse through a pre-flight wrapper (`bin/run-pulse.sh`) that
-  compile-checks `pulse.py` before each launchd start. If the check fails, it
-  logs the reason and exits cleanly so launchd keeps its schedule instead of
-  throttling on a crash-loop.
+  compile-checks `pulse.py` before each start and, on failure, logs the reason
+  and exits cleanly so a broken pulse leaves one clear log line per tick instead
+  of crashing silently. New installs get this right away; existing machines pick
+  it up on the next reinstall or reboot, since the pulse's own LaunchAgent reload
+  is deferred.
 - Enforce 100% changed-code coverage with separate Python and real-browser reports.
   Add missing failure-path tests and repeatable mutation checks for key protections.
   Bind reports to measured sources and correctly map multiline Python and JavaScript changes.
diff --git a/bin/pulse.py b/bin/pulse.py
index 3558a4c..bdc9e39 100755
--- a/bin/pulse.py
+++ b/bin/pulse.py
@@ -502,10 +502,15 @@ def self_update_pulse(pulse_idx: int) -> None:
         outcome = "failed"
         kind = "self-update-syntax-fail"
         revert = "reverted" if result.get("revert_ok") else "REVERT FAILED"
+        stash_note = ""
+        if result.get("stashed"):
+            # An aged-out dirty tree was auto-stashed before this pull. Keep the
+            # recovery path loud so the operator knows their work is parked.
+            stash_note = "; auto-stashed dirty tree (recover: git stash pop)"
         evidence = (
             f"blocked broken self-update {result.get('from_sha')}.."
             f"{result.get('to_sha')} ({revert} to {result.get('reverted_to')}): "
-            f"{result.get('syntax_error', '')}"
+            f"{result.get('syntax_error', '')}{stash_note}"
         )[:300]
         key = f"self-update-syntax-fail-p{pulse_idx}"
     else:
diff --git a/bin/run-pulse.sh b/bin/run-pulse.sh
index 27aeb6d..161f3ee 100755
--- a/bin/run-pulse.sh
+++ b/bin/run-pulse.sh
@@ -3,11 +3,17 @@
 #
 # Compile-checks bin/pulse.py before running it. If pulse.py won't parse (a
 # self-update that slipped broken code past its own gate, or any local
-# corruption), this logs the failure and exits 0 so launchd keeps the
-# StartInterval schedule alive. A non-zero exit here would make launchd
-# throttle and eventually stop retrying — exactly the silent months-long
-# outage this guards against. The next self-update (or a manual fix) heals the
-# tree and the following run succeeds.
+# corruption), this logs the reason to the launchd err file and exits 0 instead
+# of running a doomed pulse.
+#
+# The LaunchAgent is StartInterval=300 + RunAtLoad with no KeepAlive, so it
+# fires every 5 minutes regardless of exit code. The 2026 outage wasn't launchd
+# giving up — it was pulse.py raising SyntaxError on every tick, doing no work
+# for months, with the only trace in an err log nobody watches. Exiting 0 on a
+# failed compile keeps that log to one legible line per tick (no repeated crash
+# traceback, no brief crash-respawn throttle). If the broken file is pulse.py
+# ITSELF, self-update (which runs inside pulse.py) can't heal it and recovery is
+# manual — but the wrapper still turns a silent crash into a clear, logged retry.
 #
 # The plist passes the arch-resolved python3 as $1 (install.sh's __PYTHON__
 # substitution); the repo is derived from this script's own location.
diff --git a/bin/self_update.py b/bin/self_update.py
index 3336b42..696bd66 100644
--- a/bin/self_update.py
+++ b/bin/self_update.py
@@ -198,16 +198,25 @@ def should_attempt(marker: dict, now: float, interval_sec: int) -> bool:
 def _scan_conflict_markers(paths: list[Path]) -> str | None:
     """Return "path:line" of the first git conflict marker found, else None.
 
-    Only lines that START with a marker count — a git conflict marker always
-    sits at column 0, so this avoids flagging an incidental `=======` mid-line."""
+    Only column-0 markers count. The `<<<<<<<` / `>>>>>>>` markers are
+    unambiguous — no valid Python or RST starts a line with seven of them. A
+    bare `=======` line is NOT flagged on its own: a 7-char RST section
+    underline in a docstring is legitimate and would otherwise revert a good
+    pull and loop. It counts only when the file also carries an arrow marker,
+    i.e. it is the divider of a real conflict block."""
     for p in paths:
         try:
-            text = p.read_text(errors="replace")
+            lines = p.read_text(errors="replace").splitlines()
         except OSError:
             continue
-        for lineno, line in enumerate(text.splitlines(), 1):
-            if any(line.startswith(m) for m in _CONFLICT_MARKERS):
-                return f"{p}:{lineno}"
+        markers: list[tuple[int, bool]] = []  # (lineno, is_arrow)
+        for lineno, line in enumerate(lines, 1):
+            if line.startswith("<<<<<<<") or line.startswith(">>>>>>>"):
+                markers.append((lineno, True))
+            elif line.startswith("======="):
+                markers.append((lineno, False))
+        if any(is_arrow for _, is_arrow in markers):
+            return f"{p}:{min(lineno for lineno, _ in markers)}"
     return None
 
 
@@ -236,6 +245,8 @@ def syntax_gate(repo: Path, rel_files: tuple[str, ...] = SYNTAX_GATE_FILES) -> t
         )
     except subprocess.TimeoutExpired:
         return False, "py_compile timed out after 60s"
+    except Exception as e:  # noqa: BLE001 — the gate must never raise into the pulse
+        return False, f"py_compile could not run: {e}"[:500]
     if p.returncode != 0:
         return False, (p.stderr or p.stdout).strip()[:500]
     return True, "ok"
diff --git a/tests/test_run_pulse_wrapper.py b/tests/test_run_pulse_wrapper.py
index b503cfe..94f643a 100644
--- a/tests/test_run_pulse_wrapper.py
+++ b/tests/test_run_pulse_wrapper.py
@@ -43,6 +43,16 @@ def test_committed_wrapper_is_executable_with_shebang():
     assert WRAPPER.read_text().startswith("#!/bin/bash")
 
 
+def test_good_pulse_runs_with_no_extra_args(tmp_path):
+    # The real plist invocation is `run-pulse.sh <python>` with nothing after
+    # it, so exec "$@" runs with an empty $@ under `set -u`. Pin that path.
+    wrapper, home = _make_layout(tmp_path, 'print("RAN")\n')
+    r = _run(wrapper, home, "python3")
+    assert r.returncode == 0
+    assert "RAN" in r.stdout
+    assert not (home / ".assistant/logs/assistant-pulse.launchd.err").exists()
+
+
 def test_good_pulse_runs_and_passes_args(tmp_path):
     wrapper, home = _make_layout(
         tmp_path, 'import sys\nprint("RAN", " ".join(sys.argv[1:]))\n')
diff --git a/tests/test_self_update.py b/tests/test_self_update.py
index 1125968..414f2c8 100644
--- a/tests/test_self_update.py
+++ b/tests/test_self_update.py
@@ -513,6 +513,17 @@ def test_gate_flags_syntax_error_without_markers(self):
             self.assertFalse(ok)
             self.assertNotIn("conflict marker", detail)
 
+    def test_gate_ignores_rst_underline_without_arrows(self):
+        # A bare `=======` line (an RST section underline in a docstring) is
+        # NOT a conflict marker — flagging it would revert a healthy pull.
+        with TemporaryDirectory() as t:
+            tmp = Path(t)
+            (tmp / "bin").mkdir()
+            (tmp / "bin/pulse.py").write_text(
+                '"""Module.\n\nSection\n=======\n\nBody.\n"""\nx = 1\n')
+            ok, detail = su.syntax_gate(tmp, ("bin/pulse.py",))
+            self.assertTrue(ok, detail)
+
     def test_gate_skips_missing_targets(self):
         with TemporaryDirectory() as t:
             tmp = Path(t)

From 055e05d7191e08f8206d7efd6020bfa0ee4b6358 Mon Sep 17 00:00:00 2001
From: Mukul Sharma <elitecoder.mukul@gmail.com>
Date: Sun, 27 Sep 2026 11:04:54 -0700
Subject: [PATCH 4/8] Pin the clock in the focus-unpin brief test so it stops
 expiring

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 CHANGELOG.md                     | 1 +
 tests/test_renderer_brief_tab.py | 4 ++++
 2 files changed, 5 insertions(+)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 9ee1ecd..c034035 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -37,6 +37,7 @@ The version is carried in `pyproject.toml` and `src/assistant/__init__.py`
   and reminders to finish older pending work before starting another task.
 
 ### Fixed
+- Pin the clock in the review-topic focus test so it stops failing once its fixture alert is more than 4 days old.
 - Download browser-check dependencies publicly so CI doesn't require Adobe's internal network.
 - Invalidate return notes after completed tool traffic; require review before reusing older notes.
 - Block close-out for unverified terminals, and safely show incomplete question choices.
diff --git a/tests/test_renderer_brief_tab.py b/tests/test_renderer_brief_tab.py
index 510a7b2..862bd08 100644
--- a/tests/test_renderer_brief_tab.py
+++ b/tests/test_renderer_brief_tab.py
@@ -317,6 +317,10 @@ def test_alert_created_after_focus_unpins_topic_without_hiding_history(self):
         self.write_brief(brief_fixture())
         checked_at = "2026-09-19T10:10:00-07:00"
         checked_epoch = datetime.fromisoformat(checked_at).timestamp()
+        # Pin the clock an hour after the check: freshness decays to 0 within
+        # 4 days, after which "New topic alert" no longer outranks "Old topic alert".
+        self.enterContext(patch.object(self.mod.brief_store.time, "time",
+                                       return_value=checked_epoch + 3600))
         old = {"id": "old-alert", "title": "Old topic alert", "source": "github",
                "refs": {"repo": "adobe/firefly-platform", "pr": 15561},
                "created_epoch": NOW, "epoch": checked_epoch + 60}

From d85877e1031859cc68a9321aa3b62ea29191748c Mon Sep 17 00:00:00 2001
From: Mukul Sharma <elitecoder.mukul@gmail.com>
Date: Sun, 27 Sep 2026 11:59:15 -0700
Subject: [PATCH 5/8] Refuse unparseable self-updates before they land and run
 the pre-flight in Python

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 CHANGELOG.md                                  |  21 +-
 bin/pulse.py                                  |  22 +-
 bin/run-pulse.py                              |  43 ++++
 bin/run-pulse.sh                              |  43 ----
 bin/self_update.py                            | 167 +++++++--------
 .../com.assistant.assistant-pulse.plist       |   2 +-
 tests/test_pulse_topup.py                     |  21 +-
 tests/test_run_pulse_wrapper.py               | 149 ++++++++------
 tests/test_self_update.py                     | 192 +++++++++++-------
 tests/test_self_update_topup.py               |   9 +-
 10 files changed, 355 insertions(+), 314 deletions(-)
 create mode 100755 bin/run-pulse.py
 delete mode 100755 bin/run-pulse.sh

diff --git a/CHANGELOG.md b/CHANGELOG.md
index c034035..c94e0d7 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -10,16 +10,17 @@ The version is carried in `pyproject.toml` and `src/assistant/__init__.py`
 ## [Unreleased]
 
 ### Added
-- Gate every self-update pull: after a pull, check that the core pulse files
-  parse and carry no leftover merge conflict markers. If a pull brings broken
-  code, revert it, log the failure, and record it on the dashboard instead of
-  letting the next restart crash on unparseable code.
-- Run the pulse through a pre-flight wrapper (`bin/run-pulse.sh`) that
-  compile-checks `pulse.py` before each start and, on failure, logs the reason
-  and exits cleanly so a broken pulse leaves one clear log line per tick instead
-  of crashing silently. New installs get this right away; existing machines pick
-  it up on the next reinstall or reboot, since the pulse's own LaunchAgent reload
-  is deferred.
+- Check every self-update before it lands: parse each Python file under `bin/`
+  and `src/` that the fetched commits add or change, straight from git. If any
+  file has leftover merge conflict markers or won't parse, refuse the update
+  before anything is stashed or merged, record the failure on the dashboard
+  once, and skip that commit until the remote moves. Updates now fast-forward
+  to exactly the commit that was checked.
+- Run the pulse through a pre-flight (`bin/run-pulse.py`) that parses `pulse.py`
+  and the `src/` package before each start. If either won't parse, it logs one
+  line to the pulse's launchd error log and exits cleanly instead of crashing.
+  New installs get this right away; existing machines pick it up on the next
+  reinstall or reboot, since the pulse's own LaunchAgent reload is deferred.
 - Enforce 100% changed-code coverage with separate Python and real-browser reports.
   Add missing failure-path tests and repeatable mutation checks for key protections.
   Bind reports to measured sources and correctly map multiline Python and JavaScript changes.
diff --git a/bin/pulse.py b/bin/pulse.py
index bdc9e39..bd32eca 100755
--- a/bin/pulse.py
+++ b/bin/pulse.py
@@ -457,8 +457,9 @@ def self_update_pulse(pulse_idx: int) -> None:
 
     reason = result.get("skipped_reason")
     changed = result.get("changed")
-    # Silent path: attempted, nothing to do, no problem.
-    if not changed and reason is None and not result.get("error"):
+    # Silent path: attempted, nothing to do, no problem — or a refused commit
+    # whose failure was already recorded when it was first refused.
+    if not changed and reason in (None, "syntax-fail-known") and not result.get("error"):
         return
 
     kind = "self-update"
@@ -497,21 +498,12 @@ def self_update_pulse(pulse_idx: int) -> None:
         evidence = f"self-update auto-stash failed: {result.get('error', '')}"[:300]
         key = f"self-update-stash-failed-p{pulse_idx}"
     elif reason == "syntax-fail":
-        # The pull brought unparseable code (conflict markers or a SyntaxError);
-        # self_update reverted it so the pulse won't crash-loop. Surface loudly.
+        # The fetched commits carry Python that won't parse; self_update refused
+        # them before touching the working tree.
         outcome = "failed"
         kind = "self-update-syntax-fail"
-        revert = "reverted" if result.get("revert_ok") else "REVERT FAILED"
-        stash_note = ""
-        if result.get("stashed"):
-            # An aged-out dirty tree was auto-stashed before this pull. Keep the
-            # recovery path loud so the operator knows their work is parked.
-            stash_note = "; auto-stashed dirty tree (recover: git stash pop)"
-        evidence = (
-            f"blocked broken self-update {result.get('from_sha')}.."
-            f"{result.get('to_sha')} ({revert} to {result.get('reverted_to')}): "
-            f"{result.get('syntax_error', '')}{stash_note}"
-        )[:300]
+        evidence = (f"refused self-update {result.get('from_sha')}.."
+                    f"{result.get('to_sha')}: {result.get('syntax_error', '')}")[:300]
         key = f"self-update-syntax-fail-p{pulse_idx}"
     else:
         outcome = "failed"
diff --git a/bin/run-pulse.py b/bin/run-pulse.py
new file mode 100755
index 0000000..3c9c003
--- /dev/null
+++ b/bin/run-pulse.py
@@ -0,0 +1,43 @@
+#!/usr/bin/env python3
+"""Launchd pre-flight for the Assistant pulse.
+
+Parses bin/pulse.py and the src/ package it imports, then replaces this process
+with the pulse. If any of them won't parse, it prints one line to stderr (the
+LaunchAgent sends stderr to ~/.assistant/logs/assistant-pulse.launchd.err) and
+exits 0 instead of starting a pulse that would crash on import. The LaunchAgent
+fires every StartInterval whatever the exit code, so the next tick retries.
+"""
+from __future__ import annotations
+
+import os
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+
+BIN = Path(__file__).resolve().parent
+PULSE = BIN / "pulse.py"
+SRC = BIN.parent / "src"
+
+
+def first_parse_error(paths: list[Path]) -> str | None:
+    for path in paths:
+        try:
+            compile(path.read_bytes(), str(path), "exec", dont_inherit=True)
+        except (SyntaxError, ValueError) as exc:
+            return f"{path}: {type(exc).__name__}: {exc}"
+    return None
+
+
+def main(argv: list[str], *, execv=os.execv) -> int:
+    error = first_parse_error([PULSE, *sorted(SRC.rglob("*.py"))])
+    if error:
+        stamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+        print(f"[{stamp}] pulse pre-flight FAILED, skipping this run: {error}",
+              file=sys.stderr)
+        return 0
+    execv(sys.executable, [sys.executable, str(PULSE), *argv])
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv[1:]))
diff --git a/bin/run-pulse.sh b/bin/run-pulse.sh
deleted file mode 100755
index 161f3ee..0000000
--- a/bin/run-pulse.sh
+++ /dev/null
@@ -1,43 +0,0 @@
-#!/bin/bash
-# Launchd pre-flight wrapper for the Assistant pulse.
-#
-# Compile-checks bin/pulse.py before running it. If pulse.py won't parse (a
-# self-update that slipped broken code past its own gate, or any local
-# corruption), this logs the reason to the launchd err file and exits 0 instead
-# of running a doomed pulse.
-#
-# The LaunchAgent is StartInterval=300 + RunAtLoad with no KeepAlive, so it
-# fires every 5 minutes regardless of exit code. The 2026 outage wasn't launchd
-# giving up — it was pulse.py raising SyntaxError on every tick, doing no work
-# for months, with the only trace in an err log nobody watches. Exiting 0 on a
-# failed compile keeps that log to one legible line per tick (no repeated crash
-# traceback, no brief crash-respawn throttle). If the broken file is pulse.py
-# ITSELF, self-update (which runs inside pulse.py) can't heal it and recovery is
-# manual — but the wrapper still turns a silent crash into a clear, logged retry.
-#
-# The plist passes the arch-resolved python3 as $1 (install.sh's __PYTHON__
-# substitution); the repo is derived from this script's own location.
-set -u
-
-if [ "$#" -gt 0 ]; then
-    PYTHON="$1"
-    shift
-else
-    PYTHON="python3"
-fi
-
-REPO_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
-PULSE="$REPO_DIR/bin/pulse.py"
-ERR_LOG="${HOME}/.assistant/logs/assistant-pulse.launchd.err"
-
-if ! compile_out="$("$PYTHON" -m py_compile "$PULSE" 2>&1)"; then
-    mkdir -p "$(dirname "$ERR_LOG")"
-    {
-        printf '[%s] pulse.py pre-flight py_compile FAILED — skipping this run\n' \
-            "$(date -u +%Y-%m-%dT%H:%M:%SZ)"
-        printf '%s\n' "$compile_out"
-    } >> "$ERR_LOG"
-    exit 0
-fi
-
-exec "$PYTHON" "$PULSE" "$@"
diff --git a/bin/self_update.py b/bin/self_update.py
index 696bd66..b2a8eea 100644
--- a/bin/self_update.py
+++ b/bin/self_update.py
@@ -19,8 +19,14 @@
      clocked from the first pulse that observed it dirty) AND an update is
      waiting, in which case the tree is auto-stashed (`git stash push -u`,
      always recoverable via `git stash pop`) and the pull proceeds.
-  3. If behind: `git pull --ff-only <remote> <branch>`. Fast-forward only,
-     so a diverged history fails loudly rather than merging blindly.
+     Before any stash or pull, every Python file the fetched commits add or
+     change under bin/ or src/ is parsed straight from git. Code that won't
+     parse (conflict markers, a SyntaxError) is refused and never reaches the
+     working tree; that remote commit is remembered and skipped until the
+     remote moves.
+  3. If behind: `git merge --ff-only <fetched sha>` — exactly the commit the
+     gate checked. Fast-forward only, so a diverged history fails loudly
+     rather than merging blindly.
      bin/ and prompts/ are symlinked / read live, so a pull alone makes
      code + Observer-prompt changes take effect on the very next pulse.
   4. If the pull touched COPIED artifacts (skills/, launchagents/, the
@@ -37,7 +43,6 @@
 
 import json
 import subprocess
-import sys
 import time
 from pathlib import Path
 
@@ -62,21 +67,11 @@
 # (`git stash list` / `git stash pop`); it is never dropped or discarded.
 DEFAULT_DIRTY_STASH_AFTER_SEC = 86400  # 1 day
 
-# ── Post-pull syntax gate ────────────────────────────────────────────────────
-# The files a bad pull most commonly corrupts and which MUST parse for the
-# pulse to restart at all. After every pull that changed something, each of
-# these is compiled out-of-process and scanned for git conflict markers; a
-# failure reverts the pull rather than letting the next launchd restart
-# crash-loop on unparseable code (the July 2026 incident: commit 17f3862
-# pulled unresolved conflict markers into bin/pulse.py and the pulse silently
-# throttled for months).
-SYNTAX_GATE_FILES = ("bin/pulse.py", "bin/agent_session.py", "bin/todo-server.py")
-
-# Git writes these at column 0 in an unresolved merge. A committed marker makes
-# a .py file a SyntaxError, so py_compile already catches most of them — but we
-# scan explicitly too, since a marker inside a string or docstring can slip past
-# the compiler yet still corrupt behavior.
-_CONFLICT_MARKERS = ("<<<<<<<", "=======", ">>>>>>>")
+# ── Pre-pull syntax gate ─────────────────────────────────────────────────────
+# Runtime Python lives under these directories. The July 2026 incident (commit
+# 17f3862 pulled unresolved conflict markers into bin/pulse.py and the pulse
+# failed every tick for months) is what the gate exists to stop.
+SYNTAX_GATE_DIRS = ("bin/", "src/")
 
 
 def _git(repo: Path, *args: str, timeout: int = 90) -> tuple[int, str, str]:
@@ -195,60 +190,55 @@ def should_attempt(marker: dict, now: float, interval_sec: int) -> bool:
     return (now - last) >= interval_sec
 
 
-def _scan_conflict_markers(paths: list[Path]) -> str | None:
-    """Return "path:line" of the first git conflict marker found, else None.
+def _conflict_marker_line(text: str) -> int | None:
+    """Line number of the first git conflict marker in `text`, else None.
 
-    Only column-0 markers count. The `<<<<<<<` / `>>>>>>>` markers are
-    unambiguous — no valid Python or RST starts a line with seven of them. A
-    bare `=======` line is NOT flagged on its own: a 7-char RST section
-    underline in a docstring is legitimate and would otherwise revert a good
-    pull and loop. It counts only when the file also carries an arrow marker,
-    i.e. it is the divider of a real conflict block."""
-    for p in paths:
-        try:
-            lines = p.read_text(errors="replace").splitlines()
-        except OSError:
+    Only column-0 markers count. A bare `=======` line alone is a legitimate
+    RST section underline, so it counts only when the text also carries a
+    `<<<<<<<` or `>>>>>>>` line."""
+    first = None
+    arrow = False
+    for n, line in enumerate(text.splitlines(), 1):
+        if line.startswith(("<<<<<<<", ">>>>>>>")):
+            arrow = True
+        elif not line.startswith("======="):
             continue
-        markers: list[tuple[int, bool]] = []  # (lineno, is_arrow)
-        for lineno, line in enumerate(lines, 1):
-            if line.startswith("<<<<<<<") or line.startswith(">>>>>>>"):
-                markers.append((lineno, True))
-            elif line.startswith("======="):
-                markers.append((lineno, False))
-        if any(is_arrow for _, is_arrow in markers):
-            return f"{p}:{min(lineno for lineno, _ in markers)}"
-    return None
-
-
-def syntax_gate(repo: Path, rel_files: tuple[str, ...] = SYNTAX_GATE_FILES) -> tuple[bool, str]:
-    """Verify the given repo-relative files parse and carry no conflict markers.
-
-    Returns (ok, detail). Scans for conflict markers first (cheap, and catches
-    a marker even inside a string that would still compile), then compiles the
-    existing targets out-of-process with `python -m py_compile` — a true parse
-    by a fresh interpreter, the same way a launchd restart loads them. Missing
-    targets are skipped (a checkout may not ship every file). `detail` names
-    the first failure, or "ok"."""
-    paths = [repo / rel for rel in rel_files if (repo / rel).is_file()]
-
-    marker_hit = _scan_conflict_markers(paths)
-    if marker_hit:
-        return False, f"conflict marker at {marker_hit}"
-
-    if not paths:
-        return True, "ok (no target files present)"
+        first = first or n
+    return first if arrow else None
+
 
+def _blob(repo: Path, sha: str, name: str) -> bytes | None:
+    """Raw bytes of `name` at commit `sha`, or None if git can't read it."""
     try:
-        p = subprocess.run(
-            [sys.executable, "-m", "py_compile", *[str(x) for x in paths]],
-            capture_output=True, text=True, timeout=60,
-        )
+        p = subprocess.run(["git", "-C", str(repo), "cat-file", "blob", f"{sha}:{name}"],
+                           capture_output=True, timeout=90)
     except subprocess.TimeoutExpired:
-        return False, "py_compile timed out after 60s"
-    except Exception as e:  # noqa: BLE001 — the gate must never raise into the pulse
-        return False, f"py_compile could not run: {e}"[:500]
-    if p.returncode != 0:
-        return False, (p.stderr or p.stdout).strip()[:500]
+        return None
+    return p.stdout if p.returncode == 0 else None
+
+
+def syntax_gate(repo: Path, old_sha: str, new_sha: str) -> tuple[bool, str]:
+    """Check the Python files under SYNTAX_GATE_DIRS that `new_sha` adds or
+    changes relative to `old_sha`: each must parse and carry no conflict
+    markers. Reads committed blobs, never the working tree. Returns
+    (ok, detail); detail names the first failure, or "ok"."""
+    rc, names, err = _git(repo, "diff", "--name-only", "--diff-filter=ACMR", "-z",
+                          old_sha, new_sha, "--", *SYNTAX_GATE_DIRS)
+    if rc != 0:
+        return False, f"could not list incoming changes: {err}"[:500]
+    for name in names.split("\0"):
+        if not name.endswith(".py"):
+            continue
+        source = _blob(repo, new_sha, name)
+        if source is None:
+            return False, f"could not read {name} at {new_sha[:12]}"
+        marker = _conflict_marker_line(source.decode("utf-8", errors="replace"))
+        if marker:
+            return False, f"conflict marker at {name}:{marker}"
+        try:
+            compile(source, name, "exec", dont_inherit=True)
+        except (SyntaxError, ValueError) as exc:
+            return False, f"{name}: {exc}"[:500]
     return True, "ok"
 
 
@@ -333,6 +323,24 @@ def _log(msg: str) -> None:
         _log("already up to date")
         return result
 
+    # Gate the fetched commit before stashing or pulling anything, so broken
+    # code never reaches the working tree. A commit already refused is skipped
+    # quietly (its failure was recorded once) until the remote moves.
+    old_head, new_sha = status["head"], status["remote_sha"]
+    result["to_sha"] = new_sha[:12]
+    if marker.get("rejected_sha") == new_sha:
+        result["skipped_reason"] = "syntax-fail-known"
+        _log(f"{remote}/{branch} is still at refused {new_sha[:12]}; waiting for a fix")
+        return result
+    ok, detail = syntax_gate(repo, old_head, new_sha)
+    if not ok:
+        marker["rejected_sha"] = new_sha
+        _write_marker(marker_path, marker)
+        result["skipped_reason"] = "syntax-fail"
+        result["syntax_error"] = detail
+        _log(f"refused {old_head[:12]}..{new_sha[:12]}, code won't parse: {detail[:200]}")
+        return result
+
     if status["dirty"]:
         age = result["dirty_age_sec"]
         if age < dirty_stash_after_sec:
@@ -359,12 +367,11 @@ def _log(msg: str) -> None:
              "pull; recover with `git stash pop`")
 
     # Fast-forward only — a diverged history fails rather than merging blindly.
-    old_head = status["head"]
-    rc, _, err = _git(repo, "pull", "--ff-only", remote, branch)
+    rc, _, err = _git(repo, "merge", "--ff-only", new_sha)
     if rc != 0:
         result["skipped_reason"] = "pull-failed"
         result["error"] = err
-        _log(f"git pull --ff-only failed: {err}")
+        _log(f"git merge --ff-only {new_sha[:12]} failed: {err}")
         return result
 
     _, new_head, _ = _git(repo, "rev-parse", "HEAD")
@@ -373,26 +380,6 @@ def _log(msg: str) -> None:
     if not result["changed"]:
         return result
 
-    # Post-pull syntax gate. If the pull brought code that won't parse (conflict
-    # markers or a SyntaxError), revert it now rather than let the next launchd
-    # restart crash-loop on it. Resetting to the pre-pull SHA is safe here: the
-    # pull was fast-forward-only over a tree we verified clean and zero commits
-    # ahead, so every reverted commit is a REMOTE commit — never operator work.
-    ok, detail = syntax_gate(repo)
-    if not ok:
-        rc, _, rerr = _git(repo, "reset", "--hard", old_head)
-        result["changed"] = False
-        result["skipped_reason"] = "syntax-fail"
-        result["syntax_error"] = detail[:500]
-        result["reverted_to"] = old_head[:12]
-        result["revert_ok"] = rc == 0
-        if rc != 0:
-            result["error"] = f"revert failed after syntax gate: {rerr}"[:300]
-        _log(f"post-pull syntax gate FAILED — reverted {new_head[:12]} → "
-             f"{old_head[:12]} ({'ok' if rc == 0 else 'REVERT FAILED'}): "
-             f"{detail[:200]}")
-        return result
-
     rc, files_out, _ = _git(repo, "diff", "--name-only", f"{old_head}..{new_head}")
     files = [f for f in files_out.splitlines() if f.strip()]
     result["files_changed"] = files
diff --git a/launchagents/com.assistant.assistant-pulse.plist b/launchagents/com.assistant.assistant-pulse.plist
index be3144c..a0f37b0 100644
--- a/launchagents/com.assistant.assistant-pulse.plist
+++ b/launchagents/com.assistant.assistant-pulse.plist
@@ -6,8 +6,8 @@
     <string>com.assistant.assistant-pulse</string>
     <key>ProgramArguments</key>
     <array>
-        <string>__REPO__/bin/run-pulse.sh</string>
         <string>__PYTHON__</string>
+        <string>__REPO__/bin/run-pulse.py</string>
     </array>
     <key>StartInterval</key>
     <integer>300</integer>
diff --git a/tests/test_pulse_topup.py b/tests/test_pulse_topup.py
index c436892..84e4d0f 100644
--- a/tests/test_pulse_topup.py
+++ b/tests/test_pulse_topup.py
@@ -220,7 +220,6 @@ def test_self_update_reason_syntax_fail(mod, home):
     _inject_self_update(mod, {
         "changed": False, "skipped_reason": "syntax-fail",
         "from_sha": "aaaaaaaaaaaa", "to_sha": "bbbbbbbbbbbb",
-        "reverted_to": "aaaaaaaaaaaa", "revert_ok": True,
         "syntax_error": "conflict marker at bin/pulse.py:42",
     })
     try:
@@ -231,27 +230,17 @@ def test_self_update_reason_syntax_fail(mod, home):
     assert e["outcome"] == "failed"
     assert e["kind"] == "self-update-syntax-fail"
     assert e["key"] == "self-update-syntax-fail-p13"
-    assert "blocked broken self-update aaaaaaaaaaaa..bbbbbbbbbbbb" in e["evidence"]
-    assert "reverted to aaaaaaaaaaaa" in e["evidence"]
-    assert "conflict marker at bin/pulse.py:42" in e["evidence"]
+    assert e["evidence"] == ("refused self-update aaaaaaaaaaaa..bbbbbbbbbbbb: "
+                             "conflict marker at bin/pulse.py:42")
 
 
-def test_self_update_syntax_fail_revert_failed_note(mod, home):
-    _inject_self_update(mod, {
-        "changed": False, "skipped_reason": "syntax-fail",
-        "from_sha": "aaaa", "to_sha": "bbbb",
-        "reverted_to": "aaaa", "revert_ok": False,
-        "syntax_error": "SyntaxError: invalid syntax",
-        "error": "revert failed after syntax gate: detached",
-    })
+def test_self_update_already_refused_commit_is_silent(mod, home):
+    _inject_self_update(mod, {"changed": False, "skipped_reason": "syntax-fail-known"})
     try:
         mod.self_update_pulse(14)
     finally:
         sys.modules.pop("self_update", None)
-    e = _read_ledger(home)[0]
-    assert e["outcome"] == "failed"
-    assert e["kind"] == "self-update-syntax-fail"
-    assert "REVERT FAILED" in e["evidence"]
+    assert _read_ledger(home) == []
 
 
 def test_self_update_other_reason_failed(mod, home):
diff --git a/tests/test_run_pulse_wrapper.py b/tests/test_run_pulse_wrapper.py
index 94f643a..5b2085b 100644
--- a/tests/test_run_pulse_wrapper.py
+++ b/tests/test_run_pulse_wrapper.py
@@ -1,83 +1,102 @@
-"""Integration tests for bin/run-pulse.sh — the launchd pre-flight wrapper.
+"""Tests for bin/run-pulse.py — the launchd pre-flight for the pulse.
 
-Drives the REAL shell script (byte-for-byte copied into a throwaway repo layout
-with a stub pulse.py) so the compile-check / exit-0-on-failure / arg-passthrough
-behavior is exercised end to end, not asserted from reading the source. This is
-the guard that keeps a bad pulse.py from crash-looping launchd into a silent,
-throttled outage.
+In-process tests call the real module with a fake `execv` so the parse check,
+the logged skip, and the exact exec command are measured. Subprocess tests copy
+the real file next to a stub pulse and let it exec for real.
 """
 from __future__ import annotations
 
+import importlib.util
 import os
+import runpy
 import shutil
-import stat
 import subprocess
+import sys
 from pathlib import Path
+from unittest import mock
 
-REPO = Path(__file__).resolve().parent.parent
-WRAPPER = REPO / "bin/run-pulse.sh"
-
-
-def _make_layout(tmp: Path, pulse_body: str) -> tuple[Path, Path]:
-    """Copy the real wrapper into tmp/bin next to a stub pulse.py. Returns
-    (wrapper_path, fake_home)."""
-    (tmp / "bin").mkdir()
-    dst = tmp / "bin/run-pulse.sh"
-    shutil.copy2(WRAPPER, dst)
-    dst.chmod(dst.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
-    (tmp / "bin/pulse.py").write_text(pulse_body)
-    home = tmp / "home"
-    home.mkdir()
-    return dst, home
+import pytest
 
-
-def _run(wrapper: Path, home: Path, *args: str) -> subprocess.CompletedProcess:
-    env = dict(os.environ, HOME=str(home))
-    return subprocess.run([str(wrapper), *args], capture_output=True, text=True,
-                          env=env, timeout=60)
+REPO = Path(__file__).resolve().parent.parent
+WRAPPER = REPO / "bin/run-pulse.py"
 
 
-def test_committed_wrapper_is_executable_with_shebang():
-    assert WRAPPER.exists()
-    assert os.access(WRAPPER, os.X_OK), "run-pulse.sh must be executable for launchd"
-    assert WRAPPER.read_text().startswith("#!/bin/bash")
+def _load():
+    spec = importlib.util.spec_from_file_location("run_pulse_mod", str(WRAPPER))
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    return mod
 
 
-def test_good_pulse_runs_with_no_extra_args(tmp_path):
-    # The real plist invocation is `run-pulse.sh <python>` with nothing after
-    # it, so exec "$@" runs with an empty $@ under `set -u`. Pin that path.
-    wrapper, home = _make_layout(tmp_path, 'print("RAN")\n')
-    r = _run(wrapper, home, "python3")
+def _layout(tmp: Path, pulse_body: str, src_files: dict[str, str] | None = None) -> Path:
+    (tmp / "bin").mkdir()
+    shutil.copy2(WRAPPER, tmp / "bin/run-pulse.py")
+    (tmp / "bin/pulse.py").write_text(pulse_body)
+    for rel, body in (src_files or {}).items():
+        path = tmp / "src" / rel
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(body)
+    return tmp / "bin/run-pulse.py"
+
+
+def test_healthy_checkout_execs_pulse_with_same_interpreter_and_args():
+    mod = _load()
+    calls = []
+    assert mod.main(["--pulse-idx", "5"], execv=lambda *a: calls.append(a)) == 0
+    assert calls == [(sys.executable, [sys.executable, str(REPO / "bin/pulse.py"),
+                                       "--pulse-idx", "5"])]
+
+
+def test_broken_pulse_skips_the_run_and_logs(tmp_path, capsys):
+    _layout(tmp_path, "def broken(:\n    pass\n")
+    mod = _load()
+    calls = []
+    with mock.patch.object(mod, "PULSE", tmp_path / "bin/pulse.py"), \
+            mock.patch.object(mod, "SRC", tmp_path / "src"):
+        assert mod.main([], execv=lambda *a: calls.append(a)) == 0
+    assert calls == []
+    err = capsys.readouterr().err
+    assert "pulse pre-flight FAILED, skipping this run" in err
+    assert f"{tmp_path / 'bin/pulse.py'}: SyntaxError" in err
+
+
+def test_broken_src_module_skips_the_run(tmp_path, capsys):
+    _layout(tmp_path, "x = 1\n", {"assistant/__init__.py": "",
+                                  "assistant/model_tiers.py": "x = 1\n<<<<<<< HEAD\n"})
+    mod = _load()
+    calls = []
+    with mock.patch.object(mod, "PULSE", tmp_path / "bin/pulse.py"), \
+            mock.patch.object(mod, "SRC", tmp_path / "src"):
+        assert mod.main([], execv=lambda *a: calls.append(a)) == 0
+    assert calls == []
+    assert "assistant/model_tiers.py: SyntaxError" in capsys.readouterr().err
+
+
+def test_script_entry_point_execs_pulse():
+    calls = []
+    with mock.patch.object(os, "execv", lambda *a: calls.append(a)), \
+            mock.patch.object(sys, "argv", [str(WRAPPER), "--dry-run"]):
+        with pytest.raises(SystemExit) as exc:
+            runpy.run_path(str(WRAPPER), run_name="__main__")
+    assert exc.value.code == 0
+    assert calls == [(sys.executable, [sys.executable, str(REPO / "bin/pulse.py"),
+                                       "--dry-run"])]
+
+
+def test_real_exec_runs_pulse_and_passes_args(tmp_path):
+    wrapper = _layout(tmp_path, 'import sys\nprint("RAN", " ".join(sys.argv[1:]))\n',
+                      {"assistant/__init__.py": ""})
+    r = subprocess.run([sys.executable, str(wrapper), "--pulse-idx", "5"],
+                       capture_output=True, text=True, timeout=60)
     assert r.returncode == 0
-    assert "RAN" in r.stdout
-    assert not (home / ".assistant/logs/assistant-pulse.launchd.err").exists()
+    assert r.stdout.strip() == "RAN --pulse-idx 5"
+    assert r.stderr == ""
 
 
-def test_good_pulse_runs_and_passes_args(tmp_path):
-    wrapper, home = _make_layout(
-        tmp_path, 'import sys\nprint("RAN", " ".join(sys.argv[1:]))\n')
-    r = _run(wrapper, home, "python3", "--pulse-idx", "5")
-    assert r.returncode == 0
-    assert "RAN --pulse-idx 5" in r.stdout
-    # A healthy run leaves no failure log behind.
-    assert not (home / ".assistant/logs/assistant-pulse.launchd.err").exists()
-
-
-def test_broken_pulse_exits_zero_and_logs(tmp_path):
-    wrapper, home = _make_layout(tmp_path, "def broken(:\n    pass\n")
-    r = _run(wrapper, home, "python3")
-    # Exit 0 is the whole point: a non-zero exit would make launchd throttle.
-    assert r.returncode == 0, f"wrapper must exit 0 on compile failure, got {r.returncode}"
-    err_log = home / ".assistant/logs/assistant-pulse.launchd.err"
-    assert err_log.exists(), "compile failure must be logged for the operator"
-    text = err_log.read_text()
-    assert "pre-flight py_compile FAILED" in text
-    assert "SyntaxError" in text
-
-
-def test_conflict_markers_block_the_run(tmp_path):
-    wrapper, home = _make_layout(
-        tmp_path, "x = 1\n<<<<<<< HEAD\ny = 2\n=======\ny = 3\n>>>>>>> other\n")
-    r = _run(wrapper, home, "python3")
+def test_real_run_with_broken_pulse_exits_zero(tmp_path):
+    wrapper = _layout(tmp_path, "def broken(:\n")
+    r = subprocess.run([sys.executable, str(wrapper)],
+                       capture_output=True, text=True, timeout=60)
     assert r.returncode == 0
-    assert (home / ".assistant/logs/assistant-pulse.launchd.err").exists()
+    assert r.stdout == ""
+    assert "pulse pre-flight FAILED" in r.stderr
diff --git a/tests/test_self_update.py b/tests/test_self_update.py
index 414f2c8..4c59668 100644
--- a/tests/test_self_update.py
+++ b/tests/test_self_update.py
@@ -481,99 +481,151 @@ def test_git_os_error_returns_minus1(self):
 
 
 class SyntaxGateTests(unittest.TestCase):
-    """The post-pull gate: a pull that brings unparseable code (conflict
-    markers or a SyntaxError) is reverted, not applied — so the pulse can never
-    crash-loop on it the way it did in July 2026."""
+    """The pre-pull gate: fetched commits whose Python won't parse (conflict
+    markers or a SyntaxError) are refused before anything is stashed, merged,
+    or reset — so the pulse can never crash on them the way it did in July 2026."""
 
-    def test_gate_passes_clean_file(self):
-        with TemporaryDirectory() as t:
-            tmp = Path(t)
-            (tmp / "bin").mkdir()
-            (tmp / "bin/pulse.py").write_text("x = 1\n")
-            ok, detail = su.syntax_gate(tmp, ("bin/pulse.py",))
-            self.assertTrue(ok, detail)
-
-    def test_gate_flags_conflict_marker(self):
-        with TemporaryDirectory() as t:
-            tmp = Path(t)
-            (tmp / "bin").mkdir()
-            (tmp / "bin/pulse.py").write_text(
-                "x = 1\n<<<<<<< HEAD\ny = 2\n=======\ny = 3\n>>>>>>> other\n")
-            ok, detail = su.syntax_gate(tmp, ("bin/pulse.py",))
-            self.assertFalse(ok)
-            self.assertIn("conflict marker at", detail)
-            self.assertIn("bin/pulse.py:2", detail)
+    def _bad_remote(self, tmp: Path, files: dict[str, str]) -> tuple[Path, str, str]:
+        clone, remote = make_repos(tmp)
+        old_head = git(clone, "rev-parse", "HEAD")
+        new_sha = advance_remote(tmp, remote, files, "incoming")
+        return clone, old_head, new_sha
 
-    def test_gate_flags_syntax_error_without_markers(self):
+    def _gate(self, files: dict[str, str]) -> tuple[bool, str]:
         with TemporaryDirectory() as t:
-            tmp = Path(t)
-            (tmp / "bin").mkdir()
-            (tmp / "bin/pulse.py").write_text("def broken(:\n    pass\n")
-            ok, detail = su.syntax_gate(tmp, ("bin/pulse.py",))
-            self.assertFalse(ok)
-            self.assertNotIn("conflict marker", detail)
-
-    def test_gate_ignores_rst_underline_without_arrows(self):
-        # A bare `=======` line (an RST section underline in a docstring) is
-        # NOT a conflict marker — flagging it would revert a healthy pull.
-        with TemporaryDirectory() as t:
-            tmp = Path(t)
-            (tmp / "bin").mkdir()
-            (tmp / "bin/pulse.py").write_text(
-                '"""Module.\n\nSection\n=======\n\nBody.\n"""\nx = 1\n')
-            ok, detail = su.syntax_gate(tmp, ("bin/pulse.py",))
-            self.assertTrue(ok, detail)
+            clone, old_head, new_sha = self._bad_remote(Path(t), files)
+            git(clone, "fetch", "origin", "main")
+            return su.syntax_gate(clone, old_head, new_sha)
 
-    def test_gate_skips_missing_targets(self):
+    def test_conflict_markers_are_refused_before_touching_the_tree(self):
         with TemporaryDirectory() as t:
             tmp = Path(t)
-            ok, detail = su.syntax_gate(tmp, ("bin/pulse.py", "bin/absent.py"))
-            self.assertTrue(ok)
-
-    def test_pull_with_conflict_marker_is_reverted(self):
-        with TemporaryDirectory() as t:
-            tmp = Path(t)
-            clone, remote = make_repos(tmp)
-            old_head = git(clone, "rev-parse", "HEAD")
-            advance_remote(
-                tmp, remote,
-                {"bin/pulse.py": "# pulse\n<<<<<<< HEAD\na = 1\n=======\na = 2\n>>>>>>> x\n"},
-                "bad merge with markers")
+            clone, old_head, new_sha = self._bad_remote(
+                tmp, {"bin/pulse.py": "# pulse\n<<<<<<< HEAD\na = 1\n=======\na = 2\n>>>>>>> x\n"})
+            reflog = git(clone, "reflog")
             r = su.maybe_update(clone, interval_sec=0, marker_path=tmp / "m.json")
             self.assertEqual(r["skipped_reason"], "syntax-fail")
             self.assertFalse(r["changed"])
-            self.assertTrue(r["revert_ok"])
-            self.assertIn("conflict marker", r["syntax_error"])
-            # The bad commit was reverted: HEAD is back where it started and the
-            # working file is the pre-pull content, with no markers.
+            self.assertEqual(r["syntax_error"], "conflict marker at bin/pulse.py:2")
+            self.assertEqual(r["to_sha"], new_sha[:12])
             self.assertEqual(git(clone, "rev-parse", "HEAD"), old_head)
-            self.assertNotIn("<<<<<<<", (clone / "bin/pulse.py").read_text())
-            self.assertEqual(git(clone, "status", "--porcelain"), "")
+            self.assertEqual(git(clone, "reflog"), reflog)  # no merge, no reset
+            self.assertEqual((clone / "bin/pulse.py").read_text(), "# pulse\n")
+            self.assertEqual(json.loads((tmp / "m.json").read_text())["rejected_sha"], new_sha)
 
-    def test_pull_with_syntax_error_is_reverted(self):
+    def test_broken_src_module_is_refused(self):
         with TemporaryDirectory() as t:
             tmp = Path(t)
-            clone, remote = make_repos(tmp)
-            old_head = git(clone, "rev-parse", "HEAD")
-            advance_remote(tmp, remote, {"bin/pulse.py": "def broken(:\n"},
-                           "syntax error, no markers")
+            clone, old_head, _ = self._bad_remote(
+                tmp, {"src/assistant/model_tiers.py": "def broken(:\n"})
             r = su.maybe_update(clone, interval_sec=0, marker_path=tmp / "m.json")
             self.assertEqual(r["skipped_reason"], "syntax-fail")
-            self.assertFalse(r["changed"])
-            self.assertTrue(r["revert_ok"])
+            self.assertTrue(r["syntax_error"].startswith("src/assistant/model_tiers.py: "))
             self.assertEqual(git(clone, "rev-parse", "HEAD"), old_head)
+            self.assertFalse((clone / "src").exists())
 
-    def test_pull_with_valid_code_still_applies(self):
-        # The gate is transparent to a healthy pull.
+    def test_refused_commit_is_skipped_until_the_remote_moves(self):
         with TemporaryDirectory() as t:
             tmp = Path(t)
             clone, remote = make_repos(tmp)
-            advance_remote(tmp, remote, {"bin/pulse.py": "# pulse v2\nok = True\n"},
-                           "clean change")
+            advance_remote(tmp, remote, {"bin/pulse.py": "def broken(:\n"}, "bad")
+            marker = tmp / "m.json"
+            self.assertEqual(su.maybe_update(clone, interval_sec=0, marker_path=marker)
+                             ["skipped_reason"], "syntax-fail")
+            with unittest.mock.patch.object(su, "syntax_gate") as gate:
+                r = su.maybe_update(clone, interval_sec=0, marker_path=marker)
+            gate.assert_not_called()
+            self.assertEqual(r["skipped_reason"], "syntax-fail-known")
+            fixed = advance_remote(tmp, remote, {"bin/pulse.py": "fixed = True\n"}, "fix")
+            r = su.maybe_update(clone, interval_sec=0, marker_path=marker)
+            self.assertTrue(r["changed"])
+            self.assertEqual(git(clone, "rev-parse", "HEAD"), fixed)
+            self.assertEqual((clone / "bin/pulse.py").read_text(), "fixed = True\n")
+
+    def test_dirty_tree_past_window_is_not_stashed_for_a_refused_update(self):
+        with TemporaryDirectory() as t:
+            tmp = Path(t)
+            clone, _, _ = self._bad_remote(tmp, {"bin/pulse.py": "def broken(:\n"})
+            (clone / "install.sh").write_text("# operator edit\n")
+            marker = tmp / "m.json"
+            first = su.maybe_update(clone, interval_sec=0, marker_path=marker,
+                                    dirty_stash_after_sec=86400, now=1000.0)
+            later = su.maybe_update(clone, interval_sec=0, marker_path=marker,
+                                    dirty_stash_after_sec=86400, now=1000.0 + 25 * 3600)
+            self.assertEqual(first["skipped_reason"], "syntax-fail")
+            self.assertEqual(later["skipped_reason"], "syntax-fail-known")
+            self.assertNotIn("stashed", later)
+            self.assertEqual(git(clone, "stash", "list"), "")
+            self.assertEqual((clone / "install.sh").read_text(), "# operator edit\n")
+
+    def test_healthy_update_lands_on_the_gated_commit(self):
+        with TemporaryDirectory() as t:
+            tmp = Path(t)
+            clone, _, new_sha = self._bad_remote(tmp, {
+                "bin/pulse.py": '"""Pulse.\n\nSection\n=======\n"""\nok = True\n',
+                "tests/test_scratch.py": "def broken(:\n",
+                "docs/merging.md": "<<<<<<< HEAD\n=======\n>>>>>>> x\n",
+            })
             r = su.maybe_update(clone, interval_sec=0, marker_path=tmp / "m.json")
             self.assertTrue(r["changed"])
             self.assertIsNone(r["skipped_reason"])
-            self.assertIn("ok = True", (clone / "bin/pulse.py").read_text())
+            self.assertEqual(git(clone, "rev-parse", "HEAD"), new_sha)
+
+    def test_marker_inside_a_string_is_flagged_even_though_it_compiles(self):
+        ok, detail = self._gate({"bin/pulse.py": 'X = """\n>>>>>>> theirs\n"""\n'})
+        self.assertFalse(ok)
+        self.assertEqual(detail, "conflict marker at bin/pulse.py:2")
+
+    def test_marker_line_ignores_rst_underline_alone(self):
+        self.assertIsNone(su._conflict_marker_line("Title\n=======\nbody\n"))
+        self.assertEqual(su._conflict_marker_line("a\n=======\n>>>>>>> b\n"), 2)
+
+    def test_unreadable_diff_is_refused(self):
+        with TemporaryDirectory() as t:
+            clone, _ = make_repos(Path(t))
+            ok, detail = su.syntax_gate(clone, "0" * 40, "HEAD")
+            self.assertFalse(ok)
+            self.assertTrue(detail.startswith("could not list incoming changes: "))
+
+    def test_unreadable_blob_is_refused(self):
+        with TemporaryDirectory() as t:
+            clone, old_head, new_sha = self._bad_remote(Path(t), {"bin/new.py": "x = 1\n"})
+            git(clone, "fetch", "origin", "main")
+            real_run = subprocess.run
+
+            def run(cmd, **kwargs):
+                if "cat-file" in cmd:
+                    raise subprocess.TimeoutExpired(cmd, 90)
+                return real_run(cmd, **kwargs)
+
+            with unittest.mock.patch.object(su.subprocess, "run", run):
+                ok, detail = su.syntax_gate(clone, old_head, new_sha)
+            self.assertFalse(ok)
+            self.assertEqual(detail, f"could not read bin/new.py at {new_sha[:12]}")
+
+    def test_blob_missing_from_commit_reads_as_none(self):
+        with TemporaryDirectory() as t:
+            clone, _ = make_repos(Path(t))
+            self.assertIsNone(su._blob(clone, "HEAD", "bin/absent.py"))
+            self.assertEqual(su._blob(clone, "HEAD", "bin/pulse.py"), b"# pulse\n")
+
+    def test_failed_fast_forward_is_reported(self):
+        with TemporaryDirectory() as t:
+            tmp = Path(t)
+            clone, old_head, new_sha = self._bad_remote(tmp, {"bin/pulse.py": "ok = 1\n"})
+            real_git = su._git
+
+            def fake_git(repo, *args, **kwargs):
+                if args[0] == "merge":
+                    self.assertEqual(args, ("merge", "--ff-only", new_sha))
+                    return 1, "", "not possible to fast-forward"
+                return real_git(repo, *args, **kwargs)
+
+            with unittest.mock.patch.object(su, "_git", fake_git):
+                r = su.maybe_update(clone, interval_sec=0, marker_path=tmp / "m.json")
+            self.assertEqual(r["skipped_reason"], "pull-failed")
+            self.assertEqual(r["error"], "not possible to fast-forward")
+            self.assertEqual(git(clone, "rev-parse", "HEAD"), old_head)
 
 
 class ResolveRemoteBranchTests(unittest.TestCase):
diff --git a/tests/test_self_update_topup.py b/tests/test_self_update_topup.py
index aa391a6..ec6663a 100644
--- a/tests/test_self_update_topup.py
+++ b/tests/test_self_update_topup.py
@@ -90,7 +90,8 @@ def test_maybe_update_stash_failed(tmp_path):
                 "head": "oldsha", "remote_sha": "newsha",
                 "dirty": True, "behind": 1, "ahead": 0}):
             with mock.patch.object(su, "_stash_dirty",
-                                   return_value=(False, "fatal: stash conflict")):
+                                   return_value=(False, "fatal: stash conflict")), \
+                    mock.patch.object(su, "syntax_gate", return_value=(True, "ok")):
                 # First pass stamps dirty_since at t=1000.
                 su.maybe_update(tmp_path, interval_sec=0, marker_path=marker,
                                 dirty_stash_after_sec=86400, now=1000.0)
@@ -107,12 +108,12 @@ def test_maybe_update_stash_failed(tmp_path):
 # ─── maybe_update: pull-failed path (291-294) ─────────────────────────────────
 
 def test_maybe_update_pull_failed(tmp_path):
-    """Clean tree, behind, but `git pull --ff-only` fails (e.g. diverged) →
+    """Clean tree, behind, but `git merge --ff-only` fails (e.g. diverged) →
     skipped_reason 'pull-failed' with the git error captured."""
     marker = tmp_path / "m.json"
 
     def fake_git(repo, *args, **kw):
-        if args[:1] == ("pull",):
+        if args[:1] == ("merge",):
             return (1, "", "fatal: Not possible to fast-forward, aborting.")
         # rev-parse HEAD after a failed pull would not be reached.
         return (0, "", "")
@@ -137,7 +138,7 @@ def test_maybe_update_pull_noop_to_sha_set(tmp_path):
     marker = tmp_path / "m.json"
 
     def fake_git(repo, *args, **kw):
-        if args[:1] == ("pull",):
+        if args[:1] == ("merge",):
             return (0, "Already up to date.", "")
         if args[:2] == ("rev-parse", "HEAD"):
             return (0, "oldsha", "")  # same as status head → no change

From 12aa74b25f6213dea0845ece200ec1492d13539d Mon Sep 17 00:00:00 2001
From: Mukul Sharma <elitecoder.mukul@gmail.com>
Date: Sun, 27 Sep 2026 14:38:43 -0700
Subject: [PATCH 6/8] Keep git errors out of the refusal memory, scan all
 runtime files, and narrow the pre-flight

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 CHANGELOG.md                      | 20 +++----
 bin/run-pulse.py                  | 34 +++++-------
 bin/self_update.py                | 89 +++++++++++++++++++------------
 src/assistant/subsystems/pulse.py |  5 +-
 tests/test_pulse_topup.py         | 13 +++++
 tests/test_run_pulse_wrapper.py   | 32 +++++------
 tests/test_self_update.py         | 87 +++++++++++++++++++++++-------
 7 files changed, 173 insertions(+), 107 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index c94e0d7..149bf31 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -10,17 +10,17 @@ The version is carried in `pyproject.toml` and `src/assistant/__init__.py`
 ## [Unreleased]
 
 ### Added
-- Check every self-update before it lands: parse each Python file under `bin/`
-  and `src/` that the fetched commits add or change, straight from git. If any
-  file has leftover merge conflict markers or won't parse, refuse the update
-  before anything is stashed or merged, record the failure on the dashboard
-  once, and skip that commit until the remote moves. Updates now fast-forward
-  to exactly the commit that was checked.
+- Check every self-update before it lands: read each runtime file the fetched
+  commits add or change straight from git, scan it for leftover merge conflict
+  markers, and parse it if it's Python. If a file fails, refuse the update
+  before anything is stashed or merged, and record the failure on the
+  dashboard. Skip that commit quietly until the remote moves, with a reminder
+  once a day. Updates now fast-forward to exactly the commit that was checked.
 - Run the pulse through a pre-flight (`bin/run-pulse.py`) that parses `pulse.py`
-  and the `src/` package before each start. If either won't parse, it logs one
-  line to the pulse's launchd error log and exits cleanly instead of crashing.
-  New installs get this right away; existing machines pick it up on the next
-  reinstall or reboot, since the pulse's own LaunchAgent reload is deferred.
+  before each start. If it won't parse, the pre-flight logs one line to the
+  pulse's launchd error log and exits cleanly instead of crashing. New installs
+  get this right away. Existing machines pick it up after a reboot, a logout, or
+  a manual reload of the pulse LaunchAgent, since self-update defers that reload.
 - Enforce 100% changed-code coverage with separate Python and real-browser reports.
   Add missing failure-path tests and repeatable mutation checks for key protections.
   Bind reports to measured sources and correctly map multiline Python and JavaScript changes.
diff --git a/bin/run-pulse.py b/bin/run-pulse.py
index 3c9c003..0058ef7 100755
--- a/bin/run-pulse.py
+++ b/bin/run-pulse.py
@@ -1,11 +1,13 @@
 #!/usr/bin/env python3
 """Launchd pre-flight for the Assistant pulse.
 
-Parses bin/pulse.py and the src/ package it imports, then replaces this process
-with the pulse. If any of them won't parse, it prints one line to stderr (the
-LaunchAgent sends stderr to ~/.assistant/logs/assistant-pulse.launchd.err) and
-exits 0 instead of starting a pulse that would crash on import. The LaunchAgent
-fires every StartInterval whatever the exit code, so the next tick retries.
+Parses bin/pulse.py, then replaces this process with it. If pulse.py won't
+parse, it prints one line to stderr (the LaunchAgent sends stderr to
+~/.assistant/logs/assistant-pulse.launchd.err) and exits 0 instead of starting a
+pulse that would crash on its first line. The LaunchAgent fires every
+StartInterval whatever the exit code, so the next tick retries. Other modules
+aren't checked here: the pulse guards its own optional imports, and
+self_update.py refuses incoming commits that don't parse.
 """
 from __future__ import annotations
 
@@ -14,26 +16,16 @@
 from datetime import datetime, timezone
 from pathlib import Path
 
-BIN = Path(__file__).resolve().parent
-PULSE = BIN / "pulse.py"
-SRC = BIN.parent / "src"
-
-
-def first_parse_error(paths: list[Path]) -> str | None:
-    for path in paths:
-        try:
-            compile(path.read_bytes(), str(path), "exec", dont_inherit=True)
-        except (SyntaxError, ValueError) as exc:
-            return f"{path}: {type(exc).__name__}: {exc}"
-    return None
+PULSE = Path(__file__).resolve().parent / "pulse.py"
 
 
 def main(argv: list[str], *, execv=os.execv) -> int:
-    error = first_parse_error([PULSE, *sorted(SRC.rglob("*.py"))])
-    if error:
+    try:
+        compile(PULSE.read_bytes(), str(PULSE), "exec", dont_inherit=True)
+    except (SyntaxError, ValueError) as exc:
         stamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
-        print(f"[{stamp}] pulse pre-flight FAILED, skipping this run: {error}",
-              file=sys.stderr)
+        print(f"[{stamp}] pulse pre-flight FAILED, skipping this run: "
+              f"{PULSE}: {type(exc).__name__}: {exc}", file=sys.stderr)
         return 0
     execv(sys.executable, [sys.executable, str(PULSE), *argv])
     return 0
diff --git a/bin/self_update.py b/bin/self_update.py
index b2a8eea..d3022ff 100644
--- a/bin/self_update.py
+++ b/bin/self_update.py
@@ -19,11 +19,11 @@
      clocked from the first pulse that observed it dirty) AND an update is
      waiting, in which case the tree is auto-stashed (`git stash push -u`,
      always recoverable via `git stash pop`) and the pull proceeds.
-     Before any stash or pull, every Python file the fetched commits add or
-     change under bin/ or src/ is parsed straight from git. Code that won't
-     parse (conflict markers, a SyntaxError) is refused and never reaches the
-     working tree; that remote commit is remembered and skipped until the
-     remote moves.
+     Before any stash or pull, every runtime file the fetched commits add or
+     change is read straight from git: all are scanned for conflict markers,
+     and Python files are parsed. A broken commit is refused and never reaches
+     the working tree; it's skipped quietly until the remote moves, with a
+     reminder once a day.
   3. If behind: `git merge --ff-only <fetched sha>` — exactly the commit the
      gate checked. Fast-forward only, so a diverged history fails loudly
      rather than merging blindly.
@@ -43,6 +43,7 @@
 
 import json
 import subprocess
+import sys
 import time
 from pathlib import Path
 
@@ -68,10 +69,16 @@
 DEFAULT_DIRTY_STASH_AFTER_SEC = 86400  # 1 day
 
 # ── Pre-pull syntax gate ─────────────────────────────────────────────────────
-# Runtime Python lives under these directories. The July 2026 incident (commit
-# 17f3862 pulled unresolved conflict markers into bin/pulse.py and the pulse
-# failed every tick for months) is what the gate exists to stop.
-SYNTAX_GATE_DIRS = ("bin/", "src/")
+# Paths the running system loads, runs, or installs from the checkout. Add new
+# runtime paths here. The July 2026 incident (commit 17f3862 pulled unresolved
+# conflict markers into bin/pulse.py and the pulse failed every tick for
+# months) is what the gate exists to stop.
+SYNTAX_GATE_PATHS = ("bin/", "src/", "hooks/", "install/", "prompts/", "skills/",
+                     "launchagents/", "config/", "slack-reactor/", "install.sh",
+                     "install-bootstrap.sh")
+
+# A refused commit is re-checked, and its failure re-recorded, this often.
+REJECT_REMIND_SEC = 86400  # 1 day
 
 
 def _git(repo: Path, *args: str, timeout: int = 90) -> tuple[int, str, str]:
@@ -217,29 +224,32 @@ def _blob(repo: Path, sha: str, name: str) -> bytes | None:
     return p.stdout if p.returncode == 0 else None
 
 
-def syntax_gate(repo: Path, old_sha: str, new_sha: str) -> tuple[bool, str]:
-    """Check the Python files under SYNTAX_GATE_DIRS that `new_sha` adds or
-    changes relative to `old_sha`: each must parse and carry no conflict
-    markers. Reads committed blobs, never the working tree. Returns
-    (ok, detail); detail names the first failure, or "ok"."""
+def syntax_gate(repo: Path, old_sha: str, new_sha: str) -> tuple[str, str]:
+    """Check the files under SYNTAX_GATE_PATHS that `new_sha` adds or changes
+    relative to `old_sha`: none may carry conflict markers, and Python files
+    must parse. Reads committed blobs, never the working tree.
+
+    Returns (verdict, detail). verdict is "ok", "broken" (a file has conflict
+    markers or won't parse), or "unchecked" (git couldn't list or read the
+    files). detail names the first problem, or "ok"."""
     rc, names, err = _git(repo, "diff", "--name-only", "--diff-filter=ACMR", "-z",
-                          old_sha, new_sha, "--", *SYNTAX_GATE_DIRS)
+                          old_sha, new_sha, "--", *SYNTAX_GATE_PATHS)
     if rc != 0:
-        return False, f"could not list incoming changes: {err}"[:500]
-    for name in names.split("\0"):
-        if not name.endswith(".py"):
-            continue
+        return "unchecked", f"could not list incoming changes: {err}"[:500]
+    for name in filter(None, names.split("\0")):
         source = _blob(repo, new_sha, name)
         if source is None:
-            return False, f"could not read {name} at {new_sha[:12]}"
+            return "unchecked", f"could not read {name} at {new_sha[:12]}"
         marker = _conflict_marker_line(source.decode("utf-8", errors="replace"))
         if marker:
-            return False, f"conflict marker at {name}:{marker}"
+            return "broken", f"conflict marker at {name}:{marker}"
+        if not name.endswith(".py"):
+            continue
         try:
             compile(source, name, "exec", dont_inherit=True)
         except (SyntaxError, ValueError) as exc:
-            return False, f"{name}: {exc}"[:500]
-    return True, "ok"
+            return "broken", f"{name}: {exc}"[:500]
+    return "ok", "ok"
 
 
 def maybe_update(
@@ -323,18 +333,33 @@ def _log(msg: str) -> None:
         _log("already up to date")
         return result
 
+    if status["dirty"] and result["dirty_age_sec"] < dirty_stash_after_sec:
+        age = result["dirty_age_sec"]
+        result["skipped_reason"] = "dirty"
+        _log(f"working tree dirty for {age / 3600.0:.1f}h "
+             f"(< {dirty_stash_after_sec / 3600.0:.0f}h) — refusing to pull "
+             "(surfacing instead)")
+        return result
+
     # Gate the fetched commit before stashing or pulling anything, so broken
-    # code never reaches the working tree. A commit already refused is skipped
-    # quietly (its failure was recorded once) until the remote moves.
+    # code never reaches the working tree. A commit this interpreter already
+    # refused is skipped quietly until the remote moves or a day passes.
     old_head, new_sha = status["head"], status["remote_sha"]
     result["to_sha"] = new_sha[:12]
-    if marker.get("rejected_sha") == new_sha:
+    rejected = f"{new_sha} python{sys.version_info[0]}.{sys.version_info[1]}"
+    if (marker.get("rejected") == rejected
+            and now - marker.get("rejected_ts", 0) < REJECT_REMIND_SEC):
         result["skipped_reason"] = "syntax-fail-known"
         _log(f"{remote}/{branch} is still at refused {new_sha[:12]}; waiting for a fix")
         return result
-    ok, detail = syntax_gate(repo, old_head, new_sha)
-    if not ok:
-        marker["rejected_sha"] = new_sha
+    verdict, detail = syntax_gate(repo, old_head, new_sha)
+    if verdict == "unchecked":
+        result["skipped_reason"] = "gate-error"
+        result["error"] = detail
+        _log(f"could not check {new_sha[:12]}; not updating this time: {detail[:200]}")
+        return result
+    if verdict == "broken":
+        marker["rejected"], marker["rejected_ts"] = rejected, now
         _write_marker(marker_path, marker)
         result["skipped_reason"] = "syntax-fail"
         result["syntax_error"] = detail
@@ -343,12 +368,6 @@ def _log(msg: str) -> None:
 
     if status["dirty"]:
         age = result["dirty_age_sec"]
-        if age < dirty_stash_after_sec:
-            result["skipped_reason"] = "dirty"
-            _log(f"working tree dirty for {age / 3600.0:.1f}h "
-                 f"(< {dirty_stash_after_sec / 3600.0:.0f}h) — refusing to pull "
-                 "(surfacing instead)")
-            return result
         # Dirty past the window AND an update is waiting → stash, then pull.
         # The stash is recoverable (`git stash list` / `git stash pop`); it is
         # never dropped.
diff --git a/src/assistant/subsystems/pulse.py b/src/assistant/subsystems/pulse.py
index 7a8c9df..8e81252 100644
--- a/src/assistant/subsystems/pulse.py
+++ b/src/assistant/subsystems/pulse.py
@@ -17,8 +17,9 @@
   - The pulse spawns its own Observer subprocesses, writes its own heartbeat,
     and is the most safety-critical component. A subprocess gives us complete
     isolation and byte-for-byte compatibility with the system that runs today:
-    the daemon runs EXACTLY `python3 bin/pulse.py`, the same command the
-    com.assistant.assistant-pulse LaunchAgent runs.
+    the daemon runs EXACTLY `python3 bin/pulse.py`, the command the
+    com.assistant.assistant-pulse LaunchAgent execs after its bin/run-pulse.py
+    pre-flight.
 
 So this subsystem is a clean supervisor loop: run one pulse, sleep
 `pulse_interval_sec`, repeat — interruptible on shutdown. Bedrock env is merged
diff --git a/tests/test_pulse_topup.py b/tests/test_pulse_topup.py
index 84e4d0f..c3a5eea 100644
--- a/tests/test_pulse_topup.py
+++ b/tests/test_pulse_topup.py
@@ -243,6 +243,19 @@ def test_self_update_already_refused_commit_is_silent(mod, home):
     assert _read_ledger(home) == []
 
 
+def test_self_update_gate_error_is_recorded(mod, home):
+    _inject_self_update(mod, {"changed": False, "skipped_reason": "gate-error",
+                              "error": "could not read bin/pulse.py at bbbb"})
+    try:
+        mod.self_update_pulse(15)
+    finally:
+        sys.modules.pop("self_update", None)
+    e = _read_ledger(home)[0]
+    assert e["outcome"] == "failed"
+    assert e["kind"] == "self-update"
+    assert e["evidence"] == "self-update gate-error: could not read bin/pulse.py at bbbb"
+
+
 def test_self_update_other_reason_failed(mod, home):
     _inject_self_update(mod, {
         "changed": False, "skipped_reason": "pull-failed",
diff --git a/tests/test_run_pulse_wrapper.py b/tests/test_run_pulse_wrapper.py
index 5b2085b..a8e5b90 100644
--- a/tests/test_run_pulse_wrapper.py
+++ b/tests/test_run_pulse_wrapper.py
@@ -28,14 +28,10 @@ def _load():
     return mod
 
 
-def _layout(tmp: Path, pulse_body: str, src_files: dict[str, str] | None = None) -> Path:
+def _layout(tmp: Path, pulse_body: str) -> Path:
     (tmp / "bin").mkdir()
     shutil.copy2(WRAPPER, tmp / "bin/run-pulse.py")
     (tmp / "bin/pulse.py").write_text(pulse_body)
-    for rel, body in (src_files or {}).items():
-        path = tmp / "src" / rel
-        path.parent.mkdir(parents=True, exist_ok=True)
-        path.write_text(body)
     return tmp / "bin/run-pulse.py"
 
 
@@ -51,8 +47,7 @@ def test_broken_pulse_skips_the_run_and_logs(tmp_path, capsys):
     _layout(tmp_path, "def broken(:\n    pass\n")
     mod = _load()
     calls = []
-    with mock.patch.object(mod, "PULSE", tmp_path / "bin/pulse.py"), \
-            mock.patch.object(mod, "SRC", tmp_path / "src"):
+    with mock.patch.object(mod, "PULSE", tmp_path / "bin/pulse.py"):
         assert mod.main([], execv=lambda *a: calls.append(a)) == 0
     assert calls == []
     err = capsys.readouterr().err
@@ -60,16 +55,16 @@ def test_broken_pulse_skips_the_run_and_logs(tmp_path, capsys):
     assert f"{tmp_path / 'bin/pulse.py'}: SyntaxError" in err
 
 
-def test_broken_src_module_skips_the_run(tmp_path, capsys):
-    _layout(tmp_path, "x = 1\n", {"assistant/__init__.py": "",
-                                  "assistant/model_tiers.py": "x = 1\n<<<<<<< HEAD\n"})
-    mod = _load()
-    calls = []
-    with mock.patch.object(mod, "PULSE", tmp_path / "bin/pulse.py"), \
-            mock.patch.object(mod, "SRC", tmp_path / "src"):
-        assert mod.main([], execv=lambda *a: calls.append(a)) == 0
-    assert calls == []
-    assert "assistant/model_tiers.py: SyntaxError" in capsys.readouterr().err
+def test_broken_src_module_does_not_block_the_pulse(tmp_path):
+    # The pulse guards its optional imports itself; an unrelated broken module
+    # must not stop every run.
+    wrapper = _layout(tmp_path, 'print("RAN")\n')
+    (tmp_path / "src/assistant").mkdir(parents=True)
+    (tmp_path / "src/assistant/narrator.py").write_text("def broken(:\n")
+    r = subprocess.run([sys.executable, str(wrapper)], capture_output=True, text=True,
+                       timeout=60)
+    assert r.returncode == 0
+    assert r.stdout.strip() == "RAN"
 
 
 def test_script_entry_point_execs_pulse():
@@ -84,8 +79,7 @@ def test_script_entry_point_execs_pulse():
 
 
 def test_real_exec_runs_pulse_and_passes_args(tmp_path):
-    wrapper = _layout(tmp_path, 'import sys\nprint("RAN", " ".join(sys.argv[1:]))\n',
-                      {"assistant/__init__.py": ""})
+    wrapper = _layout(tmp_path, 'import sys\nprint("RAN", " ".join(sys.argv[1:]))\n')
     r = subprocess.run([sys.executable, str(wrapper), "--pulse-idx", "5"],
                        capture_output=True, text=True, timeout=60)
     assert r.returncode == 0
diff --git a/tests/test_self_update.py b/tests/test_self_update.py
index 4c59668..c608af9 100644
--- a/tests/test_self_update.py
+++ b/tests/test_self_update.py
@@ -12,6 +12,7 @@
 import json
 import os
 import subprocess
+import sys
 import unittest
 import unittest.mock
 from pathlib import Path
@@ -491,7 +492,7 @@ def _bad_remote(self, tmp: Path, files: dict[str, str]) -> tuple[Path, str, str]
         new_sha = advance_remote(tmp, remote, files, "incoming")
         return clone, old_head, new_sha
 
-    def _gate(self, files: dict[str, str]) -> tuple[bool, str]:
+    def _gate(self, files: dict[str, str]) -> tuple[str, str]:
         with TemporaryDirectory() as t:
             clone, old_head, new_sha = self._bad_remote(Path(t), files)
             git(clone, "fetch", "origin", "main")
@@ -511,18 +512,25 @@ def test_conflict_markers_are_refused_before_touching_the_tree(self):
             self.assertEqual(git(clone, "rev-parse", "HEAD"), old_head)
             self.assertEqual(git(clone, "reflog"), reflog)  # no merge, no reset
             self.assertEqual((clone / "bin/pulse.py").read_text(), "# pulse\n")
-            self.assertEqual(json.loads((tmp / "m.json").read_text())["rejected_sha"], new_sha)
+            self.assertEqual(json.loads((tmp / "m.json").read_text())["rejected"],
+                             f"{new_sha} python{sys.version_info[0]}.{sys.version_info[1]}")
+
+    def test_broken_runtime_python_outside_bin_is_refused(self):
+        for rel in ("src/assistant/model_tiers.py", "hooks/cmux-session-ledger.py",
+                    "install/patch-settings.py"):
+            with self.subTest(rel=rel), TemporaryDirectory() as t:
+                tmp = Path(t)
+                clone, old_head, _ = self._bad_remote(tmp, {rel: "def broken(:\n"})
+                r = su.maybe_update(clone, interval_sec=0, marker_path=tmp / "m.json")
+                self.assertEqual(r["skipped_reason"], "syntax-fail")
+                self.assertTrue(r["syntax_error"].startswith(f"{rel}: "))
+                self.assertEqual(git(clone, "rev-parse", "HEAD"), old_head)
+                self.assertFalse((clone / rel).exists())
 
-    def test_broken_src_module_is_refused(self):
-        with TemporaryDirectory() as t:
-            tmp = Path(t)
-            clone, old_head, _ = self._bad_remote(
-                tmp, {"src/assistant/model_tiers.py": "def broken(:\n"})
-            r = su.maybe_update(clone, interval_sec=0, marker_path=tmp / "m.json")
-            self.assertEqual(r["skipped_reason"], "syntax-fail")
-            self.assertTrue(r["syntax_error"].startswith("src/assistant/model_tiers.py: "))
-            self.assertEqual(git(clone, "rev-parse", "HEAD"), old_head)
-            self.assertFalse((clone / "src").exists())
+    def test_conflict_markers_in_runtime_non_python_files_are_refused(self):
+        verdict, detail = self._gate({"install.sh": "<<<<<<< HEAD\necho a\n=======\necho b\n"})
+        self.assertEqual(verdict, "broken")
+        self.assertEqual(detail, "conflict marker at install.sh:1")
 
     def test_refused_commit_is_skipped_until_the_remote_moves(self):
         with TemporaryDirectory() as t:
@@ -536,12 +544,49 @@ def test_refused_commit_is_skipped_until_the_remote_moves(self):
                 r = su.maybe_update(clone, interval_sec=0, marker_path=marker)
             gate.assert_not_called()
             self.assertEqual(r["skipped_reason"], "syntax-fail-known")
+            self.assertEqual(r["to_sha"], git(clone, "rev-parse", "origin/main")[:12])
             fixed = advance_remote(tmp, remote, {"bin/pulse.py": "fixed = True\n"}, "fix")
             r = su.maybe_update(clone, interval_sec=0, marker_path=marker)
             self.assertTrue(r["changed"])
             self.assertEqual(git(clone, "rev-parse", "HEAD"), fixed)
             self.assertEqual((clone / "bin/pulse.py").read_text(), "fixed = True\n")
 
+    def test_refused_commit_is_recorded_again_after_a_day(self):
+        with TemporaryDirectory() as t:
+            tmp = Path(t)
+            clone, _, _ = self._bad_remote(tmp, {"bin/pulse.py": "def broken(:\n"})
+            marker = tmp / "m.json"
+            runs = [su.maybe_update(clone, interval_sec=0, marker_path=marker, now=when)
+                    ["skipped_reason"] for when in (1000.0, 1000.0 + 3600, 1000.0 + 86400)]
+            self.assertEqual(runs, ["syntax-fail", "syntax-fail-known", "syntax-fail"])
+            self.assertEqual(json.loads(marker.read_text())["rejected_ts"], 1000.0 + 86400)
+
+    def test_refusal_by_another_python_is_checked_again(self):
+        with TemporaryDirectory() as t:
+            tmp = Path(t)
+            clone, _, new_sha = self._bad_remote(tmp, {"bin/pulse.py": "def broken(:\n"})
+            marker = tmp / "m.json"
+            su.maybe_update(clone, interval_sec=0, marker_path=marker, now=1000.0)
+            saved = json.loads(marker.read_text())
+            marker.write_text(json.dumps({**saved, "rejected": f"{new_sha} python3.0"}))
+            r = su.maybe_update(clone, interval_sec=0, marker_path=marker, now=1001.0)
+            self.assertEqual(r["skipped_reason"], "syntax-fail")
+
+    def test_git_trouble_during_the_check_is_not_remembered(self):
+        with TemporaryDirectory() as t:
+            tmp = Path(t)
+            clone, _, new_sha = self._bad_remote(tmp, {"bin/pulse.py": "ok = 1\n"})
+            marker = tmp / "m.json"
+            with unittest.mock.patch.object(su, "syntax_gate",
+                                            return_value=("unchecked", "git timed out")):
+                r = su.maybe_update(clone, interval_sec=0, marker_path=marker)
+            self.assertEqual(r["skipped_reason"], "gate-error")
+            self.assertEqual(r["error"], "git timed out")
+            self.assertNotIn("rejected", json.loads(marker.read_text()))
+            r = su.maybe_update(clone, interval_sec=0, marker_path=marker)
+            self.assertTrue(r["changed"])
+            self.assertEqual(git(clone, "rev-parse", "HEAD"), new_sha)
+
     def test_dirty_tree_past_window_is_not_stashed_for_a_refused_update(self):
         with TemporaryDirectory() as t:
             tmp = Path(t)
@@ -552,8 +597,8 @@ def test_dirty_tree_past_window_is_not_stashed_for_a_refused_update(self):
                                     dirty_stash_after_sec=86400, now=1000.0)
             later = su.maybe_update(clone, interval_sec=0, marker_path=marker,
                                     dirty_stash_after_sec=86400, now=1000.0 + 25 * 3600)
-            self.assertEqual(first["skipped_reason"], "syntax-fail")
-            self.assertEqual(later["skipped_reason"], "syntax-fail-known")
+            self.assertEqual(first["skipped_reason"], "dirty")
+            self.assertEqual(later["skipped_reason"], "syntax-fail")
             self.assertNotIn("stashed", later)
             self.assertEqual(git(clone, "stash", "list"), "")
             self.assertEqual((clone / "install.sh").read_text(), "# operator edit\n")
@@ -565,6 +610,8 @@ def test_healthy_update_lands_on_the_gated_commit(self):
                 "bin/pulse.py": '"""Pulse.\n\nSection\n=======\n"""\nok = True\n',
                 "tests/test_scratch.py": "def broken(:\n",
                 "docs/merging.md": "<<<<<<< HEAD\n=======\n>>>>>>> x\n",
+                "prompts/observer.md": "Heading\n=======\n",
+                "skills/logo.bin": "\x00\x01binary",
             })
             r = su.maybe_update(clone, interval_sec=0, marker_path=tmp / "m.json")
             self.assertTrue(r["changed"])
@@ -572,8 +619,8 @@ def test_healthy_update_lands_on_the_gated_commit(self):
             self.assertEqual(git(clone, "rev-parse", "HEAD"), new_sha)
 
     def test_marker_inside_a_string_is_flagged_even_though_it_compiles(self):
-        ok, detail = self._gate({"bin/pulse.py": 'X = """\n>>>>>>> theirs\n"""\n'})
-        self.assertFalse(ok)
+        verdict, detail = self._gate({"bin/pulse.py": 'X = """\n>>>>>>> theirs\n"""\n'})
+        self.assertEqual(verdict, "broken")
         self.assertEqual(detail, "conflict marker at bin/pulse.py:2")
 
     def test_marker_line_ignores_rst_underline_alone(self):
@@ -583,8 +630,8 @@ def test_marker_line_ignores_rst_underline_alone(self):
     def test_unreadable_diff_is_refused(self):
         with TemporaryDirectory() as t:
             clone, _ = make_repos(Path(t))
-            ok, detail = su.syntax_gate(clone, "0" * 40, "HEAD")
-            self.assertFalse(ok)
+            verdict, detail = su.syntax_gate(clone, "0" * 40, "HEAD")
+            self.assertEqual(verdict, "unchecked")
             self.assertTrue(detail.startswith("could not list incoming changes: "))
 
     def test_unreadable_blob_is_refused(self):
@@ -599,8 +646,8 @@ def run(cmd, **kwargs):
                 return real_run(cmd, **kwargs)
 
             with unittest.mock.patch.object(su.subprocess, "run", run):
-                ok, detail = su.syntax_gate(clone, old_head, new_sha)
-            self.assertFalse(ok)
+                verdict, detail = su.syntax_gate(clone, old_head, new_sha)
+            self.assertEqual(verdict, "unchecked")
             self.assertEqual(detail, f"could not read bin/new.py at {new_sha[:12]}")
 
     def test_blob_missing_from_commit_reads_as_none(self):

From 06c1e71c20f8dbdd068a2896fb3fd2baf999b876 Mon Sep 17 00:00:00 2001
From: Mukul Sharma <elitecoder.mukul@gmail.com>
Date: Sun, 27 Sep 2026 16:05:21 -0700
Subject: [PATCH 7/8] Show pre-flight failures on the dashboard and check
 startup imports, symlinks, and docs

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 CHANGELOG.md                      |  5 +--
 bin/pulse.py                      |  9 +++---
 bin/render-assistant-page.py      | 22 ++++++++++++-
 bin/run-pulse.py                  | 44 ++++++++++++++++---------
 bin/self_update.py                | 22 +++++++------
 tests/test_pulse_topup.py         |  3 +-
 tests/test_renderer_in_process.py | 32 +++++++++++++++++++
 tests/test_run_pulse_wrapper.py   | 53 +++++++++++++++++++++++++------
 tests/test_self_update.py         | 19 ++++++++++-
 tests/test_self_update_topup.py   |  2 +-
 10 files changed, 168 insertions(+), 43 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 149bf31..c2f177d 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -17,8 +17,9 @@ The version is carried in `pyproject.toml` and `src/assistant/__init__.py`
   dashboard. Skip that commit quietly until the remote moves, with a reminder
   once a day. Updates now fast-forward to exactly the commit that was checked.
 - Run the pulse through a pre-flight (`bin/run-pulse.py`) that parses `pulse.py`
-  before each start. If it won't parse, the pre-flight logs one line to the
-  pulse's launchd error log and exits cleanly instead of crashing. New installs
+  and the modules it loads at startup before each start. If one won't parse,
+  the pre-flight skips the run and exits cleanly instead of crashing, and the
+  dashboard's pulse banner turns red and shows the error. New installs
   get this right away. Existing machines pick it up after a reboot, a logout, or
   a manual reload of the pulse LaunchAgent, since self-update defers that reload.
 - Enforce 100% changed-code coverage with separate Python and real-browser reports.
diff --git a/bin/pulse.py b/bin/pulse.py
index bd32eca..2860de8 100755
--- a/bin/pulse.py
+++ b/bin/pulse.py
@@ -458,7 +458,7 @@ def self_update_pulse(pulse_idx: int) -> None:
     reason = result.get("skipped_reason")
     changed = result.get("changed")
     # Silent path: attempted, nothing to do, no problem — or a refused commit
-    # whose failure was already recorded when it was first refused.
+    # whose failure was recorded in the last day.
     if not changed and reason in (None, "syntax-fail-known") and not result.get("error"):
         return
 
@@ -498,12 +498,13 @@ def self_update_pulse(pulse_idx: int) -> None:
         evidence = f"self-update auto-stash failed: {result.get('error', '')}"[:300]
         key = f"self-update-stash-failed-p{pulse_idx}"
     elif reason == "syntax-fail":
-        # The fetched commits carry Python that won't parse; self_update refused
-        # them before touching the working tree.
+        # A fetched file has conflict markers or Python that won't parse;
+        # self_update refused the commits before touching the working tree.
         outcome = "failed"
         kind = "self-update-syntax-fail"
         evidence = (f"refused self-update {result.get('from_sha')}.."
-                    f"{result.get('to_sha')}: {result.get('syntax_error', '')}")[:300]
+                    f"{result.get('to_sha')} (pull by hand if this is wrong): "
+                    f"{result.get('syntax_error', '')}")[:300]
         key = f"self-update-syntax-fail-p{pulse_idx}"
     else:
         outcome = "failed"
diff --git a/bin/render-assistant-page.py b/bin/render-assistant-page.py
index 9b89114..cf97672 100755
--- a/bin/render-assistant-page.py
+++ b/bin/render-assistant-page.py
@@ -2254,6 +2254,18 @@ def _ws_num(c):
     return f'<div class="fleet-board">{"".join(col_html)}</div>', total
 
 
+def _preflight_failure_since(last_pulse_ts: int) -> str | None:
+    """The error bin/run-pulse.py recorded when it last refused to start the
+    pulse, if no pulse has run since. None when there's nothing to show."""
+    try:
+        record = json.loads((HOME / ".assistant/pulse-preflight.json").read_text())
+        if float(record["failed_at"]) > last_pulse_ts:
+            return str(record["error"])
+    except (OSError, ValueError, TypeError, KeyError):
+        pass
+    return None
+
+
 def render_pulse_health() -> str:
     """One-line banner showing whether the assistant-pulse cron is alive.
     Reads ~/.assistant/heartbeat.json and color-codes by age:
@@ -2291,6 +2303,14 @@ def render_pulse_health() -> str:
     else:
         cls = "pulse-bad"
         msg = "Pulse stale — orchestrator may be down"
+    # Without data-pulse-at, the page script leaves the failure text in place
+    # instead of replacing it with an age-based status.
+    pulse_at = f' data-pulse-at="{last_ts}"'
+    failure = _preflight_failure_since(last_ts)
+    if failure is not None:
+        cls = "pulse-bad"
+        msg = f"Pulse can't start: {e(failure[:200])}"
+        pulse_at = ""
     if age_sec < 60:
         age_str = f"{age_sec}s"
     elif age_sec < 3600:
@@ -2302,7 +2322,7 @@ def render_pulse_health() -> str:
     pulse_idx = hb.get("pulse_idx", "?")
     model = hb.get("model", "?")
     return (
-        f'<div class="pulse-health {cls}" data-pulse-at="{last_ts}">'
+        f'<div class="pulse-health {cls}"{pulse_at}>'
         f'<span class="pulse-dot"></span>'
         f'<span class="pulse-text">{msg}</span>'
         f'<span class="pulse-meta">last pulse {age_str} ago · #{pulse_idx} · {e(str(model))}</span>'
diff --git a/bin/run-pulse.py b/bin/run-pulse.py
index 0058ef7..f6b803a 100755
--- a/bin/run-pulse.py
+++ b/bin/run-pulse.py
@@ -1,32 +1,46 @@
 #!/usr/bin/env python3
 """Launchd pre-flight for the Assistant pulse.
 
-Parses bin/pulse.py, then replaces this process with it. If pulse.py won't
-parse, it prints one line to stderr (the LaunchAgent sends stderr to
-~/.assistant/logs/assistant-pulse.launchd.err) and exits 0 instead of starting a
-pulse that would crash on its first line. The LaunchAgent fires every
-StartInterval whatever the exit code, so the next tick retries. Other modules
-aren't checked here: the pulse guards its own optional imports, and
-self_update.py refuses incoming commits that don't parse.
+Parses the files the pulse loads at startup, then replaces this process with
+bin/pulse.py. If one won't parse, it records the error in
+~/.assistant/pulse-preflight.json (the dashboard's pulse banner shows it), prints
+it to stderr (the LaunchAgent's err log), and exits 0 instead of starting a
+pulse that would crash on import. The LaunchAgent fires every StartInterval
+whatever the exit code, so the next tick retries. Modules the pulse loads later
+aren't checked here: it guards those imports itself, and self_update.py refuses
+incoming commits that don't parse.
 """
 from __future__ import annotations
 
+import json
 import os
 import sys
+import time
 from datetime import datetime, timezone
 from pathlib import Path
 
-PULSE = Path(__file__).resolve().parent / "pulse.py"
+BIN = Path(__file__).resolve().parent
+PULSE = BIN / "pulse.py"
+# pulse.py plus its unguarded module-level imports. Keep in sync with pulse.py.
+STARTUP_FILES = (PULSE, BIN.parent / "src/assistant/__init__.py",
+                 BIN.parent / "src/assistant/model_tiers.py")
 
 
 def main(argv: list[str], *, execv=os.execv) -> int:
-    try:
-        compile(PULSE.read_bytes(), str(PULSE), "exec", dont_inherit=True)
-    except (SyntaxError, ValueError) as exc:
-        stamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
-        print(f"[{stamp}] pulse pre-flight FAILED, skipping this run: "
-              f"{PULSE}: {type(exc).__name__}: {exc}", file=sys.stderr)
-        return 0
+    for path in STARTUP_FILES:
+        try:
+            compile(path.read_bytes(), str(path), "exec", dont_inherit=True)
+        except (SyntaxError, ValueError) as exc:
+            error = f"{path}: {type(exc).__name__}: {exc}"
+            record = Path.home() / ".assistant/pulse-preflight.json"
+            record.parent.mkdir(parents=True, exist_ok=True)
+            tmp = record.with_suffix(".json.tmp")
+            tmp.write_text(json.dumps({"failed_at": time.time(), "error": error}))
+            tmp.replace(record)
+            stamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+            print(f"[{stamp}] pulse pre-flight FAILED, skipping this run: {error}",
+                  file=sys.stderr)
+            return 0
     execv(sys.executable, [sys.executable, str(PULSE), *argv])
     return 0
 
diff --git a/bin/self_update.py b/bin/self_update.py
index d3022ff..5e68346 100644
--- a/bin/self_update.py
+++ b/bin/self_update.py
@@ -70,12 +70,12 @@
 
 # ── Pre-pull syntax gate ─────────────────────────────────────────────────────
 # Paths the running system loads, runs, or installs from the checkout. Add new
-# runtime paths here. The July 2026 incident (commit 17f3862 pulled unresolved
-# conflict markers into bin/pulse.py and the pulse failed every tick for
-# months) is what the gate exists to stop.
+# runtime paths here. In July 2026, conflict markers left in the working copy of
+# bin/pulse.py made the pulse fail every tick for months. The gate keeps a
+# pulled commit from doing the same; bin/run-pulse.py covers the working copy.
 SYNTAX_GATE_PATHS = ("bin/", "src/", "hooks/", "install/", "prompts/", "skills/",
-                     "launchagents/", "config/", "slack-reactor/", "install.sh",
-                     "install-bootstrap.sh")
+                     "launchagents/", "config/", "docs/", "slack-reactor/",
+                     "install.sh", "install-bootstrap.sh")
 
 # A refused commit is re-checked, and its failure re-recorded, this often.
 REJECT_REMIND_SEC = 86400  # 1 day
@@ -232,11 +232,14 @@ def syntax_gate(repo: Path, old_sha: str, new_sha: str) -> tuple[str, str]:
     Returns (verdict, detail). verdict is "ok", "broken" (a file has conflict
     markers or won't parse), or "unchecked" (git couldn't list or read the
     files). detail names the first problem, or "ok"."""
-    rc, names, err = _git(repo, "diff", "--name-only", "--diff-filter=ACMR", "-z",
-                          old_sha, new_sha, "--", *SYNTAX_GATE_PATHS)
+    rc, raw, err = _git(repo, "diff", "--raw", "--no-renames", "-z", "--diff-filter=ACMT",
+                        old_sha, new_sha, "--", *SYNTAX_GATE_PATHS)
     if rc != 0:
         return "unchecked", f"could not list incoming changes: {err}"[:500]
-    for name in filter(None, names.split("\0")):
+    fields = raw.split("\0")
+    for meta, name in zip(fields[::2], fields[1::2]):
+        if not meta.split()[1].startswith("100"):
+            continue  # a symlink or submodule has no file content to check
         source = _blob(repo, new_sha, name)
         if source is None:
             return "unchecked", f"could not read {name} at {new_sha[:12]}"
@@ -363,7 +366,8 @@ def _log(msg: str) -> None:
         _write_marker(marker_path, marker)
         result["skipped_reason"] = "syntax-fail"
         result["syntax_error"] = detail
-        _log(f"refused {old_head[:12]}..{new_sha[:12]}, code won't parse: {detail[:200]}")
+        _log(f"refused {old_head[:12]}..{new_sha[:12]}, a file failed the check: "
+             f"{detail[:200]}")
         return result
 
     if status["dirty"]:
diff --git a/tests/test_pulse_topup.py b/tests/test_pulse_topup.py
index c3a5eea..ef9c8ca 100644
--- a/tests/test_pulse_topup.py
+++ b/tests/test_pulse_topup.py
@@ -230,7 +230,8 @@ def test_self_update_reason_syntax_fail(mod, home):
     assert e["outcome"] == "failed"
     assert e["kind"] == "self-update-syntax-fail"
     assert e["key"] == "self-update-syntax-fail-p13"
-    assert e["evidence"] == ("refused self-update aaaaaaaaaaaa..bbbbbbbbbbbb: "
+    assert e["evidence"] == ("refused self-update aaaaaaaaaaaa..bbbbbbbbbbbb "
+                             "(pull by hand if this is wrong): "
                              "conflict marker at bin/pulse.py:42")
 
 
diff --git a/tests/test_renderer_in_process.py b/tests/test_renderer_in_process.py
index ecc6512..c4b7380 100644
--- a/tests/test_renderer_in_process.py
+++ b/tests/test_renderer_in_process.py
@@ -104,6 +104,38 @@ def test_stale_pulse_renders_red_banner(self):
         self.assertIn("pulse-bad", html)
         self.assertIn("Pulse stale", html)
 
+    def _write_preflight(self, payload) -> None:
+        (self._tmp / ".assistant/pulse-preflight.json").write_text(
+            payload if isinstance(payload, str) else json.dumps(payload))
+
+    def test_preflight_failure_after_last_pulse_renders_red_banner(self):
+        now = int(time.time())
+        self._write_heartbeat({"last_pulse_ts": now - 30, "pulse_idx": 99, "model": "m"})
+        self._write_preflight({"failed_at": now - 10,
+                               "error": "bin/pulse.py: SyntaxError: <bad> (line 1)"})
+        html = self.mod.render_pulse_health()
+        self.assertIn("pulse-bad", html)
+        self.assertIn("Pulse can't start: bin/pulse.py: SyntaxError: &lt;bad&gt; (line 1)",
+                      html)
+        self.assertNotIn("data-pulse-at", html)
+
+    def test_preflight_failure_before_last_pulse_is_ignored(self):
+        now = int(time.time())
+        self._write_heartbeat({"last_pulse_ts": now - 30, "pulse_idx": 99, "model": "m"})
+        self._write_preflight({"failed_at": now - 60, "error": "old failure"})
+        html = self.mod.render_pulse_health()
+        self.assertIn("Pulse healthy", html)
+        self.assertIn(f'data-pulse-at="{now - 30}"', html)
+        self.assertNotIn("old failure", html)
+
+    def test_unreadable_preflight_record_is_ignored(self):
+        self._write_heartbeat({"last_pulse_ts": int(time.time()) - 30,
+                               "pulse_idx": 99, "model": "m"})
+        for payload in ("{ corrupt", {"error": "no timestamp"}):
+            with self.subTest(payload=payload):
+                self._write_preflight(payload)
+                self.assertIn("Pulse healthy", self.mod.render_pulse_health())
+
     def test_age_formatting_includes_unit(self):
         now = int(time.time())
         with mock.patch.object(self.mod, "utc_now",
diff --git a/tests/test_run_pulse_wrapper.py b/tests/test_run_pulse_wrapper.py
index a8e5b90..a058778 100644
--- a/tests/test_run_pulse_wrapper.py
+++ b/tests/test_run_pulse_wrapper.py
@@ -6,8 +6,11 @@
 """
 from __future__ import annotations
 
+import ast
 import importlib.util
+import json
 import os
+import plistlib
 import runpy
 import shutil
 import subprocess
@@ -28,10 +31,13 @@ def _load():
     return mod
 
 
-def _layout(tmp: Path, pulse_body: str) -> Path:
+def _layout(tmp: Path, pulse_body: str, model_tiers: str = "TIERS = {}\n") -> Path:
     (tmp / "bin").mkdir()
     shutil.copy2(WRAPPER, tmp / "bin/run-pulse.py")
     (tmp / "bin/pulse.py").write_text(pulse_body)
+    (tmp / "src/assistant").mkdir(parents=True)
+    (tmp / "src/assistant/__init__.py").write_text("")
+    (tmp / "src/assistant/model_tiers.py").write_text(model_tiers)
     return tmp / "bin/run-pulse.py"
 
 
@@ -43,30 +49,51 @@ def test_healthy_checkout_execs_pulse_with_same_interpreter_and_args():
                                        "--pulse-idx", "5"])]
 
 
-def test_broken_pulse_skips_the_run_and_logs(tmp_path, capsys):
+def test_broken_pulse_skips_the_run_and_records_why(tmp_path, capsys, monkeypatch):
+    monkeypatch.setenv("HOME", str(tmp_path / "home"))
     _layout(tmp_path, "def broken(:\n    pass\n")
     mod = _load()
     calls = []
-    with mock.patch.object(mod, "PULSE", tmp_path / "bin/pulse.py"):
+    with mock.patch.object(mod, "STARTUP_FILES", (tmp_path / "bin/pulse.py",)):
         assert mod.main([], execv=lambda *a: calls.append(a)) == 0
     assert calls == []
     err = capsys.readouterr().err
     assert "pulse pre-flight FAILED, skipping this run" in err
     assert f"{tmp_path / 'bin/pulse.py'}: SyntaxError" in err
+    record = json.loads((tmp_path / "home/.assistant/pulse-preflight.json").read_text())
+    assert record["error"].startswith(f"{tmp_path / 'bin/pulse.py'}: SyntaxError")
+    assert isinstance(record["failed_at"], float)
 
 
-def test_broken_src_module_does_not_block_the_pulse(tmp_path):
-    # The pulse guards its optional imports itself; an unrelated broken module
+def test_broken_startup_import_skips_the_run(tmp_path):
+    wrapper = _layout(tmp_path, 'print("RAN")\n', model_tiers="x = 1\n<<<<<<< HEAD\n")
+    r = subprocess.run([sys.executable, str(wrapper)], capture_output=True, text=True,
+                       timeout=60, env=dict(os.environ, HOME=str(tmp_path / "home")))
+    assert r.returncode == 0
+    assert r.stdout == ""
+    assert "src/assistant/model_tiers.py: SyntaxError" in r.stderr
+
+
+def test_broken_later_module_does_not_block_the_pulse(tmp_path):
+    # The pulse guards the modules it loads later; an unrelated broken module
     # must not stop every run.
     wrapper = _layout(tmp_path, 'print("RAN")\n')
-    (tmp_path / "src/assistant").mkdir(parents=True)
     (tmp_path / "src/assistant/narrator.py").write_text("def broken(:\n")
     r = subprocess.run([sys.executable, str(wrapper)], capture_output=True, text=True,
-                       timeout=60)
+                       timeout=60, env=dict(os.environ, HOME=str(tmp_path / "home")))
     assert r.returncode == 0
     assert r.stdout.strip() == "RAN"
 
 
+def test_startup_files_match_pulse_module_level_imports():
+    tree = ast.parse((REPO / "bin/pulse.py").read_text())
+    imported = {f"src/assistant/{alias.name}.py" for node in tree.body
+                if isinstance(node, ast.ImportFrom) and node.module == "assistant"
+                for alias in node.names}
+    listed = {str(path.relative_to(REPO)) for path in _load().STARTUP_FILES}
+    assert imported | {"bin/pulse.py", "src/assistant/__init__.py"} == listed
+
+
 def test_script_entry_point_execs_pulse():
     calls = []
     with mock.patch.object(os, "execv", lambda *a: calls.append(a)), \
@@ -81,7 +108,8 @@ def test_script_entry_point_execs_pulse():
 def test_real_exec_runs_pulse_and_passes_args(tmp_path):
     wrapper = _layout(tmp_path, 'import sys\nprint("RAN", " ".join(sys.argv[1:]))\n')
     r = subprocess.run([sys.executable, str(wrapper), "--pulse-idx", "5"],
-                       capture_output=True, text=True, timeout=60)
+                       capture_output=True, text=True, timeout=60,
+                       env=dict(os.environ, HOME=str(tmp_path / "home")))
     assert r.returncode == 0
     assert r.stdout.strip() == "RAN --pulse-idx 5"
     assert r.stderr == ""
@@ -90,7 +118,14 @@ def test_real_exec_runs_pulse_and_passes_args(tmp_path):
 def test_real_run_with_broken_pulse_exits_zero(tmp_path):
     wrapper = _layout(tmp_path, "def broken(:\n")
     r = subprocess.run([sys.executable, str(wrapper)],
-                       capture_output=True, text=True, timeout=60)
+                       capture_output=True, text=True, timeout=60,
+                       env=dict(os.environ, HOME=str(tmp_path / "home")))
     assert r.returncode == 0
     assert r.stdout == ""
     assert "pulse pre-flight FAILED" in r.stderr
+
+
+def test_pulse_launch_agent_runs_the_preflight():
+    plist = plistlib.loads((REPO / "launchagents/com.assistant.assistant-pulse.plist")
+                           .read_bytes())
+    assert plist["ProgramArguments"] == ["__PYTHON__", "__REPO__/bin/run-pulse.py"]
diff --git a/tests/test_self_update.py b/tests/test_self_update.py
index c608af9..110943e 100644
--- a/tests/test_self_update.py
+++ b/tests/test_self_update.py
@@ -609,7 +609,7 @@ def test_healthy_update_lands_on_the_gated_commit(self):
             clone, _, new_sha = self._bad_remote(tmp, {
                 "bin/pulse.py": '"""Pulse.\n\nSection\n=======\n"""\nok = True\n',
                 "tests/test_scratch.py": "def broken(:\n",
-                "docs/merging.md": "<<<<<<< HEAD\n=======\n>>>>>>> x\n",
+                "README.md": "<<<<<<< HEAD\n=======\n>>>>>>> x\n",
                 "prompts/observer.md": "Heading\n=======\n",
                 "skills/logo.bin": "\x00\x01binary",
             })
@@ -618,6 +618,23 @@ def test_healthy_update_lands_on_the_gated_commit(self):
             self.assertIsNone(r["skipped_reason"])
             self.assertEqual(git(clone, "rev-parse", "HEAD"), new_sha)
 
+    def test_symlinked_python_file_is_not_compiled_as_its_link_text(self):
+        with TemporaryDirectory() as t:
+            tmp = Path(t)
+            clone, remote = make_repos(tmp)
+            old_head = git(clone, "rev-parse", "HEAD")
+            scratch = tmp / "scratch-link"
+            git(tmp, "clone", str(remote), str(scratch))
+            git(scratch, "config", "user.email", "t@t")
+            git(scratch, "config", "user.name", "t")
+            (scratch / "bin/alias.py").symlink_to("../bin/pulse.py")
+            git(scratch, "add", "-A")
+            git(scratch, "commit", "-m", "add symlink")
+            git(scratch, "push", "origin", "main")
+            git(clone, "fetch", "origin", "main")
+            verdict, detail = su.syntax_gate(clone, old_head, git(scratch, "rev-parse", "HEAD"))
+            self.assertEqual((verdict, detail), ("ok", "ok"))
+
     def test_marker_inside_a_string_is_flagged_even_though_it_compiles(self):
         verdict, detail = self._gate({"bin/pulse.py": 'X = """\n>>>>>>> theirs\n"""\n'})
         self.assertEqual(verdict, "broken")
diff --git a/tests/test_self_update_topup.py b/tests/test_self_update_topup.py
index ec6663a..ee08e6c 100644
--- a/tests/test_self_update_topup.py
+++ b/tests/test_self_update_topup.py
@@ -91,7 +91,7 @@ def test_maybe_update_stash_failed(tmp_path):
                 "dirty": True, "behind": 1, "ahead": 0}):
             with mock.patch.object(su, "_stash_dirty",
                                    return_value=(False, "fatal: stash conflict")), \
-                    mock.patch.object(su, "syntax_gate", return_value=(True, "ok")):
+                    mock.patch.object(su, "syntax_gate", return_value=("ok", "ok")):
                 # First pass stamps dirty_since at t=1000.
                 su.maybe_update(tmp_path, interval_sec=0, marker_path=marker,
                                 dirty_stash_after_sec=86400, now=1000.0)

From d9333f4c37ba42bd3be6a371bec34552e6e179ea Mon Sep 17 00:00:00 2001
From: Mukul Sharma <elitecoder.mukul@gmail.com>
Date: Sun, 27 Sep 2026 17:14:26 -0700
Subject: [PATCH 8/8] Alert pre-flight failures at the top of the dashboard and
 in the ledger

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 CHANGELOG.md                      |  5 +-
 bin/render-assistant-page.py      | 35 +++++++------
 bin/run-pulse.py                  | 62 ++++++++++++++++------
 tests/test_renderer_in_process.py | 28 +++++-----
 tests/test_run_pulse_wrapper.py   | 87 ++++++++++++++++++++++++++++---
 5 files changed, 164 insertions(+), 53 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index c2f177d..14b91b9 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -18,8 +18,9 @@ The version is carried in `pyproject.toml` and `src/assistant/__init__.py`
   once a day. Updates now fast-forward to exactly the commit that was checked.
 - Run the pulse through a pre-flight (`bin/run-pulse.py`) that parses `pulse.py`
   and the modules it loads at startup before each start. If one won't parse,
-  the pre-flight skips the run and exits cleanly instead of crashing, and the
-  dashboard's pulse banner turns red and shows the error. New installs
+  the pre-flight skips the run and exits cleanly instead of crashing. The
+  dashboard shows the error at the top of the page until a pulse runs again, and
+  an actions-ledger entry reaches Slack for a new error, then once a day. New installs
   get this right away. Existing machines pick it up after a reboot, a logout, or
   a manual reload of the pulse LaunchAgent, since self-update defers that reload.
 - Enforce 100% changed-code coverage with separate Python and real-browser reports.
diff --git a/bin/render-assistant-page.py b/bin/render-assistant-page.py
index cf97672..e154dac 100755
--- a/bin/render-assistant-page.py
+++ b/bin/render-assistant-page.py
@@ -2254,16 +2254,25 @@ def _ws_num(c):
     return f'<div class="fleet-board">{"".join(col_html)}</div>', total
 
 
-def _preflight_failure_since(last_pulse_ts: int) -> str | None:
-    """The error bin/run-pulse.py recorded when it last refused to start the
-    pulse, if no pulse has run since. None when there's nothing to show."""
+def render_pulse_alert() -> str:
+    """A top-of-page alert when bin/run-pulse.py refused to start the pulse and
+    no pulse has run since. Empty string when there's nothing to show."""
     try:
         record = json.loads((HOME / ".assistant/pulse-preflight.json").read_text())
-        if float(record["failed_at"]) > last_pulse_ts:
-            return str(record["error"])
+        failed_at, error = float(record["failed_at"]), str(record["error"])
     except (OSError, ValueError, TypeError, KeyError):
-        pass
-    return None
+        return ""
+    try:
+        last_ts = float(json.loads((HOME / ".assistant/heartbeat.json").read_text())
+                        .get("last_pulse_ts") or 0)
+    except (OSError, ValueError, TypeError, AttributeError):
+        last_ts = 0
+    if failed_at <= last_ts:
+        return ""
+    return ('<div class="pulse-health pulse-bad" role="alert">'
+            '<span class="pulse-dot"></span>'
+            f'<span class="pulse-text">Pulse can\'t start: {e(error[:200])}</span>'
+            '</div>')
 
 
 def render_pulse_health() -> str:
@@ -2303,14 +2312,6 @@ def render_pulse_health() -> str:
     else:
         cls = "pulse-bad"
         msg = "Pulse stale — orchestrator may be down"
-    # Without data-pulse-at, the page script leaves the failure text in place
-    # instead of replacing it with an age-based status.
-    pulse_at = f' data-pulse-at="{last_ts}"'
-    failure = _preflight_failure_since(last_ts)
-    if failure is not None:
-        cls = "pulse-bad"
-        msg = f"Pulse can't start: {e(failure[:200])}"
-        pulse_at = ""
     if age_sec < 60:
         age_str = f"{age_sec}s"
     elif age_sec < 3600:
@@ -2322,7 +2323,7 @@ def render_pulse_health() -> str:
     pulse_idx = hb.get("pulse_idx", "?")
     model = hb.get("model", "?")
     return (
-        f'<div class="pulse-health {cls}"{pulse_at}>'
+        f'<div class="pulse-health {cls}" data-pulse-at="{last_ts}">'
         f'<span class="pulse-dot"></span>'
         f'<span class="pulse-text">{msg}</span>'
         f'<span class="pulse-meta">last pulse {age_str} ago · #{pulse_idx} · {e(str(model))}</span>'
@@ -2471,6 +2472,7 @@ def render():
     brief_html, brief_n = render_brief_tab()
     connections_html, connected_n = render_connections_panel(world)
     pulse_health_html = render_pulse_health()
+    pulse_alert_html = render_pulse_alert()
     counts = world.get("counts", {})
     snapshot_at = _overview_timestamp(world.get("_meta", {}).get("built_at"))
     rendered_at = utc_now().timestamp()
@@ -4096,6 +4098,7 @@ def render():
   <button class="btn" id="refresh-dashboard" onclick="refreshDashboard(true)">Reload page</button>
   <span class="meta" id="refresh-note">Checks for updates every 15 seconds.</span>
 </div>
+{pulse_alert_html}
 <details class="service-details" data-context-key="service-health"><summary>Background services and saved data</summary>
 {pulse_health_html}
 <p>The page checks for updates every 15 seconds, except while you're reading expanded details.</p>
diff --git a/bin/run-pulse.py b/bin/run-pulse.py
index f6b803a..ff05248 100755
--- a/bin/run-pulse.py
+++ b/bin/run-pulse.py
@@ -1,14 +1,20 @@
 #!/usr/bin/env python3
 """Launchd pre-flight for the Assistant pulse.
 
-Parses the files the pulse loads at startup, then replaces this process with
-bin/pulse.py. If one won't parse, it records the error in
-~/.assistant/pulse-preflight.json (the dashboard's pulse banner shows it), prints
-it to stderr (the LaunchAgent's err log), and exits 0 instead of starting a
-pulse that would crash on import. The LaunchAgent fires every StartInterval
-whatever the exit code, so the next tick retries. Modules the pulse loads later
-aren't checked here: it guards those imports itself, and self_update.py refuses
-incoming commits that don't parse.
+Parses the files the pulse loads at startup, clears any earlier failure record,
+then replaces this process with bin/pulse.py. If one won't parse, it skips the
+run and exits 0 instead of starting a pulse that would crash on import:
+
+  - ~/.assistant/pulse-preflight.json holds the error; the dashboard shows it
+    at the top of the page until a pulse runs again.
+  - An actions-ledger entry (so Slack hears about it) is written for a new
+    error, then once a day while it lasts.
+  - stderr (the LaunchAgent's err log) gets one line per skipped run.
+
+The LaunchAgent fires every StartInterval whatever the exit code, so the next
+tick retries. Modules the pulse loads later aren't checked here: it guards
+those imports itself, and self_update.py refuses incoming commits that don't
+parse.
 """
 from __future__ import annotations
 
@@ -24,23 +30,49 @@
 # pulse.py plus its unguarded module-level imports. Keep in sync with pulse.py.
 STARTUP_FILES = (PULSE, BIN.parent / "src/assistant/__init__.py",
                  BIN.parent / "src/assistant/model_tiers.py")
+LEDGER_EVERY_SEC = 86400  # 1 day
+
+
+def _record_failure(assistant_dir: Path, error: str, now: float) -> None:
+    record = assistant_dir / "pulse-preflight.json"
+    try:
+        previous = json.loads(record.read_text())
+    except (OSError, ValueError):
+        previous = {}
+    ledgered_at = previous.get("ledgered_at") if previous.get("error") == error else None
+    try:
+        assistant_dir.mkdir(parents=True, exist_ok=True)
+        if ledgered_at is None or now - ledgered_at >= LEDGER_EVERY_SEC:
+            stamp = datetime.fromtimestamp(now, timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+            with open(assistant_dir / "actions-ledger.jsonl", "a") as ledger:
+                ledger.write(json.dumps({
+                    "ts": stamp, "epoch": int(now), "key": f"pulse-preflight-fail-{int(now)}",
+                    "kind": "pulse-preflight-fail", "ws_ref": "(launchd)", "outcome": "failed",
+                    "evidence": f"pulse can't start: {error}"[:300],
+                }) + "\n")
+            ledgered_at = now
+        tmp = record.with_suffix(".json.tmp")
+        tmp.write_text(json.dumps({"failed_at": now, "error": error, "ledgered_at": ledgered_at}))
+        tmp.replace(record)
+    except OSError as exc:
+        print(f"pulse pre-flight could not record the failure in {assistant_dir}: {exc}",
+              file=sys.stderr)
 
 
 def main(argv: list[str], *, execv=os.execv) -> int:
+    assistant_dir = Path.home() / ".assistant"
     for path in STARTUP_FILES:
         try:
             compile(path.read_bytes(), str(path), "exec", dont_inherit=True)
-        except (SyntaxError, ValueError) as exc:
+        except (OSError, SyntaxError, ValueError) as exc:
             error = f"{path}: {type(exc).__name__}: {exc}"
-            record = Path.home() / ".assistant/pulse-preflight.json"
-            record.parent.mkdir(parents=True, exist_ok=True)
-            tmp = record.with_suffix(".json.tmp")
-            tmp.write_text(json.dumps({"failed_at": time.time(), "error": error}))
-            tmp.replace(record)
-            stamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+            now = time.time()
+            _record_failure(assistant_dir, error, now)
+            stamp = datetime.fromtimestamp(now, timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
             print(f"[{stamp}] pulse pre-flight FAILED, skipping this run: {error}",
                   file=sys.stderr)
             return 0
+    (assistant_dir / "pulse-preflight.json").unlink(missing_ok=True)
     execv(sys.executable, [sys.executable, str(PULSE), *argv])
     return 0
 
diff --git a/tests/test_renderer_in_process.py b/tests/test_renderer_in_process.py
index c4b7380..6db3f79 100644
--- a/tests/test_renderer_in_process.py
+++ b/tests/test_renderer_in_process.py
@@ -108,33 +108,35 @@ def _write_preflight(self, payload) -> None:
         (self._tmp / ".assistant/pulse-preflight.json").write_text(
             payload if isinstance(payload, str) else json.dumps(payload))
 
-    def test_preflight_failure_after_last_pulse_renders_red_banner(self):
+    def test_preflight_failure_after_last_pulse_renders_top_alert(self):
         now = int(time.time())
         self._write_heartbeat({"last_pulse_ts": now - 30, "pulse_idx": 99, "model": "m"})
         self._write_preflight({"failed_at": now - 10,
                                "error": "bin/pulse.py: SyntaxError: <bad> (line 1)"})
-        html = self.mod.render_pulse_health()
-        self.assertIn("pulse-bad", html)
-        self.assertIn("Pulse can't start: bin/pulse.py: SyntaxError: &lt;bad&gt; (line 1)",
-                      html)
-        self.assertNotIn("data-pulse-at", html)
+        self.assertEqual(self.mod.render_pulse_alert(), (
+            '<div class="pulse-health pulse-bad" role="alert"><span class="pulse-dot"></span>'
+            "<span class=\"pulse-text\">Pulse can't start: bin/pulse.py: SyntaxError: "
+            "&lt;bad&gt; (line 1)</span></div>"))
+        self.assertIn("Pulse healthy", self.mod.render_pulse_health())
+
+    def test_preflight_failure_with_no_heartbeat_renders_top_alert(self):
+        self._write_preflight({"failed_at": 5, "error": "first run broke"})
+        self.assertIn("Pulse can't start: first run broke", self.mod.render_pulse_alert())
 
     def test_preflight_failure_before_last_pulse_is_ignored(self):
         now = int(time.time())
         self._write_heartbeat({"last_pulse_ts": now - 30, "pulse_idx": 99, "model": "m"})
         self._write_preflight({"failed_at": now - 60, "error": "old failure"})
-        html = self.mod.render_pulse_health()
-        self.assertIn("Pulse healthy", html)
-        self.assertIn(f'data-pulse-at="{now - 30}"', html)
-        self.assertNotIn("old failure", html)
+        self.assertEqual(self.mod.render_pulse_alert(), "")
 
     def test_unreadable_preflight_record_is_ignored(self):
-        self._write_heartbeat({"last_pulse_ts": int(time.time()) - 30,
-                               "pulse_idx": 99, "model": "m"})
         for payload in ("{ corrupt", {"error": "no timestamp"}):
             with self.subTest(payload=payload):
                 self._write_preflight(payload)
-                self.assertIn("Pulse healthy", self.mod.render_pulse_health())
+                self.assertEqual(self.mod.render_pulse_alert(), "")
+
+    def test_no_preflight_record_renders_nothing(self):
+        self.assertEqual(self.mod.render_pulse_alert(), "")
 
     def test_age_formatting_includes_unit(self):
         now = int(time.time())
diff --git a/tests/test_run_pulse_wrapper.py b/tests/test_run_pulse_wrapper.py
index a058778..7df10a1 100644
--- a/tests/test_run_pulse_wrapper.py
+++ b/tests/test_run_pulse_wrapper.py
@@ -41,12 +41,17 @@ def _layout(tmp: Path, pulse_body: str, model_tiers: str = "TIERS = {}\n") -> Pa
     return tmp / "bin/run-pulse.py"
 
 
-def test_healthy_checkout_execs_pulse_with_same_interpreter_and_args():
+def test_healthy_checkout_clears_old_failure_and_execs_pulse(tmp_path, monkeypatch):
+    monkeypatch.setenv("HOME", str(tmp_path))
+    record = tmp_path / ".assistant/pulse-preflight.json"
+    record.parent.mkdir()
+    record.write_text('{"failed_at": 1, "error": "old"}')
     mod = _load()
     calls = []
     assert mod.main(["--pulse-idx", "5"], execv=lambda *a: calls.append(a)) == 0
     assert calls == [(sys.executable, [sys.executable, str(REPO / "bin/pulse.py"),
                                        "--pulse-idx", "5"])]
+    assert not record.exists()
 
 
 def test_broken_pulse_skips_the_run_and_records_why(tmp_path, capsys, monkeypatch):
@@ -62,7 +67,46 @@ def test_broken_pulse_skips_the_run_and_records_why(tmp_path, capsys, monkeypatc
     assert f"{tmp_path / 'bin/pulse.py'}: SyntaxError" in err
     record = json.loads((tmp_path / "home/.assistant/pulse-preflight.json").read_text())
     assert record["error"].startswith(f"{tmp_path / 'bin/pulse.py'}: SyntaxError")
-    assert isinstance(record["failed_at"], float)
+    assert record["ledgered_at"] == record["failed_at"]
+    [entry] = [json.loads(line) for line in
+               (tmp_path / "home/.assistant/actions-ledger.jsonl").read_text().splitlines()]
+    assert entry["kind"] == "pulse-preflight-fail"
+    assert entry["outcome"] == "failed"
+    assert entry["evidence"] == f"pulse can't start: {record['error']}"[:300]
+
+
+def test_ledger_entry_is_written_for_a_new_error_then_once_a_day(tmp_path):
+    mod = _load()
+    ledger = tmp_path / "actions-ledger.jsonl"
+
+    def entries():
+        return [json.loads(line)["evidence"] for line in ledger.read_text().splitlines()]
+
+    mod._record_failure(tmp_path, "err A", 1000.0)
+    mod._record_failure(tmp_path, "err A", 1000.0 + 3600)
+    assert entries() == ["pulse can't start: err A"]
+    mod._record_failure(tmp_path, "err B", 1000.0 + 7200)
+    mod._record_failure(tmp_path, "err B", 1000.0 + 7200 + 86400)
+    assert entries() == ["pulse can't start: err A", "pulse can't start: err B",
+                         "pulse can't start: err B"]
+    record = json.loads((tmp_path / "pulse-preflight.json").read_text())
+    assert record == {"failed_at": 1000.0 + 7200 + 86400, "error": "err B",
+                      "ledgered_at": 1000.0 + 7200 + 86400}
+
+
+def test_unwritable_record_still_skips_cleanly(tmp_path, capsys, monkeypatch):
+    monkeypatch.setenv("HOME", str(tmp_path / "home"))
+    (tmp_path / "home").mkdir()
+    (tmp_path / "home/.assistant").write_text("a file where the folder should be")
+    _layout(tmp_path, "def broken(:\n")
+    mod = _load()
+    calls = []
+    with mock.patch.object(mod, "STARTUP_FILES", (tmp_path / "bin/pulse.py",)):
+        assert mod.main([], execv=lambda *a: calls.append(a)) == 0
+    assert calls == []
+    err = capsys.readouterr().err
+    assert "could not record the failure" in err
+    assert "pulse pre-flight FAILED, skipping this run" in err
 
 
 def test_broken_startup_import_skips_the_run(tmp_path):
@@ -74,6 +118,16 @@ def test_broken_startup_import_skips_the_run(tmp_path):
     assert "src/assistant/model_tiers.py: SyntaxError" in r.stderr
 
 
+def test_missing_startup_file_skips_the_run(tmp_path):
+    wrapper = _layout(tmp_path, 'print("RAN")\n')
+    (tmp_path / "src/assistant/model_tiers.py").rename(tmp_path / "src/assistant/moved.py")
+    r = subprocess.run([sys.executable, str(wrapper)], capture_output=True, text=True,
+                       timeout=60, env=dict(os.environ, HOME=str(tmp_path / "home")))
+    assert r.returncode == 0
+    assert r.stdout == ""
+    assert "model_tiers.py: FileNotFoundError" in r.stderr
+
+
 def test_broken_later_module_does_not_block_the_pulse(tmp_path):
     # The pulse guards the modules it loads later; an unrelated broken module
     # must not stop every run.
@@ -85,16 +139,35 @@ def test_broken_later_module_does_not_block_the_pulse(tmp_path):
     assert r.stdout.strip() == "RAN"
 
 
+def _module_level_imports(nodes):
+    for node in nodes:
+        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
+            continue
+        if isinstance(node, (ast.Import, ast.ImportFrom)):
+            yield node
+        for field in ("body", "orelse", "finalbody", "handlers"):
+            yield from _module_level_imports(getattr(node, field, []))
+
+
 def test_startup_files_match_pulse_module_level_imports():
     tree = ast.parse((REPO / "bin/pulse.py").read_text())
-    imported = {f"src/assistant/{alias.name}.py" for node in tree.body
-                if isinstance(node, ast.ImportFrom) and node.module == "assistant"
-                for alias in node.names}
+    imported = {"bin/pulse.py"}
+    for node in _module_level_imports(tree.body):
+        modules = ([alias.name for alias in node.names] if isinstance(node, ast.Import)
+                   else [node.module])
+        for module in modules:
+            top = module.split(".")[0]
+            if top in sys.stdlib_module_names or top == "__future__":
+                continue
+            assert module == "assistant", f"add {module} to run-pulse.py STARTUP_FILES"
+            imported.add("src/assistant/__init__.py")
+            imported.update(f"src/assistant/{alias.name}.py" for alias in node.names)
     listed = {str(path.relative_to(REPO)) for path in _load().STARTUP_FILES}
-    assert imported | {"bin/pulse.py", "src/assistant/__init__.py"} == listed
+    assert imported == listed
 
 
-def test_script_entry_point_execs_pulse():
+def test_script_entry_point_execs_pulse(tmp_path, monkeypatch):
+    monkeypatch.setenv("HOME", str(tmp_path))
     calls = []
     with mock.patch.object(os, "execv", lambda *a: calls.append(a)), \
             mock.patch.object(sys, "argv", [str(WRAPPER), "--dry-run"]):