From ec1c270a9e0a676c776fd2fa6910c96a5efdb255 Mon Sep 17 00:00:00 2001 From: Matthew Wesney Date: Thu, 23 Jul 2026 15:09:54 -0400 Subject: [PATCH] Wire authenticated recipe worker broker loop --- backend/cortex_backend/execution/__init__.py | 12 + backend/cortex_backend/execution/broker.py | 11 +- .../execution/native_launcher.py | 4 + .../execution/worker_protocol.py | 204 +++++--- .../execution/worker_runtime.py | 457 ++++++++++++++++++ ...bility-tiered-agentic-execution-harness.md | 3 +- docs/adr/0001-phase2-broker-contract.md | 4 +- docs/adr/0001-phase2-evidence.md | 10 +- docs/adr/0001-phase2-native-broker.md | 10 +- docs/adr/0001-phase2-native-launcher.md | 10 +- docs/adr/0001-phase2-sandbox-qualification.md | 6 +- docs/adr/0001-phase2-worker-protocol.md | 40 +- packaging/recipe_worker/README.md | 11 +- packaging/recipe_worker/recipe_worker.py | 100 +++- packaging/recipe_worker/recipe_worker.spec | 2 + tests/test_phase2_native_launcher.py | 4 + tests/test_phase2_worker_runtime.py | 387 +++++++++++++++ tools/execution_spikes/README.md | 8 +- .../native_launcher_qualification.py | 2 +- .../recipe_sandbox_qualification.py | 2 +- 20 files changed, 1157 insertions(+), 130 deletions(-) create mode 100644 backend/cortex_backend/execution/worker_runtime.py create mode 100644 tests/test_phase2_worker_runtime.py diff --git a/backend/cortex_backend/execution/__init__.py b/backend/cortex_backend/execution/__init__.py index 91a9dec..321a6fd 100644 --- a/backend/cortex_backend/execution/__init__.py +++ b/backend/cortex_backend/execution/__init__.py @@ -90,6 +90,13 @@ verify_active_worker, verify_installed_worker, ) +from .worker_runtime import ( + DEFAULT_MAX_MESSAGES, + DEFAULT_WATCHDOG_TIMEOUT_MS, + RecipeWorkerBrokerRuntime, + WorkerRuntimeError, + WorkerRuntimeReport, +) from .recipes import ( CalculatorPlan, CheckPlan, @@ -195,6 +202,11 @@ "WorkerProvenanceError", "verify_active_worker", "verify_installed_worker", + "DEFAULT_MAX_MESSAGES", + "DEFAULT_WATCHDOG_TIMEOUT_MS", + "RecipeWorkerBrokerRuntime", + "WorkerRuntimeError", + "WorkerRuntimeReport", "decode_frame", "decode_message", "encode_frame", diff --git a/backend/cortex_backend/execution/broker.py b/backend/cortex_backend/execution/broker.py index aaa8c00..0dc2152 100644 --- a/backend/cortex_backend/execution/broker.py +++ b/backend/cortex_backend/execution/broker.py @@ -36,7 +36,16 @@ ) BrokerDirection = Literal["to_broker", "to_executor"] -BrokerOperation = Literal["prepare", "start", "cancel", "collect"] +BrokerOperation = Literal[ + "prepare", + "input_chunk", + "input_complete", + # Kept for transport-level compatibility; the fixed worker runtime rejects + # this legacy operation instead of guessing whether it means input complete. + "start", + "cancel", + "collect", +] IntegrityLevel = Literal["low", "medium", "high", "system"] diff --git a/backend/cortex_backend/execution/native_launcher.py b/backend/cortex_backend/execution/native_launcher.py index 49d6302..672ae2a 100644 --- a/backend/cortex_backend/execution/native_launcher.py +++ b/backend/cortex_backend/execution/native_launcher.py @@ -366,6 +366,10 @@ def prepare( binding.pipe_name, "--broker-pid", str(binding.broker_process_id), + "--broker-principal", + binding.installation_principal_id, + "--job-id", + binding.job_id, ] ) return NativeWorkerLaunchPlan( diff --git a/backend/cortex_backend/execution/worker_protocol.py b/backend/cortex_backend/execution/worker_protocol.py index ec519af..5aa75cb 100644 --- a/backend/cortex_backend/execution/worker_protocol.py +++ b/backend/cortex_backend/execution/worker_protocol.py @@ -14,6 +14,7 @@ import hashlib import json import re +from threading import RLock from typing import Any, Final, Literal from pydantic import BaseModel, ConfigDict, Field, field_validator @@ -34,6 +35,7 @@ _HASH = re.compile(r"^[0-9a-f]{64}$") _SAFE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,127}$") _SAFE_CODE = re.compile(r"^[a-z][a-z0-9_]{0,63}$") +WorkerOperation = Literal["prepare", "input_chunk", "input_complete", "cancel", "collect"] class WorkerProtocolError(ValueError): @@ -113,6 +115,15 @@ class WorkerPrepare(_WorkerModel): _job_id = field_validator("job_id")(_safe_id) _input_sha256 = field_validator("input_sha256")(_safe_hash) + @field_validator("plan", mode="before") + @classmethod + def _wire_plan_arrays(cls, value: Any) -> Any: + """Accept canonical JSON arrays while keeping the internal plan immutable.""" + + if isinstance(value, dict) and isinstance(value.get("steps"), list): + value = {**value, "steps": tuple(value["steps"])} + return value + class WorkerInputChunk(_WorkerModel): """Append one in-order, independently hashed input chunk.""" @@ -246,8 +257,20 @@ class WorkerError(_WorkerModel): _code = field_validator("code")(_safe_code) +class WorkerAck(_WorkerModel): + """Redacted acknowledgement for a command with no result payload.""" + + schema_version: Literal["recipe.worker.ack.v1"] + request_id: str + job_id: str + acknowledged_operation: WorkerOperation + + _request_id = field_validator("request_id")(_safe_id) + _job_id = field_validator("job_id")(_safe_id) + + class RecipeWorkerSession: - """Bounded state machine used by the future native worker loop. + """Bounded state machine used by the authenticated native worker loop. The session has no filesystem or process capabilities. A native launcher must provide the already-started provider and authenticated transport; this class @@ -264,105 +287,124 @@ def __init__(self, provider: Any) -> None: self._input = bytearray() self._result: RecipeProviderResult | None = None self._cancelled = False + self._lock = RLock() @property def state(self) -> Literal["idle", "receiving", "running", "complete", "cancelled", "failed"]: - return self._state # type: ignore[return-value] + with self._lock: + return self._state # type: ignore[return-value] def prepare(self, message: WorkerPrepare) -> None: - if self._state != "idle": - raise WorkerProtocolError("request_state_invalid") - self._prepare = message - self._input = bytearray() - self._result = None - self._cancelled = False - self._state = "receiving" + with self._lock: + if self._state != "idle": + raise WorkerProtocolError("request_state_invalid") + self._prepare = message + self._input = bytearray() + self._result = None + self._cancelled = False + self._state = "receiving" def input_chunk(self, message: WorkerInputChunk) -> None: - if self._state != "receiving" or self._prepare is None: - raise WorkerProtocolError("request_state_invalid") - if message.request_id != self._prepare.request_id or message.job_id != self._prepare.job_id: - raise WorkerProtocolError("request_identity_mismatch") - if message.offset != len(self._input): - raise WorkerProtocolError("chunk_out_of_order") content = message.decoded() - if len(self._input) + len(content) > self._prepare.input_size: - raise WorkerProtocolError("input_size_exceeded") - self._input.extend(content) + with self._lock: + if self._state != "receiving" or self._prepare is None: + raise WorkerProtocolError("request_state_invalid") + if message.request_id != self._prepare.request_id or message.job_id != self._prepare.job_id: + raise WorkerProtocolError("request_identity_mismatch") + if message.offset != len(self._input): + raise WorkerProtocolError("chunk_out_of_order") + if len(self._input) + len(content) > self._prepare.input_size: + raise WorkerProtocolError("input_size_exceeded") + self._input.extend(content) def cancel(self, message: WorkerCancel) -> None: - if self._prepare is None or message.request_id != self._prepare.request_id: - raise WorkerProtocolError("request_identity_mismatch") - if message.job_id != self._prepare.job_id: - raise WorkerProtocolError("request_identity_mismatch") - if self._state in {"complete", "failed"}: - raise WorkerProtocolError("request_state_invalid") - self._cancelled = True - self._state = "cancelled" - self._input.clear() - self._result = None + with self._lock: + if self._prepare is None or message.request_id != self._prepare.request_id: + raise WorkerProtocolError("request_identity_mismatch") + if message.job_id != self._prepare.job_id: + raise WorkerProtocolError("request_identity_mismatch") + if self._state in {"complete", "failed"}: + raise WorkerProtocolError("request_state_invalid") + self._cancelled = True + self._state = "cancelled" + self._input.clear() + self._result = None + + def _cancel_requested(self) -> bool: + with self._lock: + return self._cancelled def complete_input(self, message: WorkerInputComplete) -> WorkerResult: - if self._state != "receiving" or self._prepare is None: - raise WorkerProtocolError("request_state_invalid") - if message.request_id != self._prepare.request_id or message.job_id != self._prepare.job_id: - raise WorkerProtocolError("request_identity_mismatch") - if message.input_size != self._prepare.input_size or message.input_sha256 != self._prepare.input_sha256: - raise WorkerProtocolError("input_claim_mismatch") - if len(self._input) != self._prepare.input_size: - raise WorkerProtocolError("input_size_mismatch") - if hashlib.sha256(self._input).hexdigest() != self._prepare.input_sha256: - raise WorkerProtocolError("input_hash_mismatch") - self._state = "running" + with self._lock: + if self._state != "receiving" or self._prepare is None: + raise WorkerProtocolError("request_state_invalid") + if message.request_id != self._prepare.request_id or message.job_id != self._prepare.job_id: + raise WorkerProtocolError("request_identity_mismatch") + if message.input_size != self._prepare.input_size or message.input_sha256 != self._prepare.input_sha256: + raise WorkerProtocolError("input_claim_mismatch") + if len(self._input) != self._prepare.input_size: + raise WorkerProtocolError("input_size_mismatch") + if hashlib.sha256(self._input).hexdigest() != self._prepare.input_sha256: + raise WorkerProtocolError("input_hash_mismatch") + prepare = self._prepare + content = bytes(self._input) + self._state = "running" + self._input.clear() try: result = self._provider.transform( - self._prepare.plan, - bytes(self._input), - cancel_check=lambda: self._cancelled, + prepare.plan, + content, + cancel_check=self._cancel_requested, ) except RecipeProviderError as error: - self._state = "cancelled" if error.code == "cancelled" else "failed" - self._input.clear() - raise WorkerProtocolError(error.code) from None + with self._lock: + cancelled = self._cancelled or error.code == "cancelled" + self._state = "cancelled" if cancelled else "failed" + self._input.clear() + raise WorkerProtocolError("cancelled" if cancelled else error.code) from None except Exception: - self._state = "failed" - self._input.clear() - raise WorkerProtocolError("provider_failed") from None - if self._cancelled: - self._state = "cancelled" + with self._lock: + cancelled = self._cancelled + self._state = "cancelled" if cancelled else "failed" + self._input.clear() + raise WorkerProtocolError("cancelled" if cancelled else "provider_failed") from None + with self._lock: + if self._cancelled or self._state == "cancelled": + self._state = "cancelled" + self._input.clear() + raise WorkerProtocolError("cancelled") self._input.clear() - raise WorkerProtocolError("cancelled") - self._input.clear() - self._result = result - self._state = "complete" + self._result = result + self._state = "complete" return WorkerResult.from_provider( result, - request_id=self._prepare.request_id, - job_id=self._prepare.job_id, + request_id=prepare.request_id, + job_id=prepare.job_id, ) def collect(self, message: WorkerCollect) -> WorkerOutputChunk: - if self._state != "complete" or self._prepare is None or self._result is None: - raise WorkerProtocolError("request_state_invalid") - if message.request_id != self._prepare.request_id or message.job_id != self._prepare.job_id: - raise WorkerProtocolError("request_identity_mismatch") - output = self._result.content - if message.offset > len(output): - raise WorkerProtocolError("output_offset_invalid") - end = min(message.offset + message.max_bytes, len(output)) - chunk = output[message.offset:end] - if not chunk: - raise WorkerProtocolError("output_offset_invalid") - encoded = base64.urlsafe_b64encode(chunk).decode("ascii").rstrip("=") - return WorkerOutputChunk( - schema_version="recipe.worker.output_chunk.v1", - request_id=message.request_id, - job_id=message.job_id, - offset=message.offset, - data=encoded, - sha256=hashlib.sha256(chunk).hexdigest(), - final=end == len(output), - ) + with self._lock: + if self._state != "complete" or self._prepare is None or self._result is None: + raise WorkerProtocolError("request_state_invalid") + if message.request_id != self._prepare.request_id or message.job_id != self._prepare.job_id: + raise WorkerProtocolError("request_identity_mismatch") + output = self._result.content + if message.offset > len(output): + raise WorkerProtocolError("output_offset_invalid") + end = min(message.offset + message.max_bytes, len(output)) + chunk = output[message.offset:end] + if not chunk: + raise WorkerProtocolError("output_offset_invalid") + encoded = base64.urlsafe_b64encode(chunk).decode("ascii").rstrip("=") + return WorkerOutputChunk( + schema_version="recipe.worker.output_chunk.v1", + request_id=message.request_id, + job_id=message.job_id, + offset=message.offset, + data=encoded, + sha256=hashlib.sha256(chunk).hexdigest(), + final=end == len(output), + ) class RecipeWorkerDispatcher: @@ -384,6 +426,12 @@ class RecipeWorkerDispatcher: def __init__(self, session: RecipeWorkerSession) -> None: self._session = session + @property + def state(self) -> str: + """Expose only the bounded session state to the broker loop.""" + + return self._session.state + def dispatch(self, operation: str, body: dict[str, Any]) -> WorkerResult | WorkerOutputChunk | None: model_type = self._MODELS.get(operation) if model_type is None or not isinstance(body, dict): @@ -415,10 +463,12 @@ def dispatch(self, operation: str, body: dict[str, Any]) -> WorkerResult | Worke "WorkerCancel", "WorkerCollect", "WorkerError", + "WorkerAck", "WorkerInputChunk", "WorkerInputComplete", "WorkerOutputChunk", "WorkerPrepare", "WorkerProtocolError", "WorkerResult", + "WorkerOperation", ] diff --git a/backend/cortex_backend/execution/worker_runtime.py b/backend/cortex_backend/execution/worker_runtime.py new file mode 100644 index 0000000..e8eb6bb --- /dev/null +++ b/backend/cortex_backend/execution/worker_runtime.py @@ -0,0 +1,457 @@ +"""Authenticated native-broker loop for the fixed recipe worker. + +The package entrypoint owns only this loop. It accepts commands after the native +broker has authenticated the peer and the launcher has bound the exact job, +principal, PID, and AppContainer token. Every command is decoded into the +transport-neutral worker protocol; no command line, path, shell, or host-process +fallback is ever interpreted here. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from queue import Empty, Queue +import re +from threading import Event, Thread +from time import monotonic +from typing import Any, Callable, Final, Literal, Protocol + +from .broker import BrokerMessage +from .lifecycle import RuntimeHealth +from .recipe_provider import RecipeImageProvider +from .worker_protocol import ( + RecipeWorkerDispatcher, + RecipeWorkerSession, + WorkerAck, + WorkerCancel, + WorkerError, + WorkerCollect, + WorkerInputChunk, + WorkerInputComplete, + WorkerOperation, + WorkerOutputChunk, + WorkerPrepare, + WorkerProtocolError, + WorkerResult, +) + + +_PRINCIPAL = re.compile(r"^[0-9a-f]{64}$") +_SAFE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,127}$") +_SAFE_CODE = re.compile(r"^[a-z][a-z0-9_]{0,63}$") +DEFAULT_MAX_MESSAGES: Final[int] = 16_384 +DEFAULT_WATCHDOG_TIMEOUT_MS: Final[int] = 30_000 + + +class WorkerRuntimeError(ValueError): + """Stable, redacted worker-loop failure category.""" + + def __init__(self, code: str) -> None: + if _SAFE_CODE.fullmatch(code) is None: + raise ValueError("invalid worker runtime code") + self.code = code + super().__init__("The recipe worker runtime stopped safely.") + + +class WorkerProvider(Protocol): + def start(self, sandbox_health: RuntimeHealth) -> RuntimeHealth: + """Enable the fixed provider after the authenticated native boundary.""" + + def stop(self) -> RuntimeHealth: + """Release provider state before the worker exits.""" + + def transform(self, plan: Any, content: bytes, *, cancel_check: Callable[[], bool]) -> Any: + """Transform only validated in-memory bytes.""" + + +class WorkerBrokerConnection(Protocol): + """The narrow authenticated transport surface consumed by the worker loop.""" + + def receive_message(self) -> BrokerMessage: + """Receive one already-authenticated broker message.""" + + def send_message(self, message: BrokerMessage) -> None: + """Send one broker response using the transport's authenticated direction.""" + + def close(self) -> None: + """Close the connection and release native handles.""" + + +@dataclass(frozen=True, slots=True) +class WorkerRuntimeReport: + """Non-sensitive terminal state for package probes and tests.""" + + processed_messages: int + terminal_state: Literal["complete", "cancelled", "failed", "message_budget"] + + +def _validate_principal(value: str) -> str: + if _PRINCIPAL.fullmatch(value) is None: + raise ValueError("worker principal is invalid") + return value + + +def _validate_id(value: str, field: str) -> str: + if _SAFE_ID.fullmatch(value) is None: + raise ValueError(f"worker {field} is invalid") + return value + + +class RecipeWorkerBrokerRuntime: + """Run one bounded worker request over an authenticated broker connection.""" + + _MODELS: dict[str, type[Any]] = { + "prepare": WorkerPrepare, + "input_chunk": WorkerInputChunk, + "input_complete": WorkerInputComplete, + "cancel": WorkerCancel, + "collect": WorkerCollect, + } + + def __init__( + self, + connection: WorkerBrokerConnection, + *, + expected_principal_id: str, + job_id: str, + provider_factory: Callable[[], WorkerProvider] = RecipeImageProvider, + max_messages: int = DEFAULT_MAX_MESSAGES, + watchdog_timeout_ms: int = DEFAULT_WATCHDOG_TIMEOUT_MS, + ) -> None: + if not all( + callable(getattr(connection, method, None)) + for method in ("receive_message", "send_message", "close") + ): + raise TypeError("worker runtime requires an authenticated broker connection") + self._connection = connection + self._principal = _validate_principal(expected_principal_id) + self._job_id = _validate_id(job_id, "job ID") + if not callable(provider_factory): + raise TypeError("worker provider factory must be callable") + if type(max_messages) is not int or not 1 <= max_messages <= DEFAULT_MAX_MESSAGES: + raise ValueError("worker message budget is invalid") + if type(watchdog_timeout_ms) is not int or not 1 <= watchdog_timeout_ms <= 600_000: + raise ValueError("worker watchdog timeout is invalid") + self._provider_factory = provider_factory + self._max_messages = max_messages + self._watchdog_timeout_ms = watchdog_timeout_ms + + @staticmethod + def _envelope_body(message: BrokerMessage) -> Any: + model_type = RecipeWorkerBrokerRuntime._MODELS.get(message.operation) + if model_type is None: + raise WorkerProtocolError("operation_invalid") + try: + model = model_type.model_validate(message.body) + except (TypeError, ValueError): + raise WorkerProtocolError("message_invalid") from None + if model.request_id != message.request_id or model.job_id != message.job_id: + raise WorkerProtocolError("request_identity_mismatch") + return model + + def _validate_envelope(self, message: BrokerMessage) -> None: + if message.direction != "to_executor": + raise WorkerRuntimeError("broker_direction_invalid") + if message.installation_principal_id != self._principal: + raise WorkerRuntimeError("broker_principal_mismatch") + if message.job_id != self._job_id: + raise WorkerRuntimeError("broker_job_mismatch") + self._envelope_body(message) + + def _response( + self, + request: BrokerMessage, + body: WorkerAck | WorkerError | WorkerOutputChunk | WorkerResult, + ) -> BrokerMessage: + return BrokerMessage( + schema_version="broker.message.v1", + direction="to_broker", + operation=request.operation, + request_id=request.request_id, + job_id=request.job_id, + installation_principal_id=self._principal, + body=body.model_dump(mode="json"), + ) + + def _dispatch( + self, + dispatcher: RecipeWorkerDispatcher, + request: BrokerMessage, + ) -> WorkerAck | WorkerOutputChunk | WorkerResult | None: + self._validate_envelope(request) + result = dispatcher.dispatch(request.operation, request.body) + if result is not None: + return result + if request.operation == "cancel": + acknowledged: WorkerOperation = "cancel" + else: + acknowledged = request.operation # type: ignore[assignment] + return WorkerAck( + schema_version="recipe.worker.ack.v1", + request_id=request.request_id, + job_id=request.job_id, + acknowledged_operation=acknowledged, + ) + + @staticmethod + def _receive_loop( + connection: WorkerBrokerConnection, + incoming: Queue[Any], + stop_event: Event, + ) -> None: + """Keep broker reads live while a provider transform is running.""" + + while not stop_event.is_set(): + try: + incoming.put(connection.receive_message()) + except Exception as error: + if not stop_event.is_set(): + incoming.put(error) + return + + def _dispatch_completion( + self, + dispatcher: RecipeWorkerDispatcher, + request: BrokerMessage, + completion: Queue[Any], + ) -> None: + try: + completion.put((request, self._dispatch(dispatcher, request), None)) + except WorkerRuntimeError as error: + completion.put((request, None, error)) + except WorkerProtocolError as error: + completion.put((request, None, error)) + except Exception: + completion.put((request, None, WorkerProtocolError("provider_failed"))) + + def run(self) -> WorkerRuntimeReport: + """Serve one authenticated request and close transport/provider state. + + Broker reads run on a dedicated daemon thread so a cancellation frame can + reach the session while Pillow is decoding or encoding. The transform + itself is also isolated to a daemon thread; the native Job Object remains + the final watchdog if a provider ignores its cancellation callback. + """ + + provider: WorkerProvider | None = None + processed = 0 + terminal: Literal["complete", "cancelled", "failed", "message_budget"] = "failed" + receiver_stop = Event() + incoming: Queue[Any] = Queue(maxsize=8) + completion: Queue[Any] = Queue(maxsize=1) + receiver: Thread | None = None + completion_thread: Thread | None = None + try: + try: + provider = self._provider_factory() + health = provider.start( + RuntimeHealth.ready("Native broker identity and job binding established.") + ) + except WorkerRuntimeError: + raise + except Exception: + raise WorkerRuntimeError("provider_start_failed") from None + if not isinstance(health, RuntimeHealth) or not health.available: + raise WorkerRuntimeError("provider_unavailable") + dispatcher = RecipeWorkerDispatcher(RecipeWorkerSession(provider)) + receiver = Thread( + target=self._receive_loop, + args=(self._connection, incoming, receiver_stop), + name="cortex-worker-broker-reader", + daemon=True, + ) + receiver.start() + pending_collect: BrokerMessage | None = None + cancel_acknowledged = False + active_request: BrokerMessage | None = None + transform_started_at = 0.0 + + while processed < self._max_messages: + if ( + active_request is not None + and completion_thread is not None + and completion_thread.is_alive() + and not cancel_acknowledged + and monotonic() - transform_started_at + > self._watchdog_timeout_ms / 1000 + ): + timeout_cancel = BrokerMessage( + schema_version="broker.message.v1", + direction="to_executor", + operation="cancel", + request_id=active_request.request_id, + job_id=active_request.job_id, + installation_principal_id=self._principal, + body={ + "schema_version": "recipe.worker.cancel.v1", + "request_id": active_request.request_id, + "job_id": active_request.job_id, + "reason": "timeout", + }, + ) + try: + dispatcher.dispatch("cancel", timeout_cancel.body) + except WorkerProtocolError: + pass + self._connection.send_message( + self._response( + active_request, + WorkerError( + schema_version="recipe.worker.error.v1", + request_id=active_request.request_id, + job_id=active_request.job_id, + code="timeout", + ), + ) + ) + terminal = "failed" + return WorkerRuntimeReport(processed, terminal) + try: + request = incoming.get(timeout=0.05) + except Empty: + try: + completed_request, response, error = completion.get_nowait() + except Empty: + continue + completion_thread = None + active_request = None + if isinstance(error, WorkerRuntimeError): + raise error + if isinstance(error, WorkerProtocolError): + if error.code == "cancelled" and cancel_acknowledged: + terminal = "cancelled" + return WorkerRuntimeReport(processed, terminal) + response = WorkerError( + schema_version="recipe.worker.error.v1", + request_id=completed_request.request_id, + job_id=completed_request.job_id, + code=error.code, + ) + self._connection.send_message( + self._response(completed_request, response) + ) + terminal = "cancelled" if error.code == "cancelled" else "failed" + return WorkerRuntimeReport(processed, terminal) + if response is not None: + self._connection.send_message( + self._response(completed_request, response) + ) + if pending_collect is not None: + deferred = pending_collect + pending_collect = None + try: + deferred_response = self._dispatch(dispatcher, deferred) + except WorkerProtocolError as deferred_error: + deferred_response = WorkerError( + schema_version="recipe.worker.error.v1", + request_id=deferred.request_id, + job_id=deferred.job_id, + code=deferred_error.code, + ) + if deferred_response is not None: + self._connection.send_message( + self._response(deferred, deferred_response) + ) + if isinstance(deferred_response, WorkerOutputChunk): + if deferred_response.final: + terminal = "complete" + return WorkerRuntimeReport(processed, terminal) + if cancel_acknowledged: + terminal = "cancelled" + return WorkerRuntimeReport(processed, terminal) + continue + if isinstance(request, Exception): + raise WorkerRuntimeError("broker_receive_failed") from None + processed += 1 + if not isinstance(request, BrokerMessage): + raise WorkerRuntimeError("broker_message_invalid") + if request.operation == "input_complete" and completion_thread is not None: + try: + response = self._dispatch(dispatcher, request) + except WorkerProtocolError as error: + response = WorkerError( + schema_version="recipe.worker.error.v1", + request_id=request.request_id, + job_id=request.job_id, + code=error.code, + ) + if response is not None: + self._connection.send_message(self._response(request, response)) + continue + if request.operation == "collect" and completion_thread is not None: + if pending_collect is None: + pending_collect = request + else: + self._connection.send_message( + self._response( + request, + WorkerError( + schema_version="recipe.worker.error.v1", + request_id=request.request_id, + job_id=request.job_id, + code="request_state_invalid", + ), + ) + ) + continue + if request.operation == "input_complete": + self._validate_envelope(request) + active_request = request + transform_started_at = monotonic() + completion_thread = Thread( + target=self._dispatch_completion, + args=(dispatcher, request, completion), + name="cortex-worker-transform", + daemon=True, + ) + completion_thread.start() + continue + try: + response = self._dispatch(dispatcher, request) + except WorkerRuntimeError: + raise + except WorkerProtocolError as error: + response = WorkerError( + schema_version="recipe.worker.error.v1", + request_id=request.request_id, + job_id=request.job_id, + code=error.code, + ) + if response is not None: + self._connection.send_message(self._response(request, response)) + session_state = dispatcher.state + if request.operation == "cancel": + if isinstance(response, WorkerAck): + cancel_acknowledged = True + if completion_thread is None: + terminal = "cancelled" + return WorkerRuntimeReport(processed, terminal) + elif isinstance(response, WorkerError): + cancel_acknowledged = False + if session_state == "failed": + terminal = "failed" + return WorkerRuntimeReport(processed, terminal) + if request.operation == "collect" and isinstance(response, WorkerOutputChunk): + if response.final: + terminal = "complete" + return WorkerRuntimeReport(processed, terminal) + + terminal = "message_budget" + raise WorkerRuntimeError("message_budget_exceeded") + finally: + receiver_stop.set() + if provider is not None: + try: + provider.stop() + except Exception: + pass + self._connection.close() + + +__all__ = [ + "DEFAULT_MAX_MESSAGES", + "DEFAULT_WATCHDOG_TIMEOUT_MS", + "RecipeWorkerBrokerRuntime", + "WorkerRuntimeError", + "WorkerRuntimeReport", +] diff --git a/docs/adr/0001-capability-tiered-agentic-execution-harness.md b/docs/adr/0001-capability-tiered-agentic-execution-harness.md index 1725ab1..a1f24d8 100644 --- a/docs/adr/0001-capability-tiered-agentic-execution-harness.md +++ b/docs/adr/0001-capability-tiered-agentic-execution-harness.md @@ -948,7 +948,8 @@ pass without executing code. signed worker provenance now binds exactly one image-transform role to the fixed worker entrypoint, and the disposable launcher now qualifies pre-resume Job Object resource policy. The reviewed Win32 suspended factory and launcher-side broker - PID/AppContainer-token binder are now qualified; signed-worker end-to-end broker + PID/AppContainer-token binder are now qualified, and the packaged worker-side + authenticated broker loop is implemented; signed-worker end-to-end broker execution remains blocked. Collect opt-in aggregate reliability metrics, never content, only after the provider is sandbox-qualified. diff --git a/docs/adr/0001-phase2-broker-contract.md b/docs/adr/0001-phase2-broker-contract.md index 7fef8e4..04a39d7 100644 --- a/docs/adr/0001-phase2-broker-contract.md +++ b/docs/adr/0001-phase2-broker-contract.md @@ -28,7 +28,9 @@ the authorization step remains a distinct trusted-peer/job-owner check. The frame payload is canonical ASCII JSON for `broker.message.v1`: - direction: `to_broker` or `to_executor`; -- operation: `prepare`, `start`, `cancel`, or `collect`; +- operation: `prepare`, `input_chunk`, `input_complete`, `cancel`, or `collect` + (`start` remains a transport-compatibility value and is rejected by the fixed + worker rather than being guessed as an input operation); - bounded request and job identifiers; - the 256-bit installation principal; and - a bounded body for a later operation-specific schema. diff --git a/docs/adr/0001-phase2-evidence.md b/docs/adr/0001-phase2-evidence.md index 5dd3cc4..059482e 100644 --- a/docs/adr/0001-phase2-evidence.md +++ b/docs/adr/0001-phase2-evidence.md @@ -1,7 +1,7 @@ # ADR-0001 Phase 2 evidence log - **Phase:** 2 — signed image recipes and calculator/check primitives -- **Status:** Typed contract, signed-manifest verification, native broker transport, signed bundle installation, trusted artifact boundary, and qualification-only provider core complete; OS sandbox/provider and release gates remain open +- **Status:** Typed contract, signed-manifest verification, native broker transport, authenticated worker loop, signed bundle installation, trusted artifact boundary, and qualification-only provider core complete; OS sandbox/provider and release gates remain open - **Scope:** Provider-independent contracts plus a qualification-only fixed-function core - **Source decision:** [Capability-tiered agentic execution harness](0001-capability-tiered-agentic-execution-harness.md) - **Contract ADR:** [Phase 2 typed recipe and primitive contract](0001-phase2-recipe-contract.md) @@ -18,14 +18,14 @@ | Signed recipe manifest | **Complete (verification only)** | Ed25519 signature verification uses a pinned key-id allowlist; every declared bundle entry is path-, size-, and SHA-256-verified; monotonic updates and explicit rollback authorization are enforced. | | Signed bundle installation/update | **Complete (storage-only)** | Digest-named immutable generations, exclusive staging, atomic activation state, chained keyring rotation, explicit rollback authorization, and previous-generation recovery are covered by installer tests. No provider is loaded. | | Signed worker provenance binding | **Complete (storage-only)** | `verify_active_worker()` rechecks the active signed generation, binds exactly one `image_transform` role to `recipe_worker.exe`, revalidates byte identity, and rejects missing/ambiguous/mismatched/tampered/reparse entries without launching. | -| Fixed worker protocol and package closure | **Complete (qualification-only)** | `worker_protocol.py` enforces bounded prepare/chunk/complete/cancel/collect state, canonical per-chunk hashes, whole-input claims, redacted output metadata, and no-capability bodies. `packaging/recipe_worker/recipe_worker.spec` builds the fixed `recipe_worker.exe`; the entrypoint exits `78` until native broker wiring exists. | +| Fixed worker protocol and package closure | **Complete (qualification-only)** | `worker_protocol.py` and `worker_runtime.py` enforce bounded prepare/chunk/complete/cancel/collect state, authenticated envelope identity, concurrent cancellation, redacted output/errors, and no-capability bodies. `packaging/recipe_worker/recipe_worker.spec` builds the fixed `recipe_worker.exe` (Windows build verified 2026-07-23); the entrypoint accepts only the fixed native-broker identity arguments and returns `78` on direct or failed launches. | | Authenticated broker contract | **Complete (transport-neutral)** | Bounded versioned frames, direction-specific HMAC keys, canonical messages, peer ACL/integrity policy, and owner-scoped authorization are covered by adversarial tests. | | Native named-pipe adapter/DACL/peer-token binding | **Complete (transport-only)** | Protected local pipe, expected PID, OS token identity, X25519/HKDF handshake, direction keys, and close-on-error lifecycle are covered by native broker tests. | | User-artifact copy-in, output validation, and publication | **Complete (boundary only)** | Explicit owner/turn grants, bounded stable snapshots, link/reparse/hardlink/sparse/ADS rejection, byte-derived MIME policy, exact output claims, quarantine, hash/size limits, atomic repository publication, rollback, and cleanup categories are covered by `tests/test_phase2_artifact_boundary.py`. | | Fixed-function image provider core | **Complete (qualification-only)** | `RecipeImageProvider` validates allowlisted PNG/JPEG/WebP bytes, verifies/loads one frame with Pillow bomb/resource limits, applies only parsed steps, strips metadata, revalidates encoded output, checks cancellation, and remains disabled until external sandbox health passes. | | Windows recipe sandbox qualification harness | **Complete (qualification harness; worker gate blocked)** | `recipe_sandbox_qualification.py` composes out-of-process AppContainer isolation and Job Object cancellation with a fixed decoder corpus, then fails closed because the signed `recipe_worker.exe` bundle and trust-root launch verification are not shipped. | | Suspended native launcher/resource policy | **Complete (factory + binder + disposable control spike)** | `NativeWin32ProcessFactory` creates a suspended zero-capability AppContainer child and verifies Job Object policy before resume. `NativeBrokerIdentityBinder` pins the live server to the worker PID/AppContainer SID and launcher cleanup closes it on failure. The fixed qualification helper remains separate evidence. | -| OS sandbox provider and provider-produced image outputs | **Blocked / release gate** | The package/protocol/launch boundary is qualified, but the actual provider worker still needs a signed installed generation, worker-side broker loop, end-to-end authenticated input/output, watchdog, hostile decoder execution inside the sandbox, external review, and lifecycle wiring. | +| OS sandbox provider and provider-produced image outputs | **Blocked / release gate** | The package/protocol/launch boundary and worker-side broker loop are qualified, but the actual provider worker still needs a signed installed generation, end-to-end authenticated input/output through the suspended process, watchdog, hostile decoder execution inside the sandbox, external review, and lifecycle wiring. | ## Security invariants @@ -117,8 +117,8 @@ artifact-boundary tests, 17 recipe-provider tests, 6 worker-provenance tests, 7 worker-protocol tests, 16 native-launcher/factory tests, 4 native-launcher tests, and 5 sandbox-qualification tests -passed; the full Python suite -passed (242 tests total) with one +passed; 9 worker-runtime tests passed; the full Python suite +passed (251 tests total) with one native-platform skip and one pre-existing `pytest-asyncio` deprecation warning. Frontend lint, typecheck, production build, and all 39 frontend tests passed. Contract generation, compileall, and `git diff --check` passed. No production execution diff --git a/docs/adr/0001-phase2-native-broker.md b/docs/adr/0001-phase2-native-broker.md index a7fca54..6c5ddc9 100644 --- a/docs/adr/0001-phase2-native-broker.md +++ b/docs/adr/0001-phase2-native-broker.md @@ -1,6 +1,6 @@ # ADR-0001 Phase 2 native broker adapter -- **Status:** Transport and launcher identity binder implemented and verified; signed-worker end-to-end enablement remains blocked +- **Status:** Transport, launcher identity binder, and worker-side client loop implemented and verified; signed-worker end-to-end enablement remains blocked - **Parent:** [Phase 2 authenticated broker contract](0001-phase2-broker-contract.md) - **Scope:** Windows named-pipe transport, protected DACL, OS peer identity, and authenticated session-key establishment @@ -45,7 +45,10 @@ worker PID and allows exactly the worker's AppContainer SID in the protected DAC the broker PID is required to equal the current server process. The binder pins the installation principal and durable job for `accept()`, and closes the endpoint on resume or cancellation failure. This is the live identity binding seam; the -signed worker package and worker-side client loop remain separate release gates. +signed worker package remains a separate release gate. The packaged worker now +uses `NativeBrokerClient` with the expected broker PID, principal, and exact job +owner binding. It receives only authenticated `broker.message.v1` commands and +returns redacted worker acknowledgements, results, output chunks, and errors. ## Lifecycle and failure contract @@ -74,6 +77,9 @@ decoding and lifecycle enablement remain separate gates. deterministic SDDL, X25519/HKDF directional key agreement, malformed-handshake and PID-mismatch rejection, direction-violation close, local pipe-name and expected-PID validation, native pipe create/close, and OS token identity extraction. The Windows +worker-loop contract is covered separately by `tests/test_phase2_worker_runtime.py`, +which uses the same narrow connection surface to exercise authenticated command +mapping, redacted responses, watchdog, cancellation, and close-on-terminal state. API choices follow Microsoft's contracts for [CreateNamedPipe](https://learn.microsoft.com/en-us/windows/win32/api/winbase/nf-winbase-createnamedpipea), [GetNamedPipeClientProcessId](https://learn.microsoft.com/en-us/windows/win32/api/winbase/nf-winbase-getnamedpipeclientprocessid), diff --git a/docs/adr/0001-phase2-native-launcher.md b/docs/adr/0001-phase2-native-launcher.md index 8b22ba5..6078088 100644 --- a/docs/adr/0001-phase2-native-launcher.md +++ b/docs/adr/0001-phase2-native-launcher.md @@ -16,8 +16,8 @@ accepts only an installer-verified worker and a trusted `BrokerWorkerBinding`: 1. recheck the active signed generation through `verify_active_worker()`; 2. revalidate the fixed `recipe_worker.exe` identity, size, link count, and hash immediately before launch planning; -3. construct a fixed command line containing only the native-broker endpoint and - expected broker PID; +3. construct a fixed command line containing only the native-broker endpoint, + expected broker PID, installation principal, and job ID; 4. require a reviewed native process factory and live broker binder before any process is created; and 5. enforce the order `create suspended -> apply Job policy -> bind worker PID and @@ -83,9 +83,9 @@ behavior across supported Windows versions, or external launcher review. - The fixed signed `recipe_worker.exe` bundle is not installed in the immutable generation used by the launcher, so no real worker has completed the broker handshake yet. -- The packaged worker loop still exits with its launch-refusing status; the - end-to-end authenticated input/output, watchdog, and cancellation path must be - wired before provider execution is authorized. +- The packaged worker loop is now broker-only and fail-closed; the end-to-end + authenticated input/output, watchdog, and cancellation path still needs evidence + against the signed installed generation before provider execution is authorized. - Watchdog progress, output framing, staging ACLs, hostile decoder execution, and lifecycle health-gated wiring remain separate release gates. diff --git a/docs/adr/0001-phase2-sandbox-qualification.md b/docs/adr/0001-phase2-sandbox-qualification.md index a0360f9..f66f369 100644 --- a/docs/adr/0001-phase2-sandbox-qualification.md +++ b/docs/adr/0001-phase2-sandbox-qualification.md @@ -24,7 +24,7 @@ The harness is deliberately fail-closed and has independent checks for: the native launcher applies and verifies them; and 6. it reports end-to-end broker execution as blocked until the launcher binds the qualified transport to a signed worker's actual PID/AppContainer token and the - worker completes the authenticated client handshake. + packaged worker completes the authenticated client handshake and hostile corpus. The fourth check is intentionally blocked in this stage. A directory, executable, self-reported digest, or unverified manifest cannot authorize a launch. The future @@ -54,8 +54,8 @@ control produces `blocked` or `fail`, and no weaker host-process path is attempt ## Required future worker qualification -The next implementation must add a signed, pinned worker bundle and a native -launcher that, per attempt: +The remaining release gate must install a signed, pinned worker bundle and run the +existing native launcher/worker loop per attempt: 1. verifies the installed immutable generation and image-worker entrypoint; 2. creates private staging and grants only the sandbox identity and required diff --git a/docs/adr/0001-phase2-worker-protocol.md b/docs/adr/0001-phase2-worker-protocol.md index 8336034..e6bc0dd 100644 --- a/docs/adr/0001-phase2-worker-protocol.md +++ b/docs/adr/0001-phase2-worker-protocol.md @@ -1,6 +1,6 @@ # ADR-0001 Phase 2 fixed recipe worker protocol and package boundary -- **Status:** Protocol and packaging qualification complete; native execution blocked +- **Status:** Protocol, authenticated worker loop, and packaging qualification complete; signed end-to-end execution remains blocked - **Phase:** 2 - fixed-function image provider - **Parent:** [Capability-tiered agentic execution harness](0001-capability-tiered-agentic-execution-harness.md) - **Depends on:** [Signed worker provenance](0001-phase2-worker-provenance.md), [native broker adapter](0001-phase2-native-broker.md), and [native launcher](0001-phase2-native-launcher.md) @@ -34,12 +34,15 @@ replayed offsets, identity mismatches, size/digest mismatches, provider failures and output offsets fail closed with stable categories. No worker message accepts a filesystem path, shell command, executable, network target, token, or model source. -The package entrypoint is intentionally launch-refusing (exit code `78`) until the -reviewed native broker adapter is wired in. This prevents an unsigned or directly -started package from becoming a stdio, shell, or host-process fallback. The -PyInstaller definition and Windows build script qualify dependency closure and the -fixed `recipe_worker.exe` path only; they do not sign, install, or authorize the -worker. +The package entrypoint now accepts exactly the native launcher's fixed argument +shape: protected pipe name, expected broker PID, installation principal, and job +ID. It creates only a `NativeBrokerClient`, authenticates the broker session, and +serves the bounded worker loop. Direct launches, malformed arguments, missing +broker identity, provider startup failure, transport failure, and message-budget +exhaustion return the safe refusal status (`78`); there is no stdio, shell, path, +or host-process fallback. The PyInstaller definition and Windows build script +qualify dependency closure and the fixed `recipe_worker.exe` path only; they do +not sign, install, or authorize the worker. ## Security invariants @@ -61,16 +64,19 @@ worker. `tests/test_phase2_worker_protocol.py` covers successful in-order streaming and collection, malformed operations, replay/order failure, claim mismatch, -cancellation terminality, and chunk tampering. On the controlled Windows host -(2026-07-23), the PyInstaller build produced -`dist/recipe-runtime/recipe_worker.exe` (11,320,824 bytes); direct launch returned -exit code `78` as required. +cancellation terminality, and chunk tampering. `tests/test_phase2_worker_runtime.py` +(9 tests) covers authenticated envelope binding, redacted repairable failures, +provider failure, message budgets, terminal cleanup, watchdog expiry, +fixed-entrypoint parsing, and cancellation delivered while a transform is running. +On the controlled Windows host (2026-07-23), the +PyInstaller build produced `dist/recipe-runtime/recipe_worker.exe`; direct launch +without the exact native broker arguments returned exit code `78` as required. ## Remaining blockers -This ADR does not authorize provider execution. The next stage must replace the -launch-refusing entrypoint with the reviewed native broker loop, launch the verified -worker suspended under AppContainer/LPAC and Job Object policy, bind the live broker -session to the actual worker PID and AppContainer token, and run the hostile decoder -and cancellation corpus inside that process. Lifecycle/UI enablement remains behind -those gates and external security review. +This ADR does not authorize provider execution. The remaining stage must install a +real signed generation, launch it through the reviewed suspended AppContainer/Job +Object factory, bind the live broker session to the actual worker PID and +AppContainer token, and run the hostile decoder/cancellation corpus through that +packaged process with watchdog and artifact-boundary evidence. Lifecycle/UI +enablement remains behind those gates and external security review. diff --git a/packaging/recipe_worker/README.md b/packaging/recipe_worker/README.md index 0822449..d9c474c 100644 --- a/packaging/recipe_worker/README.md +++ b/packaging/recipe_worker/README.md @@ -1,9 +1,10 @@ # Recipe worker package This directory defines the fixed-function worker package boundary. The package -entrypoint deliberately exits with status `78` until the reviewed native broker -adapter is added. That refusal is a safety property: a package build must not -silently become a host-process or stdio execution fallback. +entrypoint accepts only the reviewed native broker launch shape (protected pipe, +expected broker PID, installation principal, and exact job ID). Direct launches, +malformed identity, transport failures, provider failures, and watchdog expiry +return status `78`; there is no host-process, shell, path, or stdio fallback. Build it on a supported Windows machine with: @@ -23,4 +24,6 @@ repository. Release packaging must, outside source control: be considered. The native launcher and live broker PID/AppContainer-token binding remain required -before this package can perform any transform. +before this package can perform any transform. The worker loop itself is covered +by `tests/test_phase2_worker_runtime.py`, including in-flight cancellation and +watchdog cleanup. diff --git a/packaging/recipe_worker/recipe_worker.py b/packaging/recipe_worker/recipe_worker.py index f2f5ec5..c2d6293 100644 --- a/packaging/recipe_worker/recipe_worker.py +++ b/packaging/recipe_worker/recipe_worker.py @@ -1,23 +1,105 @@ """Fixed recipe worker package entrypoint. -This entrypoint intentionally refuses to run until the native broker adapter is -provided by the reviewed launcher stage. It exists so packaging, dependency -closure, and signed-bundle provenance can be qualified without creating an -unsafe stdio, shell, or host-process fallback. +The executable has one supported launch shape: the native launcher supplies a +protected named-pipe endpoint plus the exact broker principal and job identity. +There is deliberately no stdio, shell, path, or direct-process fallback. """ from __future__ import annotations +from dataclasses import dataclass +import re import sys +from cortex_backend.execution.native_broker import ( + NativeBrokerClient, + NativeBrokerClientConfig, +) +from cortex_backend.execution.worker_runtime import ( + RecipeWorkerBrokerRuntime, +) + + +EXIT_SAFE_REFUSAL = 78 +_PIPE = re.compile(r"^\\\\\.\\pipe\\cortex-[A-Za-z0-9._-]{1,200}$") +_PRINCIPAL = re.compile(r"^[0-9a-f]{64}$") +_SAFE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,127}$") + + +@dataclass(frozen=True, slots=True) +class WorkerArguments: + pipe_name: str + broker_process_id: int + installation_principal_id: str + job_id: str + + +def _parse_args(argv: list[str]) -> WorkerArguments: + expected = ( + "--native-broker", + "--broker-pipe", + "--broker-pid", + "--broker-principal", + "--job-id", + ) + if len(argv) != 9 or argv[0] != expected[0]: + raise ValueError("native broker arguments are required") + values = dict(zip(argv[1::2], argv[2::2], strict=True)) + if tuple(values) != expected[1:] or any(not isinstance(value, str) for value in values.values()): + raise ValueError("native broker arguments are invalid") + pipe_name = values["--broker-pipe"] + if _PIPE.fullmatch(pipe_name) is None: + raise ValueError("native broker pipe is invalid") + try: + broker_process_id = int(values["--broker-pid"], 10) + except (TypeError, ValueError): + raise ValueError("native broker process ID is invalid") from None + if not 1 <= broker_process_id <= 0xFFFFFFFF: + raise ValueError("native broker process ID is invalid") + principal = values["--broker-principal"] + if _PRINCIPAL.fullmatch(principal) is None: + raise ValueError("native broker principal is invalid") + job_id = values["--job-id"] + if _SAFE_ID.fullmatch(job_id) is None: + raise ValueError("native broker job ID is invalid") + return WorkerArguments(pipe_name, broker_process_id, principal, job_id) + + +def _run(arguments: WorkerArguments) -> int: + client = NativeBrokerClient( + NativeBrokerClientConfig( + pipe_name=arguments.pipe_name, + expected_server_process_id=arguments.broker_process_id, + ) + ) + connection = client.connect( + expected_principal_id=arguments.installation_principal_id, + owner_for_job=lambda job_id: ( + arguments.installation_principal_id if job_id == arguments.job_id else None + ), + ) + try: + report = RecipeWorkerBrokerRuntime( + connection, + expected_principal_id=arguments.installation_principal_id, + job_id=arguments.job_id, + ).run() + except Exception: + connection.close() + raise + return 0 if report.terminal_state in {"complete", "cancelled"} else EXIT_SAFE_REFUSAL + def main(argv: list[str] | None = None) -> int: - """Refuse every launch until the native broker contract is wired in.""" + """Run only when the native broker launch contract is complete.""" - del argv - # A non-zero deterministic exit is important: a packaged but unbound worker - # must never look healthy to a caller that accidentally starts it directly. - return 78 + try: + arguments = _parse_args(list(sys.argv[1:] if argv is None else argv)) + return _run(arguments) + except Exception: + # Keep direct launches and all runtime failures indistinguishable and + # avoid printing broker, token, path, or decoder details from a package. + return EXIT_SAFE_REFUSAL if __name__ == "__main__": # pragma: no cover - exercised by the packaged exe. diff --git a/packaging/recipe_worker/recipe_worker.spec b/packaging/recipe_worker/recipe_worker.spec index de5d3be..5892429 100644 --- a/packaging/recipe_worker/recipe_worker.spec +++ b/packaging/recipe_worker/recipe_worker.spec @@ -12,6 +12,8 @@ a = Analysis( datas=[], hiddenimports=[ "cortex_backend.execution.worker_protocol", + "cortex_backend.execution.worker_runtime", + "cortex_backend.execution.native_broker", "cortex_backend.execution.recipe_provider", "PIL", "PIL.Image", diff --git a/tests/test_phase2_native_launcher.py b/tests/test_phase2_native_launcher.py index d69bc5b..a567c65 100644 --- a/tests/test_phase2_native_launcher.py +++ b/tests/test_phase2_native_launcher.py @@ -129,6 +129,10 @@ def test_prepare_reverifies_active_worker_and_builds_fixed_command_line(tmp_path _binding().pipe_name, "--broker-pid", "321", + "--broker-principal", + _binding().installation_principal_id, + "--job-id", + _binding().job_id, ] ) == plan.command_line assert "shell" not in plan.command_line.casefold() diff --git a/tests/test_phase2_worker_runtime.py b/tests/test_phase2_worker_runtime.py new file mode 100644 index 0000000..d109c36 --- /dev/null +++ b/tests/test_phase2_worker_runtime.py @@ -0,0 +1,387 @@ +"""Authenticated worker-loop contract and hostile recovery tests.""" + +from __future__ import annotations + +import base64 +from hashlib import sha256 +from io import BytesIO +from pathlib import Path +from queue import Queue +import runpy +from threading import Event +import time +from typing import Any + +import pytest +from PIL import Image + +from cortex_backend.execution.broker import BrokerMessage +from cortex_backend.execution.lifecycle import RuntimeHealth +from cortex_backend.execution.recipe_provider import RecipeImageProvider, RecipeProviderError +from cortex_backend.execution.recipes import parse_image_transform +from cortex_backend.execution.worker_protocol import ( + WorkerCancel, + WorkerCollect, + WorkerInputChunk, + WorkerInputComplete, + WorkerPrepare, +) +from cortex_backend.execution.worker_runtime import ( + RecipeWorkerBrokerRuntime, + WorkerRuntimeError, +) + + +PRINCIPAL = "a" * 64 +JOB_ID = "job_1" + + +class _FakeConnection: + def __init__(self, incoming: list[BrokerMessage]) -> None: + self.incoming: Queue[BrokerMessage | None] = Queue() + for message in incoming: + self.incoming.put(message) + self.sent: list[BrokerMessage] = [] + self.closed = False + + def receive_message(self) -> BrokerMessage: + message = self.incoming.get() + if message is None: + raise AssertionError("runtime read past the expected terminal response") + return message + + def send_message(self, message: BrokerMessage) -> None: + self.sent.append(message) + + def close(self) -> None: + self.closed = True + self.incoming.put(None) + + +def _image_bytes() -> bytes: + image = Image.new("RGB", (4, 3), (120, 80, 40)) + try: + with BytesIO() as stream: + image.save(stream, format="PNG") + return stream.getvalue() + finally: + image.close() + + +def _plan(): + return parse_image_transform( + { + "schema_version": "artifact.transform.v1", + "input_artifact_id": "artifact-1", + "steps": [{"op": "grayscale"}], + "output_format": "png", + } + ) + + +def _prepare(content: bytes, *, request_id: str = "request_1") -> WorkerPrepare: + return WorkerPrepare( + schema_version="recipe.worker.prepare.v1", + request_id=request_id, + job_id=JOB_ID, + plan=_plan(), + input_size=len(content), + input_sha256=sha256(content).hexdigest(), + input_mime_type="image/png", + ) + + +def _chunk(content: bytes, *, offset: int, request_id: str = "request_1") -> WorkerInputChunk: + return WorkerInputChunk( + schema_version="recipe.worker.input_chunk.v1", + request_id=request_id, + job_id=JOB_ID, + offset=offset, + data=base64.urlsafe_b64encode(content).decode("ascii").rstrip("="), + sha256=sha256(content).hexdigest(), + ) + + +def _message(operation: str, model: Any, *, request_id: str = "request_1") -> BrokerMessage: + return BrokerMessage( + schema_version="broker.message.v1", + direction="to_executor", + operation=operation, + request_id=request_id, + job_id=JOB_ID, + installation_principal_id=PRINCIPAL, + body=model.model_dump(mode="json"), + ) + + +def _successful_messages(content: bytes) -> list[BrokerMessage]: + prepare = _prepare(content) + complete = WorkerInputComplete( + schema_version="recipe.worker.input_complete.v1", + request_id="request_1", + job_id=JOB_ID, + input_size=len(content), + input_sha256=sha256(content).hexdigest(), + ) + collect = WorkerCollect( + schema_version="recipe.worker.collect.v1", + request_id="request_1", + job_id=JOB_ID, + offset=0, + max_bytes=48 * 1024, + ) + return [ + _message("prepare", prepare), + _message("input_chunk", _chunk(content, offset=0)), + _message("input_complete", complete), + _message("collect", collect), + ] + + +def test_runtime_processes_authenticated_request_and_closes_provider_and_transport(): + connection = _FakeConnection(_successful_messages(_image_bytes())) + providers: list[RecipeImageProvider] = [] + + def provider_factory() -> RecipeImageProvider: + provider = RecipeImageProvider() + providers.append(provider) + return provider + + report = RecipeWorkerBrokerRuntime( + connection, + expected_principal_id=PRINCIPAL, + job_id=JOB_ID, + provider_factory=provider_factory, + ).run() + + assert report.terminal_state == "complete" + assert report.processed_messages == 4 + assert connection.closed + assert providers[0].health_snapshot.code == "recipe_provider_stopped" + assert [message.operation for message in connection.sent] == [ + "prepare", + "input_chunk", + "input_complete", + "collect", + ] + assert connection.sent[0].body["schema_version"] == "recipe.worker.ack.v1" + assert connection.sent[2].body["schema_version"] == "recipe.worker.result.v1" + assert connection.sent[3].body["schema_version"] == "recipe.worker.output_chunk.v1" + assert connection.sent[3].body["final"] is True + + +def test_runtime_redacts_malformed_body_and_allows_bounded_repair(): + content = _image_bytes() + malformed = BrokerMessage( + schema_version="broker.message.v1", + direction="to_executor", + operation="prepare", + request_id="request_1", + job_id=JOB_ID, + installation_principal_id=PRINCIPAL, + body={"schema_version": "recipe.worker.prepare.v1"}, + ) + connection = _FakeConnection([malformed, *_successful_messages(content)]) + + report = RecipeWorkerBrokerRuntime( + connection, + expected_principal_id=PRINCIPAL, + job_id=JOB_ID, + ).run() + + assert report.terminal_state == "complete" + assert connection.sent[0].body == { + "schema_version": "recipe.worker.error.v1", + "request_id": "request_1", + "job_id": JOB_ID, + "code": "message_invalid", + } + assert "path" not in str(connection.sent[0].body) + + +def test_runtime_cancellation_is_terminal_and_acknowledged(): + content = _image_bytes() + prepare = _prepare(content) + cancel = WorkerCancel( + schema_version="recipe.worker.cancel.v1", + request_id="request_1", + job_id=JOB_ID, + reason="user", + ) + connection = _FakeConnection([_message("prepare", prepare), _message("cancel", cancel)]) + + report = RecipeWorkerBrokerRuntime( + connection, + expected_principal_id=PRINCIPAL, + job_id=JOB_ID, + ).run() + + assert report.terminal_state == "cancelled" + assert connection.sent[-1].body["schema_version"] == "recipe.worker.ack.v1" + assert connection.sent[-1].body["acknowledged_operation"] == "cancel" + assert connection.closed + + +class _CancellableProvider: + def __init__(self) -> None: + self.entered = Event() + self.stopped = False + self.on_entered: Any = None + + def start(self, _health: RuntimeHealth) -> RuntimeHealth: + return RuntimeHealth.ready("test") + + def stop(self) -> RuntimeHealth: + self.stopped = True + return RuntimeHealth.blocked("test_stopped", "test provider stopped") + + def transform(self, _plan: Any, _content: bytes, *, cancel_check: Any) -> Any: + self.entered.set() + if self.on_entered is not None: + self.on_entered() + while not cancel_check(): + time.sleep(0.001) + raise RecipeProviderError("cancelled") + + +def test_runtime_cancellation_reaches_provider_during_transform(): + content = _image_bytes() + provider = _CancellableProvider() + messages = _successful_messages(content)[:3] + connection = _FakeConnection(messages) + provider.on_entered = lambda: connection.incoming.put( + _message( + "cancel", + WorkerCancel( + schema_version="recipe.worker.cancel.v1", + request_id="request_1", + job_id=JOB_ID, + reason="user", + ), + ) + ) + + report = RecipeWorkerBrokerRuntime( + connection, + expected_principal_id=PRINCIPAL, + job_id=JOB_ID, + provider_factory=lambda: provider, + ).run() + + assert provider.entered.is_set() + assert provider.stopped + assert report.terminal_state == "cancelled" + assert connection.sent[-1].body["schema_version"] == "recipe.worker.ack.v1" + + +def test_runtime_rejects_job_confusion_before_dispatch_and_closes(): + content = _image_bytes() + message = _message("prepare", _prepare(content)) + message = message.model_copy(update={"job_id": "other_job"}) + connection = _FakeConnection([message]) + + with pytest.raises(WorkerRuntimeError) as error: + RecipeWorkerBrokerRuntime( + connection, + expected_principal_id=PRINCIPAL, + job_id=JOB_ID, + ).run() + assert error.value.code == "broker_job_mismatch" + assert connection.sent == [] + assert connection.closed + + +class _FailingProvider: + def __init__(self) -> None: + self.stopped = False + + def start(self, _health: RuntimeHealth) -> RuntimeHealth: + return RuntimeHealth.ready("test") + + def stop(self) -> RuntimeHealth: + self.stopped = True + return RuntimeHealth.blocked("test_stopped", "test provider stopped") + + def transform(self, _plan: Any, _content: bytes, *, cancel_check: Any) -> Any: + del cancel_check + raise RecipeProviderError("decode_failed") + + +def test_runtime_returns_redacted_provider_failure_and_stops(): + content = _image_bytes() + provider = _FailingProvider() + complete = _successful_messages(content)[:3] + connection = _FakeConnection(complete) + + report = RecipeWorkerBrokerRuntime( + connection, + expected_principal_id=PRINCIPAL, + job_id=JOB_ID, + provider_factory=lambda: provider, + ).run() + + assert report.terminal_state == "failed" + assert connection.sent[-1].body == { + "schema_version": "recipe.worker.error.v1", + "request_id": "request_1", + "job_id": JOB_ID, + "code": "decode_failed", + } + assert provider.stopped + + +def test_runtime_enforces_message_budget_and_cleans_up(): + content = _image_bytes() + connection = _FakeConnection([_message("prepare", _prepare(content))]) + with pytest.raises(WorkerRuntimeError) as error: + RecipeWorkerBrokerRuntime( + connection, + expected_principal_id=PRINCIPAL, + job_id=JOB_ID, + max_messages=1, + ).run() + assert error.value.code == "message_budget_exceeded" + assert connection.closed + + +def test_runtime_watchdog_captures_stalled_transform_and_closes(): + content = _image_bytes() + provider = _CancellableProvider() + connection = _FakeConnection(_successful_messages(content)[:3]) + + report = RecipeWorkerBrokerRuntime( + connection, + expected_principal_id=PRINCIPAL, + job_id=JOB_ID, + provider_factory=lambda: provider, + watchdog_timeout_ms=10, + ).run() + + assert report.terminal_state == "failed" + assert connection.sent[-1].body["schema_version"] == "recipe.worker.error.v1" + assert connection.sent[-1].body["code"] == "timeout" + assert connection.closed + + +def test_packaged_entrypoint_accepts_only_fixed_native_launch_shape(): + entry = runpy.run_path( + str(Path(__file__).parents[1] / "packaging" / "recipe_worker" / "recipe_worker.py") + ) + assert entry["main"]([]) == 78 + parsed = entry["_parse_args"]( + [ + "--native-broker", + "--broker-pipe", + r"\\.\pipe\cortex-test", + "--broker-pid", + "321", + "--broker-principal", + PRINCIPAL, + "--job-id", + JOB_ID, + ] + ) + assert parsed.pipe_name == r"\\.\pipe\cortex-test" + assert parsed.broker_process_id == 321 + with pytest.raises(ValueError): + entry["_parse_args"](["--native-broker", "--broker-pipe", "unsafe"]) diff --git a/tools/execution_spikes/README.md b/tools/execution_spikes/README.md index 2031b37..4b3fa44 100644 --- a/tools/execution_spikes/README.md +++ b/tools/execution_spikes/README.md @@ -56,9 +56,11 @@ powershell -ExecutionPolicy Bypass -File packaging/build_recipe_worker.ps1 ``` This produces `dist/recipe-runtime/recipe_worker.exe` and verifies dependency -closure only. The entrypoint exits with status `78` until the native broker loop is -implemented; the output is unsigned and must not be installed or launched as a -provider. The package contract is covered by `tests/test_phase2_worker_protocol.py`. +closure only. The entrypoint accepts only the fixed native broker identity +arguments and exits with status `78` for direct or failed launches; the output is +unsigned and must not be installed or launched as a provider. The package and +runtime contracts are covered by `tests/test_phase2_worker_protocol.py` and +`tests/test_phase2_worker_runtime.py`. ## What the probes prove diff --git a/tools/execution_spikes/native_launcher_qualification.py b/tools/execution_spikes/native_launcher_qualification.py index bacf0a9..c581970 100644 --- a/tools/execution_spikes/native_launcher_qualification.py +++ b/tools/execution_spikes/native_launcher_qualification.py @@ -192,7 +192,7 @@ def _probe_broker_binding() -> dict[str, Any]: return _result( "native_launcher_broker_binding", BLOCKED, - "The native broker transport is qualified separately; a reviewed launcher has not yet bound it to a worker PID and OS token.", + "The native broker transport and launcher binder are qualified separately; a signed installed worker still needs a real PID/token handshake.", launch_refused=True, ) diff --git a/tools/execution_spikes/recipe_sandbox_qualification.py b/tools/execution_spikes/recipe_sandbox_qualification.py index 3b6f1cd..9612e6c 100644 --- a/tools/execution_spikes/recipe_sandbox_qualification.py +++ b/tools/execution_spikes/recipe_sandbox_qualification.py @@ -310,7 +310,7 @@ def _probe_future_worker_controls() -> list[dict[str, Any]]: _result( "recipe_broker_identity", BLOCKED, - "The native broker transport is qualified separately but is not yet bound to a recipe worker PID and OS token by a launcher.", + "The native broker transport and launcher binder are qualified separately; a signed installed worker still needs a real PID/token handshake.", launch_refused=True, ), ]