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
8import subprocess
9import sys
10import zipfile
11from datetime import UTC, datetime
12from pathlib import Path
13from typing import Any
14
15import pytest
16
17import scripts.release_build_host as build_rail
18import scripts.release_install_smoke as smoke
19import scripts.release_proof_host as proof_rail
20from scripts.channel_adapters import adapter_common as common
21from scripts.channel_adapters import build_host_macos, proof_host
22from scripts.check_release_preflight import expected_presign_lane_tool_evidence
23from scripts.release_digest import candidate_digest
24from scripts.release_tool_pins import (
25 HOST_VARIANT_TOOL_KEYS,
26 MACOS_SWIFT_FLATTENED_BANNER,
27 UV_MACOS_FIXTURE_BANNER,
28)
29from tests.helpers.release_wheel_fixtures import ROOT_LAUNCHER_BYTES, record_hash
30
31
32def _completed(
33 stdout: str = "",
34 *,
35 stderr: str = "",
36 returncode: int = 0,
37) -> subprocess.CompletedProcess[str]:
38 return subprocess.CompletedProcess(
39 args=[],
40 returncode=returncode,
41 stdout=stdout,
42 stderr=stderr,
43 )
44
45
46def _lane(
47 name: str = "lane",
48 *,
49 host: str = "build-host.example",
50 port: int | None = 2222,
51) -> common.LaneConfig:
52 return common.LaneConfig(
53 name=name,
54 mode="ssh",
55 host=host,
56 port=port,
57 user="builder",
58 identity_file="~/.ssh/solstone-channel-adapter",
59 extra_ssh_options=("-o", "BatchMode=yes"),
60 remote_python="python3",
61 remote_work_prefix="/tmp/solstone-channel-adapter",
62 remote_run_wrapper="operator-session-wrapper",
63 tmux_window="adapter:build",
64 unlock_workdir="~/projects/build-worktree",
65 )
66
67
68def _local_lane() -> common.LaneConfig:
69 return common.LaneConfig(
70 name="proof.linux-x86_64-musl",
71 mode="local",
72 )
73
74
75def _write_metadata_wheel(path: Path) -> None:
76 distribution, version = path.name.removesuffix(".whl").split("-")[:2]
77 metadata_name = f"{distribution}-{version}.dist-info/METADATA"
78 metadata = f"Name: {distribution.replace('_', '-')}\nVersion: {version}\n"
79 with zipfile.ZipFile(path, "w") as wheel:
80 members = {metadata_name: metadata.encode("utf-8")}
81 if path.name.startswith("solstone-"):
82 members[f"solstone-{version}.dist-info/WHEEL"] = b"Wheel-Version: 1.0\n"
83 for name, content in ROOT_LAUNCHER_BYTES.items():
84 members[f"solstone-{version}.data/scripts/{name}"] = content
85 record_name = f"solstone-{version}.dist-info/RECORD"
86 record = "\n".join(
87 f"{name},{record_hash(content)},{len(content)}"
88 for name, content in members.items()
89 )
90 members[record_name] = f"{record}\n{record_name},,".encode("utf-8")
91 for name, content in members.items():
92 info = zipfile.ZipInfo(name)
93 info.external_attr = (
94 0o755 << 16
95 if Path(name).name in smoke.ROOT_LAUNCHER_NAMES
96 else 0o644 << 16
97 )
98 wheel.writestr(info, content)
99
100
101def _file_entry(path: Path) -> dict[str, Any]:
102 data = path.read_bytes()
103 return {
104 "name": path.name,
105 "bytes": len(data),
106 "sha256": hashlib.sha256(data).hexdigest(),
107 }
108
109
110def _tool_stdout(*, uv: str = UV_MACOS_FIXTURE_BANNER) -> str:
111 expected = expected_presign_lane_tool_evidence("macos-arm64")
112 observed = {
113 **expected,
114 "uv": uv,
115 "swift": MACOS_SWIFT_FLATTENED_BANNER,
116 }
117 return (
118 "\n".join(f"{key}\t{observed[key]}" for key in sorted(observed))
119 + f"\n{build_host_macos.TOOLCHAIN_TOKEN}\n"
120 )
121
122
123def _artifact_listing_stdout(artifact_bytes: dict[str, bytes]) -> str:
124 lines = []
125 for name, data in artifact_bytes.items():
126 lines.append(
127 "\t".join(
128 (
129 build_host_macos.ARTIFACT_TOKEN,
130 name,
131 hashlib.sha256(data).hexdigest(),
132 str(len(data)),
133 )
134 )
135 )
136 lines.append(build_host_macos.DIST_TOKEN)
137 return "\n".join(lines) + "\n"
138
139
140def _write_build_request(tmp_path: Path) -> tuple[Path, build_rail.SourceBundle, dict]:
141 bundle = tmp_path / "source.bundle"
142 bundle.write_bytes(b"bundle")
143 sha256, byte_count = common.sha256_size(bundle)
144 source_bundle = build_rail.SourceBundle(
145 path=bundle,
146 source_commit="a" * 40,
147 sha256=sha256,
148 bytes=byte_count,
149 )
150 channel = build_rail.ExternalBuildHostChannel(["adapter"])
151 payload = channel._request_payload(
152 cohort_id="cohort",
153 source_bundle=source_bundle,
154 expected_commit=source_bundle.source_commit,
155 )
156 request_dir = tmp_path / "request"
157 request_dir.mkdir()
158 request_bundle = request_dir / "source.bundle"
159 request_bundle.write_bytes(bundle.read_bytes())
160 request_path = request_dir / "request.json"
161 request_path.write_text(json.dumps(payload), encoding="utf-8")
162 return request_path, source_bundle, payload
163
164
165def _write_proof_request(tmp_path: Path) -> tuple[Path, dict[str, Any], dict[str, Any]]:
166 request_dir = tmp_path / "proof-request"
167 candidate_dir = request_dir / "candidate"
168 output_dir = request_dir / "output"
169 candidate_dir.mkdir(parents=True)
170 output_dir.mkdir()
171 for name in (
172 "solstone-1.0.0-py3-none-any.whl",
173 "solstone_core-1.0.0-py3-none-linux_x86_64.whl",
174 ):
175 _write_metadata_wheel(candidate_dir / name)
176 native_members = {
177 name: {
178 "path": f"linux-x86/{name}",
179 "sha256": hashlib.sha256(name.encode("utf-8")).hexdigest(),
180 "bytes": len(name.encode("utf-8")),
181 }
182 for name in smoke.CORE_SCRIPT_NAMES
183 }
184 digest = candidate_digest(candidate_dir)
185 ledger: dict[str, Any] = {
186 "source_commit": "b" * 40,
187 "core_lock_sha256": "c" * 64,
188 "candidate": {
189 "candidate_digest": digest,
190 "files": [
191 _file_entry(path)
192 for path in sorted(candidate_dir.iterdir(), key=lambda item: item.name)
193 ],
194 },
195 "native_members": {"linux-x86_64-musl": native_members},
196 }
197 ledger_bytes = json.dumps(ledger, sort_keys=True).encode("utf-8")
198 ledger_sha = hashlib.sha256(ledger_bytes).hexdigest()
199 (request_dir / "ledger.json").write_bytes(ledger_bytes)
200 install_paths = smoke.target_install_paths_from_ledger(
201 ledger,
202 target="linux-x86_64-musl",
203 candidate_dir=candidate_dir,
204 )
205 channel = proof_rail.ExternalProofHostChannel(
206 "linux-x86_64-musl",
207 ["adapter"],
208 )
209 payload = channel._request_payload(
210 cohort_id="cohort",
211 target="linux-x86_64-musl",
212 version="1.0.0",
213 source_commit="b" * 40,
214 core_lock_sha256="c" * 64,
215 candidate_digest=digest,
216 ledger_sha256=ledger_sha,
217 install_paths=install_paths,
218 )
219 request_path = request_dir / "request.json"
220 request_path.write_text(json.dumps(payload), encoding="utf-8")
221 return request_path, payload, ledger
222
223
224def _write_valid_install_proof(
225 proof_path: Path,
226 *,
227 request_payload: dict[str, Any],
228 ledger: dict[str, Any],
229) -> None:
230 request_dir = proof_path.parents[1]
231 candidate_dir = request_dir / request_payload["paths"]["candidate_dir"]
232 env_root = request_dir / "env"
233 (env_root / "bin").mkdir(parents=True, exist_ok=True)
234 python_path = env_root / "bin" / "python"
235 python_path.write_bytes(b"python")
236 executable_paths = {
237 name: env_root / "bin" / name for name in smoke.INSTALL_SCRIPT_NAMES
238 }
239 for name in smoke.ROOT_LAUNCHER_NAMES:
240 executable_paths[name].write_bytes(ROOT_LAUNCHER_BYTES[name])
241 for name in smoke.CORE_SCRIPT_NAMES:
242 executable_paths[name].write_text(name, encoding="utf-8")
243 install_paths = smoke.target_install_paths_from_ledger(
244 ledger,
245 target=request_payload["target"],
246 candidate_dir=candidate_dir,
247 )
248 expected_members, expected_failures = smoke._expected_install_members(
249 ledger,
250 request_payload["target"],
251 candidate_dir=candidate_dir,
252 install_paths=install_paths,
253 )
254 assert expected_failures == []
255 proof = smoke.build_install_proof(
256 target=request_payload["target"],
257 version=request_payload["version"],
258 source_commit=request_payload["source_commit"],
259 core_lock_sha256=request_payload["core_lock_sha256"],
260 candidate_digest=request_payload["candidate_digest"],
261 ledger_sha256=request_payload["ledger_sha256"],
262 candidate_dir=candidate_dir,
263 candidate_paths=install_paths,
264 ledger_payload=ledger,
265 observation=smoke.InstallObservation(
266 env_root=env_root,
267 preexisting_distributions=(),
268 install=smoke.CommandResult(
269 argv=(
270 str(python_path),
271 "-m",
272 "pip",
273 "install",
274 "--no-index",
275 "--no-deps",
276 *(str(path) for path in install_paths),
277 ),
278 exit_code=0,
279 stdout="installed",
280 env=smoke.SCRUBBED_COMMAND_ENV,
281 ),
282 installed_distributions=smoke.expected_distribution_entries(install_paths),
283 installed_members=tuple(
284 {
285 "name": name,
286 "path": path,
287 "sha256": expected_members[name]["sha256"],
288 "symlink": False,
289 }
290 for name, path in sorted(executable_paths.items())
291 ),
292 smoke={
293 name: smoke.CommandResult(
294 argv=(str(path), "--version"),
295 exit_code=0,
296 stdout=(
297 f"{smoke.CORE_SMOKE_STDOUT[name]} {request_payload['version']}"
298 ),
299 env=smoke.SCRUBBED_COMMAND_ENV,
300 )
301 for name, path in sorted(executable_paths.items())
302 },
303 ),
304 recorded_at=datetime(2026, 7, 20, 12, tzinfo=UTC),
305 )
306 smoke.write_install_proof(
307 proof_path,
308 proof,
309 target=request_payload["target"],
310 version=request_payload["version"],
311 source_commit=request_payload["source_commit"],
312 core_lock_sha256=request_payload["core_lock_sha256"],
313 candidate_digest=request_payload["candidate_digest"],
314 ledger_sha256=request_payload["ledger_sha256"],
315 candidate_dir=candidate_dir,
316 ledger_payload=ledger,
317 )
318
319
320def test_build_request_response_round_trip_through_rail_parser(
321 tmp_path: Path,
322 monkeypatch: pytest.MonkeyPatch,
323) -> None:
324 request_path, source_bundle, payload = _write_build_request(tmp_path)
325 expected_files = list(payload["expected_outputs"].values())
326 artifact_bytes = {
327 name: f"artifact:{name}".encode("utf-8") for name in expected_files
328 }
329
330 def fake_runner(argv, **kwargs):
331 if argv[0] == "scp" and ":" in argv[-2]:
332 name = str(argv[-2]).rsplit("/", 1)[-1]
333 Path(argv[-1]).write_bytes(artifact_bytes[name])
334 script = kwargs.get("input_text") or ""
335 if "emit python" in script:
336 return _completed(_tool_stdout())
337 if "git checkout" in script:
338 return _completed(f"{build_host_macos.CHECKOUT_TOKEN}\n")
339 if "for f in" in script:
340 return _completed(_artifact_listing_stdout(artifact_bytes))
341 return _completed()
342
343 monkeypatch.setattr(common, "run", fake_runner)
344 monkeypatch.chdir(request_path.parent)
345
346 build_host_macos.build_macos(_lane(), request_path)
347
348 response = json.loads((request_path.parent / "response.json").read_text())
349 build_rail._validate_attestation(
350 response,
351 expected_commit=source_bundle.source_commit,
352 source_bundle=source_bundle,
353 )
354 evidence = build_rail._validate_macos_tool_evidence(response)
355 wheel_names, record_names = build_rail._names_from_payload(response)
356 assert set(evidence) == set(expected_presign_lane_tool_evidence("macos-arm64"))
357 assert len(wheel_names) == 2
358 assert tuple(record_names) == (
359 build_rail.MACOS_ROOT_RECORD,
360 build_rail.MACOS_CORE_RECORD,
361 )
362
363
364def test_build_retrieved_artifact_digest_mismatch_writes_no_response(
365 tmp_path: Path,
366 monkeypatch: pytest.MonkeyPatch,
367) -> None:
368 request_path, _source_bundle, payload = _write_build_request(tmp_path)
369 expected_files = list(payload["expected_outputs"].values())
370 artifact_bytes = {
371 name: f"artifact:{name}".encode("utf-8") for name in expected_files
372 }
373 truncated_name = expected_files[0]
374
375 def fake_runner(argv, **kwargs):
376 if argv[0] == "scp" and ":" in argv[-2]:
377 name = str(argv[-2]).rsplit("/", 1)[-1]
378 data = artifact_bytes[name]
379 Path(argv[-1]).write_bytes(data[:-1] if name == truncated_name else data)
380 script = kwargs.get("input_text") or ""
381 if "emit python" in script:
382 return _completed(_tool_stdout())
383 if "git checkout" in script:
384 return _completed(f"{build_host_macos.CHECKOUT_TOKEN}\n")
385 if "for f in" in script:
386 return _completed(_artifact_listing_stdout(artifact_bytes))
387 return _completed()
388
389 monkeypatch.setattr(common, "run", fake_runner)
390 monkeypatch.chdir(request_path.parent)
391
392 with pytest.raises(SystemExit):
393 build_host_macos.build_macos(_lane(), request_path)
394
395 assert not (request_path.parent / "response.json").exists()
396
397
398def test_proof_request_response_round_trip_through_rail_parser(
399 tmp_path: Path,
400 monkeypatch: pytest.MonkeyPatch,
401) -> None:
402 request_path, request_payload, ledger = _write_proof_request(tmp_path)
403 proof_path = request_path.parent / "output" / "proof.json"
404
405 def fake_runner(argv, **kwargs):
406 if argv == ["uname", "-s"]:
407 return _completed("Linux\n")
408 if argv == ["uname", "-m"]:
409 return _completed("x86_64\n")
410 if argv[:2] == [sys.executable, "-c"]:
411 _write_valid_install_proof(
412 proof_path,
413 request_payload=request_payload,
414 ledger=ledger,
415 )
416 sha256, byte_count = common.sha256_size(proof_path)
417 return _completed(
418 f'{proof_host.PROOF_TOKEN} {{"bytes": {byte_count}, "sha256": "{sha256}"}}\n'
419 )
420 raise AssertionError(argv)
421
422 monkeypatch.setattr(proof_host, "run", fake_runner)
423
424 proof_host.prove("linux-x86_64-musl", _local_lane(), request_path)
425
426 response = json.loads((request_path.parent / "response.json").read_text())
427 channel = proof_rail.ExternalProofHostChannel("linux-x86_64-musl", ["adapter"])
428 proof_descriptor = channel._validate_response(
429 response,
430 cohort_id="cohort",
431 target="linux-x86_64-musl",
432 candidate_digest=request_payload["candidate_digest"],
433 ledger_sha256=request_payload["ledger_sha256"],
434 )
435 assert proof_descriptor["path"] == "output/proof.json"
436 proof_failures = smoke.validate_install_proof_bytes(
437 proof_path.read_bytes(),
438 target=request_payload["target"],
439 version=request_payload["version"],
440 source_commit=request_payload["source_commit"],
441 core_lock_sha256=request_payload["core_lock_sha256"],
442 candidate_digest=request_payload["candidate_digest"],
443 ledger_sha256=request_payload["ledger_sha256"],
444 candidate_dir=request_path.parent / request_payload["paths"]["candidate_dir"],
445 ledger_payload=ledger,
446 )
447 assert proof_failures == []
448
449
450def test_source_and_retrieved_digest_verification(tmp_path: Path) -> None:
451 proof = tmp_path / "proof.json"
452 proof.write_bytes(b"proof")
453 sha256, byte_count = common.sha256_size(proof)
454
455 common.verify_retrieved_file(
456 proof,
457 expected_sha256=sha256,
458 expected_bytes=byte_count,
459 label="proof.json",
460 )
461 with pytest.raises(SystemExit):
462 common.verify_retrieved_file(
463 proof,
464 expected_sha256="0" * 64,
465 expected_bytes=byte_count,
466 label="proof.json",
467 )
468
469
470def test_macos_tool_evidence_derives_from_rail_pins(
471 monkeypatch: pytest.MonkeyPatch,
472) -> None:
473 monkeypatch.setattr(
474 common, "run", lambda _argv, **_kwargs: _completed(_tool_stdout())
475 )
476
477 evidence = build_host_macos._derive_tool_evidence(_lane())
478 expected = expected_presign_lane_tool_evidence("macos-arm64")
479
480 assert set(evidence) == set(expected)
481 assert evidence["uv"] == UV_MACOS_FIXTURE_BANNER
482 assert evidence["swift"] == MACOS_SWIFT_FLATTENED_BANNER
483 for key in set(expected) - set(HOST_VARIANT_TOOL_KEYS):
484 assert evidence[key] == expected[key]
485
486
487def test_host_variant_banner_mutation_writes_no_evidence_and_exits_nonzero(
488 tmp_path: Path,
489 monkeypatch: pytest.MonkeyPatch,
490) -> None:
491 request_path, _source_bundle, _payload = _write_build_request(tmp_path)
492 mutated = "uv 0.11.5 (aarch64-apple-darwin)"
493
494 monkeypatch.setattr(
495 common,
496 "run",
497 lambda _argv, **kwargs: (
498 _completed(_tool_stdout(uv=mutated))
499 if "emit python" in (kwargs.get("input_text") or "")
500 else _completed(f"{build_host_macos.CHECKOUT_TOKEN}\n")
501 ),
502 )
503 monkeypatch.chdir(request_path.parent)
504
505 with pytest.raises(SystemExit):
506 build_host_macos.build_macos(_lane(), request_path)
507
508 assert not (request_path.parent / "response.json").exists()
509
510
511def test_public_evidence_failure_writes_no_response(
512 tmp_path: Path,
513 monkeypatch: pytest.MonkeyPatch,
514) -> None:
515 request_path, _source_bundle, _payload = _write_build_request(tmp_path)
516 private_shape = "uv 0.11.4 (builder.local)"
517
518 monkeypatch.setattr(
519 common,
520 "run",
521 lambda _argv, **kwargs: (
522 _completed(_tool_stdout(uv=private_shape))
523 if "emit python" in (kwargs.get("input_text") or "")
524 else _completed(f"{build_host_macos.CHECKOUT_TOKEN}\n")
525 ),
526 )
527 monkeypatch.chdir(request_path.parent)
528
529 with pytest.raises(SystemExit):
530 build_host_macos.build_macos(_lane(), request_path)
531
532 assert not (request_path.parent / "response.json").exists()
533
534
535def test_sentinel_requires_exit_zero_and_token() -> None:
536 common.require_success_token(_completed("TOKEN\n"), "TOKEN", "label")
537 with pytest.raises(SystemExit):
538 common.require_success_token(_completed("", returncode=0), "TOKEN", "label")
539 with pytest.raises(SystemExit):
540 common.require_success_token(
541 _completed("TOKEN\n", returncode=1), "TOKEN", "label"
542 )
543
544
545def test_ssh_argv_from_structured_config() -> None:
546 flag = "-" + "p"
547
548 argv = common.build_ssh_argv(_lane(), ["bash", "-s"])
549
550 assert argv == [
551 "ssh",
552 "-o",
553 "BatchMode=yes",
554 flag,
555 "2222",
556 "-i",
557 "~/.ssh/solstone-channel-adapter",
558 "builder@build-host.example",
559 "bash",
560 "-s",
561 ]
562
563
564def test_scp_argv_from_structured_config() -> None:
565 flag = "-" + "P"
566
567 argv = common.build_scp_argv(_lane(), "local", "remote", direction="to")
568
569 assert argv == [
570 "scp",
571 "-q",
572 "-o",
573 "BatchMode=yes",
574 flag,
575 "2222",
576 "-i",
577 "~/.ssh/solstone-channel-adapter",
578 "local",
579 "builder@build-host.example:remote",
580 ]
581
582
583def test_cleanup_argument_handling(monkeypatch: pytest.MonkeyPatch) -> None:
584 calls: list[tuple[list[str], str]] = []
585
586 def fake_runner(argv, **kwargs):
587 calls.append((list(argv), kwargs.get("input_text") or ""))
588 return _completed()
589
590 monkeypatch.setattr(common, "run", fake_runner)
591
592 build_host_macos.cleanup(_lane(), "cohort", "f" * 64)
593 proof_host.cleanup(
594 "macos-arm64", _lane(name="proof.macos-arm64"), "cohort", "e" * 64
595 )
596
597 assert len(calls) == 2
598 assert all("cohort" in script for _argv, script in calls)
599
600
601def test_config_validation_fails_before_side_effects(
602 tmp_path: Path,
603 monkeypatch: pytest.MonkeyPatch,
604) -> None:
605 config = tmp_path / "channel-adapters.json"
606 config.write_text(
607 '{"schema_version": 1, "build": {}, "proof": {}}', encoding="utf-8"
608 )
609 calls: list[list[str]] = []
610
611 def fake_runner(argv, **_kwargs):
612 calls.append(list(argv))
613 return _completed()
614
615 monkeypatch.setenv(common.CONFIG_ENV, str(config))
616 monkeypatch.setattr(common, "run", fake_runner)
617
618 with pytest.raises(SystemExit):
619 build_host_macos.main(["build-macos", "request.json"])
620
621 assert calls == []
622
623
624def test_target_env_keys_coupling() -> None:
625 config_targets = {
626 "linux-x86_64-musl",
627 "linux-aarch64-musl",
628 "macos-arm64",
629 }
630
631 assert config_targets == set(proof_rail.TARGET_ENV_KEYS)