diff --git a/docs/operations/brainlayer-health-check.md b/docs/operations/brainlayer-health-check.md index a3f7371a9..2c9d92788 100644 --- a/docs/operations/brainlayer-health-check.md +++ b/docs/operations/brainlayer-health-check.md @@ -14,6 +14,18 @@ brainlayer health-check --json --heal - No running hotlane command line is disabling the embedding backlog with `--backlog-batch 0`. - 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 JSONL backup attempt log has a fresh terminal receipt. A verified upload and a + verified `no-op` are healthy but remain distinct in `jsonl_backup.state`; failed, + malformed, missing, or older-than-36-hour receipts are critical issues. + +The backup process appends a timezone-aware `attempted_at` to every normal result and +to caught terminal failures. The independent five-minute health check reads that +receipt, so a backup process that never starts cannot make its own absence look +healthy. The threshold can be changed with +`BRAINLAYER_JSONL_BACKUP_MAX_AGE_SECONDS`; the monitored log follows +`BRAINLAYER_JSONL_BACKUP_LOG_PATH` when set. Legacy receipts without +`attempted_at` are interpreted at the launchd schedule of 05:00 in the producer +machine's local timezone; this fallback disappears after the next stamped run. 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. diff --git a/src/brainlayer/health_check.py b/src/brainlayer/health_check.py index ee9d3246b..854027a43 100644 --- a/src/brainlayer/health_check.py +++ b/src/brainlayer/health_check.py @@ -48,6 +48,8 @@ DEFAULT_HEAL_MIN_CONSECUTIVE_FAILURES = 2 DEFAULT_HEAL_CIRCUIT_BREAKER_LIMIT = 3 DEFAULT_MAX_DURATION_SECONDS = 45.0 +DEFAULT_JSONL_BACKUP_LOG_PATH = Path("~/.local/share/brainlayer/logs/jsonl-backup.log").expanduser() +DEFAULT_JSONL_BACKUP_MAX_AGE_SECONDS = 36 * 60 * 60 HEAL_MIN_CONSECUTIVE_FAILURES_ENV = "BRAINLAYER_HEAL_MIN_CONSECUTIVE_FAILURES" MISSING_EMBEDDINGS_SQL = """ @@ -134,6 +136,18 @@ class HealthCheckConfig: default_factory=lambda: Path("~/.local/share/brainlayer/drain-health.json").expanduser() ) t3_health_path: Path = field(default_factory=lambda: Path("~/.local/share/brainlayer/t3-health.json").expanduser()) + jsonl_backup_log_path: Path = field( + default_factory=lambda: Path( + os.environ.get("BRAINLAYER_JSONL_BACKUP_LOG_PATH", str(DEFAULT_JSONL_BACKUP_LOG_PATH)) + ).expanduser() + ) + jsonl_backup_max_age_seconds: int = field( + default_factory=lambda: _env_int( + "BRAINLAYER_JSONL_BACKUP_MAX_AGE_SECONDS", + DEFAULT_JSONL_BACKUP_MAX_AGE_SECONDS, + minimum=60 * 60, + ) + ) queue_dir: Path = field(default_factory=lambda: Path("~/.brainlayer/queue").expanduser()) pending_stores_path: Path = field( default_factory=lambda: Path("~/.local/share/brainlayer/pending-stores.jsonl").expanduser() @@ -176,6 +190,17 @@ class LockHolder: held_ticks: int = 0 +@dataclass(frozen=True) +class JsonlBackupHealth: + state: str + status: str | None = None + attempted_at: str | None = None + age_seconds: float | None = None + verified: bool | None = None + archive: str | None = None + detail: str | None = None + + @dataclass class HealthCheckResult: checked_at: str @@ -194,6 +219,7 @@ class HealthCheckResult: slow_check: bool = False slow_check_stage: str | None = None t3_health: dict[str, Any] | None = None + jsonl_backup: JsonlBackupHealth | None = None def to_dict(self) -> dict[str, Any]: return asdict(self) @@ -787,6 +813,110 @@ def _load_json(path: Path) -> dict[str, Any]: return payload if isinstance(payload, dict) else {} +def _jsonl_backup_attempt_time(payload: dict[str, Any]) -> datetime | None: + raw_attempted_at = payload.get("attempted_at") + if isinstance(raw_attempted_at, str) and raw_attempted_at: + try: + attempted_at = datetime.fromisoformat(raw_attempted_at.replace("Z", "+00:00")) + except ValueError: + return None + if attempted_at.tzinfo is None: + return None + return attempted_at.astimezone(UTC) + + archive = payload.get("archive") + if not isinstance(archive, str): + return None + legacy_date = re.search(r"claude-jsonl-(\d{4}-\d{2}-\d{2})\.tar\.gz$", archive) + if legacy_date is None: + return None + try: + # Legacy archive names carry only the UTC date, while launchd schedules + # 05:00 in the producer machine's local timezone. Interpret that date at + # the local scheduled hour; every new receipt has an exact attempted_at, + # so this machine-local compatibility path disappears after one run. + scheduled_local = datetime.fromisoformat(f"{legacy_date.group(1)}T05:00:00") + return scheduled_local.astimezone(UTC) + except ValueError: + return None + + +def inspect_jsonl_backup_health( + log_path: Path, + *, + now: datetime, + max_age_seconds: int, +) -> tuple[JsonlBackupHealth, HealthIssue | None]: + """Classify the latest durable backup attempt from an independent process.""" + resolved = log_path.expanduser() + try: + lines = [line for line in resolved.read_text(encoding="utf-8").splitlines() if line.strip()] + except FileNotFoundError: + status = JsonlBackupHealth(state="missing", detail=f"attempt log does not exist: {resolved}") + return status, HealthIssue("jsonl_backup_attempt_missing", "critical", status.detail) + except OSError as exc: + status = JsonlBackupHealth(state="invalid", detail=f"attempt log unreadable: {resolved}: {exc}") + return status, HealthIssue("jsonl_backup_attempt_invalid", "critical", status.detail) + + if not lines: + status = JsonlBackupHealth(state="missing", detail=f"attempt log is empty: {resolved}") + return status, HealthIssue("jsonl_backup_attempt_missing", "critical", status.detail) + try: + payload = json.loads(lines[-1]) + except json.JSONDecodeError as exc: + status = JsonlBackupHealth(state="invalid", detail=f"latest attempt receipt is malformed JSON: {exc}") + return status, HealthIssue("jsonl_backup_attempt_invalid", "critical", status.detail) + if not isinstance(payload, dict): + status = JsonlBackupHealth(state="invalid", detail="latest attempt receipt is not a JSON object") + return status, HealthIssue("jsonl_backup_attempt_invalid", "critical", status.detail) + + attempted_at = _jsonl_backup_attempt_time(payload) + raw_status = payload.get("status") + backup_status = raw_status if isinstance(raw_status, str) else None + verified = payload.get("verified") if isinstance(payload.get("verified"), bool) else None + archive = payload.get("archive") if isinstance(payload.get("archive"), str) else None + if attempted_at is None: + status = JsonlBackupHealth( + state="invalid", + status=backup_status, + verified=verified, + archive=archive, + detail="latest attempt receipt has no valid timezone-aware attempted_at or legacy archive date", + ) + return status, HealthIssue("jsonl_backup_attempt_invalid", "critical", status.detail) + + normalized_now = now.astimezone(UTC) if now.tzinfo is not None else now.replace(tzinfo=UTC) + age_seconds = max(0.0, (normalized_now - attempted_at).total_seconds()) + common = { + "status": backup_status, + "attempted_at": attempted_at.isoformat(), + "age_seconds": age_seconds, + "verified": verified, + "archive": archive, + } + if age_seconds > max_age_seconds: + detail = ( + f"latest JSONL backup attempt is stale: attempted_at={attempted_at.isoformat()} " + f"age_seconds={age_seconds:.0f} threshold_seconds={max_age_seconds} status={backup_status}" + ) + status = JsonlBackupHealth(state="stale", detail=detail, **common) + return status, HealthIssue("jsonl_backup_attempt_stale", "critical", detail) + + if backup_status == "no-op" and verified is True: + return JsonlBackupHealth( + state="no_op", detail=str(payload.get("message") or "legitimate no-op"), **common + ), None + if backup_status == "created" and verified is True: + return JsonlBackupHealth(state="verified_bundle", **common), None + if backup_status == "uploaded" and verified is True and payload.get("uploaded") is True: + return JsonlBackupHealth(state="verified_bundle", **common), None + + error = payload.get("error") or payload.get("verification_error") or "unverified terminal result" + detail = f"latest JSONL backup attempt failed: status={backup_status} error={error}" + status = JsonlBackupHealth(state="failed", detail=detail, **common) + return status, HealthIssue("jsonl_backup_attempt_failed", "critical", detail) + + def _t3_health_issue(payload: dict[str, Any]) -> HealthIssue | None: """Turn the T3 adapter's durable health snapshot into a check issue.""" if not payload.get("alerting"): @@ -1096,6 +1226,14 @@ def deadline_reached(stage: str) -> HealthCheckResult | None: return None return finish_slow(stage, f"health-check exceeded {config.max_duration_seconds:.0f}s during {stage}") + result.jsonl_backup, jsonl_backup_issue = inspect_jsonl_backup_health( + config.jsonl_backup_log_path, + now=now, + max_age_seconds=config.jsonl_backup_max_age_seconds, + ) + if jsonl_backup_issue is not None: + add_issue(jsonl_backup_issue.code, jsonl_backup_issue.severity, jsonl_backup_issue.message) + pause_payload, pause_active, pause_stale = _pause_sentinel_state(config, now) t3_health = _load_json(config.t3_health_path) diff --git a/src/brainlayer/jsonl_backup.py b/src/brainlayer/jsonl_backup.py index b757f37d4..28fe18b7d 100644 --- a/src/brainlayer/jsonl_backup.py +++ b/src/brainlayer/jsonl_backup.py @@ -526,6 +526,7 @@ def run_backup( ) -> dict[str, Any]: date_stamp = date_stamp or _today() now = time.time() if now is None else now + attempted_at = dt.datetime.fromtimestamp(now, dt.UTC).isoformat() roots = source_roots or DEFAULT_SOURCE_ROOTS state_path = Path(state_path).expanduser() state = _load_state(state_path) @@ -554,6 +555,7 @@ def run_backup( if not changed: result: dict[str, Any] = { + "attempted_at": attempted_at, "status": "no-op", "uploaded": False, "verified": True, @@ -570,6 +572,7 @@ def run_backup( archive_path, bundle_digests = create_jsonl_bundle_with_digests(changed, staging_dir, date_stamp=date_stamp) archive_size = archive_path.stat().st_size result = { + "attempted_at": attempted_at, "status": "uploaded" if upload else "created", "archive": str(archive_path), "bytes": archive_size, @@ -645,8 +648,17 @@ def _raise_backup_timeout(signum, frame) -> None: # noqa: ARG001 raise backup_daily.BackupTimeoutError("jsonl backup exceeded configured wall-clock timeout") +def _append_terminal_failure(log_path: Path, result: dict[str, Any]) -> None: + """Persist terminal failures without hiding the original failure if logging also breaks.""" + try: + _append_json_log(log_path, result) + except Exception as exc: + result["attempt_log_error"] = str(exc) + + def main() -> int: timeout_seconds = _configured_backup_timeout_seconds() + log_path = Path(os.environ.get("BRAINLAYER_JSONL_BACKUP_LOG_PATH", str(DEFAULT_LOG_PATH))) previous_alarm_handler = None if timeout_seconds is not None: previous_alarm_handler = signal.getsignal(signal.SIGALRM) @@ -656,28 +668,32 @@ def main() -> int: result = run_backup( staging_dir=Path(os.environ.get("BRAINLAYER_JSONL_BACKUP_STAGING_DIR", str(DEFAULT_STAGING_DIR))), state_path=Path(os.environ.get("BRAINLAYER_JSONL_BACKUP_STATE_PATH", str(DEFAULT_STATE_PATH))), - log_path=Path(os.environ.get("BRAINLAYER_JSONL_BACKUP_LOG_PATH", str(DEFAULT_LOG_PATH))), + log_path=log_path, folder_parts=os.environ.get("BRAINLAYER_JSONL_BACKUP_DRIVE_FOLDER", "/".join(DEFAULT_FOLDER_PARTS)).split( "/" ), ) except backup_daily.BackupTimeoutError: result = { + "attempted_at": dt.datetime.now(dt.UTC).isoformat(), "status": "failed", "uploaded": False, "verified": False, "error": f"timed out after {timeout_seconds}s", } + _append_terminal_failure(log_path, result) print(json.dumps(result, sort_keys=True), flush=True) return 124 except Exception as exc: result = { + "attempted_at": dt.datetime.now(dt.UTC).isoformat(), "status": "failed", "uploaded": False, "verified": False, "error": str(exc), "traceback": traceback.format_exc(), } + _append_terminal_failure(log_path, result) print(json.dumps(result, sort_keys=True), flush=True) return 1 finally: diff --git a/tests/test_jsonl_backup.py b/tests/test_jsonl_backup.py index 72df1d63f..ccab4ff1f 100644 --- a/tests/test_jsonl_backup.py +++ b/tests/test_jsonl_backup.py @@ -524,6 +524,8 @@ def _upload(file_path, folder_id, credentials): assert first["status"] == "uploaded" assert second["status"] == "no-op" + assert first["attempted_at"] == second["attempted_at"] + assert second["attempted_at"].endswith("+00:00") assert second["uploaded"] is False assert second["already_covered_files"] == 1 assert second["message"] == "no-op, 1 files already covered" @@ -647,6 +649,28 @@ def fake_run_backup(**kwargs): # noqa: ARG001 assert payload["verified"] is False +def test_jsonl_backup_main_persists_terminal_failure_to_attempt_log(tmp_path, monkeypatch, capsys): + from brainlayer import jsonl_backup + + log_path = tmp_path / "jsonl-backup.log" + monkeypatch.setenv("BRAINLAYER_JSONL_BACKUP_LOG_PATH", str(log_path)) + monkeypatch.setattr(jsonl_backup, "_configured_backup_timeout_seconds", lambda: None) + monkeypatch.setattr( + jsonl_backup, + "run_backup", + lambda **_kwargs: (_ for _ in ()).throw(RuntimeError("upload exploded")), + ) + + assert jsonl_backup.main() == 1 + + stdout_payload = json.loads(capsys.readouterr().out) + logged_payload = json.loads(log_path.read_text(encoding="utf-8")) + assert logged_payload == stdout_payload + assert logged_payload["status"] == "failed" + assert logged_payload["verified"] is False + assert logged_payload["attempted_at"].endswith("+00:00") + + def test_jsonl_backup_launchd_plist_and_docstring_install_note_are_committed(): module_path = Path("src/brainlayer/jsonl_backup.py") plist_path = Path("launchd/com.brainlayer.jsonl-backup.plist") diff --git a/tests/test_stability_health_check.py b/tests/test_stability_health_check.py index 4b73c148e..57c49ff50 100644 --- a/tests/test_stability_health_check.py +++ b/tests/test_stability_health_check.py @@ -67,6 +67,26 @@ def _isolated_pause_state(config, now): return real_pause_state(config, now) monkeypatch.setattr(health_check, "_pause_sentinel_state", _isolated_pause_state) + + live_backup_log_default = Path("~/.local/share/brainlayer/logs/jsonl-backup.log").expanduser() + real_backup_health = health_check.inspect_jsonl_backup_health + + def _isolated_backup_health(log_path, *, now, max_age_seconds): + if log_path.expanduser() == live_backup_log_default: + return ( + health_check.JsonlBackupHealth( + state="no_op", + status="no-op", + attempted_at=now.isoformat(), + age_seconds=0.0, + verified=True, + detail="isolated test no-op", + ), + None, + ) + return real_backup_health(log_path, now=now, max_age_seconds=max_age_seconds) + + monkeypatch.setattr(health_check, "inspect_jsonl_backup_health", _isolated_backup_health) yield @@ -167,6 +187,154 @@ def test_health_check_consumes_alerting_t3_health_snapshot(tmp_path): assert "schema_drift" in issue.message +@pytest.mark.parametrize( + ("payload", "expected_state", "expected_issue"), + [ + ( + { + "attempted_at": "2026-09-09T02:00:00+00:00", + "status": "uploaded", + "uploaded": True, + "verified": True, + "archive": "claude-jsonl-2026-09-09.tar.gz", + }, + "verified_bundle", + None, + ), + ( + { + "status": "uploaded", + "uploaded": True, + "verified": True, + "archive": "claude-jsonl-2026-09-09.tar.gz", + }, + "verified_bundle", + None, + ), + ( + { + "attempted_at": "2026-09-09T02:00:00+00:00", + "status": "no-op", + "uploaded": False, + "verified": True, + "message": "no-op, 10 files already covered", + }, + "no_op", + None, + ), + ( + { + "attempted_at": "2026-09-09T02:00:00+00:00", + "status": "failed", + "uploaded": False, + "verified": False, + "error": "upload exploded", + }, + "failed", + "jsonl_backup_attempt_failed", + ), + ], +) +def test_jsonl_backup_detector_distinguishes_verified_noop_and_failure( + tmp_path, payload, expected_state, expected_issue +): + log_path = tmp_path / "jsonl-backup.log" + log_path.write_text(json.dumps(payload) + "\n", encoding="utf-8") + + status, issue = health_check.inspect_jsonl_backup_health( + log_path, + now=datetime(2026, 9, 9, 12, 0, tzinfo=UTC), + max_age_seconds=36 * 60 * 60, + ) + + assert status.state == expected_state + assert status.status == payload["status"] + assert (issue.code if issue else None) == expected_issue + + +def test_jsonl_backup_detector_reports_absent_malformed_and_stale_attempts(tmp_path): + log_path = tmp_path / "jsonl-backup.log" + now = datetime(2026, 9, 9, 12, 0, tzinfo=UTC) + + missing, missing_issue = health_check.inspect_jsonl_backup_health( + log_path, + now=now, + max_age_seconds=36 * 60 * 60, + ) + assert missing.state == "missing" + assert missing_issue.code == "jsonl_backup_attempt_missing" + + log_path.write_text("not-json\n", encoding="utf-8") + malformed, malformed_issue = health_check.inspect_jsonl_backup_health( + log_path, + now=now, + max_age_seconds=36 * 60 * 60, + ) + assert malformed.state == "invalid" + assert malformed_issue.code == "jsonl_backup_attempt_invalid" + + log_path.write_text( + json.dumps( + { + "attempted_at": "2026-09-07T02:00:00+00:00", + "status": "uploaded", + "uploaded": True, + "verified": True, + "archive": "claude-jsonl-2026-09-07.tar.gz", + } + ) + + "\n", + encoding="utf-8", + ) + stale, stale_issue = health_check.inspect_jsonl_backup_health( + log_path, + now=now, + max_age_seconds=36 * 60 * 60, + ) + assert stale.state == "stale" + assert stale_issue.code == "jsonl_backup_attempt_stale" + + +def test_run_health_check_surfaces_jsonl_backup_failure(tmp_path): + db_path = tmp_path / "brainlayer.db" + log_path = tmp_path / "jsonl-backup.log" + _make_db(db_path, total=1, vector_rows=1) + log_path.write_text( + json.dumps( + { + "attempted_at": "2026-09-09T02:00:00+00:00", + "status": "failed", + "uploaded": False, + "verified": False, + "error": "upload exploded", + } + ) + + "\n", + encoding="utf-8", + ) + + result = run_health_check( + HealthCheckConfig( + db_path=db_path, + state_path=tmp_path / "health-state.json", + jsonl_backup_log_path=log_path, + source_jsonl_globs=[], + queue_dir=tmp_path / "queue", + pending_stores_path=tmp_path / "pending-stores.jsonl", + watcher_health_path=tmp_path / "watcher-health.json", + drain_health_path=tmp_path / "drain-health.json", + ), + ps_output_fn=lambda: "123 /usr/bin/python scripts/hotlane_brainbar_daemon.py --interval 1 --backlog-batch 4\n", + socket_request_fn=_ok_canary, + command_runner=lambda _args: SimpleNamespace(returncode=0, stdout="", stderr=""), + now_fn=lambda: datetime(2026, 9, 9, 12, 0, tzinfo=UTC), + ) + + assert result.jsonl_backup is not None + assert result.jsonl_backup.state == "failed" + assert "jsonl_backup_attempt_failed" in [issue.code for issue in result.issues] + + def test_backlog_batch_zero_alarms_but_waits_until_repeated_failure_to_kickstart_hotlane(tmp_path, capsys): db_path = tmp_path / "brainlayer.db" state_path = tmp_path / "health-state.json"