fix(backup): detect missing daily attempts

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.

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)

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:

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")