diff --git a/launchd/com.brainlayer.jsonl-backup.plist b/launchd/com.brainlayer.jsonl-backup.plist
index 8ca8d008..b15baff1 100644
--- a/launchd/com.brainlayer.jsonl-backup.plist
+++ b/launchd/com.brainlayer.jsonl-backup.plist
@@ -10,8 +10,9 @@
ProgramArguments
- /usr/bin/env
- python3
+
+ __BRAINLAYER_PYTHON__
-m
brainlayer.jsonl_backup
@@ -30,8 +31,6 @@
PATH
/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin:__HOME__/.local/bin
- PYTHONPATH
- __BRAINLAYER_DIR__/src
BRAINLAYER_BACKUP_TIMEOUT_SECONDS
1800
BRAINLAYER_JSONL_BACKUP_DRIVE_FOLDER
diff --git a/scripts/launchd/install.sh b/scripts/launchd/install.sh
index 8d25cbaa..4bf1ac9d 100755
--- a/scripts/launchd/install.sh
+++ b/scripts/launchd/install.sh
@@ -138,11 +138,45 @@ fi
# PATH happens to front. On the M4 that is the framework Python's `brainlayer`, and installing from a
# source checkout fronts `~/Gits/brainlayer/.venv/bin/brainlayer` -- neither of which any release moves.
BRAINLAYER_BIN="$(stable_brainlayer_path "${BRAINLAYER_BIN:-${BRAINLAYER_KEG_CLI:-$(which brainlayer 2>/dev/null || echo "$HOME/.local/bin/brainlayer")}}")"
+# Capture only caller intent before PYTHON_BIN receives its legacy PATH fallback.
+BRAINLAYER_PYTHON_REQUESTED="${BRAINLAYER_PYTHON:-}"
# In a keg, an unset PYTHON_BIN must NOT fall through to `command -v python3`: on a Mac whose PATH
# puts /Library/Frameworks/Python.framework first, that renders a framework interpreter that never
# sees the keg's site-packages, and no release can move it. An explicit override still wins.
PYTHON_BIN="$(stable_brainlayer_path "${PYTHON_BIN:-${BRAINLAYER_KEG_PYTHON:-$(command -v python3)}}")"
+# The backup wrappers import BrainLayer itself, so a source install must not inherit
+# `command -v python3`: on the M4 that is the framework interpreter whose global
+# .pth injects the mutable root checkout. Explicit overrides remain explicit; a keg
+# uses its stable opt/ path; otherwise reuse hook_python's ARM/Intel-aware resolver
+# and fail closed when neither Homebrew prefix exists.
BRAINLAYER_PYTHON="$(stable_brainlayer_path "${BRAINLAYER_PYTHON:-$PYTHON_BIN}")"
+
+resolve_jsonl_backup_python() {
+ if [ -n "$BRAINLAYER_KEG_PYTHON" ] && [ -z "$BRAINLAYER_PYTHON_REQUESTED" ] && [ -z "${BRAINLAYER_HOOK_PYTHON:-}" ]; then
+ BRAINLAYER_PYTHON="$(stable_brainlayer_path "$BRAINLAYER_KEG_PYTHON")"
+ return 0
+ fi
+
+ HOOK_PYTHON_RESOLVER="$BRAINLAYER_DIR/src/brainlayer/hook_python.py"
+ if [ ! -f "$HOOK_PYTHON_RESOLVER" ]; then
+ HOOK_PYTHON_RESOLVER="$BRAINLAYER_DIR/brainlayer/hook_python.py"
+ fi
+ if [ ! -f "$HOOK_PYTHON_RESOLVER" ]; then
+ echo "ERROR: hook_python.py not found; refusing a PATH-derived BrainLayer interpreter" >&2
+ return 1
+ fi
+
+ # `python hook_python.py` would put brainlayer/ itself on sys.path, where
+ # brainlayer/types.py shadows the stdlib `types` module under Apple's Python.
+ # run_path keeps the resolver executable as a standalone stdlib-only script
+ # without adding its package directory to import resolution.
+ if [ -n "$BRAINLAYER_PYTHON_REQUESTED" ]; then
+ BRAINLAYER_PYTHON="$(BRAINLAYER_HOOK_PYTHON="$BRAINLAYER_PYTHON_REQUESTED" /usr/bin/python3 -c 'import runpy, sys; path = sys.argv.pop(1); runpy.run_path(path, run_name="__main__")' "$HOOK_PYTHON_RESOLVER" --print-interpreter)" || return 1
+ else
+ BRAINLAYER_PYTHON="$(/usr/bin/python3 -c 'import runpy, sys; path = sys.argv.pop(1); runpy.run_path(path, run_name="__main__")' "$HOOK_PYTHON_RESOLVER" --print-interpreter)" || return 1
+ fi
+ BRAINLAYER_PYTHON="$(stable_brainlayer_path "$BRAINLAYER_PYTHON")"
+}
BRAINLAYER_ENV_FILE="${BRAINLAYER_ENV_FILE:-$HOME/.config/brainlayer/brainlayer.env}"
BRAINLAYER_ENV_RUN="$BRAINLAYER_LIB_DIR/brainlayer-env-run.sh"
TIER0_WATCHDOG_DST="$BRAINLAYER_LIB_DIR/tier0-watchdog.sh"
@@ -622,6 +656,13 @@ install_plist() {
verify_gemini_env_file || return 1
fi
+ # XML-escape the interpreter path, then escape sed replacement metacharacters.
+ # `&` is legal in a filename but means "the matched placeholder" to sed.
+ local brainlayer_python_xml
+ local brainlayer_python_sed
+ brainlayer_python_xml="$(printf '%s' "$BRAINLAYER_PYTHON" | sed -e 's/&/\&/g' -e 's//\>/g')" || return 1
+ brainlayer_python_sed="$(printf '%s' "$brainlayer_python_xml" | sed -e 's/[\\&|]/\\&/g')" || return 1
+
# Replace placeholders
sed \
-e "s|__HOME__|$HOME|g" \
@@ -629,7 +670,7 @@ install_plist() {
-e "s|__BRAINLAYER_DIR__|$BRAINLAYER_DIR|g" \
-e "s|__BRAINLAYER_LAUNCHD_DIR__|$BRAINLAYER_LAUNCHD_DIR|g" \
-e "s|__PYTHON_BIN__|$PYTHON_BIN|g" \
- -e "s|__BRAINLAYER_PYTHON__|$BRAINLAYER_PYTHON|g" \
+ -e "s|__BRAINLAYER_PYTHON__|$brainlayer_python_sed|g" \
-e "s|__REPO_ROOT__|$BRAINLAYER_DIR|g" \
-e "s|__BRAINLAYER_ENV_FILE__|$BRAINLAYER_ENV_FILE|g" \
-e "s|__BRAINLAYER_ENV_RUN__|$BRAINLAYER_ENV_RUN|g" \
@@ -692,6 +733,8 @@ install_jsonl_backup_script() {
return 1
fi
+ resolve_jsonl_backup_python || return 1
+
escaped_brainlayer_dir="$(printf '%s' "$BRAINLAYER_DIR" | sed 's/[\\&|]/\\&/g')" || return 1
sed \
-e "s|__BRAINLAYER_DIR_VALUE__|$escaped_brainlayer_dir|g" \
diff --git a/scripts/launchd/jsonl-backup.sh b/scripts/launchd/jsonl-backup.sh
index 5a824561..baf665fb 100755
--- a/scripts/launchd/jsonl-backup.sh
+++ b/scripts/launchd/jsonl-backup.sh
@@ -5,10 +5,7 @@ export PATH="/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin:$HOME/.local/bin"
export PYTHONUNBUFFERED=1
: "${BRAINLAYER_BACKUP_TIMEOUT_SECONDS:=1800}"
export BRAINLAYER_BACKUP_TIMEOUT_SECONDS
-BRAINLAYER_DIR="${BRAINLAYER_DIR:-__BRAINLAYER_DIR_VALUE__}"
-case "$BRAINLAYER_DIR" in
- __BRAINLAYER_DIR_*) BRAINLAYER_DIR="$HOME/Gits/brainlayer" ;;
-esac
-export PYTHONPATH="$BRAINLAYER_DIR/src${PYTHONPATH:+:$PYTHONPATH}"
+: "${BRAINLAYER_PYTHON:?installer must render the prefix-aware keg interpreter}"
+unset PYTHONPATH
-exec "${BRAINLAYER_PYTHON:-python3}" -m brainlayer.jsonl_backup
+exec "$BRAINLAYER_PYTHON" -m brainlayer.jsonl_backup
diff --git a/src/brainlayer/hook_python.py b/src/brainlayer/hook_python.py
index cfbe3a4f..101bd8fd 100644
--- a/src/brainlayer/hook_python.py
+++ b/src/brainlayer/hook_python.py
@@ -23,6 +23,7 @@
import shlex
from dataclasses import dataclass
from typing import Iterable, Iterator, Mapping, Sequence
+from xml.sax.saxutils import escape
__all__ = [
"BRAINLAYER_HOOK_SCRIPTS",
@@ -36,6 +37,7 @@
"is_system_python",
"main",
"render_hook_command",
+ "render_launchd_plist",
"resolve_hook_python",
"shebang_of",
]
@@ -54,6 +56,8 @@
"/usr/local/opt/brainlayer/libexec/venv/bin/python",
)
+_XML_10_FORBIDDEN = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f\ud800-\udfff\ufffe\uffff]")
+
#: Hook scripts this repo owns. The settings.json lint matches on these basenames so
#: it never touches a hook belonging to another repo. `tests/test_hook_python.py`
#: asserts every shebang-bearing file under `hooks/` appears here.
@@ -115,7 +119,10 @@ def _tokens(value: str | None) -> list[str]:
token = value.strip()
if token.startswith("#!"):
token = token[2:].strip()
- return token.split()
+ try:
+ return shlex.split(token)
+ except ValueError:
+ return []
def is_bare_python3(value: str | None) -> bool:
@@ -245,10 +252,16 @@ def resolve_hook_python(
"substitute another interpreter for an override that was set on purpose — "
f"fix the path or unset {HOOK_PYTHON_ENV} to use the keg."
)
+ if (
+ not is_pinned_interpreter(shlex.quote(override))
+ or not os.path.isfile(override)
+ or not os.access(override, os.X_OK)
+ ):
+ raise HookPythonUnresolved(f"{HOOK_PYTHON_ENV}={override!r} is not an executable Python interpreter")
return override
for candidate in candidates:
- if os.path.exists(candidate):
+ if os.path.isfile(candidate) and os.access(candidate, os.X_OK):
return candidate
looked_at.append(candidate)
@@ -271,6 +284,30 @@ def render_hook_command(
return f"{shlex.quote(interpreter) if ' ' in interpreter else interpreter} {script_path}"
+def render_launchd_plist(
+ template: str,
+ *,
+ python: str | None = None,
+ env: Mapping[str, str] | None = None,
+) -> str:
+ """Render the prefix-aware keg interpreter into a launchd template.
+
+ Templates stay portable across ARM and Intel Homebrew prefixes. Resolution
+ uses the same fail-closed candidate order and override rules as hook command
+ rendering; a caller-supplied interpreter is accepted only when the existing
+ affirmative pin gate can vouch for it.
+ """
+ interpreter = python or resolve_hook_python(env=env)
+ if (
+ not is_pinned_interpreter(shlex.quote(interpreter))
+ or not os.path.isfile(interpreter)
+ or not os.access(interpreter, os.X_OK)
+ or _XML_10_FORBIDDEN.search(interpreter)
+ ):
+ raise HookPythonUnresolved(f"launchd interpreter is not explicitly pinned: {interpreter!r}")
+ return template.replace("__BRAINLAYER_PYTHON__", escape(interpreter))
+
+
def _iter_hook_entries(settings: Mapping) -> Iterator[tuple[str, str]]:
"""Yield `(event, command)` for every command hook configured in `settings`."""
hooks = settings.get("hooks") if isinstance(settings, Mapping) else None
@@ -381,7 +418,7 @@ def _why_unpinned(interpreter: str) -> str:
def main(argv: Sequence[str] | None = None) -> int:
- """`python -m brainlayer.hook_python [settings.json]` — lint a settings file.
+ """Lint settings, or print the prefix-aware interpreter for installers.
Exits 0 when every BrainLayer hook names its interpreter, 1 when any is
PATH-resolved, 2 when the file cannot be read. Hooks owned by other repos are
@@ -389,8 +426,14 @@ def main(argv: Sequence[str] | None = None) -> int:
"""
import argparse
import json
+ import sys
parser = argparse.ArgumentParser(prog="brainlayer.hook_python")
+ parser.add_argument(
+ "--print-interpreter",
+ action="store_true",
+ help="print the affirmative prefix-aware interpreter and exit",
+ )
parser.add_argument(
"settings",
nargs="?",
@@ -399,6 +442,14 @@ def main(argv: Sequence[str] | None = None) -> int:
)
args = parser.parse_args(argv)
+ if args.print_interpreter:
+ try:
+ print(resolve_hook_python())
+ except HookPythonUnresolved as exc:
+ print(f"cannot resolve BrainLayer interpreter: {exc}", file=sys.stderr, flush=True)
+ return 2
+ return 0
+
try:
with open(args.settings, encoding="utf-8") as handle:
settings = json.load(handle)
diff --git a/src/brainlayer/jsonl_backup.py b/src/brainlayer/jsonl_backup.py
index 077d7abb..789c5f5f 100644
--- a/src/brainlayer/jsonl_backup.py
+++ b/src/brainlayer/jsonl_backup.py
@@ -1,8 +1,9 @@
-"""Nightly JSONL transcript backups to Google Drive.
+"""Nightly JSONL transcript backups to Google Drive and iCloud Drive.
-Install note: commit `launchd/com.brainlayer.jsonl-backup.plist`, then install it
-after merge with the repo's launchd flow or a manual `launchctl bootstrap`; this
-module intentionally does not install the agent itself.
+Install note: commit `launchd/com.brainlayer.jsonl-backup.plist`, render its
+`__BRAINLAYER_PYTHON__` placeholder through `hook_python.render_launchd_plist`,
+then install it after merge with the repo's launchd flow; this module
+intentionally does not install the agent itself.
Source format policy: Claude/Codex/Cursor/Gemini JSONL files are backed up as
plain transcript files. Antigravity has no stable text export contract, so this
@@ -16,6 +17,7 @@
import datetime as dt
import fcntl
import functools
+import gzip
import hashlib
import json
import os
@@ -26,6 +28,7 @@
import tempfile
import time
import traceback
+import uuid
from dataclasses import dataclass
from pathlib import Path
from typing import Any
@@ -38,8 +41,14 @@
DEFAULT_STATE_PATH = Path.home() / ".local" / "share" / "brainlayer" / "jsonl-backup-state.json"
DEFAULT_STAGING_DIR = Path.home() / ".local" / "share" / "brainlayer" / "jsonl-backups"
DEFAULT_LOG_PATH = Path.home() / ".local" / "share" / "brainlayer" / "logs" / "jsonl-backup.log"
+DEFAULT_ICLOUD_DIR = (
+ Path.home() / "Library" / "Mobile Documents" / "com~apple~CloudDocs" / "Archives" / "brainlayer-jsonl-backups"
+)
+# A distinct sibling of golems' reserved path avoids shared naming/pruning ownership.
DEFAULT_ACTIVE_SKIP_SECONDS = 10 * 60
DEFAULT_TIMEOUT_SECONDS = 1800
+DEFAULT_ICLOUD_TIMEOUT_SECONDS = 300
+ICLOUD_DIR_ENV = "BRAINLAYER_JSONL_BACKUP_ICLOUD_DIR"
JSONL_RETENTION = backup_daily.DriveRetentionPolicy(
keep_latest=30,
filename_prefix="claude-jsonl-",
@@ -125,6 +134,12 @@ def _configured_backup_timeout_seconds() -> int | None:
return seconds if seconds > 0 else None
+def _configured_icloud_dir() -> Path | None:
+ """Return the opt-in iCloud destination; Drive-only is the default."""
+ raw = os.environ.get(ICLOUD_DIR_ENV, "").strip()
+ return Path(raw).expanduser() if raw else None
+
+
def _load_state(path: Path) -> dict[str, Any]:
path = Path(path).expanduser()
if not path.exists():
@@ -317,14 +332,480 @@ def _forever_enabled() -> bool:
return os.environ.get("BRAINLAYER_JSONL_FOREVER", "").strip().lower() in {"1", "true", "yes", "on"}
-def _sha256_file(path: Path) -> str:
+class ICloudDeadlineExceeded(RuntimeError):
+ """The local iCloud operation budget expired without disproving an object."""
+
+
+class ICloudProbeError(RuntimeError):
+ """Foundation state could not be read, so the object was not disproved."""
+
+
+def _check_icloud_deadline(deadline: float | None, phase: str) -> None:
+ if deadline is not None and time.monotonic() >= deadline:
+ raise ICloudDeadlineExceeded(f"iCloud operation deadline exceeded during {phase}")
+
+
+def _sha256_file(path: Path, *, deadline: float | None = None) -> str:
digest = hashlib.sha256()
with path.open("rb") as handle:
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+ _check_icloud_deadline(deadline, f"hashing {path}")
digest.update(chunk)
+ _check_icloud_deadline(deadline, f"hashing {path}")
+ return digest.hexdigest()
+
+
+def _sha256_gzip_payload(path: Path, *, deadline: float | None = None) -> str:
+ """Address valid gzip archives by their logical payload, not header metadata.
+
+ Retrying a bundle can change the gzip header timestamp while preserving the
+ tar payload. A payload-derived destination therefore reuses the same iCloud
+ object after a later Drive failure instead of accumulating untracked copies.
+ Non-gzip inputs retain byte-addressed behavior for compatibility.
+ """
+ digest = hashlib.sha256()
+ try:
+ with gzip.open(path, "rb") as handle:
+ for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+ _check_icloud_deadline(deadline, f"hashing gzip payload {path}")
+ digest.update(chunk)
+ except backup_daily.BackupTimeoutError:
+ raise
+ except (gzip.BadGzipFile, EOFError, OSError):
+ return _sha256_file(path, deadline=deadline)
+ _check_icloud_deadline(deadline, f"hashing gzip payload {path}")
return digest.hexdigest()
+_ICLOUD_STATUS_SCRIPT = r"""
+ObjC.import("Foundation");
+const args = $.NSProcessInfo.processInfo.arguments;
+const action = ObjC.unwrap(args.objectAtIndex(args.count - 2));
+const path = ObjC.unwrap(args.lastObject);
+const url = $.NSURL.fileURLWithPath(path);
+function resourceValue(key) {
+ const value = Ref();
+ const error = Ref();
+ if (!url.getResourceValueForKeyError(value, key, error)) {
+ const detail = error[0] ? ObjC.unwrap(error[0].localizedDescription) : "unknown error";
+ throw new Error(detail);
+ }
+ if (value[0] === undefined || value[0] === null) return null;
+ const unwrapped = ObjC.unwrap(value[0]);
+ return unwrapped === undefined ? null : unwrapped;
+}
+if (action === "download") {
+ const error = Ref();
+ if (!$.NSFileManager.defaultManager.startDownloadingUbiquitousItemAtURLError(url, error)) {
+ const detail = error[0] ? ObjC.unwrap(error[0].localizedDescription) : "unknown error";
+ throw new Error(detail);
+ }
+}
+const uploadError = resourceValue($.NSURLUbiquitousItemUploadingErrorKey);
+JSON.stringify({
+ is_ubiquitous: resourceValue($.NSURLIsUbiquitousItemKey),
+ is_uploaded: resourceValue($.NSURLUbiquitousItemIsUploadedKey),
+ is_uploading: resourceValue($.NSURLUbiquitousItemIsUploadingKey),
+ downloading_status: resourceValue($.NSURLUbiquitousItemDownloadingStatusKey),
+ uploading_error: uploadError === null ? null : String(uploadError)
+});
+"""
+
+
+def _normalized_icloud_download_status(value: Any) -> str | None:
+ if value is None:
+ return None
+ text = str(value)
+ prefix = "NSURLUbiquitousItemDownloadingStatus"
+ if text.startswith(prefix):
+ text = text[len(prefix) :]
+ return text[:1].lower() + text[1:] if text else text
+
+
+def _quarantine_unverified_icloud_item(path: Path) -> None:
+ """Hide an unverified iCloud item without deleting personal backup data."""
+ if not path.exists():
+ return
+ hidden_name = path.name if path.name.startswith(".") else f".{path.name}"
+ os.replace(path, path.with_name(f"{hidden_name}.{uuid.uuid4().hex}.unverified"))
+
+
+def _icloud_item_state(
+ path: Path,
+ *,
+ request_download: bool = False,
+ timeout_seconds: float | None = None,
+) -> dict[str, Any]:
+ """Read authoritative iCloud state, optionally forcing cloud materialization."""
+ try:
+ completed = subprocess.run(
+ [
+ "/usr/bin/osascript",
+ "-l",
+ "JavaScript",
+ "-e",
+ _ICLOUD_STATUS_SCRIPT,
+ "--",
+ "download" if request_download else "status",
+ str(path),
+ ],
+ check=True,
+ capture_output=True,
+ text=True,
+ timeout=timeout_seconds,
+ )
+ except backup_daily.BackupTimeoutError:
+ raise
+ except subprocess.CalledProcessError as exc:
+ detail = (exc.stderr or "").strip() or str(exc)
+ raise ICloudProbeError(f"iCloud status probe failed for {path}: {detail}") from exc
+ except OSError as exc:
+ raise ICloudProbeError(f"iCloud status probe could not run for {path}: {exc}") from exc
+ try:
+ state = json.loads(completed.stdout)
+ except json.JSONDecodeError as exc:
+ raise ICloudProbeError(f"iCloud status returned invalid JSON: {completed.stdout!r}") from exc
+ if not isinstance(state, dict):
+ raise ICloudProbeError(f"iCloud status returned a non-object: {state!r}")
+ state["downloading_status"] = _normalized_icloud_download_status(state.get("downloading_status"))
+ return state
+
+
+def copy_archive_to_icloud(
+ archive_path: Path,
+ icloud_dir: Path,
+ *,
+ timeout_seconds: float = DEFAULT_ICLOUD_TIMEOUT_SECONDS,
+ poll_interval_seconds: float = 2.0,
+ deadline: float | None = None,
+) -> dict[str, Any]:
+ """Copy to iCloud, force materialization, then verify authoritative state and bytes."""
+ archive_path = Path(archive_path).expanduser()
+ icloud_dir = Path(icloud_dir).expanduser()
+ deadline = deadline if deadline is not None else time.monotonic() + timeout_seconds
+
+ def remaining_seconds() -> float:
+ return max(deadline - time.monotonic(), 0.001)
+
+ icloud_dir.mkdir(parents=True, exist_ok=True)
+ _check_icloud_deadline(deadline, "preparing the iCloud directory")
+ expected_size = archive_path.stat().st_size
+ expected_sha256 = _sha256_file(archive_path, deadline=deadline)
+ logical_sha256 = _sha256_gzip_payload(archive_path, deadline=deadline)
+ suffix = "".join(archive_path.suffixes)
+ destination = icloud_dir / f"claude-jsonl-{logical_sha256}{suffix}"
+ placeholder = destination.with_name(f".{destination.name}.icloud")
+
+ def receipt(*, reused: bool) -> dict[str, Any]:
+ _check_icloud_deadline(deadline, "building the iCloud receipt")
+ actual_size = destination.stat().st_size
+ actual_sha256 = _sha256_file(destination, deadline=deadline)
+ return {
+ "path": str(destination),
+ "uploaded": True,
+ "materialization": "MATERIALIZED",
+ "bytes": actual_size,
+ "sha256": actual_sha256,
+ "source_sha256": expected_sha256,
+ "logical_sha256": logical_sha256,
+ "downloading_status": "current",
+ "reused": reused,
+ }
+
+ # A logical address is immutable. If a prior verified copy already occupies
+ # it, materialize and validate that object rather than replacing it before a
+ # retry is proven. Gzip header metadata may differ while the tar payload is
+ # identical, which is exactly what the logical address represents.
+ if destination.exists() or placeholder.exists():
+ try:
+ status_path = placeholder if not destination.exists() and placeholder.exists() else destination
+ state = _icloud_item_state(
+ status_path,
+ request_download=True,
+ timeout_seconds=remaining_seconds(),
+ )
+ while True:
+ uploading_error = state.get("uploading_error")
+ if uploading_error:
+ raise RuntimeError(f"existing iCloud upload failed for {destination}: {uploading_error}")
+ uploaded = state.get("is_uploaded") is True and state.get("is_uploading") is False
+ materialized = state.get("downloading_status") == "current" and destination.is_file()
+ if state.get("is_ubiquitous") is True and uploaded and materialized:
+ actual_logical_sha256 = _sha256_gzip_payload(destination, deadline=deadline)
+ if actual_logical_sha256 != logical_sha256:
+ raise RuntimeError(
+ "iCloud logical-address collision: "
+ f"path={destination} expected={logical_sha256} actual={actual_logical_sha256}"
+ )
+ return receipt(reused=True)
+ if time.monotonic() >= deadline:
+ raise ICloudDeadlineExceeded(
+ f"existing iCloud copy was not uploaded and materialized within {timeout_seconds}s: "
+ f"path={destination} state={state!r}"
+ )
+ time.sleep(min(poll_interval_seconds, remaining_seconds()))
+ status_path = placeholder if not destination.exists() and placeholder.exists() else destination
+ state = _icloud_item_state(
+ status_path,
+ request_download=True,
+ timeout_seconds=remaining_seconds(),
+ )
+ except Exception as exc:
+ # This path was not proven usable. Preserve it under a unique hidden
+ # name, then let the normal fresh-copy path repair the logical address.
+ if isinstance(
+ exc,
+ (
+ backup_daily.BackupTimeoutError,
+ ICloudDeadlineExceeded,
+ ICloudProbeError,
+ subprocess.TimeoutExpired,
+ ),
+ ):
+ raise
+ _quarantine_unverified_icloud_item(destination)
+ _quarantine_unverified_icloud_item(placeholder)
+ if time.monotonic() >= deadline:
+ raise
+
+ temp_path = icloud_dir / f".{destination.name}.{os.getpid()}.partial"
+ try:
+ with archive_path.open("rb") as source, temp_path.open("xb") as destination_handle:
+ while chunk := source.read(1024 * 1024):
+ _check_icloud_deadline(deadline, f"copying {archive_path} to iCloud")
+ destination_handle.write(chunk)
+ _check_icloud_deadline(deadline, f"copying {archive_path} to iCloud")
+ destination_handle.flush()
+ os.fsync(destination_handle.fileno())
+ _check_icloud_deadline(deadline, "publishing the iCloud copy")
+ os.replace(temp_path, destination)
+ finally:
+ temp_path.unlink(missing_ok=True)
+
+ verified = False
+ try:
+ _check_icloud_deadline(deadline, "starting iCloud status verification")
+ state = _icloud_item_state(destination, request_download=True, timeout_seconds=remaining_seconds())
+ while True:
+ uploading_error = state.get("uploading_error")
+ if uploading_error:
+ raise RuntimeError(f"iCloud upload failed for {destination}: {uploading_error}")
+ uploaded = state.get("is_uploaded") is True and state.get("is_uploading") is False
+ materialized = state.get("downloading_status") == "current" and destination.is_file()
+ if state.get("is_ubiquitous") is True and uploaded and materialized:
+ actual_size = destination.stat().st_size
+ actual_sha256 = _sha256_file(destination, deadline=deadline)
+ if actual_size != expected_size or actual_sha256 != expected_sha256:
+ raise RuntimeError(
+ "iCloud copy content mismatch: "
+ f"expected size={expected_size} sha256={expected_sha256}, "
+ f"actual size={actual_size} sha256={actual_sha256}"
+ )
+ verified = True
+ return receipt(reused=False)
+ if time.monotonic() >= deadline:
+ placeholder = destination.with_name(f".{destination.name}.icloud")
+ materialization = "PLACEHOLDER" if placeholder.exists() or not destination.exists() else "PENDING"
+ raise RuntimeError(
+ f"iCloud copy was not uploaded and materialized within {timeout_seconds}s: "
+ f"path={destination} materialization={materialization} state={state!r}"
+ )
+ time.sleep(min(poll_interval_seconds, remaining_seconds()))
+ placeholder = destination.with_name(f".{destination.name}.icloud")
+ status_path = placeholder if not destination.exists() and placeholder.exists() else destination
+ state = _icloud_item_state(status_path, request_download=True, timeout_seconds=remaining_seconds())
+ finally:
+ if not verified:
+ _quarantine_unverified_icloud_item(destination)
+ _quarantine_unverified_icloud_item(destination.with_name(f".{destination.name}.icloud"))
+
+
+def _icloud_copy_receipt(copy_result: dict[str, Any], icloud_dir: Path) -> tuple[str, dict[str, Any]]:
+ """Reduce a verified copy result to the durable state needed for later revalidation."""
+ path = Path(copy_result.get("path", "")).expanduser()
+ directory = Path(icloud_dir).expanduser()
+ if not path.name or path.parent != directory:
+ raise RuntimeError(f"iCloud copy receipt is outside the configured directory: {path}")
+ size = copy_result.get("bytes")
+ sha256 = copy_result.get("sha256")
+ if not isinstance(size, int) or size < 0 or not isinstance(sha256, str) or len(sha256) != 64:
+ raise RuntimeError(f"iCloud copy receipt is missing exact-byte proof: {copy_result!r}")
+ return path.name, {"bytes": size, "sha256": sha256}
+
+
+def _icloud_inventory_is_verified(
+ state: dict[str, Any],
+ candidates: list[JsonlCandidate],
+ icloud_dir: Path,
+ *,
+ timeout_seconds: float = DEFAULT_ICLOUD_TIMEOUT_SECONDS,
+ poll_interval_seconds: float = 2.0,
+ validated_sources: set[str] | None = None,
+ deadline: float | None = None,
+) -> bool:
+ """Revalidate every iCloud object that current source-state entries rely on.
+
+ Finder placeholders are not proof. Each referenced object is requested for
+ download, required to report authoritative uploaded/current state, then
+ checked against the exact size and SHA-256 persisted after its original copy.
+ Legacy marker-only state therefore bootstraps once instead of being trusted.
+ """
+ directory = Path(icloud_dir).expanduser()
+ files = state.get("files")
+ candidates_by_path = {candidate.path.as_posix(): candidate for candidate in candidates}
+ if state.get("icloud_directory") != str(directory):
+ unavailable_sources = []
+ if isinstance(files, dict):
+ unavailable_sources = sorted(
+ source_path
+ for source_path, entry in files.items()
+ if source_path not in candidates_by_path
+ and isinstance(entry, dict)
+ and (entry.get("icloud_required") is True or bool(entry.get("icloud_archive")))
+ )
+ if unavailable_sources:
+ raise RuntimeError(
+ "iCloud directory changed and the new destination cannot be seeded because "
+ f"previously covered sources are unavailable: sources={unavailable_sources!r}"
+ )
+ return False
+
+ receipts = state.get("icloud_archives")
+ if not isinstance(files, dict):
+ return not candidates
+ if not isinstance(receipts, dict):
+ receipts = {}
+
+ referenced_by_sources: dict[str, set[str]] = {}
+ complete = True
+ for source_path, entry in files.items():
+ candidate = candidates_by_path.get(source_path)
+ if not isinstance(entry, dict):
+ continue
+ # Changed/new sources are selected for this run and receive a new receipt.
+ if candidate is not None and (entry.get("mtime") != candidate.mtime or entry.get("size") != candidate.size):
+ continue
+ archive_name = entry.get("icloud_archive")
+ if not isinstance(archive_name, str) or not archive_name or Path(archive_name).name != archive_name:
+ if candidate is None:
+ if entry.get("icloud_required") is True:
+ raise RuntimeError(
+ "iCloud coverage cannot be repaired because a required source is unavailable: "
+ f"source={source_path}"
+ )
+ # State predating the optional iCloud leg can retain Drive-only
+ # entries after their local source disappears. Such an entry never
+ # claimed iCloud coverage, so it is outside this inventory.
+ continue
+ complete = False
+ continue
+ referenced_by_sources.setdefault(archive_name, set()).add(source_path)
+
+ if candidates and not referenced_by_sources:
+ return False
+
+ def invalid_archive(archive_name: str, reason: str, *, quarantine: bool = False) -> bool:
+ unavailable_sources = sorted(referenced_by_sources[archive_name] - candidates_by_path.keys())
+ if unavailable_sources:
+ raise RuntimeError(
+ "iCloud coverage cannot be repaired because its archive is invalid and a source is unavailable: "
+ f"archive={archive_name} reason={reason} sources={unavailable_sources!r}"
+ )
+ if quarantine:
+ destination = directory / archive_name
+ _quarantine_unverified_icloud_item(destination)
+ _quarantine_unverified_icloud_item(directory / f".{archive_name}.icloud")
+ return False
+
+ deadline = deadline if deadline is not None else time.monotonic() + timeout_seconds
+ for archive_name in sorted(referenced_by_sources):
+ try:
+ _check_icloud_deadline(deadline, f"validating iCloud inventory archive {archive_name}")
+ receipt = receipts.get(archive_name)
+ if not isinstance(receipt, dict):
+ invalid_archive(archive_name, "missing exact-byte receipt")
+ complete = False
+ continue
+ expected_size = receipt.get("bytes")
+ expected_sha256 = receipt.get("sha256")
+ if (
+ not isinstance(expected_size, int)
+ or expected_size < 0
+ or not isinstance(expected_sha256, str)
+ or len(expected_sha256) != 64
+ ):
+ invalid_archive(archive_name, "malformed exact-byte receipt")
+ complete = False
+ continue
+
+ destination = directory / archive_name
+ placeholder = directory / f".{archive_name}.icloud"
+ if not destination.exists() and not placeholder.exists():
+ invalid_archive(archive_name, "archive and placeholder are missing")
+ complete = False
+ continue
+
+ archive_valid = True
+ while True:
+ _check_icloud_deadline(deadline, f"probing iCloud inventory archive {archive_name}")
+ remaining = max(deadline - time.monotonic(), 0.001)
+ status_path = placeholder if not destination.exists() and placeholder.exists() else destination
+ item_state = _icloud_item_state(
+ status_path,
+ request_download=True,
+ timeout_seconds=remaining,
+ )
+ uploaded = item_state.get("is_uploaded") is True and item_state.get("is_uploading") is False
+ materialized = item_state.get("downloading_status") == "current" and destination.is_file()
+ if item_state.get("uploading_error"):
+ invalid_archive(
+ archive_name,
+ f"upload error: {item_state['uploading_error']}",
+ quarantine=True,
+ )
+ archive_valid = False
+ break
+ if item_state.get("is_ubiquitous") is True and uploaded and materialized:
+ if (
+ destination.stat().st_size != expected_size
+ or _sha256_file(destination, deadline=deadline) != expected_sha256
+ ):
+ invalid_archive(
+ archive_name,
+ "materialized bytes do not match the receipt",
+ quarantine=True,
+ )
+ archive_valid = False
+ elif validated_sources is not None:
+ validated_sources.update(referenced_by_sources[archive_name])
+ break
+ if time.monotonic() >= deadline:
+ raise ICloudDeadlineExceeded(
+ f"iCloud operation deadline exceeded while materializing inventory archive {archive_name}"
+ )
+ time.sleep(min(poll_interval_seconds, remaining))
+ if not archive_valid:
+ complete = False
+ except (OSError, RuntimeError, subprocess.TimeoutExpired) as exc:
+ if isinstance(
+ exc,
+ (
+ backup_daily.BackupTimeoutError,
+ ICloudDeadlineExceeded,
+ ICloudProbeError,
+ subprocess.TimeoutExpired,
+ ),
+ ):
+ raise
+ if isinstance(exc, RuntimeError) and str(exc).startswith("iCloud coverage cannot be repaired"):
+ raise
+ invalid_archive(archive_name, str(exc))
+ complete = False
+ continue
+ referenced_sources = set().union(*referenced_by_sources.values()) if referenced_by_sources else set()
+ return complete and candidates_by_path.keys() <= referenced_sources
+
+
def _upload_forever_files(
candidates: list[JsonlCandidate],
*,
@@ -549,6 +1030,9 @@ def _update_state_for_uploaded(
archive_id: str | None = None,
archive_md5: str | None = None,
digests: dict[str, str] | None = None,
+ icloud_dir: Path | None = None,
+ icloud_copy: dict[str, Any] | None = None,
+ clear_icloud_verification: bool = False,
) -> dict[str, Any]:
"""Record which archive object carries each file, and the bytes it carried.
@@ -557,6 +1041,13 @@ def _update_state_for_uploaded(
unrecoverable while state still reports it as backed up.
"""
files = dict(state.get("files") or {})
+ existing_icloud_directory = state.get("icloud_directory")
+ icloud_archive_name: str | None = None
+ icloud_receipt: dict[str, Any] | None = None
+ if icloud_copy is not None:
+ if icloud_dir is None:
+ raise RuntimeError("an iCloud copy receipt requires its configured directory")
+ icloud_archive_name, icloud_receipt = _icloud_copy_receipt(icloud_copy, icloud_dir)
for candidate in candidates:
entry: dict[str, Any] = {"mtime": candidate.mtime, "size": candidate.size}
if archive_name and archive_id:
@@ -566,8 +1057,35 @@ def _update_state_for_uploaded(
entry["archive_md5"] = archive_md5
digest = (digests or {}).get(candidate.path.as_posix())
entry["sha256"] = digest if digest else _sha256_file(candidate.path)
+ if icloud_archive_name is not None:
+ entry["icloud_archive"] = icloud_archive_name
+ elif isinstance(existing_icloud_directory, str) and existing_icloud_directory:
+ entry["icloud_required"] = True
files[candidate.path.as_posix()] = entry
- return {"files": files, "updated_at": dt.datetime.now(dt.UTC).isoformat()}
+ updated = {"files": files, "updated_at": dt.datetime.now(dt.UTC).isoformat()}
+
+ archives = dict(state.get("icloud_archives") or {})
+ if icloud_archive_name is not None and icloud_receipt is not None:
+ archives[icloud_archive_name] = icloud_receipt
+ referenced_archives = {
+ entry.get("icloud_archive")
+ for entry in files.values()
+ if isinstance(entry, dict) and isinstance(entry.get("icloud_archive"), str)
+ }
+ retained_archives = {name: receipt for name, receipt in archives.items() if name in referenced_archives}
+ if retained_archives:
+ updated["icloud_archives"] = retained_archives
+
+ if icloud_dir is not None:
+ updated["icloud_directory"] = str(Path(icloud_dir).expanduser())
+ if not clear_icloud_verification:
+ updated["icloud_verified"] = True
+ elif not clear_icloud_verification and state.get("icloud_verified") is True:
+ existing_directory = state.get("icloud_directory")
+ if isinstance(existing_directory, str) and existing_directory:
+ updated["icloud_directory"] = existing_directory
+ updated["icloud_verified"] = True
+ return updated
def _enqueue_run_summary(result: dict[str, Any], *, queue_dir: Path | None) -> None:
@@ -626,6 +1144,7 @@ def run_backup(
upload: bool = True,
active_skip_seconds: int = DEFAULT_ACTIVE_SKIP_SECONDS,
forever_folder_parts: list[str] = DEFAULT_FOREVER_FOLDER_PARTS,
+ icloud_dir: Path | None = None,
) -> dict[str, Any]:
date_stamp = date_stamp or _today()
now = time.time() if now is None else now
@@ -634,6 +1153,31 @@ def run_backup(
state_path = Path(state_path).expanduser()
state = _load_state(state_path)
candidates = _discover_jsonl_candidates(roots)
+ selection_state = state
+ icloud_bootstrap_pending = False
+ icloud_deadline: float | None = None
+ if upload and icloud_dir is not None:
+ icloud_deadline = time.monotonic() + DEFAULT_ICLOUD_TIMEOUT_SECONDS
+ validated_icloud_sources: set[str] = set()
+ icloud_covered = _icloud_inventory_is_verified(
+ state,
+ candidates,
+ icloud_dir,
+ validated_sources=validated_icloud_sources,
+ deadline=icloud_deadline,
+ )
+ if not icloud_covered:
+ # Legacy state proves only Drive coverage. The first iCloud-enabled run
+ # must seed iCloud with every source rather than falsely returning no-op.
+ state_files = state.get("files") or {}
+ selection_state = {
+ "files": {
+ source_path: entry
+ for source_path, entry in state_files.items()
+ if source_path in validated_icloud_sources
+ }
+ }
+ icloud_bootstrap_pending = True
credentials = None
service = None
surviving_archives: dict[str, str | None] | None = None
@@ -655,12 +1199,34 @@ def run_backup(
surviving_archives = {}
changed, active, covered, vanished = _select_backup_candidates(
candidates,
- state=state,
+ state=selection_state,
now=now,
active_skip_seconds=active_skip_seconds,
surviving_archives=surviving_archives,
)
+ if not changed and icloud_bootstrap_pending and active:
+ error = (
+ "iCloud coverage is not verified; repair deferred because "
+ f"{len(active)} discovered source(s) are still active"
+ )
+ result = {
+ "attempted_at": attempted_at,
+ "status": "deferred",
+ "uploaded": False,
+ "verified": False,
+ "already_covered_files": covered,
+ "discovered_file_count": len(candidates),
+ "skipped_active_count": len(active),
+ "vanished_source_count": vanished,
+ "icloud_repair_deferred": True,
+ "error": error,
+ "message": error,
+ }
+ _append_json_log(log_path, result)
+ _enqueue_run_summary(result, queue_dir=queue_dir)
+ return result
+
if not changed:
result: dict[str, Any] = {
"attempted_at": attempted_at,
@@ -707,6 +1273,14 @@ def run_backup(
)
)
if result["verified"] and upload:
+ # iCloud goes first: a failed iCloud verification must not create an
+ # unrecorded duplicate Drive object that consumes the retention window.
+ if icloud_dir is not None:
+ result["icloud_copy"] = copy_archive_to_icloud(
+ archive_path,
+ icloud_dir,
+ deadline=icloud_deadline,
+ )
if service is None:
credentials = backup_daily.get_drive_credentials()
service = backup_daily.build_drive_service()
@@ -722,6 +1296,10 @@ def run_backup(
expected_size=archive_size,
)
result.update({"status": "uploaded", "uploaded": True, "drive_file": uploaded})
+ # The same incident was two individually reasonable deletions composed together:
+ # successful upload removed local staging, then Drive retention removed the remote
+ # bundle. Persist the exact Drive object and archived-source digests before either
+ # deletion path runs so the next selection cannot silently trust the dead copy.
_atomic_write_json(
state_path,
_update_state_for_uploaded(
@@ -731,6 +1309,11 @@ def run_backup(
archive_id=file_id,
archive_md5=uploaded.get("md5Checksum"),
digests=bundle_digests,
+ # An active source was deliberately omitted from this bootstrap.
+ # Leave the global marker unset so the next run seeds that source.
+ icloud_dir=icloud_dir,
+ icloud_copy=result.get("icloud_copy"),
+ clear_icloud_verification=bool(icloud_bootstrap_pending and active),
),
)
try:
@@ -798,6 +1381,7 @@ def main() -> int:
folder_parts=os.environ.get("BRAINLAYER_JSONL_BACKUP_DRIVE_FOLDER", "/".join(DEFAULT_FOLDER_PARTS)).split(
"/"
),
+ icloud_dir=_configured_icloud_dir(),
)
except backup_daily.BackupTimeoutError:
result = {
diff --git a/tests/test_hook_python.py b/tests/test_hook_python.py
index 9463a2cc..4aee2ff0 100644
--- a/tests/test_hook_python.py
+++ b/tests/test_hook_python.py
@@ -167,6 +167,7 @@ class TestIsPinnedInterpreter:
# this as an override, so the linter must accept it too, or the escape hatch and
# the gate contradict each other (review round 1, medium).
"/tmp/myvenv/bin/python",
+ "'/Users/Jane Doe/.venv/bin/python'",
"/Users/x/Gits/brainlayer/.venv/bin/python3.13",
],
)
@@ -251,10 +252,20 @@ def test_an_absolute_venv_override_outside_a_keg_is_accepted(tmp_path):
target = tmp_path / "myvenv" / "bin" / "python"
target.parent.mkdir(parents=True)
target.write_text("#!/bin/sh\n")
+ target.chmod(0o755)
resolved = resolve_hook_python(env={HOOK_PYTHON_ENV: str(target)}, candidates=())
assert resolved == str(target)
assert is_pinned_interpreter(resolved), "the linter must accept what the hatch returns"
+ @staticmethod
+ def test_an_absolute_venv_override_with_spaces_is_accepted(tmp_path):
+ target = tmp_path / "Jane Doe" / ".venv" / "bin" / "python"
+ target.parent.mkdir(parents=True)
+ target.write_text("#!/bin/sh\n")
+ target.chmod(0o755)
+
+ assert resolve_hook_python(env={HOOK_PYTHON_ENV: str(target)}, candidates=()) == str(target)
+
@staticmethod
def test_first_existing_candidate_wins(tmp_path):
missing = tmp_path / "missing" / "python"
@@ -263,6 +274,20 @@ def test_first_existing_candidate_wins(tmp_path):
present.chmod(0o755)
assert resolve_hook_python(env={}, candidates=(str(missing), str(present))) == str(present)
+ @staticmethod
+ def test_candidate_must_be_a_regular_executable_file(tmp_path):
+ directory = tmp_path / "directory" / "python"
+ directory.mkdir(parents=True)
+ non_executable = tmp_path / "not-executable" / "python"
+ non_executable.parent.mkdir()
+ non_executable.write_text("#!/bin/sh\n")
+ usable = tmp_path / "usable" / "python"
+ usable.parent.mkdir()
+ usable.write_text("#!/bin/sh\n")
+ usable.chmod(0o755)
+
+ assert resolve_hook_python(env={}, candidates=(str(directory), str(non_executable), str(usable))) == str(usable)
+
@staticmethod
def test_never_falls_back_to_path():
"""A silent `python3` fallback is the bug, not the remedy."""
@@ -298,6 +323,15 @@ def test_default_candidate_is_the_opt_symlink():
"""`opt/` outlives the Cellar version a command was rendered against."""
assert DEFAULT_KEG_PYTHON == "/opt/homebrew/opt/brainlayer/libexec/venv/bin/python"
+ @staticmethod
+ def test_non_python_executable_override_is_refused(tmp_path):
+ target = tmp_path / "bin" / "bash"
+ target.parent.mkdir()
+ target.write_text("#!/bin/sh\n")
+ target.chmod(0o755)
+ with pytest.raises(HookPythonUnresolved, match="executable Python"):
+ resolve_hook_python(env={HOOK_PYTHON_ENV: str(target)}, candidates=())
+
class TestRenderHookCommand:
@staticmethod
@@ -481,6 +515,17 @@ def test_exits_two_on_invalid_json(tmp_path, capsys):
assert main([str(path)]) == 2
assert "cannot read" in capsys.readouterr().out
+ @staticmethod
+ def test_print_interpreter_uses_affirmative_resolver(tmp_path, monkeypatch, capsys):
+ python = tmp_path / "venv" / "bin" / "python"
+ python.parent.mkdir(parents=True)
+ python.write_text("#!/bin/sh\n")
+ python.chmod(0o755)
+ monkeypatch.setenv(HOOK_PYTHON_ENV, str(python))
+
+ assert main(["--print-interpreter"]) == 0
+ assert capsys.readouterr().out.strip() == str(python)
+
#: conftest sandboxes HOME for every test, so `~/.claude/settings.json` is not
#: reachable by default — deliberately: a unit suite must not read Etan's home. Point
@@ -506,3 +551,87 @@ def test_installed_brainlayer_hooks_are_pinned():
assert findings == [], "BrainLayer hooks still resolve their interpreter through PATH: " + "; ".join(
f"{f.event}: {f.command}" for f in findings
)
+
+
+def test_launchd_plist_templates_pin_their_interpreter(tmp_path):
+ """Render machine-specific placeholders, then apply the existing pin gate."""
+ import plistlib
+
+ from brainlayer.hook_python import render_launchd_plist
+
+ plists = sorted((REPO_ROOT / "launchd").glob("*.plist"))
+ assert plists, "expected launchd templates to exist"
+ python = tmp_path / "intel" / "opt" / "brainlayer" / "libexec" / "venv" / "bin" / "python"
+ python.parent.mkdir(parents=True)
+ python.write_text("#!/bin/sh\n")
+ python.chmod(0o755)
+
+ unpinned: list[str] = []
+ for path in plists:
+ template = path.read_text(encoding="utf-8")
+ rendered = render_launchd_plist(template, python=str(python))
+ args = plistlib.loads(rendered.encode()).get("ProgramArguments") or []
+ if not args:
+ continue
+ interpreter = args[0]
+ # A wrapper or installed CLI is not a direct interpreter claim.
+ if "python" not in interpreter and not interpreter.endswith("/env"):
+ continue
+ if interpreter.endswith("/env"):
+ interpreter = f"{interpreter} {args[1] if len(args) > 1 else ''}".strip()
+ if not is_pinned_interpreter(interpreter):
+ unpinned.append(f"{path.name}: {interpreter}")
+
+ assert not unpinned, "launchd templates must name a pinned interpreter, not PATH: " + "; ".join(unpinned)
+
+
+def test_launchd_plist_render_uses_available_intel_keg(monkeypatch):
+ from brainlayer import hook_python
+
+ template = "__BRAINLAYER_PYTHON__"
+ intel = "/usr/local/opt/brainlayer/libexec/venv/bin/python"
+ monkeypatch.setattr(hook_python.os.path, "exists", lambda path: path == intel)
+ monkeypatch.setattr(hook_python.os.path, "isfile", lambda path: path == intel)
+ monkeypatch.setattr(hook_python.os, "access", lambda path, mode: path == intel and mode == hook_python.os.X_OK)
+
+ rendered = hook_python.render_launchd_plist(template, env={})
+
+ assert rendered == f"{intel}"
+
+
+def test_launchd_plist_render_rejects_non_executable_interpreter(tmp_path):
+ from brainlayer.hook_python import HookPythonUnresolved, render_launchd_plist
+
+ python = tmp_path / "venv" / "bin" / "python"
+ python.parent.mkdir(parents=True)
+ python.write_text("#!/bin/sh\n")
+
+ with pytest.raises(HookPythonUnresolved):
+ render_launchd_plist("__BRAINLAYER_PYTHON__", python=str(python))
+
+
+def test_launchd_plist_render_accepts_executable_interpreter_with_spaces(tmp_path):
+ import xml.etree.ElementTree as ET
+
+ from brainlayer.hook_python import render_launchd_plist
+
+ python = tmp_path / "Jane Doe" / ".venv" / "bin" / "python"
+ python.parent.mkdir(parents=True)
+ python.write_text("#!/bin/sh\n")
+ python.chmod(0o755)
+
+ rendered = render_launchd_plist("__BRAINLAYER_PYTHON__", python=str(python))
+
+ assert ET.fromstring(rendered).text == str(python)
+
+
+def test_launchd_plist_render_rejects_xml_forbidden_interpreter_path(tmp_path):
+ from brainlayer.hook_python import HookPythonUnresolved, render_launchd_plist
+
+ python = tmp_path / "bad\x01path" / ".venv" / "bin" / "python"
+ python.parent.mkdir(parents=True)
+ python.write_text("#!/bin/sh\n")
+ python.chmod(0o755)
+
+ with pytest.raises(HookPythonUnresolved):
+ render_launchd_plist("__BRAINLAYER_PYTHON__", python=str(python))
diff --git a/tests/test_installable_build.py b/tests/test_installable_build.py
index c53ecdc2..a4ee94a9 100644
--- a/tests/test_installable_build.py
+++ b/tests/test_installable_build.py
@@ -187,7 +187,7 @@ def _copy_packaged_launchd(launchd_dir: Path) -> None:
target.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(origin, target)
package_dir = launchd_dir.parent
- for name in ("__init__.py", "config.py", "paths.py", "spotlight.py"):
+ for name in ("__init__.py", "config.py", "hook_python.py", "paths.py", "spotlight.py"):
shutil.copy2(REPO_ROOT / "src" / "brainlayer" / name, package_dir / name)
@@ -1623,7 +1623,10 @@ def test_launchd_installer_renders_brainlayer_python_override(tmp_path: Path) ->
env_file = tmp_path / "brainlayer.env"
env_file.write_text("BRAINLAYER_ENRICH_ENABLED=0\n", encoding="utf-8")
env_file.chmod(0o600)
- brainlayer_python = tmp_path / "tool" / "bin" / "python"
+ brainlayer_python = tmp_path / "tool&operator" / "bin" / "python"
+ brainlayer_python.parent.mkdir(parents=True)
+ brainlayer_python.write_text("#!/bin/sh\n", encoding="utf-8")
+ brainlayer_python.chmod(0o755)
result = subprocess.run(
[str(REPO_ROOT / "scripts" / "launchd" / "install.sh"), "backup"],
@@ -1644,10 +1647,57 @@ def test_launchd_installer_renders_brainlayer_python_override(tmp_path: Path) ->
assert result.returncode == 0, result.stdout + result.stderr
rendered = home / "Library" / "LaunchAgents" / "com.brainlayer.backup-daily.plist"
- assert f"{brainlayer_python}" in rendered.read_text(encoding="utf-8")
+ assert plistlib.loads(rendered.read_bytes())["EnvironmentVariables"]["BRAINLAYER_PYTHON"] == str(brainlayer_python)
assert "__BRAINLAYER_PYTHON__" not in rendered.read_text(encoding="utf-8")
+def test_source_jsonl_installer_uses_prefix_aware_resolver_not_path_python(tmp_path: Path) -> None:
+ fake_bin = tmp_path / "bin"
+ fake_bin.mkdir()
+ authoring_python = fake_bin / "python3"
+ authoring_python.write_text(f'#!/bin/sh\nexec "{sys.executable}" "$@"\n', encoding="utf-8")
+ authoring_python.chmod(0o755)
+ fake_launchctl = fake_bin / "launchctl"
+ fake_launchctl.write_text("\n".join(_fake_launchctl_lines()), encoding="utf-8")
+ fake_launchctl.chmod(0o755)
+ home = tmp_path / "home"
+ home.mkdir()
+ env_file = tmp_path / "brainlayer.env"
+ env_file.write_text("BRAINLAYER_ENRICH_ENABLED=0\n", encoding="utf-8")
+ env_file.chmod(0o600)
+ pinned_python = tmp_path / "intel-prefix" / "opt" / "brainlayer" / "libexec" / "venv" / "bin" / "python"
+ pinned_python.parent.mkdir(parents=True)
+ pinned_python.write_text("#!/bin/sh\n", encoding="utf-8")
+ pinned_python.chmod(0o755)
+ child_env = {
+ key: value
+ for key, value in os.environ.items()
+ if key not in {"BRAINLAYER_PYTHON", "PYTHON_BIN", "BRAINLAYER_HOOK_PYTHON"}
+ }
+
+ result = subprocess.run(
+ [str(REPO_ROOT / "scripts" / "launchd" / "install.sh"), "jsonl-backup"],
+ env={
+ **child_env,
+ "PATH": f"{fake_bin}:{os.environ['PATH']}",
+ "HOME": str(home),
+ "BRAINLAYER_BIN": sys.executable,
+ "BRAINLAYER_HOOK_PYTHON": str(pinned_python),
+ "BRAINLAYER_ENV_FILE": str(env_file),
+ "FAKE_LAUNCHCTL_LOG": str(tmp_path / "launchctl.log"),
+ },
+ capture_output=True,
+ text=True,
+ check=False,
+ )
+
+ assert result.returncode == 0, result.stdout + result.stderr
+ rendered = home / "Library" / "LaunchAgents" / "com.brainlayer.jsonl-backup.plist"
+ content = rendered.read_text(encoding="utf-8")
+ assert f"{pinned_python}" in content
+ assert str(authoring_python) not in content
+
+
def test_packaged_launchd_installer_renders_p0_counter_console_shim(tmp_path: Path) -> None:
launchd_dir = tmp_path / "site-packages" / "brainlayer" / "launchd"
_copy_packaged_launchd(launchd_dir)
diff --git a/tests/test_jsonl_backup.py b/tests/test_jsonl_backup.py
index 1f9d7bd5..a5cff66e 100644
--- a/tests/test_jsonl_backup.py
+++ b/tests/test_jsonl_backup.py
@@ -1,7 +1,9 @@
+import gzip
import hashlib
import io
import json
import os
+import subprocess
import tarfile
import threading
import time
@@ -17,6 +19,49 @@ def _write_jsonl(path: Path, line: str = '{"type":"message"}\n', *, mtime: float
return path
+def _mock_drive_success(jsonl_backup, monkeypatch, uploads: list[Path] | None = None) -> None:
+ monkeypatch.setattr(jsonl_backup.backup_daily, "get_drive_credentials", lambda: object())
+ monkeypatch.setattr(jsonl_backup.backup_daily, "build_drive_service", lambda: object())
+ monkeypatch.setattr(jsonl_backup.backup_daily, "ensure_drive_folder_chain", lambda *args: "folder-id")
+
+ def upload(path, *args): # noqa: ARG001
+ if uploads is not None:
+ uploads.append(Path(path))
+ return {"id": "drive-id", "name": Path(path).name, "size": str(Path(path).stat().st_size)}
+
+ monkeypatch.setattr(jsonl_backup.backup_daily, "upload_file_to_drive_raw", upload)
+ monkeypatch.setattr(jsonl_backup.backup_daily, "verify_drive_upload", lambda *args, **kwargs: None)
+ monkeypatch.setattr(jsonl_backup.backup_daily, "prune_drive_backups", lambda *args, **kwargs: [])
+
+
+def _icloud_state(*, uploaded: bool, status: str) -> dict:
+ return {
+ "is_ubiquitous": True,
+ "is_uploaded": uploaded,
+ "is_uploading": not uploaded,
+ "downloading_status": status,
+ }
+
+
+def _copy_to_icloud_receipt(archive: Path, destination: Path, **kwargs) -> dict: # noqa: ARG001
+ destination.mkdir(parents=True, exist_ok=True)
+ try:
+ logical_bytes = gzip.decompress(archive.read_bytes())
+ except (gzip.BadGzipFile, EOFError):
+ logical_bytes = archive.read_bytes()
+ suffix = "".join(archive.suffixes)
+ logical_sha256 = hashlib.sha256(logical_bytes).hexdigest()
+ target = destination / f"claude-jsonl-{logical_sha256}{suffix}"
+ target.write_bytes(archive.read_bytes())
+ return {
+ "path": str(target),
+ "uploaded": True,
+ "materialization": "MATERIALIZED",
+ "bytes": target.stat().st_size,
+ "sha256": hashlib.sha256(target.read_bytes()).hexdigest(),
+ }
+
+
def test_jsonl_retention_invariant_is_a_ci_guard_not_only_a_behavior_fixture():
"""Fail CI when #815's surviving-copy predicate or call-site ordering is loosened."""
from brainlayer.backup_retention_invariant import inspect_jsonl_retention_invariant
@@ -163,6 +208,11 @@ def _state_matches(entry, candidate, surviving_archives=None):
archive_id=file_id,
archive_md5=uploaded.get("md5Checksum"),
digests=bundle_digests,
+ # An active source was deliberately omitted from this bootstrap.
+ # Leave the global marker unset so the next run seeds that source.
+ icloud_dir=icloud_dir,
+ icloud_copy=result.get("icloud_copy"),
+ clear_icloud_verification=bool(icloud_bootstrap_pending and active),
),
)
"""
@@ -173,6 +223,11 @@ def _state_matches(entry, candidate, surviving_archives=None):
archive_id=file_id,
archive_md5=uploaded.get("md5Checksum"),
digests=bundle_digests,
+ # An active source was deliberately omitted from this bootstrap.
+ # Leave the global marker unset so the next run seeds that source.
+ icloud_dir=icloud_dir,
+ icloud_copy=result.get("icloud_copy"),
+ clear_icloud_verification=bool(icloud_bootstrap_pending and active),
)
"""
assert persisted_state_block in source
@@ -600,6 +655,1300 @@ def fake_prune(service, *, folder_parts, retention_policy): # noqa: ARG001
assert "JSONL backup uploaded 3 files" in queued[0].read_text()
+def test_icloud_copy_requires_uploaded_materialized_exact_bytes(tmp_path, monkeypatch):
+ from brainlayer import jsonl_backup
+
+ archive = tmp_path / "staging" / "claude-jsonl-2026-09-09.tar.gz"
+ archive.parent.mkdir()
+ archive.write_bytes(b"verified archive bytes")
+ icloud_dir = tmp_path / "CloudDocs" / "Archives" / "brainlayer-jsonl-backups"
+ states = iter(
+ [
+ _icloud_state(uploaded=False, status="notDownloaded"),
+ _icloud_state(uploaded=True, status="current"),
+ ]
+ )
+ actions: list[tuple[str, float | None]] = []
+ expected_name = f"claude-jsonl-{hashlib.sha256(archive.read_bytes()).hexdigest()}.tar.gz"
+
+ def fake_icloud_item_state(path, *, request_download=False, timeout_seconds=None):
+ assert Path(path) == icloud_dir / expected_name
+ actions.append(("download" if request_download else "status", timeout_seconds))
+ return next(states)
+
+ monkeypatch.setattr(jsonl_backup, "_icloud_item_state", fake_icloud_item_state)
+
+ result = jsonl_backup.copy_archive_to_icloud(
+ archive,
+ icloud_dir,
+ timeout_seconds=1,
+ poll_interval_seconds=0,
+ )
+
+ assert [action for action, _ in actions] == ["download", "download"]
+ assert all(timeout is not None and 0 < timeout <= 1 for _, timeout in actions)
+ assert result["materialization"] == "MATERIALIZED"
+ assert result["sha256"] == hashlib.sha256(archive.read_bytes()).hexdigest()
+ assert Path(result["path"]).read_bytes() == archive.read_bytes()
+
+
+def test_icloud_destination_is_strictly_opt_in(monkeypatch):
+ from brainlayer import jsonl_backup
+
+ monkeypatch.delenv("BRAINLAYER_JSONL_BACKUP_ICLOUD_DIR", raising=False)
+ assert jsonl_backup._configured_icloud_dir() is None
+ monkeypatch.setenv("BRAINLAYER_JSONL_BACKUP_ICLOUD_DIR", "/CloudDocs/Archives/brainlayer-jsonl-backups")
+ assert jsonl_backup._configured_icloud_dir() == Path("/CloudDocs/Archives/brainlayer-jsonl-backups")
+
+
+def test_logical_gzip_hash_does_not_swallow_wall_clock_timeout(tmp_path, monkeypatch):
+ from brainlayer import jsonl_backup
+
+ archive = tmp_path / "archive.tar.gz"
+ archive.write_bytes(b"bytes")
+ monkeypatch.setattr(
+ jsonl_backup.gzip,
+ "open",
+ lambda *args, **kwargs: (_ for _ in ()).throw(jsonl_backup.backup_daily.BackupTimeoutError("deadline")),
+ )
+ monkeypatch.setattr(jsonl_backup, "_sha256_file", lambda path: "fallback")
+
+ with pytest.raises(jsonl_backup.backup_daily.BackupTimeoutError, match="deadline"):
+ jsonl_backup._sha256_gzip_payload(archive)
+
+
+def test_icloud_copy_rehydrates_placeholder_before_hashing(tmp_path, monkeypatch):
+ from brainlayer import jsonl_backup
+
+ archive = tmp_path / "claude-jsonl-2026-09-09.tar.gz"
+ archive.write_bytes(b"archive bytes")
+ icloud_dir = tmp_path / "CloudDocs" / "Archives" / "brainlayer-jsonl-backups"
+ calls: list[Path] = []
+ placeholder: Path | None = None
+
+ def fake_icloud_item_state(path, *, request_download=False, timeout_seconds=None): # noqa: ARG001
+ nonlocal placeholder
+ calls.append(Path(path))
+ assert request_download is True
+ if len(calls) == 1:
+ destination = Path(path)
+ placeholder = destination.with_name(f".{destination.name}.icloud")
+ destination.unlink()
+ placeholder.write_bytes(b"")
+ return _icloud_state(uploaded=True, status="notDownloaded")
+ assert placeholder is not None
+ assert Path(path) == placeholder
+ placeholder.unlink()
+ Path(calls[0]).write_bytes(archive.read_bytes())
+ return _icloud_state(uploaded=True, status="current")
+
+ monkeypatch.setattr(jsonl_backup, "_icloud_item_state", fake_icloud_item_state)
+
+ result = jsonl_backup.copy_archive_to_icloud(
+ archive,
+ icloud_dir,
+ timeout_seconds=1,
+ poll_interval_seconds=0,
+ )
+
+ assert calls == [Path(result["path"]), placeholder]
+ assert result["materialization"] == "MATERIALIZED"
+
+
+def test_icloud_copy_rejects_uploaded_item_with_wrong_materialized_bytes(tmp_path, monkeypatch):
+ from brainlayer import jsonl_backup
+
+ archive = tmp_path / "claude-jsonl-2026-09-09.tar.gz"
+ archive.write_bytes(b"expected bytes")
+ icloud_dir = tmp_path / "CloudDocs" / "Archives" / "brainlayer-jsonl-backups"
+
+ def fake_icloud_item_state(path, *, request_download=False, timeout_seconds=None): # noqa: ARG001
+ Path(path).write_bytes(b"remote bytes changed")
+ return _icloud_state(uploaded=True, status="current")
+
+ monkeypatch.setattr(jsonl_backup, "_icloud_item_state", fake_icloud_item_state)
+
+ with pytest.raises(RuntimeError, match="iCloud copy content mismatch"):
+ jsonl_backup.copy_archive_to_icloud(
+ archive,
+ icloud_dir,
+ timeout_seconds=1,
+ poll_interval_seconds=0,
+ )
+ quarantined = list(icloud_dir.iterdir())
+ assert len(quarantined) == 1
+ assert quarantined[0].name.endswith(".unverified")
+
+
+def test_icloud_status_reports_stderr_when_osascript_fails(tmp_path, monkeypatch):
+ from brainlayer import jsonl_backup
+
+ error = subprocess.CalledProcessError(1, ["osascript"], stderr="Foundation failed")
+ monkeypatch.setattr(jsonl_backup.subprocess, "run", lambda *args, **kwargs: (_ for _ in ()).throw(error))
+
+ with pytest.raises(jsonl_backup.ICloudProbeError, match="Foundation failed"):
+ jsonl_backup._icloud_item_state(tmp_path / "archive.tar.gz", timeout_seconds=0.5)
+
+
+def test_icloud_status_marks_malformed_json_as_inconclusive_probe_failure(tmp_path, monkeypatch):
+ from brainlayer import jsonl_backup
+
+ completed = subprocess.CompletedProcess(["osascript"], 0, stdout="{truncated", stderr="")
+ monkeypatch.setattr(jsonl_backup.subprocess, "run", lambda *args, **kwargs: completed)
+
+ with pytest.raises(jsonl_backup.ICloudProbeError, match="invalid JSON"):
+ jsonl_backup._icloud_item_state(tmp_path / "archive.tar.gz", timeout_seconds=0.5)
+
+
+def test_icloud_status_does_not_wrap_wall_clock_timeout(tmp_path, monkeypatch):
+ from brainlayer import jsonl_backup
+
+ timeout = jsonl_backup.backup_daily.BackupTimeoutError("wall clock expired")
+ monkeypatch.setattr(jsonl_backup.subprocess, "run", lambda *args, **kwargs: (_ for _ in ()).throw(timeout))
+
+ with pytest.raises(jsonl_backup.backup_daily.BackupTimeoutError, match="wall clock expired"):
+ jsonl_backup._icloud_item_state(tmp_path / "archive.tar.gz", timeout_seconds=0.5)
+
+
+def test_icloud_timeout_quarantines_unverified_placeholder(tmp_path, monkeypatch):
+ from brainlayer import jsonl_backup
+
+ archive = tmp_path / "archive.tar.gz"
+ archive.write_bytes(b"bytes")
+ icloud_dir = tmp_path / "CloudDocs"
+ clock = [0.0]
+ monkeypatch.setattr(jsonl_backup.time, "monotonic", lambda: clock[0])
+
+ def leave_placeholder(path, **kwargs): # noqa: ARG001
+ destination = Path(path)
+ destination.unlink()
+ destination.with_name(f".{destination.name}.icloud").write_bytes(b"")
+ clock[0] = 2.0
+ return _icloud_state(uploaded=False, status="notDownloaded")
+
+ monkeypatch.setattr(
+ jsonl_backup,
+ "_icloud_item_state",
+ leave_placeholder,
+ )
+
+ with pytest.raises(RuntimeError, match="not uploaded and materialized"):
+ jsonl_backup.copy_archive_to_icloud(archive, icloud_dir, timeout_seconds=1, poll_interval_seconds=0)
+ quarantined = list(icloud_dir.iterdir())
+ assert len(quarantined) == 1
+ assert ".icloud." in quarantined[0].name
+ assert quarantined[0].name.endswith(".unverified")
+
+
+def test_icloud_upload_error_quarantines_unverified_destination(tmp_path, monkeypatch):
+ from brainlayer import jsonl_backup
+
+ archive = tmp_path / "archive.tar.gz"
+ archive.write_bytes(b"bytes")
+ icloud_dir = tmp_path / "CloudDocs"
+ state = _icloud_state(uploaded=False, status="current") | {"uploading_error": "quota"}
+ monkeypatch.setattr(jsonl_backup, "_icloud_item_state", lambda *args, **kwargs: state)
+
+ with pytest.raises(RuntimeError, match="quota"):
+ jsonl_backup.copy_archive_to_icloud(archive, icloud_dir, timeout_seconds=1)
+ quarantined = list(icloud_dir.iterdir())
+ assert len(quarantined) == 1
+ assert quarantined[0].name.endswith(".unverified")
+
+
+def test_repeated_icloud_failures_preserve_every_quarantined_copy(tmp_path):
+ from brainlayer import jsonl_backup
+
+ destination = tmp_path / "archive.tar.gz"
+ destination.write_bytes(b"first failed copy")
+ jsonl_backup._quarantine_unverified_icloud_item(destination)
+ destination.write_bytes(b"second failed copy")
+ jsonl_backup._quarantine_unverified_icloud_item(destination)
+
+ quarantined = list(tmp_path.glob("*.unverified"))
+ assert len(quarantined) == 2
+ assert {path.read_bytes() for path in quarantined} == {b"first failed copy", b"second failed copy"}
+
+
+def test_same_day_incremental_icloud_bundles_do_not_overwrite(tmp_path, monkeypatch):
+ from brainlayer import jsonl_backup
+
+ first = tmp_path / "first" / "claude-jsonl-2026-09-09.tar.gz"
+ second = tmp_path / "second" / "claude-jsonl-2026-09-10.tar.gz"
+ first.parent.mkdir()
+ second.parent.mkdir()
+ first.write_bytes(b"first incremental bundle")
+ second.write_bytes(b"second incremental bundle")
+ icloud_dir = tmp_path / "CloudDocs"
+ monkeypatch.setattr(
+ jsonl_backup,
+ "_icloud_item_state",
+ lambda *args, **kwargs: _icloud_state(uploaded=True, status="current"),
+ )
+
+ first_result = jsonl_backup.copy_archive_to_icloud(first, icloud_dir, timeout_seconds=1)
+ second_result = jsonl_backup.copy_archive_to_icloud(second, icloud_dir, timeout_seconds=1, poll_interval_seconds=0)
+
+ assert first_result["path"] != second_result["path"]
+ assert sorted(path.read_bytes() for path in icloud_dir.iterdir()) == sorted(
+ [first.read_bytes(), second.read_bytes()]
+ )
+
+
+def test_icloud_retry_reuses_logical_gzip_payload_destination(tmp_path, monkeypatch):
+ from brainlayer import jsonl_backup
+
+ first = tmp_path / "first" / "claude-jsonl-2026-09-09.tar.gz"
+ second = tmp_path / "second" / "claude-jsonl-2026-09-10.tar.gz"
+ first.parent.mkdir()
+ second.parent.mkdir()
+ first.write_bytes(gzip.compress(b"same tar payload", mtime=1))
+ second.write_bytes(gzip.compress(b"same tar payload", mtime=2))
+ assert first.read_bytes() != second.read_bytes()
+ icloud_dir = tmp_path / "CloudDocs"
+ monkeypatch.setattr(
+ jsonl_backup,
+ "_icloud_item_state",
+ lambda *args, **kwargs: _icloud_state(uploaded=True, status="current"),
+ )
+
+ first_result = jsonl_backup.copy_archive_to_icloud(first, icloud_dir, timeout_seconds=1)
+ second_result = jsonl_backup.copy_archive_to_icloud(second, icloud_dir, timeout_seconds=1)
+
+ assert first_result["path"] == second_result["path"]
+ assert list(icloud_dir.iterdir()) == [Path(second_result["path"])]
+ assert second_result["reused"] is True
+ assert Path(second_result["path"]).read_bytes() == first.read_bytes()
+
+
+def test_icloud_retry_waits_for_existing_logical_object_to_materialize(tmp_path, monkeypatch):
+ from brainlayer import jsonl_backup
+
+ first = tmp_path / "first" / "claude-jsonl-2026-09-09.tar.gz"
+ second = tmp_path / "second" / first.name
+ first.parent.mkdir()
+ second.parent.mkdir()
+ first.write_bytes(gzip.compress(b"same tar payload", mtime=1))
+ second.write_bytes(gzip.compress(b"same tar payload", mtime=2))
+ icloud_dir = tmp_path / "CloudDocs"
+ monkeypatch.setattr(
+ jsonl_backup,
+ "_icloud_item_state",
+ lambda *args, **kwargs: _icloud_state(uploaded=True, status="current"),
+ )
+ jsonl_backup.copy_archive_to_icloud(first, icloud_dir, timeout_seconds=1)
+
+ states = iter(
+ [
+ _icloud_state(uploaded=False, status="notDownloaded"),
+ _icloud_state(uploaded=True, status="current"),
+ ]
+ )
+ observed: list[dict] = []
+
+ def pending_then_current(*args, **kwargs): # noqa: ARG001
+ state = next(states)
+ observed.append(state)
+ return state
+
+ monkeypatch.setattr(jsonl_backup, "_icloud_item_state", pending_then_current)
+ result = jsonl_backup.copy_archive_to_icloud(second, icloud_dir, timeout_seconds=1, poll_interval_seconds=0)
+
+ assert len(observed) == 2
+ assert result["reused"] is True
+
+
+def test_existing_icloud_probe_cannot_restart_timeout_for_repair(tmp_path, monkeypatch):
+ from brainlayer import jsonl_backup
+
+ archive = tmp_path / "archive.tar.gz"
+ archive.write_bytes(gzip.compress(b"same tar payload", mtime=1))
+ icloud_dir = tmp_path / "CloudDocs"
+ icloud_dir.mkdir()
+ logical_sha256 = jsonl_backup._sha256_gzip_payload(archive)
+ destination = icloud_dir / f"claude-jsonl-{logical_sha256}.tar.gz"
+ destination.write_bytes(archive.read_bytes())
+ clock = [0.0]
+ calls = 0
+
+ def exhaust_deadline(*args, **kwargs): # noqa: ARG001
+ nonlocal calls
+ calls += 1
+ clock[0] = 2.0
+ raise RuntimeError("existing iCloud probe timed out")
+
+ monkeypatch.setattr(jsonl_backup.time, "monotonic", lambda: clock[0])
+ monkeypatch.setattr(jsonl_backup, "_icloud_item_state", exhaust_deadline)
+
+ with pytest.raises(RuntimeError, match="existing iCloud probe timed out"):
+ jsonl_backup.copy_archive_to_icloud(archive, icloud_dir, timeout_seconds=1)
+
+ assert calls == 1
+
+
+def test_existing_verified_object_is_not_quarantined_when_local_deadline_expires(tmp_path, monkeypatch):
+ from brainlayer import jsonl_backup
+
+ archive = tmp_path / "archive.tar.gz"
+ archive.write_bytes(gzip.compress(b"same tar payload", mtime=1))
+ icloud_dir = tmp_path / "CloudDocs"
+ icloud_dir.mkdir()
+ logical_sha256 = jsonl_backup._sha256_gzip_payload(archive)
+ destination = icloud_dir / f"claude-jsonl-{logical_sha256}.tar.gz"
+ destination.write_bytes(archive.read_bytes())
+ original_hash = jsonl_backup._sha256_gzip_payload
+
+ def expire_on_existing(path, **kwargs):
+ if Path(path) == destination:
+ raise jsonl_backup.ICloudDeadlineExceeded("deadline")
+ return original_hash(path, **kwargs)
+
+ monkeypatch.setattr(
+ jsonl_backup,
+ "_icloud_item_state",
+ lambda *args, **kwargs: _icloud_state(uploaded=True, status="current"),
+ )
+ monkeypatch.setattr(jsonl_backup, "_sha256_gzip_payload", expire_on_existing)
+
+ with pytest.raises(jsonl_backup.ICloudDeadlineExceeded):
+ jsonl_backup.copy_archive_to_icloud(archive, icloud_dir, timeout_seconds=1)
+
+ assert destination.is_file()
+ assert list(icloud_dir.glob("*.unverified")) == []
+
+
+def test_existing_verified_object_is_not_quarantined_when_polling_reaches_deadline(tmp_path, monkeypatch):
+ from brainlayer import jsonl_backup
+
+ archive = tmp_path / "archive.tar.gz"
+ archive.write_bytes(gzip.compress(b"same tar payload", mtime=1))
+ icloud_dir = tmp_path / "CloudDocs"
+ icloud_dir.mkdir()
+ logical_sha256 = jsonl_backup._sha256_gzip_payload(archive)
+ destination = icloud_dir / f"claude-jsonl-{logical_sha256}.tar.gz"
+ destination.write_bytes(archive.read_bytes())
+ clock = [0.0]
+
+ def pending_until_deadline(*args, **kwargs): # noqa: ARG001
+ clock[0] = 2.0
+ return _icloud_state(uploaded=True, status="notDownloaded")
+
+ monkeypatch.setattr(jsonl_backup.time, "monotonic", lambda: clock[0])
+ monkeypatch.setattr(jsonl_backup, "_icloud_item_state", pending_until_deadline)
+
+ with pytest.raises(jsonl_backup.ICloudDeadlineExceeded):
+ jsonl_backup.copy_archive_to_icloud(archive, icloud_dir, timeout_seconds=1)
+
+ assert destination.is_file()
+ assert list(icloud_dir.glob("*.unverified")) == []
+
+
+def test_existing_verified_object_is_not_quarantined_when_status_probe_times_out(tmp_path, monkeypatch):
+ from brainlayer import jsonl_backup
+
+ archive = tmp_path / "archive.tar.gz"
+ archive.write_bytes(gzip.compress(b"same tar payload", mtime=1))
+ icloud_dir = tmp_path / "CloudDocs"
+ icloud_dir.mkdir()
+ logical_sha256 = jsonl_backup._sha256_gzip_payload(archive)
+ destination = icloud_dir / f"claude-jsonl-{logical_sha256}.tar.gz"
+ destination.write_bytes(archive.read_bytes())
+ timeout = subprocess.TimeoutExpired(["osascript"], 1)
+ monkeypatch.setattr(
+ jsonl_backup,
+ "_icloud_item_state",
+ lambda *args, **kwargs: (_ for _ in ()).throw(timeout),
+ )
+
+ with pytest.raises(subprocess.TimeoutExpired):
+ jsonl_backup.copy_archive_to_icloud(archive, icloud_dir, timeout_seconds=1)
+
+ assert destination.is_file()
+ assert list(icloud_dir.glob("*.unverified")) == []
+
+
+@pytest.mark.parametrize("message", ["Foundation transport failed", "iCloud status returned invalid JSON"])
+def test_existing_verified_object_is_not_quarantined_when_status_probe_is_inconclusive(tmp_path, monkeypatch, message):
+ from brainlayer import jsonl_backup
+
+ archive = tmp_path / "archive.tar.gz"
+ archive.write_bytes(gzip.compress(b"same tar payload", mtime=1))
+ icloud_dir = tmp_path / "CloudDocs"
+ icloud_dir.mkdir()
+ logical_sha256 = jsonl_backup._sha256_gzip_payload(archive)
+ destination = icloud_dir / f"claude-jsonl-{logical_sha256}.tar.gz"
+ destination.write_bytes(archive.read_bytes())
+ error = jsonl_backup.ICloudProbeError(message)
+ monkeypatch.setattr(
+ jsonl_backup,
+ "_icloud_item_state",
+ lambda *args, **kwargs: (_ for _ in ()).throw(error),
+ )
+
+ with pytest.raises(jsonl_backup.ICloudProbeError, match=message):
+ jsonl_backup.copy_archive_to_icloud(archive, icloud_dir, timeout_seconds=1)
+
+ assert destination.is_file()
+ assert list(icloud_dir.glob("*.unverified")) == []
+
+
+def test_icloud_deadline_starts_before_first_archive_scan(tmp_path, monkeypatch):
+ from brainlayer import jsonl_backup
+
+ archive = tmp_path / "archive.tar.gz"
+ archive.write_bytes(gzip.compress(b"payload", mtime=1))
+ observed_deadlines: list[float | None] = []
+
+ def stop_first_scan(path, *, deadline=None): # noqa: ARG001
+ observed_deadlines.append(deadline)
+ raise RuntimeError("scan stopped")
+
+ monkeypatch.setattr(jsonl_backup.time, "monotonic", lambda: 10.0)
+ monkeypatch.setattr(jsonl_backup, "_sha256_file", stop_first_scan)
+
+ with pytest.raises(RuntimeError, match="scan stopped"):
+ jsonl_backup.copy_archive_to_icloud(archive, tmp_path / "CloudDocs", timeout_seconds=5)
+
+ assert observed_deadlines == [15.0]
+
+
+def test_icloud_copy_deadline_blocks_status_probe(tmp_path, monkeypatch):
+ from brainlayer import jsonl_backup
+
+ archive = tmp_path / "archive.tar.gz"
+ archive.write_bytes(b"payload")
+ clock = [0.0]
+ status_calls = 0
+ real_check = jsonl_backup._check_icloud_deadline
+
+ monkeypatch.setattr(jsonl_backup.time, "monotonic", lambda: clock[0])
+ monkeypatch.setattr(jsonl_backup, "_sha256_file", lambda path, **kwargs: "a" * 64)
+ monkeypatch.setattr(jsonl_backup, "_sha256_gzip_payload", lambda path, **kwargs: "b" * 64)
+
+ def expire_during_copy(deadline, phase):
+ if phase.startswith("copying "):
+ clock[0] = 2.0
+ real_check(deadline, phase)
+
+ def count_status(*args, **kwargs): # noqa: ARG001
+ nonlocal status_calls
+ status_calls += 1
+ return _icloud_state(uploaded=True, status="current")
+
+ monkeypatch.setattr(jsonl_backup, "_check_icloud_deadline", expire_during_copy)
+ monkeypatch.setattr(jsonl_backup, "_icloud_item_state", count_status)
+
+ with pytest.raises(RuntimeError, match="deadline exceeded during copying"):
+ jsonl_backup.copy_archive_to_icloud(archive, tmp_path / "CloudDocs", timeout_seconds=1)
+
+ assert status_calls == 0
+
+
+def test_icloud_retry_preserves_prior_verified_logical_object(tmp_path, monkeypatch):
+ from brainlayer import jsonl_backup
+
+ first = tmp_path / "first" / "claude-jsonl-2026-09-09.tar.gz"
+ second = tmp_path / "second" / first.name
+ first.parent.mkdir()
+ second.parent.mkdir()
+ first.write_bytes(gzip.compress(b"same tar payload", mtime=1))
+ second.write_bytes(gzip.compress(b"same tar payload", mtime=2))
+ icloud_dir = tmp_path / "CloudDocs"
+ monkeypatch.setattr(
+ jsonl_backup,
+ "_icloud_item_state",
+ lambda *args, **kwargs: _icloud_state(uploaded=True, status="current"),
+ )
+ first_result = jsonl_backup.copy_archive_to_icloud(first, icloud_dir, timeout_seconds=1)
+ destination = Path(first_result["path"])
+
+ def only_prior_object_is_verified(path, **kwargs): # noqa: ARG001
+ if Path(path).read_bytes() == first.read_bytes():
+ return _icloud_state(uploaded=True, status="current")
+ raise RuntimeError("replacement upload failed")
+
+ monkeypatch.setattr(jsonl_backup, "_icloud_item_state", only_prior_object_is_verified)
+ second_result = jsonl_backup.copy_archive_to_icloud(second, icloud_dir, timeout_seconds=1)
+
+ assert second_result["reused"] is True
+ assert destination.read_bytes() == first.read_bytes()
+ assert list(icloud_dir.iterdir()) == [destination]
+
+
+def test_stale_icloud_object_is_quarantined_before_fresh_upload(tmp_path, monkeypatch):
+ from brainlayer import jsonl_backup
+
+ archive = tmp_path / "archive.tar.gz"
+ archive.write_bytes(b"fresh archive")
+ logical_sha256 = hashlib.sha256(archive.read_bytes()).hexdigest()
+ icloud_dir = tmp_path / "CloudDocs"
+ icloud_dir.mkdir()
+ destination = icloud_dir / f"claude-jsonl-{logical_sha256}.tar.gz"
+ destination.write_bytes(b"stale object")
+ states = iter(
+ [
+ _icloud_state(uploaded=False, status="notDownloaded") | {"uploading_error": "stale"},
+ _icloud_state(uploaded=True, status="current"),
+ ]
+ )
+ monkeypatch.setattr(jsonl_backup, "_icloud_item_state", lambda *args, **kwargs: next(states))
+
+ result = jsonl_backup.copy_archive_to_icloud(archive, icloud_dir, timeout_seconds=1)
+
+ assert result["reused"] is False
+ assert destination.read_bytes() == archive.read_bytes()
+ quarantined = list(icloud_dir.glob("*.unverified"))
+ assert len(quarantined) == 1
+ assert quarantined[0].read_bytes() == b"stale object"
+
+
+def test_icloud_poll_sleep_cannot_overshoot_deadline(tmp_path, monkeypatch):
+ from brainlayer import jsonl_backup
+
+ archive = tmp_path / "archive.tar.gz"
+ archive.write_bytes(b"bytes")
+ clock = [0.0]
+ states = iter(
+ [_icloud_state(uploaded=False, status="notDownloaded"), _icloud_state(uploaded=True, status="current")]
+ )
+ sleeps: list[float] = []
+ monkeypatch.setattr(jsonl_backup.time, "monotonic", lambda: clock[0])
+ monkeypatch.setattr(jsonl_backup.time, "sleep", sleeps.append)
+
+ def pending_then_current(*args, **kwargs): # noqa: ARG001
+ clock[0] = 0.75
+ return next(states)
+
+ monkeypatch.setattr(jsonl_backup, "_icloud_item_state", pending_then_current)
+
+ jsonl_backup.copy_archive_to_icloud(archive, tmp_path / "CloudDocs", timeout_seconds=1, poll_interval_seconds=10)
+
+ assert sleeps == [0.25]
+
+
+def test_jsonl_backup_does_not_advance_state_until_icloud_copy_is_verified(tmp_path, monkeypatch):
+ from brainlayer import jsonl_backup
+
+ now = time.time()
+ source_root = tmp_path / "sessions"
+ source_file = _write_jsonl(source_root / "changed.jsonl", mtime=now - 3600)
+ state_path = tmp_path / "state.json"
+ drive_uploads: list[Path] = []
+
+ _mock_drive_success(jsonl_backup, monkeypatch, drive_uploads)
+ monkeypatch.setattr(
+ jsonl_backup,
+ "copy_archive_to_icloud",
+ lambda *args, **kwargs: (_ for _ in ()).throw(RuntimeError("iCloud verification failed")),
+ )
+
+ with pytest.raises(RuntimeError, match="iCloud verification failed"):
+ jsonl_backup.run_backup(
+ source_roots=[source_root],
+ state_path=state_path,
+ staging_dir=tmp_path / "staging",
+ log_path=tmp_path / "jsonl-backup.log",
+ queue_dir=tmp_path / "queue",
+ icloud_dir=tmp_path / "CloudDocs" / "Archives" / "brainlayer-jsonl-backups",
+ date_stamp="2026-09-09",
+ now=now,
+ upload=True,
+ )
+
+ assert not state_path.exists()
+ assert source_file.exists()
+ assert drive_uploads == []
+
+
+def test_run_backup_reuses_one_icloud_deadline_for_inventory_and_repair(tmp_path, monkeypatch):
+ from brainlayer import jsonl_backup
+
+ now = time.time()
+ source_root = tmp_path / "sessions"
+ _write_jsonl(source_root / "changed.jsonl", mtime=now - 3600)
+ icloud_dir = tmp_path / "CloudDocs"
+ observed: list[tuple[str, float | None]] = []
+
+ _mock_drive_success(jsonl_backup, monkeypatch)
+
+ def inventory(*args, deadline=None, **kwargs): # noqa: ARG001
+ observed.append(("inventory", deadline))
+ return False
+
+ def copy(archive, destination, *, deadline=None, **kwargs): # noqa: ARG001
+ observed.append(("copy", deadline))
+ return _copy_to_icloud_receipt(Path(archive), Path(destination))
+
+ monkeypatch.setattr(jsonl_backup, "_icloud_inventory_is_verified", inventory)
+ monkeypatch.setattr(jsonl_backup, "copy_archive_to_icloud", copy)
+
+ jsonl_backup.run_backup(
+ source_roots=[source_root],
+ state_path=tmp_path / "state.json",
+ staging_dir=tmp_path / "staging",
+ log_path=tmp_path / "backup.log",
+ queue_dir=tmp_path / "queue",
+ icloud_dir=icloud_dir,
+ now=now,
+ upload=True,
+ )
+
+ assert observed[0][0] == "inventory"
+ assert observed[1][0] == "copy"
+ assert observed[0][1] is not None
+ assert observed[0][1] == observed[1][1]
+
+
+def test_enabling_icloud_bootstraps_files_covered_only_by_legacy_drive_state(tmp_path, monkeypatch):
+ from brainlayer import jsonl_backup
+
+ now = time.time()
+ source_root = tmp_path / "sessions"
+ source_file = _write_jsonl(source_root / "legacy-covered.jsonl", mtime=now - 3600)
+ state_path = tmp_path / "state.json"
+ state_path.write_text(
+ json.dumps(
+ {
+ "files": {
+ source_file.as_posix(): {"mtime": source_file.stat().st_mtime, "size": source_file.stat().st_size}
+ }
+ }
+ )
+ )
+ icloud_dir = tmp_path / "CloudDocs" / "Archives" / "brainlayer-jsonl-backups"
+
+ _mock_drive_success(jsonl_backup, monkeypatch)
+ monkeypatch.setattr(
+ jsonl_backup,
+ "copy_archive_to_icloud",
+ _copy_to_icloud_receipt,
+ )
+
+ result = jsonl_backup.run_backup(
+ source_roots=[source_root],
+ state_path=state_path,
+ staging_dir=tmp_path / "staging",
+ log_path=tmp_path / "jsonl-backup.log",
+ queue_dir=tmp_path / "queue",
+ icloud_dir=icloud_dir,
+ date_stamp="2026-09-09",
+ now=now,
+ upload=True,
+ )
+
+ assert result["status"] == "uploaded"
+ assert result["bundled_file_count"] == 1
+ state = json.loads(state_path.read_text())
+ assert state["icloud_directory"] == str(icloud_dir)
+ assert state["icloud_verified"] is True
+ assert state["files"][source_file.as_posix()]["icloud_archive"]
+ assert state["icloud_archives"]
+
+
+def test_missing_recorded_icloud_archive_forces_opt_in_rebootstrap(tmp_path, monkeypatch):
+ from brainlayer import jsonl_backup
+
+ now = time.time()
+ source_root = tmp_path / "sessions"
+ source_file = _write_jsonl(source_root / "covered.jsonl", mtime=now - 3600)
+ icloud_dir = tmp_path / "CloudDocs" / "Archives" / "brainlayer-jsonl-backups"
+ missing_name = "claude-jsonl-missing.tar.gz"
+ state_path = tmp_path / "state.json"
+ state_path.write_text(
+ json.dumps(
+ {
+ "files": {
+ source_file.as_posix(): {
+ "mtime": source_file.stat().st_mtime,
+ "size": source_file.stat().st_size,
+ "sha256": hashlib.sha256(source_file.read_bytes()).hexdigest(),
+ "archive": "drive.tar.gz",
+ "archive_id": "drive-id",
+ "icloud_archive": missing_name,
+ }
+ },
+ "icloud_directory": str(icloud_dir),
+ "icloud_verified": True,
+ "icloud_archives": {
+ missing_name: {"bytes": 123, "sha256": "0" * 64},
+ },
+ }
+ )
+ )
+ copied: list[Path] = []
+
+ _mock_drive_success(jsonl_backup, monkeypatch)
+ monkeypatch.setattr(jsonl_backup, "_list_surviving_archives", lambda *args, **kwargs: {"drive-id": None})
+
+ def copy(archive, destination, **kwargs): # noqa: ARG001
+ copied.append(Path(archive))
+ return _copy_to_icloud_receipt(Path(archive), Path(destination))
+
+ monkeypatch.setattr(jsonl_backup, "copy_archive_to_icloud", copy)
+
+ result = jsonl_backup.run_backup(
+ source_roots=[source_root],
+ state_path=state_path,
+ staging_dir=tmp_path / "staging",
+ log_path=tmp_path / "jsonl-backup.log",
+ queue_dir=tmp_path / "queue",
+ icloud_dir=icloud_dir,
+ date_stamp="2026-09-09",
+ now=now,
+ upload=True,
+ )
+
+ assert result["status"] == "uploaded"
+ assert result["bundled_file_count"] == 1
+ assert len(copied) == 1
+
+
+def test_missing_icloud_archive_for_vanished_source_fails_loudly(tmp_path):
+ from brainlayer import jsonl_backup
+
+ now = time.time()
+ source_root = tmp_path / "sessions"
+ vanished = _write_jsonl(source_root / "vanished.jsonl", mtime=now - 3600)
+ stat = vanished.stat()
+ vanished.unlink()
+ icloud_dir = tmp_path / "CloudDocs"
+ state_path = tmp_path / "state.json"
+ state_path.write_text(
+ json.dumps(
+ {
+ "files": {
+ vanished.as_posix(): {
+ "mtime": stat.st_mtime,
+ "size": stat.st_size,
+ "icloud_archive": "missing.tar.gz",
+ }
+ },
+ "icloud_directory": str(icloud_dir),
+ "icloud_verified": True,
+ "icloud_archives": {"missing.tar.gz": {"bytes": 123, "sha256": "0" * 64}},
+ }
+ )
+ )
+
+ with pytest.raises(RuntimeError, match="source is unavailable"):
+ jsonl_backup.run_backup(
+ source_roots=[source_root],
+ state_path=state_path,
+ staging_dir=tmp_path / "staging",
+ log_path=tmp_path / "jsonl-backup.log",
+ queue_dir=tmp_path / "queue",
+ icloud_dir=icloud_dir,
+ date_stamp="2026-09-09",
+ now=now,
+ upload=True,
+ )
+
+
+def test_icloud_inventory_still_validates_archives_for_vanished_sources(tmp_path, monkeypatch):
+ from brainlayer import jsonl_backup
+
+ source_path = (tmp_path / "sessions" / "vanished.jsonl").as_posix()
+ icloud_dir = tmp_path / "CloudDocs"
+ icloud_dir.mkdir()
+ archive_name = "vanished-source.tar.gz"
+ archive = icloud_dir / archive_name
+ archive.write_bytes(b"only remaining copy")
+ probes: list[Path] = []
+
+ def probe(path, **kwargs):
+ probes.append(Path(path))
+ return _icloud_state(uploaded=True, status="current")
+
+ monkeypatch.setattr(jsonl_backup, "_icloud_item_state", probe)
+ state = {
+ "files": {
+ source_path: {
+ "mtime": 1.0,
+ "size": 10,
+ "icloud_archive": archive_name,
+ }
+ },
+ "icloud_directory": str(icloud_dir),
+ "icloud_verified": True,
+ "icloud_archives": {
+ archive_name: {
+ "bytes": archive.stat().st_size,
+ "sha256": hashlib.sha256(archive.read_bytes()).hexdigest(),
+ }
+ },
+ }
+ hash_deadlines: list[float | None] = []
+ original_hash = jsonl_backup._sha256_file
+
+ def hash_with_deadline(path, *, deadline=None):
+ hash_deadlines.append(deadline)
+ return original_hash(path, deadline=deadline)
+
+ monkeypatch.setattr(jsonl_backup, "_sha256_file", hash_with_deadline)
+
+ assert jsonl_backup._icloud_inventory_is_verified(state, [], icloud_dir, timeout_seconds=1)
+ assert probes == [archive]
+ assert len(hash_deadlines) == 1
+ assert hash_deadlines[0] is not None
+
+
+def test_icloud_inventory_preserves_later_valid_receipts_after_one_archive_fails(tmp_path, monkeypatch):
+ from brainlayer import jsonl_backup
+
+ icloud_dir = tmp_path / "CloudDocs"
+ icloud_dir.mkdir()
+ source_root = tmp_path / "sessions"
+ first = _write_jsonl(source_root / "first.jsonl", mtime=1.0)
+ second = _write_jsonl(source_root / "second.jsonl", mtime=1.0)
+ candidates = jsonl_backup._discover_jsonl_candidates([source_root])
+ good_archive = icloud_dir / "z-good.tar.gz"
+ good_archive.write_bytes(b"verified archive")
+ state = {
+ "files": {
+ first.as_posix(): {
+ "mtime": first.stat().st_mtime,
+ "size": first.stat().st_size,
+ "icloud_archive": "a-missing.tar.gz",
+ },
+ second.as_posix(): {
+ "mtime": second.stat().st_mtime,
+ "size": second.stat().st_size,
+ "icloud_archive": good_archive.name,
+ },
+ },
+ "icloud_directory": str(icloud_dir),
+ "icloud_verified": True,
+ "icloud_archives": {
+ "a-missing.tar.gz": {"bytes": 1, "sha256": "0" * 64},
+ good_archive.name: {
+ "bytes": good_archive.stat().st_size,
+ "sha256": hashlib.sha256(good_archive.read_bytes()).hexdigest(),
+ },
+ },
+ }
+ validated_sources: set[str] = set()
+ monkeypatch.setattr(
+ jsonl_backup,
+ "_icloud_item_state",
+ lambda *args, **kwargs: _icloud_state(uploaded=True, status="current"),
+ )
+
+ assert not jsonl_backup._icloud_inventory_is_verified(
+ state,
+ candidates,
+ icloud_dir,
+ timeout_seconds=1,
+ validated_sources=validated_sources,
+ )
+ assert validated_sources == {second.as_posix()}
+
+
+def test_icloud_inventory_stops_when_receipt_hash_exhausts_shared_deadline(tmp_path, monkeypatch):
+ from brainlayer import jsonl_backup
+
+ source_root = tmp_path / "sessions"
+ source = _write_jsonl(source_root / "covered.jsonl", mtime=1.0)
+ candidates = jsonl_backup._discover_jsonl_candidates([source_root])
+ icloud_dir = tmp_path / "CloudDocs"
+ icloud_dir.mkdir()
+ archive = icloud_dir / "covered.tar.gz"
+ archive.write_bytes(b"verified archive")
+ state = {
+ "files": {
+ source.as_posix(): {
+ "mtime": source.stat().st_mtime,
+ "size": source.stat().st_size,
+ "icloud_archive": archive.name,
+ }
+ },
+ "icloud_directory": str(icloud_dir),
+ "icloud_verified": True,
+ "icloud_archives": {
+ archive.name: {
+ "bytes": archive.stat().st_size,
+ "sha256": hashlib.sha256(archive.read_bytes()).hexdigest(),
+ }
+ },
+ }
+ clock = [0.0]
+
+ def exhaust_during_probe(*args, **kwargs): # noqa: ARG001
+ clock[0] = 2.0
+ return _icloud_state(uploaded=True, status="current")
+
+ monkeypatch.setattr(jsonl_backup.time, "monotonic", lambda: clock[0])
+ monkeypatch.setattr(jsonl_backup, "_icloud_item_state", exhaust_during_probe)
+
+ with pytest.raises(jsonl_backup.ICloudDeadlineExceeded):
+ jsonl_backup._icloud_inventory_is_verified(
+ state,
+ candidates,
+ icloud_dir,
+ timeout_seconds=1,
+ )
+
+
+def test_icloud_inventory_ignores_vanished_legacy_entry_without_icloud_receipt(tmp_path):
+ from brainlayer import jsonl_backup
+
+ icloud_dir = tmp_path / "CloudDocs"
+ state = {
+ "files": {
+ (tmp_path / "vanished.jsonl").as_posix(): {
+ "mtime": 1.0,
+ "size": 10,
+ "archive": "drive-only.tar.gz",
+ "archive_id": "drive-id",
+ }
+ },
+ "icloud_directory": str(icloud_dir),
+ "icloud_verified": True,
+ "icloud_archives": {},
+ }
+
+ assert jsonl_backup._icloud_inventory_is_verified(state, [], icloud_dir)
+
+
+def test_icloud_inventory_rehydrates_placeholder_and_checks_exact_receipt(tmp_path, monkeypatch):
+ from brainlayer import jsonl_backup
+
+ now = time.time()
+ source_root = tmp_path / "sessions"
+ source_file = _write_jsonl(source_root / "covered.jsonl", mtime=now - 3600)
+ candidate = jsonl_backup._discover_jsonl_candidates([source_root])[0]
+ icloud_dir = tmp_path / "CloudDocs"
+ icloud_dir.mkdir()
+ archive_name = "claude-jsonl-covered.tar.gz"
+ archive_bytes = b"durable iCloud archive"
+ placeholder = icloud_dir / f".{archive_name}.icloud"
+ placeholder.write_bytes(b"")
+ calls: list[Path] = []
+
+ def materialize(path, *, request_download=False, timeout_seconds=None):
+ calls.append(Path(path))
+ assert request_download is True
+ assert timeout_seconds is not None and timeout_seconds > 0
+ placeholder.unlink()
+ (icloud_dir / archive_name).write_bytes(archive_bytes)
+ return _icloud_state(uploaded=True, status="current")
+
+ monkeypatch.setattr(jsonl_backup, "_icloud_item_state", materialize)
+ state = {
+ "files": {
+ source_file.as_posix(): {
+ "mtime": source_file.stat().st_mtime,
+ "size": source_file.stat().st_size,
+ "icloud_archive": archive_name,
+ }
+ },
+ "icloud_directory": str(icloud_dir),
+ "icloud_verified": True,
+ "icloud_archives": {
+ archive_name: {
+ "bytes": len(archive_bytes),
+ "sha256": hashlib.sha256(archive_bytes).hexdigest(),
+ }
+ },
+ }
+
+ assert jsonl_backup._icloud_inventory_is_verified(state, [candidate], icloud_dir, timeout_seconds=1)
+ assert calls == [placeholder]
+
+
+def test_icloud_inventory_does_not_swallow_wall_clock_timeout(tmp_path, monkeypatch):
+ from brainlayer import jsonl_backup
+
+ source = _write_jsonl(tmp_path / "source.jsonl", mtime=time.time() - 3600)
+ candidate = jsonl_backup.JsonlCandidate(
+ path=source,
+ root=tmp_path,
+ root_index=0,
+ mtime=source.stat().st_mtime,
+ size=source.stat().st_size,
+ )
+ icloud_dir = tmp_path / "CloudDocs"
+ icloud_dir.mkdir()
+ archive_name = "archive.tar.gz"
+ (icloud_dir / archive_name).write_bytes(b"archive")
+ state = {
+ "files": {
+ source.as_posix(): {
+ "mtime": candidate.mtime,
+ "size": candidate.size,
+ "icloud_archive": archive_name,
+ }
+ },
+ "icloud_directory": str(icloud_dir),
+ "icloud_verified": True,
+ "icloud_archives": {archive_name: {"bytes": 7, "sha256": "0" * 64}},
+ }
+ monkeypatch.setattr(
+ jsonl_backup,
+ "_icloud_item_state",
+ lambda *args, **kwargs: (_ for _ in ()).throw(jsonl_backup.backup_daily.BackupTimeoutError("deadline")),
+ )
+
+ with pytest.raises(jsonl_backup.backup_daily.BackupTimeoutError, match="deadline"):
+ jsonl_backup._icloud_inventory_is_verified(state, [candidate], icloud_dir)
+
+
+def test_icloud_inventory_quarantines_known_bad_materialized_archive(tmp_path, monkeypatch):
+ from brainlayer import jsonl_backup
+
+ source = _write_jsonl(tmp_path / "source.jsonl", mtime=time.time() - 3600)
+ candidate = jsonl_backup.JsonlCandidate(
+ path=source,
+ root=tmp_path,
+ root_index=0,
+ mtime=source.stat().st_mtime,
+ size=source.stat().st_size,
+ )
+ icloud_dir = tmp_path / "CloudDocs"
+ icloud_dir.mkdir()
+ archive_name = "archive.tar.gz"
+ archive = icloud_dir / archive_name
+ archive.write_bytes(b"corrupt archive")
+ state = {
+ "files": {
+ source.as_posix(): {
+ "mtime": candidate.mtime,
+ "size": candidate.size,
+ "icloud_archive": archive_name,
+ }
+ },
+ "icloud_directory": str(icloud_dir),
+ "icloud_verified": True,
+ "icloud_archives": {
+ archive_name: {
+ "bytes": len(b"expected archive"),
+ "sha256": hashlib.sha256(b"expected archive").hexdigest(),
+ }
+ },
+ }
+ monkeypatch.setattr(
+ jsonl_backup,
+ "_icloud_item_state",
+ lambda *args, **kwargs: _icloud_state(uploaded=True, status="current"),
+ )
+
+ assert not jsonl_backup._icloud_inventory_is_verified(state, [candidate], icloud_dir)
+ assert not archive.exists()
+ quarantined = list(icloud_dir.glob("*.unverified"))
+ assert len(quarantined) == 1
+ assert quarantined[0].read_bytes() == b"corrupt archive"
+
+
+def test_drive_only_change_invalidates_that_sources_icloud_receipt(tmp_path):
+ from brainlayer import jsonl_backup
+
+ now = time.time()
+ source_file = _write_jsonl(tmp_path / "changed.jsonl", mtime=now - 3600)
+ candidate = jsonl_backup.JsonlCandidate(
+ path=source_file,
+ root=tmp_path,
+ root_index=0,
+ mtime=source_file.stat().st_mtime,
+ size=source_file.stat().st_size,
+ )
+ state = {
+ "files": {
+ source_file.as_posix(): {
+ "mtime": candidate.mtime - 1,
+ "size": candidate.size,
+ "icloud_archive": "old.tar.gz",
+ }
+ },
+ "icloud_directory": str(tmp_path / "CloudDocs"),
+ "icloud_verified": True,
+ "icloud_archives": {"old.tar.gz": {"bytes": 1, "sha256": "0" * 64}},
+ }
+
+ updated = jsonl_backup._update_state_for_uploaded(
+ state,
+ [candidate],
+ "drive.tar.gz",
+ archive_id="drive-id",
+ icloud_dir=None,
+ )
+
+ assert updated["icloud_verified"] is True
+ assert "icloud_archive" not in updated["files"][source_file.as_posix()]
+ assert updated["files"][source_file.as_posix()]["icloud_required"] is True
+ assert "icloud_archives" not in updated
+ assert not jsonl_backup._icloud_inventory_is_verified(updated, [candidate], tmp_path / "CloudDocs")
+
+
+def test_vanished_drive_only_update_fails_icloud_revalidation_loudly(tmp_path):
+ from brainlayer import jsonl_backup
+
+ source_path = (tmp_path / "vanished.jsonl").as_posix()
+ state = {
+ "files": {source_path: {"mtime": 1.0, "size": 10, "icloud_required": True}},
+ "icloud_directory": str(tmp_path / "CloudDocs"),
+ "icloud_verified": True,
+ }
+
+ with pytest.raises(RuntimeError, match="required source is unavailable"):
+ jsonl_backup._icloud_inventory_is_verified(state, [], tmp_path / "CloudDocs")
+
+
+def test_changed_icloud_directory_fails_when_prior_covered_source_is_unavailable(tmp_path):
+ from brainlayer import jsonl_backup
+
+ source_path = (tmp_path / "vanished.jsonl").as_posix()
+ state = {
+ "files": {
+ source_path: {
+ "mtime": 1.0,
+ "size": 10,
+ "icloud_archive": "prior.tar.gz",
+ }
+ },
+ "icloud_directory": str(tmp_path / "OldCloudDocs"),
+ "icloud_verified": True,
+ "icloud_archives": {"prior.tar.gz": {"bytes": 10, "sha256": "0" * 64}},
+ }
+
+ with pytest.raises(RuntimeError, match="new destination cannot be seeded"):
+ jsonl_backup._icloud_inventory_is_verified(state, [], tmp_path / "NewCloudDocs")
+
+
+def test_icloud_bootstrap_with_active_source_does_not_mark_complete(tmp_path, monkeypatch):
+ from brainlayer import jsonl_backup
+
+ now = time.time()
+ source_root = tmp_path / "sessions"
+ inactive = _write_jsonl(source_root / "inactive.jsonl", mtime=now - 3600)
+ active = _write_jsonl(source_root / "active.jsonl", mtime=now - 60)
+ state_path = tmp_path / "state.json"
+ state_path.write_text(
+ json.dumps(
+ {
+ "files": {
+ path.as_posix(): {"mtime": path.stat().st_mtime, "size": path.stat().st_size}
+ for path in (inactive, active)
+ }
+ }
+ )
+ )
+ icloud_dir = tmp_path / "CloudDocs" / "Archives" / "brainlayer-jsonl-backups"
+
+ _mock_drive_success(jsonl_backup, monkeypatch)
+ monkeypatch.setattr(
+ jsonl_backup,
+ "copy_archive_to_icloud",
+ _copy_to_icloud_receipt,
+ )
+
+ result = jsonl_backup.run_backup(
+ source_roots=[source_root],
+ state_path=state_path,
+ staging_dir=tmp_path / "staging",
+ log_path=tmp_path / "jsonl-backup.log",
+ queue_dir=tmp_path / "queue",
+ icloud_dir=icloud_dir,
+ date_stamp="2026-09-09",
+ now=now,
+ upload=True,
+ )
+
+ assert result["bundled_file_count"] == 1
+ assert result["skipped_active_count"] == 1
+ state = json.loads(state_path.read_text())
+ assert state["icloud_directory"] == str(icloud_dir)
+ assert "icloud_verified" not in state
+ candidates = jsonl_backup._discover_jsonl_candidates([source_root])
+ validated_sources: set[str] = set()
+ monkeypatch.setattr(
+ jsonl_backup,
+ "_icloud_item_state",
+ lambda *args, **kwargs: _icloud_state(uploaded=True, status="current"),
+ )
+
+ assert not jsonl_backup._icloud_inventory_is_verified(
+ state,
+ candidates,
+ icloud_dir,
+ validated_sources=validated_sources,
+ )
+ assert validated_sources == {inactive.as_posix()}
+
+ first_archive = state["files"][inactive.as_posix()]["icloud_archive"]
+ monkeypatch.setattr(jsonl_backup, "_list_surviving_archives", lambda *args, **kwargs: {"drive-id": None})
+ second = jsonl_backup.run_backup(
+ source_roots=[source_root],
+ state_path=state_path,
+ staging_dir=tmp_path / "staging",
+ log_path=tmp_path / "jsonl-backup.log",
+ queue_dir=tmp_path / "queue",
+ icloud_dir=icloud_dir,
+ date_stamp="2026-09-10",
+ now=now + 3600,
+ upload=True,
+ )
+
+ assert second["bundled_file_count"] == 1
+ completed_state = json.loads(state_path.read_text())
+ assert completed_state["files"][inactive.as_posix()]["icloud_archive"] == first_archive
+ assert completed_state["files"][active.as_posix()]["icloud_archive"] != first_archive
+ assert completed_state["icloud_verified"] is True
+
+
+def test_invalid_icloud_inventory_with_only_active_sources_defers_instead_of_verifying(tmp_path, monkeypatch):
+ from brainlayer import jsonl_backup
+
+ now = time.time()
+ source_root = tmp_path / "sessions"
+ source = _write_jsonl(source_root / "active.jsonl", mtime=now - 60)
+ state_path = tmp_path / "state.json"
+ state_path.write_text(
+ json.dumps(
+ {
+ "files": {
+ source.as_posix(): {
+ "mtime": source.stat().st_mtime,
+ "size": source.stat().st_size,
+ }
+ }
+ }
+ )
+ )
+
+ result = jsonl_backup.run_backup(
+ source_roots=[source_root],
+ state_path=state_path,
+ staging_dir=tmp_path / "staging",
+ log_path=tmp_path / "jsonl-backup.log",
+ queue_dir=tmp_path / "queue",
+ icloud_dir=tmp_path / "CloudDocs" / "Archives" / "brainlayer-jsonl-backups",
+ date_stamp="2026-09-09",
+ now=now,
+ upload=True,
+ )
+
+ assert result["status"] == "deferred"
+ assert result["verified"] is False
+ assert result["uploaded"] is False
+ assert result["skipped_active_count"] == 1
+ assert "iCloud coverage" in result["error"]
+ assert json.loads(state_path.read_text())["files"][source.as_posix()]["size"] == source.stat().st_size
+
+
+def test_drive_only_state_update_preserves_prior_icloud_coverage():
+ from brainlayer import jsonl_backup
+
+ state = {
+ "files": {},
+ "icloud_directory": "/CloudDocs/Archives/brainlayer-jsonl-backups",
+ "icloud_verified": True,
+ }
+
+ updated = jsonl_backup._update_state_for_uploaded(state, [], icloud_dir=None)
+
+ assert updated["icloud_directory"] == state["icloud_directory"]
+ assert updated["icloud_verified"] is True
+
+
def test_default_source_roots_append_all_agent_cli_transcript_roots(monkeypatch):
from brainlayer import jsonl_backup
@@ -1259,14 +2608,26 @@ def test_jsonl_backup_launchd_plist_and_docstring_install_note_are_committed():
assert "0" in plist
assert "0" in script_plist
assert "BRAINLAYER_BACKUP_TIMEOUT_SECONDS" in plist
+ assert "__BRAINLAYER_PYTHON__" in plist
assert "BRAINLAYER_BACKUP_TIMEOUT_SECONDS" in wrapper
+ assert "BRAINLAYER_JSONL_BACKUP_ICLOUD_DIR" not in plist
+ assert "BRAINLAYER_JSONL_BACKUP_ICLOUD_DIR" not in script_plist
+ assert "BRAINLAYER_JSONL_BACKUP_ICLOUD_DIR" in module
+ assert "Archives/claude-sessions" not in plist
+ assert "Archives/claude-sessions" not in script_plist
assert "1800" in plist
assert "1800" in wrapper
assert ".local/share/brainlayer/logs/jsonl-backup.log" in plist
assert "jsonl-backup" in install
assert "install_jsonl_backup_script" in install
- assert "__BRAINLAYER_DIR_VALUE__" in wrapper
- assert "PYTHONPATH" in wrapper
+ assert "HOOK_PYTHON_RESOLVER" in install
+ assert "resolve_jsonl_backup_python || return 1" in install
+ assert 'runpy.run_path(path, run_name="__main__")' in install
+ assert "--print-interpreter" in install
+ assert "__BRAINLAYER_DIR_VALUE__" not in wrapper
+ assert '"${BRAINLAYER_PYTHON:?' in wrapper
+ assert "unset PYTHONPATH" in wrapper
+ assert "PYTHONPATH" not in plist
assert "__HOME__/.local/lib/brainlayer/jsonl-backup.sh" in script_plist
assert "SoftResourceLimits" in plist
assert "SoftResourceLimits" in script_plist
diff --git a/tests/test_launchd_hygiene.py b/tests/test_launchd_hygiene.py
index bbdc6742..17c25e54 100644
--- a/tests/test_launchd_hygiene.py
+++ b/tests/test_launchd_hygiene.py
@@ -517,6 +517,7 @@ def test_launchd_installer_rejects_key_only_enrichment_config(tmp_path):
"HOME": str(tmp_path),
"BRAINLAYER_BIN": "/usr/bin/true",
"PYTHON_BIN": "/usr/bin/true",
+ "BRAINLAYER_PYTHON": "/usr/bin/true",
"BRAINLAYER_ENV_FILE": str(env_file),
},
capture_output=True,