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
6from pathlib import Path
7from types import SimpleNamespace
8
9import pytest
10
11from solstone.think import offload_measurement
12from solstone.think.offload_measurement import (
13 device_free_bytes,
14 device_total_bytes,
15 measure_raw_media_usage,
16 suggest_offload_defaults,
17)
18from solstone.think.retention import compute_storage_summary
19
20GB = 10**9
21
22
23def _segment(journal: Path, day: str, stream: str = "archon") -> Path:
24 path = journal / "chronicle" / day / stream / "120000_300"
25 path.mkdir(parents=True, exist_ok=True)
26 return path
27
28
29def _write_bytes(path: Path, size: int) -> Path:
30 path.parent.mkdir(parents=True, exist_ok=True)
31 path.write_bytes(b"x" * size)
32 return path
33
34
35def test_raw_media_measurement_matches_retention_predicate_non_recursive(
36 tmp_path: Path, monkeypatch: pytest.MonkeyPatch
37) -> None:
38 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
39 segment = _segment(tmp_path, "20260101")
40 _write_bytes(segment / "audio.wav", 10)
41 _write_bytes(segment / "clip.mp4", 20)
42 _write_bytes(segment / "frame.png", 30)
43 _write_bytes(segment / "monitor_0_diff.png", 40)
44 _write_bytes(segment / "audio.jsonl", 50)
45 _write_bytes(segment / "note.json", 60)
46 _write_bytes(segment / "talents" / "frame.png", 70)
47
48 usage = measure_raw_media_usage()
49
50 assert usage.total_bytes == 100
51 assert usage.total_files == 4
52 assert usage.per_day == (offload_measurement.RawMediaDayUsage("20260101", 100, 4),)
53 assert compute_storage_summary().raw_media_bytes == usage.total_bytes
54
55
56def test_raw_media_per_day_breakdown_is_chronological(
57 tmp_path: Path, monkeypatch: pytest.MonkeyPatch
58) -> None:
59 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
60 _write_bytes(_segment(tmp_path, "20260301") / "audio.wav", 3)
61 _write_bytes(_segment(tmp_path, "20260101") / "audio.wav", 1)
62 _write_bytes(_segment(tmp_path, "20260201") / "audio.wav", 2)
63
64 usage = measure_raw_media_usage()
65
66 assert [day.day for day in usage.per_day] == [
67 "20260101",
68 "20260201",
69 "20260301",
70 ]
71 assert [day.bytes for day in usage.per_day] == [1, 2, 3]
72
73
74def test_raw_media_measurement_tolerates_file_vanishing_before_stat(
75 tmp_path: Path, monkeypatch: pytest.MonkeyPatch
76) -> None:
77 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
78 segment = _segment(tmp_path, "20260101")
79 vanishing = _write_bytes(segment / "vanishing.wav", 10)
80 stable = _write_bytes(segment / "stable.wav", 20)
81
82 def fake_get_raw_media_files(_segment_path: Path) -> list[Path]:
83 vanishing.unlink()
84 return [vanishing, stable]
85
86 monkeypatch.setattr(
87 offload_measurement, "get_raw_media_files", fake_get_raw_media_files
88 )
89
90 usage = measure_raw_media_usage()
91
92 assert usage.total_bytes == 20
93 assert usage.total_files == 1
94
95
96def test_device_free_bytes_uses_disk_usage_free(
97 tmp_path: Path, monkeypatch: pytest.MonkeyPatch
98) -> None:
99 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
100
101 def fake_disk_usage(path: Path) -> SimpleNamespace:
102 assert path == tmp_path
103 return SimpleNamespace(total=1000 * GB, used=100 * GB, free=850 * GB)
104
105 monkeypatch.setattr(offload_measurement.shutil, "disk_usage", fake_disk_usage)
106
107 assert device_free_bytes() == 850 * GB
108
109
110def test_device_total_bytes_uses_disk_usage_total(
111 tmp_path: Path, monkeypatch: pytest.MonkeyPatch
112) -> None:
113 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
114
115 def fake_disk_usage(path: Path) -> SimpleNamespace:
116 assert path == tmp_path
117 return SimpleNamespace(total=1000 * GB, used=100 * GB, free=850 * GB)
118
119 monkeypatch.setattr(offload_measurement.shutil, "disk_usage", fake_disk_usage)
120
121 assert device_total_bytes() == 1000 * GB
122
123
124@pytest.mark.parametrize(
125 ("total", "budget", "floor"),
126 [
127 (1000 * GB, 500 * GB, 100 * GB),
128 (100 * GB, 50 * GB, 20 * GB),
129 (30 * GB, 15 * GB, 7_500_000_000),
130 ],
131)
132def test_suggest_offload_defaults_decimal_gb(
133 total: int, budget: int, floor: int
134) -> None:
135 defaults = suggest_offload_defaults(total)
136
137 assert defaults.budget_bytes == budget
138 assert defaults.floor_bytes == floor