diff --git a/docs/plans/topology-target.yaml b/docs/plans/topology-target.yaml index 6a1a38a56b..0adfa37b5b 100644 --- a/docs/plans/topology-target.yaml +++ b/docs/plans/topology-target.yaml @@ -70,7 +70,7 @@ files: owner: stable cross_cut: { api: async } - path: polylogue/api/archive.py - loc: 6788 + loc: 6918 target: polylogue/api/archive.py owner: stable cross_cut: { api: async } @@ -426,7 +426,7 @@ files: owner: archive-query reason: archive-domain query semantics - path: polylogue/archive/query/runtime.py - loc: 32 + loc: 34 target: polylogue/archive/query/runtime.py owner: archive-query reason: archive-domain query semantics @@ -436,7 +436,7 @@ files: owner: archive-query reason: archive-domain query semantics - path: polylogue/archive/query/runtime_matching.py - loc: 184 + loc: 199 target: polylogue/archive/query/runtime_matching.py owner: archive-query reason: archive-domain query semantics @@ -936,7 +936,7 @@ files: target: polylogue/cli/commands/maintenance/_blob_publications.py owner: stable - path: polylogue/cli/commands/maintenance/_embeddings.py - loc: 173 + loc: 143 target: polylogue/cli/commands/maintenance/_embeddings.py owner: stable - path: polylogue/cli/commands/maintenance/_embeddings_rescue.py @@ -1072,7 +1072,7 @@ files: target: polylogue/cli/query_group.py owner: stable - path: polylogue/cli/query_output.py - loc: 983 + loc: 972 target: polylogue/cli/query_output.py owner: stable - path: polylogue/cli/query_output_contracts.py @@ -1080,7 +1080,7 @@ files: target: polylogue/cli/query_output_contracts.py owner: stable - path: polylogue/cli/query_semantic.py - loc: 264 + loc: 275 target: polylogue/cli/query_semantic.py owner: stable - path: polylogue/cli/query_set_read.py @@ -1088,7 +1088,7 @@ files: target: polylogue/cli/query_set_read.py owner: stable - path: polylogue/cli/query_stats.py - loc: 722 + loc: 236 target: polylogue/cli/query_stats.py owner: stable - path: polylogue/cli/query_verbs.py @@ -1487,6 +1487,10 @@ files: loc: 1057 target: polylogue/daemon/backup.py owner: stable + - path: polylogue/daemon/blob_gc_periodic.py + loc: 81 + target: polylogue/daemon/blob_gc_periodic.py + owner: stable - path: polylogue/daemon/blob_integrity_alerts.py loc: 51 target: polylogue/daemon/blob_integrity_alerts.py @@ -1504,7 +1508,7 @@ files: target: polylogue/daemon/catchup_status.py owner: stable - path: polylogue/daemon/cli.py - loc: 2740 + loc: 2760 target: polylogue/daemon/cli.py owner: stable - path: polylogue/daemon/compare.py @@ -2307,7 +2311,7 @@ files: target: polylogue/operations/import_operations.py owner: stable - path: polylogue/operations/mutation_actuators.py - loc: 991 + loc: 1377 target: polylogue/operations/mutation_actuators.py owner: stable - path: polylogue/operations/mutation_transaction.py @@ -2327,7 +2331,7 @@ files: target: polylogue/operations/route_observation.py owner: stable - path: polylogue/operations/specs.py - loc: 1118 + loc: 1256 target: polylogue/operations/specs.py owner: stable - path: polylogue/operations/status_protocol.py @@ -3212,7 +3216,7 @@ files: target: polylogue/sources/live/append_ingest.py owner: stable - path: polylogue/sources/live/batch.py - loc: 2587 + loc: 2610 target: polylogue/sources/live/batch.py owner: stable - path: polylogue/sources/live/batch_observability.py @@ -3260,7 +3264,7 @@ files: target: polylogue/sources/live/sqlite_locking.py owner: stable - path: polylogue/sources/live/watcher.py - loc: 1390 + loc: 1413 target: polylogue/sources/live/watcher.py owner: stable - path: polylogue/sources/origin_specs.py @@ -3622,7 +3626,7 @@ files: target: polylogue/storage/fts/freshness.py owner: stable - path: polylogue/storage/fts/fts_lifecycle.py - loc: 1182 + loc: 1209 target: polylogue/storage/fts/fts_lifecycle.py owner: stable - path: polylogue/storage/fts/pl_fold.py @@ -4009,7 +4013,7 @@ files: target: polylogue/storage/sqlite/archive_tiers/__init__.py owner: stable - path: polylogue/storage/sqlite/archive_tiers/archive.py - loc: 12808 + loc: 12949 target: polylogue/storage/sqlite/archive_tiers/archive.py owner: stable - path: polylogue/storage/sqlite/archive_tiers/archive_init.py @@ -4077,7 +4081,7 @@ files: target: polylogue/storage/sqlite/archive_tiers/pricing_seed.py owner: stable - path: polylogue/storage/sqlite/archive_tiers/revision_application.py - loc: 396 + loc: 402 target: polylogue/storage/sqlite/archive_tiers/revision_application.py owner: stable - path: polylogue/storage/sqlite/archive_tiers/self_verify.py @@ -4121,7 +4125,7 @@ files: target: polylogue/storage/sqlite/archive_tiers/user_write.py owner: stable - path: polylogue/storage/sqlite/archive_tiers/write.py - loc: 5278 + loc: 5304 target: polylogue/storage/sqlite/archive_tiers/write.py owner: stable - path: polylogue/storage/sqlite/async_sqlite.py diff --git a/docs/topology-status.md b/docs/topology-status.md index a7f994c0cb..b32969a709 100644 --- a/docs/topology-status.md +++ b/docs/topology-status.md @@ -28,12 +28,12 @@ Generated by `devtools render topology-status`. Reads `docs/plans/topology-targe ### Summary -- **Stable** (no move scoped): 905 +- **Stable** (no move scoped): 906 - **Kernel** (polylogue/ root): 8 - **Primitives** (storage-root): 19 - **TBD** (cell needs explicit assignment): 10 -- **Total declared**: 1077 -- **Realized polylogue/**/*.py**: 1077 files declared +- **Total declared**: 1078 +- **Realized polylogue/**/*.py**: 1078 files declared ### TBD cells (require explicit routing) diff --git a/polylogue/cli/commands/maintenance/_embeddings.py b/polylogue/cli/commands/maintenance/_embeddings.py index 97bc4c34d5..3c4533fe92 100644 --- a/polylogue/cli/commands/maintenance/_embeddings.py +++ b/polylogue/cli/commands/maintenance/_embeddings.py @@ -1,4 +1,12 @@ -"""``maintenance embedding-orphan-reconcile``: reconcile embeddings.db orphans.""" +"""``maintenance embedding-orphan-reconcile``: inspect embeddings.db orphans. + +Read-only by design (automagic-invariants, polylogue-gd6v/4jsk): daemon +convergence (``periodic_embedding_orphan_reconcile_check``, +``polylogue/daemon/embedding_backlog.py``) already reconciles this backlog +automatically in bounded batches, so a manual mutate/apply path here would +be exactly the redundant "demoted to break-glass" surface that ruling +forbids rather than deletes. This command stays as pure diagnostic preview. +""" from __future__ import annotations @@ -7,8 +15,7 @@ import click -from polylogue.config import Config -from polylogue.paths import archive_file_set_root_for_paths, archive_root, db_path, render_root +from polylogue.paths import archive_file_set_root_for_paths, archive_root, db_path if TYPE_CHECKING: from polylogue.storage.embeddings.reconcile import EmbeddingOrphanReconcileReport @@ -26,7 +33,7 @@ "--max-count", type=int, default=None, - help="Maximum number of orphan embedding rows to reconcile or preview (default: unbounded for inspect).", + help="Maximum number of orphan embedding rows to preview (default: unbounded).", ) @click.option( "--quiet-window-seconds", @@ -42,11 +49,6 @@ show_default=True, help="Maximum number of representative samples to include.", ) -@click.option( - "--yes", - is_flag=True, - help="Delete orphan embedding rows. Without this flag, the command only inspects and reports (break-glass).", -) @click.option( "--output-format", "output_format", @@ -59,58 +61,33 @@ def embedding_orphan_reconcile_command( max_count: int | None, quiet_window_seconds: int, sample_limit: int, - yes: bool, output_format: str, ) -> None: - """Inspect (default) or reconcile embeddings.db rows orphaned by an index rebuild. + """Inspect embeddings.db rows orphaned by an index rebuild. Read-only. An index rebuild (full re-ingest, ``ops reset --index``, a provider full-replace parse) can leave ``message_embeddings_meta`` / ``message_embeddings`` / ``embedding_status`` rows in ``embeddings.db`` pointing at message/session identities that no longer exist in the rebuilt ``index.db``. Daemon convergence reconciles these automatically - in bounded batches; this command is the manual inspect/break-glass path. + in bounded batches; this command is diagnostic preview only. """ - from polylogue.storage.embeddings.reconcile import DEFAULT_MAX_COUNT, reconcile_embedding_orphans + from polylogue.storage.embeddings.reconcile import reconcile_embedding_orphans root = archive_file_set_root_for_paths(archive_root_path=archive_root(), db_anchor=db_path()) index_db = root / "index.db" embeddings_db = root / "embeddings.db" - if yes: - from polylogue.maintenance.offline_guard import running_daemon_pid - from polylogue.storage.index_generation import RebuildLease, RebuildLeaseUnavailableError - - active_config = Config(archive_root=root, render_root=render_root(), sources=[], db_path=index_db) - try: - with RebuildLease(root): - daemon_pid = running_daemon_pid(active_config) - if daemon_pid is not None: - raise click.ClickException( - f"embedding orphan reconcile refused while polylogued PID {daemon_pid} is running" - ) - report = reconcile_embedding_orphans( - index_db, - embeddings_db, - dry_run=False, - max_count=DEFAULT_MAX_COUNT if max_count is None else max_count, - sample_size=sample_limit, - quiet_window_ms=quiet_window_seconds * 1000, - mutation_authority="offline-exclusive", - ) - except RebuildLeaseUnavailableError as exc: - raise click.ClickException(str(exc)) from exc - else: - report = reconcile_embedding_orphans( - index_db, - embeddings_db, - dry_run=True, - max_count=max_count, - sample_size=sample_limit, - quiet_window_ms=quiet_window_seconds * 1000, - ) + report = reconcile_embedding_orphans( + index_db, + embeddings_db, + dry_run=True, + max_count=max_count, + sample_size=sample_limit, + quiet_window_ms=quiet_window_seconds * 1000, + ) payload = { "mode": "embedding_orphan_reconcile", - "mutates": bool(yes), + "mutates": False, **report.to_dict(), } @@ -133,10 +110,9 @@ def _render_embedding_orphan_reconcile_plain(report: EmbeddingOrphanReconcileRep happens. ``scanned_message_meta_rows``/``scanned_vector_rows`` remain informational counts of the (deduped) content-addressed tables. """ - click.echo("Embedding orphan reconcile") + click.echo("Embedding orphan reconcile (inspect, read-only)") click.echo(f"Index DB: {report.index_db}") click.echo(f"Embeddings DB: {report.embeddings_db}") - click.echo(f"Mode: {'dry-run' if report.dry_run else 'apply'}") click.echo( f"Scanned: {report.scanned_message_meta_rows:,} distinct vector meta row(s) " f"(content-addressed, shared), {report.scanned_vector_rows:,} distinct vector row(s), " @@ -153,16 +129,10 @@ def _render_embedding_orphan_reconcile_plain(report: EmbeddingOrphanReconcileRep f"{report.skipped_recent_status_rows:,} status row(s) " f"(within {report.quiet_window_ms // 1000}s)" ) - if report.dry_run: - click.echo( - f"Would remove: {report.candidate_message_rows:,} message ref(s), " - f"{report.candidate_status_rows:,} status row(s)" - ) - else: - click.echo( - f"Removed: {report.removed_message_rows:,} message ref(s), " - f"{report.removed_status_rows:,} status row(s)" - ) + click.echo( + f"Would remove: {report.candidate_message_rows:,} message ref(s), " + f"{report.candidate_status_rows:,} status row(s)" + ) if report.sessions_recounted: click.echo(f"Recounted: {report.sessions_recounted:,} session(s) message_count_embedded") click.echo(f"More pending: {report.more_pending}") diff --git a/polylogue/daemon/blob_gc_periodic.py b/polylogue/daemon/blob_gc_periodic.py new file mode 100644 index 0000000000..cde27d645c --- /dev/null +++ b/polylogue/daemon/blob_gc_periodic.py @@ -0,0 +1,81 @@ +"""Bounded ambient blob-GC drain for the daemon. + +``polylogue ops maintenance blob-gc`` is a mechanical, non-judgment +operation (delete a content-addressed blob once its refcount is zero and +it clears the generation-age gate — see ``storage/blob_gc.py``'s module +docstring for the full safety invariant list) with no daemon-side +equivalent before this module: the CLI command was the *only* way to +reclaim disk space, in the same shape ``embedding-orphan-reconcile`` was +before its own daemon loop was added. This mirrors that precedent — +periodic ambient drain in bounded batches, same as +``periodic_embedding_orphan_reconcile_check``. +""" + +from __future__ import annotations + +import asyncio +import sqlite3 +from pathlib import Path +from typing import TYPE_CHECKING + +from polylogue.logging import get_logger +from polylogue.sources.live.sqlite_locking import is_transient_sqlite_lock + +if TYPE_CHECKING: + from polylogue.storage.blob_gc import BlobGCResult + +logger = get_logger(__name__) + +BLOB_GC_INTERVAL_SECONDS = 900 +BLOB_GC_MAX_BATCH = 200 + + +async def periodic_blob_gc_check(*, catch_up_complete: asyncio.Event | None = None) -> None: + """Periodically reclaim one bounded batch of unreferenced, aged-out blobs.""" + from polylogue.paths import archive_root, source_db_path + + if catch_up_complete is not None: + await catch_up_complete.wait() + while True: + await asyncio.sleep(BLOB_GC_INTERVAL_SECONDS) + try: + from polylogue.daemon.write_coordinator import daemon_write_coordinator + + result = await daemon_write_coordinator().run_sync( + "maintenance.blob_gc", + run_blob_gc_once, + source_db_path(), + archive_root() / "blob", + ) + if result is not None and result.deleted_count: + logger.info( + "blob gc: reclaimed %d blob(s), %d byte(s)", + result.deleted_count, + result.reclaimed_bytes, + ) + except sqlite3.OperationalError as exc: + if is_transient_sqlite_lock(exc): + logger.info("blob gc: archive busy; retrying on next tick: %s", exc) + continue + logger.warning("blob gc: periodic reclaim failed", exc_info=True) + except Exception: + logger.warning("blob gc: periodic reclaim failed", exc_info=True) + + +def run_blob_gc_once(source_db_path_arg: Path, blob_dir: Path) -> BlobGCResult | None: + """Run one bounded daemon blob-GC pass, or ``None`` if the blob store is absent.""" + from polylogue.storage.blob_gc import run_blob_gc_report + + if not blob_dir.is_dir(): + return None + if not source_db_path_arg.is_file(): + return None + return run_blob_gc_report(source_db_path_arg, blob_dir, max_batch=BLOB_GC_MAX_BATCH, dry_run=False) + + +__all__ = [ + "BLOB_GC_INTERVAL_SECONDS", + "BLOB_GC_MAX_BATCH", + "periodic_blob_gc_check", + "run_blob_gc_once", +] diff --git a/polylogue/daemon/cli.py b/polylogue/daemon/cli.py index d1a83ca27d..b55252df9d 100644 --- a/polylogue/daemon/cli.py +++ b/polylogue/daemon/cli.py @@ -2015,6 +2015,7 @@ async def run_daemon_services( # both write-heavy; running them concurrently makes SQLite maintenance # time out behind the daemon's own writer. if not watcher_blocked: + from polylogue.daemon.blob_gc_periodic import periodic_blob_gc_check from polylogue.daemon.convergence import DaemonConverger from polylogue.daemon.convergence_stages import make_default_convergence_stages from polylogue.daemon.embedding_backlog import ( @@ -2060,6 +2061,7 @@ async def run_daemon_services( _periodic_status_snapshot_refresh(), periodic_judgment_automation_sweep(catch_up_complete=catch_up_complete_gate), periodic_fts_identity_drift_recompute(catch_up_complete=catch_up_complete_gate), + periodic_blob_gc_check(catch_up_complete=catch_up_complete_gate), ] if enable_source_catchup: periodic_loops.append(_periodic_drive_source_catchup(catch_up_complete=catch_up_complete_gate)) diff --git a/tests/unit/cli/test_archive_maintenance_cli.py b/tests/unit/cli/test_archive_maintenance_cli.py index 7a8867f77f..863af2370b 100644 --- a/tests/unit/cli/test_archive_maintenance_cli.py +++ b/tests/unit/cli/test_archive_maintenance_cli.py @@ -1290,147 +1290,26 @@ def test_embedding_orphan_reconcile_cli_plain_dry_run_reports_would_remove_count assert conn.execute("SELECT COUNT(*) FROM message_embeddings_meta").fetchone()[0] == 1 -def test_embedding_orphan_reconcile_cli_apply_removes_rows( +def test_embedding_orphan_reconcile_cli_has_no_mutate_flag( cli_workspace: dict[str, Path], cli_runner: CliRunner, ) -> None: - session_id, message_id = _seed_orphan_embedding_row(cli_workspace["archive_root"]) - - result = cli_runner.invoke( - cli, - [ - "--plain", - "ops", - "maintenance", - "embedding-orphan-reconcile", - "--yes", - "--output-format", - "json", - ], - catch_exceptions=False, - ) - - assert result.exit_code == 0 - payload = json.loads(result.output) - assert payload["mutates"] is True - assert payload["dry_run"] is False - assert payload["removed_message_rows"] == 1 - # v4: message_embeddings/message_embeddings_meta are content-addressed and - # never deleted by this reconciler -- only the message_id -> hash ref is. - assert payload["removed_vector_rows"] == 0 - assert payload["sessions_recounted"] == 1 - with sqlite3.connect(cli_workspace["archive_root"] / "embeddings.db") as conn: - assert ( - conn.execute("SELECT COUNT(*) FROM message_embedding_refs WHERE message_id = ?", (message_id,)).fetchone()[ - 0 - ] - == 0 - ) - status = conn.execute( - "SELECT message_count_embedded, needs_reindex FROM embedding_status WHERE session_id = ?", (session_id,) - ).fetchone() - assert status is not None - assert status[0] == 0 - assert status[1] == 1 - - -def test_embedding_orphan_reconcile_cli_apply_is_bounded_by_default( - cli_workspace: dict[str, Path], - cli_runner: CliRunner, -) -> None: - session_id, _message_id = _seed_orphan_embedding_row(cli_workspace["archive_root"]) - embeddings_db = cli_workspace["archive_root"] / "embeddings.db" - with sqlite3.connect(embeddings_db) as conn: - # v4: per-message identity for the reconciler's scan is - # message_embedding_refs (message_id-keyed) -- message_embeddings_meta - # is content-addressed and never independently orphaned, so 500 - # distinct orphan candidates means 500 distinct refs, not meta rows. - conn.executemany( - """ - INSERT INTO message_embedding_refs ( - message_id, session_id, origin, embedding_input_hash, embedded_at_ms - ) VALUES (?, ?, 'codex-session', ?, 1700000000000) - """, - [ - ( - f"{session_id}:zz-orphan-{position:03d}", - f"{session_id}:zz-orphan-{position:03d}", - hashlib.sha256(f"{session_id}:zz-orphan-{position:03d}".encode()).digest(), - ) - for position in range(500) - ], - ) - - result = cli_runner.invoke( - cli, - [ - "--plain", - "ops", - "maintenance", - "embedding-orphan-reconcile", - "--yes", - "--output-format", - "json", - ], - catch_exceptions=False, - ) - - assert result.exit_code == 0 - payload = json.loads(result.output) - assert payload["candidate_message_rows"] == 500 - assert payload["removed_message_rows"] == 500 - assert payload["more_pending"] is True - with sqlite3.connect(embeddings_db) as conn: - assert conn.execute("SELECT COUNT(*) FROM message_embeddings_meta").fetchone()[0] == 1 - - -def test_embedding_orphan_reconcile_cli_apply_requires_exclusive_offline_lease( - cli_workspace: dict[str, Path], - cli_runner: CliRunner, -) -> None: - """Production CLI proof: a live/shared writer lease blocks break-glass apply.""" - from polylogue.storage.index_generation import ActiveWriterLease - - _seed_orphan_embedding_row(cli_workspace["archive_root"]) - writer = ActiveWriterLease(cli_workspace["archive_root"]) - writer.acquire() - try: - result = cli_runner.invoke( - cli, - ["--plain", "ops", "maintenance", "embedding-orphan-reconcile", "--yes"], - ) - finally: - writer.close() - - assert result.exit_code != 0 - assert "index rebuild lease is already held" in result.output - - -def test_embedding_orphan_reconcile_cli_apply_refuses_stale_index_schema( - cli_workspace: dict[str, Path], - cli_runner: CliRunner, -) -> None: - """The break-glass writer cannot treat a pre-rebuild index as deletion truth.""" - _session_id, message_id = _seed_orphan_embedding_row(cli_workspace["archive_root"]) - index_db = cli_workspace["archive_root"] / "index.db" - with sqlite3.connect(index_db) as conn: - conn.execute("PRAGMA user_version = 32") - + """Read-only by design (automagic-invariants, polylogue-gd6v/4jsk): daemon + convergence already reconciles this backlog automatically, so a manual + apply path would be a redundant, doctrine-forbidden break-glass surface. + """ result = cli_runner.invoke( cli, ["--plain", "ops", "maintenance", "embedding-orphan-reconcile", "--yes"], ) - assert result.exit_code != 0 - assert isinstance(result.exception, RuntimeError) - assert "active index is v32, packaged index is v" in str(result.exception) - with sqlite3.connect(cli_workspace["archive_root"] / "embeddings.db") as conn: - assert ( - conn.execute("SELECT COUNT(*) FROM message_embedding_refs WHERE message_id = ?", (message_id,)).fetchone()[ - 0 - ] - == 1 - ) + # Click's CliRunner surfaces a rejected/unknown option as SystemExit(2) + # (its own UsageError is caught and converted before invoke() returns), + # so exit code 2 plus the option name in the rejection message is the + # stable, public contract to assert on -- not Click's internal + # exception wording, which isn't guaranteed across versions. + assert result.exit_code == 2 + assert "--yes" in result.output def test_archive_init_cli_is_dry_run_without_yes(cli_workspace: dict[str, Path], cli_runner: CliRunner) -> None: diff --git a/tests/unit/daemon/test_blob_gc_periodic.py b/tests/unit/daemon/test_blob_gc_periodic.py new file mode 100644 index 0000000000..6a8dc8491d --- /dev/null +++ b/tests/unit/daemon/test_blob_gc_periodic.py @@ -0,0 +1,122 @@ +"""Daemon-side wiring for the periodic blob-GC drain (automagic-invariants gap). + +``polylogue ops maintenance blob-gc`` was, before this module existed, the +*only* path that reclaimed unreferenced blobs — every other mechanical, +non-judgment maintenance operation (FTS merge, WAL checkpoint, embedding +orphan reconcile) already has a daemon-owned periodic equivalent. These +tests exercise ``run_blob_gc_once`` — the bounded sync helper the periodic +daemon loop (``periodic_blob_gc_check``) invokes through the write +coordinator — against real on-disk blob-store fixtures, and prove the +production route (``DaemonWriteCoordinator.run_sync``) actually performs +the reclaim rather than a bypassed direct call. +""" + +from __future__ import annotations + +import asyncio +import os +import sqlite3 +from pathlib import Path + +from polylogue.daemon.blob_gc_periodic import run_blob_gc_once +from polylogue.daemon.write_coordinator import DaemonWriteCoordinator +from polylogue.storage.blob_store import BlobStore + + +def _make_source_db(path: Path) -> None: + conn = sqlite3.connect(str(path)) + try: + conn.execute( + """CREATE TABLE raw_sessions ( + raw_id TEXT PRIMARY KEY, + blob_hash BLOB NOT NULL, + blob_size INTEGER NOT NULL DEFAULT 0, + acquired_at TEXT NOT NULL DEFAULT '' + )""" + ) + conn.execute( + """CREATE TABLE blob_refs ( + blob_hash BLOB NOT NULL CHECK(length(blob_hash) = 32), + ref_id TEXT NOT NULL, + ref_type TEXT NOT NULL CHECK(ref_type IN ('raw_payload', 'attachment', 'sidecar')), + source_path TEXT, + size_bytes INTEGER NOT NULL DEFAULT 0 CHECK(size_bytes >= 0), + acquired_at_ms INTEGER NOT NULL DEFAULT 0, + PRIMARY KEY (blob_hash, ref_type, ref_id) + )""" + ) + conn.execute( + """CREATE TABLE gc_generations ( + generation_id TEXT PRIMARY KEY, + started_at_ms INTEGER NOT NULL, + completed_at_ms INTEGER, + reclaimed_count INTEGER NOT NULL DEFAULT 0, + reclaimed_bytes INTEGER NOT NULL DEFAULT 0 + )""" + ) + conn.commit() + finally: + conn.close() + + +def _backdate(blob_store: BlobStore, blob_hash: str) -> None: + """Backdate a blob's mtime well past MIN_AGE_S (60s) so it is GC-eligible. + + A fixed epoch (not a `time.time()`-relative offset) avoids any host-clock + read in the test itself -- GC eligibility only needs "old enough", not a + real elapsed duration. + """ + os.utime(blob_store.blob_path(blob_hash), (0, 0)) + + +def test_run_blob_gc_once_returns_none_when_blob_dir_absent(tmp_path: Path) -> None: + db_path = tmp_path / "source.db" + _make_source_db(db_path) + result = run_blob_gc_once(db_path, tmp_path / "blob") + assert result is None + + +def test_run_blob_gc_once_returns_none_when_source_db_absent(tmp_path: Path) -> None: + blob_dir = tmp_path / "blob" + blob_dir.mkdir() + result = run_blob_gc_once(tmp_path / "source.db", blob_dir) + assert result is None + + +def test_run_blob_gc_once_reclaims_unreferenced_aged_blob(tmp_path: Path) -> None: + db_path = tmp_path / "source.db" + _make_source_db(db_path) + blob_dir = tmp_path / "blob" + blob_store = BlobStore(blob_dir) + + blob_hash, _ = blob_store.write_from_bytes(b"orphan blob for daemon gc") + _backdate(blob_store, blob_hash) + + result = run_blob_gc_once(db_path, blob_dir) + + assert result is not None + assert result.deleted_count == 1 + assert not blob_store.blob_path(blob_hash).exists() + + +def test_daemon_coordinator_owns_real_blob_gc_mutation(tmp_path: Path) -> None: + """Production-route proof; bypassing run_sync or its authority token makes this fail.""" + db_path = tmp_path / "source.db" + _make_source_db(db_path) + blob_dir = tmp_path / "blob" + blob_store = BlobStore(blob_dir) + + blob_hash, _ = blob_store.write_from_bytes(b"orphan blob via coordinator") + _backdate(blob_store, blob_hash) + + coordinator = DaemonWriteCoordinator() + + async def run() -> object: + return await coordinator.run_sync("maintenance.blob_gc", run_blob_gc_once, db_path, blob_dir) + + result = asyncio.run(run()) + + assert result is not None + assert result.deleted_count == 1 # type: ignore[attr-defined] + assert coordinator.snapshot().active_actor is None + assert not blob_store.blob_path(blob_hash).exists()