From b553c59b484a4e5a383df3500ad9e4cab9d51272 Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Tue, 29 Sep 2026 04:41:15 +0800 Subject: [PATCH 1/7] test(architecture): expose runner port drift with cloud regressions Refs #160 #82 #83 #85. Record source-bound CI evidence and an independent progress note before changing implementation. --- .github/workflows/architecture-boundaries.yml | 80 +++++++++++ docs/ARCHITECTURE_PROGRESS_20260929.md | 32 +++++ tests/test_tool_runner_contract.py | 131 ++++++++++++++++++ 3 files changed, 243 insertions(+) create mode 100644 .github/workflows/architecture-boundaries.yml create mode 100644 docs/ARCHITECTURE_PROGRESS_20260929.md create mode 100644 tests/test_tool_runner_contract.py diff --git a/.github/workflows/architecture-boundaries.yml b/.github/workflows/architecture-boundaries.yml new file mode 100644 index 00000000..25a9b25a --- /dev/null +++ b/.github/workflows/architecture-boundaries.yml @@ -0,0 +1,80 @@ +name: Architecture boundaries +on: + push: + branches: ['fix/architecture-boundaries-*'] + pull_request: + paths: + - 'src/msg/core/**' + - 'src/msg/workers/**' + - 'tests/test_tool_runner_contract.py' + - '.github/workflows/architecture-boundaries.yml' + workflow_dispatch: +permissions: + contents: read +concurrency: + group: architecture-${{ github.ref }} + cancel-in-progress: true +jobs: + runner-contract: + runs-on: ubuntu-latest + timeout-minutes: 15 + 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 + valkey: + image: valkey/valkey:9.0.6 + ports: ['6379:6379'] + options: >- + --health-cmd "valkey-cli ping" + --health-interval 10s + --health-timeout 5s + --health-retries 5 + 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: Record exact tested source + run: | + mkdir -p artifacts + python - <<'PY' + import json, os, pathlib, subprocess + def git(*args): + return subprocess.check_output(['git', *args], text=True).strip() + pathlib.Path('artifacts/source.json').write_text(json.dumps({ + 'commit': git('rev-parse', 'HEAD'), 'tree': git('rev-parse', 'HEAD^{tree}'), + 'event': os.environ['GITHUB_EVENT_NAME'], 'run_id': os.environ['GITHUB_RUN_ID'], + }, indent=2) + '\n') + PY + - name: Install isolated test prerequisites + run: | + sudo apt-get update -qq + sudo apt-get install -y -qq age git-lfs + python -m pip install -e '.[dev]' + - name: Check runner ownership, real worker publication and safety fences + run: >- + python -m pytest tests/test_tool_runner_contract.py tests/test_tools.py + tests/test_effect_completion_fencing.py + --junitxml=artifacts/runner-contract.xml + - uses: actions/upload-artifact@v4 + if: always() + with: + name: architecture-boundaries + path: artifacts/ diff --git a/docs/ARCHITECTURE_PROGRESS_20260929.md b/docs/ARCHITECTURE_PROGRESS_20260929.md new file mode 100644 index 00000000..78c089be --- /dev/null +++ b/docs/ARCHITECTURE_PROGRESS_20260929.md @@ -0,0 +1,32 @@ +# Architecture issue progress — 2026-09-29 + +## Source and working boundary + +- Repository: TokenNotIncluded/msg.lmm.best; coordinator: #83; architecture umbrella: #85. +- Starting main: `476e64f04ed1d9c7dcab4a41c4f3cdb72faad6a7`, tree `40ca207a3c49be347120e058e47ed328ec4e3098`. +- Independent branch: `fix/architecture-boundaries-20260929`. Do not write or force-push the recovery branch for #155. +- Main CI runs 36472507327 (complete contracts) and 36472507300 (recovery) were read as success. Shard-0 source artifact 10992168743 was downloaded and its SHA-256 verified as `9abbc69988b8ffcb45f9931dc87e291145a7680f275bf7aadbedb7bed2189d10`. This is baseline evidence, not evidence for new changes. +- Follow the authoritative design and valid completed archive linked by #83. Preserve historical schemas/signatures, current authorization, one transaction publisher, quarantine and lease fences. + +## Active slice: #160 + +Observed: `ToolExecutor.invoke(...) -> ResourceRef` has no consumer; the worker actually injects an async callable returning a local `ToolResult`. That result is owned by `workers.effects`, forcing the sandbox adapter to import its orchestrator. + +1. Regression-first commit: new ownership/import/type/assembly tests plus real PostgreSQL worker success and invalid-result refusal. Existing tool and completion-fence tests remain unchanged. +2. Cloud-only execution: `.github/workflows/architecture-boundaries.yml` records exact commit/tree and uploads JUnit; it does not replace the full four-shard/conformance/build gate. +3. Planned minimal change: neutral `core.tool_execution` owns a callable ToolRunner and local ToolResult; worker and sandbox consume them. Keep the existing effects.ToolResult import as the same type. Retire the unused invoke declaration explicitly, retaining ToolExecutor only as an import alias for the actual runner port. +4. Pending: read the red run, implement, read focused and full final-head evidence, review diff and merge only if the exact combination is verified. + +No implementation success or issue closure is claimed by this regression-first note. + +## Queue and separate work + +- #156: executor currently imports batch, discovery and communication helpers and retains Application; needs explicit narrow assembly dependencies, not just one renamed import. +- #157: PostgreSQL inherits concrete SqliteSession; separate genuinely neutral session behavior without changing SQL/locks/schema. +- #158: client imports server executor/Git/vault helpers; minimal-client packaging remains a distinct acceptance requirement. +- #159: market/plugin bidirectional helper ownership requires preserving all published order versions and the sole protected ledger path. +- #155: live head observed `0e4dbe777053e354bcf4845b851a0c9b86ab8310`; title says verifier fix submitted while body still describes older failed head. Do not infer success; refresh its actual diff/CI and resolve main conflicts in a separate serial integration. +- Functional umbrella issues #68–#83/#85 remain open unless every remaining acceptance item has exact evidence. Existing completed functionality must not be rewritten from stale issue descriptions. +- #64/#65/#84 need authorized production logs/configuration, real protected snapshots and field/capacity evidence; #68/#69/#70 also retain external trust/retirement/console requirements. Code CI cannot supply those facts. + +No deployment, production Root/PIN, real money, outbound notification, key destruction or recovery promotion is authorized or performed. diff --git a/tests/test_tool_runner_contract.py b/tests/test_tool_runner_contract.py new file mode 100644 index 00000000..c1d915a0 --- /dev/null +++ b/tests/test_tool_runner_contract.py @@ -0,0 +1,131 @@ +"""The runner owns local output; only the worker may publish a ResourceRef.""" +from __future__ import annotations + +import ast +from dataclasses import FrozenInstanceError, fields +import inspect +from pathlib import Path +from typing import get_type_hints + +import pytest + +from msg.core.errors import Failure +from msg.core.models import JsonMap, NetworkPolicy, ResourceRef, ToolSpec +from msg.workers.effects import EffectWorker +from msg.workers.sandbox import BubblewrapRunner +from test_service import call, register +from test_authorization import approve, scoped + + +def test_tool_result_has_one_neutral_owner_and_compatible_export(tmp_path): + from msg.core.tool_execution import ToolResult + from msg.workers.effects import ToolResult as CompatibilityResult + + assert CompatibilityResult is ToolResult + assert ToolResult.__module__ == 'msg.core.tool_execution' + assert [field.name for field in fields(ToolResult)] == ['path', 'media_type', 'metadata'] + result = ToolResult(tmp_path / 'output.bin', 'application/octet-stream', {'size': 1}) + assert not isinstance(result, ResourceRef) + assert not hasattr(result, '__dict__') + with pytest.raises(FrozenInstanceError): + result.path = tmp_path / 'other.bin' + + +def test_actual_worker_dependency_is_the_declared_callable_port(): + from msg.core.tool_execution import ToolRunner + + assert get_type_hints(EffectWorker.__init__)['tool_runner'] == ToolRunner | None + assert isinstance(BubblewrapRunner(None), ToolRunner) + + +def test_default_runner_and_port_have_the_same_input_and_local_output_types(): + from msg.core.tool_execution import ToolResult, ToolRunner + + expected = {'tool': ToolSpec, 'arguments': JsonMap, + 'policies': tuple[NetworkPolicy, ...], 'directory': Path, 'return': ToolResult} + assert get_type_hints(ToolRunner.__call__) == expected + assert get_type_hints(BubblewrapRunner.__call__) == expected + assert inspect.iscoroutinefunction(BubblewrapRunner.__call__) + assert inspect.iscoroutinefunction(ToolRunner.__call__) + + +def test_unused_publishing_port_is_retired_not_a_second_runner_contract(): + from msg.core.contracts import ToolExecutor + from msg.core.tool_execution import ToolRunner + + # Keep the old import spelling, not the unused invoke -> ResourceRef interface. + assert ToolExecutor is ToolRunner + assert not hasattr(ToolExecutor, 'invoke') + + +def test_sandbox_does_not_import_its_worker_or_application_composition(): + import msg.workers.sandbox as sandbox + + tree = ast.parse(inspect.getsource(sandbox)) + dependencies = [] + for node in ast.walk(tree): + if isinstance(node, ast.ImportFrom): + dependencies.append(node.module or '') + elif isinstance(node, ast.Import): + dependencies.extend(alias.name for alias in node.names) + assert 'msg.core.tool_execution' in dependencies + assert not any(name == 'msg.workers.effects' or name.startswith('msg.workers.effects.') + or name == 'msg.application' for name in dependencies) + + +async def test_default_runner_still_refuses_missing_mandatory_sandbox(monkeypatch, tmp_path): + import msg.workers.sandbox as sandbox + + monkeypatch.setattr(sandbox.shutil, 'which', lambda name: None) + with pytest.raises(Failure, match='tool_isolation_unavailable'): + await BubblewrapRunner(None)(None, {}, (), tmp_path) + assert list(tmp_path.iterdir()) == [] + + +@pytest.mark.parametrize('invalid_result', [False, True]) +async def test_injected_port_cannot_publish_a_resource_itself(installed, invalid_result): + from msg.core.tool_execution import ToolResult, ToolRunner + + app, root = installed + key, subject, _ = await register(app, 'runner-contract') + cap = scoped(app, 'tool.use', 'tool_dns', app.registry.capability('tool.use').operations) + certificate = await approve(app, root, subject, key, (cap,)) + certs = (certificate.resource_id,) + request = await call(app, 'tool.run', + {'id': '/tools/dns', 'arguments': {'name': 'example.org', 'type': 'A'}}, + key=key, subject=subject, certs=certs) + assert request.status == 'accepted' + invocations = [] + + async def runner(tool: ToolSpec, arguments: JsonMap, + policies: tuple[NetworkPolicy, ...], directory: Path) -> ToolResult: + assert tool.executor_key == 'dns' + assert arguments['name'] == 'example.org' + assert isinstance(policies, tuple) and policies + assert all(isinstance(policy, NetworkPolicy) for policy in policies) + invocations.append(tool.resource) + if invalid_result: + # An untrusted/misconfigured runner must not publish an existing ref. + return ResourceRef(id='tool_dns', revision=tool.resource.revision) + output = directory / 'output.bin' + output.write_bytes(b'[]') + return ToolResult(output, 'application/json', {'kind': 'dns'}) + + assert isinstance(runner, ToolRunner) + worker = EffectWorker(app, tool_runner=runner) + assert worker.tool_runner is runner + assert await worker.run_once() + assert not await worker.run_once() + assert len(invocations) == 1 + async with app.metadata.transaction(write=False) as tx: + job = await tx.job(request.data['job_id']) + outputs = list(tx.execute('SELECT id FROM resources WHERE name=?', ('tool-' + job.id,))) + if invalid_result: + assert job.state == 'uncertain' + assert job.result is None + assert outputs == [] + else: + assert job.state == 'done' + assert isinstance(job.result, ResourceRef) + assert job.result.id != 'tool_dns' + assert [row[0] for row in outputs] == [job.result.id] From f14fd1533f70983b046df4b4c813993e8a536d12 Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Tue, 29 Sep 2026 04:43:50 +0800 Subject: [PATCH 2/7] refactor(tools): make the actual runner port and local result backend-neutral Keep publication, current authorization and lease fences in EffectWorker; preserve the effects.ToolResult export and explicitly retire the unused ToolExecutor.invoke declaration. Refs #160 #82 #85. --- docs/TOOL_RUNNER_BOUNDARY.md | 23 +++++++++++++++++++++++ src/msg/core/contracts.py | 10 +++++----- src/msg/core/tool_execution.py | 28 ++++++++++++++++++++++++++++ src/msg/workers/effects.py | 24 +++++++++--------------- src/msg/workers/sandbox.py | 6 ++++-- 5 files changed, 69 insertions(+), 22 deletions(-) create mode 100644 docs/TOOL_RUNNER_BOUNDARY.md create mode 100644 src/msg/core/tool_execution.py diff --git a/docs/TOOL_RUNNER_BOUNDARY.md b/docs/TOOL_RUNNER_BOUNDARY.md new file mode 100644 index 00000000..8b15f79f --- /dev/null +++ b/docs/TOOL_RUNNER_BOUNDARY.md @@ -0,0 +1,23 @@ +# Tool execution boundary + +Refs #160, #82, #85. This is a module-boundary change, not a new tool authority or scheduler. + +Before: EffectWorker injected an untyped callable; BubblewrapRunner imported ToolResult from EffectWorker's implementation module. The unused core ToolExecutor.invoke declaration instead claimed to return a published ResourceRef. + +After: + +```text +EffectWorker --> core.tool_execution.ToolRunner / ToolResult +BubblewrapRunner --> core.tool_execution.ToolResult +EffectWorker -- trusted default construction --> BubblewrapRunner +runner -- local ToolResult --> worker validation/current authority/attempt fence +worker -- existing storage transaction --> published ResourceRef +``` + +ToolRunner is the actual asynchronous callable port: ToolSpec, arguments, the tuple of already intersected NetworkPolicy alternatives and a per-attempt output directory. ToolResult is only a frozen local-artifact record (path, media_type, metadata). It is not proof of authorization, successful publication, or exactly-once execution. Runtime validation, mandatory isolation, output bounds, quarantine, current credential/certificate checks and attempt/state/deadline fences remain unchanged in their existing owners. + +EffectWorker's constructor and BubblewrapRunner's callable now share explicit input/output annotations; the assembly tests compare the annotations and exercise a correctly injected runner in real PostgreSQL. Returning an existing ResourceRef instead of a local artifact cannot bypass the publisher; the attempt becomes uncertain and no output resource is created. + +Compatibility: `msg.workers.effects.ToolResult` still exports the exact same class, not a copy or adapter. `msg.core.contracts.ToolExecutor` remains an import alias for ToolRunner. Its unused `invoke(...) -> ResourceRef` declaration is retired; no in-repository implementation or caller used it. This preserves the old import spelling, not that never-implemented invocation interface. New callers should use ToolRunner. No SMTP/Webhook/Git backend is made to implement this tool-specific port. + +Evidence lives in `tests/test_tool_runner_contract.py`, existing `tests/test_tools.py` and `tests/test_effect_completion_fencing.py`, the focused cloud workflow and the full CI gate. A missing-bwrap refusal test is not a claim of successful real bubblewrap or production acceptance. diff --git a/src/msg/core/contracts.py b/src/msg/core/contracts.py index 912b1caa..65eeb841 100644 --- a/src/msg/core/contracts.py +++ b/src/msg/core/contracts.py @@ -1,6 +1,7 @@ from __future__ import annotations from .models import * +from .tool_execution import ToolRunner class MetadataSession(Protocol): def on_rollback(self, effect: Callable[[], Awaitable[None]]) -> None: ... @@ -111,11 +112,10 @@ async def discover(self) -> TransportLimits: ... async def call(self, request: OperationRequest) -> OperationResult: ... -class ToolExecutor(Protocol): - async def invoke( - self, context: ExecutionContext, tool: ToolSpec, - arguments: JsonMap, policy: NetworkPolicy - ) -> ResourceRef: ... +# Compatibility import only. The unused invoke -> ResourceRef declaration never +# described the injected runner. ToolRunner is now the one actual execution port; +# publication and current-authority/attempt checks remain owned by EffectWorker. +ToolExecutor = ToolRunner class NotificationSender(Protocol): diff --git a/src/msg/core/tool_execution.py b/src/msg/core/tool_execution.py new file mode 100644 index 00000000..43c02efd --- /dev/null +++ b/src/msg/core/tool_execution.py @@ -0,0 +1,28 @@ +"""Local tool execution port, independent of worker orchestration and backends. + +A runner returns a local artifact, never a committed ResourceRef. The worker +validates it, rechecks current authority/attempt/deadline and publishes it in its +existing transaction. Implementing this Protocol grants no execution authority. +""" +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path +from typing import Protocol, runtime_checkable + +from .models import JsonMap, NetworkPolicy, ToolSpec + + +@dataclass(frozen=True, slots=True) +class ToolResult: + path: Path + media_type: str + metadata: dict + + +@runtime_checkable +class ToolRunner(Protocol): + async def __call__( + self, tool: ToolSpec, arguments: JsonMap, + policies: tuple[NetworkPolicy, ...], directory: Path, / + ) -> ToolResult: ... diff --git a/src/msg/workers/effects.py b/src/msg/workers/effects.py index ec43dbe1..108614a4 100644 --- a/src/msg/workers/effects.py +++ b/src/msg/workers/effects.py @@ -7,7 +7,7 @@ from __future__ import annotations import asyncio -from dataclasses import dataclass, replace +from dataclasses import replace from datetime import timedelta from pathlib import Path import tempfile @@ -21,19 +21,13 @@ CapabilityGrant, EffectJob, EmailSettings, ExecutionContext, Principal, ResourceRef, ) from msg.core.requests import request_for +from msg.core.tool_execution import ToolResult, ToolRunner from msg.plugins.common import check_access, create_resource from msg.security.policy import scope_subset from msg.security.network import intersect_policy from msg.workers.leases import current_attempt -@dataclass(frozen=True, slots=True) -class ToolResult: - path: Path - media_type: str - metadata: dict - - async def current_principal(app, original: Principal, tx) -> Principal: """Intersect the captured credential with its *current* authority ceiling.""" require(original.method not in {'anonymous', 'local'}, 'worker_identity_required') @@ -98,7 +92,7 @@ def effect_request(app, job, principal): class EffectWorker: - def __init__(self, app, *, tool_runner=None, mail_sender=None, webhook_sender=None, + def __init__(self, app, *, tool_runner: ToolRunner | None = None, mail_sender=None, webhook_sender=None, lease_seconds=120): self.app = app if tool_runner is None: @@ -275,12 +269,12 @@ async def _market_mail(self, job): from msg.market.email import render_verification require(mail_enabled(self.app), 'mail_disabled') async with self.app.metadata.transaction(write=False) as tx: - await current_principal(self.app, job.principal, tx) - outgoing = await (render_verification(self.app,tx,job) if job.kind=='market_email_verify' - else render_notification(self.app,tx,job)) - state = await self.mail_sender.send(outgoing) - require(state in {'sent', 'uncertain'}, 'invalid_delivery_result') - await self._finish(job, 'done' if state == 'sent' else 'uncertain', state) + await current_principal(self.app,job.principal,tx) + outgoing=await (render_verification(self.app,tx,job) if job.kind=='market_email_verify' + else render_notification(self.app,tx,job)) + state=await self.mail_sender.send(outgoing) + require(state in {'sent','uncertain'},'invalid_delivery_result') + await self._finish(job,'done' if state=='sent' else 'uncertain',state) async def _webhook(self, job): from msg.core.codec import canonical,decode,loads diff --git a/src/msg/workers/sandbox.py b/src/msg/workers/sandbox.py index 6aa8b226..ea3cf00e 100644 --- a/src/msg/workers/sandbox.py +++ b/src/msg/workers/sandbox.py @@ -6,14 +6,16 @@ import sys from msg.core.codec import canonical, loads, wire from msg.core.errors import Failure, require +from msg.core.models import JsonMap, NetworkPolicy, ToolSpec +from msg.core.tool_execution import ToolResult class BubblewrapRunner: def __init__(self, app): self.app = app - async def __call__(self, tool, arguments, policies, directory): - from msg.workers.effects import ToolResult + async def __call__(self, tool: ToolSpec, arguments: JsonMap, + policies: tuple[NetworkPolicy, ...], directory: Path) -> ToolResult: bwrap = shutil.which('bwrap') require(bwrap is not None, 'tool_isolation_unavailable') source = Path(__file__).resolve().parents[2] From 1953beeda08cb34570f2c56535c4e4b1249546eb Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Tue, 29 Sep 2026 04:47:09 +0800 Subject: [PATCH 3/7] test(core): expose executor composition and plugin dependency debt Refs #156 #83 #85. Run independent assembly and existing transaction/security regressions in cloud before changing implementation. --- .github/workflows/executor-boundary.yml | 85 +++++++++++ docs/EXECUTOR_PROGRESS_20260929.md | 21 +++ tests/test_executor_boundary.py | 189 ++++++++++++++++++++++++ 3 files changed, 295 insertions(+) create mode 100644 .github/workflows/executor-boundary.yml create mode 100644 docs/EXECUTOR_PROGRESS_20260929.md create mode 100644 tests/test_executor_boundary.py diff --git a/.github/workflows/executor-boundary.yml b/.github/workflows/executor-boundary.yml new file mode 100644 index 00000000..6e94921f --- /dev/null +++ b/.github/workflows/executor-boundary.yml @@ -0,0 +1,85 @@ +name: Executor boundary +on: + push: + branches: ['fix/executor-boundary-*'] + pull_request: + paths: + - 'src/msg/core/**' + - 'src/msg/application.py' + - 'src/msg/plugins/batch.py' + - 'src/msg/plugins/communication.py' + - 'src/msg/extensions/ssh.py' + - 'tests/test_executor_boundary.py' + - '.github/workflows/executor-boundary.yml' + workflow_dispatch: +permissions: + contents: read +concurrency: + group: executor-boundary-${{ github.ref }} + cancel-in-progress: true +jobs: + assembly-and-transactions: + runs-on: ubuntu-latest + timeout-minutes: 15 + 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 -d postgres" + --health-interval 10s + --health-timeout 5s + --health-retries 5 + valkey: + image: valkey/valkey:9.0.6 + ports: ['6379:6379'] + options: >- + --health-cmd "valkey-cli ping" + --health-interval 10s + --health-timeout 5s + --health-retries 5 + 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: Record tested source + run: | + mkdir -p artifacts + python - <<'PY' + import json, os, pathlib, subprocess + def git(*args): + return subprocess.check_output(['git', *args], text=True).strip() + pathlib.Path('artifacts/source.json').write_text(json.dumps({ + 'commit': git('rev-parse', 'HEAD'), 'tree': git('rev-parse', 'HEAD^{tree}'), + 'run_id': os.environ['GITHUB_RUN_ID'], 'event': os.environ['GITHUB_EVENT_NAME'], + }, indent=2) + '\n') + PY + - name: Install test prerequisites + run: | + sudo apt-get update -qq + sudo apt-get install -y -qq age git-lfs + python -m pip install -e '.[dev]' + - name: Exercise assembly and unchanged security transaction contracts + run: >- + python -m pytest tests/test_executor_boundary.py tests/test_batch.py + tests/test_batch_secret_release.py tests/test_webhook_domain_events.py + tests/test_ssh_rss.py tests/test_security_transfer_rollback.py + tests/test_service.py::test_requested_write_projection_is_returned_without_readback + --junitxml=artifacts/executor-boundary.xml + - uses: actions/upload-artifact@v4 + if: always() + with: + name: executor-boundary + path: artifacts/ diff --git a/docs/EXECUTOR_PROGRESS_20260929.md b/docs/EXECUTOR_PROGRESS_20260929.md new file mode 100644 index 00000000..156b1cd9 --- /dev/null +++ b/docs/EXECUTOR_PROGRESS_20260929.md @@ -0,0 +1,21 @@ +# Executor architecture progress — 2026-09-29 + +Refs #156/#83/#85. Independent stacked branch `fix/executor-boundary-20260929`, starting from tool boundary implementation `f14fd1533f70983b046df4b4c813993e8a536d12` (PR #165). Do not overwrite #165 or #155 branches. This note supplements `ARCHITECTURE_PROGRESS_20260929.md`. + +## Regression-first + +New tests expose core imports of plugin implementations and the Application backreference, inconsistent standalone/factory assembly, and missing narrow services reported as internal errors. Real PostgreSQL tests retain batch idempotency, write projections, and verify Event projection failure rolls back both business state and Event/result in the existing transaction. Unchanged batch secret-delivery, webhook domain, SSH and transfer rollback tests run alongside them in `.github/workflows/executor-boundary.yml`. + +This commit only introduces tests/workflow/notes. No success is claimed before cloud execution. Existing full four-shard/conformance/build CI remains mandatory and unchanged. + +## Planned ownership + +- Core BatchPolicy owns the single child admission algorithm and consumes an explicit bounded packet decoder; core must not import a transport or plugin to find one. +- Core event identity owns the existing deterministic digest; communication keeps only a compatibility import. +- OperationExecutor receives a packet decoder/byte limit, read-projection function and in-transaction Event projection function. It holds no Application/settings backreference. +- Application is the composition root for those dependencies; normal and SSH entrypoints use the same factory with their own authenticator and unchanged recovery/secret-delivery fences. +- Missing decoder/projection is a stable fail-closed error, not an internal error or implicit authorization fallback. Event delivery effects stay in the existing transaction; this is not a configurable workflow/hook engine. + +## Remaining evidence + +Read RED result, implement without weakening old assertions, check focused/full final head and reviewed exact combination. #156 remains open until those gates pass. No production/log/snapshot/Root/PIN/funds/outbound/recovery promotion work is performed. Production-blocked issues remain blocked. diff --git a/tests/test_executor_boundary.py b/tests/test_executor_boundary.py new file mode 100644 index 00000000..856a2969 --- /dev/null +++ b/tests/test_executor_boundary.py @@ -0,0 +1,189 @@ +"""Explicit executor assembly, with unchanged real transaction/security paths.""" +from __future__ import annotations + +import ast +from contextlib import asynccontextmanager +from datetime import timedelta +import inspect +from types import SimpleNamespace + +import pytest + +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 + + +def test_executor_has_no_business_plugin_imports_or_application_backreference(): + import msg.core.executor as executor + + tree = ast.parse(inspect.getsource(executor)) + forbidden = [] + for node in ast.walk(tree): + if isinstance(node, ast.ImportFrom) and (node.module or '').startswith( + ('msg.plugins', 'msg.application', 'msg.transports')): + forbidden.append(node.module) + elif isinstance(node, ast.Import): + forbidden.extend(alias.name for alias in node.names if alias.name.startswith( + ('msg.plugins', 'msg.application', 'msg.transports'))) + assert forbidden == [] + assert not any(isinstance(node, ast.Attribute) and node.attr == 'application' + for node in ast.walk(tree)) + + +def test_batch_structure_has_one_neutral_owner(): + import msg.core.batch as batch + import msg.plugins.batch as plugin + + tree = ast.parse(inspect.getsource(batch)) + assert not any(isinstance(node, ast.ImportFrom) and (node.module or '').startswith( + ('msg.plugins', 'msg.transports', 'msg.application')) for node in ast.walk(tree)) + # The plugin must not retain a second copy of the child admission algorithm. + assert not any(isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name == 'packets' + for node in ast.walk(ast.parse(inspect.getsource(plugin)))) + assert batch.NO_BATCH >= {'identity.register', 'identity.upgrade', 'identity.token_recover'} + + +def test_event_identity_is_shared_without_importing_the_communication_plugin(): + from msg.core.events import event_id + from msg.plugins.communication import event_id as compatibility_id + + request = request_for('testing.read', {}, 'https://unit.invalid', request_id='event-once') + assert event_id is compatibility_id + assert event_id(request, 'u_test') == event_id(request, 'u_test') + assert event_id(request, 'u_test') != event_id(request, 'u_other') + + +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): + assert active == [transaction] and tx is transaction + assert request.return_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, projection_reader=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 result.data == {'original': True, 'projection': [{'id': 'r_test'}]} + assert active == [] + + +async def test_application_factory_configures_normal_and_alternate_authentication(installed): + app, _ = installed + alternate = object() + executor = app.make_executor(authenticator=alternate) + assert executor.authenticator is alternate + assert not hasattr(executor, 'application') + assert executor.projection_reader is not None + assert executor.event_projector is not None + assert executor.batch_policy.max_request_bytes == app.settings.server.limits.max_request_bytes + assert executor.response_hook == app._secrets_for_caller + assert executor.recovery_drill_marker == app.executor.recovery_drill_marker + assert executor.recovery_quarantined == app.executor.recovery_quarantined + import msg.extensions.ssh as ssh + source = inspect.getsource(ssh) + assert 'app.make_executor(' in source + assert 'executor.application' not in source + + +async def test_factory_batch_and_projection_work_without_application_backreference(installed): + app, _ = installed + key, subject, _ = await register(app, 'executor-factory') + app.executor = app.make_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.make_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_projector = 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 + + +@pytest.mark.parametrize('case', ['batch', 'projection']) +async def test_missing_optional_service_fails_closed_without_internal_error_or_commit(installed, case): + app, _ = installed + key, subject, _ = await register(app, 'unconfigured-executor') + executor = OperationExecutor(app.registry, app.metadata, app.contents, + app.authenticator, app.authorizer, app.clock, app.receipt_signer) + if case == 'batch': + child = packet(app, key, subject, 'content.post_create', + {'parent': '/main', 'body': 'not committed'}, 'unconfigured-child') + request = request_for('batch.atomic', {'requests': [child]}, app.settings.service_url, + subject=subject, signer=key, expires_at=NOW + timedelta(seconds=60)) + expected = 'batch_decoder_not_configured' + else: + request = request_for('content.post_create', {'parent': '/main', 'body': 'not committed'}, + app.settings.service_url, subject=subject, signer=key, + expires_at=NOW + timedelta(seconds=60), return_fields=('id',)) + expected = 'projection_unavailable' + result = await executor.execute(request) + assert result.status == 'error' and result.error.code == expected, wire(result) + async with app.metadata.transaction(write=False) as tx: + assert tx.one("SELECT COUNT(*) FROM resources WHERE type='post'")[0] == 0 From 9b45cfa359800c5d14322203e3b8eecee85da918 Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Tue, 29 Sep 2026 04:53:37 +0800 Subject: [PATCH 4/7] ci(core): apply the reviewed executor patch on the isolated cloud branch Record exact parent and nine allowed source paths; no force update, no test or production execution with write credentials. Subsequent PR CI verifies the generated source independently. --- .../workflows/prepare-executor-boundary.yml | 80 +++++ scripts/patches/executor-boundary.patch | 331 ++++++++++++++++++ 2 files changed, 411 insertions(+) create mode 100644 .github/workflows/prepare-executor-boundary.yml create mode 100644 scripts/patches/executor-boundary.patch diff --git a/.github/workflows/prepare-executor-boundary.yml b/.github/workflows/prepare-executor-boundary.yml new file mode 100644 index 00000000..a90fc7f2 --- /dev/null +++ b/.github/workflows/prepare-executor-boundary.yml @@ -0,0 +1,80 @@ +name: Apply reviewed executor boundary patch +on: + push: + branches: ['fix/executor-boundary-20260929'] + paths: ['scripts/patches/executor-boundary.patch'] +permissions: + contents: read +jobs: + apply-reviewed-patch: + if: github.repository == 'TokenNotIncluded/msg.lmm.best' + runs-on: ubuntu-latest + timeout-minutes: 5 + permissions: + contents: write + steps: + - uses: actions/checkout@v7 + with: + persist-credentials: false + fetch-depth: 2 + - name: Apply exactly the reviewed patch to its recorded parent + run: | + test "$(git rev-parse HEAD^)" = 1953beeda08cb34570f2c56535c4e4b1249546eb + git apply --index --check scripts/patches/executor-boundary.patch + git apply --index scripts/patches/executor-boundary.patch + git diff --cached --check + - name: Publish only to this feature branch, without force + env: + GH_TOKEN: ${{ github.token }} + run: | + python - <<'PY' + import hashlib, json, os, pathlib, subprocess, urllib.request + repo = 'TokenNotIncluded/msg.lmm.best' + branch = 'fix/executor-boundary-20260929' + assert os.environ['GITHUB_REPOSITORY'] == repo + assert os.environ['GITHUB_REF'] == 'refs/heads/' + branch + paths = [ + 'src/msg/application.py', 'src/msg/core/batch.py', 'src/msg/core/events.py', + 'src/msg/core/execution_ports.py', 'src/msg/core/executor.py', + 'src/msg/extensions/ssh.py', 'src/msg/plugins/batch.py', + 'src/msg/plugins/communication.py', 'tests/issue_78_80/conftest.py', + ] + changed = subprocess.check_output(['git', 'diff', '--cached', '--name-only'], text=True).splitlines() + assert sorted(changed) == sorted(paths), changed + def api(method, endpoint, payload=None): + data = None if payload is None else json.dumps(payload).encode() + request = urllib.request.Request('https://api.github.com/repos/' + repo + endpoint, + data=data, method=method, headers={ + 'Authorization': 'Bearer ' + os.environ['GH_TOKEN'], + 'Accept': 'application/vnd.github+json', 'Content-Type': 'application/json', + 'X-GitHub-Api-Version': '2022-11-28', + }) + with urllib.request.urlopen(request, timeout=60) as response: + return json.load(response) + parent = os.environ['GITHUB_SHA'] + assert api('GET', '/git/ref/heads/' + branch)['object']['sha'] == parent + old = api('GET', '/git/commits/' + parent) + assert [p['sha'] for p in old['parents']] == ['1953beeda08cb34570f2c56535c4e4b1249546eb'] + entries = [{'path': p, 'mode': '100644', 'type': 'blob', + 'content': pathlib.Path(p).read_text()} for p in paths] + # The one-shot patch publisher is retained in history, not in the final tree. + for p in ['scripts/patches/executor-boundary.patch', '.github/workflows/prepare-executor-boundary.yml']: + entries.append({'path': p, 'mode': '100644', 'type': 'blob', 'sha': None}) + tree = api('POST', '/git/trees', {'base_tree': old['tree']['sha'], 'tree': entries}) + commit = api('POST', '/git/commits', { + 'message': 'refactor(core): inject bounded executor services at the composition root\n\nRefs #156 #83 #85. Preserve the existing transaction, authentication and secret-delivery paths.', + 'tree': tree['sha'], 'parents': [parent], + }) + api('PATCH', '/git/refs/heads/' + branch, {'sha': commit['sha'], 'force': False}) + pathlib.Path('artifacts').mkdir(exist_ok=True) + pathlib.Path('artifacts/patch-result.json').write_text(json.dumps({ + 'source': parent, 'commit': commit['sha'], 'tree': tree['sha'], + 'patch_sha256': hashlib.sha256(pathlib.Path('scripts/patches/executor-boundary.patch').read_bytes()).hexdigest(), + 'paths': paths, + }, indent=2) + '\n') + print('Published reviewed source:', commit['sha'], tree['sha']) + PY + - uses: actions/upload-artifact@v4 + with: + name: executor-patch-result + path: artifacts/ diff --git a/scripts/patches/executor-boundary.patch b/scripts/patches/executor-boundary.patch new file mode 100644 index 00000000..c4453c94 --- /dev/null +++ b/scripts/patches/executor-boundary.patch @@ -0,0 +1,331 @@ +diff --git a/src/msg/application.py b/src/msg/application.py +index 591920d..df294fb 100644 +--- a/src/msg/application.py ++++ b/src/msg/application.py +@@ -119,15 +119,37 @@ class Application: + 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.make_executor() + 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 make_executor(self, *, authenticator=None): ++ """Compose the same finite services for network and SSH execution.""" ++ from functools import partial ++ from msg.plugins.discovery import read_projection ++ from msg.plugins.communication import enqueue_domain_webhooks ++ from msg.transports.packet import decode_packet ++ ++ async def projection(context, request, tx, ref): ++ return await read_projection(self,context,request,tx,ref.id, ++ revision=ref.revision,fields=request.return_fields) ++ ++ 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, ++ packet_decoder=decode_packet,projection_reader=projection, ++ event_projector=partial(enqueue_domain_webhooks,self)) ++ 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_drill_marker=self.executor.recovery_drill_marker ++ executor.recovery_quarantined=self.executor.recovery_quarantined ++ return executor ++ + 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/batch.py b/src/msg/core/batch.py +new file mode 100644 +index 0000000..aea934e +--- /dev/null ++++ b/src/msg/core/batch.py +@@ -0,0 +1,34 @@ ++"""One bounded child-admission policy for both named batch primitives.""" ++from __future__ import annotations ++ ++from .codec import canonical ++from .errors import require ++from .execution_ports import PacketDecoder ++from .requests import SECRET_DELIVERY_MIN_VERSION ++ ++# Response-only credentials require their own non-batched delivery call. ++NO_BATCH = frozenset(SECRET_DELIVERY_MIN_VERSION) | {'identity.register', 'identity.upgrade'} ++ ++ ++class BatchPolicy: ++ def __init__(self, registry, decoder: PacketDecoder | None, max_request_bytes: int): ++ self.registry = registry ++ self.decoder = decoder ++ self.max_request_bytes = max_request_bytes ++ ++ def packets(self, request, subject, *, bounded=True): ++ require(self.decoder is not None, 'batch_decoder_not_configured') ++ # Independent batches historically use the validated parent envelope size. ++ max_bytes = self.max_request_bytes if bounded else len(canonical(request)) ++ values = [] ++ for value in request.arguments['requests']: ++ packet = self.decoder(value, max_bytes) ++ spec = self.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/events.py b/src/msg/core/events.py +new file mode 100644 +index 0000000..aa28f84 +--- /dev/null ++++ b/src/msg/core/events.py +@@ -0,0 +1,6 @@ ++"""Stable request Event identity shared by execution and domain adapters.""" ++from .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 0000000..b9798a3 +--- /dev/null ++++ b/src/msg/core/execution_ports.py +@@ -0,0 +1,19 @@ ++"""Narrow trusted assembly dependencies, not resource-configurable hooks.""" ++from __future__ import annotations ++ ++from typing import Protocol ++from .contracts import MetadataSession ++from .models import Event, ExecutionContext, JsonMap, OperationRequest, ResourceRef ++ ++ ++class PacketDecoder(Protocol): ++ def __call__(self, value: object, max_bytes: int, /) -> OperationRequest: ... ++ ++ ++class ProjectionReader(Protocol): ++ async def __call__(self, context: ExecutionContext, request: OperationRequest, ++ session: MetadataSession, ref: ResourceRef, /) -> JsonMap: ... ++ ++ ++class EventProjector(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 d00ac08..ba27481 100644 +--- a/src/msg/core/executor.py ++++ b/src/msg/core/executor.py +@@ -5,6 +5,9 @@ import time + from dataclasses import replace + from uuid import uuid4 + from msg.core.codec import wire,digest ++from msg.core.batch import BatchPolicy ++from msg.core.events import event_id ++from msg.core.execution_ports import PacketDecoder, ProjectionReader, EventProjector + 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 +16,19 @@ log=logging.getLogger(__name__) + + + 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=1048576, packet_decoder: PacketDecoder | None = None, ++ projection_reader: ProjectionReader | None = None, ++ event_projector: EventProjector | 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 ++ self.batch_policy=BatchPolicy(registry,packet_decoder,max_request_bytes) ++ self.projection_reader=projection_reader ++ self.event_projector=event_projector + + def recovery_drill_active(self): + marker=self.recovery_drill_marker +@@ -49,9 +57,7 @@ class OperationExecutor: + 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.batch_policy.packets(request,principal.subject) + 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,10 +100,9 @@ class OperationExecutor: + 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, +- fields=request.return_fields) for ref in output.resources] ++ require(output.resources and self.projection_reader is not None,'projection_unavailable') ++ projections=[await self.projection_reader(context,request,session,ref) ++ for ref in output.resources] + output=replace(output,data={**(output.data or {}),'projection':projections}) + if audited: + from msg.core.models import AuditEvent,ResourceRef +@@ -114,14 +119,12 @@ class OperationExecutor: + 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_projector is not None: ++ await self.event_projector(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 +153,10 @@ class OperationExecutor: + 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=self.batch_policy.packets(request,principal.subject,bounded=False) + 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) +diff --git a/src/msg/extensions/ssh.py b/src/msg/extensions/ssh.py +index d12baad..c855eea 100644 +--- a/src/msg/extensions/ssh.py ++++ b/src/msg/extensions/ssh.py +@@ -21,7 +21,7 @@ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey + 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.executor 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,8 @@ 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 ++ # Use the same finite services and recovery fences with SSH authentication. ++ executor = app.make_executor(authenticator=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/plugins/batch.py b/src/msg/plugins/batch.py +index beb05db..327edfb 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.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 +- ++from msg.transports.packet import REQUEST_SCHEMA + + def install(app): + op,finish=registration(app,'batch',('identity',)) +@@ -35,7 +14,7 @@ def install(app): + @op('batch.atomic',schema) + async def atomic(ctx,request,tx): + await app.authorizer._ceiling(ctx.principal,operation_id(request),ctx.principal.subject,tx) +- children=packets(app.registry,request,ctx.principal.subject,app.settings.server.limits.max_request_bytes) ++ children=app.executor.batch_policy.packets(request,ctx.principal.subject) + results=[];resources=[] + for index,packet in enumerate(children): + result=await app.executor.execute(packet,entry=ctx.entry) +diff --git a/src/msg/plugins/communication.py b/src/msg/plugins/communication.py +index d704aa7..579350a 100644 +--- a/src/msg/plugins/communication.py ++++ b/src/msg/plugins/communication.py +@@ -11,15 +11,13 @@ from msg.core.codec import wire,canonical,loads,decode,digest,parse_time,b64,unb + from msg.core.errors import Failure,require + from msg.core.models import HandlerOutput,ResourceRef,ResourceTypeSpec,EmailSettings,EffectJob,Principal + from msg.core.requests import signing_bytes ++from msg.core.events import event_id + from msg.plugins.common import * + from msg.plugins.schemas import * + from msg.plugins.discovery import visible + from msg.storage.capacity import require_webhook_capacity + + +-def event_id(request,subject): +- return 'e_'+digest((subject,request.request_id))[7:39] +- + + # A deliberately small event vocabulary. An Event is an audit fact, not blanket + # permission to disclose its data to an external receiver. +diff --git a/tests/issue_78_80/conftest.py b/tests/issue_78_80/conftest.py +index ada0f4a..954c25d 100644 +--- a/tests/issue_78_80/conftest.py ++++ b/tests/issue_78_80/conftest.py +@@ -125,9 +125,16 @@ async def harness(tmp_path): + 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)) ++ from msg.transports.packet import decode_packet ++ ++ async def project(context, request, tx, ref): ++ return await discovery.read_projection(app,context,request,tx,ref.id, ++ revision=ref.revision,fields=request.return_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, ++ packet_decoder=decode_packet,projection_reader=project) + app._loaded=True + + def packet(operation,args,version=1, *, signed=True,**kwargs): From c639a520ab69e22cccaabb07999d7cb54b8be4ca Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Mon, 28 Sep 2026 20:58:30 +0000 Subject: [PATCH 5/7] refactor(core): inject bounded executor services at the composition root Refs #156 #83 #85. Preserve the existing transaction, authentication and secret-delivery paths. --- .../workflows/prepare-executor-boundary.yml | 80 ----- scripts/patches/executor-boundary.patch | 331 ------------------ src/msg/application.py | 28 +- src/msg/core/batch.py | 34 ++ src/msg/core/events.py | 6 + src/msg/core/execution_ports.py | 19 + src/msg/core/executor.py | 33 +- src/msg/extensions/ssh.py | 8 +- src/msg/plugins/batch.py | 25 +- src/msg/plugins/communication.py | 4 +- tests/issue_78_80/conftest.py | 11 +- 11 files changed, 116 insertions(+), 463 deletions(-) delete mode 100644 .github/workflows/prepare-executor-boundary.yml delete mode 100644 scripts/patches/executor-boundary.patch create mode 100644 src/msg/core/batch.py create mode 100644 src/msg/core/events.py create mode 100644 src/msg/core/execution_ports.py diff --git a/.github/workflows/prepare-executor-boundary.yml b/.github/workflows/prepare-executor-boundary.yml deleted file mode 100644 index a90fc7f2..00000000 --- a/.github/workflows/prepare-executor-boundary.yml +++ /dev/null @@ -1,80 +0,0 @@ -name: Apply reviewed executor boundary patch -on: - push: - branches: ['fix/executor-boundary-20260929'] - paths: ['scripts/patches/executor-boundary.patch'] -permissions: - contents: read -jobs: - apply-reviewed-patch: - if: github.repository == 'TokenNotIncluded/msg.lmm.best' - runs-on: ubuntu-latest - timeout-minutes: 5 - permissions: - contents: write - steps: - - uses: actions/checkout@v7 - with: - persist-credentials: false - fetch-depth: 2 - - name: Apply exactly the reviewed patch to its recorded parent - run: | - test "$(git rev-parse HEAD^)" = 1953beeda08cb34570f2c56535c4e4b1249546eb - git apply --index --check scripts/patches/executor-boundary.patch - git apply --index scripts/patches/executor-boundary.patch - git diff --cached --check - - name: Publish only to this feature branch, without force - env: - GH_TOKEN: ${{ github.token }} - run: | - python - <<'PY' - import hashlib, json, os, pathlib, subprocess, urllib.request - repo = 'TokenNotIncluded/msg.lmm.best' - branch = 'fix/executor-boundary-20260929' - assert os.environ['GITHUB_REPOSITORY'] == repo - assert os.environ['GITHUB_REF'] == 'refs/heads/' + branch - paths = [ - 'src/msg/application.py', 'src/msg/core/batch.py', 'src/msg/core/events.py', - 'src/msg/core/execution_ports.py', 'src/msg/core/executor.py', - 'src/msg/extensions/ssh.py', 'src/msg/plugins/batch.py', - 'src/msg/plugins/communication.py', 'tests/issue_78_80/conftest.py', - ] - changed = subprocess.check_output(['git', 'diff', '--cached', '--name-only'], text=True).splitlines() - assert sorted(changed) == sorted(paths), changed - def api(method, endpoint, payload=None): - data = None if payload is None else json.dumps(payload).encode() - request = urllib.request.Request('https://api.github.com/repos/' + repo + endpoint, - data=data, method=method, headers={ - 'Authorization': 'Bearer ' + os.environ['GH_TOKEN'], - 'Accept': 'application/vnd.github+json', 'Content-Type': 'application/json', - 'X-GitHub-Api-Version': '2022-11-28', - }) - with urllib.request.urlopen(request, timeout=60) as response: - return json.load(response) - parent = os.environ['GITHUB_SHA'] - assert api('GET', '/git/ref/heads/' + branch)['object']['sha'] == parent - old = api('GET', '/git/commits/' + parent) - assert [p['sha'] for p in old['parents']] == ['1953beeda08cb34570f2c56535c4e4b1249546eb'] - entries = [{'path': p, 'mode': '100644', 'type': 'blob', - 'content': pathlib.Path(p).read_text()} for p in paths] - # The one-shot patch publisher is retained in history, not in the final tree. - for p in ['scripts/patches/executor-boundary.patch', '.github/workflows/prepare-executor-boundary.yml']: - entries.append({'path': p, 'mode': '100644', 'type': 'blob', 'sha': None}) - tree = api('POST', '/git/trees', {'base_tree': old['tree']['sha'], 'tree': entries}) - commit = api('POST', '/git/commits', { - 'message': 'refactor(core): inject bounded executor services at the composition root\n\nRefs #156 #83 #85. Preserve the existing transaction, authentication and secret-delivery paths.', - 'tree': tree['sha'], 'parents': [parent], - }) - api('PATCH', '/git/refs/heads/' + branch, {'sha': commit['sha'], 'force': False}) - pathlib.Path('artifacts').mkdir(exist_ok=True) - pathlib.Path('artifacts/patch-result.json').write_text(json.dumps({ - 'source': parent, 'commit': commit['sha'], 'tree': tree['sha'], - 'patch_sha256': hashlib.sha256(pathlib.Path('scripts/patches/executor-boundary.patch').read_bytes()).hexdigest(), - 'paths': paths, - }, indent=2) + '\n') - print('Published reviewed source:', commit['sha'], tree['sha']) - PY - - uses: actions/upload-artifact@v4 - with: - name: executor-patch-result - path: artifacts/ diff --git a/scripts/patches/executor-boundary.patch b/scripts/patches/executor-boundary.patch deleted file mode 100644 index c4453c94..00000000 --- a/scripts/patches/executor-boundary.patch +++ /dev/null @@ -1,331 +0,0 @@ -diff --git a/src/msg/application.py b/src/msg/application.py -index 591920d..df294fb 100644 ---- a/src/msg/application.py -+++ b/src/msg/application.py -@@ -119,15 +119,37 @@ class Application: - 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.make_executor() - 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 make_executor(self, *, authenticator=None): -+ """Compose the same finite services for network and SSH execution.""" -+ from functools import partial -+ from msg.plugins.discovery import read_projection -+ from msg.plugins.communication import enqueue_domain_webhooks -+ from msg.transports.packet import decode_packet -+ -+ async def projection(context, request, tx, ref): -+ return await read_projection(self,context,request,tx,ref.id, -+ revision=ref.revision,fields=request.return_fields) -+ -+ 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, -+ packet_decoder=decode_packet,projection_reader=projection, -+ event_projector=partial(enqueue_domain_webhooks,self)) -+ 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_drill_marker=self.executor.recovery_drill_marker -+ executor.recovery_quarantined=self.executor.recovery_quarantined -+ return executor -+ - 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/batch.py b/src/msg/core/batch.py -new file mode 100644 -index 0000000..aea934e ---- /dev/null -+++ b/src/msg/core/batch.py -@@ -0,0 +1,34 @@ -+"""One bounded child-admission policy for both named batch primitives.""" -+from __future__ import annotations -+ -+from .codec import canonical -+from .errors import require -+from .execution_ports import PacketDecoder -+from .requests import SECRET_DELIVERY_MIN_VERSION -+ -+# Response-only credentials require their own non-batched delivery call. -+NO_BATCH = frozenset(SECRET_DELIVERY_MIN_VERSION) | {'identity.register', 'identity.upgrade'} -+ -+ -+class BatchPolicy: -+ def __init__(self, registry, decoder: PacketDecoder | None, max_request_bytes: int): -+ self.registry = registry -+ self.decoder = decoder -+ self.max_request_bytes = max_request_bytes -+ -+ def packets(self, request, subject, *, bounded=True): -+ require(self.decoder is not None, 'batch_decoder_not_configured') -+ # Independent batches historically use the validated parent envelope size. -+ max_bytes = self.max_request_bytes if bounded else len(canonical(request)) -+ values = [] -+ for value in request.arguments['requests']: -+ packet = self.decoder(value, max_bytes) -+ spec = self.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/events.py b/src/msg/core/events.py -new file mode 100644 -index 0000000..aa28f84 ---- /dev/null -+++ b/src/msg/core/events.py -@@ -0,0 +1,6 @@ -+"""Stable request Event identity shared by execution and domain adapters.""" -+from .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 0000000..b9798a3 ---- /dev/null -+++ b/src/msg/core/execution_ports.py -@@ -0,0 +1,19 @@ -+"""Narrow trusted assembly dependencies, not resource-configurable hooks.""" -+from __future__ import annotations -+ -+from typing import Protocol -+from .contracts import MetadataSession -+from .models import Event, ExecutionContext, JsonMap, OperationRequest, ResourceRef -+ -+ -+class PacketDecoder(Protocol): -+ def __call__(self, value: object, max_bytes: int, /) -> OperationRequest: ... -+ -+ -+class ProjectionReader(Protocol): -+ async def __call__(self, context: ExecutionContext, request: OperationRequest, -+ session: MetadataSession, ref: ResourceRef, /) -> JsonMap: ... -+ -+ -+class EventProjector(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 d00ac08..ba27481 100644 ---- a/src/msg/core/executor.py -+++ b/src/msg/core/executor.py -@@ -5,6 +5,9 @@ import time - from dataclasses import replace - from uuid import uuid4 - from msg.core.codec import wire,digest -+from msg.core.batch import BatchPolicy -+from msg.core.events import event_id -+from msg.core.execution_ports import PacketDecoder, ProjectionReader, EventProjector - 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 +16,19 @@ log=logging.getLogger(__name__) - - - 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=1048576, packet_decoder: PacketDecoder | None = None, -+ projection_reader: ProjectionReader | None = None, -+ event_projector: EventProjector | 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 -+ self.batch_policy=BatchPolicy(registry,packet_decoder,max_request_bytes) -+ self.projection_reader=projection_reader -+ self.event_projector=event_projector - - def recovery_drill_active(self): - marker=self.recovery_drill_marker -@@ -49,9 +57,7 @@ class OperationExecutor: - 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.batch_policy.packets(request,principal.subject) - 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,10 +100,9 @@ class OperationExecutor: - 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, -- fields=request.return_fields) for ref in output.resources] -+ require(output.resources and self.projection_reader is not None,'projection_unavailable') -+ projections=[await self.projection_reader(context,request,session,ref) -+ for ref in output.resources] - output=replace(output,data={**(output.data or {}),'projection':projections}) - if audited: - from msg.core.models import AuditEvent,ResourceRef -@@ -114,14 +119,12 @@ class OperationExecutor: - 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_projector is not None: -+ await self.event_projector(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 +153,10 @@ class OperationExecutor: - 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=self.batch_policy.packets(request,principal.subject,bounded=False) - 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) -diff --git a/src/msg/extensions/ssh.py b/src/msg/extensions/ssh.py -index d12baad..c855eea 100644 ---- a/src/msg/extensions/ssh.py -+++ b/src/msg/extensions/ssh.py -@@ -21,7 +21,7 @@ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey - 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.executor 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,8 @@ 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 -+ # Use the same finite services and recovery fences with SSH authentication. -+ executor = app.make_executor(authenticator=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/plugins/batch.py b/src/msg/plugins/batch.py -index beb05db..327edfb 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.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 -- -+from msg.transports.packet import REQUEST_SCHEMA - - def install(app): - op,finish=registration(app,'batch',('identity',)) -@@ -35,7 +14,7 @@ def install(app): - @op('batch.atomic',schema) - async def atomic(ctx,request,tx): - await app.authorizer._ceiling(ctx.principal,operation_id(request),ctx.principal.subject,tx) -- children=packets(app.registry,request,ctx.principal.subject,app.settings.server.limits.max_request_bytes) -+ children=app.executor.batch_policy.packets(request,ctx.principal.subject) - results=[];resources=[] - for index,packet in enumerate(children): - result=await app.executor.execute(packet,entry=ctx.entry) -diff --git a/src/msg/plugins/communication.py b/src/msg/plugins/communication.py -index d704aa7..579350a 100644 ---- a/src/msg/plugins/communication.py -+++ b/src/msg/plugins/communication.py -@@ -11,15 +11,13 @@ from msg.core.codec import wire,canonical,loads,decode,digest,parse_time,b64,unb - from msg.core.errors import Failure,require - from msg.core.models import HandlerOutput,ResourceRef,ResourceTypeSpec,EmailSettings,EffectJob,Principal - from msg.core.requests import signing_bytes -+from msg.core.events import event_id - from msg.plugins.common import * - from msg.plugins.schemas import * - from msg.plugins.discovery import visible - from msg.storage.capacity import require_webhook_capacity - - --def event_id(request,subject): -- return 'e_'+digest((subject,request.request_id))[7:39] -- - - # A deliberately small event vocabulary. An Event is an audit fact, not blanket - # permission to disclose its data to an external receiver. -diff --git a/tests/issue_78_80/conftest.py b/tests/issue_78_80/conftest.py -index ada0f4a..954c25d 100644 ---- a/tests/issue_78_80/conftest.py -+++ b/tests/issue_78_80/conftest.py -@@ -125,9 +125,16 @@ async def harness(tmp_path): - 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)) -+ from msg.transports.packet import decode_packet -+ -+ async def project(context, request, tx, ref): -+ return await discovery.read_projection(app,context,request,tx,ref.id, -+ revision=ref.revision,fields=request.return_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, -+ packet_decoder=decode_packet,projection_reader=project) - app._loaded=True - - def packet(operation,args,version=1, *, signed=True,**kwargs): diff --git a/src/msg/application.py b/src/msg/application.py index 591920d2..df294fba 100644 --- a/src/msg/application.py +++ b/src/msg/application.py @@ -119,15 +119,37 @@ 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.make_executor() 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 make_executor(self, *, authenticator=None): + """Compose the same finite services for network and SSH execution.""" + from functools import partial + from msg.plugins.discovery import read_projection + from msg.plugins.communication import enqueue_domain_webhooks + from msg.transports.packet import decode_packet + + async def projection(context, request, tx, ref): + return await read_projection(self,context,request,tx,ref.id, + revision=ref.revision,fields=request.return_fields) + + 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, + packet_decoder=decode_packet,projection_reader=projection, + event_projector=partial(enqueue_domain_webhooks,self)) + 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_drill_marker=self.executor.recovery_drill_marker + executor.recovery_quarantined=self.executor.recovery_quarantined + return executor + 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/batch.py b/src/msg/core/batch.py new file mode 100644 index 00000000..aea934ec --- /dev/null +++ b/src/msg/core/batch.py @@ -0,0 +1,34 @@ +"""One bounded child-admission policy for both named batch primitives.""" +from __future__ import annotations + +from .codec import canonical +from .errors import require +from .execution_ports import PacketDecoder +from .requests import SECRET_DELIVERY_MIN_VERSION + +# Response-only credentials require their own non-batched delivery call. +NO_BATCH = frozenset(SECRET_DELIVERY_MIN_VERSION) | {'identity.register', 'identity.upgrade'} + + +class BatchPolicy: + def __init__(self, registry, decoder: PacketDecoder | None, max_request_bytes: int): + self.registry = registry + self.decoder = decoder + self.max_request_bytes = max_request_bytes + + def packets(self, request, subject, *, bounded=True): + require(self.decoder is not None, 'batch_decoder_not_configured') + # Independent batches historically use the validated parent envelope size. + max_bytes = self.max_request_bytes if bounded else len(canonical(request)) + values = [] + for value in request.arguments['requests']: + packet = self.decoder(value, max_bytes) + spec = self.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/events.py b/src/msg/core/events.py new file mode 100644 index 00000000..aa28f84e --- /dev/null +++ b/src/msg/core/events.py @@ -0,0 +1,6 @@ +"""Stable request Event identity shared by execution and domain adapters.""" +from .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..b9798a35 --- /dev/null +++ b/src/msg/core/execution_ports.py @@ -0,0 +1,19 @@ +"""Narrow trusted assembly dependencies, not resource-configurable hooks.""" +from __future__ import annotations + +from typing import Protocol +from .contracts import MetadataSession +from .models import Event, ExecutionContext, JsonMap, OperationRequest, ResourceRef + + +class PacketDecoder(Protocol): + def __call__(self, value: object, max_bytes: int, /) -> OperationRequest: ... + + +class ProjectionReader(Protocol): + async def __call__(self, context: ExecutionContext, request: OperationRequest, + session: MetadataSession, ref: ResourceRef, /) -> JsonMap: ... + + +class EventProjector(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..ba274817 100644 --- a/src/msg/core/executor.py +++ b/src/msg/core/executor.py @@ -5,6 +5,9 @@ from dataclasses import replace from uuid import uuid4 from msg.core.codec import wire,digest +from msg.core.batch import BatchPolicy +from msg.core.events import event_id +from msg.core.execution_ports import PacketDecoder, ProjectionReader, EventProjector 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 +16,19 @@ 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=1048576, packet_decoder: PacketDecoder | None = None, + projection_reader: ProjectionReader | None = None, + event_projector: EventProjector | 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 + self.batch_policy=BatchPolicy(registry,packet_decoder,max_request_bytes) + self.projection_reader=projection_reader + self.event_projector=event_projector def recovery_drill_active(self): marker=self.recovery_drill_marker @@ -49,9 +57,7 @@ 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.batch_policy.packets(request,principal.subject) 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,10 +100,9 @@ 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, - fields=request.return_fields) for ref in output.resources] + require(output.resources and self.projection_reader is not None,'projection_unavailable') + projections=[await self.projection_reader(context,request,session,ref) + for ref in output.resources] output=replace(output,data={**(output.data or {}),'projection':projections}) if audited: from msg.core.models import AuditEvent,ResourceRef @@ -114,14 +119,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_projector is not None: + await self.event_projector(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 +153,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=self.batch_policy.packets(request,principal.subject,bounded=False) 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) diff --git a/src/msg/extensions/ssh.py b/src/msg/extensions/ssh.py index d12baadc..c855eea2 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.executor 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,8 @@ 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 + # Use the same finite services and recovery fences with SSH authentication. + executor = app.make_executor(authenticator=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/plugins/batch.py b/src/msg/plugins/batch.py index beb05dbf..327edfb2 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.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 - +from msg.transports.packet import REQUEST_SCHEMA def install(app): op,finish=registration(app,'batch',('identity',)) @@ -35,7 +14,7 @@ def install(app): @op('batch.atomic',schema) async def atomic(ctx,request,tx): await app.authorizer._ceiling(ctx.principal,operation_id(request),ctx.principal.subject,tx) - children=packets(app.registry,request,ctx.principal.subject,app.settings.server.limits.max_request_bytes) + children=app.executor.batch_policy.packets(request,ctx.principal.subject) results=[];resources=[] for index,packet in enumerate(children): result=await app.executor.execute(packet,entry=ctx.entry) diff --git a/src/msg/plugins/communication.py b/src/msg/plugins/communication.py index d704aa7e..579350ac 100644 --- a/src/msg/plugins/communication.py +++ b/src/msg/plugins/communication.py @@ -11,15 +11,13 @@ from msg.core.errors import Failure,require from msg.core.models import HandlerOutput,ResourceRef,ResourceTypeSpec,EmailSettings,EffectJob,Principal from msg.core.requests import signing_bytes +from msg.core.events import event_id from msg.plugins.common import * from msg.plugins.schemas import * from msg.plugins.discovery import visible from msg.storage.capacity import require_webhook_capacity -def event_id(request,subject): - return 'e_'+digest((subject,request.request_id))[7:39] - # A deliberately small event vocabulary. An Event is an audit fact, not blanket # permission to disclose its data to an external receiver. diff --git a/tests/issue_78_80/conftest.py b/tests/issue_78_80/conftest.py index ada0f4a7..954c25d2 100644 --- a/tests/issue_78_80/conftest.py +++ b/tests/issue_78_80/conftest.py @@ -125,9 +125,16 @@ 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)) + from msg.transports.packet import decode_packet + + async def project(context, request, tx, ref): + return await discovery.read_projection(app,context,request,tx,ref.id, + revision=ref.revision,fields=request.return_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, + packet_decoder=decode_packet,projection_reader=project) app._loaded=True def packet(operation,args,version=1, *, signed=True,**kwargs): From f7f8a890d91e4eda5d3da5f3c2ac4d8988e70848 Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Tue, 29 Sep 2026 05:02:12 +0800 Subject: [PATCH 6/7] docs(core): record verified RED evidence and the exact executor implementation Refs #156 #160 #83 #85. Keep final-head full verification and production acceptance explicitly pending. --- docs/ARCHITECTURE_PROGRESS_20260929.md | 40 ++++++++++++++------------ docs/EXECUTOR_PROGRESS_20260929.md | 28 +++++++++--------- 2 files changed, 36 insertions(+), 32 deletions(-) diff --git a/docs/ARCHITECTURE_PROGRESS_20260929.md b/docs/ARCHITECTURE_PROGRESS_20260929.md index 78c089be..cce8dc1a 100644 --- a/docs/ARCHITECTURE_PROGRESS_20260929.md +++ b/docs/ARCHITECTURE_PROGRESS_20260929.md @@ -4,29 +4,31 @@ - Repository: TokenNotIncluded/msg.lmm.best; coordinator: #83; architecture umbrella: #85. - Starting main: `476e64f04ed1d9c7dcab4a41c4f3cdb72faad6a7`, tree `40ca207a3c49be347120e058e47ed328ec4e3098`. -- Independent branch: `fix/architecture-boundaries-20260929`. Do not write or force-push the recovery branch for #155. -- Main CI runs 36472507327 (complete contracts) and 36472507300 (recovery) were read as success. Shard-0 source artifact 10992168743 was downloaded and its SHA-256 verified as `9abbc69988b8ffcb45f9931dc87e291145a7680f275bf7aadbedb7bed2189d10`. This is baseline evidence, not evidence for new changes. -- Follow the authoritative design and valid completed archive linked by #83. Preserve historical schemas/signatures, current authorization, one transaction publisher, quarantine and lease fences. +- Main CI runs 36472507327 (complete contracts) and 36472507300 (recovery) were read as success. Shard-0 source artifact 10992168743 was downloaded and hash verified: `9abbc69988b8ffcb45f9931dc87e291145a7680f275bf7aadbedb7bed2189d10`. Baseline evidence does not validate later changes. +- Follow the authoritative design and valid completed archive linked by #83. Preserve historical schemas/signatures, current authorization, one transaction publisher, quarantine and lease fences. Never force-push another work stream or treat an old GREEN as the current combination. -## Active slice: #160 +## #160 — tool runner boundary, PR #165 -Observed: `ToolExecutor.invoke(...) -> ResourceRef` has no consumer; the worker actually injects an async callable returning a local `ToolResult`. That result is owned by `workers.effects`, forcing the sandbox adapter to import its orchestrator. +Implemented on `fix/architecture-boundaries-20260929` at **f14fd1533f70983b046df4b4c813993e8a536d12**, tree **8a62150d8273a2c75775872694749cb529835966**. -1. Regression-first commit: new ownership/import/type/assembly tests plus real PostgreSQL worker success and invalid-result refusal. Existing tool and completion-fence tests remain unchanged. -2. Cloud-only execution: `.github/workflows/architecture-boundaries.yml` records exact commit/tree and uploads JUnit; it does not replace the full four-shard/conformance/build gate. -3. Planned minimal change: neutral `core.tool_execution` owns a callable ToolRunner and local ToolResult; worker and sandbox consume them. Keep the existing effects.ToolResult import as the same type. Retire the unused invoke declaration explicitly, retaining ToolExecutor only as an import alias for the actual runner port. -4. Pending: read the red run, implement, read focused and full final-head evidence, review diff and merge only if the exact combination is verified. +The neutral `core.tool_execution` owns callable ToolRunner and local ToolResult. EffectWorker and BubblewrapRunner share the actual contract. The effects.ToolResult export remains the same type; ToolExecutor is an explicit compatibility alias, not a second unused invoke-to-ResourceRef contract. Existing worker authorization, mandatory sandbox, output and completion fences are unchanged. See `TOOL_RUNNER_BOUNDARY.md`. -No implementation success or issue closure is claimed by this regression-first note. +- RED commit b553c59b484a4e5a383df3500ad9e4cab9d51272; run **36480982432**: **7 failed / 31 passed**. Old tools/fence tests passed. +- Focused GREEN PR run **36481370662**: downloaded artifact **10996748629**, SHA-256 `6a86edb23bac1f6463a65ca9b1118a9c54b0b25c274b81539defdc8ee0e03e9f`. Independently checked JUnit **38 tests / 0 failures / 0 errors / 0 skips**. Synthetic merge 0af4b1910bd31615e0203ae1cce5571b8a048142 has the exact implementation tree above. +- Recovery safety run **36481370672** was read as success. Full run **36481370643** still in progress at this checkpoint. No merge or closure is claimed by this note; refresh final PR/base/CI before acting. -## Queue and separate work +## #156 — executor boundary -- #156: executor currently imports batch, discovery and communication helpers and retains Application; needs explicit narrow assembly dependencies, not just one renamed import. -- #157: PostgreSQL inherits concrete SqliteSession; separate genuinely neutral session behavior without changing SQL/locks/schema. -- #158: client imports server executor/Git/vault helpers; minimal-client packaging remains a distinct acceptance requirement. -- #159: market/plugin bidirectional helper ownership requires preserving all published order versions and the sole protected ledger path. -- #155: live head observed `0e4dbe777053e354bcf4845b851a0c9b86ab8310`; title says verifier fix submitted while body still describes older failed head. Do not infer success; refresh its actual diff/CI and resolve main conflicts in a separate serial integration. -- Functional umbrella issues #68–#83/#85 remain open unless every remaining acceptance item has exact evidence. Existing completed functionality must not be rewritten from stale issue descriptions. -- #64/#65/#84 need authorized production logs/configuration, real protected snapshots and field/capacity evidence; #68/#69/#70 also retain external trust/retirement/console requirements. Code CI cannot supply those facts. +Separate stacked branch `fix/executor-boundary-20260929`. RED run **36481661656** independently verified **9 failed / 25 passed / 0 errors / 0 skips**. Implementation applied at c639a520ab69e22cccaabb07999d7cb54b8be4ca and final-head verification remains pending. See `EXECUTOR_PROGRESS_20260929.md` for exact artifacts and ownership. This removes all three categories of core-to-plugin dependency, not just one renamed import. -No deployment, production Root/PIN, real money, outbound notification, key destruction or recovery promotion is authorized or performed. +## Other queue and parallel work + +- #157 is being handled separately by PR **#166**, metadata-session invariants versus concrete SQLite/PostgreSQL adapters. Do not duplicate or overwrite that branch. +- #158 now has independent branch `fix/client-boundary-20260929` with regression-first client/wheel/import tests at 6ea463d6fa8b256cb07d0ede2d7795a5336eda4f. Minimal packaging and actual four transport calls must be proved; no client-boundary success is claimed here. +- #159: market/plugin helper ownership remains open; preserve published order versions and sole protected ledger path. +- New #161–#163 need their own source/overlap review; do not infer they are resolved by the above module moves. +- #155 recovery head last read as 0e4dbe777053e354bcf4845b851a0c9b86ab8310 while body retained an older failure. Refresh actual diff/review/CI; do not infer success or rewrite its active branch. +- Functional umbrellas #68–#83/#85 remain open unless every remaining acceptance item has exact evidence. Do not reconstruct completed functionality from stale issue bodies. +- #64/#65/#84 need authorized production logs/configuration, real protected snapshots and field/capacity evidence. #68/#69/#70 also retain external trust/retirement/console requirements. Code CI cannot supply those facts. + +No deployment, production Root/PIN, real money, outbound notification, key destruction or recovery promotion is authorized or performed. All project execution/testing in this work stream is cloud-only; local work is source inspection/editing and artifact verification. diff --git a/docs/EXECUTOR_PROGRESS_20260929.md b/docs/EXECUTOR_PROGRESS_20260929.md index 156b1cd9..6493ffa8 100644 --- a/docs/EXECUTOR_PROGRESS_20260929.md +++ b/docs/EXECUTOR_PROGRESS_20260929.md @@ -1,21 +1,23 @@ # Executor architecture progress — 2026-09-29 -Refs #156/#83/#85. Independent stacked branch `fix/executor-boundary-20260929`, starting from tool boundary implementation `f14fd1533f70983b046df4b4c813993e8a536d12` (PR #165). Do not overwrite #165 or #155 branches. This note supplements `ARCHITECTURE_PROGRESS_20260929.md`. +Refs #156/#83/#85. Independent stacked branch `fix/executor-boundary-20260929`, starting from tool boundary implementation `f14fd1533f70983b046df4b4c813993e8a536d12` (PR #165). Do not overwrite #165/#155/#166 branches. This note supplements `ARCHITECTURE_PROGRESS_20260929.md`. -## Regression-first +## Evidence and current state -New tests expose core imports of plugin implementations and the Application backreference, inconsistent standalone/factory assembly, and missing narrow services reported as internal errors. Real PostgreSQL tests retain batch idempotency, write projections, and verify Event projection failure rolls back both business state and Event/result in the existing transaction. Unchanged batch secret-delivery, webhook domain, SSH and transfer rollback tests run alongside them in `.github/workflows/executor-boundary.yml`. +- RED: `1953beeda08cb34570f2c56535c4e4b1249546eb`, tree `28cd98e8d3cb1e54df9714efc29ea06a41107898`, cloud run **36481661656**. Downloaded artifact **10996698881**, SHA-256 `0dfce2e01becb21a5660017ddc017d2ad3102cdc7cb98b99c86a42bdf1f1e8e9`. Source JSON and JUnit independently inspected: **9 failed / 25 passed / 0 errors / 0 skips**. All failures were the new boundary/assembly/error tests; existing batch, secret-release, domain-webhook, SSH and transfer-rollback tests passed. +- Reviewed implementation patch SHA-256: `0389f11f5609abc7c54ecf67159b6aae75245d81c94611c8a4ecf7017f3ed3dc`. Cloud run **36482413751** applied only the nine enumerated paths to the fixed parent and non-force updated this feature branch. Its final commit is **c639a520ab69e22cccaabb07999d7cb54b8be4ca**, tree **07ad9c86bf26b7d9e3b294200ebd9709f7e5f888**. +- Patch-result artifact **10998370091**, SHA-256 `9914220c3ff71d1dcca02491a3cada0dd95e4babf06b1838f9b09298a827a4e9`, was downloaded. Its source, path allowlist and patch hash match the reviewed change. The one-shot publisher and patch are removed from the final tree and remain inspectable in history. No project code was executed by that write-enabled job. +- This note commit follows the implementation. Focused/full CI must run against the resulting final head and actual main combination. **GREEN is not yet claimed.** Existing full four-shard/conformance/build CI remains mandatory and unchanged. No project tests were run locally. -This commit only introduces tests/workflow/notes. No success is claimed before cloud execution. Existing full four-shard/conformance/build CI remains mandatory and unchanged. +## Implemented ownership -## Planned ownership +- `core.batch.BatchPolicy` owns the single child-admission algorithm and consumes an explicit bounded `PacketDecoder`. Atomic and independent batch primitives retain their existing size/idempotency/secret-delivery rules. +- `core.events.event_id` owns the unchanged deterministic digest. Communication keeps a compatibility import of that same function. +- `OperationExecutor` receives packet decoder/byte limit, `ProjectionReader` and in-transaction `EventProjector`. It has no Application/settings backreference and no imports of business plugins or transport implementations. +- `Application.make_executor` is the composition root. Normal and forced SSH execution use it with their respective authenticator and existing recovery/secret-release fences. A standalone executor can use its own explicit services. +- Missing decoder/projection is a stable fail-closed error, not an internal error or implicit authorization fallback. Event callback failure still rolls back the existing business/Event/result transaction; no separate writer, transaction or configurable hook engine was added. +- The test-only issue_78_80 harness now provides its decoder and projection explicitly. No existing behavioral assertion was deleted or weakened. -- Core BatchPolicy owns the single child admission algorithm and consumes an explicit bounded packet decoder; core must not import a transport or plugin to find one. -- Core event identity owns the existing deterministic digest; communication keeps only a compatibility import. -- OperationExecutor receives a packet decoder/byte limit, read-projection function and in-transaction Event projection function. It holds no Application/settings backreference. -- Application is the composition root for those dependencies; normal and SSH entrypoints use the same factory with their own authenticator and unchanged recovery/secret-delivery fences. -- Missing decoder/projection is a stable fail-closed error, not an internal error or implicit authorization fallback. Event delivery effects stay in the existing transaction; this is not a configurable workflow/hook engine. +## Remaining gates -## Remaining evidence - -Read RED result, implement without weakening old assertions, check focused/full final head and reviewed exact combination. #156 remains open until those gates pass. No production/log/snapshot/Root/PIN/funds/outbound/recovery promotion work is performed. Production-blocked issues remain blocked. +Read focused/full final-head artifacts, review the final diff and exact merge tree, then decide #156 closure. #165 must be integrated first or included in that exact tested combination. #155 recovery and #166 metadata-session changes require their own reviewed integration. No production/log/snapshot/Root/PIN/funds/outbound/recovery-promotion work was performed; production-blocked issues remain blocked. From 3db973ab1784926eb4e545cabdf832e7faa0a4bc Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Tue, 29 Sep 2026 05:11:24 +0800 Subject: [PATCH 7/7] fix(core): preserve live event projection binding and compare frozen JSON correctly Cloud run36483396339 exposed 2 failures; retain the old webhook fault-injection/rollback test unchanged and fix the assembly binding. The new standalone test compares the complete wire value, not list against frozen tuple. Refs #156 #83 #85. --- docs/EXECUTOR_PROGRESS_20260929.md | 24 ++++++++++++------------ src/msg/application.py | 10 +++++++--- tests/test_executor_boundary.py | 3 ++- 3 files changed, 21 insertions(+), 16 deletions(-) diff --git a/docs/EXECUTOR_PROGRESS_20260929.md b/docs/EXECUTOR_PROGRESS_20260929.md index 6493ffa8..bdb5eee1 100644 --- a/docs/EXECUTOR_PROGRESS_20260929.md +++ b/docs/EXECUTOR_PROGRESS_20260929.md @@ -1,23 +1,23 @@ # Executor architecture progress — 2026-09-29 -Refs #156/#83/#85. Independent stacked branch `fix/executor-boundary-20260929`, starting from tool boundary implementation `f14fd1533f70983b046df4b4c813993e8a536d12` (PR #165). Do not overwrite #165/#155/#166 branches. This note supplements `ARCHITECTURE_PROGRESS_20260929.md`. +Refs #156/#83/#85. Branch `fix/executor-boundary-20260929`, PR #169. Its prerequisite #165 was merged as 08c58928ee3f6c6cb350d9e9fba75d072bd78f08 with exact tested tool tree 8a62150d8273a2c75775872694749cb529835966; #160 is closed. Do not overwrite #155/#166 or other work streams. ## Evidence and current state -- RED: `1953beeda08cb34570f2c56535c4e4b1249546eb`, tree `28cd98e8d3cb1e54df9714efc29ea06a41107898`, cloud run **36481661656**. Downloaded artifact **10996698881**, SHA-256 `0dfce2e01becb21a5660017ddc017d2ad3102cdc7cb98b99c86a42bdf1f1e8e9`. Source JSON and JUnit independently inspected: **9 failed / 25 passed / 0 errors / 0 skips**. All failures were the new boundary/assembly/error tests; existing batch, secret-release, domain-webhook, SSH and transfer-rollback tests passed. -- Reviewed implementation patch SHA-256: `0389f11f5609abc7c54ecf67159b6aae75245d81c94611c8a4ecf7017f3ed3dc`. Cloud run **36482413751** applied only the nine enumerated paths to the fixed parent and non-force updated this feature branch. Its final commit is **c639a520ab69e22cccaabb07999d7cb54b8be4ca**, tree **07ad9c86bf26b7d9e3b294200ebd9709f7e5f888**. -- Patch-result artifact **10998370091**, SHA-256 `9914220c3ff71d1dcca02491a3cada0dd95e4babf06b1838f9b09298a827a4e9`, was downloaded. Its source, path allowlist and patch hash match the reviewed change. The one-shot publisher and patch are removed from the final tree and remain inspectable in history. No project code was executed by that write-enabled job. -- This note commit follows the implementation. Focused/full CI must run against the resulting final head and actual main combination. **GREEN is not yet claimed.** Existing full four-shard/conformance/build CI remains mandatory and unchanged. No project tests were run locally. +- RED: 1953beeda08cb34570f2c56535c4e4b1249546eb, tree 28cd98e8d3cb1e54df9714efc29ea06a41107898, cloud run36481661656. Artifact10996698881 SHA-256 `0dfce2e01becb21a5660017ddc017d2ad3102cdc7cb98b99c86a42bdf1f1e8e9`; independently inspected source/JUnit: **9 failed / 25 passed / 0 errors / 0 skips**. +- Reviewed implementation patch SHA-256 `0389f11f5609abc7c54ecf67159b6aae75245d81c94611c8a4ecf7017f3ed3dc` was applied by cloud run36482413751 only to its nine enumerated source paths. Generated commit c639a520ab69e22cccaabb07999d7cb54b8be4ca/tree07ad9c86bf26b7d9e3b294200ebd9709f7e5f888. Artifact10998370091 SHA-256 `9914220c3ff71d1dcca02491a3cada0dd95e4babf06b1838f9b09298a827a4e9` matched source, paths and patch. The temporary patch publisher and patch are absent from the final tree; it did not execute project code with write credentials. +- First implementation/note head f7f8a890d91e4eda5d3da5f3c2ac4d8988e70848/treea9e18de677098798cc9f518cdd9ccb4c1a82c336: focused run36483396339 produced **2 failed / 32 passed / 0 errors / 0 skips**. Artifact10997282312 SHA-256 `d726cd26b800c6415ef7f0edc8c94f5a39f7509036fa66c8590f0cf9d79c26ca`, source/JUnit independently inspected. Recovery36483396239 and tool-boundary36483396136 succeeded; full gate was still queued. That head is **not GREEN**. +- Two precise corrections follow: the new standalone test compares the entire wire representation because models freeze nested lists/maps; the composition root uses a live module binding for Event projection rather than capturing a function with partial, retaining the existing post-load fault-injection/instrumentation contract. Existing webhook rollback assertions are untouched. This commit requires a new final-head cloud run; success is not yet claimed. ## Implemented ownership -- `core.batch.BatchPolicy` owns the single child-admission algorithm and consumes an explicit bounded `PacketDecoder`. Atomic and independent batch primitives retain their existing size/idempotency/secret-delivery rules. -- `core.events.event_id` owns the unchanged deterministic digest. Communication keeps a compatibility import of that same function. -- `OperationExecutor` receives packet decoder/byte limit, `ProjectionReader` and in-transaction `EventProjector`. It has no Application/settings backreference and no imports of business plugins or transport implementations. -- `Application.make_executor` is the composition root. Normal and forced SSH execution use it with their respective authenticator and existing recovery/secret-release fences. A standalone executor can use its own explicit services. -- Missing decoder/projection is a stable fail-closed error, not an internal error or implicit authorization fallback. Event callback failure still rolls back the existing business/Event/result transaction; no separate writer, transaction or configurable hook engine was added. -- The test-only issue_78_80 harness now provides its decoder and projection explicitly. No existing behavioral assertion was deleted or weakened. +- `core.batch.BatchPolicy` owns the single child-admission algorithm and consumes explicit bounded `PacketDecoder`. Atomic/independent batch size, idempotency and secret-delivery rules stay unchanged. +- `core.events.event_id` owns the unchanged deterministic digest; communication has a compatibility import of that exact function. +- `OperationExecutor` has explicit decoder/byte-limit, `ProjectionReader` and in-transaction `EventProjector` dependencies. It has no Application/settings backreference or business-plugin/transport implementation imports. +- `Application.make_executor` composes normal and forced-SSH execution with their respective authenticator and existing recovery/secret-release fences. It exposes finite trusted services, not a resource-configured hook engine. +- Missing decoder/projection fails closed with stable errors. Event callback failure rolls back business state, Event and request result in the existing transaction. No second writer/transaction is introduced. +- The issue_78_80 test-only harness supplies its decoder/projection explicitly. No original behavioral assertion was deleted or weakened. ## Remaining gates -Read focused/full final-head artifacts, review the final diff and exact merge tree, then decide #156 closure. #165 must be integrated first or included in that exact tested combination. #155 recovery and #166 metadata-session changes require their own reviewed integration. No production/log/snapshot/Root/PIN/funds/outbound/recovery-promotion work was performed; production-blocked issues remain blocked. +Read corrected focused/full final-head artifacts, review the full diff and exact current-main merge tree, then decide #156 closure. All project execution/tests are cloud-only. No production/log/snapshot/Root/PIN/funds/outbound/recovery-promotion work was performed; production-blocked issues remain blocked. diff --git a/src/msg/application.py b/src/msg/application.py index df294fba..865f8a63 100644 --- a/src/msg/application.py +++ b/src/msg/application.py @@ -128,21 +128,25 @@ async def load(self): def make_executor(self, *, authenticator=None): """Compose the same finite services for network and SSH execution.""" - from functools import partial from msg.plugins.discovery import read_projection - from msg.plugins.communication import enqueue_domain_webhooks + from msg.plugins import communication from msg.transports.packet import decode_packet async def projection(context, request, tx, ref): return await read_projection(self,context,request,tx,ref.id, revision=ref.revision,fields=request.return_fields) + async def project_event(tx, event): + # Keep the installed adapter's live binding for instrumentation and + # fault injection; the executor still knows only its narrow port. + await communication.enqueue_domain_webhooks(self,tx,event) + 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, packet_decoder=decode_packet,projection_reader=projection, - event_projector=partial(enqueue_domain_webhooks,self)) + event_projector=project_event) executor.response_hook=self._secrets_for_caller executor.recovery_drill_marker=self.settings.config_dir/'recovery-drill.json' if self.executor is not None: diff --git a/tests/test_executor_boundary.py b/tests/test_executor_boundary.py index 856a2969..7dec3ac0 100644 --- a/tests/test_executor_boundary.py +++ b/tests/test_executor_boundary.py @@ -101,7 +101,8 @@ async def project(context, request, tx, ref): result = await executor.execute(request_for('testing.read', {}, 'https://unit.invalid', return_fields=('id',))) assert result.status == 'ok', result - assert result.data == {'original': True, 'projection': [{'id': 'r_test'}]} + # Models freeze nested JSON; compare its complete wire representation. + assert wire(result.data) == {'original': True, 'projection': [{'id': 'r_test'}]} assert active == []