test(storage): extend backend parity coverage for the shared-session refactor

.github/workflows/metadata-session-boundary.yml

@@ -0,0 +1,66 @@
+name: Metadata session contracts
+on:
+  push:
+    branches: [fix/metadata-session-boundary-20260929]
+  workflow_dispatch:
+permissions:
+  contents: read
+concurrency:
+  group: metadata-session-${{ github.ref }}
+  cancel-in-progress: false
+jobs:
+  contracts:
+    runs-on: ubuntu-latest
+    timeout-minutes: 20
+    services:
+      postgres:
+        image: postgres:16
+        env:
+          POSTGRES_USER: msgtest
+          POSTGRES_PASSWORD: msgtest
+          POSTGRES_DB: postgres
+        ports: ['5432:5432']
+        options: >-
+          --health-cmd "pg_isready -U msgtest"
+          --health-interval 10s --health-timeout 5s --health-retries 5
+    env:
+      MSG_TEST_POSTGRES_URL_TEMPLATE: postgresql://msgtest:msgtest@127.0.0.1:5432/{database}
+    steps:
+      - uses: actions/checkout@v7
+        with:
+          persist-credentials: false
+      - uses: actions/setup-python@v7
+        with:
+          python-version: '3.15'
+          allow-prereleases: true
+          cache: pip
+      - name: Install isolated test dependencies
+        run: |
+          sudo apt-get update -qq
+          sudo apt-get install -y -qq age
+          python -m pip install -e '.[dev]'
+      - name: Record exact source
+        env:
+          SOURCE_SHA: ${{ github.sha }}
+        run: |
+          mkdir -p artifacts
+          test "$(git rev-parse HEAD)" = "$SOURCE_SHA"
+          python - <<'PY'
+          import json, os, subprocess
+          from pathlib import Path
+          git=lambda *args: subprocess.check_output(['git',*args],text=True).strip()
+          Path('artifacts/source.json').write_text(json.dumps({'commit':git('rev-parse','HEAD'),
+              'tree':git('rev-parse','HEAD^{tree}'),'event_sha':os.environ['SOURCE_SHA'],
+              'dirty':bool(git('status','--porcelain'))},indent=2)+'\n')
+          PY
+      - name: Run the same contract on both real database adapters
+        run: python -m pytest tests/test_metadata_session_boundary.py --junitxml=artifacts/metadata-session.xml
+      - name: Preserve PostgreSQL locks, migration and recovery regressions
+        run: |
+          python -m pytest tests/test_postgres.py tests/test_security_storage_rollback.py tests/test_security_rollback_fences.py tests/test_ledger_accounts.py tests/test_ledger_account_isolation.py tests/test_ledger_migration_json.py tests/test_ledger_migration_references.py tests/test_recovery_safety_units.py tests/test_recovery_metadata.py --junitxml=artifacts/metadata-regressions.xml
+      - uses: actions/upload-artifact@v4
+        if: always()
+        with:
+          name: metadata-session-${{ github.sha }}
+          path: artifacts/
+          if-no-files-found: error

docs/METADATA_SESSION_PROGRESS.md

@@ -0,0 +1,67 @@
+# Metadata session boundary progress
+
+## Current integration decision
+
+Issue #157. PR #166 is now supplemental backend-parity coverage stacked on #167,
+not a competing source implementation. Coordination: #83 comment5878756363.
+The canonical source is #167 commit `a85b74456be8eb20977ddd144b94483b0b21202f`,
+tree `042422943d4f5a9c43e5d531797dcc3d9c92a787`.
+
+That implementation includes the explicit QuerySession/QueryResult consumer port,
+transaction-lifetime result guards and missing declared MetadataSession domain
+methods. Reusing it better covers the issue's declaration/actual-dependency
+acceptance condition than maintaining this branch's parallel base extraction.
+All #167 source files are adopted unchanged. #166 adds only this note, its focused
+workflow and the same 16 independent parity tests. The four class-name references
+change from MetadataSessionBase to the canonical RelationalSession; no behavior
+assertion, test, skip condition or existing upstream source is removed or weakened.
+Both histories are preserved with a non-force two-parent commit.
+
+No #167 branch, #169 executor branch, #155 recovery branch or main is overwritten.
+The #167/#169 executor overlap must still converge to one implementation before
+main integration. #165 has meanwhile merged to main `08c58928ee3f6c6cb350d9e9fba75d072bd78f08`;
+old main/branch green results do not certify the new combined tree.
+
+## Original red evidence, preserved
+
+Original base main `476e64f04ed1d9c7dcab4a41c4f3cdb72faad6a7`, tree
+`40ca207a3c49be347120e058e47ed328ec4e3098`.
+Regression commit `7f8752e37fe5b5a8021f192b87e731caec2bbc2d`, tree
+`686dbddc1bdfc52d97c127bdb7338f002cabf735`.
+Run36481798533: **4 failures, 12 passes, 0 errors/skips**; all failures were
+new architecture assertions, while six behaviors passed on both actual adapters.
+Artifact10997260030 SHA-256 independently verified:
+`50ae1ff1b71684d60e0ff133cde41c6cb81658c0855aeb9c376fff110dc27e2e`.
+
+## Original implementation audit, preserved but superseded
+
+The original source assembly run36482780156 verified 42 shared methods, both
+complete Store classes plus FakeMetadataStore, driver execute/enqueue, PostgreSQL
+SQL translation and both schemas for unchanged ASTs. Output5474eaf475e7f844244413b1f3e5109d4add4dce.
+Artifact10998045352 hash:
+`521d6a2f3c5d0d6b8e701302f671fad8b7d3b2641ac346b3ce4c48d282525823`.
+One-shot tooling was removed from the final original tree.
+
+Original implementation head `0ef416ec99e20604de2398622e72d7f479a84ceb`, tree
+`ad1d3f4a321c25c0c3d4b6c1be8421824809e54f`.
+Focused run36483460243 passed **16 new + 80 existing = 96 tests**, with
+**0 failures, errors or skips**. Actual source.json and both JUnit files were read.
+Artifact10997317418 SHA-256 independently verified:
+`8a9cc70e9c31d19baf18050f42eda11818786d671863ded05d35b5e9ab098354`.
+These results apply to that original head only, not the adopted #167 source.
+
+## Coverage and next gate
+
+The same tests verify isolated PostgreSQL imports, neutral-base ownership,
+MetadataSession method declarations, both adapters' resource generations/path/
+paging, result/event/audit/job round trips, idempotency conflicts, nested savepoint
+compensation, read-only/closed/cross-task guards, unique-constraint rollback and
+question-mark/percent bound-parameter handling.
+
+The focused workflow also retains PostgreSQL locks and concurrency, security
+rollback fences, ledger migrations and recovery metadata/unit suites. All project
+execution stays in GitHub Actions. Final combined-head focused/full four-shard,
+exact node-ID union/disjointness, conformance/build and Recovery safety outcomes
+must be inspected and recorded in #166/#83 before merge; no outcome is predicted.
+No production deployment, migration, Root/PIN, real funds/outbound effects, backup
+destruction or recovery promotion. #65/#69/#85 umbrella acceptance remains separate.

tests/test_metadata_session_boundary.py

@@ -0,0 +1,186 @@
+"""One storage contract exercised independently against PostgreSQL and SQLite."""
+from __future__ import annotations
+
+import asyncio
+from dataclasses import replace
+from datetime import UTC, datetime
+import importlib
+import inspect
+from pathlib import Path
+import subprocess
+import sys
+
+import pytest
+
+from msg.core.contracts import MetadataSession, MetadataStore
+from msg.core.errors import Failure
+from msg.core.models import AuditEvent, EffectJob, Event, OperationResult, Principal
+from test_postgres import item
+
+
+def test_postgres_import_does_not_load_a_sqlite_adapter_or_driver():
+    code = '''
+import importlib.abc, sys
+class NoSqlite(importlib.abc.MetaPathFinder):
+    def find_spec(self, fullname, path=None, target=None):
+        if fullname == 'msg.storage.sqlite' or fullname == 'sqlite3' or fullname.startswith('sqlite3.'):
+            raise AssertionError('PostgreSQL imported SQLite: ' + fullname)
+sys.meta_path.insert(0, NoSqlite())
+import msg.storage.postgres
+assert 'msg.storage.sqlite' not in sys.modules
+assert 'sqlite3' not in sys.modules
+'''
+    result = subprocess.run([sys.executable, '-c', code], capture_output=True, text=True)
+    assert result.returncode == 0, result.stderr
+
+
+def test_backends_share_only_the_neutral_session_implementation():
+    from msg.storage.postgres import PostgresSession
+    from msg.storage.sqlite import SqliteSession
+    common = importlib.import_module('msg.storage.session').RelationalSession
+    assert not issubclass(PostgresSession, SqliteSession)
+    assert PostgresSession.__bases__ == SqliteSession.__bases__ == (common,)
+    for name in ('resource','children','save_result','append_audit','on_rollback','check'):
+        assert getattr(PostgresSession, name) is getattr(SqliteSession, name) is getattr(common, name)
+    assert PostgresSession.execute is not SqliteSession.execute
+    assert 'execute' in common.__abstractmethods__
+
+
+def test_shared_layer_does_not_own_connections_dialects_schemas_or_commits():
+    module = importlib.import_module('msg.storage.session')
+    text = Path(module.__file__).read_text()
+    for forbidden in ('sqlite3', 'psycopg', '_connection', 'PRAGMA ', 'CREATE TABLE ',
+                      'INSERT OR IGNORE', '.commit(', '.rollback(', 'pg_advisory'):
+        assert forbidden not in text
+    assert set(module.RelationalSession.__abstractmethods__) == {'execute'}
+    assert set(inspect.signature(module.RelationalSession).parameters) == {'write'}
+
+
+def test_shared_implementation_covers_the_declared_metadata_contract():
+    common = importlib.import_module('msg.storage.session').RelationalSession
+    for name, member in vars(MetadataSession).items():
+        if not name.startswith('_') and callable(member):
+            assert name in vars(common), name
+            assert callable(getattr(common, name)), name
+    assert 'transaction' in vars(MetadataStore)
+    assert {'one','rows','check','setting','set_setting','job','save_job',
+            'organization','path','ancestors','run_rollback_effects'} <= set(vars(common))
+
+
+@pytest.fixture(params=['postgres','sqlite'])
+async def metadata(request, pg_dsn, tmp_path):
+    from msg.storage.postgres import PostgresMetadataStore
+    from msg.storage.sqlite import SqliteMetadataStore
+    store = (PostgresMetadataStore(pg_dsn) if request.param == 'postgres' else
+             SqliteMetadataStore(tmp_path/'metadata.sqlite'))
+    try:
+        yield store
+    finally:
+        await store.close()
+
+
+async def test_shared_resource_generation_path_and_paging(metadata):
+    async with metadata.transaction(write=True) as tx:
+        await tx.insert(item())
+        await tx.insert(item('a','root'))
+        await tx.insert(item('b','root'))
+        assert await tx.resolve('/a') == 'a'
+        assert await tx.path('a') == '/a'
+        assert [r.id for r in await tx.ancestors('a')] == ['root']
+        first = await tx.children('root', limit=1)
+        assert [r.id for r in first.items] == ['a'] and first.next_cursor == 'a'
+        assert [r.id for r in (await tx.children('root', cursor=first.next_cursor)).items] == ['b']
+        old = await tx.resource('a')
+        await tx.replace(replace(old, generation=1, mode=0o700), 0)
+        assert tx.setting('authorization_epoch') == 1
+        with pytest.raises(Failure, match='generation_conflict'):
+            await tx.replace(replace(old, generation=1), 0)
+    async with metadata.transaction(write=False) as tx:
+        assert (await tx.resource('a')).generation == 1
+        assert (await tx.resource('a')).mode == 0o700
+
+
+async def test_shared_result_event_audit_and_job_roundtrips(metadata):
+    now = datetime.now(UTC)
+    principal = Principal(actor='owner', subject='owner', credential_id=None,
+                          method='local', certificates=(), ceiling=())
+    result = OperationResult(request_id='once', operation='test', status='ok', actor=None, subject=None)
+    event = Event(id='event-one', type='test', time=now, request_id='once',
+                  actor='owner', subject='owner', resources=(), data={})
+    audit = AuditEvent(event=event, authority=(), before_digest=None, after_digest=None,
+                       previous_digest=None, entry_digest='', result='ok')
+    job = EffectJob(id='job-one', event_id=event.id, kind='mail', dedupe_key='unique-job',
+                    principal=principal, operation='mail.send', arguments={}, state='pending',
+                    attempts=0, next_attempt_at=now, lease_until=None)
+    async with metadata.transaction(write=True) as tx:
+        await tx.save_result('owner','digest-one',result)
+        await tx.append_event(event)
+        await tx.append_audit(audit)
+        await tx.append_audit(replace(audit, event=replace(event,id='event-two')))
+        await tx.enqueue(job)
+    async with metadata.transaction(write=False) as tx:
+        assert await tx.request_result('owner','once','digest-one') == result
+        with pytest.raises(Failure, match='idempotency_conflict'):
+            await tx.request_result('owner','once','another-digest')
+        rows = tx.rows('SELECT digest,previous FROM audit ORDER BY seq')
+        assert rows[0][1] is None and rows[1][1] == rows[0][0]
+        assert await tx.job(job.id) == job
+        assert tx.one('SELECT id FROM events WHERE id=?',(event.id,))[0] == event.id
+
+
+async def test_shared_savepoints_and_compensation_keep_their_scope(metadata):
+    effects = []
+    async def compensate(value):
+        effects.append(value)
+    async with metadata.transaction(write=True) as outer:
+        outer.set_setting('outer', True)
+        outer.on_rollback(lambda: compensate('outer'))
+        with pytest.raises(ValueError, match='inner'):
+            async with metadata.transaction(write=True) as inner:
+                assert inner is outer
+                inner.set_setting('inner', True)
+                inner.on_rollback(lambda: compensate('inner'))
+                raise ValueError('inner')
+        assert effects == ['inner'] and outer.setting('inner') is None
+    assert effects == ['inner']
+    with pytest.raises(ValueError, match='whole'):
+        async with metadata.transaction(write=True) as tx:
+            tx.set_setting('whole', True)
+            tx.on_rollback(lambda: compensate('whole'))
+            raise ValueError('whole')
+    async with metadata.transaction(write=False) as tx:
+        assert tx.setting('outer') is True and tx.setting('whole') is None
+    assert effects == ['inner','whole']
+
+
+async def test_shared_read_only_closed_and_cross_task_guards(metadata):
+    async with metadata.transaction(write=False) as tx:
+        with pytest.raises(Failure, match='read_only_transaction'):
+            tx.set_setting('forbidden', True)
+        with pytest.raises(Failure, match='read_only_transaction'):
+            async with metadata.transaction(write=True):
+                pytest.fail('read-only transaction was upgraded')
+        async def other_task():
+            tx.setting('anything')
+        with pytest.raises(Failure, match='transaction_cross_task'):
+            await asyncio.create_task(other_task())
+    with pytest.raises(Failure, match='transaction_closed'):
+        tx.setting('anything')
+
+
+async def test_each_driver_translates_unique_errors_and_rolls_back(metadata):
+    async with metadata.transaction(write=True) as tx:
+        await tx.insert(item())
+    with pytest.raises(Failure, match='constraint_conflict'):
+        async with metadata.transaction(write=True) as tx:
+            await tx.insert(item('child','root'))
+            await tx.insert(item('child','root'))
+    async with metadata.transaction(write=False) as tx:
+        assert tx.one('SELECT COUNT(*) FROM resources')[0] == 1
+        assert tx.one('SELECT id FROM resources WHERE id=?',('child',)) is None
+
+
+async def test_each_driver_preserves_literal_percent_and_question_mark(metadata):
+    async with metadata.transaction(write=False) as tx:
+        row = tx.one("SELECT '?', ?, '50%'", ('literal?%',))
+        assert tuple(row) == ('?', 'literal?%', '50%')