fix(persistence): replace online SQLite VACUUM with incremental reclamation

- Add migration 0042 to enable incremental vacuum with verified backup and restore
- Enable incremental vacuum before creating tables in new SQLite databases
- Adapt reclamation batches to elapsed time and lock contention
- Cover migration recovery, WAL mode, and concurrent writes with regression tests

ARCHITECTURE.md

@@ -273,6 +273,7 @@ Maintenance rule: when adding, removing, or changing an HTTP endpoint that shoul
 Persistence is centered on `pkg/persistence/mgr.py`.
 
 - SQLite is the default database; PostgreSQL is supported.
+- SQLite files use incremental auto-vacuum. Migration `0042_sqlite_incremental_vacuum` converts existing files before producers start, with a verified backup and restoration on conversion failure when a full rebuild is required. Fresh files enable the format before creating tables. Monitoring cleanup reclaims free pages in independent adaptive batches (1–512 pages, starting at 8, targeting 25 ms), pausing 250 ms between batches and starting no new batches after a two-second round budget. Busy writers take priority. No full `VACUUM` or forced WAL checkpoint runs during online cleanup.
 - Models live under `pkg/entity/persistence/`.
 - `pkg/persistence/media.py::MediaReferenceText` validates bound conversation payloads on ORM/Core writes while preserving the existing TEXT schema and string API. It rejects recognized structured inline images: retained images go through `MediaCache.externalize_chain_dump()` before serialization, while metadata-only snapshots and transcripts use `redact_inline_images()`. Reads and comparisons preserve historical values for incremental cleanup. Runtime/adapter messages, arbitrary diagnostic text, and opaque strings are outside this contract.
 - `pkg/persistence/resource_urls.py` interns structured resource URL fields in retained messages, history and monitoring at the typed SQL boundary, in the source transaction. `MediaReferenceJSON` preserves native JSON monitoring details. Payloads store opaque references to the global `resource_urls` catalog; typed SQL reads and comparisons expand them to current URLs, preserving API shapes and source-record authorization. Configuration URLs and URLs in free text are excluded. Catalog lookups use only the primary-key and URL-hash unique indexes. Successful refreshes overwrite the current URL and persist the check time even when standard `get_msg` returns the same URL. Converging URLs retain only ID redirects, with no old URL. PostgreSQL row locks and SQLite's immediate write transaction serialize concurrent refreshes; SQLite holds its write lock during the upstream call. Bounded in-memory aliases let reused Runner payloads resolve old URLs without retaining those URLs in the database. Migration `0041_resource_urls` normalizes existing payloads using their source creation timestamps before producers start; downgrade expands references before dropping the catalog.

src/langbot/pkg/api/http/service/monitoring.py

@@ -394,10 +394,9 @@ class MonitoringService(ExecutionDetailsMixin):
         else:
             deleted_counts = await delete_records()
 
-        # Receipt expiry alone must not trigger database-wide VACUUM work.
-        if any(count for table, count in deleted_counts.items() if table != 'monitoring_session_receipts'):
-            await self._release_sqlite_space()
+        from ....persistence.sqlite_reclamation import reclaim_free_pages
 
+        await reclaim_free_pages(self.ap.persistence_mgr.get_db_engine())
         return deleted_counts
 
     async def _delete_expired_in_batches(
@@ -503,16 +502,6 @@ class MonitoringService(ExecutionDetailsMixin):
                 break
         return deleted
 
-    async def _release_sqlite_space(self) -> None:
-        database_type = self.ap.instance_config.data.get('database', {}).get('use', 'sqlite')
-        if database_type != 'sqlite':
-            return
-
-        async with self.ap.persistence_mgr.get_db_engine().connect() as conn:
-            autocommit_conn = await conn.execution_options(isolation_level='AUTOCOMMIT')
-            await autocommit_conn.execute(sqlalchemy.text('PRAGMA wal_checkpoint(TRUNCATE)'))
-            await autocommit_conn.execute(sqlalchemy.text('VACUUM'))
-
     def _serialize_tool_payload(self, payload: object, max_length: int = 20000) -> str | None:
         """Serialize tool arguments/results for monitoring storage."""
         if payload is None:

src/langbot/pkg/persistence/alembic/versions/0042_sqlite_incremental_vacuum.py

@@ -0,0 +1,28 @@
+"""Enable bounded online SQLite space reclamation."""
+
+from alembic import op
+
+revision = '0042_sqlite_incremental_vacuum'
+down_revision = '0041_resource_urls'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    connection = op.get_bind()
+    if connection.dialect.name != 'sqlite':
+        return
+    with op.get_context().autocommit_block():
+        mode = connection.exec_driver_sql('PRAGMA auto_vacuum').scalar_one()
+        if mode != 2:
+            connection.exec_driver_sql('PRAGMA auto_vacuum=INCREMENTAL')
+            if mode == 0:
+                connection.exec_driver_sql('VACUUM')
+        if connection.exec_driver_sql('PRAGMA auto_vacuum').scalar_one() != 2:
+            raise RuntimeError('SQLite incremental vacuum conversion did not complete')
+
+
+def downgrade():
+    # The physical format remains compatible with earlier releases. Do not
+    # rebuild the database merely to undo a space reclamation preference.
+    pass

src/langbot/pkg/persistence/alembic_runner.py

@@ -70,6 +70,13 @@ def get_alembic_head() -> str:
     return head
 
 
+def pending_revisions(current: str, target: str) -> set[str]:
+    """Resolve migration boundaries without relying on revision name ordering."""
+    cfg = Config()
+    cfg.set_main_option('script_location', _ALEMBIC_DIR)
+    return {step.revision for step in ScriptDirectory.from_config(cfg).iterate_revisions(target, current)}
+
+
 def _do_autogenerate(connection: Connection, message: str = 'auto migration') -> None:
     """Synchronous autogenerate — runs inside run_sync."""
     cfg = _build_config(connection)

src/langbot/pkg/persistence/databases/sqlite.py

@@ -17,5 +17,7 @@ class SQLiteDatabaseManager(database.BaseDatabaseManager):
         # Reader transactions must not block independent plugin callback commits.
         # Configure the persistent journal before serving requests, not per write.
         async with self.engine.connect() as connection:
+            # Fresh files can adopt the format without rebuilding existing data.
+            await connection.execute(sqlalchemy.text('PRAGMA auto_vacuum=INCREMENTAL'))
             await connection.execute(sqlalchemy.text('PRAGMA journal_mode=WAL'))
             await connection.commit()

src/langbot/pkg/persistence/mgr.py

@@ -1715,9 +1715,23 @@ class PersistenceManager:
                         target_revision=_RESOURCE_SCOPE_ALEMBIC_REVISION,
                     )
 
-            # PostgreSQL has transactional DDL. SQLite has already crossed the
-            # two destructive tenancy boundaries under verified backups; this
-            # final call is a no-op today and applies future migrations.
+            if engine.dialect.name == 'sqlite':
+                current_rev = await alembic_runner.get_alembic_current(engine)
+                conversion = '0042_sqlite_incremental_vacuum'
+                if conversion in alembic_runner.pending_revisions(current_rev, target_revision):
+                    await alembic_runner.run_alembic_upgrade(engine, '0041_resource_urls')
+                    async with engine.connect() as connection:
+                        vacuum_mode = (await connection.exec_driver_sql('PRAGMA auto_vacuum')).scalar_one()
+                    if vacuum_mode == 0:
+                        await self._run_verified_sqlite_migration(
+                            engine,
+                            source_revision='0041_resource_urls',
+                            target_revision=conversion,
+                        )
+                    else:
+                        await alembic_runner.run_alembic_upgrade(engine, conversion)
+
+            # Apply any remaining release migrations after verified boundaries.
             await alembic_runner.run_alembic_upgrade(engine, target_revision)
             self.ap.logger.info(f'Alembic migrations completed at {target_revision}.')
         except Exception as e:

src/langbot/pkg/persistence/sqlite_reclamation.py

@@ -0,0 +1,75 @@
+"""Reclaim free pages without rebuilding a live SQLite database."""
+
+import asyncio
+import sqlite3
+import time
+from weakref import WeakKeyDictionary
+
+
+_states = WeakKeyDictionary()
+MIN_PAGES = 1
+MAX_PAGES = 512
+TARGET_SECONDS = 0.025
+ROUND_SECONDS = 2.0
+PAUSE_SECONDS = 0.25
+
+
+class _State:
+    def __init__(self):
+        self.lock = asyncio.Lock()
+        self.pages = 8
+
+
+async def reclaim_free_pages(engine):
+    """Adapt batch size to measured latency; yield immediately to other writers.
+
+    The round budget stops new batches, but cannot bound an in-flight SQLite
+    operation. Each completed statement releases its write transaction before
+    sleeping. The connection's original busy timeout is restored before pooling.
+    """
+    if engine.dialect.name != 'sqlite':
+        return
+    if engine.url.database in (None, '', ':memory:') or engine.url.query.get('mode') == 'memory':
+        return
+    state = _states.setdefault(engine, _State())
+    if state.lock.locked():
+        return
+    async with state.lock:
+        deadline = time.monotonic() + ROUND_SECONDS
+        async with engine.connect() as connection:
+            connection = await connection.execution_options(isolation_level='AUTOCOMMIT')
+            if (await connection.exec_driver_sql('PRAGMA auto_vacuum')).scalar_one() != 2:
+                raise RuntimeError('SQLite incremental vacuum migration is required')
+            timeout = (await connection.exec_driver_sql('PRAGMA busy_timeout')).scalar_one()
+            await connection.exec_driver_sql('PRAGMA busy_timeout=0')
+            driver = (await connection.get_raw_connection()).driver_connection
+            try:
+                while time.monotonic() < deadline:
+                    if (await connection.exec_driver_sql('PRAGMA freelist_count')).scalar_one() == 0:
+                        break
+                    started = time.monotonic()
+                    try:
+                        # SQLite advances incremental vacuum while stepping the
+                        # cursor despite exposing no result columns. SQLAlchemy
+                        # closes such results early, so drain the driver cursor.
+                        cursor = await driver.execute(f'PRAGMA incremental_vacuum({state.pages})')
+                        try:
+                            await cursor.fetchall()
+                        finally:
+                            await cursor.close()
+                    except sqlite3.OperationalError as exc:
+                        code = getattr(exc, 'sqlite_errorcode', None)
+                        if code is None or code & 255 not in (sqlite3.SQLITE_BUSY, sqlite3.SQLITE_LOCKED):
+                            raise
+                        state.pages = max(MIN_PAGES, state.pages // 2)
+                        break
+                    elapsed = time.monotonic() - started
+                    if elapsed > TARGET_SECONDS:
+                        state.pages = max(MIN_PAGES, state.pages // 2)
+                    elif elapsed < TARGET_SECONDS / 2:
+                        state.pages = min(MAX_PAGES, state.pages * 2)
+                    if time.monotonic() + PAUSE_SECONDS >= deadline:
+                        break
+                    await asyncio.sleep(PAUSE_SECONDS)
+            finally:
+                await connection.exec_driver_sql(f'PRAGMA busy_timeout={timeout}')

tests/integration/persistence/test_sqlite_migration_backup.py

@@ -80,12 +80,13 @@ async def test_tenancy_migrations_retain_verified_boundary_backups(tmp_path):
 
         assert await alembic_runner.get_alembic_current(engine) == alembic_runner.get_alembic_head()
         payloads = _manifest_payloads(tmp_path / 'migration-backups')
-        assert len(payloads) == 2
+        assert len(payloads) == 3
         assert {
             (payload['source_revision'], payload['target_revision'], payload['status']) for payload in payloads
         } == {
             ('0008_mcp_resource_prefs', '0009_workspace_tenancy', 'migration_succeeded'),
             ('0009_workspace_tenancy', '0010_scope_resources', 'migration_succeeded'),
+            ('0041_resource_urls', '0042_sqlite_incremental_vacuum', 'migration_succeeded'),
         }
         for payload in payloads:
             _assert_verified_backup(payload)
@@ -198,6 +199,6 @@ async def test_backup_retries_transient_reopen_failure_after_replace(tmp_path, m
 
         assert transient_failures == 1
         assert await alembic_runner.get_alembic_current(engine) == alembic_runner.get_alembic_head()
-        assert len(_manifest_payloads(tmp_path / 'migration-backups')) == 2
+        assert len(_manifest_payloads(tmp_path / 'migration-backups')) == 3
     finally:
         await engine.dispose()

tests/integration/persistence/test_sqlite_reclamation.py

@@ -0,0 +1,111 @@
+"""Exercise physical format conversion and bounded reclamation on real files."""
+
+import pytest
+import sqlalchemy as sa
+from sqlalchemy.ext.asyncio import create_async_engine
+
+from langbot.pkg.persistence import alembic_runner, sqlite_reclamation
+from tests.integration.persistence.test_sqlite_migration_backup import _manager
+
+pytestmark = [pytest.mark.integration, pytest.mark.asyncio]
+
+
+@pytest.mark.parametrize('mode', [0, 1, 2])
+async def test_conversion_preserves_data_and_is_repeatable(tmp_path, mode):
+    engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "convert.db"}')
+    try:
+        async with engine.begin() as connection:
+            await connection.exec_driver_sql(f'PRAGMA auto_vacuum={mode}')
+            await connection.exec_driver_sql('CREATE TABLE retained (id INTEGER PRIMARY KEY, value TEXT)')
+            await connection.exec_driver_sql("INSERT INTO retained VALUES (1, 'keep')")
+        await alembic_runner.run_alembic_stamp(engine, '0041_resource_urls')
+        for _ in range(2):
+            await _manager(engine)._run_alembic_migrations()
+        async with engine.connect() as connection:
+            assert (await connection.exec_driver_sql('PRAGMA auto_vacuum')).scalar_one() == 2
+            assert (await connection.exec_driver_sql('SELECT value FROM retained')).scalar_one() == 'keep'
+            assert (await connection.exec_driver_sql('PRAGMA quick_check')).scalar_one() == 'ok'
+        backups = list((tmp_path / 'migration-backups').glob('*.json'))
+        assert len(backups) == (1 if mode == 0 else 0)
+    finally:
+        await engine.dispose()
+
+
+async def test_conversion_failure_restores_original_format_and_revision(tmp_path):
+    engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "failure.db"}')
+
+    @sa.event.listens_for(engine.sync_engine, 'after_cursor_execute')
+    def fail_after_conversion(connection, cursor, statement, parameters, context, executemany):
+        if statement == 'VACUUM':
+            raise RuntimeError('injected conversion failure')
+
+    try:
+        async with engine.begin() as connection:
+            await connection.exec_driver_sql('CREATE TABLE retained (value TEXT)')
+            await connection.exec_driver_sql("INSERT INTO retained VALUES ('keep')")
+        await alembic_runner.run_alembic_stamp(engine, '0041_resource_urls')
+        with pytest.raises(RuntimeError, match='injected conversion failure'):
+            await _manager(engine)._run_alembic_migrations()
+        assert await alembic_runner.get_alembic_current(engine) == '0041_resource_urls'
+        async with engine.connect() as connection:
+            assert (await connection.exec_driver_sql('PRAGMA auto_vacuum')).scalar_one() == 0
+            assert (await connection.exec_driver_sql('SELECT value FROM retained')).scalar_one() == 'keep'
+    finally:
+        await engine.dispose()
+
+
+@pytest.mark.parametrize('slow', [False, True])
+@pytest.mark.parametrize('journal', ['DELETE', 'WAL'])
+async def test_reclamation_shrinks_file_and_adapts_without_full_vacuum(tmp_path, monkeypatch, slow, journal):
+    engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "reclaim.db"}')
+    statements = []
+
+    @sa.event.listens_for(engine.sync_engine, 'before_cursor_execute')
+    def capture(connection, cursor, statement, parameters, context, executemany):
+        statements.append(statement)
+
+    try:
+        async with engine.begin() as connection:
+            await connection.exec_driver_sql('PRAGMA auto_vacuum=INCREMENTAL')
+            await connection.exec_driver_sql(f'PRAGMA journal_mode={journal}')
+            await connection.exec_driver_sql('CREATE TABLE payloads (id INTEGER PRIMARY KEY, data BLOB)')
+            await connection.exec_driver_sql('INSERT INTO payloads(data) VALUES (randomblob(4194304))')
+            await connection.exec_driver_sql('DELETE FROM payloads')
+        async with engine.connect() as connection:
+            before = (await connection.exec_driver_sql('PRAGMA page_count')).scalar_one()
+            timeout = (await connection.exec_driver_sql('PRAGMA busy_timeout')).scalar_one()
+            driver = (await connection.get_raw_connection()).driver_connection
+            await driver.set_trace_callback(statements.append)
+        monkeypatch.setattr(sqlite_reclamation, 'ROUND_SECONDS', 0.1)
+        monkeypatch.setattr(sqlite_reclamation, 'PAUSE_SECONDS', 0.001)
+        monkeypatch.setattr(sqlite_reclamation, 'TARGET_SECONDS', 0 if slow else 100)
+        await sqlite_reclamation.reclaim_free_pages(engine)
+        batches = [int(s.split('(')[1].split(')')[0]) for s in statements if 'incremental_vacuum(' in s]
+        assert len(batches) > 1
+        assert max(batches) <= 512 and min(batches) >= 1
+        assert batches[1] < batches[0] if slow else batches[1] > batches[0]
+        assert 'VACUUM' not in statements
+        async with engine.connect() as connection:
+            assert (await connection.exec_driver_sql('PRAGMA page_count')).scalar_one() < before
+            assert (await connection.exec_driver_sql('PRAGMA busy_timeout')).scalar_one() == timeout
+            await connection.exec_driver_sql('INSERT INTO payloads(data) VALUES (NULL)')
+            await connection.commit()
+    finally:
+        await engine.dispose()
+
+
+async def test_reclamation_yields_to_existing_writer(tmp_path):
+    engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "busy.db"}')
+    try:
+        async with engine.begin() as connection:
+            await connection.exec_driver_sql('PRAGMA auto_vacuum=INCREMENTAL')
+            await connection.exec_driver_sql('CREATE TABLE payloads (data BLOB)')
+            await connection.exec_driver_sql('INSERT INTO payloads VALUES (randomblob(1048576))')
+            await connection.exec_driver_sql('DELETE FROM payloads')
+        async with engine.connect() as writer:
+            await writer.exec_driver_sql('BEGIN IMMEDIATE')
+            await sqlite_reclamation.reclaim_free_pages(engine)
+            assert sqlite_reclamation._states[engine].pages == 4
+            await writer.rollback()
+    finally:
+        await engine.dispose()

tests/unit_tests/api/service/test_monitoring_sessions.py

@@ -1814,7 +1814,7 @@ async def test_historical_ingress_replay_after_initialization_does_not_increment
 
 
 @pytest.mark.contract('summary_retention')
-async def test_active_session_receipts_expire_in_bounded_tenant_scoped_batches(service, monkeypatch):
+async def test_active_session_receipts_expire_in_bounded_tenant_scoped_batches(service):
     now = dt.datetime.now(dt.timezone.utc).replace(tzinfo=None)
     for index in range(5):
         await service.record_message(
@@ -1838,10 +1838,6 @@ async def test_active_session_receipts_expire_in_bounded_tenant_scoped_batches(s
             )
         )
 
-    async def forbid_vacuum():
-        pytest.fail('Receipt expiry must not trigger database-wide VACUUM')
-
-    monkeypatch.setattr(service, '_release_sqlite_space', forbid_vacuum)
     deleted = await service.cleanup_expired_records(context(), retention_days=1, batch_size=2, max_batches_per_table=1)
     assert deleted['monitoring_session_receipts'] == 2 and sum(deleted.values()) == 2
     for expected in [2, 1, 0]:

tests/unit_tests/api/service/test_monitoring_tenancy.py

@@ -69,6 +69,7 @@ class _PersistenceManager:
 async def service(tmp_path):
     engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "monitoring.db"}')
     async with engine.begin() as connection:
+        await connection.exec_driver_sql('PRAGMA auto_vacuum=INCREMENTAL')
         await connection.run_sync(Base.metadata.create_all)
         await connection.execute(
             sqlalchemy.insert(Workspace),
@@ -349,7 +350,7 @@ async def test_token_statistics_aggregate_and_limit_groups_in_database(service):
     ]
 
 
-async def test_cleanup_commits_sqlite_delete_before_vacuum(tmp_path):
+async def test_cleanup_deletes_bounded_batches_without_exclusive_sqlite_maintenance(tmp_path):
     engine = create_async_engine(
         f'sqlite+aiosqlite:///{tmp_path / "monitoring-cleanup.db"}',
         connect_args={'timeout': 0.1},
@@ -362,6 +363,7 @@ async def test_cleanup_commits_sqlite_delete_before_vacuum(tmp_path):
     application.persistence_mgr = manager
     try:
         async with engine.begin() as connection:
+            await connection.exec_driver_sql('PRAGMA auto_vacuum=INCREMENTAL')
             await connection.run_sync(Base.metadata.create_all)
             await connection.execute(
                 sqlalchemy.insert(Workspace).values(
@@ -393,6 +395,12 @@ async def test_cleanup_commits_sqlite_delete_before_vacuum(tmp_path):
                 ],
             )
 
+        statements = []
+
+        @sqlalchemy.event.listens_for(engine.sync_engine, 'before_cursor_execute')
+        def capture_statement(connection, cursor, statement, parameters, context, executemany):
+            statements.append(statement.strip().upper())
+
         deleted = await MonitoringService(application).cleanup_expired_records(
             _context(WORKSPACE_A),
             retention_days=1,
@@ -406,6 +414,7 @@ async def test_cleanup_commits_sqlite_delete_before_vacuum(tmp_path):
                 sqlalchemy.select(sqlalchemy.func.count()).select_from(MonitoringMessage)
             )
         assert remaining == 3
+        assert not any(statement.startswith(('VACUUM', 'PRAGMA WAL_CHECKPOINT')) for statement in statements)
     finally:
         await engine.dispose()