Guard the self-update pipeline against shipping broken Python

CHANGELOG.md

@@ -10,6 +10,19 @@ The version is carried in `pyproject.toml` and `src/assistant/__init__.py`
 ## [Unreleased]
 
 ### Added
+- 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 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. 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.
   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.
@@ -27,6 +40,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.

bin/pulse.py

@@ -457,10 +457,12 @@ 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 recorded in the last day.
+    if not changed and reason in (None, "syntax-fail-known") and not result.get("error"):
         return
 
+    kind = "self-update"
     if changed:
         files = result.get("files_changed", [])
         installed = result.get("installed")
@@ -495,6 +497,15 @@ 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":
+        # 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')} (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"
         evidence = f"self-update {reason or 'error'}: {result.get('error', '')}"[:300]
@@ -505,7 +516,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,

bin/render-assistant-page.py

@@ -2254,6 +2254,27 @@ def _ws_num(c):
     return f'<div class="fleet-board">{"".join(col_html)}</div>', total
 
 
+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())
+        failed_at, error = float(record["failed_at"]), str(record["error"])
+    except (OSError, ValueError, TypeError, KeyError):
+        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:
     """One-line banner showing whether the assistant-pulse cron is alive.
     Reads ~/.assistant/heartbeat.json and color-codes by age:
@@ -2451,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()
@@ -4076,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>

bin/run-pulse.py

@@ -0,0 +1,81 @@
+#!/usr/bin/env python3
+"""Launchd pre-flight for the Assistant pulse.
+
+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
+
+import json
+import os
+import sys
+import time
+from datetime import datetime, timezone
+from pathlib import Path
+
+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")
+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 (OSError, SyntaxError, ValueError) as exc:
+            error = f"{path}: {type(exc).__name__}: {exc}"
+            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
+
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv[1:]))

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 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.
      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,6 +43,7 @@
 
 import json
 import subprocess
+import sys
 import time
 from pathlib import Path
 
@@ -61,6 +68,18 @@
 # (`git stash list` / `git stash pop`); it is never dropped or discarded.
 DEFAULT_DIRTY_STASH_AFTER_SEC = 86400  # 1 day
 
+# ── Pre-pull syntax gate ─────────────────────────────────────────────────────
+# Paths the running system loads, runs, or installs from the checkout. Add new
+# 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/", "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
+
 
 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 +197,64 @@ def should_attempt(marker: dict, now: float, interval_sec: int) -> bool:
     return (now - last) >= interval_sec
 
 
+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. 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
+        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(["git", "-C", str(repo), "cat-file", "blob", f"{sha}:{name}"],
+                           capture_output=True, timeout=90)
+    except subprocess.TimeoutExpired:
+        return None
+    return p.stdout if p.returncode == 0 else None
+
+
+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, 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]
+    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]}"
+        marker = _conflict_marker_line(source.decode("utf-8", errors="replace"))
+        if 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 "broken", f"{name}: {exc}"[:500]
+    return "ok", "ok"
+
+
 def maybe_update(
     repo: Path,
     *,
@@ -259,14 +336,42 @@ 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 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]
+    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
+    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
+        _log(f"refused {old_head[:12]}..{new_sha[:12]}, a file failed the check: "
+             f"{detail[:200]}")
+        return result
+
     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.
@@ -285,12 +390,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")

launchagents/com.assistant.assistant-pulse.plist

@@ -7,7 +7,7 @@
     <key>ProgramArguments</key>
     <array>
         <string>__PYTHON__</string>
-        <string>__REPO__/bin/pulse.py</string>
+        <string>__REPO__/bin/run-pulse.py</string>
     </array>
     <key>StartInterval</key>
     <integer>300</integer>

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