solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4"""Shared isolation harness for API baseline tests and baseline regeneration.
5
6Used by:
7- tests/test_api_baselines.py - module-scoped fixtures
8- tests/verify_api.py - verify/update CLI mode
9
10Keeps both paths on identical isolation so generated baselines match the test
11oracle.
12"""
13
14from __future__ import annotations
15
16import atexit
17import json
18import os
19import shutil
20import subprocess
21import tempfile
22import threading
23from contextlib import contextmanager
24from pathlib import Path
25from typing import Iterator
26from unittest.mock import patch
27
28FROZEN_DATE = "2026-04-15"
29FROZEN_TZ_OFFSET = -7
30
31# The fixture journal is built once per process from an immutable snapshot of
32# HEAD rather than copied file-by-file from the live working tree. make dev /
33# make sandbox write runtime artifacts into tests/fixtures/journal (AGENTS.md
34# §6), so a concurrent writer can delete or replace a tracked file between
35# `git ls-files` and the copy — raising FileNotFoundError at fixture *setup*
36# (a spurious pytest ERROR, not a real failure). `git archive HEAD` reads the
37# committed tree from the object store, never the working tree, so it is immune
38# to concurrent working-tree mutation.
39_REPO_ROOT = Path(__file__).resolve().parent.parent
40_FIXTURE_JOURNAL_REL = "tests/fixtures/journal"
41_LIVE_FIXTURE_JOURNAL = (_REPO_ROOT / _FIXTURE_JOURNAL_REL).resolve()
42
43_snapshot_lock = threading.Lock()
44_snapshot_root: Path | None = None
45
46
47def _fixture_journal_snapshot() -> Path:
48 """Return an immutable, process-scoped snapshot of the tracked fixture journal.
49
50 Built once per process via `git archive HEAD` → tar, which reads HEAD from
51 the git object store and is therefore unaffected by concurrent writes to the
52 live tests/fixtures/journal tree. The same set of git-tracked files that
53 `git ls-files` selects (working tree clean ⇒ HEAD ≡ index), with symlinks
54 preserved as symlinks. The temp dir is removed at interpreter exit.
55 """
56 global _snapshot_root
57 with _snapshot_lock:
58 if _snapshot_root is not None and _snapshot_root.exists():
59 return _snapshot_root
60 tmp = Path(tempfile.mkdtemp(prefix="solstone-journal-snapshot-"))
61 archive = subprocess.run(
62 ["git", "archive", "HEAD", "--", _FIXTURE_JOURNAL_REL],
63 cwd=str(_REPO_ROOT),
64 capture_output=True,
65 check=True,
66 )
67 subprocess.run(
68 ["tar", "-x", "-C", str(tmp)],
69 input=archive.stdout,
70 check=True,
71 )
72 # `git archive` preserves the tests/fixtures/journal/ prefix in the tar.
73 _snapshot_root = (tmp / _FIXTURE_JOURNAL_REL).resolve()
74 atexit.register(shutil.rmtree, tmp, ignore_errors=True)
75 return _snapshot_root
76
77
78def copytree_tracked(src: Path, dst: Path) -> None:
79 """Copy only git-tracked files from src to dst.
80
81 For paths within tests/fixtures/journal (the journal fixture and any subtree
82 of it), copy from an immutable, process-scoped snapshot of HEAD instead of
83 enumerating and copying from the live working tree — see
84 `_fixture_journal_snapshot` for why this defeats the concurrent-write race.
85 For any other src, fall back to enumerating tracked files from the live tree.
86 """
87 src = Path(src).resolve()
88 dst = Path(dst)
89 try:
90 rel = src.relative_to(_LIVE_FIXTURE_JOURNAL)
91 except ValueError:
92 rel = None
93 if rel is not None:
94 snap_src = _fixture_journal_snapshot() / rel
95 shutil.copytree(snap_src, dst, symlinks=True, dirs_exist_ok=True)
96 return
97 result = subprocess.run(
98 ["git", "ls-files", "."],
99 cwd=str(src),
100 capture_output=True,
101 text=True,
102 check=True,
103 )
104 for rel_path in result.stdout.splitlines():
105 if not rel_path:
106 continue
107 src_file = src / rel_path
108 dst_file = dst / rel_path
109 dst_file.parent.mkdir(parents=True, exist_ok=True)
110 if src_file.is_symlink():
111 os.symlink(os.readlink(src_file), dst_file)
112 else:
113 shutil.copy2(src_file, dst_file)
114
115
116def prepare_isolated_journal(dst: Path) -> Path:
117 """Copy the git-tracked fixture journal into dst and return the absolute path."""
118 src = Path("tests/fixtures/journal").resolve()
119 dst = dst.resolve()
120 copytree_tracked(src, dst)
121 return dst
122
123
124@contextmanager
125def isolated_app_env(journal: Path) -> Iterator[Path]:
126 """Patch env so create_app(journal) is fully isolated."""
127
128 journal = Path(journal).resolve()
129 overrides = {
130 "SOLSTONE_JOURNAL": str(journal),
131 # Match tests/conftest.py's deterministic unit-test runtime. Without
132 # these, baseline regeneration probes host GPU/supervisor state while
133 # test_api_baselines compares against the isolated test contract.
134 "SOL_SKIP_SUPERVISOR_CHECK": "1",
135 "SOLSTONE_DISABLE_CONVEY_SIDE_RUNTIMES": "1",
136 }
137 previous = {key: os.environ.get(key) for key in overrides}
138 os.environ.update(overrides)
139 try:
140 from solstone.think.providers import local_cuda, local_install, local_vulkan
141
142 deterministic_devices = [
143 local_vulkan.VulkanDevice(
144 1,
145 "NVIDIA GeForce GTX 1660 Ti",
146 local_vulkan.VK_TYPE_DISCRETE,
147 6390,
148 )
149 ]
150 with (
151 patch.object(
152 local_cuda,
153 "probe_nvidia_gpu",
154 lambda: local_cuda.NvidiaProbe(
155 index=None,
156 compute_cap=None,
157 driver_cuda_version=None,
158 vram_mib=None,
159 tiering_memory_mib=None,
160 memory_source=local_cuda.MEMORY_SOURCE_UNAVAILABLE,
161 detected=False,
162 ),
163 ),
164 patch.object(
165 local_install,
166 "probe_cuda_runtime_artifact_trust",
167 lambda _pin, **_kwargs: local_cuda.ArtifactTrust.ABSENT,
168 ),
169 patch.object(
170 local_install,
171 "has_persisted_installed_cuda_target",
172 lambda **_kwargs: False,
173 ),
174 patch.object(
175 local_vulkan,
176 "detect_gpus",
177 lambda: deterministic_devices,
178 ),
179 ):
180 yield journal
181 finally:
182 for key, value in previous.items():
183 if value is None:
184 os.environ.pop(key, None)
185 else:
186 os.environ[key] = value
187
188
189def make_test_client(journal: Path):
190 """Create a Flask test client for an isolated journal."""
191 from solstone.convey import create_app
192
193 app = create_app(journal=str(Path(journal).resolve()))
194 app.config["TESTING"] = True
195 client = app.test_client()
196 return client
197
198
199def mark_setup_complete(journal: Path, completed_at: int = 1700000000000) -> None:
200 """Mark a minimal test journal as past first-run setup."""
201
202 config_path = Path(journal) / "config" / "journal.json"
203 config_path.parent.mkdir(parents=True, exist_ok=True)
204 if config_path.exists():
205 config = json.loads(config_path.read_text(encoding="utf-8"))
206 else:
207 config = {}
208 config["setup"] = {"completed_at": completed_at}
209 config_path.write_text(json.dumps(config, indent=2) + "\n", encoding="utf-8")