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/.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/ARCHITECTURE_PROGRESS_20260929.md b/docs/ARCHITECTURE_PROGRESS_20260929.md new file mode 100644 index 00000000..cce8dc1a --- /dev/null +++ b/docs/ARCHITECTURE_PROGRESS_20260929.md @@ -0,0 +1,34 @@ +# 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`. +- 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. + +## #160 — tool runner boundary, PR #165 + +Implemented on `fix/architecture-boundaries-20260929` at **f14fd1533f70983b046df4b4c813993e8a536d12**, tree **8a62150d8273a2c75775872694749cb529835966**. + +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`. + +- 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. + +## #156 — executor boundary + +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. + +## 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 new file mode 100644 index 00000000..bdb5eee1 --- /dev/null +++ b/docs/EXECUTOR_PROGRESS_20260929.md @@ -0,0 +1,23 @@ +# Executor architecture progress — 2026-09-29 + +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 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 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 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/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/application.py b/src/msg/application.py index 591920d2..865f8a63 100644 --- a/src/msg/application.py +++ b/src/msg/application.py @@ -119,15 +119,41 @@ 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 msg.plugins.discovery import read_projection + 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=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: + 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/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/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/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/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/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] 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): diff --git a/tests/test_executor_boundary.py b/tests/test_executor_boundary.py new file mode 100644 index 00000000..7dec3ac0 --- /dev/null +++ b/tests/test_executor_boundary.py @@ -0,0 +1,190 @@ +"""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 + # Models freeze nested JSON; compare its complete wire representation. + assert wire(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 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]