From 4c2a40367e05f085608a6d472e847a49cfd7218e Mon Sep 17 00:00:00 2001 From: Matthew Wesney Date: Wed, 29 Jul 2026 08:45:51 -0400 Subject: [PATCH] Add durable recipe coordinator and publication path --- backend/cortex_backend/execution/__init__.py | 40 +- .../execution/recipe_coordinator.py | 977 ++++++++++++++++++ .../cortex_backend/execution/repository.py | 90 +- docs/adr/0001-phase2-artifact-boundary.md | 7 +- docs/adr/0001-phase2-evidence.md | 11 +- .../0001-phase2-qualification-lifecycle.md | 11 +- docs/adr/0001-phase2-recipe-contract.md | 13 +- docs/adr/0001-phase2-recipe-coordinator.md | 111 ++ docs/adr/0001-phase2-recipe-provider.md | 13 +- docs/adr/0001-phase2-sandbox-qualification.md | 10 +- tests/test_phase2_recipe_coordinator.py | 365 +++++++ 11 files changed, 1615 insertions(+), 33 deletions(-) create mode 100644 backend/cortex_backend/execution/recipe_coordinator.py create mode 100644 docs/adr/0001-phase2-recipe-coordinator.md create mode 100644 tests/test_phase2_recipe_coordinator.py diff --git a/backend/cortex_backend/execution/__init__.py b/backend/cortex_backend/execution/__init__.py index 7399cd0..c97199d 100644 --- a/backend/cortex_backend/execution/__init__.py +++ b/backend/cortex_backend/execution/__init__.py @@ -1,10 +1,10 @@ """Durable execution primitives and provider-independent safety contracts. Only deterministic contracts, the storage-only signed bundle installer, the -reviewed broker transports, and the explicit local qualification composition are -exposed in this phase. Native transport and bundle storage never load a provider; -the normal application remains default-off and a recipe request coordinator is a -later ADR slice. +reviewed broker transports, the explicit local qualification composition, and the +internal qualified recipe coordinator are exposed in this phase. Native transport +and bundle storage never load a provider; the normal application remains +default-off until a later application-integration decision. """ from .broker import ( @@ -127,6 +127,23 @@ LeaseConflict, ExecutionRepositoryError, ) +from .recipe_coordinator import ( + DEFAULT_CANCEL_GRACE_SECONDS, + DEFAULT_RECIPE_RETENTION_SECONDS, + DEFAULT_WORKER_TIMEOUT_SECONDS, + MAX_RECIPE_RETENTION_SECONDS, + RECIPE_IMAGE_PROFILE, + RECIPE_PAYLOAD_SCHEMA, + RECIPE_RESULT_SCHEMA, + RecipeExecutionCoordinator, + RecipeExecutionError, + RecipeImageRequest, + RecipeWorkerAttempt, + RecipeWorkerAttemptFactory, + RecipeWorkerClient, + RecipeWorkerConnection, + RecipeWorkerOutput, +) from .resource_accounting import ( MAX_CONSOLE_BYTES, MAX_COUNTER, @@ -183,6 +200,21 @@ "DEFAULT_NATIVE_PIPE_BUFFER_BYTES", "ExecutionRepository", "ExecutionRepositoryError", + "DEFAULT_CANCEL_GRACE_SECONDS", + "DEFAULT_RECIPE_RETENTION_SECONDS", + "DEFAULT_WORKER_TIMEOUT_SECONDS", + "MAX_RECIPE_RETENTION_SECONDS", + "RECIPE_IMAGE_PROFILE", + "RECIPE_PAYLOAD_SCHEMA", + "RECIPE_RESULT_SCHEMA", + "RecipeExecutionCoordinator", + "RecipeExecutionError", + "RecipeImageRequest", + "RecipeWorkerAttempt", + "RecipeWorkerAttemptFactory", + "RecipeWorkerClient", + "RecipeWorkerConnection", + "RecipeWorkerOutput", "ExecutionLifecycle", "ExecutionProfile", "CoordinatorFactory", diff --git a/backend/cortex_backend/execution/recipe_coordinator.py b/backend/cortex_backend/execution/recipe_coordinator.py new file mode 100644 index 0000000..7ebd052 --- /dev/null +++ b/backend/cortex_backend/execution/recipe_coordinator.py @@ -0,0 +1,977 @@ +"""Durable coordinator for the qualified fixed-function image recipe. + +The coordinator is intentionally an internal composition seam. It accepts an +opaque, owner-scoped artifact identifier and a typed :class:`ImageTransformPlan`, +then delegates the bounded transform to an injected authenticated worker attempt. +It never accepts source paths, model code, shell commands, or provider authority. +Outputs are staged privately and published only through ``ArtifactBoundary`` after +the complete result has been validated. +""" + +from __future__ import annotations + +from collections.abc import Callable, Mapping +import base64 +from dataclasses import dataclass +from hashlib import sha256 +import json +import os +from pathlib import Path +from queue import Empty, Queue +import re +import tempfile +from threading import Event, Lock, Thread +import time +from typing import Any, Protocol +from uuid import uuid4 + +from .artifact_boundary import ( + ArtifactBoundary, + ArtifactBoundaryError, + OutputClaim, + PublishedArtifact, + sniff_artifact_mime, +) +from .broker import BrokerMessage +from .models import ExecutionJob, TerminalExecutionStatus +from .recipes import ImageTransformPlan, RecipeValidationError, parse_image_transform +from .repository import ( + ExecutionRepository, + ExecutionRepositoryError, + LeaseConflict, +) +from .worker_protocol import ( + MAX_WORKER_CHUNK_BYTES, + MAX_WORKER_INPUT_BYTES, + MAX_WORKER_OUTPUT_BYTES, + WorkerAck, + WorkerCancel, + WorkerCollect, + WorkerError, + WorkerInputChunk, + WorkerInputComplete, + WorkerOutputChunk, + WorkerPrepare, + WorkerProtocolError, + WorkerResult, +) + + +RECIPE_IMAGE_PROFILE = "recipe.image.v1" +RECIPE_PAYLOAD_SCHEMA = "recipe.execution.v1" +RECIPE_RESULT_SCHEMA = "recipe.result.v1" +DEFAULT_RECIPE_RETENTION_SECONDS = 86_400 +MAX_RECIPE_RETENTION_SECONDS = 30 * 86_400 +DEFAULT_WORKER_TIMEOUT_SECONDS = 120.0 +DEFAULT_CANCEL_GRACE_SECONDS = 5.0 +_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}$") +_MIME_TO_FORMAT = {"image/png": "PNG", "image/jpeg": "JPEG", "image/webp": "WEBP"} + + +class RecipeExecutionError(RuntimeError): + """Stable, redacted coordinator failure category.""" + + def __init__(self, code: str) -> None: + if _SAFE_CODE.fullmatch(code) is None: + raise ValueError("invalid recipe execution error code") + self.code = code + super().__init__("The image recipe could not be completed safely.") + + +def _safe_id(value: Any, field: str) -> str: + if not isinstance(value, str) or _SAFE_ID.fullmatch(value) is None: + raise ValueError(f"{field} must be a bounded opaque identifier") + return value + + +@dataclass(frozen=True, slots=True) +class RecipeImageRequest: + """A durable image request with no host path or executable authority.""" + + owner: str + request_id: str + source_artifact_id: str + plan: ImageTransformPlan + retention_seconds: int = DEFAULT_RECIPE_RETENTION_SECONDS + + def __post_init__(self) -> None: + _safe_id(self.owner, "owner") + _safe_id(self.request_id, "request_id") + _safe_id(self.source_artifact_id, "source_artifact_id") + if not isinstance(self.plan, ImageTransformPlan): + raise TypeError("plan must be an ImageTransformPlan") + if self.plan.input_artifact_id != self.source_artifact_id: + raise RecipeExecutionError("input_artifact_mismatch") + if ( + isinstance(self.retention_seconds, bool) + or not isinstance(self.retention_seconds, int) + or not 1 <= self.retention_seconds <= MAX_RECIPE_RETENTION_SECONDS + ): + raise ValueError("retention_seconds is outside the bounded recipe limit") + + +@dataclass(frozen=True, slots=True) +class RecipeWorkerOutput: + """Validated in-memory output returned by one worker attempt.""" + + content: bytes + mime_type: str + format: str + width: int + height: int + sha256: str + + def __post_init__(self) -> None: + if not isinstance(self.content, bytes) or not self.content: + raise RecipeExecutionError("worker_output_invalid") + if len(self.content) > MAX_WORKER_OUTPUT_BYTES: + raise RecipeExecutionError("worker_output_too_large") + if self.mime_type not in _MIME_TO_FORMAT or _MIME_TO_FORMAT[self.mime_type] != self.format: + raise RecipeExecutionError("worker_output_invalid") + if type(self.width) is not int or type(self.height) is not int: + raise RecipeExecutionError("worker_output_invalid") + if not 1 <= self.width <= 16_384 or not 1 <= self.height <= 16_384: + raise RecipeExecutionError("worker_output_invalid") + if not isinstance(self.sha256, str) or re.fullmatch(r"[0-9a-f]{64}", self.sha256) is None: + raise RecipeExecutionError("worker_output_invalid") + if sha256(self.content).hexdigest() != self.sha256: + raise RecipeExecutionError("worker_output_hash_mismatch") + try: + detected = sniff_artifact_mime(self.content) + except ArtifactBoundaryError: + raise RecipeExecutionError("worker_output_invalid") from None + if detected != self.mime_type: + raise RecipeExecutionError("worker_output_mime_mismatch") + + +class RecipeWorkerAttempt(Protocol): + """One already-authenticated, bounded worker invocation.""" + + def transform( + self, + request_id: str, + job_id: str, + plan: ImageTransformPlan, + content: bytes, + cancel_event: Event, + ) -> RecipeWorkerOutput: + """Transform immutable bytes and return a validated output record.""" + + def cancel(self, reason: str = "user") -> None: + """Send a bounded cancellation request to the worker.""" + + def close(self) -> None: + """Close the authenticated worker transport and release handles.""" + + +RecipeWorkerAttemptFactory = Callable[[ExecutionJob], RecipeWorkerAttempt] + + +class RecipeWorkerConnection(Protocol): + """The narrow surface of an authenticated broker connection.""" + + def send_message(self, message: BrokerMessage) -> None: + ... + + def receive_message(self) -> BrokerMessage: + ... + + def close(self) -> None: + ... + + +class _ResponseReader: + """Make a blocking broker receive cancellable without changing the transport.""" + + def __init__(self, connection: RecipeWorkerConnection) -> None: + self.connection = connection + self.items: Queue[Any] = Queue(maxsize=16) + self.stop_event = Event() + self.thread = Thread(target=self._run, name="cortex-recipe-worker-reader", daemon=True) + + def start(self) -> None: + self.thread.start() + + def _run(self) -> None: + while not self.stop_event.is_set(): + try: + self.items.put(self.connection.receive_message()) + except Exception as error: + if not self.stop_event.is_set(): + try: + self.items.put(error, timeout=0.1) + except Exception: + pass + return + + def next(self, timeout: float) -> Any: + try: + return self.items.get(timeout=max(0.001, timeout)) + except Empty: + return None + + def close(self) -> None: + self.stop_event.set() + try: + self.connection.close() + except Exception: + pass + self.thread.join(timeout=1.0) + + +class RecipeWorkerClient: + """Client for the authenticated worker protocol, adapted as one attempt. + + The client accepts only in-memory bytes and typed plans. It validates every + response envelope, chunk offset, digest, MIME claim, and terminal condition. + """ + + def __init__( + self, + connection: RecipeWorkerConnection, + *, + installation_principal_id: str, + timeout_seconds: float = DEFAULT_WORKER_TIMEOUT_SECONDS, + cancel_grace_seconds: float = DEFAULT_CANCEL_GRACE_SECONDS, + ) -> None: + if not all(callable(getattr(connection, name, None)) for name in ("send_message", "receive_message", "close")): + raise TypeError("worker client requires an authenticated broker connection") + _safe_id(installation_principal_id, "installation_principal_id") + if re.fullmatch(r"[0-9a-f]{64}", installation_principal_id) is None: + raise ValueError("installation_principal_id must be a broker principal") + if not isinstance(timeout_seconds, (int, float)) or isinstance(timeout_seconds, bool) or not 0.1 <= timeout_seconds <= 600: + raise ValueError("timeout_seconds is outside the bounded worker limit") + if not isinstance(cancel_grace_seconds, (int, float)) or isinstance(cancel_grace_seconds, bool) or not 0.1 <= cancel_grace_seconds <= 30: + raise ValueError("cancel_grace_seconds is outside the bounded worker limit") + self._connection = connection + self._principal = installation_principal_id + self._timeout = float(timeout_seconds) + self._cancel_grace = float(cancel_grace_seconds) + self._reader: _ResponseReader | None = None + self._active_cancel: Event | None = None + self._cancel_sent = False + self._started = False + self._closed = False + self._lock = Lock() + + @property + def principal_id(self) -> str: + return self._principal + + def _message(self, operation: str, request_id: str, job_id: str, model: Any) -> BrokerMessage: + return BrokerMessage( + schema_version="broker.message.v1", + direction="to_executor", + operation=operation, + request_id=request_id, + job_id=job_id, + installation_principal_id=self._principal, + body=model.model_dump(mode="json"), + ) + + def _send(self, message: BrokerMessage) -> None: + try: + self._connection.send_message(message) + except Exception: + raise RecipeExecutionError("worker_transport_failed") from None + + def _validate_response(self, message: Any, request_id: str, job_id: str, operation: str) -> Any: + if not isinstance(message, BrokerMessage): + raise RecipeExecutionError("worker_message_invalid") + if ( + message.direction != "to_broker" + or message.operation not in {operation, "cancel"} + or message.request_id != request_id + or message.job_id != job_id + or message.installation_principal_id != self._principal + ): + raise RecipeExecutionError("worker_identity_mismatch") + body = message.body + if not isinstance(body, dict): + raise RecipeExecutionError("worker_message_invalid") + schema = body.get("schema_version") + try: + if schema == "recipe.worker.ack.v1": + return WorkerAck.model_validate(body) + if schema == "recipe.worker.result.v1": + return WorkerResult.model_validate(body) + if schema == "recipe.worker.error.v1": + return WorkerError.model_validate(body) + if schema == "recipe.worker.output_chunk.v1": + return WorkerOutputChunk.model_validate(body) + except (TypeError, ValueError): + raise RecipeExecutionError("worker_message_invalid") from None + raise RecipeExecutionError("worker_message_invalid") + + def _send_cancel(self, request_id: str, job_id: str, reason: str) -> None: + if self._cancel_sent: + return + if reason not in {"user", "timeout", "shutdown"}: + reason = "user" + self._cancel_sent = True + self._send( + self._message( + "cancel", + request_id, + job_id, + WorkerCancel( + schema_version="recipe.worker.cancel.v1", + request_id=request_id, + job_id=job_id, + reason=reason, + ), + ) + ) + + def _wait( + self, + *, + request_id: str, + job_id: str, + operation: str, + expected: tuple[type[Any], ...], + cancel_event: Event, + deadline: float, + ) -> Any: + if self._reader is None: + raise RecipeExecutionError("worker_transport_failed") + cancel_deadline: float | None = None + while True: + now = time.monotonic() + if cancel_event.is_set() and not self._cancel_sent: + self._send_cancel(request_id, job_id, "user") + cancel_deadline = now + self._cancel_grace + elif not self._cancel_sent and now >= deadline: + self._send_cancel(request_id, job_id, "timeout") + cancel_deadline = now + self._cancel_grace + if self._cancel_sent and cancel_deadline is not None and now >= cancel_deadline: + raise RecipeExecutionError("cancelled" if cancel_event.is_set() else "worker_timeout") + wait_for = 0.05 + if not self._cancel_sent: + wait_for = min(wait_for, max(0.001, deadline - now)) + elif cancel_deadline is not None: + wait_for = min(wait_for, max(0.001, cancel_deadline - now)) + item = self._reader.next(wait_for) + if item is None: + continue + if isinstance(item, BaseException): + raise RecipeExecutionError("cancelled" if self._cancel_sent and cancel_event.is_set() else "worker_transport_failed") + response = self._validate_response(item, request_id, job_id, operation) + if isinstance(response, WorkerError): + if response.code == "cancelled" or self._cancel_sent and cancel_event.is_set(): + raise RecipeExecutionError("cancelled") + if response.code == "timeout": + raise RecipeExecutionError("worker_timeout") + raise RecipeExecutionError(response.code) + if self._cancel_sent: + if isinstance(response, WorkerAck) and response.acknowledged_operation == "cancel": + raise RecipeExecutionError("cancelled" if cancel_event.is_set() else "worker_timeout") + continue + if not isinstance(response, expected): + raise RecipeExecutionError("worker_response_unexpected") + return response + + def transform( + self, + request_id: str, + job_id: str, + plan: ImageTransformPlan, + content: bytes, + cancel_event: Event, + ) -> RecipeWorkerOutput: + if self._started or self._closed: + raise RecipeExecutionError("worker_attempt_reused") + _safe_id(request_id, "request_id") + _safe_id(job_id, "job_id") + if not isinstance(plan, ImageTransformPlan): + raise RecipeExecutionError("invalid_plan") + if not isinstance(content, bytes) or not 1 <= len(content) <= MAX_WORKER_INPUT_BYTES: + raise RecipeExecutionError("input_too_large" if isinstance(content, bytes) else "invalid_input") + if not isinstance(cancel_event, Event): + raise TypeError("cancel_event must be a threading.Event") + self._started = True + self._active_cancel = cancel_event + self._cancel_sent = False + self._reader = _ResponseReader(self._connection) + self._reader.start() + digest = sha256(content).hexdigest() + try: + try: + input_mime = sniff_artifact_mime(content) + except ArtifactBoundaryError: + raise RecipeExecutionError("invalid_input") from None + if input_mime not in _MIME_TO_FORMAT: + raise RecipeExecutionError("invalid_input") + prepare = WorkerPrepare( + schema_version="recipe.worker.prepare.v1", + request_id=request_id, + job_id=job_id, + plan=plan, + input_size=len(content), + input_sha256=digest, + input_mime_type=input_mime, + ) + deadline = time.monotonic() + self._timeout + self._send(self._message("prepare", request_id, job_id, prepare)) + self._wait( + request_id=request_id, + job_id=job_id, + operation="prepare", + expected=(WorkerAck,), + cancel_event=cancel_event, + deadline=deadline, + ) + for offset in range(0, len(content), MAX_WORKER_CHUNK_BYTES): + chunk = content[offset : offset + MAX_WORKER_CHUNK_BYTES] + message = WorkerInputChunk( + schema_version="recipe.worker.input_chunk.v1", + request_id=request_id, + job_id=job_id, + offset=offset, + data=base64.urlsafe_b64encode(chunk).decode("ascii").rstrip("="), + sha256=sha256(chunk).hexdigest(), + ) + self._send(self._message("input_chunk", request_id, job_id, message)) + self._wait( + request_id=request_id, + job_id=job_id, + operation="input_chunk", + expected=(WorkerAck,), + cancel_event=cancel_event, + deadline=deadline, + ) + complete = WorkerInputComplete( + schema_version="recipe.worker.input_complete.v1", + request_id=request_id, + job_id=job_id, + input_size=len(content), + input_sha256=digest, + ) + self._send(self._message("input_complete", request_id, job_id, complete)) + result = self._wait( + request_id=request_id, + job_id=job_id, + operation="input_complete", + expected=(WorkerResult,), + cancel_event=cancel_event, + deadline=deadline, + ) + assert isinstance(result, WorkerResult) + output = bytearray() + while len(output) < result.output_size: + collect = WorkerCollect( + schema_version="recipe.worker.collect.v1", + request_id=request_id, + job_id=job_id, + offset=len(output), + max_bytes=MAX_WORKER_CHUNK_BYTES, + ) + self._send(self._message("collect", request_id, job_id, collect)) + chunk = self._wait( + request_id=request_id, + job_id=job_id, + operation="collect", + expected=(WorkerOutputChunk,), + cancel_event=cancel_event, + deadline=deadline, + ) + assert isinstance(chunk, WorkerOutputChunk) + if chunk.offset != len(output): + raise RecipeExecutionError("worker_output_offset_invalid") + try: + decoded = chunk.decoded() + except WorkerProtocolError: + raise RecipeExecutionError("worker_output_chunk_invalid") from None + if not decoded or len(output) + len(decoded) > result.output_size: + raise RecipeExecutionError("worker_output_size_mismatch") + output.extend(decoded) + if chunk.final and len(output) != result.output_size: + raise RecipeExecutionError("worker_output_size_mismatch") + if not output or len(output) != result.output_size or not chunk.final: + raise RecipeExecutionError("worker_output_size_mismatch") + content_out = bytes(output) + digest_out = sha256(content_out).hexdigest() + if digest_out != result.output_sha256: + raise RecipeExecutionError("worker_output_hash_mismatch") + if result.mime_type not in _MIME_TO_FORMAT or _MIME_TO_FORMAT[result.mime_type] != result.format: + raise RecipeExecutionError("worker_output_mime_mismatch") + return RecipeWorkerOutput( + content=content_out, + mime_type=result.mime_type, + format=result.format, + width=result.width, + height=result.height, + sha256=digest_out, + ) + except RecipeExecutionError: + raise + except (TypeError, ValueError, WorkerProtocolError): + raise RecipeExecutionError("worker_protocol_failed") from None + + def cancel(self, reason: str = "user") -> None: + if reason not in {"user", "timeout", "shutdown"}: + reason = "user" + active = self._active_cancel + if active is not None: + active.set() + + def close(self) -> None: + if self._closed: + return + self._closed = True + if self._reader is not None: + self._reader.close() + else: + try: + self._connection.close() + except Exception: + pass + + +class RecipeExecutionCoordinator: + """Durable, owner-scoped coordinator for the qualified image recipe.""" + + def __init__( + self, + repository: ExecutionRepository, + worker_factory: RecipeWorkerAttemptFactory, + *, + artifact_boundary: ArtifactBoundary | None = None, + lease_seconds: float = 30.0, + supervisor_lease_seconds: float = 30.0, + auto_recover: bool = False, + ) -> None: + if not isinstance(repository, ExecutionRepository): + raise TypeError("repository must be an ExecutionRepository") + if not callable(worker_factory): + raise TypeError("worker_factory must be callable") + if lease_seconds <= 0 or supervisor_lease_seconds <= 0: + raise ValueError("lease durations must be positive") + self.repository = repository + self.worker_factory = worker_factory + self.artifact_boundary = artifact_boundary or ArtifactBoundary(repository) + self.lease_seconds = float(lease_seconds) + self.supervisor_lease_seconds = float(supervisor_lease_seconds) + self._supervisor_owner = f"recipe-supervisor-{uuid4().hex}" + self._supervisor_lease_active = False + self._lock = Lock() + self._threads: dict[str, Thread] = {} + self._cancel_events: dict[str, Event] = {} + self._attempts: dict[str, RecipeWorkerAttempt] = {} + if auto_recover: + self.startup_recover() + + def start_image_transform(self, request: RecipeImageRequest) -> ExecutionJob: + if not isinstance(request, RecipeImageRequest): + raise TypeError("request must be a RecipeImageRequest") + self._load_input(request.owner, request.source_artifact_id) + payload = self._payload_for_request(request) + job, created = self.repository.create_job( + job_id=uuid4().hex, + owner=request.owner, + request_id=request.request_id, + profile=RECIPE_IMAGE_PROFILE, + payload=payload, + ) + if not created: + if job.profile != RECIPE_IMAGE_PROFILE or self._canonical_payload(job.payload) != self._canonical_payload(payload): + raise RecipeExecutionError("request_conflict") + return job + self._launch(job.job_id, request) + return job + + start = start_image_transform + + def resume(self, job_id: str) -> ExecutionJob: + job = self.repository.get_job(job_id) + if job is None: + raise ValueError("execution job does not exist") + if job.profile != RECIPE_IMAGE_PROFILE: + raise RecipeExecutionError("recovery_profile_invalid") + if job.status in TerminalExecutionStatus: + return job + request = self._request_from_job(job) + self._launch(job.job_id, request) + return self.repository.get_job(job.job_id) or job + + def wait(self, job_id: str, *, timeout: float = 5.0) -> ExecutionJob: + if timeout < 0: + raise ValueError("timeout must be non-negative") + deadline = time.monotonic() + timeout + while True: + job = self.repository.get_job(job_id) + if job is None: + raise ValueError("execution job does not exist") + if job.status in TerminalExecutionStatus: + return job + if time.monotonic() >= deadline: + raise TimeoutError("execution job did not reach a terminal state") + time.sleep(0.005) + + def cancel(self, job_id: str, *, owner: str) -> ExecutionJob: + job = self.repository.get_job(job_id, owner=owner) + if job is None: + raise ValueError("execution job does not exist or is not owned by caller") + with self._lock: + event = self._cancel_events.get(job_id) + attempt = self._attempts.get(job_id) + if event is not None: + event.set() + if attempt is not None: + try: + attempt.cancel("user") + except Exception: + pass + return self.repository.request_cancel(job_id) + + def shutdown(self, *, timeout: float = 5.0) -> None: + if timeout < 0: + raise ValueError("timeout must be non-negative") + with self._lock: + events = list(self._cancel_events.values()) + attempts = list(self._attempts.values()) + threads = list(self._threads.values()) + for event in events: + event.set() + for attempt in attempts: + try: + attempt.cancel("shutdown") + except Exception: + pass + deadline = time.monotonic() + timeout + for thread in threads: + thread.join(timeout=max(0.0, deadline - time.monotonic())) + if self._supervisor_lease_active: + self.repository.release_supervisor_lease(lease_owner=self._supervisor_owner) + self._supervisor_lease_active = False + + def startup_recover(self) -> list[str]: + if self._supervisor_lease_active: + return [] + self.repository.claim_supervisor_lease( + lease_owner=self._supervisor_owner, + ttl_seconds=self.supervisor_lease_seconds, + ) + self._supervisor_lease_active = True + recovered = self.repository.recover_expired_leases() + self.repository.expire_approvals() + for job_id in recovered: + job = self.repository.get_job(job_id) + if job is None or job.status in TerminalExecutionStatus: + continue + if job.profile != RECIPE_IMAGE_PROFILE: + continue + events = self.repository.events(job_id) + if len(events) >= 2 and events[-2].event == "cancelling": + try: + self.repository.transition( + job_id, + status="cancelled", + event="cancelled", + phase="recovery", + data={"message": "Recipe cancellation was recovered."}, + error="cancelled", + ) + except Exception: + pass + continue + try: + request = self._request_from_job(job) + self._load_input(request.owner, request.source_artifact_id) + except (RecipeExecutionError, ValueError, TypeError, KeyError, RecipeValidationError): + self._fail_recovery(job_id, "recovery_invalid_payload") + continue + self._launch(job_id, request) + return recovered + + def _fail_recovery(self, job_id: str, code: str) -> None: + try: + self.repository.transition( + job_id, + status="failed", + event="failed", + phase="recovery", + data={"message": "Recipe recovery metadata is invalid."}, + error=code, + ) + except Exception: + pass + + @staticmethod + def _canonical_payload(payload: Mapping[str, Any]) -> str: + return json.dumps(dict(payload), ensure_ascii=True, sort_keys=True, separators=(",", ":")) + + @staticmethod + def _payload_for_request(request: RecipeImageRequest) -> Mapping[str, Any]: + return { + "schema_version": RECIPE_PAYLOAD_SCHEMA, + "provider": "recipe-image-v1", + "source_artifact_id": request.source_artifact_id, + "plan": json.loads(request.plan.canonical_json()), + "plan_digest": request.plan.digest(), + "retention_seconds": request.retention_seconds, + } + + @staticmethod + def _request_from_job(job: ExecutionJob) -> RecipeImageRequest: + payload = job.payload + if ( + set(payload) != { + "schema_version", + "provider", + "source_artifact_id", + "plan", + "plan_digest", + "retention_seconds", + } + or payload.get("schema_version") != RECIPE_PAYLOAD_SCHEMA + or payload.get("provider") != "recipe-image-v1" + ): + raise RecipeExecutionError("recovery_invalid_payload") + source_artifact_id = _safe_id(payload.get("source_artifact_id"), "source_artifact_id") + plan_payload = payload.get("plan") + try: + plan = parse_image_transform(plan_payload if isinstance(plan_payload, Mapping) else {}) + except RecipeValidationError: + raise RecipeExecutionError("recovery_invalid_payload") from None + if payload.get("plan_digest") != plan.digest(): + raise RecipeExecutionError("recovery_invalid_payload") + retention = payload.get("retention_seconds") + try: + return RecipeImageRequest( + owner=job.owner, + request_id=job.request_id, + source_artifact_id=source_artifact_id, + plan=plan, + retention_seconds=retention, + ) + except (TypeError, ValueError, RecipeExecutionError): + raise RecipeExecutionError("recovery_invalid_payload") from None + + def _load_input(self, owner: str, artifact_id: str) -> bytes: + artifact = self.repository.get_artifact(artifact_id, owner=owner) + if artifact is None: + raise RecipeExecutionError("input_artifact_unavailable") + if artifact.mime_type not in _MIME_TO_FORMAT: + raise RecipeExecutionError("input_artifact_invalid") + try: + content = self.repository.read_artifact(artifact.artifact_id) + detected = sniff_artifact_mime(content) + except (ExecutionRepositoryError, ArtifactBoundaryError): + raise RecipeExecutionError("input_artifact_unavailable") from None + if detected != artifact.mime_type or len(content) > MAX_WORKER_INPUT_BYTES: + raise RecipeExecutionError("input_artifact_invalid") + return content + + def _launch(self, job_id: str, request: RecipeImageRequest) -> None: + with self._lock: + existing = self._threads.get(job_id) + if existing is not None and existing.is_alive(): + return + event = self._cancel_events.setdefault(job_id, Event()) + thread = Thread( + target=self._run, + args=(job_id, request, event), + name=f"cortex-recipe-{job_id}", + daemon=True, + ) + self._threads[job_id] = thread + thread.start() + + @staticmethod + def _write_staging(root: Path, content: bytes) -> None: + path = root / "output" + try: + with path.open("xb") as stream: + stream.write(content) + stream.flush() + os.fsync(stream.fileno()) + except OSError: + raise RecipeExecutionError("artifact_staging_failed") from None + + def _publish( + self, + job: ExecutionJob, + output: RecipeWorkerOutput, + retention_seconds: int, + ) -> tuple[PublishedArtifact, ...]: + if len(output.content) > self.repository.max_artifact_bytes: + raise RecipeExecutionError("worker_output_too_large") + try: + with tempfile.TemporaryDirectory( + prefix=f".recipe-{job.job_id}-", dir=str(self.repository.artifact_root) + ) as staging: + root = Path(staging) + self._write_staging(root, output.content) + return self.artifact_boundary.collect_outputs( + job.job_id, + job.owner, + root, + (OutputClaim("output", output.mime_type),), + retention_seconds=retention_seconds, + ) + except ArtifactBoundaryError: + raise RecipeExecutionError("artifact_publication_failed") from None + + @staticmethod + def _result(output: RecipeWorkerOutput, published: PublishedArtifact, plan: ImageTransformPlan) -> Mapping[str, Any]: + artifact = published.artifact + return { + "schema_version": RECIPE_RESULT_SCHEMA, + "artifact_id": artifact.artifact_id, + "mime_type": output.mime_type, + "format": output.format, + "size": artifact.size, + "sha256": artifact.sha256, + "width": output.width, + "height": output.height, + "plan_digest": plan.digest(), + } + + def _delete_published(self, published: tuple[PublishedArtifact, ...]) -> None: + for item in published: + try: + self.repository.delete_artifact(item.artifact.artifact_id) + except ExecutionRepositoryError: + raise RecipeExecutionError("artifact_cleanup_pending") from None + + def _run(self, job_id: str, request: RecipeImageRequest, cancel_event: Event) -> None: + lease_owner = f"recipe-coordinator-{uuid4().hex}" + attempt: RecipeWorkerAttempt | None = None + with self._lock: + self._cancel_events[job_id] = cancel_event + published: tuple[PublishedArtifact, ...] = () + try: + self.repository.claim_lease(job_id, lease_owner=lease_owner, ttl_seconds=self.lease_seconds) + current = self.repository.get_job(job_id) + if current is None: + return + if current.status == "cancelling" or cancel_event.is_set(): + self.repository.transition( + job_id, + status="cancelled", + event="cancelled", + phase="cancelled", + data={"message": "Recipe cancelled before start."}, + error="cancelled", + ) + return + self.repository.transition( + job_id, + status="running", + event="started", + phase="prepare", + data={"message": "Image recipe started."}, + ) + content = self._load_input(request.owner, request.source_artifact_id) + self.repository.transition( + job_id, + status="running", + event="progress", + phase="worker", + data={"message": "Image recipe worker is processing the staged artifact."}, + ) + current = self.repository.get_job(job_id) or current + attempt = self.worker_factory(current) + if not all(callable(getattr(attempt, name, None)) for name in ("transform", "cancel", "close")): + raise RecipeExecutionError("worker_attempt_invalid") + with self._lock: + self._attempts[job_id] = attempt + output = attempt.transform( + request.request_id, + job_id, + request.plan, + content, + cancel_event, + ) + if not isinstance(output, RecipeWorkerOutput): + raise RecipeExecutionError("worker_output_invalid") + current = self.repository.get_job(job_id) + if cancel_event.is_set() or (current is not None and current.status == "cancelling"): + raise RecipeExecutionError("cancelled") + if current is None: + raise RecipeExecutionError("coordinator_failed") + published = self._publish(current, output, request.retention_seconds) + current = self.repository.get_job(job_id) + if cancel_event.is_set() or (current is not None and current.status == "cancelling"): + self._delete_published(published) + published = () + raise RecipeExecutionError("cancelled") + self.repository.transition( + job_id, + status="succeeded", + event="completed", + phase="completed", + data={"message": "Image recipe completed."}, + result=self._result(output, published[0], request.plan), + ) + except RecipeExecutionError as execution_error: + failure_code = execution_error.code + if published: + try: + self._delete_published(published) + except RecipeExecutionError: + failure_code = "artifact_cleanup_pending" + current = self.repository.get_job(job_id) + cancelled = failure_code == "cancelled" or cancel_event.is_set() or (current is not None and current.status == "cancelling") + try: + self.repository.transition( + job_id, + status="cancelled" if cancelled else "failed", + event="cancelled" if cancelled else "failed", + phase="cancelled" if cancelled else "failed", + data={"message": "Image recipe was cancelled." if cancelled else "Image recipe failed safely."}, + error="cancelled" if cancelled else failure_code, + ) + except Exception: + pass + except LeaseConflict: + self._fail_recovery(job_id, "lease_unavailable") + except Exception: + current = self.repository.get_job(job_id) + cancelled = cancel_event.is_set() or (current is not None and current.status == "cancelling") + try: + self.repository.transition( + job_id, + status="cancelled" if cancelled else "failed", + event="cancelled" if cancelled else "failed", + phase="cancelled" if cancelled else "failed", + data={"message": "Image recipe was cancelled." if cancelled else "Image recipe failed safely."}, + error="cancelled" if cancelled else "coordinator_failed", + ) + except Exception: + pass + finally: + if attempt is not None: + try: + attempt.close() + except Exception: + pass + with self._lock: + self._attempts.pop(job_id, None) + self._cancel_events.pop(job_id, None) + self._threads.pop(job_id, None) + try: + self.repository.release_lease(job_id, lease_owner=lease_owner) + except Exception: + pass + + +__all__ = [ + "DEFAULT_CANCEL_GRACE_SECONDS", + "DEFAULT_RECIPE_RETENTION_SECONDS", + "DEFAULT_WORKER_TIMEOUT_SECONDS", + "MAX_RECIPE_RETENTION_SECONDS", + "RecipeExecutionCoordinator", + "RecipeExecutionError", + "RecipeImageRequest", + "RecipeWorkerAttempt", + "RecipeWorkerAttemptFactory", + "RecipeWorkerClient", + "RecipeWorkerConnection", + "RecipeWorkerOutput", + "RECIPE_IMAGE_PROFILE", + "RECIPE_PAYLOAD_SCHEMA", + "RECIPE_RESULT_SCHEMA", +] diff --git a/backend/cortex_backend/execution/repository.py b/backend/cortex_backend/execution/repository.py index 342f854..822b3c7 100644 --- a/backend/cortex_backend/execution/repository.py +++ b/backend/cortex_backend/execution/repository.py @@ -343,6 +343,60 @@ def get_job(self, job_id: str, *, owner: str | None = None) -> ExecutionJob | No return None return self._job_from_row(row) + def replace_job_payload( + self, + job_id: str, + payload: Mapping[str, Any], + *, + expected_status: ExecutionStatus = "queued", + ) -> ExecutionJob: + """Replace a queued job payload before its first worker lease. + + The payload is control-plane metadata, not an event. It is updated under + an immediate transaction and cannot be changed once a worker has claimed + the job or the job has left the expected state. + """ + + if not isinstance(payload, Mapping): + raise ExecutionRepositoryError("Execution payload is invalid.") + try: + encoded = json.dumps( + dict(payload), + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ) + except (TypeError, ValueError, OverflowError): + raise ExecutionRepositoryError("Execution payload is invalid.") from None + if len(encoded.encode("utf-8")) > MAX_EVENT_BYTES: + raise ExecutionRepositoryError("Execution payload is too large.") + if expected_status not in {"queued", "running", "cancelling"}: + raise ValueError("expected_status is invalid") + now = self._now() + with self.connect() as connection: + connection.execute("BEGIN IMMEDIATE") + row = connection.execute( + "SELECT status FROM execution_jobs WHERE job_id = ?", (job_id,) + ).fetchone() + if row is None: + raise ExecutionRepositoryError("Execution job does not exist.") + if row["status"] != expected_status: + raise ExecutionRepositoryError("Execution payload is no longer mutable.") + lease = connection.execute( + "SELECT 1 FROM execution_leases WHERE job_id = ?", (job_id,) + ).fetchone() + if lease is not None: + raise ExecutionRepositoryError("Execution payload is no longer mutable.") + connection.execute( + "UPDATE execution_jobs SET payload_json = ?, updated_at = ? WHERE job_id = ?", + (encoded, now, job_id), + ) + updated = self.get_job(job_id) + if updated is None: + raise ExecutionRepositoryError("Execution job does not exist.") + return updated + def list_jobs( self, *, @@ -711,7 +765,6 @@ def claim_lease(self, job_id: str, *, lease_owner: str, ttl_seconds: float = 30. raise ValueError("ttl_seconds must be positive") now = datetime.now(timezone.utc) expires = now + timedelta(seconds=ttl_seconds) - now_text = now.isoformat() expires_text = expires.isoformat() with self.connect() as connection: connection.execute("BEGIN IMMEDIATE") @@ -865,6 +918,41 @@ def publish_artifact( expires_at=expires.isoformat(), ) + def get_artifact( + self, + artifact_id: str, + *, + owner: str | None = None, + ) -> ExecutionArtifact | None: + """Return artifact metadata only when its owning job is visible.""" + + if not isinstance(artifact_id, str) or not _SAFE_NAME.fullmatch(artifact_id): + return None + with self.connect() as connection: + row = connection.execute( + """ + SELECT a.* + FROM execution_artifacts a + JOIN execution_jobs j ON j.job_id = a.job_id + WHERE a.artifact_id = ? + AND (? IS NULL OR j.owner = ?) + """, + (artifact_id, owner, owner), + ).fetchone() + if row is None: + return None + return ExecutionArtifact( + artifact_id=row["artifact_id"], + job_id=row["job_id"], + name=row["name"], + mime_type=row["mime_type"], + size=int(row["size"]), + sha256=row["sha256"], + path=row["path"], + created_at=row["created_at"], + expires_at=row["expires_at"], + ) + def delete_artifact(self, artifact_id: str) -> None: """Remove one unpublished/rolled-back artifact record and file safely.""" diff --git a/docs/adr/0001-phase2-artifact-boundary.md b/docs/adr/0001-phase2-artifact-boundary.md index d7c9182..cac20e6 100644 --- a/docs/adr/0001-phase2-artifact-boundary.md +++ b/docs/adr/0001-phase2-artifact-boundary.md @@ -105,6 +105,7 @@ recorded in the [Phase 2 evidence log](0001-phase2-evidence.md). This ADR does not authorize image decoding, thumbnail generation, archive extraction, provider loading, code execution, model tool exposure, network access, or automatic -execution. The artifact security review is complete; provider enablement remains -blocked on resource/watchdog accounting, external security review, and lifecycle -health gates in the parent ADR. +execution. The artifact security review is complete. The internal recipe +coordinator now consumes this boundary only after the separate resource/watchdog +and lifecycle gates pass; official external review and production trust remain +optional release hardening. diff --git a/docs/adr/0001-phase2-evidence.md b/docs/adr/0001-phase2-evidence.md index 451e8ec..8bbbd5d 100644 --- a/docs/adr/0001-phase2-evidence.md +++ b/docs/adr/0001-phase2-evidence.md @@ -35,8 +35,9 @@ | 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, and checks cancellation. The provider request route remains separate and default-off. | | Windows recipe sandbox qualification harness | **Complete (signed launch/broker/hostile/cancellation/resource/watchdog)** | `recipe_worker_e2e_qualification.py` signs a disposable package with an in-memory key, installs/verifies one immutable generation, binds the live AppContainer identity to the broker, and exercises the fixed PNG transform, truncated-PNG decoder rejection, active-SVG decoder rejection, and in-flight cancellation corpus. `resource_watchdog_qualification.py` separately proves immutable budgets, actual Job Object accounting, and kill-on-close tree reaping. The native broker uses bounded availability polling before reads so the cancellation reader remains live while the packaged provider transforms. | | Suspended native launcher/resource policy | **Complete (factory + binder + ACL cleanup + qualification evidence)** | `NativeWin32ProcessFactory` grants only inherited read/execute access to the fresh AppContainer SID on the verified package root, applies and verifies Job Object policy before resume, and removes the per-launch ACE during cleanup. `NativeBrokerIdentityBinder` pins the live server to the worker PID/AppContainer SID and launcher cleanup closes it on failure. | -| OS sandbox provider and provider-produced image outputs | **Complete (qualification; default-off in the app)** | Signed installation, provenance, AppContainer/job identity, broker handshake, `prepare`, `input_chunk`, `input_complete`, `collect_output`, hostile decoder rejection, in-flight cancellation acknowledgement, artifact security review, resource accounting, and watchdog tree reaping are qualified. The explicit lifecycle composition is now wired; the next core slice is the recipe coordinator/request path. Official-release review/signing remains optional hardening. | +| OS sandbox provider and provider-produced image outputs | **Complete (qualification; default-off in the app)** | Signed installation, provenance, AppContainer/job identity, broker handshake, `prepare`, `input_chunk`, `input_complete`, `collect_output`, hostile decoder rejection, in-flight cancellation acknowledgement, artifact security review, resource accounting, and watchdog tree reaping are qualified. The explicit lifecycle composition and internal recipe coordinator/publication path are now wired behind injection; UI/API exposure remains separate and default-off. Official-release review/signing remains optional hardening. | | Explicit qualification-profile lifecycle wiring | **Complete (local/CI composition; default-off in the app)** | `build_execution_lifecycle()` accepts only exact `disabled`/`qualification` selection, requires a qualification release gate, coordinator factory, and provider-health probe, and composes them in a fail-closed order. `Cortex_Preview.build_preview_app` exposes this only as an explicit injection; the normal app supplies no profile or controls. | +| Durable recipe coordinator/request and artifact publication | **Complete (internal qualification composition; default-off in the app)** | `RecipeExecutionCoordinator` persists only opaque artifact IDs and canonical plan digests, enforces owner-scoped input reads and idempotency conflicts, leases/recoveries/cancels attempts, validates worker envelopes/chunks, and publishes exactly one output through `ArtifactBoundary.collect_outputs`. `tests/test_phase2_recipe_coordinator.py` covers the hostile and cancellation paths. | ## Security invariants @@ -292,6 +293,8 @@ hardening. They do not block the open-source source checkout or local qualification. The release gate exposes an explicit `release_profile="qualification"`, and the new lifecycle builder composes that profile only when a caller injects the qualification gate, coordinator, and -provider-health probe. The normal application remains default-off. The next core -implementation slice is the recipe-specific coordinator/request and artifact -publication path; no outside reviewer or production key is required for it. +provider-health probe. The normal application remains default-off. The durable +recipe-specific coordinator/request and artifact-publication path is now +implemented as an internal qualification composition; no outside reviewer or +production key was required for it. The next core slice is the explicit UI/API +request surface and qualified worker-attempt factory wiring. diff --git a/docs/adr/0001-phase2-qualification-lifecycle.md b/docs/adr/0001-phase2-qualification-lifecycle.md index 2f4746c..879184a 100644 --- a/docs/adr/0001-phase2-qualification-lifecycle.md +++ b/docs/adr/0001-phase2-qualification-lifecycle.md @@ -1,6 +1,6 @@ # ADR-0001 Phase 2 explicit qualification-profile lifecycle -- **Status:** Implemented as a local/CI lifecycle composition boundary; recipe request execution remains a separate slice +- **Status:** Implemented as a local/CI lifecycle composition boundary; the internal recipe coordinator is now available behind it and application exposure remains separate - **Phase:** 2 - fixed-function image provider - **Parent:** [Capability-tiered agentic execution harness](0001-capability-tiered-agentic-execution-harness.md) - **Depends on:** [Phase 2 release/lifecycle preflight](0001-phase2-release-lifecycle-gate.md), [recipe provider](0001-phase2-recipe-provider.md), and [Phase 1 lifecycle](0001-phase1-production-lifecycle.md) @@ -53,10 +53,11 @@ qualification code without an outside reviewer, production signing key, or trusted release root. The normal application cannot be enabled accidentally by importing the provider or setting an implicit default. -The next Phase 2 slice is the recipe-specific coordinator/request and artifact -publication path that can be injected behind this lifecycle. It must preserve -the same release gate, native launcher, broker identity, resource/watchdog, and -trusted artifact controls. +The internal recipe-specific coordinator/request and artifact-publication path is +now implemented behind this lifecycle boundary. It preserves the same release +gate, native launcher, broker identity, resource/watchdog, and trusted artifact +controls. The next slice is the explicit UI/API request surface and qualified +worker-attempt factory wiring; it must remain default-off in the application. ## Verification diff --git a/docs/adr/0001-phase2-recipe-contract.md b/docs/adr/0001-phase2-recipe-contract.md index 9cb094a..b72382f 100644 --- a/docs/adr/0001-phase2-recipe-contract.md +++ b/docs/adr/0001-phase2-recipe-contract.md @@ -1,6 +1,6 @@ # ADR-0001 Phase 2 typed recipe and primitive contract -- **Status:** Typed contract, signed-manifest verification, native broker transport, signed bundle installation, trusted artifact boundary, qualification-only provider core, and explicit qualification-profile lifecycle composition implemented and verified; recipe request execution remains next +- **Status:** Typed contract, signed-manifest verification, native broker transport, signed bundle installation, trusted artifact boundary, qualification-only provider core, explicit qualification-profile lifecycle composition, and the internal durable recipe coordinator implemented and verified; application integration remains separate - **Parent:** [Capability-tiered agentic execution harness](0001-capability-tiered-agentic-execution-harness.md) - **Depends on:** [Phase 1 production lifecycle gate](0001-phase1-production-lifecycle.md) - **Scope:** Typed fixed-function image plans, calculator/check primitives, canonical @@ -71,11 +71,12 @@ hardening, not open-source prerequisites. ## Required next gates -1. Add the recipe-specific coordinator/request and artifact-publication path behind - the passing lifecycle health check for the explicit - `release_profile="qualification"`. The lifecycle builder intentionally does not - create processes or expose a recipe route by itself. Official prebuilt releases - may add external review and production trust as optional hardening. +1. Wire an explicit UI/API request surface and a qualified worker-attempt factory + behind the passing lifecycle health check. The implemented coordinator is + documented in [the coordinator ADR](0001-phase2-recipe-coordinator.md); the + lifecycle builder intentionally does not create processes or expose a recipe + route by itself. Official prebuilt releases may add external review and + production trust as optional hardening. ## Verification diff --git a/docs/adr/0001-phase2-recipe-coordinator.md b/docs/adr/0001-phase2-recipe-coordinator.md new file mode 100644 index 0000000..22aa9cb --- /dev/null +++ b/docs/adr/0001-phase2-recipe-coordinator.md @@ -0,0 +1,111 @@ +# ADR-0001 Phase 2 durable recipe coordinator and artifact publication + +- **Status:** Implemented and verified as an internal qualification composition + boundary; application/API exposure remains a separate default-off decision +- **Phase:** 2 - fixed-function image recipe +- **Parent:** [Capability-tiered agentic execution harness](0001-capability-tiered-agentic-execution-harness.md) +- **Depends on:** [typed recipe contract](0001-phase2-recipe-contract.md), + [trusted artifact boundary](0001-phase2-artifact-boundary.md), + [worker protocol](0001-phase2-worker-protocol.md), and + [qualification lifecycle](0001-phase2-qualification-lifecycle.md) +- **Scope:** Durable request/attempt coordination, cancellation and recovery, + and all-or-nothing publication for the fixed image recipe + +## Decision + +The qualified image recipe is coordinated by +`cortex_backend.execution.recipe_coordinator.RecipeExecutionCoordinator`. +The coordinator is an internal composition seam, not a new application route. +It can be injected by the explicit `release_profile="qualification"` +lifecycle, while the normal application remains disabled by default. + +The request contract is `RecipeImageRequest`. It contains an owner, an +idempotency request identifier, an opaque `source_artifact_id`, a validated +`ImageTransformPlan`, and a bounded retention period. The plan's input artifact +identifier must match the request identifier exactly. No source path, filename, +command, model name, shell text, network target, or executable authority is +accepted or written to durable job state. + +The worker seam is `RecipeWorkerAttempt`. A factory receives the durable job and +returns one already-authenticated attempt. The attempt receives immutable input +bytes and a typed plan, and returns `RecipeWorkerOutput` with byte-derived MIME, +format, dimensions, and SHA-256. The concrete `RecipeWorkerClient` speaks only +the existing authenticated broker/worker protocol; it does not open a process, +choose a provider, or invent a fallback transport. + +## Durable lifecycle + +1. The coordinator owner-checks and reads the staged source artifact through the + repository. The artifact's stored size, digest, and byte-derived MIME are + rechecked before a worker is created. +2. A queued `recipe.image.v1` job stores only the versioned recipe payload, + opaque artifact ID, canonical plan, plan digest, provider version, and bounded + retention. SQLite uniqueness on `(owner, request_id)` makes retries + idempotent; a different payload with the same key is a stable + `request_conflict`. +3. A per-job lease and background thread own the attempt. Startup recovery claims + the supervisor lease, recovers expired job leases, validates the persisted + payload, and resumes only the known recipe profile. Invalid recovery metadata + becomes `recovery_invalid_payload`; it is never interpreted as executable + input. +4. Cancellation sets a durable `cancelling` state, signals the worker attempt, + and waits only within the worker's bounded cancellation path. A cancelled + attempt cannot publish a successful result. +5. A valid output is written to a private temporary staging directory under the + artifact root. `ArtifactBoundary.collect_outputs` requires the exact single + `output` claim, re-sniffs bytes, enforces size/link/reparse/hash limits, and + publishes atomically. Any publication failure rolls back records and + quarantines or reports cleanup failure through a stable category. +6. The terminal result contains only the published artifact ID, safe MIME/format, + size, digest, dimensions, and plan digest. It never returns the staging or + repository path. A cancellation race after publication deletes the unpublished + records before recording `cancelled`. + +## Worker protocol client + +`RecipeWorkerClient` uses a reader thread around the existing blocking +authenticated connection so cancellation can be sent while the provider is +transforming. It sends `prepare`, bounded independently hashed input chunks, +`input_complete`, and `collect` messages. Every response is checked for direction, +principal, request/job identity, operation, schema, chunk offset, chunk digest, +output size, output digest, MIME, and format. A worker error is reduced to its +stable code; transport, timeout, malformed-message, and cancellation categories +are distinct. A late result after cancellation is never accepted. + +## Failure and recovery policy + +The coordinator exposes stable categories such as `input_artifact_unavailable`, +`worker_transport_failed`, `worker_timeout`, `worker_output_invalid`, +`artifact_publication_failed`, `artifact_cleanup_pending`, `cancelled`, and +`coordinator_failed`. Raw paths, decoder errors, broker payloads, process IDs, +tokens, and stack traces do not enter job events or results. Terminal repository +state is immutable, so late worker callbacks cannot overwrite a validated result. + +The worker factory is deliberately injected. This stage does not pretend that an +unsigned package, missing broker identity binding, or absent production trust root +is a usable runtime. Those are release/qualification composition inputs owned by +the existing lifecycle and launcher gates. + +## Explicit non-goals + +This ADR does not add a public API route, automatic model tool selection, arbitrary +Python/WASI execution, source-path access, direct user-file mutation, network +access, application-exit persistence, production signing, or external-review +requirements. It also does not make the qualification profile the application +default. + +## Verification + +`tests/test_phase2_recipe_coordinator.py` covers opaque request binding, output +digest/MIME revalidation, owner-scoped publication, idempotency conflicts, +redacted worker failure, cancellation cleanup, and a real authenticated worker +runtime round trip including in-flight cancellation. The repository-wide matrix +passed with **324 passed, 1 skipped** on 2026-07-29. + +## Next stage + +The next implementation decision is application integration: define the explicit +UI/API request surface and wire a qualified worker-attempt factory into the +qualification lifecycle. That stage must preserve this coordinator contract and +keep the normal application default-off. No external reviewer or production key +is required to continue the open-source qualification path. diff --git a/docs/adr/0001-phase2-recipe-provider.md b/docs/adr/0001-phase2-recipe-provider.md index ef721dc..1502283 100644 --- a/docs/adr/0001-phase2-recipe-provider.md +++ b/docs/adr/0001-phase2-recipe-provider.md @@ -1,6 +1,6 @@ # ADR-0001 Phase 2 fixed-function recipe provider qualification -- **Status:** Qualification core and explicit qualification-profile lifecycle composition implemented and verified; recipe request execution remains next +- **Status:** Qualification core, explicit qualification-profile lifecycle composition, and the internal recipe coordinator/publication path implemented and verified; application exposure remains separate - **Phase:** 2 - fixed-function image provider - **Parent:** [Capability-tiered agentic execution harness](0001-capability-tiered-agentic-execution-harness.md) - **Depends on:** [Phase 2 typed recipe contract](0001-phase2-recipe-contract.md) and [trusted artifact boundary](0001-phase2-artifact-boundary.md) @@ -113,8 +113,9 @@ limits, crop bounds, cancellation, stop behavior, and non-raiseable configuratio The explicit qualification-profile lifecycle composition now consumes the signed worker provenance, disposable [Windows sandbox qualification harness](0001-phase2-sandbox-qualification.md), native broker identity, watchdog, and accounting controls through an injected -health-gated builder. The next gate is the recipe-specific coordinator/request -and artifact-publication path. The provider must still launch out of process -under the bounded worker contract; it must never fall back to a host process or -in-process execution. External review and production signing remain optional -official-release hardening. +health-gated builder. The internal recipe coordinator now binds that worker +contract to owner-scoped staged artifacts and atomic publication. The next gate +is the explicit UI/API request surface and qualified worker-attempt factory; the +provider must still launch out of process under the bounded worker contract and +must never fall back to a host process or in-process execution. External review +and production signing remain optional official-release hardening. diff --git a/docs/adr/0001-phase2-sandbox-qualification.md b/docs/adr/0001-phase2-sandbox-qualification.md index 10aea5c..b8f4592 100644 --- a/docs/adr/0001-phase2-sandbox-qualification.md +++ b/docs/adr/0001-phase2-sandbox-qualification.md @@ -1,6 +1,6 @@ # ADR-0001 Phase 2 Windows recipe sandbox qualification -- **Status:** Qualification harness, packaged worker/release preflight, and explicit qualification-profile lifecycle composition implemented; recipe request execution remains next +- **Status:** Qualification harness, packaged worker/release preflight, explicit qualification-profile lifecycle composition, and internal recipe coordination implemented; application exposure remains separate - **Phase:** 2 - fixed-function image provider - **Parent:** [Capability-tiered agentic execution harness](0001-capability-tiered-agentic-execution-harness.md) - **Depends on:** [Phase 2 recipe provider core](0001-phase2-recipe-provider.md), [signed bundle installation](0001-phase2-bundle-installation.md), [native broker adapter](0001-phase2-native-broker.md), and [trusted artifact boundary](0001-phase2-artifact-boundary.md) @@ -51,7 +51,7 @@ fallback. | Decoder hostile corpus | Fixed one-pixel PNG, truncated PNG, and active SVG against the core | Qualification-only evidence; not OS-sandbox evidence | | Signed worker provenance | Storage-only `verify_active_worker()` role binding plus strict packaged-worker corpus | **Qualification complete** for the fixed worker role; production trust remains separate | | Broker identity and framed IPC | Native broker transport tests and strict packaged-worker corpus | **Qualification complete** when bound to the actual worker PID/token; no host fallback | -| Lifecycle enablement | `ExecutionLifecycle` remains disabled by default; `build_execution_lifecycle()` composes an explicit `release_profile="qualification"` only with an injected gate, coordinator, and provider-health probe | No provider can become reachable accidentally; the next core slice is the recipe coordinator/request path | +| Lifecycle enablement | `ExecutionLifecycle` remains disabled by default; `build_execution_lifecycle()` composes an explicit `release_profile="qualification"` only with an injected gate, coordinator, and provider-health probe | No provider can become reachable accidentally; the internal coordinator is now available and the next slice is explicit UI/API exposure | No single green smoke result closes the gate. A missing, failed, or unverified control produces `blocked` or `fail`, and no weaker host-process path is attempted. @@ -106,5 +106,7 @@ prevents accidental false-green qualification. The source checkout remains default-off; the explicit qualification profile is now a deliberate, injected lifecycle composition rather than an application default. Official production readiness, signing, and external review are separate optional maintainer concerns -and are not prerequisites for open-source development. The next implementation -slice is the recipe coordinator/request path behind this boundary. +and are not prerequisites for open-source development. The internal recipe +coordinator/request path is now implemented behind this boundary; the next +implementation slice is explicit UI/API exposure and qualified attempt-factory +wiring. diff --git a/tests/test_phase2_recipe_coordinator.py b/tests/test_phase2_recipe_coordinator.py new file mode 100644 index 0000000..28ee0b5 --- /dev/null +++ b/tests/test_phase2_recipe_coordinator.py @@ -0,0 +1,365 @@ +"""Adversarial tests for the qualified recipe coordinator and worker client.""" + +from __future__ import annotations + +from io import BytesIO +import hashlib +from pathlib import Path +from queue import Queue +from threading import Event, Thread +import time +from typing import Any + +import pytest +from PIL import Image + +from cortex_backend.execution import ( + ExecutionRepository, + RecipeExecutionCoordinator, + RecipeExecutionError, + RecipeImageRequest, + RecipeWorkerClient, + RecipeWorkerOutput, + parse_image_transform, +) +from cortex_backend.execution.lifecycle import RuntimeHealth +from cortex_backend.execution.recipe_provider import RecipeProviderError +from cortex_backend.execution.worker_runtime import RecipeWorkerBrokerRuntime + + +PRINCIPAL = "a" * 64 +OWNER = PRINCIPAL + + +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(artifact_id: str): + return parse_image_transform( + { + "schema_version": "artifact.transform.v1", + "input_artifact_id": artifact_id, + "steps": [{"op": "grayscale"}], + "output_format": "png", + } + ) + + +def _repository(tmp_path: Path) -> tuple[ExecutionRepository, str, str]: + repository = ExecutionRepository( + tmp_path / "execution.sqlite", + tmp_path / "artifacts", + max_artifact_bytes=2 * 1024 * 1024, + ) + source_job, _ = repository.create_job( + job_id="source-job", + owner=OWNER, + request_id="source-request", + profile="artifact.transform.v1", + payload={}, + ) + source = repository.publish_artifact( + source_job.job_id, + name="source.png", + content=_image_bytes(), + mime_type="image/png", + ) + return repository, source_job.job_id, source.artifact_id + + +class _FakeAttempt: + def __init__(self, output: RecipeWorkerOutput | None = None, error: str | None = None) -> None: + self.output = output + self.error = error + self.started = Event() + self.cancelled = Event() + self.closed = False + + def transform(self, _request_id: str, _job_id: str, _plan: Any, _content: bytes, cancel_event: Event) -> RecipeWorkerOutput: + self.started.set() + while not cancel_event.is_set() and not self.cancelled.is_set(): + if self.error is not None: + raise RecipeExecutionError(self.error) + break + if cancel_event.is_set() or self.cancelled.is_set(): + raise RecipeExecutionError("cancelled") + assert self.output is not None + return self.output + + def cancel(self, _reason: str = "user") -> None: + self.cancelled.set() + + def close(self) -> None: + self.closed = True + + +def _request(source_artifact_id: str, *, request_id: str = "recipe-request") -> RecipeImageRequest: + return RecipeImageRequest( + owner=OWNER, + request_id=request_id, + source_artifact_id=source_artifact_id, + plan=_plan(source_artifact_id), + ) + + +def _output() -> RecipeWorkerOutput: + content = _image_bytes() + return RecipeWorkerOutput( + content=content, + mime_type="image/png", + format="PNG", + width=4, + height=3, + sha256=hashlib.sha256(content).hexdigest(), + ) + + +def test_request_is_opaque_and_plan_must_bind_to_input_artifact(): + with pytest.raises(RecipeExecutionError) as mismatch: + RecipeImageRequest( + owner=OWNER, + request_id="recipe-request", + source_artifact_id="artifact-a", + plan=_plan("artifact-b"), + ) + assert mismatch.value.code == "input_artifact_mismatch" + with pytest.raises(ValueError): + _request(r"C:\Users\Admin\secret.png") + + +def test_worker_output_rechecks_digest_and_mime(): + content = _image_bytes() + with pytest.raises(RecipeExecutionError) as error: + RecipeWorkerOutput( + content=content, + mime_type="image/jpeg", + format="JPEG", + width=4, + height=3, + sha256=hashlib.sha256(content).hexdigest(), + ) + assert error.value.code == "worker_output_mime_mismatch" + + +def test_coordinator_publishes_owner_scoped_result_once(tmp_path: Path): + repository, _source_job_id, source_artifact_id = _repository(tmp_path) + attempt = _FakeAttempt(_output()) + coordinator = RecipeExecutionCoordinator(repository, lambda _job: attempt) + request = _request(source_artifact_id) + + accepted = coordinator.start_image_transform(request) + completed = coordinator.wait(accepted.job_id, timeout=3) + + assert completed.status == "succeeded" + assert completed.result is not None + assert "path" not in completed.result + result_artifact_id = completed.result["artifact_id"] + artifact = repository.get_artifact(result_artifact_id, owner=OWNER) + assert artifact is not None + assert repository.read_artifact(result_artifact_id) == _image_bytes() + assert attempt.closed + + duplicate = coordinator.start_image_transform(request) + assert duplicate.job_id == accepted.job_id + + +def test_coordinator_rejects_request_conflict_and_wrong_owner_artifact(tmp_path: Path): + repository, _source_job_id, source_artifact_id = _repository(tmp_path) + coordinator = RecipeExecutionCoordinator(repository, lambda _job: _FakeAttempt(_output())) + accepted = coordinator.start_image_transform(_request(source_artifact_id)) + conflicting = RecipeImageRequest( + owner=OWNER, + request_id="recipe-request", + source_artifact_id=source_artifact_id, + plan=parse_image_transform( + { + "schema_version": "artifact.transform.v1", + "input_artifact_id": source_artifact_id, + "steps": [{"op": "grayscale"}], + "output_format": "jpeg", + } + ), + ) + with pytest.raises(RecipeExecutionError) as conflict: + coordinator.start_image_transform(conflicting) + assert accepted.job_id + assert conflict.value.code == "request_conflict" + + with pytest.raises(RecipeExecutionError) as missing: + coordinator.start_image_transform(_request("not-an-artifact")) + assert missing.value.code == "input_artifact_unavailable" + + +def test_coordinator_failure_is_redacted_and_does_not_publish(tmp_path: Path): + repository, _source_job_id, source_artifact_id = _repository(tmp_path) + attempt = _FakeAttempt(error="provider_failed") + coordinator = RecipeExecutionCoordinator(repository, lambda _job: attempt) + accepted = coordinator.start_image_transform(_request(source_artifact_id)) + completed = coordinator.wait(accepted.job_id, timeout=3) + + assert completed.status == "failed" + assert completed.error == "provider_failed" + assert completed.result is None + assert list((repository.artifact_root / accepted.job_id).glob("*")) == [] + + +def test_coordinator_cancellation_is_terminal_and_cleans_staging(tmp_path: Path): + repository, _source_job_id, source_artifact_id = _repository(tmp_path) + attempt = _FakeAttempt(_output()) + coordinator = RecipeExecutionCoordinator(repository, lambda _job: attempt) + accepted = coordinator.start_image_transform(_request(source_artifact_id)) + assert attempt.started.wait(2) + coordinator.cancel(accepted.job_id, owner=OWNER) + completed = coordinator.wait(accepted.job_id, timeout=3) + + assert completed.status == "cancelled" + assert completed.error == "cancelled" + assert completed.result is None + assert attempt.closed + + +def test_recovery_rejects_tampered_payload_and_never_interprets_a_path(tmp_path: Path): + repository, _source_job_id, source_artifact_id = _repository(tmp_path) + job, _ = repository.create_job( + job_id="recovery-job", + owner=OWNER, + request_id="recovery-request", + profile="recipe.image.v1", + payload={ + "schema_version": "recipe.execution.v1", + "provider": "recipe-image-v1", + "source_artifact_id": source_artifact_id, + "plan": _plan(source_artifact_id).model_dump(mode="json"), + "plan_digest": _plan(source_artifact_id).digest(), + "retention_seconds": 86_400, + "path": r"C:\Windows\System32\cmd.exe", + }, + ) + repository.claim_lease(job.job_id, lease_owner="crashed-coordinator", ttl_seconds=0.01) + time.sleep(0.03) + coordinator = RecipeExecutionCoordinator(repository, lambda _job: _FakeAttempt(_output())) + recovered = coordinator.startup_recover() + + assert job.job_id in recovered + failed = repository.get_job(job.job_id) + assert failed is not None + assert failed.status == "failed" + assert failed.error == "recovery_invalid_payload" + + +class _Duplex: + def __init__(self) -> None: + self.to_worker: Queue[Any] = Queue() + self.to_client: Queue[Any] = Queue() + self.closed = False + + def endpoint(self, *, worker: bool): + parent = self + + class Endpoint: + def send_message(self, message: Any) -> None: + (parent.to_client if worker else parent.to_worker).put(message) + + def receive_message(self) -> Any: + item = (parent.to_worker if worker else parent.to_client).get() + if item is None: + raise RuntimeError("closed") + return item + + def close(self) -> None: + if not parent.closed: + parent.closed = True + parent.to_worker.put(None) + parent.to_client.put(None) + + return Endpoint() + + +def test_worker_client_round_trip_uses_only_typed_bytes_and_chunks(): + transport = _Duplex() + worker_endpoint = transport.endpoint(worker=True) + client_endpoint = transport.endpoint(worker=False) + runtime = RecipeWorkerBrokerRuntime( + worker_endpoint, + expected_principal_id=PRINCIPAL, + job_id="worker-job", + ) + runtime_thread = Thread(target=runtime.run, daemon=True) + runtime_thread.start() + client = RecipeWorkerClient( + client_endpoint, + installation_principal_id=PRINCIPAL, + timeout_seconds=5, + ) + output = client.transform("worker-request", "worker-job", _plan("artifact-1"), _image_bytes(), Event()) + client.close() + runtime_thread.join(timeout=2) + + assert output.mime_type == "image/png" + assert output.sha256 + assert transport.closed + + +class _CancellableProvider: + def __init__(self) -> None: + self.entered = Event() + + def start(self, _health: RuntimeHealth) -> RuntimeHealth: + return RuntimeHealth.ready("test") + + def stop(self) -> RuntimeHealth: + return RuntimeHealth.blocked("test_stopped", "stopped") + + def transform(self, _plan: Any, _content: bytes, *, cancel_check: Any) -> Any: + self.entered.set() + while not cancel_check(): + time.sleep(0.001) + raise RecipeProviderError("cancelled") + + +def test_worker_client_cancellation_reaches_running_worker(): + transport = _Duplex() + worker_endpoint = transport.endpoint(worker=True) + client_endpoint = transport.endpoint(worker=False) + provider = _CancellableProvider() + runtime_thread = Thread( + target=lambda: RecipeWorkerBrokerRuntime( + worker_endpoint, + expected_principal_id=PRINCIPAL, + job_id="worker-job", + provider_factory=lambda: provider, + ).run(), + daemon=True, + ) + runtime_thread.start() + client = RecipeWorkerClient( + client_endpoint, + installation_principal_id=PRINCIPAL, + timeout_seconds=5, + ) + cancel_event = Event() + result: list[Any] = [] + + def run() -> None: + try: + client.transform("worker-request", "worker-job", _plan("artifact-1"), _image_bytes(), cancel_event) + except Exception as error: + result.append(error) + + transform_thread = Thread(target=run, daemon=True) + transform_thread.start() + assert provider.entered.wait(2) + cancel_event.set() + transform_thread.join(timeout=3) + runtime_thread.join(timeout=2) + client.close() + + assert result and isinstance(result[0], RecipeExecutionError) + assert result[0].code == "cancelled"