From 63b4a3f83f2a128aaab555a2b14f4cc0df7fe12c Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Tue, 29 Sep 2026 04:41:56 +0800 Subject: [PATCH] test: lock execution and tool-runner boundaries before refactoring --- .github/workflows/execution-boundaries.yml | 65 +++++++ docs/EXECUTION_BOUNDARY_PROGRESS.md | 36 ++++ tests/test_execution_boundaries.py | 205 +++++++++++++++++++++ 3 files changed, 306 insertions(+) create mode 100644 .github/workflows/execution-boundaries.yml create mode 100644 docs/EXECUTION_BOUNDARY_PROGRESS.md create mode 100644 tests/test_execution_boundaries.py diff --git a/.github/workflows/execution-boundaries.yml b/.github/workflows/execution-boundaries.yml new file mode 100644 index 00000000..84b79e95 --- /dev/null +++ b/.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 diff --git a/docs/EXECUTION_BOUNDARY_PROGRESS.md b/docs/EXECUTION_BOUNDARY_PROGRESS.md new file mode 100644 index 00000000..7c2d44fc --- /dev/null +++ b/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. diff --git a/tests/test_execution_boundaries.py b/tests/test_execution_boundaries.py new file mode 100644 index 00000000..87f0ce57 --- /dev/null +++ b/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)]