Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
80 changes: 80 additions & 0 deletions .github/workflows/architecture-boundaries.yml
Original file line number Diff line number Diff line change
@@ -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/
32 changes: 32 additions & 0 deletions docs/ARCHITECTURE_PROGRESS_20260929.md
Original file line number Diff line number Diff line change
@@ -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.
23 changes: 23 additions & 0 deletions docs/TOOL_RUNNER_BOUNDARY.md
Original file line number Diff line number Diff line change
@@ -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.
10 changes: 5 additions & 5 deletions src/msg/core/contracts.py
Original file line number Diff line number Diff line change
@@ -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: ...
Expand Down Expand Up @@ -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):
Expand Down
28 changes: 28 additions & 0 deletions src/msg/core/tool_execution.py
Original file line number Diff line number Diff line change
@@ -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: ...
24 changes: 9 additions & 15 deletions src/msg/workers/effects.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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')
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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
Expand Down
6 changes: 4 additions & 2 deletions src/msg/workers/sandbox.py
Original file line number Diff line number Diff line change
Expand Up @@ -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]
Expand Down
131 changes: 131 additions & 0 deletions tests/test_tool_runner_contract.py
Original file line number Diff line number Diff line change
@@ -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]
Loading