diff --git a/docs/operations/brainlayer-health-check.md b/docs/operations/brainlayer-health-check.md index a0ba7b11..a3f7371a 100644 --- a/docs/operations/brainlayer-health-check.md +++ b/docs/operations/brainlayer-health-check.md @@ -15,6 +15,14 @@ brainlayer health-check --json --heal - Active chunks missing semantic vectors are decreasing across ticks. One unchanged tick is tolerated; the second unchanged tick alarms. - BrainBar's served MCP socket can answer a `brain_search` canary with at least one result. +The missing-vector count is exact, but it computes the ID difference through the +covering `chunks` and `chunk_vectors_rowids` indexes before reading chunk payloads. +This keeps the scheduled check independent of database payload size. + +The check has a 45-second internal deadline. A query that exhausts that budget is +interrupted and the state file is still refreshed with `slow_check: true`, the +stage name, and the elapsed duration. + ## Self-Heal With `--heal`, the check uses `launchctl kickstart -k` for cheap recovery: @@ -22,7 +30,19 @@ With `--heal`, the check uses `launchctl kickstart -k` for cheap recovery: - `com.brainlayer.hotlane-brainbar` when hotlane is dead, backlog is disabled, or missing vectors are climbing/stalled. - `com.brainlayer.brainbar-daemon` when the BrainBar MCP canary fails or returns zero results. -The command always writes the latest missing-vector count to `~/.local/share/brainlayer/health-check-state.json`. +Before a kickstart, the check reads the launchd process state. Processes in an +uninterruptible `U`/`D` state are not kickstarted repeatedly; the check records a +`heal_backoff` action and the existing circuit breaker escalates after repeated +failed ticks. + +The command writes the latest successfully measured missing-vector count to +`~/.local/share/brainlayer/health-check-state.json`; a timed-out tick preserves +the previous count and marks the state as slow. + +The independent Tier-0 watchdog treats `slow_check: true` as unhealthy even +when the state file was just refreshed, and treats any state as stale after 900 +seconds. Its first alert is immediate; repeat alerts for the same failure class +are suppressed for 1,800 seconds while recovery attempts continue. ## Logs diff --git a/scripts/launchd/com.brainlayer.tier0-watchdog.plist b/scripts/launchd/com.brainlayer.tier0-watchdog.plist index 8140e760..42a07fd6 100644 --- a/scripts/launchd/com.brainlayer.tier0-watchdog.plist +++ b/scripts/launchd/com.brainlayer.tier0-watchdog.plist @@ -15,6 +15,12 @@ __HOME__ PATH /usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin + TIER0_ALERT_STATE_PATH + __HOME__/.local/share/brainlayer/tier0-watchdog-alert-state + TIER0_REPEAT_ALERT_SECONDS + 1800 + TIER0_STALE_SECONDS + 900 StandardOutPath __HOME__/Library/Logs/brainlayer/tier0-watchdog.out.log diff --git a/scripts/tier0-watchdog.sh b/scripts/tier0-watchdog.sh index 95c2dfed..e12b8232 100755 --- a/scripts/tier0-watchdog.sh +++ b/scripts/tier0-watchdog.sh @@ -12,6 +12,7 @@ TIER0_ID=${TIER0_ID:-/usr/bin/id} TIER0_SLEEP=${TIER0_SLEEP:-/bin/sleep} TIER0_DIRNAME=${TIER0_DIRNAME:-/usr/bin/dirname} TIER0_MKDIR=${TIER0_MKDIR:-/bin/mkdir} +TIER0_GREP=${TIER0_GREP:-/usr/bin/grep} TIER0_LABEL=${TIER0_LABEL:-com.brainlayer.health-check} if [ -z "${TIER0_DOMAIN:-}" ]; then @@ -26,8 +27,10 @@ fi TIER0_STATE_PATH=${TIER0_STATE_PATH:-$HOME/.local/share/brainlayer/health-check-state.json} TIER0_HEALTH_PLIST_PATH=${TIER0_HEALTH_PLIST_PATH:-$HOME/Library/LaunchAgents/com.brainlayer.health-check.plist} TIER0_LOG_PATH=${TIER0_LOG_PATH:-$HOME/.local/share/brainlayer/logs/tier0-watchdog.log} +TIER0_ALERT_STATE_PATH=${TIER0_ALERT_STATE_PATH:-$HOME/.local/share/brainlayer/tier0-watchdog-alert-state} TIER0_NOTIFY_ENDPOINT=${TIER0_NOTIFY_ENDPOINT:-http://localhost:3847/notify} -TIER0_STALE_SECONDS=${TIER0_STALE_SECONDS:-1200} +TIER0_STALE_SECONDS=${TIER0_STALE_SECONDS:-900} +TIER0_REPEAT_ALERT_SECONDS=${TIER0_REPEAT_ALERT_SECONDS:-1800} TIER0_ALERT_TIMEOUT_SECONDS=${TIER0_ALERT_TIMEOUT_SECONDS:-3} TIER0_NOTIFY_TIMEOUT_SECONDS=${TIER0_NOTIFY_TIMEOUT_SECONDS:-3} @@ -54,6 +57,7 @@ require_epoch() { } require_positive_integer TIER0_STALE_SECONDS "$TIER0_STALE_SECONDS" +require_positive_integer TIER0_REPEAT_ALERT_SECONDS "$TIER0_REPEAT_ALERT_SECONDS" require_positive_integer TIER0_ALERT_TIMEOUT_SECONDS "$TIER0_ALERT_TIMEOUT_SECONDS" require_positive_integer TIER0_NOTIFY_TIMEOUT_SECONDS "$TIER0_NOTIFY_TIMEOUT_SECONDS" @@ -121,46 +125,92 @@ alert_all_channels() { wait_for_alerts "$log_pid" "$osascript_pid" "$curl_pid" } +should_alert() { + failure_key=$1 + if [ ! -f "$TIER0_ALERT_STATE_PATH" ]; then + return 0 + fi + + last_alert_epoch= + last_failure_key= + IFS="$(printf '\t')" read -r last_alert_epoch last_failure_key < "$TIER0_ALERT_STATE_PATH" || return 0 + case "$last_alert_epoch" in + ''|*[!0-9]*) return 0 ;; + esac + if [ "$last_failure_key" != "$failure_key" ] || [ "$last_alert_epoch" -gt "$now_epoch" ]; then + return 0 + fi + alert_age=$((now_epoch - last_alert_epoch)) + [ "$alert_age" -ge "$TIER0_REPEAT_ALERT_SECONDS" ] +} + +record_alert() { + failure_key=$1 + alert_state_dir=$($TIER0_DIRNAME "$TIER0_ALERT_STATE_PATH" 2>/dev/null) || return 1 + "$TIER0_MKDIR" -p "$alert_state_dir" 2>/dev/null || return 1 + printf '%s\t%s\n' "$now_epoch" "$failure_key" > "$TIER0_ALERT_STATE_PATH" +} + +reset_alert_cooldown() { + if [ -f "$TIER0_ALERT_STATE_PATH" ]; then + printf '0\tok\n' > "$TIER0_ALERT_STATE_PATH" + fi +} + target="$TIER0_DOMAIN/$TIER0_LABEL" failure_reason= +failure_key= label_loaded=0 if "$TIER0_LAUNCHCTL" print "$target" >/dev/null 2>&1; then label_loaded=1 else failure_reason=label_unloaded + failure_key=label_unloaded fi if [ "$label_loaded" -eq 1 ]; then if [ ! -f "$TIER0_STATE_PATH" ]; then failure_reason=state_missing + failure_key=state_missing else state_mtime=$($TIER0_STAT -f %m "$TIER0_STATE_PATH" 2>/dev/null) || state_mtime= case "$state_mtime" in ''|*[!0-9]*) failure_reason=state_mtime_unreadable + failure_key=state_mtime_unreadable ;; *) if [ "$state_mtime" -gt "$now_epoch" ]; then future_offset=$((state_mtime - now_epoch)) failure_reason="state_mtime_future offset=${future_offset}s" + failure_key=state_mtime_future else state_age=$((now_epoch - state_mtime)) if [ "$state_age" -ge "$TIER0_STALE_SECONDS" ]; then failure_reason="state_stale age=${state_age}s threshold=${TIER0_STALE_SECONDS}s" + failure_key=state_stale fi fi ;; esac + if [ -z "$failure_reason" ] && "$TIER0_GREP" -Eq '"slow_check"[[:space:]]*:[[:space:]]*true' "$TIER0_STATE_PATH" 2>/dev/null; then + failure_reason=state_slow_check + failure_key=state_slow_check + fi fi fi if [ -z "$failure_reason" ]; then + reset_alert_cooldown exit 0 fi -# Detection and all alert attempts intentionally precede every recovery command. -alert_all_channels "$failure_reason" +# Detection and all due alert attempts intentionally precede every recovery command. +if should_alert "$failure_key"; then + alert_all_channels "$failure_reason" + record_alert "$failure_key" || : +fi if [ "$label_loaded" -eq 0 ]; then "$TIER0_LAUNCHCTL" bootstrap "$TIER0_DOMAIN" "$TIER0_HEALTH_PLIST_PATH" >/dev/null 2>&1 || : diff --git a/src/brainlayer/doctor.py b/src/brainlayer/doctor.py index 63083c3a..633d90e7 100644 --- a/src/brainlayer/doctor.py +++ b/src/brainlayer/doctor.py @@ -387,7 +387,7 @@ def _check_brainbar_version_consistency( def run_doctor( config: DoctorConfig, *, - ps_output_fn: Callable[[], str] = _default_ps_output, + ps_output_fn: Callable[[], str | None] = _default_ps_output, command_runner: CommandRunner = _default_command_runner, now_fn: Callable[[], datetime] = lambda: datetime.now(UTC), ) -> DoctorResult: @@ -500,10 +500,17 @@ def warning(code: str, message: str, **details: Any) -> None: command_runner=command_runner, ) - hotlane_processes = parse_hotlane_processes(ps_output_fn()) - result.hotlane_running = bool(hotlane_processes) - if not hotlane_processes: - fatal("hotlane_dead", "hot-lane embedding process is not running") + ps_output = ps_output_fn() + if ps_output is None: + fatal( + "process_snapshot_failed", + "could not read the process table; hot-lane daemon liveness was not evaluated", + ) + else: + hotlane_processes = parse_hotlane_processes(ps_output) + result.hotlane_running = bool(hotlane_processes) + if not hotlane_processes: + fatal("hotlane_dead", "hot-lane embedding process is not running") if result.enrichment_backlog and result.enrichment_backlog > 0: warning( diff --git a/src/brainlayer/health_check.py b/src/brainlayer/health_check.py index 260e87c6..e1bb402f 100644 --- a/src/brainlayer/health_check.py +++ b/src/brainlayer/health_check.py @@ -10,6 +10,8 @@ import sqlite3 import subprocess import sys +import threading +import time from dataclasses import asdict, dataclass, field, replace from datetime import UTC, datetime from pathlib import Path @@ -43,8 +45,45 @@ DEFAULT_BACKLOG_BATCH = 4 DEFAULT_HEAL_MIN_CONSECUTIVE_FAILURES = 2 DEFAULT_HEAL_CIRCUIT_BREAKER_LIMIT = 3 +DEFAULT_MAX_DURATION_SECONDS = 45.0 HEAL_MIN_CONSECUTIVE_FAILURES_ENV = "BRAINLAYER_HEAL_MIN_CONSECUTIVE_FAILURES" +MISSING_EMBEDDINGS_SQL = """ + SELECT COUNT(*) + FROM chunks c + JOIN ( + SELECT id FROM chunks + EXCEPT + SELECT id FROM chunk_vectors_rowids + ) missing ON missing.id = c.id + WHERE c.content IS NOT NULL + AND c.content != '' + AND c.archived_at IS NULL + AND c.superseded_by IS NULL + AND c.aggregated_into IS NULL + AND COALESCE(c.archived, 0) = 0 + AND COALESCE(c.status, 'active') = 'active' +""" + +ENRICHMENT_BACKLOG_SQL = """ + SELECT COUNT(*) + FROM chunks + WHERE enriched_at IS NULL + AND enrich_status IS NULL + AND COALESCE(char_count, length(content), 0) >= 50 + AND content IS NOT NULL + AND content != '' + AND archived_at IS NULL + AND superseded_by IS NULL + AND aggregated_into IS NULL + AND COALESCE(archived, 0) = 0 + AND COALESCE(status, 'active') = 'active' +""" + + +class HealthCheckDeadlineExceeded(RuntimeError): + """Raised when a read-only health query consumes the check's time budget.""" + def _env_int(name: str, default: int, *, minimum: int = 1) -> int: raw_value = os.environ.get(name) @@ -113,6 +152,7 @@ class HealthCheckConfig: queue_page_oldest_seconds: int = 4 * 60 * 60 queue_page_bytes: int = 2 * 1024 * 1024 * 1024 heal_circuit_breaker_limit: int = DEFAULT_HEAL_CIRCUIT_BREAKER_LIMIT + max_duration_seconds: float = DEFAULT_MAX_DURATION_SECONDS heal: bool = False socket_timeout_seconds: float = 5.0 max_stalled_ticks: int = 2 @@ -153,6 +193,9 @@ class HealthCheckResult: lock_holder: LockHolder | None = None canary_ok: bool = False canary_result_count: int | None = None + duration_seconds: float = 0.0 + slow_check: bool = False + slow_check_stage: str | None = None def to_dict(self) -> dict[str, Any]: return asdict(self) @@ -162,16 +205,29 @@ def to_dict(self) -> dict[str, Any]: SocketRequestFn = Callable[[Path, str, float], dict[str, Any]] -def _default_ps_output() -> str: - result = subprocess.run(["ps", "axo", "pid=,command="], text=True, capture_output=True, check=False) - return result.stdout +def _default_ps_output() -> str | None: + try: + result = subprocess.run( + ["ps", "axo", "pid=,command="], + text=True, + capture_output=True, + check=False, + timeout=5, + ) + return result.stdout if result.returncode == 0 else None + except (FileNotFoundError, subprocess.TimeoutExpired): + return None def _default_command_runner(args: list[str]) -> subprocess.CompletedProcess[str]: try: - return subprocess.run(args, text=True, capture_output=True, check=False) + return subprocess.run(args, text=True, capture_output=True, check=False, timeout=5) except FileNotFoundError as exc: return subprocess.CompletedProcess(args=args, returncode=127, stdout="", stderr=str(exc)) + except subprocess.TimeoutExpired as exc: + return subprocess.CompletedProcess( + args=args, returncode=124, stdout=exc.stdout or "", stderr="command timed out" + ) def _command_returncode(result: Any) -> int: @@ -196,6 +252,28 @@ def _kickstart(label: str, command_runner: CommandRunner) -> str: return f"kickstart:{label}" +def _launchd_process_state(label: str, command_runner: CommandRunner) -> tuple[int, str] | None: + launchd_result = command_runner(["launchctl", "print", _launchd_target(label)]) + if _command_returncode(launchd_result) != 0: + return None + match = re.search(r"\bpid\s*=\s*(\d+)\b", _command_stdout(launchd_result)) + if match is None: + return None + pid = int(match.group(1)) + ps_result = command_runner(["ps", "-o", "state=", "-p", str(pid)]) + if _command_returncode(ps_result) != 0: + return None + process_state = _command_stdout(ps_result).strip().splitlines() + if not process_state: + return None + return pid, process_state[0].strip() + + +def _state_is_uninterruptible(process_state: str) -> bool: + normalized = process_state.upper() + return "U" in normalized or normalized.startswith("D") + + def _bootstrap_if_absent(label: str, plist_path: Path, command_runner: CommandRunner) -> str: loaded = _launchd_label_loaded(label, command_runner) if loaded is True: @@ -408,55 +486,59 @@ def _holder_message(holder: LockHolder) -> str: return f"holder pid={holder.pid} command={command} db_path={holder.db_path}" -def count_missing_embeddings(db_path: Path) -> int: +def _read_only_count( + db_path: Path, + sql: str, + *, + stage: str, + deadline_at: float | None = None, + monotonic_fn: Callable[[], float] = time.monotonic, +) -> int: uri = f"file:{db_path.expanduser()}?mode=ro" conn = sqlite3.connect(uri, uri=True, timeout=5) try: conn.execute("PRAGMA query_only = ON") - row = conn.execute( - """ - SELECT COUNT(*) - FROM chunks c - LEFT JOIN chunk_vectors_rowids r ON r.id = c.id - WHERE r.id IS NULL - AND c.content IS NOT NULL - AND c.content != '' - AND c.archived_at IS NULL - AND c.superseded_by IS NULL - AND c.aggregated_into IS NULL - AND COALESCE(c.archived, 0) = 0 - AND COALESCE(c.status, 'active') = 'active' - """ - ).fetchone() + if deadline_at is not None: + conn.set_progress_handler(lambda: int(monotonic_fn() >= deadline_at), 1_000) + try: + row = conn.execute(sql).fetchone() + except sqlite3.OperationalError as exc: + if deadline_at is not None and "interrupted" in str(exc).lower(): + raise HealthCheckDeadlineExceeded(f"health-check deadline exceeded during {stage}") from exc + raise return int(row[0] if row else 0) finally: conn.close() -def _enrichment_backlog(db_path: Path) -> int: - uri = f"file:{db_path.expanduser()}?mode=ro" - conn = sqlite3.connect(uri, uri=True, timeout=5) - try: - conn.execute("PRAGMA query_only = ON") - row = conn.execute( - """ - SELECT COUNT(*) - FROM chunks - WHERE enriched_at IS NULL - AND enrich_status IS NULL - AND COALESCE(char_count, length(content), 0) >= 50 - AND content IS NOT NULL - AND content != '' - AND archived_at IS NULL - AND superseded_by IS NULL - AND aggregated_into IS NULL - AND COALESCE(archived, 0) = 0 - AND COALESCE(status, 'active') = 'active' - """ - ).fetchone() - return int(row[0] if row else 0) - finally: - conn.close() +def count_missing_embeddings( + db_path: Path, + *, + deadline_at: float | None = None, + monotonic_fn: Callable[[], float] = time.monotonic, +) -> int: + return _read_only_count( + db_path, + MISSING_EMBEDDINGS_SQL, + stage="missing_embeddings", + deadline_at=deadline_at, + monotonic_fn=monotonic_fn, + ) + + +def _enrichment_backlog( + db_path: Path, + *, + deadline_at: float | None = None, + monotonic_fn: Callable[[], float] = time.monotonic, +) -> int: + return _read_only_count( + db_path, + ENRICHMENT_BACKLOG_SQL, + stage="enrichment_backlog", + deadline_at=deadline_at, + monotonic_fn=monotonic_fn, + ) def _load_state(path: Path) -> dict[str, Any]: @@ -626,11 +708,28 @@ def _apply_heals( ) continue if consecutive_failures >= threshold: - action = ( - _bootstrap_if_absent(label, plist_path, command_runner) - if issue_code in bootstrap_issue_codes - else _kickstart(label, command_runner) - ) + if issue_code in bootstrap_issue_codes: + action = _bootstrap_if_absent(label, plist_path, command_runner) + else: + process = _launchd_process_state(label, command_runner) + if process is not None and _state_is_uninterruptible(process[1]): + pid, process_state = process + action = f"heal_backoff:{label}:{issue_code}:pid={pid}:state={process_state}" + if action not in result.actions: + result.actions.append(action) + _emit_heal_event( + { + "_type": "heal_backoff", + "label": label, + "issue_code": issue_code, + "consecutive_failures": consecutive_failures, + "pid": pid, + "process_state": process_state, + **details, + } + ) + continue + action = _kickstart(label, command_runner) if action.startswith("bootstrap:"): tripped.discard(key) heal_failures.pop(key, None) @@ -694,18 +793,51 @@ def _pause_sentinel_state(config: HealthCheckConfig, now: datetime) -> tuple[dic return payload, not stale, stale -def _source_recent(config: HealthCheckConfig, now: datetime, window_seconds: int) -> bool: +def _source_recent( + config: HealthCheckConfig, + now: datetime, + window_seconds: int, + *, + deadline_at: float | None = None, + monotonic_fn: Callable[[], float] = time.monotonic, +) -> bool: import glob - cutoff = now.timestamp() - window_seconds - for pattern in config.source_jsonl_globs: - for raw_path in glob.iglob(str(Path(pattern).expanduser()), recursive=True): - try: - if Path(raw_path).stat().st_mtime >= cutoff: - return True - except OSError: - continue - return False + def scan() -> bool: + cutoff = now.timestamp() - window_seconds + for pattern in config.source_jsonl_globs: + for raw_path in glob.iglob(str(Path(pattern).expanduser()), recursive=True): + try: + if Path(raw_path).stat().st_mtime >= cutoff: + return True + except OSError: + continue + return False + + if deadline_at is None: + return scan() + + remaining = deadline_at - monotonic_fn() + if remaining <= 0: + raise HealthCheckDeadlineExceeded("health-check deadline exceeded during source_recent") + + completed = threading.Event() + outcome: dict[str, Any] = {} + + def bounded_scan() -> None: + try: + outcome["value"] = scan() + except Exception as exc: + outcome["error"] = exc + finally: + completed.set() + + threading.Thread(target=bounded_scan, name="brainlayer-health-source-scan", daemon=True).start() + if not completed.wait(timeout=remaining): + raise HealthCheckDeadlineExceeded("health-check deadline exceeded during source_recent") + if error := outcome.get("error"): + raise error + return bool(outcome.get("value", False)) def _queue_stats(queue_dir: Path, now: datetime) -> tuple[int, int, float | None]: @@ -757,11 +889,14 @@ def _plist_for_label(config: HealthCheckConfig, label: str) -> Path: def run_health_check( config: HealthCheckConfig, *, - ps_output_fn: Callable[[], str] = _default_ps_output, + ps_output_fn: Callable[[], str | None] = _default_ps_output, socket_request_fn: SocketRequestFn = send_brainbar_search_canary, command_runner: CommandRunner = _default_command_runner, now_fn: Callable[[], datetime] = lambda: datetime.now(UTC), + monotonic_fn: Callable[[], float] = time.monotonic, ) -> HealthCheckResult: + started_monotonic = monotonic_fn() + deadline_at = started_monotonic + max(1.0, config.max_duration_seconds) now = now_fn() result = HealthCheckResult(checked_at=now.isoformat(), ok=True) state = _load_state(config.state_path) @@ -781,33 +916,79 @@ def add_issue(code: str, severity: str, message: str) -> None: } ) + def finish_slow(stage: str, message: str) -> HealthCheckResult: + result.slow_check = True + result.slow_check_stage = stage + result.duration_seconds = max(0.0, monotonic_fn() - started_monotonic) + add_issue("slow_check", "critical", message) + state_payload: dict[str, Any] = dict(state) + state_payload["ts"] = now.isoformat() + state_payload["slow_check"] = True + state_payload["slow_check_stage"] = stage + state_payload["duration_seconds"] = result.duration_seconds + _write_state(config.state_path, state_payload) + result.ok = False + return result + + def deadline_reached(stage: str) -> HealthCheckResult | None: + if monotonic_fn() < deadline_at: + return None + return finish_slow(stage, f"health-check exceeded {config.max_duration_seconds:.0f}s during {stage}") + ps_output = ps_output_fn() - lock_holder = _read_writer_lock_holder(config.db_path, ps_output) - result.lock_holder = lock_holder - hotlane_processes = parse_hotlane_processes(ps_output) - result.hotlane_running = bool(hotlane_processes) - if not hotlane_processes: - add_issue("hotlane_dead", "critical", "hotlane BrainBar embedding daemon is not running") - heal_issue_labels["hotlane_dead"] = (config.hotlane_label, _plist_for_label(config, config.hotlane_label)) + lock_holder = None + if ps_output is None: + add_issue( + "process_snapshot_failed", + "critical", + "could not read the process table; daemon liveness and writer ownership were not evaluated", + ) else: - result.backlog_batch = min(process.backlog_batch for process in hotlane_processes) - if any(process.backlog_batch <= 0 for process in hotlane_processes): - add_issue("hotlane_backlog_disabled", "critical", "--backlog-batch is 0; embeddings will not drain") - heal_issue_labels["hotlane_backlog_disabled"] = ( + lock_holder = _read_writer_lock_holder(config.db_path, ps_output) + result.lock_holder = lock_holder + hotlane_processes = parse_hotlane_processes(ps_output) + result.hotlane_running = bool(hotlane_processes) + if not hotlane_processes: + add_issue("hotlane_dead", "critical", "hotlane BrainBar embedding daemon is not running") + heal_issue_labels["hotlane_dead"] = ( config.hotlane_label, _plist_for_label(config, config.hotlane_label), ) + else: + result.backlog_batch = min(process.backlog_batch for process in hotlane_processes) + if any(process.backlog_batch <= 0 for process in hotlane_processes): + add_issue("hotlane_backlog_disabled", "critical", "--backlog-batch is 0; embeddings will not drain") + heal_issue_labels["hotlane_backlog_disabled"] = ( + config.hotlane_label, + _plist_for_label(config, config.hotlane_label), + ) previous_missing = state.get("missing_vectors") result.previous_missing_vectors = int(previous_missing) if isinstance(previous_missing, int) else None try: - result.missing_vectors = count_missing_embeddings(config.db_path) + result.missing_vectors = count_missing_embeddings( + config.db_path, + deadline_at=deadline_at, + monotonic_fn=monotonic_fn, + ) + except HealthCheckDeadlineExceeded as exc: + return finish_slow("missing_embeddings", str(exc)) + except sqlite3.OperationalError as exc: + if "interrupted" in str(exc).lower(): + return finish_slow("missing_embeddings", "health-check deadline interrupted missing_embeddings") + add_issue( + "missing_embeddings_count_failed", + "critical", + f"could not count missing embeddings: {exc}", + ) except Exception as exc: add_issue( "missing_embeddings_count_failed", "critical", f"could not count missing embeddings: {exc}", ) + if slow_result := deadline_reached("missing_embeddings"): + return slow_result stalled_ticks = 0 if result.missing_vectors is not None: @@ -861,6 +1042,8 @@ def add_issue(code: str, severity: str, message: str) -> None: config.brainbar_daemon_label, _plist_for_label(config, config.brainbar_daemon_label), ) + if slow_result := deadline_reached("brain_search_canary"): + return slow_result if config.heal: for label in ( @@ -889,6 +1072,8 @@ def add_issue(code: str, severity: str, message: str) -> None: if loaded is False: add_issue(issue_code, "critical", message) heal_issue_labels[issue_code] = (label, _plist_for_label(config, label)) + if slow_result := deadline_reached("launchd_status"): + return slow_result pause_payload, pause_active, pause_stale = _pause_sentinel_state(config, now) if pause_stale: @@ -927,11 +1112,22 @@ def add_issue(code: str, severity: str, message: str) -> None: or (queue_oldest_age is not None and queue_oldest_age >= config.queue_page_oldest_seconds) ): _push_notification("BrainLayer queue backlog", f"queue_count={queue_count} queue_bytes={queue_bytes}") + if slow_result := deadline_reached("queue_stats"): + return slow_result watcher_health = _load_json(config.watcher_health_path) watcher_poll_count = watcher_health.get("poll_count") previous_watcher_poll_count = state.get("watcher_poll_count") - source_recent = _source_recent(config, now, config.max_offsets_age_seconds) + try: + source_recent = _source_recent( + config, + now, + config.max_offsets_age_seconds, + deadline_at=deadline_at, + monotonic_fn=monotonic_fn, + ) + except HealthCheckDeadlineExceeded as exc: + return finish_slow("source_recent", str(exc)) offsets_age = _path_age_seconds(config.offsets_path, now) watcher_health_age = _path_age_seconds(config.watcher_health_path, now) if ( @@ -956,7 +1152,22 @@ def add_issue(code: str, severity: str, message: str) -> None: previous_drain_total = state.get("drain_drained_total") drain_starved = _drain_health_has_sqlite_busy_signal(drain_health) try: - enrichment_backlog = _enrichment_backlog(config.db_path) + enrichment_backlog = _enrichment_backlog( + config.db_path, + deadline_at=deadline_at, + monotonic_fn=monotonic_fn, + ) + except HealthCheckDeadlineExceeded as exc: + return finish_slow("enrichment_backlog", str(exc)) + except sqlite3.OperationalError as exc: + if "interrupted" in str(exc).lower(): + return finish_slow("enrichment_backlog", "health-check deadline interrupted enrichment_backlog") + enrichment_backlog = 0 + add_issue( + "enrichment_backlog_count_failed", + "critical", + f"could not count enrichment backlog: {exc}", + ) except Exception as exc: enrichment_backlog = 0 add_issue( @@ -964,6 +1175,8 @@ def add_issue(code: str, severity: str, message: str) -> None: "critical", f"could not count enrichment backlog: {exc}", ) + if slow_result := deadline_reached("enrichment_backlog"): + return slow_result drain_liveness_issue = check_drain_liveness( drain_label=config.drain_label, drain_loaded=drain_loaded, @@ -1050,6 +1263,20 @@ def add_issue(code: str, severity: str, message: str) -> None: state_payload["lock_holder_held_ticks"] = 0 if pause_payload: state_payload["pause_sentinel"] = pause_payload + result.duration_seconds = max(0.0, monotonic_fn() - started_monotonic) + result.slow_check = result.duration_seconds >= config.max_duration_seconds + state_payload["duration_seconds"] = result.duration_seconds + state_payload["slow_check"] = result.slow_check + if result.slow_check: + result.slow_check_stage = "finalize" + state_payload["slow_check_stage"] = result.slow_check_stage + add_issue( + "slow_check", + "critical", + f"health-check exceeded {config.max_duration_seconds:.0f}s before state write", + ) + else: + state_payload.pop("slow_check_stage", None) _write_state(config.state_path, state_payload) result.ok = not result.issues return result diff --git a/tests/test_doctor.py b/tests/test_doctor.py index 0e9f9307..9ad1ab4e 100644 --- a/tests/test_doctor.py +++ b/tests/test_doctor.py @@ -343,6 +343,26 @@ def test_run_doctor_exits_zero_on_healthy_fixture(tmp_path): assert not [issue for issue in result.issues if issue.severity == "fatal"] +def test_run_doctor_reports_unavailable_process_snapshot_without_false_daemon_death(tmp_path): + from brainlayer.doctor import run_doctor + + db_path = tmp_path / "healthy.db" + _build_db(db_path) + + result = run_doctor( + _doctor_config(tmp_path, db_path), + ps_output_fn=lambda: None, + command_runner=_loaded_launchctl, + now_fn=lambda: NOW, + ) + + issue_codes = {issue.code for issue in result.issues} + assert result.exit_code == 1 + assert result.ok is False + assert "process_snapshot_failed" in issue_codes + assert "hotlane_dead" not in issue_codes + + def test_run_doctor_can_skip_roundtrip_probe_for_bounded_status_gate(tmp_path, monkeypatch): import brainlayer.doctor as doctor diff --git a/tests/test_installable_build.py b/tests/test_installable_build.py index eed2f93e..4c1573cd 100644 --- a/tests/test_installable_build.py +++ b/tests/test_installable_build.py @@ -562,6 +562,11 @@ def test_packaged_launchd_installer_installs_tier0_watchdog_without_env_runner(t rendered = home / "Library" / "LaunchAgents" / "com.brainlayer.tier0-watchdog.plist" plist = plistlib.loads(rendered.read_bytes()) assert plist["ProgramArguments"] == ["/bin/sh", str(installed_script)] + assert plist["EnvironmentVariables"]["TIER0_STALE_SECONDS"] == "900" + assert plist["EnvironmentVariables"]["TIER0_REPEAT_ALERT_SECONDS"] == "1800" + assert plist["EnvironmentVariables"]["TIER0_ALERT_STATE_PATH"] == str( + home / ".local" / "share" / "brainlayer" / "tier0-watchdog-alert-state" + ) rendered_content = rendered.read_text(encoding="utf-8") assert "__TIER0_WATCHDOG_SCRIPT__" not in rendered_content assert "brainlayer-env-run" not in rendered_content diff --git a/tests/test_stability_health_check.py b/tests/test_stability_health_check.py index 91fefec1..775e039c 100644 --- a/tests/test_stability_health_check.py +++ b/tests/test_stability_health_check.py @@ -84,7 +84,7 @@ def _make_db(path: Path, *, total: int, vector_rows: int) -> None: enrich_status TEXT, char_count INTEGER ); - CREATE TABLE chunk_vectors_rowids (id TEXT PRIMARY KEY); + CREATE TABLE chunk_vectors_rowids (id TEXT PRIMARY KEY, chunk_id INTEGER); """ ) for index in range(total): @@ -202,6 +202,44 @@ def test_any_zero_backlog_batch_alarms_when_multiple_hotlanes_are_running(tmp_pa assert "com.brainlayer.hotlane-brainbar" in " ".join(kickstarts[0]) +def test_uninterruptible_launchd_process_backs_off_instead_of_kickstarting(tmp_path, monkeypatch): + db_path = tmp_path / "brainlayer.db" + state_path = tmp_path / "health-state.json" + _make_db(db_path, total=1, vector_rows=1) + commands: list[list[str]] = [] + events: list[dict] = [] + monkeypatch.setattr(health_check, "_emit_heal_event", events.append) + + def command_runner(args: list[str]): + commands.append(args) + if args[:2] == ["launchctl", "print"] and "hotlane-brainbar" in args[-1]: + return SimpleNamespace(returncode=0, stdout="pid = 123\n", stderr="") + if args[:3] == ["ps", "-o", "state="]: + return SimpleNamespace(returncode=0, stdout="UN\n", stderr="") + return SimpleNamespace(returncode=0, stdout="", stderr="") + + result = run_health_check( + HealthCheckConfig( + db_path=db_path, + state_path=state_path, + heal=True, + heal_min_consecutive_failures=1, + heal_circuit_breaker_limit=3, + ), + ps_output_fn=lambda: "123 /usr/bin/python scripts/hotlane_brainbar_daemon.py --interval 1 --backlog-batch 0\n", + socket_request_fn=_ok_canary, + command_runner=command_runner, + now_fn=lambda: datetime(2026, 7, 13, 9, 5, tzinfo=UTC), + ) + + assert not [command for command in commands if command[:3] == ["launchctl", "kickstart", "-k"]] + assert "heal_backoff:com.brainlayer.hotlane-brainbar:hotlane_backlog_disabled:pid=123:state=UN" in result.actions + assert any( + event.get("_type") == "heal_backoff" and event.get("pid") == 123 and event.get("process_state") == "UN" + for event in events + ) + + def test_missing_embeddings_not_draining_after_two_ticks(tmp_path): db_path = tmp_path / "brainlayer.db" state_path = tmp_path / "health-state.json" @@ -235,6 +273,159 @@ def test_missing_embeddings_not_draining_after_two_ticks(tmp_path): assert saved["stalled_ticks"] == 2 +def test_missing_embedding_query_scans_covering_id_indexes_before_chunk_payloads(tmp_path): + db_path = tmp_path / "brainlayer.db" + _make_db(db_path, total=5, vector_rows=3) + + with sqlite3.connect(db_path) as conn: + plan = conn.execute("EXPLAIN QUERY PLAN " + health_check.MISSING_EMBEDDINGS_SQL).fetchall() + + details = [str(row[-1]) for row in plan] + assert any("chunks USING COVERING INDEX" in detail for detail in details) + assert any("chunk_vectors_rowids USING COVERING INDEX" in detail for detail in details) + assert "SCAN c" not in details + + +def test_interrupted_missing_embedding_count_writes_slow_state_and_returns_early(tmp_path, monkeypatch): + db_path = tmp_path / "brainlayer.db" + state_path = tmp_path / "health-state.json" + _make_db(db_path, total=1, vector_rows=1) + canary_called = False + + def interrupted_count(_db_path, **_kwargs): + raise sqlite3.OperationalError("interrupted") + + def canary(*_args): + nonlocal canary_called + canary_called = True + return _ok_canary(*_args) + + monkeypatch.setattr(health_check, "count_missing_embeddings", interrupted_count) + + result = run_health_check( + HealthCheckConfig(db_path=db_path, state_path=state_path, max_duration_seconds=45), + ps_output_fn=lambda: "123 /usr/bin/python scripts/hotlane_brainbar_daemon.py --interval 1 --backlog-batch 4\n", + socket_request_fn=canary, + command_runner=lambda _args: SimpleNamespace(returncode=0, stdout="", stderr=""), + now_fn=lambda: datetime(2026, 7, 13, 9, 0, tzinfo=UTC), + ) + + saved = json.loads(state_path.read_text(encoding="utf-8")) + assert result.slow_check is True + assert result.slow_check_stage == "missing_embeddings" + assert saved["slow_check"] is True + assert saved["slow_check_stage"] == "missing_embeddings" + assert saved["ts"] == "2026-07-13T09:00:00+00:00" + assert canary_called is False + + +def test_missing_embedding_count_interrupts_at_internal_deadline(tmp_path, monkeypatch): + db_path = tmp_path / "brainlayer.db" + _make_db(db_path, total=1, vector_rows=0) + callbacks = [] + + class DeadlineConnection: + def execute(self, sql): + if sql == "PRAGMA query_only = ON": + return self + raise sqlite3.OperationalError("interrupted") + + def set_progress_handler(self, callback, _instructions): + callbacks.append(callback) + + def close(self): + return None + + monkeypatch.setattr(health_check.sqlite3, "connect", lambda *_args, **_kwargs: DeadlineConnection()) + + with pytest.raises(RuntimeError, match="deadline"): + health_check.count_missing_embeddings( + db_path, + deadline_at=1.0, + monotonic_fn=lambda: 2.0, + ) + + assert len(callbacks) == 1 + assert callbacks[0]() == 1 + + +def test_source_recent_returns_at_deadline_when_glob_iteration_stalls(tmp_path, monkeypatch): + import glob + import threading + import time + + release_glob = threading.Event() + + def stalled_glob(*_args, **_kwargs): + release_glob.wait() + yield str(tmp_path / "event.jsonl") + + monkeypatch.setattr(glob, "iglob", stalled_glob) + outcome: dict[str, object] = {} + + def call_source_recent() -> None: + try: + health_check._source_recent( + HealthCheckConfig(source_jsonl_globs=(str(tmp_path / "**" / "*.jsonl"),)), + datetime(2026, 7, 13, 9, 0, tzinfo=UTC), + 60, + deadline_at=time.monotonic() + 0.05, + ) + except Exception as exc: + outcome["error"] = exc + + caller = threading.Thread(target=call_source_recent) + caller.start() + caller.join(timeout=0.25) + returned_before_release = not caller.is_alive() + release_glob.set() + caller.join(timeout=1) + + assert returned_before_release is True + assert isinstance(outcome.get("error"), health_check.HealthCheckDeadlineExceeded) + + +def test_ps_snapshot_failure_does_not_heal_a_running_daemon_as_dead(tmp_path): + db_path = tmp_path / "brainlayer.db" + state_path = tmp_path / "health-state.json" + _make_db(db_path, total=1, vector_rows=1) + commands: list[list[str]] = [] + + def command_runner(args): + commands.append(args) + return SimpleNamespace(returncode=0, stdout="", stderr="") + + result = run_health_check( + HealthCheckConfig( + db_path=db_path, + state_path=state_path, + heal=True, + heal_min_consecutive_failures=1, + ), + ps_output_fn=lambda: None, + socket_request_fn=_ok_canary, + command_runner=command_runner, + now_fn=lambda: datetime(2026, 7, 13, 9, 0, tzinfo=UTC), + ) + + issue_codes = {issue.code for issue in result.issues} + assert "process_snapshot_failed" in issue_codes + assert "hotlane_dead" not in issue_codes + assert not any( + args[:3] == ["launchctl", "kickstart", "-k"] and args[-1].endswith("/com.brainlayer.hotlane") + for args in commands + ) + + +def test_default_ps_output_marks_timeout_as_unavailable(monkeypatch): + def timeout(*_args, **_kwargs): + raise health_check.subprocess.TimeoutExpired(cmd=["ps"], timeout=5) + + monkeypatch.setattr(health_check.subprocess, "run", timeout) + + assert health_check._default_ps_output() is None + + def test_lock_holder_wedge_flags_live_holder_when_drain_is_starved(tmp_path, monkeypatch): db_path = tmp_path / "brainlayer.db" state_path = tmp_path / "health-state.json" diff --git a/tests/test_tier0_drills.py b/tests/test_tier0_drills.py index f9a552ec..a52d33a2 100644 --- a/tests/test_tier0_drills.py +++ b/tests/test_tier0_drills.py @@ -25,6 +25,7 @@ class DrillResult: process: subprocess.CompletedProcess[str] events: list[str] tier0_log: str + alert_state: str def _write_executable(path: Path, contents: str) -> None: @@ -44,6 +45,10 @@ def _run_drill( curl_hangs: bool = False, osascript_hangs: bool = False, use_fake_wait_sleep: bool = False, + last_alert_epoch: int | None = None, + last_alert_reason: str = "", + repeat_alert_seconds: int = 1_800, + state_contents: str = "{}\n", ) -> DrillResult: fake_bin = tmp_path / "bin" fake_bin.mkdir() @@ -51,10 +56,17 @@ def _run_drill( state_path = tmp_path / "health-check-state.json" health_plist_path = tmp_path / "com.example.brainlayer-health-check.plist" tier0_log_path = tmp_path / "logs" / "tier0-watchdog.log" + alert_state_path = tmp_path / "tier0-watchdog-alert-state" health_plist_path.write_text("fixture\n", encoding="utf-8") + if last_alert_epoch is not None: + alert_state_path.write_text( + f"{last_alert_epoch}\t{last_alert_reason}\n", + encoding="utf-8", + ) + if state_mtime is not None: - state_path.write_text("{}\n", encoding="utf-8") + state_path.write_text(state_contents, encoding="utf-8") _write_executable( fake_bin / "launchctl", @@ -117,6 +129,7 @@ def _run_drill( "FAKE_LAUNCHCTL_PRINT_EXIT": "0" if label_loaded else "113", "FAKE_STATE_MTIME": str(state_mtime or 0), "TIER0_ALERT_TIMEOUT_SECONDS": "1", + "TIER0_ALERT_STATE_PATH": str(alert_state_path), "TIER0_CURL": str(fake_bin / "curl"), "TIER0_DOMAIN": DOMAIN, "TIER0_DRILL_EVENTS": str(events_path), @@ -126,6 +139,7 @@ def _run_drill( "TIER0_LOG_PATH": str(tier0_log_path), "TIER0_NOTIFY_ENDPOINT": NOTIFY_ENDPOINT, "TIER0_NOTIFY_TIMEOUT_SECONDS": "1", + "TIER0_REPEAT_ALERT_SECONDS": str(repeat_alert_seconds), "TIER0_NOW_EPOCH": str(NOW_EPOCH), "TIER0_OSASCRIPT": str(fake_bin / "osascript"), "TIER0_SLEEP": str(fake_bin / "wait-sleep") if use_fake_wait_sleep else "/bin/sleep", @@ -143,7 +157,8 @@ def _run_drill( ) events = events_path.read_text(encoding="utf-8").splitlines() if events_path.exists() else [] tier0_log = tier0_log_path.read_text(encoding="utf-8") if tier0_log_path.exists() else "" - return DrillResult(process=process, events=events, tier0_log=tier0_log) + alert_state = alert_state_path.read_text(encoding="utf-8") if alert_state_path.exists() else "" + return DrillResult(process=process, events=events, tier0_log=tier0_log, alert_state=alert_state) def _assert_alert_contract(result: DrillResult) -> None: @@ -188,6 +203,24 @@ def test_d2_stale_state_alerts_before_direct_kickstart(tmp_path: Path) -> None: _assert_alert_contract(result) assert not any(event.startswith("launchctl:bootstrap ") for event in result.events) assert f"state_stale age={STALE_SECONDS + 1}s threshold={STALE_SECONDS}s" in result.tier0_log + assert result.alert_state == f"{NOW_EPOCH}\tstate_stale\n" + + +def test_repeat_stale_alert_is_suppressed_during_cooldown_but_recovery_still_runs(tmp_path: Path) -> None: + result = _run_drill( + tmp_path, + label_loaded=True, + state_mtime=NOW_EPOCH - STALE_SECONDS - 1, + last_alert_epoch=NOW_EPOCH - 300, + last_alert_reason="state_stale", + ) + + assert result.process.returncode == 1 + assert not any(event.startswith("osascript:") for event in result.events) + assert not any(event.startswith("curl:") for event in result.events) + assert result.tier0_log == "" + assert any(event == f"launchctl:kickstart -k {DOMAIN}/{LABEL}" for event in result.events) + assert result.alert_state == f"{NOW_EPOCH - 300}\tstate_stale\n" def test_d3_hanging_notify_endpoint_cannot_suppress_local_alert_or_heal(tmp_path: Path) -> None: @@ -216,6 +249,19 @@ def test_d4_loaded_label_and_fresh_state_do_nothing(tmp_path: Path) -> None: assert result.tier0_log == "" +def test_fresh_state_resets_prior_incident_alert_cooldown(tmp_path: Path) -> None: + result = _run_drill( + tmp_path, + label_loaded=True, + state_mtime=NOW_EPOCH - 60, + last_alert_epoch=NOW_EPOCH - 300, + last_alert_reason="state_stale", + ) + + assert result.process.returncode == 0 + assert result.alert_state == "0\tok\n" + + def test_missing_state_alerts_and_kickstarts_without_bootstrap(tmp_path: Path) -> None: result = _run_drill(tmp_path, label_loaded=True, state_mtime=None) @@ -235,6 +281,21 @@ def test_future_state_mtime_alerts_and_kickstarts(tmp_path: Path) -> None: assert "state_mtime_future offset=60s" in result.tier0_log +def test_fresh_slow_check_state_alerts_and_kickstarts(tmp_path: Path) -> None: + result = _run_drill( + tmp_path, + label_loaded=True, + state_mtime=NOW_EPOCH - 60, + state_contents='{"slow_check": true, "slow_check_stage": "missing_embeddings"}\n', + ) + + assert result.process.returncode == 1, result.process.stdout + result.process.stderr + _assert_alert_contract(result) + assert not any(event.startswith("launchctl:bootstrap ") for event in result.events) + assert "state_slow_check" in result.tier0_log + assert result.alert_state == f"{NOW_EPOCH}\tstate_slow_check\n" + + def test_alert_fanout_uses_one_shared_deadline(tmp_path: Path) -> None: result = _run_drill( tmp_path, @@ -257,6 +318,11 @@ def test_tier0_launchagent_uses_bin_sh_without_python_wrapper() -> None: assert plist["ProgramArguments"] == ["/bin/sh", "__TIER0_WATCHDOG_SCRIPT__"] assert plist["StartInterval"] == 300 assert plist["RunAtLoad"] is True + assert plist["EnvironmentVariables"]["TIER0_STALE_SECONDS"] == "900" + assert plist["EnvironmentVariables"]["TIER0_REPEAT_ALERT_SECONDS"] == "1800" + assert plist["EnvironmentVariables"]["TIER0_ALERT_STATE_PATH"] == ( + "__HOME__/.local/share/brainlayer/tier0-watchdog-alert-state" + ) args = " ".join(plist["ProgramArguments"]) assert "ENV_RUN" not in args assert "PYTHON" not in args.upper()