diff --git a/.github/workflows/architecture-ports.yml b/.github/workflows/architecture-ports.yml
new file mode 100644
index 00000000..7782b432
--- /dev/null
+++ b/.github/workflows/architecture-ports.yml
@@ -0,0 +1,82 @@
+name: Architecture boundary regressions
+on:
+  push:
+    branches: [fix/executor-core-20260929-2045]
+  pull_request:
+    paths:
+      - 'src/msg/**'
+      - 'tests/test_architecture_ports.py'
+      - 'tests/test_execution_composition.py'
+      - '.github/workflows/architecture-ports.yml'
+  workflow_dispatch:
+permissions:
+  contents: read
+concurrency:
+  group: architecture-ports-${{ github.event.pull_request.head.repo.full_name || github.repository }}-${{ github.head_ref || github.ref_name }}
+  cancel-in-progress: true
+jobs:
+  focused:
+    runs-on: ubuntu-latest
+    timeout-minutes: 12
+    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 5s --health-timeout 5s --health-retries 10
+      valkey:
+        image: valkey/valkey:9.0.6
+        ports: ['6379:6379']
+        options: >-
+          --health-cmd "valkey-cli ping"
+          --health-interval 5s --health-timeout 5s --health-retries 10
+    env:
+      MSG_TEST_POSTGRES_URL_TEMPLATE: postgresql://msgtest:msgtest@127.0.0.1:5432/{database}
+      MSG_TEST_VALKEY_URL: redis://127.0.0.1:6379/0
+    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 source identity
+        env:
+          EVENT_HEAD: ${{ github.event.pull_request.head.sha || github.sha }}
+        run: |
+          mkdir -p artifacts
+          python - <<'PY'
+          import json, os, platform, subprocess
+          from pathlib import Path
+          def git(ref):
+              return subprocess.check_output(['git', 'rev-parse', ref], text=True).strip()
+          Path('artifacts/source.json').write_text(json.dumps({
+              'head': os.environ['EVENT_HEAD'], 'checkout': git('HEAD'), 'tree': git('HEAD^{tree}'),
+              'run': os.environ['GITHUB_RUN_ID'], 'attempt': os.environ['GITHUB_RUN_ATTEMPT'],
+              'python': platform.python_version(), 'scope': 'isolated-regression-not-production'
+          }, indent=2) + '\n')
+          PY
+      - name: Exercise boundaries and unchanged security contracts
+        run: >-
+          python -m pytest tests/test_architecture_ports.py tests/test_execution_composition.py tests/test_batch.py
+          tests/test_batch_secret_release.py tests/test_tools.py tests/test_webhook_domain_events.py
+          tests/test_read_projection_cache_transports.py tests/test_read_projection_cache_unit.py
+          tests/issue_78_80 --junitxml=artifacts/architecture-ports.xml
+      - uses: actions/upload-artifact@v4
+        if: always()
+        with:
+          name: architecture-ports-evidence
+          path: artifacts/
+          if-no-files-found: error
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index fec5fb8d..22b3ed01 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -6,7 +6,7 @@ on:
 permissions:
   contents: read
 concurrency:
-  group: rewrite-${{ github.ref }}
+  group: rewrite-${{ github.event.pull_request.head.repo.full_name || github.repository }}-${{ github.head_ref || github.ref_name }}
   cancel-in-progress: true
 jobs:
   shards:
diff --git a/.github/workflows/metadata-session-boundary.yml b/.github/workflows/metadata-session-boundary.yml
new file mode 100644
index 00000000..5a3bf086
--- /dev/null
+++ b/.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
diff --git a/.github/workflows/storage-session.yml b/.github/workflows/storage-session.yml
new file mode 100644
index 00000000..b3dd3955
--- /dev/null
+++ b/.github/workflows/storage-session.yml
@@ -0,0 +1,85 @@
+name: Storage session boundaries
+on:
+  push:
+    branches: [fix/storage-session-20260929-2050]
+  pull_request:
+    paths:
+      - 'src/msg/**'
+      - 'tests/test_session_boundary.py'
+      - 'tests/test_metadata_session_boundary.py'
+      - '.github/workflows/storage-session.yml'
+  workflow_dispatch:
+permissions:
+  contents: read
+concurrency:
+  group: storage-session-${{ github.event.pull_request.head.repo.full_name || github.repository }}-${{ github.head_ref || github.ref_name }}
+  cancel-in-progress: true
+jobs:
+  focused:
+    runs-on: ubuntu-latest
+    timeout-minutes: 12
+    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 5s --health-timeout 5s --health-retries 10
+      valkey:
+        image: valkey/valkey:9.0.6
+        ports: ['6379:6379']
+        options: >-
+          --health-cmd "valkey-cli ping"
+          --health-interval 5s --health-timeout 5s --health-retries 10
+    env:
+      MSG_TEST_POSTGRES_URL_TEMPLATE: postgresql://msgtest:msgtest@127.0.0.1:5432/{database}
+      MSG_TEST_VALKEY_URL: redis://127.0.0.1:6379/0
+    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 source identity
+        env:
+          EVENT_HEAD: ${{ github.event.pull_request.head.sha || github.sha }}
+        run: |
+          mkdir -p artifacts
+          python - <<'PY'
+          import json, os, platform, subprocess
+          from pathlib import Path
+          def git(ref):
+              return subprocess.check_output(['git', 'rev-parse', ref], text=True).strip()
+          Path('artifacts/source.json').write_text(json.dumps({
+              'head': os.environ['EVENT_HEAD'], 'checkout': git('HEAD'), 'tree': git('HEAD^{tree}'),
+              'run': os.environ['GITHUB_RUN_ID'], 'attempt': os.environ['GITHUB_RUN_ATTEMPT'],
+              'python': platform.python_version(), 'scope': 'isolated-regression-not-production'
+          }, indent=2) + '\n')
+          PY
+      - name: Exercise boundaries and unchanged security contracts
+        run: >-
+          python -m pytest tests/test_session_boundary.py tests/test_metadata_session_boundary.py tests/test_foundation.py
+          tests/test_postgres.py tests/test_security_rollback_fences.py
+          tests/test_security_storage_rollback.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/storage-session.xml
+      - uses: actions/upload-artifact@v4
+        if: always()
+        with:
+          name: storage-session-evidence
+          path: artifacts/
+          if-no-files-found: error
diff --git a/docs/ISSUE_PROGRESS.md b/docs/ISSUE_PROGRESS.md
new file mode 100644
index 00000000..b4ce7e64
--- /dev/null
+++ b/docs/ISSUE_PROGRESS.md
@@ -0,0 +1,40 @@
+# Issue 推进笔记
+
+用户入口：2026-09-28T20:34Z。PR #167 是本轮 #156/#157 的唯一组合交付入口；基准 main476e64f04ed1d9c7dcab4a41c4f3cdb72faad6a7，tree40ca207a3c49be347120e058e47ed328ec4e3098。最终提交/CI/合并状态以 #167 最新验收评论为准；本文只登记已取得的证据，不用历史绿灯代替新头。
+
+## 已完成的源码切片
+
+#156：core.batching、events、packet、schemas 各有一份中立实现，原导入路径为同对象兼容别名；URL秘密拒绝和gzip仍在传输层。executor没有Application/插件反向引用，只接收固定投影和事务通知接口。HTTP/SSH由Application.new_executor统一装配请求上限、秘密响应和恢复隔离。签名、ID、回放、权限、事务和receipt规则保留。
+
+#157：PG/SQLite独立继承不持有驱动/连接的RelationalSession。SQL、schema、迁移、锁、保存点、补偿和提交信号仍由后端管理。显式QuerySession覆盖真实消费者，MetadataSession补齐path/ancestors/organization/job/save_job；查询结果视图每次读取重验任务/生命周期，不公开connection/commit。没有新增ORM、逐表Repository或任意SQL网络接口。
+
+SQLite仍是精简测试后端。政策消费者测试只在隔离fixture复用生产仲裁表DDL，验证真实已存政策及损坏digest拒绝，不修改生产迁移或声称完整市场后端等价。
+
+## 测试先行证据（均已下载并独立核验）
+
+| 范围 | 测试头 / tree | run / artifact | JUnit |
+|---|---|---|---|
+| #156 | 453d6f858e19136261278c7042c13598035aef1a / ae0b4f5e31dc1844c1abd0142da898b75b7779d4 | 36481310584 / 10997043729 | 7 failed / 158 passed / 0 errors / 0 skipped |
+| #157 | 2daf1e0b81783d8b52685540b041ba9563333282 / e1668ddcbf5879bce9520a31f8bfe911c813370c | 36482022051 / 10998115401 | 8 failed / 35 passed / 0 errors / 0 skipped |
+
+新增用例准确失败，原有选定用例全部通过。ZIP SHA256分别为c2e1978649b6e251e38703befb0fc674fbb336db55fa075113af0edb07d81fb7、fe037acb748babcb05dcdc7c679bb42986f7ae509af86612ce510307c90ed3e3；source.json匹配head/tree/Python3.15.0rc2/attempt1。
+
+实现历史：10f46fd300ef0fa4890ca0ea7aad3d10bca7f7a5（执行器）、4a6b852961b5e9745c1fea8d37117a7a7c1bde58（存储），组合a85b74456be8eb20977ddd144b94483b0b21202f/tree042422943d4f5a9c43e5d531797dcc3d9c92a787。该历史组合存储专项36483874055已success；后续扩展回归头仍须重新验收。
+
+## 重叠PR收敛
+
+已对照#166和#169完整改动及用例，在#83评论5878614115、#166评论5878687666、#169评论5878694971协调，不覆盖其源分支。
+
+保留#166全部16个结构/双后端行为测试，仅将拟议基类名改为唯一RelationalSession；补充#169的独立装配、工厂批处理回放、事件通知失败整事务回滚3个用例。签名/认证/事务和失败不提交断言未降低。#169的missing-decoder配置用例不适用于内置中立codec，不引入第二套decoder/BatchPolicy制造该错误；既有限长、秘密拒绝、投影缺失拒绝均保留。原red证据仍归#166/#169，不冒充本分支新红基线。
+
+## 合并门槛与执行边界
+
+两个专项、完整四分片、精确node-ID全集/互斥、conformance/实际tokenizer、wheel/sdist、Recovery safety均保留。同一来源仓库分支的push/PR共享并发组以减少重复runner，不删测试、放宽断言或降低合并门槛。最终组合通过并合并后才关闭两个已修复issue和重复PR。
+
+项目执行全部在GitHub Actions；本地仅源文本/Git差异与树检查、下载证据读取。临时精确树发布/源码导出仅在辅助分支，不进入产品树。42个共享方法主体、两后端Store/schema/迁移的静态对照未发现改动，云端运行仍是验收依据。
+
+## 剩余与并行工作
+
+#165工具、#168市场、#158客户端及#155恢复由独立分支推进，不覆盖。#155已知0e4dbe7头完整36480031758及专项36480031762成功，但仍须当前main组合与本机CLI/文件边界验收。#68–83/#85各功能总体验收不能因这两个架构修复整体关闭。
+
+#64/#65/#84仍缺授权生产代理/日志、真实旧库快照、容量与恢复材料；#68–70独立新鲜检查点、备份退役、真实控制台及promotion须分别验收，CI不能替代。此次不部署、不操作真实Root/PIN、资金、外发、备份销毁或恢复放行。
diff --git a/docs/METADATA_SESSION_PROGRESS.md b/docs/METADATA_SESSION_PROGRESS.md
new file mode 100644
index 00000000..81bac604
--- /dev/null
+++ b/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.
diff --git a/src/msg/application.py b/src/msg/application.py
index 591920d2..1862a579 100644
--- a/src/msg/application.py
+++ b/src/msg/application.py
@@ -119,15 +119,35 @@ async def load(self):
         self.authenticator=AuthenticationService(self.registry,self.certificates,self.settings.service_url,self.clock,
             self.primary_ceiling,self.temporary_ceiling)
         self.authorizer=AuthorizationService(self.registry,self.certificates)
-        self.executor=OperationExecutor(self.registry,self.metadata,self.contents,self.authenticator,self.authorizer,
-                                       self.clock,self.receipt_signer)
-        self.executor.application=self
+        self.executor=self.new_executor(self.authenticator)
         self.executor.recovery_drill_marker=marker
         self.executor.recovery_quarantined=quarantined
-        self.executor.response_hook=self._secrets_for_caller
         self._loaded=True
         return self
 
+    def new_executor(self, authenticator=None):
+        """Compose identical limits/ports for HTTP and the restricted SSH entry."""
+        executor=OperationExecutor(self.registry,self.metadata,self.contents,
+            self.authenticator if authenticator is None else authenticator,
+            self.authorizer,self.clock,self.receipt_signer,
+            max_request_bytes=self.settings.server.limits.max_request_bytes,
+            result_projection=self._result_projection,
+            event_notifications=self._event_notifications)
+        executor.response_hook=self._secrets_for_caller
+        executor.recovery_drill_marker=self.settings.config_dir/'recovery-drill.json'
+        if self.executor is not None:
+            executor.recovery_quarantined=self.executor.recovery_quarantined
+        return executor
+
+    async def _result_projection(self, context, request, session, resource, *, fields):
+        from msg.plugins.discovery import read_projection
+        return await read_projection(self,context,request,session,resource.id,
+                                     revision=resource.revision,fields=fields)
+
+    async def _event_notifications(self, session, event):
+        from msg.plugins.communication import enqueue_domain_webhooks
+        await enqueue_domain_webhooks(self,session,event)
+
     async def online_issuer(self,tx):
         issuer=tx.setting('online_ca_certificate')
         require(issuer is not None,'issuer_not_ready')
diff --git a/src/msg/core/batching.py b/src/msg/core/batching.py
new file mode 100644
index 00000000..d1b7bf2f
--- /dev/null
+++ b/src/msg/core/batching.py
@@ -0,0 +1,23 @@
+"""The one child-envelope validator for atomic and independent batches."""
+from msg.core.codec import canonical
+from msg.core.errors import require
+from msg.core.packet import decode_packet
+from msg.core.requests import SECRET_DELIVERY_MIN_VERSION
+
+# Credential delivery has response-only secrets; it requires its own call.
+NO_BATCH=frozenset(SECRET_DELIVERY_MIN_VERSION) | {'identity.register','identity.upgrade'}
+
+
+def packets(registry,request,subject,max_bytes=None):
+    max_bytes=max_bytes or len(canonical(request))
+    values=[]
+    for value in request.arguments['requests']:
+        packet=decode_packet(value,max_bytes)
+        spec=registry.operation(packet.operation,packet.contract_version)
+        require(spec.effect in {'read','transaction'} and not packet.operation.startswith('batch.')
+                and packet.operation not in NO_BATCH and 'network' in spec.entries,'operation_not_batchable')
+        require(packet.subject==subject,'batch_subject_mismatch')
+        values.append(packet)
+    require(len({p.request_id for p in values})==len(values),'duplicate_batch_request_id')
+    require(request.request_id not in {p.request_id for p in values},'recursive_request_id')
+    return values
diff --git a/src/msg/core/contracts.py b/src/msg/core/contracts.py
index 65eeb841..419d5eb4 100644
--- a/src/msg/core/contracts.py
+++ b/src/msg/core/contracts.py
@@ -1,12 +1,15 @@
 from __future__ import annotations
 
 from .models import *
+from .query import QuerySession
 from .tool_execution import ToolRunner
 
-class MetadataSession(Protocol):
+class MetadataSession(QuerySession, Protocol):
     def on_rollback(self, effect: Callable[[], Awaitable[None]]) -> None: ...
     async def resource(self, id: ResourceId) -> Resource: ...
     async def resolve(self, path: str) -> ResourceId: ...
+    async def path(self, id: ResourceId) -> str: ...
+    async def ancestors(self, id: ResourceId) -> tuple[Resource, ...]: ...
     async def children(
         self, parent: ResourceId, cursor: str | None, limit: int
     ) -> Page[Resource]: ...
@@ -20,6 +23,7 @@ async def replace(
     ) -> None: ...
     async def append_revision(self, revision: Revision) -> None: ...
     async def subject(self, id: ResourceId) -> Subject: ...
+    async def organization(self, id: ResourceId) -> Organization: ...
     async def credential(self, id: ID) -> Credential: ...
     async def certificate(self, id: ResourceId) -> Certificate: ...
     async def certificate_revoked(self, id: ResourceId) -> bool: ...
@@ -67,6 +71,8 @@ async def missing_ranges(
     async def append_event(self, event: Event) -> None: ...
     async def append_audit(self, event: AuditEvent) -> None: ...
     async def enqueue(self, job: EffectJob) -> None: ...
+    async def job(self, id: ID) -> EffectJob: ...
+    async def save_job(self, job: EffectJob) -> None: ...
 
 
 class MetadataStore(Protocol):
diff --git a/src/msg/core/events.py b/src/msg/core/events.py
new file mode 100644
index 00000000..c43b242f
--- /dev/null
+++ b/src/msg/core/events.py
@@ -0,0 +1,6 @@
+"""Deterministic committed-operation event identity shared by all producers."""
+from msg.core.codec import digest
+
+
+def event_id(request, subject):
+    return 'e_' + digest((subject, request.request_id))[7:39]
diff --git a/src/msg/core/execution_ports.py b/src/msg/core/execution_ports.py
new file mode 100644
index 00000000..6f1bd92f
--- /dev/null
+++ b/src/msg/core/execution_ports.py
@@ -0,0 +1,20 @@
+"""Fixed executor dependencies, not plugin hooks or a workflow interface.
+
+ResultProjection is a permission-checked read in the existing transaction.
+TransactionalEventNotifications may enqueue durable jobs in that transaction;
+it must never perform external I/O or commit it. Execution, auth and receipt
+ownership remain in OperationExecutor.
+"""
+from typing import Protocol
+from msg.core.contracts import MetadataSession
+from msg.core.models import Event, ExecutionContext, JsonMap, OperationRequest, ResourceRef
+
+
+class ResultProjection(Protocol):
+    async def __call__(self, context: ExecutionContext, request: OperationRequest,
+                       session: MetadataSession, resource: ResourceRef,
+                       *, fields: tuple[str, ...]) -> JsonMap: ...
+
+
+class TransactionalEventNotifications(Protocol):
+    async def __call__(self, session: MetadataSession, event: Event) -> None: ...
diff --git a/src/msg/core/executor.py b/src/msg/core/executor.py
index d00ac08d..87ea5881 100644
--- a/src/msg/core/executor.py
+++ b/src/msg/core/executor.py
@@ -5,6 +5,10 @@
 from dataclasses import replace
 from uuid import uuid4
 from msg.core.codec import wire,digest
+from msg.core.batching import packets
+from msg.core.events import event_id
+from msg.core.packet import result_wire
+from msg.core.execution_ports import ResultProjection, TransactionalEventNotifications
 from msg.core.errors import Failure,require
 from msg.core.models import ExecutionContext,HandlerOutput,OperationResult,OperationError,Event,AccessRequirement
 from msg.core.requests import SECRET_DELIVERY_MIN_VERSION,receipt_bytes
@@ -13,14 +17,21 @@
 
 
 class OperationExecutor:
-    def __init__(self,registry,metadata,contents,authenticator,authorizer,clock,receipt_signer):
+    def __init__(self,registry,metadata,contents,authenticator,authorizer,clock,receipt_signer,
+                 *, max_request_bytes: int | None = None,
+                 result_projection: ResultProjection | None = None,
+                 event_notifications: TransactionalEventNotifications | None = None):
         self.registry,self.metadata,self.contents=registry,metadata,contents
         self.authenticator,self.authorizer,self.clock=authenticator,authorizer,clock
         self.receipt_signer=receipt_signer
         self.response_hook=None
         self.recovery_drill_marker=None
         self.recovery_quarantined=False
-        self.application=None
+        require(max_request_bytes is None or (type(max_request_bytes) is int and max_request_bytes>0),
+                'invalid_request_limit')
+        self.max_request_bytes=max_request_bytes
+        self.result_projection=result_projection
+        self.event_notifications=event_notifications
 
     def recovery_drill_active(self):
         marker=self.recovery_drill_marker
@@ -49,9 +60,8 @@ async def execute(self,request, *, entry='network'):
                 if spec.name=='batch.atomic':
                     # Validate the child set before an old cached parent result
                     # can bypass the handler's secret-delivery exclusion.
-                    from msg.plugins.batch import packets
                     packets(self.registry,request,principal.subject,
-                            self.application.settings.server.limits.max_request_bytes)
+                            self.max_request_bytes)
                 context=ExecutionContext(request_id=request.request_id,principal=principal,entry=entry,
                                          now=self.clock(),deadline_monotonic=time.monotonic()+30)
                 checks=await spec.requirements(request,session)
@@ -94,9 +104,8 @@ async def execute(self,request, *, entry='network'):
                                 if exc.code not in {'not_found'}:raise
                     output=await spec.handler(context,request,session)
                     if request.return_fields:
-                        from msg.plugins.discovery import read_projection
-                        require(output.resources,'projection_unavailable')
-                        projections=[await read_projection(self.application,context,request,session,ref.id,revision=ref.revision,
+                        require(output.resources and self.result_projection is not None,'projection_unavailable')
+                        projections=[await self.result_projection(context,request,session,ref,
                                      fields=request.return_fields) for ref in output.resources]
                         output=replace(output,data={**(output.data or {}),'projection':projections})
                     if audited:
@@ -114,14 +123,12 @@ async def execute(self,request, *, entry='network'):
                         resources=output.resources,committed_at=self.clock() if spec.effect!='read' else None,
                         data=output.data,output=output.output)
                     if spec.effect!='read':
-                        from msg.plugins.communication import event_id
                         event=Event(id=event_id(request,principal.subject),type=spec.name,time=result.committed_at,request_id=request.request_id,
                                     actor=principal.actor,subject=principal.subject,resources=output.resources,
                                     data={'operation':spec.name})
                         await session.append_event(event)
-                        from msg.plugins.communication import WEBHOOK_DOMAIN_EVENTS,enqueue_domain_webhooks
-                        if event.type in WEBHOOK_DOMAIN_EVENTS and self.application is not None:
-                            await enqueue_domain_webhooks(self.application,session,event)
+                        if self.event_notifications is not None:
+                            await self.event_notifications(session,event)
                         result=replace(result,receipt=self.receipt_signer.sign(receipt_bytes(result),purpose='receipt'))
                         await session.save_result(principal.subject,request.payload_digest,result)
             # Only after the enclosing transaction commits may success reach the adapter.
@@ -150,12 +157,10 @@ async def _independent(self,request,spec,entry):
         Interrupted batches are resumed by resending the original signed children.
         No child is reported rolled back merely because its sibling failed.
         """
-        from msg.plugins.batch import packets
-        from msg.plugins.communication import event_id
         async with self.metadata.transaction(write=True) as tx:
             principal=await self.authenticator.authenticate(request,tx,entry=entry)
             await self.authorizer._ceiling(principal,f'{spec.name}@{spec.version}',principal.subject,tx)
-            children=packets(self.registry,request,principal.subject)
+            children=packets(self.registry,request,principal.subject,self.max_request_bytes)
             existing=tx.one('SELECT digest FROM batches WHERE subject=? AND request_id=?',(principal.subject,request.request_id))
             require(existing is None or existing[0]==request.payload_digest,'idempotency_conflict')
             previous=await tx.request_result(principal.subject,request.request_id,request.payload_digest)
@@ -186,11 +191,3 @@ async def _independent(self,request,spec,entry):
                 request_id=request.request_id,actor=principal.actor,subject=principal.subject,resources=result.resources,data={'operation':spec.name}))
             await tx.save_result(principal.subject,request.payload_digest,result)
         return replace(result,prefer_cli=request.source!='msg',cli_url='/AGENTS.md' if request.source!='msg' else None)
-
-
-def result_wire(result):
-    value=wire(result,compact=True)
-    for name in ('replayed','prefer_cli'):
-        if not value.get(name):
-            value.pop(name,None)
-    return value
diff --git a/src/msg/core/packet.py b/src/msg/core/packet.py
new file mode 100644
index 00000000..97cc4b29
--- /dev/null
+++ b/src/msg/core/packet.py
@@ -0,0 +1,71 @@
+"""Transport-neutral request/result contracts and bounded canonical decoding."""
+from __future__ import annotations
+from collections.abc import Mapping
+import re
+from msg.core.codec import decode, loads, canonical, wire
+from msg.core.errors import require
+from msg.core.models import OperationRequest, OperationResult
+from msg.core.schemas import obj, IDENTIFIER, STRING, BYTES, REF, SIGNATURE
+
+REQUEST_SCHEMA=obj({
+    'request_id':{'type':'string','minLength':1,'maxLength':128},'protocol_version':{'const':1},
+    'operation':{'type':'string','pattern':'^[a-z][a-z0-9_]*(\\.[a-z][a-z0-9_]*)+$','maxLength':128},
+    'contract_version':{'type':'integer','minimum':1},'target_service':{'type':'string','maxLength':2048},
+    'subject':{'type':['string','null'],'maxLength':160},'arguments':{'type':'object'},
+    'expected_generations':{'type':'array','maxItems':64,'items':{'type':'array','prefixItems':[IDENTIFIER,{'type':'integer','minimum':0}],
+                                                                 'minItems':2,'maxItems':2}},
+    'expires_at':{'type':['string','null'],'maxLength':40},
+    'payload_digest':{'type':'string','pattern':'^sha256:[0-9a-f]{64}$'},
+    'proof':{'anyOf':[{'type':'null'},obj({'signature':SIGNATURE,'certificates':{'type':'array','items':IDENTIFIER,'maxItems':16}},('signature','certificates')),
+                    obj({'credential_id':IDENTIFIER,'token':BYTES},('credential_id','token'))]},
+    'return_fields':{'type':'array','maxItems':30,'items':STRING},'source':{'enum':['msg','manual','mcp','unknown']}
+},('request_id','protocol_version','operation','contract_version','target_service','subject','arguments',
+    'expected_generations','expires_at','payload_digest','proof'))
+RESULT_SCHEMA={'type':'object','properties':{
+    'request_id':STRING,'operation':STRING,'status':{'enum':['ok','accepted','error','uncertain']},
+    'actor':STRING,'subject':STRING,'resources':{'type':'array','items':REF},'committed_at':STRING,
+    'replayed':{'type':'boolean'},'receipt':SIGNATURE,'error':{'type':'object'},'data':{'type':'object'},
+    'output':REF,'prefer_cli':{'type':'boolean'},'cli_url':STRING},
+    'required':['request_id','operation','status'],'additionalProperties':False}
+
+
+def decode_packet(value,max_bytes=1048576):
+    from jsonschema import Draft202012Validator
+    if isinstance(value,(bytes,str)):
+        require(len(value.encode() if isinstance(value,str) else value)<=max_bytes,'request_too_large')
+        value=loads(value)
+    value=wire(value)
+    require(len(canonical(value))<=max_bytes,'request_too_large')
+    error=next(Draft202012Validator(REQUEST_SCHEMA).iter_errors(value),None)
+    require(error is None,'invalid_request_envelope', '.'.join(map(str,error.path)) if error else None)
+    packet=decode(OperationRequest,value)
+    require(len(set(rid for rid,generation in packet.expected_generations))==len(packet.expected_generations),
+            'duplicate_expected_generation')
+    return packet
+
+
+def safe_error_code(value, fallback):
+    """Only bounded contract identifiers, never exception text or credential URLs."""
+    return (value if isinstance(value, str) and
+            re.fullmatch(r'[a-z][a-z0-9_]{0,95}', value) else fallback)
+
+
+def decode_result(value):
+    # Reject malformed signed/unsigned results rather than rewriting their error
+    # fields. Valid receipts and persisted result bytes are left unchanged.
+    require(isinstance(value, Mapping), 'invalid_result_envelope')
+    error = value.get('error')
+    if error is not None:
+        require(isinstance(error, Mapping) and safe_error_code(error.get('code'), None) is not None,
+                'invalid_result_envelope')
+    # Compact results omit anonymous actor/subject; these required domain fields
+    # are restored explicitly, not inferred from a transport connection.
+    return decode(OperationResult,dict({'actor':None,'subject':None},**value))
+
+
+def result_wire(result):
+    value=wire(result,compact=True)
+    for name in ('replayed','prefer_cli'):
+        if not value.get(name):
+            value.pop(name,None)
+    return value
diff --git a/src/msg/core/query.py b/src/msg/core/query.py
new file mode 100644
index 00000000..13991aa9
--- /dev/null
+++ b/src/msg/core/query.py
@@ -0,0 +1,44 @@
+"""Explicit SQL query port for trusted server-side domain consumers.
+
+Queries use the shared positional '?' subset; a backend may translate it.
+This is not an ORM or an untrusted-SQL interface. Handlers/requirements, local
+administrative use cases and workers may consume it within the transaction
+provided by MetadataStore; adapters/clients must not construct sessions.
+Writes require a write transaction and write=True. Results cannot escape its
+owning task or lifetime. Integrity conflicts are constraint_conflict, and
+PostgreSQL lock/deadlock timeouts are retryable server_busy; adapter-specific
+programming errors still abort the transaction. No commit or connection port
+is exposed. Driver and database protections remain the authority for SQL.
+"""
+from __future__ import annotations
+
+from collections.abc import Iterator, Mapping, Sequence
+from decimal import Decimal
+from typing import Protocol, runtime_checkable
+
+# PostgreSQL SUM(bigint) returns Decimal; ordinary stored times/JSON are text.
+type SqlValue = None | bool | int | float | str | bytes | Decimal
+type SqlRow = tuple[SqlValue, ...]
+type SqlParameters = Sequence[SqlValue]
+type SettingValue = (
+    None | bool | int | float | str | list[SettingValue]
+    | tuple[SettingValue, ...] | Mapping[str, SettingValue]
+)
+
+
+class QueryResult(Protocol):
+    @property
+    def rowcount(self) -> int: ...
+    def fetchone(self) -> SqlRow | None: ...
+    def fetchall(self) -> list[SqlRow]: ...
+    def __iter__(self) -> Iterator[SqlRow]: ...
+
+
+@runtime_checkable
+class QuerySession(Protocol):
+    def execute(self, sql: str, parameters: SqlParameters = (), *,
+                write: bool = False) -> QueryResult: ...
+    def one(self, sql: str, parameters: SqlParameters = ()) -> SqlRow | None: ...
+    def rows(self, sql: str, parameters: SqlParameters = ()) -> list[SqlRow]: ...
+    def setting(self, key: str, default: SettingValue = None) -> SettingValue: ...
+    def set_setting(self, key: str, value: SettingValue) -> None: ...
diff --git a/src/msg/core/schemas.py b/src/msg/core/schemas.py
new file mode 100644
index 00000000..97e00ffe
--- /dev/null
+++ b/src/msg/core/schemas.py
@@ -0,0 +1,30 @@
+"""Small explicit JSON schemas; no open-ended executable configuration."""
+STRING={'type':'string'}
+IDENTIFIER={'type':'string','minLength':1,'maxLength':160}
+INTEGER={'type':'integer','minimum':0}
+BOOLEAN={'type':'boolean'}
+BYTES={'type':'string','pattern':'^[A-Za-z0-9_-]*$'}
+REF={'type':'object','properties':{'id':IDENTIFIER,'revision':{'type':['string','null']}},'required':['id'],'additionalProperties':False}
+SCOPE={'type':'object','properties':{'resource_id':IDENTIFIER,'descendants':BOOLEAN},'required':['resource_id'],'additionalProperties':False}
+NETWORK_CONSTRAINTS={'type':'object','properties':{
+    'hosts':{'type':'array','items':STRING},'schemes':{'type':'array','items':{'enum':['http','https']}},
+    'methods':{'type':'array','items':{'enum':['GET','HEAD','POST','PUT','PATCH','DELETE','OPTIONS']}},
+    'ports':{'type':'array','items':{'type':'integer','minimum':1,'maximum':65535}},
+    'timeout_ms':{'type':'integer','minimum':1},'max_response_bytes':{'type':'integer','minimum':1},
+    'max_redirects':{'type':'integer','minimum':0}},'additionalProperties':False}
+GRANT={'type':'object','properties':{'capability':IDENTIFIER,'version':{'const':1},'scope':SCOPE,
+    'operations':{'type':'array','items':IDENTIFIER,'minItems':1,'uniqueItems':True},
+    'constraints':{'type':'object'}},'required':['capability','version','scope','operations','constraints'],
+    'additionalProperties':False}
+GRANTS={'type':'array','items':GRANT}
+ISSUANCE={'type':['object','null'],'properties':{'issue_grants':GRANTS,'max_cert_ttl_seconds':INTEGER,
+    'max_child_ca_depth':INTEGER,'max_delegation_depth':INTEGER},
+    'required':['issue_grants','max_cert_ttl_seconds','max_child_ca_depth','max_delegation_depth'],'additionalProperties':False}
+SIGNATURE={'type':'object','properties':{'key_id':IDENTIFIER,'algorithm':{'const':'ed25519'},'value':BYTES},
+           'required':['key_id','algorithm','value'],'additionalProperties':False}
+OUTPUT={'type':'object','properties':{'resources':{'type':'array','items':REF},'data':{'type':['object','null']},
+    'output':{'anyOf':[REF,{'type':'null'}]}},'required':['resources','data','output'],'additionalProperties':False}
+
+
+def obj(properties=None,required=()):
+    return {'type':'object','properties':properties or {},'required':list(required),'additionalProperties':False}
diff --git a/src/msg/extensions/ssh.py b/src/msg/extensions/ssh.py
index d12baadc..1d438593 100644
--- a/src/msg/extensions/ssh.py
+++ b/src/msg/extensions/ssh.py
@@ -21,7 +21,7 @@
 from msg.constants import ROOT_SUBJECT
 from msg.core.codec import canonical, decode, digest, wire
 from msg.core.errors import Failure, require
-from msg.core.executor import OperationExecutor, result_wire
+from msg.core.packet import result_wire
 from msg.core.models import Credential, Principal, Signature, SignatureProof, CapabilityGrant, HandlerOutput, EffectJob, ResourceRef
 from msg.core.requests import request_for, payload_fields
 from msg.plugins.common import operation_id, check_access, new_id, resolve
@@ -229,10 +229,7 @@ async def forced_session(config_dir, credential_id):
     try:
         command = parse_command(os.environ.get('SSH_ORIGINAL_COMMAND', ''))
         auth = SSHAuthenticator(app, credential_id)
-        executor = OperationExecutor(app.registry, app.metadata, app.contents, auth, app.authorizer, app.clock, app.receipt_signer)
-        # The normal secret delivery hook remains gated by signed operations.
-        executor.application = app
-        executor.response_hook = app._secrets_for_caller
+        executor = app.new_executor(auth)
         async with app.metadata.transaction(write=False) as tx:
             principal = await ssh_principal(app, credential_id, tx)
         if command[0] in {'call','packet'}:
diff --git a/src/msg/market/policy.py b/src/msg/market/policy.py
index 3b414cca..9717a4da 100644
--- a/src/msg/market/policy.py
+++ b/src/msg/market/policy.py
@@ -3,6 +3,7 @@
 
 from msg.core.codec import canonical, digest, loads
 from msg.core.errors import require
+from msg.core.query import QuerySession
 
 DEFAULT_POLICY = {
     'id': 'dispute-v1', 'version': 1,
@@ -44,7 +45,7 @@ def validate(policy):
     return policy
 
 
-def load_policy(tx, policy_id):
+def load_policy(tx: QuerySession, policy_id: str):
     row = tx.one('SELECT body,digest FROM arbitration_policies WHERE id=?', (policy_id,))
     if row:
         policy = loads(row[0])
diff --git a/src/msg/plugins/batch.py b/src/msg/plugins/batch.py
index beb05dbf..b54122db 100644
--- a/src/msg/plugins/batch.py
+++ b/src/msg/plugins/batch.py
@@ -1,32 +1,11 @@
 """Bounded batching is a protocol primitive, not an executable workflow."""
 from __future__ import annotations
-from msg.core.codec import wire,canonical
 from msg.core.errors import Failure,require
-from msg.core.executor import result_wire
-from msg.core.requests import SECRET_DELIVERY_MIN_VERSION
+from msg.core.packet import result_wire, REQUEST_SCHEMA
+from msg.core.batching import NO_BATCH, packets
 from msg.core.models import HandlerOutput
 from msg.plugins.common import registration,operation_id
 from msg.plugins.schemas import obj
-from msg.transports.packet import REQUEST_SCHEMA,decode_packet
-
-# Credential delivery has response-only secrets; it requires its own call.
-NO_BATCH=frozenset(SECRET_DELIVERY_MIN_VERSION) | {'identity.register','identity.upgrade'}
-
-
-def packets(registry,request,subject,max_bytes=None):
-    max_bytes=max_bytes or len(canonical(request))
-    values=[]
-    for value in request.arguments['requests']:
-        packet=decode_packet(value,max_bytes)
-        spec=registry.operation(packet.operation,packet.contract_version)
-        require(spec.effect in {'read','transaction'} and not packet.operation.startswith('batch.')
-                and packet.operation not in NO_BATCH and 'network' in spec.entries,'operation_not_batchable')
-        require(packet.subject==subject,'batch_subject_mismatch')
-        values.append(packet)
-    require(len({p.request_id for p in values})==len(values),'duplicate_batch_request_id')
-    require(request.request_id not in {p.request_id for p in values},'recursive_request_id')
-    return values
-
 
 def install(app):
     op,finish=registration(app,'batch',('identity',))
diff --git a/src/msg/plugins/communication.py b/src/msg/plugins/communication.py
index d704aa7e..a97ce67a 100644
--- a/src/msg/plugins/communication.py
+++ b/src/msg/plugins/communication.py
@@ -17,8 +17,7 @@
 from msg.storage.capacity import require_webhook_capacity
 
 
-def event_id(request,subject):
-    return 'e_'+digest((subject,request.request_id))[7:39]
+from msg.core.events import event_id
 
 
 # A deliberately small event vocabulary. An Event is an audit fact, not blanket
diff --git a/src/msg/plugins/schemas.py b/src/msg/plugins/schemas.py
index 97e00ffe..275fc60e 100644
--- a/src/msg/plugins/schemas.py
+++ b/src/msg/plugins/schemas.py
@@ -1,30 +1,5 @@
-"""Small explicit JSON schemas; no open-ended executable configuration."""
-STRING={'type':'string'}
-IDENTIFIER={'type':'string','minLength':1,'maxLength':160}
-INTEGER={'type':'integer','minimum':0}
-BOOLEAN={'type':'boolean'}
-BYTES={'type':'string','pattern':'^[A-Za-z0-9_-]*$'}
-REF={'type':'object','properties':{'id':IDENTIFIER,'revision':{'type':['string','null']}},'required':['id'],'additionalProperties':False}
-SCOPE={'type':'object','properties':{'resource_id':IDENTIFIER,'descendants':BOOLEAN},'required':['resource_id'],'additionalProperties':False}
-NETWORK_CONSTRAINTS={'type':'object','properties':{
-    'hosts':{'type':'array','items':STRING},'schemes':{'type':'array','items':{'enum':['http','https']}},
-    'methods':{'type':'array','items':{'enum':['GET','HEAD','POST','PUT','PATCH','DELETE','OPTIONS']}},
-    'ports':{'type':'array','items':{'type':'integer','minimum':1,'maximum':65535}},
-    'timeout_ms':{'type':'integer','minimum':1},'max_response_bytes':{'type':'integer','minimum':1},
-    'max_redirects':{'type':'integer','minimum':0}},'additionalProperties':False}
-GRANT={'type':'object','properties':{'capability':IDENTIFIER,'version':{'const':1},'scope':SCOPE,
-    'operations':{'type':'array','items':IDENTIFIER,'minItems':1,'uniqueItems':True},
-    'constraints':{'type':'object'}},'required':['capability','version','scope','operations','constraints'],
-    'additionalProperties':False}
-GRANTS={'type':'array','items':GRANT}
-ISSUANCE={'type':['object','null'],'properties':{'issue_grants':GRANTS,'max_cert_ttl_seconds':INTEGER,
-    'max_child_ca_depth':INTEGER,'max_delegation_depth':INTEGER},
-    'required':['issue_grants','max_cert_ttl_seconds','max_child_ca_depth','max_delegation_depth'],'additionalProperties':False}
-SIGNATURE={'type':'object','properties':{'key_id':IDENTIFIER,'algorithm':{'const':'ed25519'},'value':BYTES},
-           'required':['key_id','algorithm','value'],'additionalProperties':False}
-OUTPUT={'type':'object','properties':{'resources':{'type':'array','items':REF},'data':{'type':['object','null']},
-    'output':{'anyOf':[REF,{'type':'null'}]}},'required':['resources','data','output'],'additionalProperties':False}
-
-
-def obj(properties=None,required=()):
-    return {'type':'object','properties':properties or {},'required':list(required),'additionalProperties':False}
+"""Compatibility exports; shared wire schemas are owned by msg.core.schemas."""
+from msg.core.schemas import (
+    STRING, IDENTIFIER, INTEGER, BOOLEAN, BYTES, REF, SCOPE, NETWORK_CONSTRAINTS,
+    GRANT, GRANTS, ISSUANCE, SIGNATURE, OUTPUT, obj,
+)
diff --git a/src/msg/storage/postgres.py b/src/msg/storage/postgres.py
index 5ac02cff..624326e8 100644
--- a/src/msg/storage/postgres.py
+++ b/src/msg/storage/postgres.py
@@ -23,7 +23,9 @@
 from psycopg.conninfo import conninfo_to_dict, make_conninfo
 
 from msg.core.errors import Failure
-from msg.storage.sqlite import SqliteSession
+from msg.core.query import QueryResult, SqlParameters
+from msg.storage.query import SessionQueryResult
+from msg.storage.session import RelationalSession
 
 _SCHEMA = """
 CREATE TABLE IF NOT EXISTS schema_version (version INTEGER PRIMARY KEY);
@@ -509,18 +511,21 @@ def _postgres_sql(sql: str, *, has_parameters: bool = False) -> str:
     return ''.join(output)
 
 
-class PostgresSession(SqliteSession):
+class PostgresSession(RelationalSession):
     def __init__(self, connection, *, write: bool):
-        super().__init__(connection, write=write)
+        super().__init__(write=write)
+        self._connection = connection
         self.pending_effect_ids: list[str] = []
 
-    def execute(self, sql, parameters=(), *, write=False):
+    def execute(self, sql: str, parameters: SqlParameters = (), *,
+                write: bool = False) -> QueryResult:
         self.check(write)
         try:
             # With an empty parameter tuple psycopg still parses literal '%' as a
             # placeholder; SQL such as LIKE 'policy:%' must be sent unchanged.
-            return self._connection.execute(
+            cursor = self._connection.execute(
                 _postgres_sql(sql, has_parameters=bool(parameters)), parameters or None)
+            return SessionQueryResult(cursor, self.check)
         except psycopg.errors.IntegrityError as exc:
             raise Failure("constraint_conflict") from exc
         except (psycopg.errors.DeadlockDetected, psycopg.errors.LockNotAvailable,
diff --git a/src/msg/storage/query.py b/src/msg/storage/query.py
new file mode 100644
index 00000000..092787e8
--- /dev/null
+++ b/src/msg/storage/query.py
@@ -0,0 +1,39 @@
+"""Session-bound query results without a public driver cursor/connection."""
+from __future__ import annotations
+
+from collections.abc import Callable, Iterator
+from typing import Protocol
+from msg.core.query import SqlRow
+
+
+class _Cursor(Protocol):
+    @property
+    def rowcount(self) -> int: ...
+    def fetchone(self) -> SqlRow | None: ...
+    def fetchall(self) -> list[SqlRow]: ...
+
+
+class SessionQueryResult:
+    """A narrow view, not a sandbox for trusted in-process Python consumers."""
+    __slots__ = ('_cursor', '_check')
+
+    def __init__(self, cursor: _Cursor, check: Callable[[], None]):
+        self._cursor = cursor
+        self._check = check
+
+    @property
+    def rowcount(self) -> int:
+        self._check()
+        return self._cursor.rowcount
+
+    def fetchone(self) -> SqlRow | None:
+        self._check()
+        return self._cursor.fetchone()
+
+    def fetchall(self) -> list[SqlRow]:
+        self._check()
+        return self._cursor.fetchall()
+
+    def __iter__(self) -> Iterator[SqlRow]:
+        while (row := self.fetchone()) is not None:
+            yield row
diff --git a/src/msg/storage/session.py b/src/msg/storage/session.py
new file mode 100644
index 00000000..554b754d
--- /dev/null
+++ b/src/msg/storage/session.py
@@ -0,0 +1,403 @@
+"""Shared relational domain methods, with no driver or transaction ownership.
+
+This SQL-oriented monolith shares the supported placeholder/query subset, not
+an imaginary database-independent object store. Adapters own SQL translation,
+schema migration, connections, writer fences, savepoints and commit/rollback.
+Only this session's owning task can access its transaction or results.
+"""
+from __future__ import annotations
+
+import asyncio
+from abc import ABC, abstractmethod
+from dataclasses import replace
+from msg.core.codec import canonical, decode, digest, loads, wire
+from msg.core.errors import require
+from msg.core.query import QueryResult, SettingValue, SqlParameters, SqlRow
+from msg.core.models import (
+    AuditEvent, Certificate, CertificateRequest, CertificateRequestState, Credential,
+    EffectJob, EmailSettings, Event, Membership, OperationResult, Organization, Page,
+    Resource, ResourceRef, Revision, Subject, TransferChunk, TransferSession,
+)
+
+
+class RelationalSession(ABC):
+    def __init__(self, *, write: bool):
+        self.write = write
+        self.owner_task = asyncio.current_task()
+        self.closed = False
+        self.rollback_effects = []
+
+    def on_rollback(self, effect):
+        """Compensate external work if this transaction or savepoint aborts."""
+        self.check(write=True)
+        self.rollback_effects.append(effect)
+
+    async def run_rollback_effects(self, cause, start=0):
+        effects = self.rollback_effects[start:]
+        del self.rollback_effects[start:]
+        if not effects:
+            return
+
+        async def compensate():
+            for effect in reversed(effects):
+                try:
+                    await effect()
+                except BaseException as exc:
+                    # Keep the primary error and try every independent cleanup.
+                    cause.add_note('rollback compensation failed: ' + type(exc).__name__)
+
+        # A second cancellation must not abandon a pin operation (including its
+        # worker thread) and release the writer fence before it finishes.
+        pending = asyncio.create_task(compensate())
+        while not pending.done():
+            try:
+                await asyncio.shield(pending)
+            except asyncio.CancelledError:
+                pass
+        pending.result()
+
+    def check(self, write=False):
+        require(not self.closed, "transaction_closed")
+        require(asyncio.current_task() is self.owner_task, "transaction_cross_task")
+        if write:
+            require(self.write, "read_only_transaction")
+
+    @abstractmethod
+    def execute(self, sql: str, parameters: SqlParameters = (), *,
+                write: bool = False) -> QueryResult:
+        """Run adapter-normalized SQL in this store-owned transaction."""
+        raise NotImplementedError
+
+    def one(self, sql: str, parameters: SqlParameters = ()) -> SqlRow | None:
+        return self.execute(sql, parameters).fetchone()
+
+    def rows(self, sql: str, parameters: SqlParameters = ()) -> list[SqlRow]:
+        return self.execute(sql, parameters).fetchall()
+
+    async def resource(self, id):
+        row = self.one("SELECT body FROM resources WHERE id=?", (id,))
+        require(row is not None, "not_found")
+        return decode(Resource, loads(row[0]))
+
+    async def resolve(self, path):
+        require(isinstance(path, str) and path.startswith("/"), "invalid_path")
+        parts = path.rstrip("/").split("/")[1:]
+        require(all(p not in {".", ".."} and "\\" not in p and "\x00" not in p for p in parts), "invalid_path")
+        if len(parts) == 2 and parts[0] == "_id":
+            return (await self.resource(parts[1])).id
+        row = self.one("SELECT id FROM resources WHERE parent IS NULL")
+        require(row is not None, "not_initialized")
+        rid = row[0]
+        if path == "/":
+            return rid
+        for part in parts:
+            require(bool(part), "invalid_path")
+            row = self.one("SELECT id FROM resources WHERE parent=? AND name=?", (rid, part))
+            require(row is not None, "not_found")
+            rid = row[0]
+        return rid
+
+    async def path(self, id):
+        segments, seen = [], set()
+        r = await self.resource(id)
+        while r.parent is not None:
+            require(r.id not in seen, "parent_cycle")
+            seen.add(r.id)
+            segments.append(r.name)
+            r = await self.resource(r.parent)
+        return "/" + "/".join(reversed(segments))
+
+    async def ancestors(self, id):
+        result, seen = [], set()
+        r = await self.resource(id)
+        while r.parent is not None:
+            require(r.id not in seen, "parent_cycle")
+            seen.add(r.id)
+            r = await self.resource(r.parent)
+            result.append(r)
+        return tuple(reversed(result))
+
+    async def children(self, parent, cursor=None, limit=50):
+        require(type(limit) is int and 1 <= limit <= 500, "invalid_limit")
+        rows = self.rows("SELECT body FROM resources WHERE parent=? AND id>? ORDER BY id LIMIT ?",
+                         (parent, cursor or "", limit + 1))
+        values = tuple(decode(Resource, loads(row[0])) for row in rows[:limit])
+        return Page(items=values, next_cursor=values[-1].id if len(rows) > limit else None)
+
+    async def revision(self, ref):
+        rid = ref.revision or (await self.resource(ref.id)).revision
+        row = self.one("SELECT body FROM revisions WHERE id=? AND resource_id=?", (rid, ref.id))
+        require(row is not None, "revision_not_found")
+        return decode(Revision, loads(row[0]))
+
+    async def history(self, id, cursor=None, limit=50):
+        require(1 <= limit <= 500, "invalid_limit")
+        rows = self.rows("SELECT body FROM revisions WHERE resource_id=? AND id>? ORDER BY id LIMIT ?",
+                         (id, cursor or "", limit+1))
+        values = tuple(decode(Revision, loads(r[0])) for r in rows[:limit])
+        return Page(items=values, next_cursor=values[-1].id if len(rows)>limit else None)
+
+    async def insert(self, resource):
+        self.check(True)
+        if resource.parent is not None:
+            await self.resource(resource.parent)
+        data = wire(resource)
+        self.execute("INSERT INTO resources VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)", (
+            resource.id, resource.type, resource.name, resource.parent, resource.owner,
+            resource.group, resource.mode, resource.generation, resource.revision,
+            resource.state, data['created_at'], data['modified_at'], canonical(resource).decode()), write=True)
+        for tag in resource.tags:
+            self.execute("INSERT INTO resource_tags VALUES (?,?)", (resource.id,tag), write=True)
+
+    async def replace(self, resource, expected_generation):
+        self.check(True)
+        old = await self.resource(resource.id)
+        require(old.generation == expected_generation, "generation_conflict",
+                details={"generation":old.generation, "revision":old.revision})
+        require(resource.generation == expected_generation + 1, "invalid_generation")
+        require(resource.type == old.type and resource.created_at == old.created_at and
+                resource.created_by == old.created_by, "immutable_creation_fact")
+        if resource.parent is not None:
+            require(resource.parent != resource.id, "parent_cycle")
+            parent = await self.resource(resource.parent)
+            require(all(a.id != resource.id for a in await self.ancestors(parent.id)), "parent_cycle")
+        if any(getattr(resource,k)!=getattr(old,k) for k in ('parent','owner','group','mode','state')):
+            self.set_setting('authorization_epoch',self.setting('authorization_epoch',0)+1)
+        data = wire(resource)
+        changed = self.execute("""UPDATE resources SET name=?,parent=?,owner=?,grp=?,mode=?,
+            generation=?,revision=?,state=?,modified_at=?,body=? WHERE id=? AND generation=?""",
+            (resource.name, resource.parent, resource.owner, resource.group, resource.mode,
+             resource.generation, resource.revision, resource.state, data['modified_at'],
+             canonical(resource).decode(), resource.id, expected_generation), write=True).rowcount
+        require(changed == 1, "generation_conflict")
+        if resource.tags != old.tags:
+            self.execute("DELETE FROM resource_tags WHERE resource_id=?", (resource.id,), write=True)
+            for tag in resource.tags:
+                self.execute("INSERT INTO resource_tags VALUES (?,?)", (resource.id,tag), write=True)
+
+    async def append_revision(self, revision):
+        self.check(True)
+        for parent in revision.parents:
+            await self.revision(ResourceRef(id=revision.resource_id, revision=parent))
+        self.execute("INSERT INTO revisions VALUES (?,?,?,?)", (
+            revision.id, revision.resource_id, wire(revision.created_at), canonical(revision).decode()), write=True)
+        for relation in revision.relations:
+            self.execute("INSERT INTO relations VALUES (?,?,?,?,?,?)", (
+                revision.id, revision.resource_id, relation.type, relation.target.id,
+                relation.target.revision, canonical(relation).decode()), write=True)
+
+    async def subject(self, id):
+        row = self.one("SELECT body FROM identities WHERE id=? AND kind='subject'", (id,))
+        require(row is not None, "subject_not_found")
+        return decode(Subject, loads(row[0]))
+
+    async def organization(self, id):
+        row = self.one("SELECT body FROM identities WHERE id=? AND kind='organization'", (id,))
+        require(row is not None, "group_not_found")
+        return decode(Organization, loads(row[0]))
+
+    async def credential(self, id):
+        row = self.one("SELECT body FROM credentials WHERE id=?", (id,))
+        require(row is not None, "credential_not_found")
+        return decode(Credential, loads(row[0]))
+
+    async def certificate(self, id):
+        row = self.one("SELECT body FROM certificates WHERE id=?", (id,))
+        require(row is not None, "certificate_not_found")
+        return decode(Certificate, loads(row[0]))
+
+    async def certificate_revoked(self, id):
+        row = self.one("SELECT revoked FROM certificates WHERE id=?", (id,))
+        require(row is not None, "certificate_not_found")
+        return bool(row[0])
+
+    async def memberships(self, subject):
+        if subject is None:
+            return ()
+        rows=self.rows("""SELECT m.org,m.body FROM memberships m
+            JOIN resources r ON r.id=m.org
+            WHERE m.subject=? AND r.type='organization' AND r.state='active'
+            ORDER BY m.org""",(subject,))
+        memberships=((org,decode(Membership,loads(body))) for org,body in rows)
+        # A stale membership row must not revive an archived/missing organization
+        # or borrow an organization/subject identifier from a mismatched body.
+        active=tuple(member for org,member in memberships if member.status=='active'
+                     and member.organization_id==org and member.subject_id==subject
+                     and org!='g_public')
+        row=self.one("SELECT body FROM identities WHERE id=? AND kind='subject'",(subject,))
+        if row is None:
+            return active
+        identity=decode(Subject,loads(row[0]))
+        if identity.kind in {'registered','custodial','system'} and not identity.local_only:
+            # /&public is a virtual group for formal subjects. Historical rows
+            # may remain in storage, but never create a duplicate authorization.
+            return (Membership(organization_id='g_public',subject_id=subject,
+                               role='member',version=0),*active)
+        return active
+
+    async def request_result(self, subject, id, digest):
+        row = self.one("SELECT digest,body FROM results WHERE subject=? AND request_id=?", (subject,id))
+        if row is None:
+            return None
+        require(row[0] == digest, "idempotency_conflict")
+        return decode(OperationResult, loads(row[1]))
+
+    async def save_result(self, subject, digest, result):
+        from msg.storage.capacity import require_result_capacity
+        require_result_capacity(self)
+        self.execute("INSERT INTO results VALUES (?,?,?,?)", (
+            subject, result.request_id, digest, canonical(result).decode()), write=True)
+
+    async def update_identity(self, record, expected_generation):
+        self.check(True)
+        if isinstance(record, Membership):
+            self.set_setting('authorization_epoch',self.setting('authorization_epoch',0)+1)
+            table, where, args = 'memberships', 'org=? AND subject=?', (record.organization_id,record.subject_id)
+            generation = record.version
+        elif isinstance(record, EmailSettings):
+            table, where, args = 'emails', 'subject=?', (record.subject_id,)
+            generation = expected_generation + 1
+        else:
+            table, where, args = 'identities', 'id=?', (record.resource_id,)
+            generation = record.auth_version if isinstance(record, Subject) else record.membership_version
+        row = self.one(f"SELECT generation FROM {table} WHERE {where}", args)
+        require((row is None and expected_generation == -1) or
+                (row is not None and row[0] == expected_generation), "generation_conflict")
+        payload = canonical(record).decode()
+        if row is None:
+            if table == 'identities':
+                self.execute("INSERT INTO identities VALUES (?,?,?,?)", (*args,
+                    'subject' if isinstance(record,Subject) else 'organization', generation,payload), write=True)
+            elif table == 'memberships':
+                self.execute("INSERT INTO memberships VALUES (?,?,?,?)", (*args,generation,payload), write=True)
+            else:
+                self.execute("INSERT INTO emails VALUES (?,?,?)", (*args,generation,payload), write=True)
+        else:
+            self.execute(f"UPDATE {table} SET generation=?,body=? WHERE {where}",
+                         (generation,payload,*args), write=True)
+
+    async def save_credential(self, credential, expected_auth_version):
+        subject = await self.subject(credential.subject_id)
+        require(subject.auth_version == expected_auth_version, "auth_version_conflict")
+        old=self.one("SELECT body FROM credentials WHERE id=?",(credential.id,))
+        if old is not None:
+            previous=decode(Credential,loads(old[0]))
+            require(previous.subject_id==credential.subject_id and previous.kind==credential.kind and
+                    previous.verifier==credential.verifier,"credential_identity_immutable")
+        self.execute("INSERT INTO credentials VALUES (?,?,?) ON CONFLICT(id) DO UPDATE SET body=excluded.body",
+                     (credential.id,credential.subject_id,canonical(credential).decode()),write=True)
+
+    async def csr(self, id):
+        row = self.one("SELECT body FROM csrs WHERE id=?",(id,))
+        require(row is not None,"csr_not_found")
+        return decode(CertificateRequest,loads(row[0]))
+
+    async def csr_state(self, id):
+        row=self.one("SELECT state_body FROM csrs WHERE id=?",(id,))
+        require(row is not None,"csr_not_found")
+        return decode(CertificateRequestState,loads(row[0]))
+
+    async def save_csr(self, request):
+        state=CertificateRequestState(request_id=request.resource_id,status='pending',generation=0)
+        self.execute("INSERT INTO csrs VALUES (?,?,?,?,?)",(request.resource_id,0,'pending',
+            canonical(request).decode(),canonical(state).decode()),write=True)
+
+    async def transition_csr(self, state, expected_generation):
+        require(state.generation == expected_generation+1,"invalid_generation")
+        old=await self.csr_state(state.request_id)
+        require(old.status=='pending',"csr_not_pending")
+        require(old.generation==expected_generation,"generation_conflict")
+        self.execute("UPDATE csrs SET generation=?,state=?,state_body=? WHERE id=? AND generation=?",
+            (state.generation,state.status,canonical(state).decode(),state.request_id,expected_generation),write=True)
+
+    async def register_certificate(self, certificate, csr_id, expected_generation):
+        self.execute("INSERT INTO certificates VALUES (?,?,?,?,?)",(
+            certificate.resource_id,certificate.subject_id,certificate.parent_certificate_id,0,
+            canonical(certificate).decode()),write=True)
+        if csr_id is not None:
+            self.set_setting('certificate_request:'+certificate.resource_id,csr_id)
+            await self.transition_csr(CertificateRequestState(request_id=csr_id,status='issued',
+                generation=expected_generation+1,certificate_id=certificate.resource_id),expected_generation)
+
+    async def revoke_certificate(self, id, event):
+        await self.certificate(id)
+        self.execute("UPDATE certificates SET revoked=1 WHERE id=?",(id,),write=True)
+        self.set_setting('authorization_epoch',self.setting('authorization_epoch',0)+1)
+        await self.append_audit(event)
+
+    async def transfer(self,id):
+        row=self.one("SELECT body FROM transfers WHERE id=?",(id,))
+        require(row is not None,"transfer_not_found")
+        return decode(TransferSession,loads(row[0]))
+
+    async def save_transfer(self,session,expected_generation):
+        if expected_generation is None:
+            self.execute("INSERT INTO transfers (id,subject,generation,body) VALUES (?,?,?,?)",
+                         (session.id,session.subject_id,session.generation,canonical(session).decode()),write=True)
+        else:
+            require(session.generation==expected_generation+1,"invalid_generation")
+            count=self.execute("UPDATE transfers SET generation=?,body=? WHERE id=? AND generation=?",
+                (session.generation,canonical(session).decode(),session.id,expected_generation),write=True).rowcount
+            require(count==1,"generation_conflict")
+
+    async def put_chunk(self,chunk):
+        self.check(True)
+        start,end=chunk.offset,chunk.offset+chunk.content.size
+        rows=self.rows("SELECT offset,length,body FROM chunks WHERE transfer_id=? AND offset<? AND offset+length>?",
+                       (chunk.transfer_id,end,start))
+        if rows:
+            require(len(rows)==1 and rows[0][0]==start and rows[0][1]==chunk.content.size and
+                    decode(TransferChunk,loads(rows[0][2])).content==chunk.content,"chunk_conflict")
+            return
+        self.execute("INSERT INTO chunks VALUES (?,?,?,?)",(chunk.transfer_id,start,chunk.content.size,
+                     canonical(chunk).decode()),write=True)
+
+    async def missing_ranges(self,id,cursor=None,limit=50):
+        session=await self.transfer(id)
+        require(session.expected_size is not None,"size_required")
+        require(1<=limit<=500,"invalid_limit")
+        start=int(cursor or 0)
+        require(0<=start<=session.expected_size,"invalid_cursor")
+        gaps=[]
+        for row in self.execute("SELECT offset,length FROM chunks WHERE transfer_id=? AND offset+length>? ORDER BY offset",(id,start)):
+            if row[0]>start:
+                gaps.append((start,row[0]))
+                if len(gaps)>limit:
+                    return Page(items=tuple(gaps[:limit]),next_cursor=str(gaps[limit][0]))
+            start=max(start,row[0]+row[1])
+        if start<session.expected_size:
+            gaps.append((start,session.expected_size))
+        return Page(items=tuple(gaps[:limit]),next_cursor=str(gaps[limit][0]) if len(gaps)>limit else None)
+
+    async def append_event(self,event):
+        self.execute("INSERT INTO events (id,body) VALUES (?,?)",(event.id,canonical(event).decode()),write=True)
+
+    async def append_audit(self,event):
+        previous=self.one("SELECT digest FROM audit ORDER BY seq DESC LIMIT 1")
+        prev=previous[0] if previous else None
+        data=wire(replace(event,previous_digest=prev,entry_digest=''))
+        data.pop('entry_digest')
+        entry=replace(event,previous_digest=prev,entry_digest=digest(data))
+        self.execute("INSERT INTO audit (digest,previous,body) VALUES (?,?,?)",
+                     (entry.entry_digest,prev,canonical(entry).decode()),write=True)
+
+    async def enqueue(self,job):
+        self.execute("INSERT INTO jobs VALUES (?,?,?,?,?,?)",(job.id,job.dedupe_key,job.kind,
+            job.state,wire(job.next_attempt_at),canonical(job).decode()),write=True)
+
+    async def job(self,id):
+        row=self.one("SELECT body FROM jobs WHERE id=?",(id,))
+        require(row is not None,'job_not_found')
+        return decode(EffectJob,loads(row[0]))
+
+    async def save_job(self,job):
+        self.execute("UPDATE jobs SET state=?,next_at=?,body=? WHERE id=?",(job.state,
+            wire(job.next_attempt_at),canonical(job).decode(),job.id),write=True)
+
+    def setting(self, key: str, default: SettingValue = None) -> SettingValue:
+        row=self.one("SELECT value FROM settings WHERE key=?",(key,))
+        return loads(row[0]) if row else default
+
+    def set_setting(self, key: str, value: SettingValue) -> None:
+        self.execute("INSERT INTO settings VALUES (?,?) ON CONFLICT(key) DO UPDATE SET value=excluded.value",
+                     (key,canonical(value).decode()),write=True)
diff --git a/src/msg/storage/sqlite.py b/src/msg/storage/sqlite.py
index 880def3d..2ec62cdf 100644
--- a/src/msg/storage/sqlite.py
+++ b/src/msg/storage/sqlite.py
@@ -9,16 +9,12 @@
 import tempfile
 import uuid
 from contextlib import asynccontextmanager
-from dataclasses import replace
 from pathlib import Path
 
-from msg.core.codec import canonical, decode, digest, loads, wire
 from msg.core.errors import Failure, require
-from msg.core.models import (
-    AuditEvent, Certificate, CertificateRequest, CertificateRequestState, Credential,
-    EffectJob, EmailSettings, Event, Membership, OperationResult, Organization, Page,
-    Resource, ResourceRef, Revision, Subject, TransferChunk, TransferSession,
-)
+from msg.core.query import QueryResult, SqlParameters
+from msg.storage.query import SessionQueryResult
+from msg.storage.session import RelationalSession
 
 _SCHEMA = """
 PRAGMA foreign_keys=ON;
@@ -219,389 +215,19 @@
 """
 
 
-class SqliteSession:
+class SqliteSession(RelationalSession):
     def __init__(self, connection, *, write: bool):
-        self._connection, self.write = connection, write
-        self.owner_task = asyncio.current_task()
-        self.closed = False
-        self.rollback_effects = []
+        super().__init__(write=write)
+        self._connection = connection
 
-    def on_rollback(self, effect):
-        """Compensate external work if this transaction or savepoint aborts."""
-        self.check(write=True)
-        self.rollback_effects.append(effect)
-
-    async def run_rollback_effects(self, cause, start=0):
-        effects = self.rollback_effects[start:]
-        del self.rollback_effects[start:]
-        if not effects:
-            return
-
-        async def compensate():
-            for effect in reversed(effects):
-                try:
-                    await effect()
-                except BaseException as exc:
-                    # Keep the primary error and try every independent cleanup.
-                    cause.add_note('rollback compensation failed: ' + type(exc).__name__)
-
-        # A second cancellation must not abandon a pin operation (including its
-        # worker thread) and release the writer fence before it finishes.
-        pending = asyncio.create_task(compensate())
-        while not pending.done():
-            try:
-                await asyncio.shield(pending)
-            except asyncio.CancelledError:
-                pass
-        pending.result()
-
-    def check(self, write=False):
-        require(not self.closed, "transaction_closed")
-        require(asyncio.current_task() is self.owner_task, "transaction_cross_task")
-        if write:
-            require(self.write, "read_only_transaction")
-
-    def execute(self, sql, parameters=(), *, write=False):
+    def execute(self, sql: str, parameters: SqlParameters = (), *,
+                write: bool = False) -> QueryResult:
         self.check(write)
         try:
-            return self._connection.execute(sql, parameters)
+            return SessionQueryResult(self._connection.execute(sql, parameters), self.check)
         except sqlite3.IntegrityError as exc:
             raise Failure("constraint_conflict") from exc
 
-    def one(self, sql, parameters=()):
-        return self.execute(sql, parameters).fetchone()
-
-    def rows(self, sql, parameters=()):
-        return self.execute(sql, parameters).fetchall()
-
-    async def resource(self, id):
-        row = self.one("SELECT body FROM resources WHERE id=?", (id,))
-        require(row is not None, "not_found")
-        return decode(Resource, loads(row[0]))
-
-    async def resolve(self, path):
-        require(isinstance(path, str) and path.startswith("/"), "invalid_path")
-        parts = path.rstrip("/").split("/")[1:]
-        require(all(p not in {".", ".."} and "\\" not in p and "\x00" not in p for p in parts), "invalid_path")
-        if len(parts) == 2 and parts[0] == "_id":
-            return (await self.resource(parts[1])).id
-        row = self.one("SELECT id FROM resources WHERE parent IS NULL")
-        require(row is not None, "not_initialized")
-        rid = row[0]
-        if path == "/":
-            return rid
-        for part in parts:
-            require(bool(part), "invalid_path")
-            row = self.one("SELECT id FROM resources WHERE parent=? AND name=?", (rid, part))
-            require(row is not None, "not_found")
-            rid = row[0]
-        return rid
-
-    async def path(self, id):
-        segments, seen = [], set()
-        r = await self.resource(id)
-        while r.parent is not None:
-            require(r.id not in seen, "parent_cycle")
-            seen.add(r.id)
-            segments.append(r.name)
-            r = await self.resource(r.parent)
-        return "/" + "/".join(reversed(segments))
-
-    async def ancestors(self, id):
-        result, seen = [], set()
-        r = await self.resource(id)
-        while r.parent is not None:
-            require(r.id not in seen, "parent_cycle")
-            seen.add(r.id)
-            r = await self.resource(r.parent)
-            result.append(r)
-        return tuple(reversed(result))
-
-    async def children(self, parent, cursor=None, limit=50):
-        require(type(limit) is int and 1 <= limit <= 500, "invalid_limit")
-        rows = self.rows("SELECT body FROM resources WHERE parent=? AND id>? ORDER BY id LIMIT ?",
-                         (parent, cursor or "", limit + 1))
-        values = tuple(decode(Resource, loads(row[0])) for row in rows[:limit])
-        return Page(items=values, next_cursor=values[-1].id if len(rows) > limit else None)
-
-    async def revision(self, ref):
-        rid = ref.revision or (await self.resource(ref.id)).revision
-        row = self.one("SELECT body FROM revisions WHERE id=? AND resource_id=?", (rid, ref.id))
-        require(row is not None, "revision_not_found")
-        return decode(Revision, loads(row[0]))
-
-    async def history(self, id, cursor=None, limit=50):
-        require(1 <= limit <= 500, "invalid_limit")
-        rows = self.rows("SELECT body FROM revisions WHERE resource_id=? AND id>? ORDER BY id LIMIT ?",
-                         (id, cursor or "", limit+1))
-        values = tuple(decode(Revision, loads(r[0])) for r in rows[:limit])
-        return Page(items=values, next_cursor=values[-1].id if len(rows)>limit else None)
-
-    async def insert(self, resource):
-        self.check(True)
-        if resource.parent is not None:
-            await self.resource(resource.parent)
-        data = wire(resource)
-        self.execute("INSERT INTO resources VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)", (
-            resource.id, resource.type, resource.name, resource.parent, resource.owner,
-            resource.group, resource.mode, resource.generation, resource.revision,
-            resource.state, data['created_at'], data['modified_at'], canonical(resource).decode()), write=True)
-        for tag in resource.tags:
-            self.execute("INSERT INTO resource_tags VALUES (?,?)", (resource.id,tag), write=True)
-
-    async def replace(self, resource, expected_generation):
-        self.check(True)
-        old = await self.resource(resource.id)
-        require(old.generation == expected_generation, "generation_conflict",
-                details={"generation":old.generation, "revision":old.revision})
-        require(resource.generation == expected_generation + 1, "invalid_generation")
-        require(resource.type == old.type and resource.created_at == old.created_at and
-                resource.created_by == old.created_by, "immutable_creation_fact")
-        if resource.parent is not None:
-            require(resource.parent != resource.id, "parent_cycle")
-            parent = await self.resource(resource.parent)
-            require(all(a.id != resource.id for a in await self.ancestors(parent.id)), "parent_cycle")
-        if any(getattr(resource,k)!=getattr(old,k) for k in ('parent','owner','group','mode','state')):
-            self.set_setting('authorization_epoch',self.setting('authorization_epoch',0)+1)
-        data = wire(resource)
-        changed = self.execute("""UPDATE resources SET name=?,parent=?,owner=?,grp=?,mode=?,
-            generation=?,revision=?,state=?,modified_at=?,body=? WHERE id=? AND generation=?""",
-            (resource.name, resource.parent, resource.owner, resource.group, resource.mode,
-             resource.generation, resource.revision, resource.state, data['modified_at'],
-             canonical(resource).decode(), resource.id, expected_generation), write=True).rowcount
-        require(changed == 1, "generation_conflict")
-        if resource.tags != old.tags:
-            self.execute("DELETE FROM resource_tags WHERE resource_id=?", (resource.id,), write=True)
-            for tag in resource.tags:
-                self.execute("INSERT INTO resource_tags VALUES (?,?)", (resource.id,tag), write=True)
-
-    async def append_revision(self, revision):
-        self.check(True)
-        for parent in revision.parents:
-            await self.revision(ResourceRef(id=revision.resource_id, revision=parent))
-        self.execute("INSERT INTO revisions VALUES (?,?,?,?)", (
-            revision.id, revision.resource_id, wire(revision.created_at), canonical(revision).decode()), write=True)
-        for relation in revision.relations:
-            self.execute("INSERT INTO relations VALUES (?,?,?,?,?,?)", (
-                revision.id, revision.resource_id, relation.type, relation.target.id,
-                relation.target.revision, canonical(relation).decode()), write=True)
-
-    async def subject(self, id):
-        row = self.one("SELECT body FROM identities WHERE id=? AND kind='subject'", (id,))
-        require(row is not None, "subject_not_found")
-        return decode(Subject, loads(row[0]))
-
-    async def organization(self, id):
-        row = self.one("SELECT body FROM identities WHERE id=? AND kind='organization'", (id,))
-        require(row is not None, "group_not_found")
-        return decode(Organization, loads(row[0]))
-
-    async def credential(self, id):
-        row = self.one("SELECT body FROM credentials WHERE id=?", (id,))
-        require(row is not None, "credential_not_found")
-        return decode(Credential, loads(row[0]))
-
-    async def certificate(self, id):
-        row = self.one("SELECT body FROM certificates WHERE id=?", (id,))
-        require(row is not None, "certificate_not_found")
-        return decode(Certificate, loads(row[0]))
-
-    async def certificate_revoked(self, id):
-        row = self.one("SELECT revoked FROM certificates WHERE id=?", (id,))
-        require(row is not None, "certificate_not_found")
-        return bool(row[0])
-
-    async def memberships(self, subject):
-        if subject is None:
-            return ()
-        rows=self.rows("""SELECT m.org,m.body FROM memberships m
-            JOIN resources r ON r.id=m.org
-            WHERE m.subject=? AND r.type='organization' AND r.state='active'
-            ORDER BY m.org""",(subject,))
-        memberships=((org,decode(Membership,loads(body))) for org,body in rows)
-        # A stale membership row must not revive an archived/missing organization
-        # or borrow an organization/subject identifier from a mismatched body.
-        active=tuple(member for org,member in memberships if member.status=='active'
-                     and member.organization_id==org and member.subject_id==subject
-                     and org!='g_public')
-        row=self.one("SELECT body FROM identities WHERE id=? AND kind='subject'",(subject,))
-        if row is None:
-            return active
-        identity=decode(Subject,loads(row[0]))
-        if identity.kind in {'registered','custodial','system'} and not identity.local_only:
-            # /&public is a virtual group for formal subjects. Historical rows
-            # may remain in storage, but never create a duplicate authorization.
-            return (Membership(organization_id='g_public',subject_id=subject,
-                               role='member',version=0),*active)
-        return active
-
-    async def request_result(self, subject, id, digest):
-        row = self.one("SELECT digest,body FROM results WHERE subject=? AND request_id=?", (subject,id))
-        if row is None:
-            return None
-        require(row[0] == digest, "idempotency_conflict")
-        return decode(OperationResult, loads(row[1]))
-
-    async def save_result(self, subject, digest, result):
-        from msg.storage.capacity import require_result_capacity
-        require_result_capacity(self)
-        self.execute("INSERT INTO results VALUES (?,?,?,?)", (
-            subject, result.request_id, digest, canonical(result).decode()), write=True)
-
-    async def update_identity(self, record, expected_generation):
-        self.check(True)
-        if isinstance(record, Membership):
-            self.set_setting('authorization_epoch',self.setting('authorization_epoch',0)+1)
-            table, where, args = 'memberships', 'org=? AND subject=?', (record.organization_id,record.subject_id)
-            generation = record.version
-        elif isinstance(record, EmailSettings):
-            table, where, args = 'emails', 'subject=?', (record.subject_id,)
-            generation = expected_generation + 1
-        else:
-            table, where, args = 'identities', 'id=?', (record.resource_id,)
-            generation = record.auth_version if isinstance(record, Subject) else record.membership_version
-        row = self.one(f"SELECT generation FROM {table} WHERE {where}", args)
-        require((row is None and expected_generation == -1) or
-                (row is not None and row[0] == expected_generation), "generation_conflict")
-        payload = canonical(record).decode()
-        if row is None:
-            if table == 'identities':
-                self.execute("INSERT INTO identities VALUES (?,?,?,?)", (*args,
-                    'subject' if isinstance(record,Subject) else 'organization', generation,payload), write=True)
-            elif table == 'memberships':
-                self.execute("INSERT INTO memberships VALUES (?,?,?,?)", (*args,generation,payload), write=True)
-            else:
-                self.execute("INSERT INTO emails VALUES (?,?,?)", (*args,generation,payload), write=True)
-        else:
-            self.execute(f"UPDATE {table} SET generation=?,body=? WHERE {where}",
-                         (generation,payload,*args), write=True)
-
-    async def save_credential(self, credential, expected_auth_version):
-        subject = await self.subject(credential.subject_id)
-        require(subject.auth_version == expected_auth_version, "auth_version_conflict")
-        old=self.one("SELECT body FROM credentials WHERE id=?",(credential.id,))
-        if old is not None:
-            previous=decode(Credential,loads(old[0]))
-            require(previous.subject_id==credential.subject_id and previous.kind==credential.kind and
-                    previous.verifier==credential.verifier,"credential_identity_immutable")
-        self.execute("INSERT INTO credentials VALUES (?,?,?) ON CONFLICT(id) DO UPDATE SET body=excluded.body",
-                     (credential.id,credential.subject_id,canonical(credential).decode()),write=True)
-
-    async def csr(self, id):
-        row = self.one("SELECT body FROM csrs WHERE id=?",(id,))
-        require(row is not None,"csr_not_found")
-        return decode(CertificateRequest,loads(row[0]))
-
-    async def csr_state(self, id):
-        row=self.one("SELECT state_body FROM csrs WHERE id=?",(id,))
-        require(row is not None,"csr_not_found")
-        return decode(CertificateRequestState,loads(row[0]))
-
-    async def save_csr(self, request):
-        state=CertificateRequestState(request_id=request.resource_id,status='pending',generation=0)
-        self.execute("INSERT INTO csrs VALUES (?,?,?,?,?)",(request.resource_id,0,'pending',
-            canonical(request).decode(),canonical(state).decode()),write=True)
-
-    async def transition_csr(self, state, expected_generation):
-        require(state.generation == expected_generation+1,"invalid_generation")
-        old=await self.csr_state(state.request_id)
-        require(old.status=='pending',"csr_not_pending")
-        require(old.generation==expected_generation,"generation_conflict")
-        self.execute("UPDATE csrs SET generation=?,state=?,state_body=? WHERE id=? AND generation=?",
-            (state.generation,state.status,canonical(state).decode(),state.request_id,expected_generation),write=True)
-
-    async def register_certificate(self, certificate, csr_id, expected_generation):
-        self.execute("INSERT INTO certificates VALUES (?,?,?,?,?)",(
-            certificate.resource_id,certificate.subject_id,certificate.parent_certificate_id,0,
-            canonical(certificate).decode()),write=True)
-        if csr_id is not None:
-            self.set_setting('certificate_request:'+certificate.resource_id,csr_id)
-            await self.transition_csr(CertificateRequestState(request_id=csr_id,status='issued',
-                generation=expected_generation+1,certificate_id=certificate.resource_id),expected_generation)
-
-    async def revoke_certificate(self, id, event):
-        await self.certificate(id)
-        self.execute("UPDATE certificates SET revoked=1 WHERE id=?",(id,),write=True)
-        self.set_setting('authorization_epoch',self.setting('authorization_epoch',0)+1)
-        await self.append_audit(event)
-
-    async def transfer(self,id):
-        row=self.one("SELECT body FROM transfers WHERE id=?",(id,))
-        require(row is not None,"transfer_not_found")
-        return decode(TransferSession,loads(row[0]))
-
-    async def save_transfer(self,session,expected_generation):
-        if expected_generation is None:
-            self.execute("INSERT INTO transfers (id,subject,generation,body) VALUES (?,?,?,?)",
-                         (session.id,session.subject_id,session.generation,canonical(session).decode()),write=True)
-        else:
-            require(session.generation==expected_generation+1,"invalid_generation")
-            count=self.execute("UPDATE transfers SET generation=?,body=? WHERE id=? AND generation=?",
-                (session.generation,canonical(session).decode(),session.id,expected_generation),write=True).rowcount
-            require(count==1,"generation_conflict")
-
-    async def put_chunk(self,chunk):
-        self.check(True)
-        start,end=chunk.offset,chunk.offset+chunk.content.size
-        rows=self.rows("SELECT offset,length,body FROM chunks WHERE transfer_id=? AND offset<? AND offset+length>?",
-                       (chunk.transfer_id,end,start))
-        if rows:
-            require(len(rows)==1 and rows[0][0]==start and rows[0][1]==chunk.content.size and
-                    decode(TransferChunk,loads(rows[0][2])).content==chunk.content,"chunk_conflict")
-            return
-        self.execute("INSERT INTO chunks VALUES (?,?,?,?)",(chunk.transfer_id,start,chunk.content.size,
-                     canonical(chunk).decode()),write=True)
-
-    async def missing_ranges(self,id,cursor=None,limit=50):
-        session=await self.transfer(id)
-        require(session.expected_size is not None,"size_required")
-        require(1<=limit<=500,"invalid_limit")
-        start=int(cursor or 0)
-        require(0<=start<=session.expected_size,"invalid_cursor")
-        gaps=[]
-        for row in self.execute("SELECT offset,length FROM chunks WHERE transfer_id=? AND offset+length>? ORDER BY offset",(id,start)):
-            if row[0]>start:
-                gaps.append((start,row[0]))
-                if len(gaps)>limit:
-                    return Page(items=tuple(gaps[:limit]),next_cursor=str(gaps[limit][0]))
-            start=max(start,row[0]+row[1])
-        if start<session.expected_size:
-            gaps.append((start,session.expected_size))
-        return Page(items=tuple(gaps[:limit]),next_cursor=str(gaps[limit][0]) if len(gaps)>limit else None)
-
-    async def append_event(self,event):
-        self.execute("INSERT INTO events (id,body) VALUES (?,?)",(event.id,canonical(event).decode()),write=True)
-
-    async def append_audit(self,event):
-        previous=self.one("SELECT digest FROM audit ORDER BY seq DESC LIMIT 1")
-        prev=previous[0] if previous else None
-        data=wire(replace(event,previous_digest=prev,entry_digest=''))
-        data.pop('entry_digest')
-        entry=replace(event,previous_digest=prev,entry_digest=digest(data))
-        self.execute("INSERT INTO audit (digest,previous,body) VALUES (?,?,?)",
-                     (entry.entry_digest,prev,canonical(entry).decode()),write=True)
-
-    async def enqueue(self,job):
-        self.execute("INSERT INTO jobs VALUES (?,?,?,?,?,?)",(job.id,job.dedupe_key,job.kind,
-            job.state,wire(job.next_attempt_at),canonical(job).decode()),write=True)
-
-    async def job(self,id):
-        row=self.one("SELECT body FROM jobs WHERE id=?",(id,))
-        require(row is not None,'job_not_found')
-        return decode(EffectJob,loads(row[0]))
-
-    async def save_job(self,job):
-        self.execute("UPDATE jobs SET state=?,next_at=?,body=? WHERE id=?",(job.state,
-            wire(job.next_attempt_at),canonical(job).decode(),job.id),write=True)
-
-    def setting(self,key,default=None):
-        row=self.one("SELECT value FROM settings WHERE key=?",(key,))
-        return loads(row[0]) if row else default
-
-    def set_setting(self,key,value):
-        self.execute("INSERT INTO settings VALUES (?,?) ON CONFLICT(key) DO UPDATE SET value=excluded.value",
-                     (key,canonical(value).decode()),write=True)
-
 
 class SqliteMetadataStore:
     def __init__(self,path: Path, *, busy_timeout: float=10):
diff --git a/src/msg/transports/packet.py b/src/msg/transports/packet.py
index 84dc47ea..dccaff9b 100644
--- a/src/msg/transports/packet.py
+++ b/src/msg/transports/packet.py
@@ -1,38 +1,15 @@
-"""Bounded wire decoding shared by every network adapter."""
+"""URL and compression boundaries; common wire contracts live in core.packet."""
 from __future__ import annotations
-from collections.abc import Mapping
-import re
 import zlib
-from msg.core.codec import decode,loads,unb64,canonical,wire
-from msg.core.errors import require,Failure
+from msg.core.codec import unb64
+from msg.core.errors import require, Failure
 from msg.core.requests import SECRET_DELIVERY_MIN_VERSION
-from msg.core.models import OperationRequest,OperationResult,TokenProof
-from msg.plugins.schemas import obj,IDENTIFIER,STRING,BYTES,REF,SIGNATURE
+from msg.core.models import OperationRequest, TokenProof
+from msg.core.packet import (
+    REQUEST_SCHEMA, RESULT_SCHEMA, decode_packet, decode_result, safe_error_code,
+)
 from msg.transports.url_safety import contains_secret_fields
 
-REQUEST_SCHEMA=obj({
-    'request_id':{'type':'string','minLength':1,'maxLength':128},'protocol_version':{'const':1},
-    'operation':{'type':'string','pattern':'^[a-z][a-z0-9_]*(\\.[a-z][a-z0-9_]*)+$','maxLength':128},
-    'contract_version':{'type':'integer','minimum':1},'target_service':{'type':'string','maxLength':2048},
-    'subject':{'type':['string','null'],'maxLength':160},'arguments':{'type':'object'},
-    'expected_generations':{'type':'array','maxItems':64,'items':{'type':'array','prefixItems':[IDENTIFIER,{'type':'integer','minimum':0}],
-                                                                 'minItems':2,'maxItems':2}},
-    'expires_at':{'type':['string','null'],'maxLength':40},
-    'payload_digest':{'type':'string','pattern':'^sha256:[0-9a-f]{64}$'},
-    'proof':{'anyOf':[{'type':'null'},obj({'signature':SIGNATURE,'certificates':{'type':'array','items':IDENTIFIER,'maxItems':16}},('signature','certificates')),
-                    obj({'credential_id':IDENTIFIER,'token':BYTES},('credential_id','token'))]},
-    'return_fields':{'type':'array','maxItems':30,'items':STRING},'source':{'enum':['msg','manual','mcp','unknown']}
-},('request_id','protocol_version','operation','contract_version','target_service','subject','arguments',
-    'expected_generations','expires_at','payload_digest','proof'))
-RESULT_SCHEMA={'type':'object','properties':{
-    'request_id':STRING,'operation':STRING,'status':{'enum':['ok','accepted','error','uncertain']},
-    'actor':STRING,'subject':STRING,'resources':{'type':'array','items':REF},'committed_at':STRING,
-    'replayed':{'type':'boolean'},'receipt':SIGNATURE,'error':{'type':'object'},'data':{'type':'object'},
-    'output':REF,'prefer_cli':{'type':'boolean'},'cli_url':STRING},
-    'required':['request_id','operation','status'],'additionalProperties':False}
-
-
-
 
 def require_url_safe_packet(packet: OperationRequest) -> None:
     """Reject credentials and recovery claims before a GET URL is dispatched.
@@ -47,40 +24,6 @@ def require_url_safe_packet(packet: OperationRequest) -> None:
     require(not contains_secret_fields(packet.arguments), 'secure_channel_required')
 
 
-def decode_packet(value,max_bytes=1048576):
-    from jsonschema import Draft202012Validator
-    if isinstance(value,(bytes,str)):
-        require(len(value.encode() if isinstance(value,str) else value)<=max_bytes,'request_too_large')
-        value=loads(value)
-    value=wire(value)
-    require(len(canonical(value))<=max_bytes,'request_too_large')
-    error=next(Draft202012Validator(REQUEST_SCHEMA).iter_errors(value),None)
-    require(error is None,'invalid_request_envelope', '.'.join(map(str,error.path)) if error else None)
-    packet=decode(OperationRequest,value)
-    require(len(set(rid for rid,generation in packet.expected_generations))==len(packet.expected_generations),
-            'duplicate_expected_generation')
-    return packet
-
-
-def safe_error_code(value, fallback):
-    """Only bounded contract identifiers, never exception text or credential URLs."""
-    return (value if isinstance(value, str) and
-            re.fullmatch(r'[a-z][a-z0-9_]{0,95}', value) else fallback)
-
-
-def decode_result(value):
-    # Reject malformed signed/unsigned results rather than rewriting their error
-    # fields. Valid receipts and persisted result bytes are left unchanged.
-    require(isinstance(value, Mapping), 'invalid_result_envelope')
-    error = value.get('error')
-    if error is not None:
-        require(isinstance(error, Mapping) and safe_error_code(error.get('code'), None) is not None,
-                'invalid_result_envelope')
-    # Compact results omit anonymous actor/subject; these required domain fields
-    # are restored explicitly, not inferred from a transport connection.
-    return decode(OperationResult,dict({'actor':None,'subject':None},**value))
-
-
 def gunzip(raw,limit):
     inflater=zlib.decompressobj(wbits=31)
     try:
diff --git a/tests/issue_78_80/conftest.py b/tests/issue_78_80/conftest.py
index ada0f4a7..578e5576 100644
--- a/tests/issue_78_80/conftest.py
+++ b/tests/issue_78_80/conftest.py
@@ -125,9 +125,14 @@ async def invoke(operation,args,version=1, *, expected=(),context=None):
     app.authorizer=AuthorizationService(registry,app.certificates)
     app.authenticator=AuthenticationService(registry,app.certificates,app.settings.service_url,
         app.clock,lambda:base_grants(registry),lambda:base_grants(registry))
+    async def result_projection(context, request, session, resource, *, fields):
+        return await discovery.read_projection(app,context,request,session,resource.id,
+                                               revision=resource.revision,fields=fields)
+
     app.executor=OperationExecutor(registry,metadata,app.contents,app.authenticator,
-        app.authorizer,app.clock,Ed25519Signer.generate())
-    app.executor.application=app
+        app.authorizer,app.clock,Ed25519Signer.generate(),
+        max_request_bytes=app.settings.server.limits.max_request_bytes,
+        result_projection=result_projection)
     app._loaded=True
 
     def packet(operation,args,version=1, *, signed=True,**kwargs):
diff --git a/tests/test_architecture_ports.py b/tests/test_architecture_ports.py
new file mode 100644
index 00000000..38fc27e3
--- /dev/null
+++ b/tests/test_architecture_ports.py
@@ -0,0 +1,129 @@
+"""Core/adapter boundaries plus real signed calls through the composed ports."""
+import ast
+import importlib
+import importlib.util
+import inspect
+from datetime import timedelta
+from pathlib import Path
+import subprocess
+import sys
+
+import pytest
+
+from msg.core.codec import digest, wire
+from msg.core.executor import OperationExecutor
+from msg.core.requests import request_for
+from test_service import NOW, register
+
+
+ROOT = Path(__file__).resolve().parents[1] / 'src' / 'msg'
+
+
+def imports(path):
+    tree = ast.parse(path.read_text())
+    return [node.module for node in ast.walk(tree)
+            if isinstance(node, ast.ImportFrom) and node.module]
+
+
+def test_executor_has_no_plugin_or_application_backdoor():
+    source = (ROOT / 'core/executor.py').read_text()
+    assert not any(name.startswith(('msg.plugins', 'msg.application'))
+                   for name in imports(ROOT / 'core/executor.py'))
+    assert not any(isinstance(node, ast.Attribute) and node.attr == 'application'
+                   for node in ast.walk(ast.parse(source)))
+
+
+def test_batch_rules_and_event_identity_have_one_neutral_owner():
+    assert importlib.util.find_spec('msg.core.batching') is not None
+    assert importlib.util.find_spec('msg.core.events') is not None
+    batching = importlib.import_module('msg.core.batching')
+    events = importlib.import_module('msg.core.events')
+    from msg.plugins import batch, communication
+    assert batch.packets is batching.packets
+    assert communication.event_id is events.event_id
+    assert not any(name.startswith('msg.plugins') for name in imports(ROOT / 'core/batching.py'))
+    request = request_for('content.post_create', {}, 'https://example.invalid',
+                          subject='u_test', request_id='stable-request')
+    assert events.event_id(request, 'u_test') == 'e_' + digest(('u_test', 'stable-request'))[7:39]
+
+
+def test_core_protocol_imports_without_loading_plugins_or_transports():
+    code = """
+import importlib.abc
+import sys
+class NoOuterLayer(importlib.abc.MetaPathFinder):
+    def find_spec(self, fullname, path=None, target=None):
+        if fullname.startswith(('msg.plugins', 'msg.transports', 'msg.application')):
+            raise AssertionError('outer layer imported: ' + fullname)
+sys.meta_path.insert(0, NoOuterLayer())
+from msg.core import executor, batching, packet
+assert executor.result_wire is packet.result_wire
+"""
+    result = subprocess.run([sys.executable, '-c', code], text=True, capture_output=True)
+    assert result.returncode == 0, result.stderr
+
+
+def test_wire_contracts_and_primitive_schemas_are_single_source():
+    assert importlib.util.find_spec('msg.core.packet') is not None
+    assert importlib.util.find_spec('msg.core.schemas') is not None
+    packet = importlib.import_module('msg.core.packet')
+    schemas = importlib.import_module('msg.core.schemas')
+    from msg.transports import packet as transport
+    from msg.plugins import schemas as plugin_schemas
+    for name in ('REQUEST_SCHEMA', 'RESULT_SCHEMA', 'decode_packet', 'decode_result'):
+        assert getattr(packet, name) is getattr(transport, name)
+    assert schemas.obj is plugin_schemas.obj
+    assert schemas.SIGNATURE is plugin_schemas.SIGNATURE
+
+
+@pytest.mark.asyncio
+async def test_composition_reuses_ports_for_alternate_authenticator(installed):
+    app, _ = installed
+    assert hasattr(app, 'new_executor')
+    executor = app.new_executor(app.authenticator)
+    assert not hasattr(executor, 'application')
+    assert executor.max_request_bytes == app.settings.server.limits.max_request_bytes
+    assert executor.result_projection == app.executor.result_projection
+    assert executor.event_notifications == app.executor.event_notifications
+    assert executor.response_hook == app.executor.response_hook
+    assert executor.recovery_drill_marker == app.executor.recovery_drill_marker
+    app.executor.recovery_quarantined = True
+    assert app.new_executor(app.authenticator).recovery_quarantined is True
+
+
+@pytest.mark.asyncio
+async def test_standalone_executor_projection_failure_rolls_back(installed):
+    app, _ = installed
+    key, uid, _ = await register(app, 'port-agent')
+    executor = OperationExecutor(app.registry, app.metadata, app.contents, app.authenticator,
+                                 app.authorizer, app.clock, app.receipt_signer)
+    packet = request_for('content.post_create', {'parent': '/main', 'body': 'not published'},
+        app.settings.service_url, subject=uid, signer=key, request_id='missing-projection-port',
+        return_fields=('id',), expires_at=NOW + timedelta(seconds=90))
+    result = await executor.execute(packet)
+    assert result.status == 'error' and result.error.code == 'projection_unavailable', wire(result)
+    async with app.metadata.transaction(write=False) as tx:
+        assert tx.one("SELECT COUNT(*) FROM resources WHERE type='post'")[0] == 0
+        assert tx.one('SELECT COUNT(*) FROM results WHERE request_id=?', (packet.request_id,))[0] == 0
+
+
+@pytest.mark.asyncio
+async def test_explicit_projection_port_preserves_signed_execution_and_replay(installed):
+    app, _ = installed
+    assert 'result_projection' in inspect.signature(OperationExecutor).parameters
+    key, uid, _ = await register(app, 'projected-agent')
+    executor = OperationExecutor(app.registry, app.metadata, app.contents, app.authenticator,
+        app.authorizer, app.clock, app.receipt_signer,
+        max_request_bytes=app.settings.server.limits.max_request_bytes,
+        result_projection=app.executor.result_projection,
+        event_notifications=app.executor.event_notifications)
+    packet = request_for('content.post_create', {'parent': '/main', 'body': 'projected'},
+        app.settings.service_url, subject=uid, signer=key, request_id='explicit-projection-port',
+        return_fields=('id',), expires_at=NOW + timedelta(seconds=90))
+    first = await executor.execute(packet)
+    replay = await executor.execute(packet)
+    assert first.status == replay.status == 'ok', wire(first)
+    assert first.data['projection'][0]['id'] == first.resources[0].id
+    assert replay.replayed and replay.resources == first.resources and replay.receipt == first.receipt
+    async with app.metadata.transaction(write=False) as tx:
+        assert tx.one("SELECT COUNT(*) FROM resources WHERE type='post'")[0] == 1
diff --git a/tests/test_execution_composition.py b/tests/test_execution_composition.py
new file mode 100644
index 00000000..fea56ccb
--- /dev/null
+++ b/tests/test_execution_composition.py
@@ -0,0 +1,106 @@
+"""Retain PR #169's useful assembly/rollback cases on the one executor port.
+
+Source cases: 3db973ab1784926eb4e545cabdf832e7faa0a4bc. No alternative
+BatchPolicy/decoder/factory is installed solely to satisfy a competing API.
+"""
+from contextlib import asynccontextmanager
+from datetime import timedelta
+from types import SimpleNamespace
+
+from msg.core.codec import wire
+from msg.core.errors import Failure
+from msg.core.executor import OperationExecutor
+from msg.core.models import HandlerOutput, Principal, ResourceRef
+from msg.core.requests import request_for
+from test_service import NOW, call, register
+from test_batch import packet
+
+
+async def test_core_executor_runs_with_explicit_services_and_no_application():
+    active = []
+    transaction = object()
+    principal = Principal(actor='u_test', subject='u_test', credential_id='k_test',
+        method='signature', certificates=(), ceiling=())
+
+    @asynccontextmanager
+    async def transaction_scope(*, write):
+        assert write is False
+        active.append(transaction)
+        try:
+            yield transaction
+        finally:
+            active.pop()
+
+    async def authenticate(request, tx, *, entry):
+        assert tx is transaction and entry == 'network'
+        return principal
+
+    async def requirements(request, tx):
+        return ()
+
+    async def authorize(context, request, checks, tx):
+        assert context.principal is principal and tx is transaction
+
+    async def handler(context, request, tx):
+        return HandlerOutput(resources=(ResourceRef(id='r_test'),), data={'original': True})
+
+    async def project(context, request, tx, ref, *, fields):
+        assert active == [transaction] and tx is transaction
+        assert request.return_fields == fields == ('id',)
+        return {'id': ref.id}
+
+    spec = SimpleNamespace(name='testing.read', version=1, entries={'network'}, effect='read',
+        input_schema=None, output_schema=None, requirements=requirements, handler=handler)
+    registry = SimpleNamespace(operation=lambda *args: spec, validate=lambda *args: None)
+    executor = OperationExecutor(registry, SimpleNamespace(transaction=transaction_scope), None,
+        SimpleNamespace(authenticate=authenticate), SimpleNamespace(require=authorize),
+        lambda: NOW, None, result_projection=project)
+    assert not hasattr(executor, 'application')
+    result = await executor.execute(request_for('testing.read', {}, 'https://unit.invalid',
+        return_fields=('id',)))
+    assert result.status == 'ok', result
+    assert wire(result.data) == {'original': True, 'projection': [{'id': 'r_test'}]}
+    assert active == []
+
+
+async def test_factory_batch_and_projection_work_without_application_backreference(installed):
+    app, _ = installed
+    key, subject, _ = await register(app, 'executor-factory')
+    app.executor = app.new_executor()
+    child = packet(app, key, subject, 'content.post_create',
+        {'parent': '/main', 'body': 'boundaries'}, 'factory-child')
+    result = await call(app, 'batch.atomic', {'requests': [child]},
+        key=key, subject=subject, rid='factory-parent')
+    assert result.status == 'ok', wire(result)
+    assert result.data['results'][0]['status'] == 'ok'
+    repeated = await call(app, 'batch.atomic', {'requests': [child]},
+        key=key, subject=subject, rid='factory-parent')
+    assert repeated.replayed
+    projected = await app.executor.execute(request_for('content.post_create',
+        {'parent': '/main', 'body': 'project'}, app.settings.service_url,
+        subject=subject, signer=key, expires_at=NOW + timedelta(seconds=60),
+        return_fields=('id', 'size')))
+    assert projected.status == 'ok', wire(projected)
+    assert projected.data['projection'][0]['size'] == 7
+
+
+async def test_event_projection_failure_rolls_back_the_existing_transaction(installed):
+    app, _ = installed
+    key, subject, _ = await register(app, 'event-rollback')
+    app.executor = app.new_executor()
+    seen = []
+
+    async def reject(tx, event):
+        assert tx.one('SELECT id FROM events WHERE id=?', (event.id,)) is not None
+        seen.append(event.id)
+        raise Failure('test_event_projection_abort')
+
+    app.executor.event_notifications = reject
+    result = await call(app, 'content.post_create', {'parent': '/main', 'body': 'not committed'},
+        key=key, subject=subject, rid='event-rollback-request')
+    assert result.error.code == 'test_event_projection_abort', wire(result)
+    assert len(seen) == 1
+    async with app.metadata.transaction(write=False) as tx:
+        assert tx.one("SELECT COUNT(*) FROM resources WHERE type='post'")[0] == 0
+        assert tx.one('SELECT id FROM events WHERE id=?', (seen[0],)) is None
+        assert tx.one('SELECT request_id FROM results WHERE request_id=?', ('event-rollback-request',)) is None
diff --git a/tests/test_metadata_session_boundary.py b/tests/test_metadata_session_boundary.py
new file mode 100644
index 00000000..725be442
--- /dev/null
+++ b/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%')
diff --git a/tests/test_session_boundary.py b/tests/test_session_boundary.py
new file mode 100644
index 00000000..e6230fec
--- /dev/null
+++ b/tests/test_session_boundary.py
@@ -0,0 +1,129 @@
+"""The same bounded query and domain contract on real SQLite and PostgreSQL."""
+import asyncio
+import importlib
+import importlib.util
+from pathlib import Path
+import subprocess
+import sys
+from typing import get_type_hints
+
+import pytest
+
+from msg.core.codec import canonical, digest
+from msg.core.errors import Failure
+from msg.core.contracts import MetadataSession
+from msg.market.policy import DEFAULT_POLICY, load_policy
+from msg.storage.postgres import PostgresMetadataStore, PostgresSession, _SCHEMA as POSTGRES_SCHEMA
+from msg.storage.sqlite import SqliteMetadataStore, SqliteSession
+
+
+@pytest.fixture(params=['sqlite', 'postgres'])
+async def store(request, tmp_path):
+    value = (PostgresMetadataStore(request.getfixturevalue('pg_dsn'))
+             if request.param == 'postgres' else SqliteMetadataStore(tmp_path / 'metadata.db'))
+    if request.param == 'sqlite':
+        # SQLite is the small test backend, not a complete production bootstrap.
+        # Install the real policy table DDL for this real consumer contract test;
+        # do not pretend the test is a production schema migration or parity claim.
+        statement = next(sql for sql in POSTGRES_SCHEMA.split(';')
+            if sql.strip().startswith('CREATE TABLE IF NOT EXISTS arbitration_policies ('))
+        async with value.transaction(write=True) as tx:
+            tx.execute(statement, write=True)
+    yield value
+    await value.close()
+
+
+def test_sessions_are_siblings_and_common_base_does_not_own_a_driver():
+    assert not issubclass(PostgresSession, SqliteSession)
+    assert importlib.util.find_spec('msg.storage.session') is not None
+    common = importlib.import_module('msg.storage.session').RelationalSession
+    assert issubclass(PostgresSession, common) and issubclass(SqliteSession, common)
+    code = """
+import importlib.abc
+import sys
+class NoSQLite(importlib.abc.MetaPathFinder):
+    def find_spec(self, fullname, path=None, target=None):
+        if fullname in {'msg.storage.sqlite', 'sqlite3'}:
+            raise AssertionError('production storage imported ' + fullname)
+sys.meta_path.insert(0, NoSQLite())
+from msg.storage.postgres import PostgresSession
+"""
+    child = subprocess.run([sys.executable, '-c', code], capture_output=True, text=True)
+    assert child.returncode == 0, child.stderr
+    source = Path(importlib.import_module('msg.storage.session').__file__).read_text()
+    assert 'import sqlite3' not in source and 'import psycopg' not in source
+    assert '_connection' not in source
+
+
+def test_declared_session_covers_real_query_consumer_without_commit_rights():
+    assert importlib.util.find_spec('msg.core.query') is not None
+    query = importlib.import_module('msg.core.query')
+    assert get_type_hints(load_policy)['tx'] is query.QuerySession
+    for name in ('execute', 'one', 'rows', 'setting', 'set_setting'):
+        assert hasattr(MetadataSession, name), name
+        assert hasattr(query.QuerySession, name), name
+    common = importlib.import_module('msg.storage.session').RelationalSession
+    domain_methods = {name for name, member in vars(common).items()
+                      if callable(member) and not name.startswith('_')}
+    # These two methods are storage lifecycle mechanics, not consumer ports.
+    domain_methods -= {'check', 'run_rollback_effects'}
+    assert all(hasattr(MetadataSession, name) for name in domain_methods)
+    for name in ('connection', 'commit', 'rollback'):
+        assert not hasattr(MetadataSession, name)
+        assert not hasattr(query.QuerySession, name)
+
+
+async def test_query_result_is_a_guarded_view_not_a_driver_cursor(store):
+    async with store.transaction(write=False) as tx:
+        result = tx.execute('SELECT 1')
+        assert not hasattr(result, 'connection')
+        assert not hasattr(result, 'commit')
+        assert result.fetchone() == (1,)
+        assert result.fetchone() is None
+        assert list(tx.execute('SELECT 2 UNION ALL SELECT 3')) == [(2,), (3,)]
+        assert tx.rows('SELECT 4') == [(4,)]
+        assert tx.one('SELECT 5') == (5,)
+    with pytest.raises(Failure, match='transaction_closed'):
+        result.fetchall()
+
+
+async def test_query_result_and_session_refuse_cross_task_and_read_only_writes(store):
+    async with store.transaction(write=False) as tx:
+        result = tx.execute('SELECT 1')
+        async def misuse_result():
+            with pytest.raises(Failure, match='transaction_cross_task'):
+                result.fetchone()
+        async def misuse_session():
+            with pytest.raises(Failure, match='transaction_cross_task'):
+                tx.one('SELECT 1')
+        await asyncio.gather(misuse_result(), misuse_session())
+        with pytest.raises(Failure, match='read_only_transaction'):
+            tx.set_setting('forbidden', True)
+        assert result.fetchone() == (1,)
+
+
+async def test_declared_policy_consumer_and_nested_rollback_work_on_both_backends(store):
+    assert importlib.util.find_spec('msg.core.query') is not None
+    query = importlib.import_module('msg.core.query')
+    async with store.transaction(write=True) as tx:
+        assert isinstance(tx, query.QuerySession)
+        assert load_policy(tx, DEFAULT_POLICY['id']) == DEFAULT_POLICY
+        policy = {**DEFAULT_POLICY, 'id': 'query-contract-policy'}
+        tx.execute('INSERT INTO arbitration_policies VALUES (?,?,?)',
+                   (policy['id'], canonical(policy).decode(), digest(policy)), write=True)
+        assert load_policy(tx, policy['id']) == policy
+        tx.execute('INSERT INTO arbitration_policies VALUES (?,?,?)',
+                   ('corrupt-contract-policy', canonical(policy).decode(), 'wrong-digest'), write=True)
+        with pytest.raises(Failure, match='arbitration_policy_corrupt'):
+            load_policy(tx, 'corrupt-contract-policy')
+        tx.set_setting('outer', 'keep')
+        with pytest.raises(ValueError, match='abort inner'):
+            async with store.transaction(write=True) as child:
+                child.set_setting('inner', 'drop')
+                raise ValueError('abort inner')
+        assert tx.setting('inner') is None
+        tx.set_setting('after-inner', True)
+    async with store.transaction(write=False) as tx:
+        assert tx.setting('outer') == 'keep'
+        assert tx.setting('after-inner') is True
+        assert tx.setting('inner') is None