refactor: isolate execution kernel and tool runner contracts

.github/workflows/execution-boundaries.yml

@@ -0,0 +1,65 @@
+name: Execution boundary contracts
+on:
+  push:
+    branches: [fix/execution-boundaries-20260929]
+  workflow_dispatch:
+permissions:
+  contents: read
+concurrency:
+  group: execution-boundaries-${{ github.ref }}
+  cancel-in-progress: false
+jobs:
+  boundaries:
+    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 fixture dependencies
+        run: |
+          sudo apt-get update -qq
+          sudo apt-get install -y -qq age
+          python -m pip install -e '.[dev]'
+      - name: Record exact tested source
+        env:
+          EXPECTED_SHA: ${{ github.sha }}
+        run: |
+          mkdir -p artifacts
+          test "$(git rev-parse HEAD)" = "$EXPECTED_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['EXPECTED_SHA'], 'dirty':bool(git('status','--porcelain')),
+          }, indent=2)+'\n')
+          PY
+      - name: Run architecture and existing semantic regressions
+        run: |
+          python -m pytest tests/test_execution_boundaries.py tests/test_batch.py tests/test_batch_secret_release.py tests/test_tools.py tests/test_effect_completion_fencing.py tests/test_effect_lease_deadlines.py tests/test_effect_retry_state.py tests/test_webhook_domain_events.py tests/test_ssh_rss.py tests/test_credential_delivery_contract.py --junitxml=artifacts/execution-boundaries.xml
+      - uses: actions/upload-artifact@v4
+        if: always()
+        with:
+          name: execution-boundaries-${{ github.sha }}
+          path: artifacts/
+          if-no-files-found: error

docs/EXECUTION_BOUNDARY_PROGRESS.md

@@ -0,0 +1,36 @@
+# Execution boundary progress
+
+## Scope and coordination
+
+- Base: `476e64f04ed1d9c7dcab4a41c4f3cdb72faad6a7`, tree `40ca207a3c49be347120e058e47ed328ec4e3098`.
+- Branch: `fix/execution-boundaries-20260929`; coordination: #83 comment 5878058356.
+- Primary issues: #156 execution-kernel dependencies; #160 tool runner contract. Shared wire ownership is a slice of #158, not complete client packaging.
+- Do not modify #155's concurrent recovery branch. Do not touch production Root/PIN, deployment, funds, outbound delivery or recovery promotion.
+
+## Evidence ledger
+
+1. Main CI 36472507327 succeeded. Source inspected from artifact 10992168743; downloaded SHA-256 `9abbc69988b8ffcb45f9931dc87e291145a7680f275bf7aadbedb7bed2189d10` matches the provider digest. This is a baseline, not evidence for new code.
+2. This commit adds regression tests and a focused cloud workflow only. It is deliberately expected to fail against the unchanged implementation. No new test result is claimed yet.
+3. All project execution is delegated to GitHub Actions. The existing full four-shard/node-ID, conformance, packaging and recovery gates remain unchanged.
+
+## Dependency inventory before the change
+
+`core.executor -> plugins.batch -> transports.packet -> plugins.schemas`
+
+`core.executor -> plugins.discovery / plugins.communication`, plus the whole `Application` back-reference for limits, projections and webhook enqueueing.
+
+`workers.effects -> workers.sandbox -> workers.effects.ToolResult`; the declared `ToolExecutor.invoke(...) -> ResourceRef` is not the actual callable runner and conflates local output with worker publication.
+
+## Planned ownership
+
+Bounded request/result envelopes and batch parsing will have one neutral core owner, with compatibility re-exports. Request-derived event IDs keep identical canonical bytes. The executor receives explicit size, result-projection and transactional event-enqueue dependencies; no concrete plugin imports or Application service locator. These are fixed trusted collaborators, not configurable before/after workflows.
+
+A neutral `ToolRunner` callable and `ToolResult` own the real isolated-runner contract. The worker retains all current-authority, output-path, policy, attempt/state/deadline, quarantine and transactional-publication checks. The unused mismatched ToolExecutor declaration will be removed, not emulated by a second lifecycle.
+
+## Completion criteria
+
+- Obtain red cloud evidence before implementation.
+- Preserve all existing semantic tests and signed/shortcode/API versions.
+- Confirm production, SSH and isolated test assembly use explicit collaborators.
+- Run focused regressions and complete final-head CI; inspect artifacts and review before any merge.
+- Close only issues whose actual acceptance criteria are met. #158/#83/#85 and production-evidence blockers remain open unless separately established.

tests/test_execution_boundaries.py

@@ -0,0 +1,205 @@
+"""Architecture contracts plus real PostgreSQL assembly, not source-only claims."""
+from __future__ import annotations
+
+import ast
+from dataclasses import fields
+from datetime import timedelta
+import importlib
+import inspect
+from pathlib import Path
+import subprocess
+import sys
+from typing import get_type_hints
+
+import pytest
+
+from msg.core.codec import digest, wire
+from msg.core.errors import Failure
+from msg.core.models import OperationResult, ResourceRef
+from msg.core.requests import request_for, receipt_bytes
+from msg.security.crypto import verify
+from test_service import register, NOW
+
+
+def source(module):
+    return Path(importlib.import_module(module).__file__).read_text()
+
+
+def imports(text):
+    return [node.module or '' for node in ast.walk(ast.parse(text))
+            if isinstance(node, ast.ImportFrom)] + [name.name
+            for node in ast.walk(ast.parse(text)) if isinstance(node, ast.Import)
+            for name in node.names]
+
+
+def test_executor_has_no_concrete_plugin_or_application_imports():
+    dependencies = imports(source('msg.core.executor'))
+    assert not [name for name in dependencies if name.startswith(
+        ('msg.plugins', 'msg.application', 'msg.transports', 'msg.workers'))]
+
+
+def test_execution_core_imports_without_loading_server_layers():
+    code = '''
+import importlib.abc, sys
+class BlockServer(importlib.abc.MetaPathFinder):
+    def find_spec(self, fullname, path=None, target=None):
+        if any(fullname == name or fullname.startswith(name + '.') for name in
+               ('msg.plugins', 'msg.application', 'msg.storage', 'msg.admin',
+                'msg.workers', 'msg.transports')):
+            raise AssertionError('unexpected server dependency: ' + fullname)
+sys.meta_path.insert(0, BlockServer())
+from msg.core.executor import OperationExecutor
+from msg.core.batch import packets
+from msg.core.envelopes import decode_packet, result_wire
+from msg.core.tool_runner import ToolRunner, ToolResult
+'''
+    result = subprocess.run([sys.executable, '-c', code], capture_output=True, text=True)
+    assert result.returncode == 0, result.stderr
+
+
+def test_batch_and_adapter_share_the_canonical_decoder_and_schema():
+    batch = importlib.import_module('msg.core.batch')
+    envelopes = importlib.import_module('msg.core.envelopes')
+    adapter = importlib.import_module('msg.transports.packet')
+    plugin = importlib.import_module('msg.plugins.batch')
+    old_schemas = importlib.import_module('msg.plugins.schemas')
+    schemas = importlib.import_module('msg.core.schemas')
+    assert plugin.packets is batch.packets
+    assert batch.decode_packet is adapter.decode_packet is envelopes.decode_packet
+    assert adapter.REQUEST_SCHEMA is envelopes.REQUEST_SCHEMA
+    assert old_schemas.SIGNATURE is schemas.SIGNATURE
+    assert old_schemas.obj is schemas.obj
+
+
+def test_result_wire_is_one_compatibly_reexported_implementation():
+    envelopes = importlib.import_module('msg.core.envelopes')
+    executor = importlib.import_module('msg.core.executor')
+    assert executor.result_wire is envelopes.result_wire
+    result = OperationResult(request_id='wire-once', operation='discovery.get',
+                             status='ok', actor=None, subject=None)
+    expected = wire(result, compact=True)
+    expected.pop('replayed', None)
+    expected.pop('prefer_cli', None)
+    assert envelopes.result_wire(result) == expected
+
+
+def test_event_identity_preserves_existing_request_subject_bytes():
+    requests = importlib.import_module('msg.core.requests')
+    communication = importlib.import_module('msg.plugins.communication')
+    packet = request_for('content.post_create', {}, 'https://test.invalid',
+                         request_id='stable-event')
+    assert communication.event_id is requests.event_id
+    assert requests.event_id(packet, 'u_alice') == 'e_' + digest(('u_alice', 'stable-event'))[7:39]
+
+
+@pytest.mark.asyncio
+async def test_production_executor_receives_only_explicit_collaborators(installed):
+    app, _ = installed
+    executor = app.executor
+    assert not hasattr(executor, 'application')
+    assert executor.max_request_bytes == app.settings.server.limits.max_request_bytes
+    assert callable(executor.project_result)
+    assert callable(executor.enqueue_event)
+
+
+@pytest.mark.asyncio
+async def test_declared_executor_ports_keep_projection_events_receipts_and_replay(installed):
+    from msg.core.executor import OperationExecutor
+    app, _ = installed
+    key, uid, cert = await register(app, 'declared-executor-ports')
+    original = app.executor
+    projected, emitted = [], []
+
+    async def project(context, request, tx, ref, selected):
+        projected.append(ref)
+        return await original.project_result(context, request, tx, ref, selected)
+
+    async def enqueue(tx, event):
+        emitted.append(event)
+        await original.enqueue_event(tx, event)
+
+    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,
+        project_result=project, enqueue_event=enqueue)
+    packet = request_for('content.post_create', {'parent':'/main','body':'port contract'},
+        app.settings.service_url, subject=uid, signer=key, certificates=(cert,),
+        request_id='declared-ports-once', expires_at=NOW+timedelta(seconds=90),
+        return_fields=('id',))
+    first = await executor.execute(packet)
+    assert first.status == 'ok', wire(first)
+    assert first.data['projection'][0]['id'] == first.resources[0].id
+    assert projected == list(first.resources)
+    assert [event.request_id for event in emitted] == [packet.request_id]
+    verify(app.receipt_signer.public_key, receipt_bytes(first), first.receipt, purpose='receipt')
+    replay = await executor.execute(packet)
+    assert replay.replayed and replay.resources == first.resources
+    assert len(projected) == len(emitted) == 1
+
+
+@pytest.mark.asyncio
+async def test_injected_event_failure_rolls_back_business_and_idempotency(installed):
+    from msg.core.executor import OperationExecutor
+    app, _ = installed
+    key, uid, cert = await register(app, 'event-port-rollback')
+
+    async def reject(tx, event):
+        assert tx.one('SELECT id FROM events WHERE id=?', (event.id,))
+        raise Failure('event_port_rejected')
+
+    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,
+        project_result=app.executor.project_result, enqueue_event=reject)
+    packet = request_for('content.post_create', {'parent':'/main','body':'not committed'},
+        app.settings.service_url, subject=uid, signer=key, certificates=(cert,),
+        request_id='event-port-failed', expires_at=NOW+timedelta(seconds=90))
+    async with app.metadata.transaction(write=False) as tx:
+        before = tuple(tx.one(f'SELECT COUNT(*) FROM {table}')[0]
+                       for table in ('events','resources','results','jobs'))
+    result = await executor.execute(packet)
+    assert result.error.code == 'event_port_rejected', wire(result)
+    async with app.metadata.transaction(write=False) as tx:
+        after = tuple(tx.one(f'SELECT COUNT(*) FROM {table}')[0]
+                      for table in ('events','resources','results','jobs'))
+    assert after == before
+
+
+def test_sandbox_does_not_import_the_worker_for_its_result_type():
+    assert 'msg.workers.effects' not in imports(source('msg.workers.sandbox'))
+
+
+def test_tool_runner_has_one_neutral_result_and_no_obsolete_publication_port():
+    contract = importlib.import_module('msg.core.tool_runner')
+    worker = importlib.import_module('msg.workers.effects')
+    contracts = importlib.import_module('msg.core.contracts')
+    sandbox = importlib.import_module('msg.workers.sandbox')
+    assert worker.ToolResult is contracts.ToolResult is contract.ToolResult
+    assert contracts.ToolRunner is contract.ToolRunner
+    assert not hasattr(contracts, 'ToolExecutor')
+    assert [field.name for field in fields(contract.ToolResult)] == ['path','media_type','metadata']
+    assert get_type_hints(sandbox.BubblewrapRunner.__call__) == get_type_hints(contract.ToolRunner.__call__)
+    assert get_type_hints(contract.ToolRunner.__call__)['return'] is not ResourceRef
+    assert inspect.iscoroutinefunction(sandbox.BubblewrapRunner.__call__)
+    assert get_type_hints(worker.EffectWorker.__init__)['tool_runner'] == contract.ToolRunner | None
+
+
+@pytest.mark.asyncio
+async def test_default_and_injected_runners_use_the_same_runtime_port(installed, tmp_path):
+    from msg.core.tool_runner import ToolRunner, ToolResult
+    from msg.workers.effects import EffectWorker
+    from msg.workers.sandbox import BubblewrapRunner
+    app, _ = installed
+    default = EffectWorker(app)
+    assert isinstance(default.tool_runner, (ToolRunner, BubblewrapRunner))
+    calls = []
+
+    async def runner(tool, arguments, policies, directory):
+        calls.append((tool, arguments, policies, directory))
+        return ToolResult(path=directory/'output.bin', media_type='text/plain', metadata={})
+
+    injected = EffectWorker(app, tool_runner=runner)
+    assert isinstance(injected.tool_runner, ToolRunner)
+    result = await injected.tool_runner(None, {}, (), tmp_path)
+    assert type(result) is ToolResult and result.path == tmp_path/'output.bin'
+    assert calls == [(None, {}, (), tmp_path)]