solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4from __future__ import annotations
5
6import hashlib
7import json
8from pathlib import Path
9from typing import Any
10from unittest.mock import Mock
11
12import pytest
13
14import solstone.think.offload as offload
15import solstone.think.offload_ledger as offload_ledger
16import solstone.think.utils as think_utils
17from solstone.think.backup import engine
18from solstone.think.backup.runner import ResticResult
19from solstone.think.offload_ledger import (
20 OffloadFile,
21 append_offload_event,
22 summarize_segment,
23)
24from solstone.think.utils import parse_duration_seconds
25
26GB = 1_000_000_000
27
28
29def _use_journal(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
30 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
31 think_utils._journal_path_cache = None
32 return tmp_path
33
34
35def _config_path(journal: Path) -> Path:
36 return journal / "config" / "journal.json"
37
38
39def _write_config(journal: Path, payload: dict[str, Any]) -> None:
40 config_path = _config_path(journal)
41 config_path.parent.mkdir(parents=True, exist_ok=True)
42 config_path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
43
44
45def _read_config(journal: Path) -> dict[str, Any]:
46 return json.loads(_config_path(journal).read_text(encoding="utf-8"))
47
48
49def _ready_config(
50 *,
51 now: int,
52 budget_bytes: int | None = 1,
53 floor_bytes: int | None = None,
54 last_ok_delta: int = 60,
55 verification_status: str = "ok",
56 verification_reason: str | None = None,
57 last_offload: dict[str, Any] | None = None,
58) -> dict[str, Any]:
59 backup: dict[str, Any] = {
60 "enabled": True,
61 "mode": "byo",
62 "destination": {
63 "repository": "s3:safe-bucket/path",
64 "backend": "s3",
65 "credentials": {
66 "access_key_id": "access-key",
67 "secret_access_key": "secret-key",
68 },
69 },
70 "daily_key": "daily-secret",
71 "recovery_key": "R" * 64,
72 "offload": {
73 "enabled": True,
74 "budget_bytes": budget_bytes,
75 "floor_bytes": floor_bytes,
76 },
77 "last_backup": {
78 "time": now - 120,
79 "snapshot_id": "full-snapshot",
80 "status": "ok",
81 "error_reason": None,
82 },
83 "last_verification": {
84 "time": now - last_ok_delta,
85 "status": verification_status,
86 "reason": verification_reason,
87 "last_ok_time": now - last_ok_delta,
88 "checked_subset": "7/52" if verification_status == "ok" else None,
89 },
90 }
91 if last_offload is not None:
92 backup["last_offload"] = last_offload
93 return {"backup": backup}
94
95
96def _make_segment(
97 journal: Path,
98 *,
99 day: str = "20260101",
100 stream: str = "default",
101 segment: str = "120000_300",
102 raw_name: str = "audio.wav",
103 content: bytes = b"raw-media",
104 size: int | None = None,
105 complete: bool = True,
106 failed: bool = False,
107) -> Path:
108 seg_path = journal / "chronicle" / day / stream / segment
109 seg_path.mkdir(parents=True, exist_ok=True)
110 raw_path = seg_path / raw_name
111 if size is None:
112 raw_path.write_bytes(content)
113 else:
114 with raw_path.open("wb") as handle:
115 handle.truncate(size)
116 if complete:
117 if failed:
118 header = {"_solstone_processing": {"state": "failed"}}
119 (seg_path / "audio.jsonl").write_text(
120 json.dumps(header) + "\n", encoding="utf-8"
121 )
122 else:
123 (seg_path / "audio.jsonl").write_text(
124 '{}\n{"start": 0}\n', encoding="utf-8"
125 )
126 (seg_path / "notes.jsonl").write_text("{}\n", encoding="utf-8")
127 talents = seg_path / "talents"
128 talents.mkdir()
129 (talents / "keep.json").write_text("{}\n", encoding="utf-8")
130 return seg_path
131
132
133def _ok_confirm(
134 snapshot_id: str,
135 expected_sizes: dict[Path, int],
136) -> engine.ArchiveCheckResult:
137 return engine.ArchiveCheckResult(
138 status="ok",
139 error_reason=None,
140 verdicts=tuple(
141 engine.ArchiveFileVerdict(
142 path=str(path),
143 confirmed=True,
144 expected_size=size,
145 observed_size=size,
146 snapshot_id=snapshot_id,
147 )
148 for path, size in expected_sizes.items()
149 ),
150 )
151
152
153def _install_successful_archive(
154 monkeypatch: pytest.MonkeyPatch,
155) -> tuple[list[tuple[Path, ...]], list[dict[Path, int]]]:
156 archive_calls: list[tuple[Path, ...]] = []
157 confirm_calls: list[dict[Path, int]] = []
158
159 def fake_archive(paths: list[Path]) -> engine.BackupResult:
160 archive_calls.append(tuple(paths))
161 return engine.BackupResult(
162 status="ok",
163 snapshot_id=f"archive-{len(archive_calls)}",
164 error_reason=None,
165 )
166
167 def fake_confirm(
168 snapshot_id: str,
169 expected_sizes: dict[Path, int],
170 ) -> engine.ArchiveCheckResult:
171 confirm_calls.append(dict(expected_sizes))
172 return _ok_confirm(snapshot_id, expected_sizes)
173
174 monkeypatch.setattr(offload, "run_archive_backup", fake_archive)
175 monkeypatch.setattr(offload, "check_archive_snapshot_files", fake_confirm)
176 monkeypatch.setattr(offload, "request_verification_now", Mock(return_value=True))
177 return archive_calls, confirm_calls
178
179
180def _assert_no_offload_side_effects(journal: Path, raw_path: Path) -> None:
181 assert raw_path.exists()
182 assert not (journal / "health" / "offload").exists()
183 assert not (journal / "health" / "pruning-runs").exists()
184
185
186def _pruning_records(journal: Path) -> list[dict[str, Any]]:
187 records: list[dict[str, Any]] = []
188 for path in sorted((journal / "health" / "pruning-runs").glob("*.jsonl")):
189 records.extend(
190 json.loads(line)
191 for line in path.read_text(encoding="utf-8").splitlines()
192 if line.strip()
193 )
194 return records
195
196
197def test_offload_max_runtime_exceeds_retry_lock_and_archive_timeout() -> None:
198 assert parse_duration_seconds(offload.OFFLOAD_MAX_RUNTIME) > (
199 parse_duration_seconds(engine.ARCHIVE_RETRY_LOCK)
200 )
201 assert parse_duration_seconds(offload.OFFLOAD_MAX_RUNTIME) > (
202 engine.ARCHIVE_BACKUP_TIMEOUT_SECONDS
203 )
204
205
206def test_offload_disabled_skips_without_recording_or_archiving(
207 tmp_path: Path,
208 monkeypatch: pytest.MonkeyPatch,
209) -> None:
210 journal = _use_journal(tmp_path, monkeypatch)
211 now = 1_800_000_000
212 config = _ready_config(now=now)
213 config["backup"]["offload"]["enabled"] = False
214 _write_config(journal, config)
215 raw_path = _make_segment(journal).joinpath("audio.wav")
216 archive = Mock()
217 monkeypatch.setattr(offload, "run_archive_backup", archive)
218
219 result = offload.run_offload()
220
221 assert result.status == "skipped"
222 archive.assert_not_called()
223 assert "last_offload" not in _read_config(journal)["backup"]
224 assert raw_path.exists()
225
226
227@pytest.mark.parametrize(
228 ("mutate", "expected_reason", "request_expected"),
229 [
230 (
231 lambda backup, now: backup.update({"enabled": False}),
232 "backup_not_ready",
233 False,
234 ),
235 (
236 lambda backup, now: backup["last_backup"].update(
237 {"status": "error", "error_reason": "incomplete", "snapshot_id": None}
238 ),
239 "backup_failing",
240 False,
241 ),
242 (
243 lambda backup, now: backup["last_verification"].update(
244 {
245 "time": None,
246 "status": None,
247 "reason": None,
248 "last_ok_time": None,
249 "checked_subset": None,
250 }
251 ),
252 "verification_missing",
253 True,
254 ),
255 (
256 lambda backup, now: backup["last_verification"].update(
257 {"last_ok_time": now - offload.VERIFICATION_MAX_AGE_SECONDS - 1}
258 ),
259 "verification_overdue",
260 True,
261 ),
262 (
263 lambda backup, now: backup["last_verification"].update(
264 {
265 "status": "error",
266 "reason": "integrity_failed",
267 "last_ok_time": now - offload.VERIFICATION_MAX_AGE_SECONDS - 1,
268 }
269 ),
270 "verification_failed",
271 False,
272 ),
273 ],
274)
275def test_precondition_stalls_happen_before_archive_and_request_only_when_needed(
276 tmp_path: Path,
277 monkeypatch: pytest.MonkeyPatch,
278 mutate,
279 expected_reason: str,
280 request_expected: bool,
281) -> None:
282 journal = _use_journal(tmp_path, monkeypatch)
283 now = 1_800_000_000
284 monkeypatch.setattr(offload.time, "time", lambda: now)
285 config = _ready_config(now=now)
286 mutate(config["backup"], now)
287 _write_config(journal, config)
288 raw_path = _make_segment(journal).joinpath("audio.wav")
289 archive = Mock()
290 confirm = Mock()
291 request = Mock(return_value=True)
292 monkeypatch.setattr(offload, "run_archive_backup", archive)
293 monkeypatch.setattr(offload, "check_archive_snapshot_files", confirm)
294 monkeypatch.setattr(offload, "request_verification_now", request)
295
296 result = offload.run_offload()
297
298 assert result.status == "stalled"
299 assert result.reason == expected_reason
300 archive.assert_not_called()
301 confirm.assert_not_called()
302 assert request.called is request_expected
303 assert _read_config(journal)["backup"]["last_offload"]["reason"] == expected_reason
304 assert raw_path.exists()
305
306
307@pytest.mark.parametrize(
308 (
309 "verification_status",
310 "verification_reason",
311 "last_ok_delta",
312 "expected_reason",
313 "request_expected",
314 "archive_expected",
315 ),
316 [
317 ("ok", None, offload.VERIFICATION_MAX_AGE_SECONDS, None, False, True),
318 (
319 "ok",
320 None,
321 offload.VERIFICATION_MAX_AGE_SECONDS + 1,
322 "verification_overdue",
323 True,
324 False,
325 ),
326 (
327 "error",
328 "integrity_failed",
329 offload.VERIFICATION_MAX_AGE_SECONDS + 1,
330 "verification_failed",
331 False,
332 False,
333 ),
334 ("error", "locked", 60, None, False, True),
335 ("error", "timeout", 60, None, False, True),
336 ],
337)
338def test_verification_gate_uses_last_success_freshness_and_integrity_precedence(
339 tmp_path: Path,
340 monkeypatch: pytest.MonkeyPatch,
341 verification_status: str,
342 verification_reason: str | None,
343 last_ok_delta: int,
344 expected_reason: str | None,
345 request_expected: bool,
346 archive_expected: bool,
347) -> None:
348 journal = _use_journal(tmp_path, monkeypatch)
349 now = 1_800_000_000
350 monkeypatch.setattr(offload.time, "time", lambda: now)
351 _write_config(
352 journal,
353 _ready_config(
354 now=now,
355 verification_status=verification_status,
356 verification_reason=verification_reason,
357 last_ok_delta=last_ok_delta,
358 ),
359 )
360 _make_segment(journal, content=b"abcdef")
361 archive_calls, _confirm_calls = _install_successful_archive(monkeypatch)
362 request = Mock(return_value=True)
363 monkeypatch.setattr(offload, "request_verification_now", request)
364
365 result = offload.run_offload()
366
367 assert result.reason == expected_reason
368 assert bool(archive_calls) is archive_expected
369 assert request.called is request_expected
370
371
372def test_dry_run_selects_without_hashing_or_side_effects(
373 tmp_path: Path,
374 monkeypatch: pytest.MonkeyPatch,
375) -> None:
376 journal = _use_journal(tmp_path, monkeypatch)
377 now = 1_800_000_000
378 _write_config(journal, _ready_config(now=now, budget_bytes=1))
379 raw_path = _make_segment(journal, content=b"abcdef").joinpath("audio.wav")
380 before_config = _config_path(journal).read_bytes()
381 archive = Mock()
382 confirm = Mock()
383 request = Mock(return_value=True)
384
385 def fail_hash():
386 raise AssertionError("dry-run must not hash raw media")
387
388 monkeypatch.setattr(offload, "run_archive_backup", archive)
389 monkeypatch.setattr(offload, "check_archive_snapshot_files", confirm)
390 monkeypatch.setattr(offload, "request_verification_now", request)
391 monkeypatch.setattr(offload.hashlib, "sha256", fail_hash)
392
393 result = offload.run_offload(dry_run=True)
394
395 assert result.status == "ok"
396 assert result.dry_run is True
397 assert result.files_offloaded == 0
398 assert result.bytes_offloaded == 0
399 assert result.details == (
400 offload.OffloadSegmentDetail(
401 day="20260101",
402 stream="default",
403 segment="120000_300",
404 files=1,
405 bytes=6,
406 ),
407 )
408 archive.assert_not_called()
409 confirm.assert_not_called()
410 request.assert_not_called()
411 assert _config_path(journal).read_bytes() == before_config
412 _assert_no_offload_side_effects(journal, raw_path)
413
414
415def test_dry_run_reports_precondition_stall_without_request_or_record(
416 tmp_path: Path,
417 monkeypatch: pytest.MonkeyPatch,
418) -> None:
419 journal = _use_journal(tmp_path, monkeypatch)
420 now = 1_800_000_000
421 config = _ready_config(now=now)
422 config["backup"]["last_verification"]["last_ok_time"] = None
423 _write_config(journal, config)
424 before_config = _config_path(journal).read_bytes()
425 request = Mock(return_value=True)
426 monkeypatch.setattr(offload, "request_verification_now", request)
427
428 result = offload.run_offload(dry_run=True)
429
430 assert result.status == "stalled"
431 assert result.reason == "verification_missing"
432 request.assert_not_called()
433 assert _config_path(journal).read_bytes() == before_config
434
435
436def test_sparse_dry_run_measurement_uses_real_fixture_sizes(
437 tmp_path: Path,
438 monkeypatch: pytest.MonkeyPatch,
439) -> None:
440 journal = _use_journal(tmp_path, monkeypatch)
441 now = 1_800_000_000
442 _write_config(journal, _ready_config(now=now, budget_bytes=100 * GB))
443 _make_segment(journal, size=137 * GB)
444 archive = Mock()
445 monkeypatch.setattr(offload, "run_archive_backup", archive)
446
447 result = offload.run_offload(dry_run=True)
448
449 assert result.status == "ok"
450 assert result.details[0].bytes == 137 * GB
451 archive.assert_not_called()
452
453
454def test_success_appends_ledger_before_unlink_and_reuses_digest_for_audit(
455 tmp_path: Path,
456 monkeypatch: pytest.MonkeyPatch,
457) -> None:
458 journal = _use_journal(tmp_path, monkeypatch)
459 now = 1_800_000_000
460 _write_config(journal, _ready_config(now=now, budget_bytes=1))
461 content = b"actual media bytes"
462 seg_path = _make_segment(journal, content=content)
463 raw_path = seg_path / "audio.wav"
464 expected_digest = hashlib.sha256(content).hexdigest()
465 events: list[tuple[str, str]] = []
466 archive_calls, confirm_calls = _install_successful_archive(monkeypatch)
467 real_append_jsonl = offload_ledger.append_jsonl
468 real_unlink = Path.unlink
469
470 def recording_append_jsonl(path: Path, record: Any) -> None:
471 events.append(("ledger_append", record["segment"]))
472 real_append_jsonl(path, record)
473
474 def recording_unlink(path: Path, *args: Any, **kwargs: Any) -> None:
475 if path == raw_path:
476 events.append(("unlink", path.name))
477 real_unlink(path, *args, **kwargs)
478
479 monkeypatch.setattr(offload_ledger, "append_jsonl", recording_append_jsonl)
480 monkeypatch.setattr(Path, "unlink", recording_unlink)
481
482 result = offload.run_offload()
483
484 assert result.status == "ok"
485 assert result.files_offloaded == 1
486 assert result.bytes_offloaded == len(content)
487 assert archive_calls == [(raw_path,)]
488 assert confirm_calls == [{raw_path: len(content)}]
489 assert events.index(("ledger_append", "120000_300")) < events.index(
490 ("unlink", "audio.wav")
491 )
492 assert not raw_path.exists()
493 assert (seg_path / "audio.jsonl").exists()
494 assert (seg_path / "notes.jsonl").exists()
495 assert (seg_path / "talents" / "keep.json").exists()
496
497 summary = summarize_segment("20260101", "default", "120000_300")
498 assert summary.currently_offloaded is True
499 assert summary.files[0].sha256 == expected_digest
500 audit_records = _pruning_records(journal)
501 assert audit_records[0]["kind"] == "raw_media_offload"
502 assert audit_records[0]["files"] == [
503 {"name": "audio.wav", "bytes": len(content), "hash": expected_digest}
504 ]
505 assert summary.files[0].sha256 == audit_records[0]["files"][0]["hash"]
506
507
508@pytest.mark.parametrize(
509 ("archive_result", "expected_reason"),
510 [
511 (
512 engine.BackupResult(status="skipped", snapshot_id=None, error_reason=None),
513 "backup_not_ready",
514 ),
515 (
516 engine.BackupResult(
517 status="error", snapshot_id=None, error_reason="locked"
518 ),
519 "locked",
520 ),
521 (
522 engine.BackupResult(
523 status="error", snapshot_id=None, error_reason="failed"
524 ),
525 "archive_failed",
526 ),
527 ],
528)
529def test_archive_failure_mapping_halts_before_confirm_or_delete(
530 tmp_path: Path,
531 monkeypatch: pytest.MonkeyPatch,
532 archive_result: engine.BackupResult,
533 expected_reason: str,
534) -> None:
535 journal = _use_journal(tmp_path, monkeypatch)
536 now = 1_800_000_000
537 _write_config(journal, _ready_config(now=now, budget_bytes=1))
538 raw_path = _make_segment(journal, content=b"abcdef").joinpath("audio.wav")
539 monkeypatch.setattr(
540 offload, "run_archive_backup", Mock(return_value=archive_result)
541 )
542 confirm = Mock()
543 monkeypatch.setattr(offload, "check_archive_snapshot_files", confirm)
544
545 result = offload.run_offload()
546
547 assert result.status == "stalled"
548 assert result.reason == expected_reason
549 confirm.assert_not_called()
550 _assert_no_offload_side_effects(journal, raw_path)
551
552
553@pytest.mark.parametrize(
554 ("confirm_case", "expected_reason"),
555 [
556 ("skipped", "backup_not_ready"),
557 ("locked", "locked"),
558 ("other_error", "confirm_tool_failed"),
559 ("unconfirmed", "confirm_failed"),
560 ],
561)
562def test_confirm_failure_mapping_halts_before_ledger_or_delete(
563 tmp_path: Path,
564 monkeypatch: pytest.MonkeyPatch,
565 confirm_case: str,
566 expected_reason: str,
567) -> None:
568 journal = _use_journal(tmp_path, monkeypatch)
569 now = 1_800_000_000
570 _write_config(journal, _ready_config(now=now, budget_bytes=1))
571 raw_path = _make_segment(journal, content=b"abcdef").joinpath("audio.wav")
572 monkeypatch.setattr(
573 offload,
574 "run_archive_backup",
575 Mock(
576 return_value=engine.BackupResult(
577 status="ok",
578 snapshot_id="archive-1",
579 error_reason=None,
580 )
581 ),
582 )
583
584 def fake_confirm(
585 snapshot_id: str,
586 expected_sizes: dict[Path, int],
587 ) -> engine.ArchiveCheckResult:
588 if confirm_case == "skipped":
589 return engine.ArchiveCheckResult(
590 status="skipped",
591 error_reason=None,
592 verdicts=None,
593 )
594 if confirm_case == "locked":
595 return engine.ArchiveCheckResult(
596 status="error",
597 error_reason="locked",
598 verdicts=None,
599 )
600 if confirm_case == "other_error":
601 return engine.ArchiveCheckResult(
602 status="error",
603 error_reason="failed",
604 verdicts=None,
605 )
606 path, size = next(iter(expected_sizes.items()))
607 return engine.ArchiveCheckResult(
608 status="ok",
609 error_reason=None,
610 verdicts=(
611 engine.ArchiveFileVerdict(
612 path=str(path),
613 confirmed=False,
614 expected_size=size,
615 observed_size=size + 1,
616 snapshot_id=snapshot_id,
617 ),
618 ),
619 )
620
621 monkeypatch.setattr(offload, "check_archive_snapshot_files", fake_confirm)
622
623 result = offload.run_offload()
624
625 assert result.status == "stalled"
626 assert result.reason == expected_reason
627 _assert_no_offload_side_effects(journal, raw_path)
628
629
630def test_partial_progress_stall_records_honest_counters_and_halts(
631 tmp_path: Path,
632 monkeypatch: pytest.MonkeyPatch,
633) -> None:
634 journal = _use_journal(tmp_path, monkeypatch)
635 now = 1_800_000_000
636 prior_last_ok_time = 1_700_000_000
637 _write_config(
638 journal,
639 _ready_config(
640 now=now,
641 budget_bytes=1,
642 last_offload={
643 "time": prior_last_ok_time,
644 "status": "ok",
645 "reason": None,
646 "last_ok_time": prior_last_ok_time,
647 "files_offloaded": 2,
648 "bytes_offloaded": 10,
649 "ran_out_of_media": False,
650 },
651 ),
652 )
653 first = _make_segment(journal, segment="090000_300", content=b"first").joinpath(
654 "audio.wav"
655 )
656 second = _make_segment(journal, segment="091000_300", content=b"second").joinpath(
657 "audio.wav"
658 )
659 third = _make_segment(journal, segment="092000_300", content=b"third").joinpath(
660 "audio.wav"
661 )
662 archive_calls: list[tuple[Path, ...]] = []
663
664 def fake_archive(paths: list[Path]) -> engine.BackupResult:
665 archive_calls.append(tuple(paths))
666 if len(archive_calls) == 1:
667 return engine.BackupResult(
668 status="ok",
669 snapshot_id="archive-1",
670 error_reason=None,
671 )
672 if len(archive_calls) == 2:
673 return engine.BackupResult(
674 status="error",
675 snapshot_id=None,
676 error_reason="failed",
677 )
678 raise AssertionError("offload must halt after first per-segment failure")
679
680 monkeypatch.setattr(offload, "run_archive_backup", fake_archive)
681 monkeypatch.setattr(offload, "check_archive_snapshot_files", _ok_confirm)
682
683 result = offload.run_offload()
684
685 assert result.status == "stalled"
686 assert result.reason == "archive_failed"
687 assert result.files_offloaded == 1
688 assert result.bytes_offloaded == len(b"first")
689 assert archive_calls == [(first,), (second,)]
690 assert not first.exists()
691 assert second.exists()
692 assert third.exists()
693 last_offload = _read_config(journal)["backup"]["last_offload"]
694 assert last_offload["files_offloaded"] == 1
695 assert last_offload["bytes_offloaded"] == len(b"first")
696 assert last_offload["last_ok_time"] == prior_last_ok_time
697
698
699def test_loop_stops_after_bounds_met_using_start_measurement_minus_freed_bytes(
700 tmp_path: Path,
701 monkeypatch: pytest.MonkeyPatch,
702) -> None:
703 journal = _use_journal(tmp_path, monkeypatch)
704 now = 1_800_000_000
705 _write_config(journal, _ready_config(now=now, budget_bytes=7))
706 first = _make_segment(journal, segment="090000_300", content=b"first").joinpath(
707 "audio.wav"
708 )
709 second = _make_segment(journal, segment="091000_300", content=b"second!").joinpath(
710 "audio.wav"
711 )
712 archive_calls, _confirm_calls = _install_successful_archive(monkeypatch)
713
714 result = offload.run_offload()
715
716 assert result.status == "ok"
717 assert result.ran_out_of_media is False
718 assert archive_calls == [(first,)]
719 assert not first.exists()
720 assert second.exists()
721
722
723def test_ledger_event_with_media_present_is_selected_and_summary_supersedes_old_event(
724 tmp_path: Path,
725 monkeypatch: pytest.MonkeyPatch,
726) -> None:
727 journal = _use_journal(tmp_path, monkeypatch)
728 now = 1_800_000_000
729 _write_config(journal, _ready_config(now=now, budget_bytes=1))
730 raw_path = _make_segment(journal, content=b"new-content").joinpath("audio.wav")
731 append_offload_event(
732 day="20260101",
733 stream="default",
734 segment="120000_300",
735 snapshot_id="old-snapshot",
736 files=[
737 OffloadFile(
738 name="audio.wav",
739 bytes=999,
740 sha256="a" * 64,
741 )
742 ],
743 )
744 archive_calls, _confirm_calls = _install_successful_archive(monkeypatch)
745
746 result = offload.run_offload()
747
748 assert result.status == "ok"
749 assert archive_calls == [(raw_path,)]
750 summary = summarize_segment("20260101", "default", "120000_300")
751 assert summary.offloaded_file_count == 1
752 assert summary.offloaded_bytes == len(b"new-content")
753
754
755def test_ledger_event_with_media_absent_is_skipped_without_reading_summary(
756 tmp_path: Path,
757 monkeypatch: pytest.MonkeyPatch,
758) -> None:
759 journal = _use_journal(tmp_path, monkeypatch)
760 now = 1_800_000_000
761 _write_config(journal, _ready_config(now=now, budget_bytes=1))
762 append_offload_event(
763 day="20260101",
764 stream="default",
765 segment="120000_300",
766 snapshot_id="old-snapshot",
767 files=[
768 OffloadFile(
769 name="audio.wav",
770 bytes=999,
771 sha256="a" * 64,
772 )
773 ],
774 )
775 archive = Mock()
776 monkeypatch.setattr(offload, "run_archive_backup", archive)
777 summary_symbols = {"summarize_segment", "summarize_day", "summarize_journal"}
778 assert summary_symbols.isdisjoint(vars(offload))
779 for name in summary_symbols:
780 monkeypatch.setattr(
781 offload_ledger,
782 name,
783 Mock(side_effect=AssertionError("pass must not read ledger summaries")),
784 raising=True,
785 )
786
787 result = offload.run_offload()
788
789 assert result.status == "ok"
790 assert result.details == ()
791 archive.assert_not_called()
792
793
794def test_gate_blocks_incomplete_and_failed_segments_without_archiving(
795 tmp_path: Path,
796 monkeypatch: pytest.MonkeyPatch,
797) -> None:
798 journal = _use_journal(tmp_path, monkeypatch)
799 now = 1_800_000_000
800 _write_config(journal, _ready_config(now=now, budget_bytes=1))
801 incomplete = _make_segment(
802 journal,
803 segment="090000_300",
804 content=b"incomplete",
805 complete=False,
806 ).joinpath("audio.wav")
807 failed = _make_segment(
808 journal,
809 segment="091000_300",
810 content=b"failed",
811 failed=True,
812 ).joinpath("audio.wav")
813 archive = Mock()
814 monkeypatch.setattr(offload, "run_archive_backup", archive)
815
816 result = offload.run_offload()
817
818 assert result.status == "ok"
819 assert result.ran_out_of_media is True
820 archive.assert_not_called()
821 assert incomplete.exists()
822 assert failed.exists()
823 assert _read_config(journal)["backup"]["last_offload"]["ran_out_of_media"] is True
824
825
826def test_unexpected_exception_records_unexpected_error(
827 tmp_path: Path,
828 monkeypatch: pytest.MonkeyPatch,
829) -> None:
830 journal = _use_journal(tmp_path, monkeypatch)
831 now = 1_800_000_000
832 _write_config(journal, _ready_config(now=now, budget_bytes=1))
833 raw_path = _make_segment(journal, content=b"abcdef").joinpath("audio.wav")
834
835 def fail_archive(paths: list[Path]) -> engine.BackupResult:
836 raise RuntimeError("boom")
837
838 monkeypatch.setattr(offload, "run_archive_backup", fail_archive)
839
840 result = offload.run_offload()
841
842 assert result.status == "stalled"
843 assert result.reason == "unexpected_error"
844 assert _read_config(journal)["backup"]["last_offload"]["reason"] == (
845 "unexpected_error"
846 )
847 assert raw_path.exists()
848
849
850def test_run_prune_keeps_archive_tag_guard_on_forget(
851 tmp_path: Path,
852 monkeypatch: pytest.MonkeyPatch,
853) -> None:
854 journal = _use_journal(tmp_path, monkeypatch)
855 _write_config(journal, _ready_config(now=1_800_000_000))
856 calls: list[list[str]] = []
857
858 def fake_run_restic(args: list[str], **kwargs: Any) -> ResticResult:
859 calls.append(args)
860 return ResticResult(
861 returncode=0,
862 stdout="",
863 stderr="",
864 json=[],
865 argv=("restic", *args),
866 )
867
868 monkeypatch.setattr(engine, "ensure_restic", Mock(return_value=Path("/restic")))
869 monkeypatch.setattr(engine, "run_restic", fake_run_restic)
870
871 result = engine.run_prune()
872
873 assert result.status == "ok"
874 forget_args = next(args for args in calls if args and args[0] == "forget")
875 assert forget_args[forget_args.index("--keep-tag") + 1] == engine.ARCHIVE_TAG