okami.mom/albedo

git clone https://git.nkp.pet/okami.mom/albedo

Aly Raffaufrefactor(cli): invoke daemon storage helpersf6da61d

main
branchesmainall refs →
36.2 KiB806 linesraw
1"""Offline cleanup and daemon startup must exclude each other, even without HTTP."""
2
3from contextlib import closing
4import hashlib
5import json
6import os
7import pickle
8import signal
9import select
10import sqlite3
11import socket
12import subprocess
13import sys
14import tempfile
15from pathlib import Path
16import time
17import unittest
18
19from harness import Albedo, ROOT, exclusive, wait_until
20
21from storage_helper_support import maintenance_command
22
23
24class StorageTests(unittest.TestCase):
25    def helper(self, app, paths=()):
26        helper = subprocess.Popen(
27            maintenance_command(app.home, app.env["ALBEDO_DAEMON"]),
28            env=app.env,
29            stdin=subprocess.PIPE,
30            stdout=subprocess.PIPE,
31            stderr=subprocess.PIPE,
32            text=True,
33        )
34        self.addCleanup(self.close_helper, helper)
35        assert helper.stdin is not None and helper.stdout is not None
36        helper.stdin.write(json.dumps({"paths": [str(path) for path in paths]}) + "\n")
37        helper.stdin.flush()
38        self.assertEqual(json.loads(helper.stdout.readline()), {"ready": True})
39        return helper
40
41    def close_helper(self, helper):
42        if helper.stdin and not helper.stdin.closed:
43            helper.stdin.close()
44        try:
45            helper.wait(timeout=10)
46        finally:
47            if helper.poll() is None:
48                helper.kill()
49                helper.wait()
50            if helper.stdout:
51                helper.stdout.close()
52            if helper.stderr:
53                helper.stderr.close()
54
55    def backup(self, app):
56        path = app.home / "backups/albedo-before-image-store-1.sqlite"
57        path.parent.mkdir(exist_ok=True)
58        path.write_bytes(b"approved backup")
59        old = time.time() - 31 * 24 * 60 * 60
60        os.utime(path, (old, old))
61        return path
62
63    # exclusive: restarts the daemon around offline maintenance
64    @exclusive
65    def test_maintenance_refuses_startup_and_second_cleanup_then_releases(self):
66        with Albedo(providers={}) as app:
67            backup = self.backup(app)
68
69            def maintenance(app):
70                database = app.home / "albedo.sqlite"
71                before = hashlib.sha256(database.read_bytes()).digest()
72                helper = self.helper(app, [backup])
73                refused = subprocess.run(
74                    [app.env["ALBEDO_DAEMON"]],
75                    env=dict(app.env, ALBEDO_TOKEN="x" * 32),
76                    cwd=ROOT,
77                    capture_output=True,
78                    timeout=20,
79                )
80                self.assertNotEqual(refused.returncode, 0)
81                self.assertIn(b"storage is in use", refused.stdout + refused.stderr)
82                self.assertEqual(hashlib.sha256(database.read_bytes()).digest(), before)
83                cleanup = subprocess.run(
84                    [
85                        str(ROOT / "cli/bin/albedo"),
86                        "storage",
87                        "prune",
88                        "--all",
89                        "--yes",
90                    ],
91                    env=app.env,
92                    cwd=ROOT,
93                    capture_output=True,
94                    timeout=20,
95                )
96                self.assertNotEqual(cleanup.returncode, 0)
97                self.assertIn(b"storage is in use", cleanup.stderr)
98                self.assertEqual(backup.read_bytes(), b"approved backup")
99                self.close_helper(helper)
100                self.assertEqual(helper.returncode, 0)
101
102            app.restart(prepare=maintenance)
103            self.assertEqual(backup.read_bytes(), b"approved backup")
104            self.assertEqual(app.api("/server").status, 200)
105
106    # exclusive: suspends the daemon and inspects global storage
107    @exclusive
108    @unittest.skipUnless(os.path.isdir("/proc/self"), "requires Linux process status")
109    def test_unreachable_daemon_keeps_cleanup_out(self):
110        with Albedo(providers={}) as app:
111            backup = self.backup(app)
112            pid = app.connection["pid"]
113            with sqlite3.connect(app.home / "albedo.sqlite") as db:
114                db.execute("CREATE TABLE reclaim(data BLOB)")
115                db.execute("INSERT INTO reclaim VALUES(zeroblob(262144))")
116                db.execute("DELETE FROM reclaim")
117            before = app.cli("storage", "--offline", "--json")
118            os.kill(pid, signal.SIGSTOP)
119            try:
120                deadline = time.monotonic() + 10
121                while True:
122                    with open(f"/proc/{pid}/status") as status:
123                        if "State:\tT" in status.read():
124                            break
125                    self.assertLess(
126                        time.monotonic(), deadline, "daemon did not suspend"
127                    )
128                unavailable = subprocess.run(
129                    [str(ROOT / "cli/bin/albedo"), "storage", "--json"],
130                    env=dict(app.env, ALBEDO_DAEMON="/nonexistent/no-launch"),
131                    cwd=ROOT,
132                    capture_output=True,
133                    timeout=30,
134                )
135                self.assertNotEqual(unavailable.returncode, 0)
136                self.assertEqual(unavailable.stdout, b"")
137                cleanup = subprocess.run(
138                    [
139                        str(ROOT / "cli/bin/albedo"),
140                        "storage",
141                        "prune",
142                        "--all",
143                        "--yes",
144                    ],
145                    env=app.env,
146                    cwd=ROOT,
147                    capture_output=True,
148                    timeout=20,
149                )
150                self.assertNotEqual(cleanup.returncode, 0)
151                self.assertEqual(backup.read_bytes(), b"approved backup")
152                self.assertEqual(
153                    json.loads(app.cli("storage", "--offline", "--json")),
154                    json.loads(before),
155                )
156            finally:
157                os.kill(pid, signal.SIGCONT)
158
159    # exclusive: restarts the daemon around offline maintenance
160    @exclusive
161    @unittest.skipUnless(hasattr(os, "pidfd_open"), "requires Linux pidfds")
162    def test_cli_death_and_signals_stop_its_mutation_helper(self):
163        with Albedo(providers={}) as app:
164            backup = self.backup(app)
165
166            def maintenance(app):
167                gate = app.home / "work-gate"
168                os.mkfifo(gate)
169                descriptor = os.open(gate, os.O_RDWR | os.O_NONBLOCK)
170                self.addCleanup(os.close, descriptor)
171                tools = app.home / "test-tools"
172                tools.mkdir()
173                wrapper = tools / "albedo-daemon"
174                held = maintenance_command(
175                    app.home, app.env["ALBEDO_DAEMON"], operation="delete", gate=gate
176                )
177                wrapper.write_text(
178                    f"#!{sys.executable}\n"
179                    "import os, sys\n"
180                    "if sys.argv[1:3] == ['storage', 'maintain']:\n"
181                    f"    command = {held!r}\n"
182                    "else:\n"
183                    f"    command = [{app.env['ALBEDO_DAEMON']!r}, *sys.argv[1:]]\n"
184                    "os.execv(command[0], command)\n"
185                )
186                wrapper.chmod(0o700)
187                for death in (signal.SIGKILL, signal.SIGINT, signal.SIGTERM):
188                    with self.subTest(signal=death):
189                        command = subprocess.Popen(
190                            [
191                                str(ROOT / "cli/bin/albedo"),
192                                "storage",
193                                "prune",
194                                "--backups",
195                                "--yes",
196                            ],
197                            env=dict(app.env, ALBEDO_DAEMON=str(wrapper)),
198                            cwd=ROOT,
199                            stdout=subprocess.PIPE,
200                            stderr=subprocess.PIPE,
201                        )
202                        self.addCleanup(self.close_helper, command)
203                        readable, _, _ = select.select([descriptor], [], [], 20)
204                        self.assertEqual(
205                            readable, [descriptor], "CLI never reached deletion"
206                        )
207                        helper_pid = int(os.read(descriptor, 1024))
208                        helper_descriptor = os.pidfd_open(helper_pid)
209                        try:
210                            command.send_signal(death)
211                            self.assertNotEqual(command.wait(timeout=10), 0)
212                            readable, _, _ = select.select(
213                                [helper_descriptor], [], [], 10
214                            )
215                            self.assertEqual(
216                                readable,
217                                [helper_descriptor],
218                                "mutation helper survived CLI",
219                            )
220                            with sqlite3.connect(
221                                app.home / "daemon.lock", timeout=0
222                            ) as db:
223                                db.execute("BEGIN EXCLUSIVE")
224                            self.assertEqual(backup.read_bytes(), b"approved backup")
225                        finally:
226                            os.close(helper_descriptor)
227                            self.close_helper(command)
228
229            app.restart(prepare=maintenance)
230
231    # exclusive: restarts the daemon and races startup against offline cleanup
232    @exclusive
233    def test_daemon_started_after_preview_refuses_cleanup_before_mutation(self):
234        with Albedo(providers={}) as app:
235            backup = self.backup(app)
236
237            def maintenance(app):
238                ready = app.home / "acquisition-ready"
239                release = app.home / "acquisition-release"
240                os.mkfifo(ready)
241                os.mkfifo(release)
242                descriptor = os.open(ready, os.O_RDWR | os.O_NONBLOCK)
243                self.addCleanup(os.close, descriptor)
244                tools = app.home / "test-tools"
245                tools.mkdir()
246                wrapper = tools / "albedo-daemon"
247                wrapper.write_text(
248                    f"#!{sys.executable}\n"
249                    "import os, sys\n"
250                    "if sys.argv[1:3] == ['storage', 'maintain']:\n"
251                    f"    with open({str(ready)!r}, 'w') as pipe:\n"
252                    "        pipe.write('acquiring')\n"
253                    f"    with open({str(release)!r}) as pipe:\n"
254                    "        pipe.read(1)\n"
255                    f"os.execv({app.env['ALBEDO_DAEMON']!r}, [{app.env['ALBEDO_DAEMON']!r}, *sys.argv[1:]])\n"
256                )
257                wrapper.chmod(0o700)
258                cleanup = subprocess.Popen(
259                    [
260                        str(ROOT / "cli/bin/albedo"),
261                        "storage",
262                        "prune",
263                        "--all",
264                        "--yes",
265                    ],
266                    env=dict(app.env, ALBEDO_DAEMON=str(wrapper)),
267                    cwd=ROOT,
268                    stdout=subprocess.PIPE,
269                    stderr=subprocess.PIPE,
270                )
271                self.addCleanup(self.close_helper, cleanup)
272                readable, _, _ = select.select([descriptor], [], [], 20)
273                self.assertEqual(
274                    readable, [descriptor], "cleanup never approved its preview"
275                )
276                self.assertEqual(os.read(descriptor, 1024), b"acquiring")
277                # The CLI has previewed and approved storage, but has not claimed
278                # ownership. Boot the runner's real daemon before releasing it.
279                app.daemon.boot()
280                with release.open("w") as pipe:
281                    pipe.write("x")
282                _, stderr = cleanup.communicate(timeout=20)
283                self.assertNotEqual(cleanup.returncode, 0)
284                self.assertIn(b"storage is in use", stderr)
285                self.assertEqual(backup.read_bytes(), b"approved backup")
286                self.close_helper(cleanup)
287
288            app.restart(prepare=maintenance)
289
290    # The private home catches filename encoding regressions and Python invocation.
291    def test_offline_inspection_and_prune_do_not_invoke_python(self):
292        with Albedo(providers={}) as app, tempfile.TemporaryDirectory() as directory:
293            root = Path(directory)
294            home = root / "home ?%é"
295            home.mkdir()
296            database = home / "albedo.sqlite"
297            with closing(sqlite3.connect(database)) as db:
298                db.executescript(
299                    "CREATE TABLE sessions(id TEXT); CREATE TABLE transcript(session TEXT,payload BLOB); CREATE TABLE reclaim(data BLOB);"
300                )
301                db.execute("INSERT INTO reclaim VALUES(zeroblob(262144))")
302                db.execute("DELETE FROM reclaim")
303                db.commit()
304            backup = home / "backups/albedo-before-image-store-1.sqlite"
305            backup.parent.mkdir()
306            backup.write_bytes(b"approved backup")
307            old = time.time() - 31 * 24 * 60 * 60
308            os.utime(backup, (old, old))
309            tools = root / "tools"
310            tools.mkdir()
311            python = tools / "python3"
312            python.write_text("#!/bin/sh\nexit 99\n")
313            python.chmod(0o700)
314            env = dict(
315                app.env,
316                ALBEDO_HOME=str(home),
317                PATH=str(tools) + os.pathsep + app.env["PATH"],
318            )
319            for arguments in (("--offline", "--json"), ("prune", "--all", "--yes")):
320                result = subprocess.run(
321                    [str(ROOT / "cli/bin/albedo"), "storage", *arguments],
322                    env=env,
323                    cwd=ROOT,
324                    capture_output=True,
325                    text=True,
326                    timeout=20,
327                )
328                self.assertEqual(result.returncode, 0, result.stderr)
329                if arguments[0] == "--offline":
330                    self.assertGreater(json.loads(result.stdout)["db"]["free_pages"], 0)
331            self.assertFalse(backup.exists())
332            self.assertFalse((home / "daemon.json").exists())
333            with closing(sqlite3.connect(database)) as db:
334                self.assertEqual(db.execute("PRAGMA freelist_count").fetchone()[0], 0)
335
336    # exclusive: restarts the daemon and mutates global storage offline
337    @exclusive
338    def test_successful_cleanup_retains_session_files_and_releases_ownership(self):
339        with Albedo() as app:
340            session = app.session()
341            history = app.history(session)
342
343            def maintenance(app):
344                kernels = app.home / "kernels"
345                kernels.mkdir(exist_ok=True)
346                backups = app.home / "backups"
347                backups.mkdir(exist_ok=True)
348                live = kernels / f"{session}.state"
349                old_orphan = kernels / "old-orphan.state"
350                recent_orphan = kernels / "recent-orphan.state"
351                old_backup = backups / "albedo-before-image-store-1.sqlite"
352                recent_backup = backups / "albedo-before-image-store-2.sqlite"
353                snapshot = pickle.dumps(
354                    {"cwd": str(app.workspace), "names": {}, "definitions": []}
355                )
356                retained = {
357                    live: snapshot,
358                    recent_orphan: b"recent state",
359                    recent_backup: b"recent backup",
360                }
361                for path, content in retained.items():
362                    path.write_bytes(content)
363                old_orphan.write_bytes(snapshot)
364                old_backup.write_bytes(b"old backup")
365                old = time.time() - 31 * 24 * 60 * 60
366                for path in (live, old_orphan, old_backup):
367                    os.utime(path, (old, old))
368
369                with closing(sqlite3.connect(app.home / "albedo.sqlite")) as db:
370                    db.execute("CREATE TABLE cleanup_reclaim(data BLOB)")
371                    db.execute("INSERT INTO cleanup_reclaim VALUES(zeroblob(1048576))")
372                    db.execute("DELETE FROM cleanup_reclaim")
373                    db.commit()
374                before = json.loads(app.cli("storage", "--json"))
375                self.assertGreater(before["db"]["free_pages"], 0)
376
377                app.cli("storage", "prune", "--all", "--yes")
378
379                for path in (old_orphan, old_backup):
380                    self.assertFalse(path.exists(), f"cleanup retained {path}")
381                for path, content in retained.items():
382                    self.assertEqual(path.read_bytes(), content)
383                after = json.loads(app.cli("storage", "--json"))
384                self.assertEqual(after["db"]["free_pages"], 0)
385                self.assertLess(after["database"], before["database"])
386                self.assertEqual(after["db"]["sessions"], before["db"]["sessions"])
387                with closing(
388                    sqlite3.connect(app.home / "daemon.lock", timeout=0)
389                ) as db:
390                    db.execute("BEGIN EXCLUSIVE")
391
392            app.restart(prepare=maintenance)
393            with app.api("/server") as response:
394                self.assertEqual(response.status, 200)
395            self.assertEqual(app.history(session), history)
396
397    # exclusive: mutates global storage and asserts complete daemon-wide reports
398    @exclusive
399    def test_online_report_uses_daemon_storage_without_client_disk_or_python(self):
400        with Albedo() as app:
401            session = app.session()
402            # The daemon reclaims unknown state files a day old on its first
403            # sweep after a start, then only daily. Let a restart's first
404            # sweep run, seen through a sentinel it removes, so the orphan the
405            # report must list is not reclaimed underneath it. Restart before
406            # the rows below are written: they are not a transcript a daemon
407            # would boot on.
408            kernels = app.home / "kernels"
409            kernels.mkdir(exist_ok=True)
410            old = time.time() - 31 * 24 * 60 * 60
411            sentinel = kernels / "sweep-sentinel.state"
412            sentinel.write_bytes(b"sentinel")
413            os.utime(sentinel, (old, old))
414            app.restart()
415            wait_until(
416                lambda: not sentinel.exists(), 30, "the first expiry sweep never ran"
417            )
418            image_reference = b'{"image_hash":"storage-report-image"}'
419            source = "é🐈"
420            trace = b"trace"
421            expected_session_bytes = (
422                3 + len(image_reference) * 2 + len(source) + len(trace)
423            )
424            with closing(sqlite3.connect(app.home / "albedo.sqlite")) as db:
425                db.execute("DELETE FROM transcript WHERE session=?", (session,))
426                db.execute(
427                    "INSERT INTO transcript(session,payload) VALUES(?,?)",
428                    (session, image_reference),
429                )
430                if "pinned_context" in {
431                    row[1] for row in db.execute("PRAGMA table_info(sessions)")
432                }:
433                    db.execute(
434                        "UPDATE sessions SET pinned_context=? WHERE id=?",
435                        ("pin", session),
436                    )
437                db.execute(
438                    "INSERT INTO images(hash,data) VALUES(?,?)",
439                    ("storage-report-image", b"imagebytes"),
440                )
441                db.execute(
442                    "INSERT INTO cells(id,session,source,status,payload) VALUES(?,?,?,?,?)",
443                    (
444                        "storage-report-cell",
445                        session,
446                        source,
447                        "finished",
448                        image_reference,
449                    ),
450                )
451                db.execute(
452                    "INSERT INTO cell_traces(id,payload) VALUES(?,?)",
453                    ("storage-report-cell", trace),
454                )
455                db.execute("CREATE TABLE report_reclaim(data BLOB)")
456                db.execute("INSERT INTO report_reclaim VALUES(zeroblob(262144))")
457                db.execute("DELETE FROM report_reclaim")
458                db.commit()
459            orphan = kernels / "report-orphan.state"
460            live = kernels / f"{session}.state"
461            orphan.write_bytes(b"orphan")
462            live.write_bytes(b"live")
463            backups = app.home / "backups"
464            baseline_backups = (
465                sum(path.stat().st_size for path in backups.iterdir() if path.is_file())
466                if backups.exists()
467                else 0
468            )
469            backup = self.backup(app)
470            recent_backup = backup.parent / "albedo-before-image-store-2.sqlite"
471            recent_backup.write_bytes(b"recent")
472            unrelated_backup = backup.parent / "personal.sqlite"
473            unrelated_backup.write_bytes(b"personal")
474            for path in (orphan, live):
475                os.utime(path, (old, old))
476            offline = json.loads(app.cli("storage", "--offline", "--json"))
477            expected_other = sum(
478                path.stat().st_size
479                for path in app.home.iterdir()
480                if path.is_file()
481                and path.name
482                not in {"albedo.sqlite", "albedo.sqlite-wal", "albedo.sqlite-shm"}
483            )
484            with tempfile.TemporaryDirectory() as directory:
485                client_home = Path(directory)
486                (client_home / "daemon.json").write_bytes(
487                    (app.home / "daemon.json").read_bytes()
488                )
489                # These paths cannot be inspected as storage, and any attempted
490                # diagnostic Python launch leaves an observable marker.
491                (client_home / "albedo.sqlite").mkdir()
492                (client_home / "kernels").write_bytes(b"not a directory")
493                tools = client_home / "tools"
494                tools.mkdir()
495                marker = client_home / "python-called"
496                python = tools / "python3"
497                python.write_text(f"#!/bin/sh\ntouch '{marker}'\nexit 91\n")
498                python.chmod(0o700)
499                command = subprocess.run(
500                    [str(ROOT / "cli/bin/albedo"), "storage", "--json"],
501                    env=dict(
502                        app.env,
503                        ALBEDO_HOME=str(client_home),
504                        PATH=str(tools) + os.pathsep + app.env["PATH"],
505                    ),
506                    cwd=ROOT,
507                    capture_output=True,
508                    text=True,
509                    timeout=20,
510                )
511                self.assertEqual(command.returncode, 0, command.stderr)
512                report = json.loads(command.stdout)
513                self.assertEqual(
514                    report["db"]["sessions"],
515                    [{"id": session, "bytes": expected_session_bytes}],
516                )
517                self.assertEqual(report["db"]["images"], 10)
518                self.assertEqual(
519                    report["old_kernels"], [{"path": str(orphan), "bytes": 6}]
520                )
521                self.assertEqual(
522                    report["old_backups"], [{"path": str(backup), "bytes": 15}]
523                )
524                self.assertEqual(report["kernels"], 10)
525                self.assertEqual(report["backups"], baseline_backups + 29)
526                self.assertEqual(report["recent_backups"], 6)
527                self.assertEqual(report["recent_backup_count"], 1)
528                self.assertEqual(report["other"], expected_other)
529                self.assertGreater(report["db"]["free_pages"], 0)
530                for key in (
531                    "old_kernels",
532                    "old_backups",
533                    "kernels",
534                    "backups",
535                    "other",
536                    "recent_backups",
537                    "recent_backup_count",
538                ):
539                    self.assertEqual(report[key], offline[key], key)
540                self.assertEqual(report["db"], offline["db"])
541                self.assertGreater(report["database"], 0)
542                self.assertGreater(report["db"]["page_size"], 0)
543                self.assertFalse(
544                    marker.exists(), "online reporting invoked client Python"
545                )
546                record = json.loads((client_home / "daemon.json").read_text())
547                self.assertEqual(record, app.connection)
548
549    # Only the private client home changes; the daemon is not stopped or mutated.
550    def test_offline_reports_supported_older_layouts_and_rejects_unknown_without_changes(
551        self,
552    ):
553        with Albedo(providers={}) as app, tempfile.TemporaryDirectory() as directory:
554            home = Path(directory) / "missing"
555
556            def report(*options, path=None):
557                return subprocess.run(
558                    [str(ROOT / "cli/bin/albedo"), "storage", *options, "--json"],
559                    env=dict(
560                        app.env,
561                        ALBEDO_HOME=str(home),
562                        ALBEDO_DAEMON=app.env["ALBEDO_DAEMON"],
563                        PATH=app.env["PATH"] if path is None else path,
564                    ),
565                    cwd=ROOT,
566                    capture_output=True,
567                    text=True,
568                    timeout=20,
569                )
570
571            missing = report("--offline", path="")
572            self.assertEqual(missing.returncode, 0, missing.stderr)
573            empty = json.loads(missing.stdout)
574            self.assertEqual(empty["db"]["sessions"], [])
575            self.assertEqual(empty["old_kernels"], [])
576            self.assertEqual(empty["old_backups"], [])
577            self.assertFalse(home.exists())
578            home.mkdir()
579            # No record also selects offline automatically, without creating one.
580            absent = report(path="")
581            self.assertEqual(absent.returncode, 0, absent.stderr)
582            self.assertEqual(json.loads(absent.stdout), empty)
583            self.assertEqual(list(home.iterdir()), [])
584            database = home / "albedo.sqlite"
585            with closing(sqlite3.connect(database)) as db:
586                db.executescript(
587                    "CREATE TABLE sessions(id TEXT); CREATE TABLE transcript(session TEXT,payload BLOB); CREATE TABLE images(data TEXT);"
588                )
589                db.execute("INSERT INTO sessions VALUES(?)", ("older",))
590                db.execute("INSERT INTO transcript VALUES(?,?)", ("older", b"history"))
591                db.execute("INSERT INTO images VALUES(?)", ("é🐈",))
592                db.commit()
593            before = database.read_bytes()
594            supported = report("--offline")
595            self.assertEqual(supported.returncode, 0, supported.stderr)
596            older = json.loads(supported.stdout)
597            self.assertEqual(older["db"]["sessions"], [{"id": "older", "bytes": 7}])
598            self.assertEqual(older["db"]["images"], 2)
599            tools = Path(directory) / "tools"
600            tools.mkdir()
601            python = tools / "python3"
602            python.write_text("#!/bin/sh\nexit 99\n")
603            python.chmod(0o700)
604            without_python = report(
605                "--offline", path=str(tools) + os.pathsep + app.env["PATH"]
606            )
607            self.assertEqual(without_python.returncode, 0, without_python.stderr)
608            self.assertEqual(json.loads(without_python.stdout), older)
609            self.assertEqual(database.read_bytes(), before)
610            self.assertEqual({path.name for path in home.iterdir()}, {"albedo.sqlite"})
611            with closing(sqlite3.connect(database)) as db:
612                db.execute(
613                    "ALTER TABLE transcript RENAME COLUMN payload TO unknown_payload"
614                )
615                db.commit()
616            kernels = home / "kernels"
617            kernels.mkdir()
618            candidate = kernels / "unproven.state"
619            candidate.write_bytes(b"retain")
620            old = time.time() - 31 * 24 * 60 * 60
621            os.utime(candidate, (old, old))
622            before = database.read_bytes()
623            unsupported = report("--offline")
624            self.assertNotEqual(unsupported.returncode, 0)
625            self.assertIn("unsupported storage layout", unsupported.stderr)
626            self.assertEqual(unsupported.stdout, "")
627            self.assertEqual(database.read_bytes(), before)
628            self.assertEqual(candidate.read_bytes(), b"retain")
629            self.assertFalse((home / "daemon.json").exists())
630            source = Path(directory) / "wal-source.sqlite"
631            with closing(sqlite3.connect(source)) as writer:
632                writer.execute("PRAGMA journal_mode=WAL")
633                writer.executescript(
634                    "CREATE TABLE sessions(id TEXT); CREATE TABLE transcript(session TEXT,payload BLOB);"
635                )
636                writer.execute("INSERT INTO sessions VALUES('wal-session')")
637                writer.execute(
638                    "INSERT INTO transcript VALUES('wal-session',?)",
639                    (b"committed in WAL",),
640                )
641                writer.commit()
642                home = Path(directory) / "wal-home"
643                home.mkdir()
644                (home / "albedo.sqlite").write_bytes(source.read_bytes())
645                (home / "albedo.sqlite-wal").write_bytes(
646                    Path(str(source) + "-wal").read_bytes()
647                )
648            originals = {path.name: path.read_bytes() for path in home.iterdir()}
649            self.assertNotIn("albedo.sqlite-shm", originals)
650            wal_report = report("--offline")
651            self.assertEqual(wal_report.returncode, 0, wal_report.stderr)
652            self.assertEqual(
653                json.loads(wal_report.stdout)["db"]["sessions"],
654                [{"id": "wal-session", "bytes": 16}],
655            )
656            self.assertEqual(
657                {path.name: path.read_bytes() for path in home.iterdir()}, originals
658            )
659
660    # Only the private client home changes; the daemon is not stopped or mutated.
661    def test_report_mode_selection_never_launches_or_hides_discovery_and_auth_failures(
662        self,
663    ):
664        with Albedo(providers={}) as app, tempfile.TemporaryDirectory() as directory:
665            home = Path(directory)
666            record = home / "daemon.json"
667
668            def report(*options):
669                return subprocess.run(
670                    [str(ROOT / "cli/bin/albedo"), "storage", *options, "--json"],
671                    env=dict(
672                        app.env,
673                        ALBEDO_HOME=str(home),
674                        ALBEDO_DAEMON=app.env["ALBEDO_DAEMON"],
675                    ),
676                    cwd=ROOT,
677                    capture_output=True,
678                    text=True,
679                    timeout=20,
680                )
681
682            absent = report()
683            self.assertEqual(absent.returncode, 0, absent.stderr)
684            empty = json.loads(absent.stdout)
685            self.assertEqual(list(home.iterdir()), [])
686            with socket.socket() as endpoint:
687                endpoint.bind(("127.0.0.1", 0))
688                refused_port = endpoint.getsockname()[1]
689            stale = dict(app.connection, pid=2147483647, port=refused_port)
690            record.write_text(json.dumps(stale))
691            stale_bytes = record.read_bytes()
692            stale_result = report()
693            self.assertEqual(stale_result.returncode, 0, stale_result.stderr)
694            self.assertEqual(
695                json.loads(stale_result.stdout), dict(empty, other=len(stale_bytes))
696            )
697            self.assertEqual(record.read_bytes(), stale_bytes)
698            for content in (
699                "not JSON",
700                json.dumps(dict(app.connection, token="wrong-token")),
701            ):
702                with self.subTest(record=content):
703                    record.write_text(content)
704                    failed = report()
705                    self.assertNotEqual(failed.returncode, 0)
706                    self.assertEqual(failed.stdout, "")
707                    self.assertEqual(record.read_text(), content)
708                    explicit = report("--offline")
709                    self.assertEqual(explicit.returncode, 0, explicit.stderr)
710                    # The discovery record itself is a regular local file.
711                    local = json.loads(explicit.stdout)
712                    self.assertEqual(local["other"], len(content.encode()))
713                    self.assertEqual(local["db"]["sessions"], [])
714                    self.assertEqual(record.read_text(), content)
715                    self.assertEqual(
716                        {path.name for path in home.iterdir()}, {"daemon.json"}
717                    )
718
719    # The diagnostic and its snapshot use private homes, tools, and scratch paths.
720    @unittest.skipUnless(hasattr(os, "pidfd_open"), "requires Linux pidfds")
721    def test_canceling_offline_diagnostic_reaps_child_and_removes_private_snapshot(
722        self,
723    ):
724        with Albedo(providers={}) as app, tempfile.TemporaryDirectory() as directory:
725            root = Path(directory)
726            home = root / "home"
727            home.mkdir()
728            database = home / "albedo.sqlite"
729            with closing(sqlite3.connect(database)) as db:
730                db.executescript(
731                    "CREATE TABLE sessions(id TEXT); CREATE TABLE transcript(session TEXT,payload BLOB);"
732                )
733                db.commit()
734            before = database.read_bytes()
735            scratch = root / "scratch"
736            scratch.mkdir()
737            ready = root / "ready"
738            os.mkfifo(ready)
739            descriptor = os.open(ready, os.O_RDWR | os.O_NONBLOCK)
740            self.addCleanup(os.close, descriptor)
741            tools = root / "tools"
742            tools.mkdir()
743            helper = tools / "albedo-daemon"
744            helper.write_text(
745                f"#!{sys.executable}\n"
746                "import os, sys, threading\n"
747                "from pathlib import Path\n"
748                "snapshot = Path(sys.argv[4])\n"
749                "(snapshot / 'albedo.sqlite').write_bytes(b'partial snapshot')\n"
750                f"with open({str(ready)!r}, 'w') as pipe:\n"
751                "    pipe.write(str(os.getpid()) + '\\n')\n"
752                "threading.Event().wait()\n"
753            )
754            helper.chmod(0o700)
755            for death in (signal.SIGINT, signal.SIGTERM):
756                with self.subTest(signal=death):
757                    command = subprocess.Popen(
758                        [
759                            str(ROOT / "cli/bin/albedo"),
760                            "storage",
761                            "--offline",
762                            "--json",
763                        ],
764                        env=dict(
765                            app.env,
766                            ALBEDO_HOME=str(home),
767                            TMPDIR=str(scratch),
768                            ALBEDO_DAEMON=str(helper),
769                        ),
770                        cwd=ROOT,
771                        stdout=subprocess.PIPE,
772                        stderr=subprocess.PIPE,
773                    )
774                    self.addCleanup(self.close_helper, command)
775                    readable, _, _ = select.select([descriptor], [], [], 20)
776                    self.assertEqual(
777                        readable,
778                        [descriptor],
779                        "diagnostic never created its private snapshot",
780                    )
781                    child = int(os.read(descriptor, 1024))
782                    child_descriptor = os.pidfd_open(child)
783                    try:
784                        self.assertEqual(len(list(scratch.iterdir())), 1)
785                        command.send_signal(death)
786                        stdout, _ = command.communicate(timeout=10)
787                        self.assertNotEqual(command.returncode, 0)
788                        self.assertEqual(stdout, b"")
789                        readable, _, _ = select.select([child_descriptor], [], [], 10)
790                        self.assertEqual(
791                            readable,
792                            [child_descriptor],
793                            "diagnostic child survived cancellation",
794                        )
795                        self.assertEqual(
796                            list(scratch.iterdir()),
797                            [],
798                            "private snapshot leaked after cancellation",
799                        )
800                        self.assertEqual(database.read_bytes(), before)
801                        self.assertEqual(
802                            {path.name for path in home.iterdir()}, {"albedo.sqlite"}
803                        )
804                    finally:
805                        os.close(child_descriptor)
806                        self.close_helper(command)