diff --git a/.github/workflows/quality.yml b/.github/workflows/quality.yml index ed4e4ca..e26404c 100644 --- a/.github/workflows/quality.yml +++ b/.github/workflows/quality.yml @@ -38,6 +38,9 @@ jobs: - name: Run artifact security qualification run: python tools/execution_spikes/artifact_security_review.py --json --strict + - name: Run resource and watchdog qualification + run: python tools/execution_spikes/resource_watchdog_qualification.py --json --strict + - name: Compile application modules run: python -m compileall -q main.py backend diff --git a/backend/cortex_backend/execution/__init__.py b/backend/cortex_backend/execution/__init__.py index 321a6fd..5570543 100644 --- a/backend/cortex_backend/execution/__init__.py +++ b/backend/cortex_backend/execution/__init__.py @@ -117,6 +117,21 @@ LeaseConflict, ExecutionRepositoryError, ) +from .resource_accounting import ( + MAX_CONSOLE_BYTES, + MAX_COUNTER, + MAX_CPU_TIME_MS, + MAX_MEMORY_BYTES, + MAX_MESSAGES, + MAX_OBSERVATION_BYTES, + MAX_WALL_TIME_MS, + MonotonicWatchdog, + ResourceAccountingError, + ResourceBudget, + ResourceGovernor, + ResourceSample, + ResourceUsage, +) __all__ = [ "ArtifactLimitError", @@ -207,6 +222,19 @@ "RecipeWorkerBrokerRuntime", "WorkerRuntimeError", "WorkerRuntimeReport", + "MAX_CONSOLE_BYTES", + "MAX_COUNTER", + "MAX_CPU_TIME_MS", + "MAX_MEMORY_BYTES", + "MAX_MESSAGES", + "MAX_OBSERVATION_BYTES", + "MAX_WALL_TIME_MS", + "MonotonicWatchdog", + "ResourceAccountingError", + "ResourceBudget", + "ResourceGovernor", + "ResourceSample", + "ResourceUsage", "decode_frame", "decode_message", "encode_frame", diff --git a/backend/cortex_backend/execution/resource_accounting.py b/backend/cortex_backend/execution/resource_accounting.py new file mode 100644 index 0000000..a510373 --- /dev/null +++ b/backend/cortex_backend/execution/resource_accounting.py @@ -0,0 +1,376 @@ +"""Bounded execution budgets, monotonic watchdogs, and redacted accounting. + +The coordinator and worker use this module as a policy-only contract. It never +discovers paths, starts processes, or grants capabilities. Native Job Objects +remain responsible for OS enforcement; this layer makes the limits and their +failure precedence deterministic, validates monotonic samples, and prevents a +missing or regressing accounting source from becoming a green result. +""" + +from __future__ import annotations + +from dataclasses import dataclass, replace +import math +import re +from time import monotonic +from typing import Callable, Final + + +_SAFE_PROFILE = re.compile(r"^[a-z][a-z0-9._-]{0,63}$") +_SAFE_CODE = re.compile(r"^[a-z][a-z0-9_]{0,63}$") +MAX_WALL_TIME_MS: Final[int] = 600_000 +MAX_CPU_TIME_MS: Final[int] = 600_000 +MAX_MEMORY_BYTES: Final[int] = 1024 * 1024 * 1024 +MAX_MESSAGES: Final[int] = 16_384 +MAX_COUNTER: Final[int] = 1 << 40 +MAX_CONSOLE_BYTES: Final[int] = 1 * 1024 * 1024 +MAX_OBSERVATION_BYTES: Final[int] = 64 * 1024 + + +class ResourceAccountingError(ValueError): + """Stable resource/watchdog category with no host details.""" + + def __init__(self, code: str) -> None: + if _SAFE_CODE.fullmatch(code) is None: + raise ValueError("invalid resource accounting code") + self.code = code + super().__init__("The execution resource budget was exceeded safely.") + + +def _bounded_int(value: int, *, maximum: int, allow_zero: bool = False) -> int: + lower = 0 if allow_zero else 1 + if type(value) is not int or not lower <= value <= maximum: + raise ValueError("resource value is outside its hard ceiling") + return value + + +@dataclass(frozen=True, slots=True) +class ResourceBudget: + """Immutable logical budget; callers can only select values under hard caps.""" + + profile: str + wall_time_ms: int + cpu_time_ms: int + memory_bytes: int + max_messages: int + max_bytes_read: int + max_bytes_written: int + max_console_bytes: int = MAX_CONSOLE_BYTES + max_observation_bytes: int = MAX_OBSERVATION_BYTES + idle_timeout_ms: int | None = None + + def __post_init__(self) -> None: + if not isinstance(self.profile, str) or _SAFE_PROFILE.fullmatch(self.profile) is None: + raise ValueError("resource profile is invalid") + _bounded_int(self.wall_time_ms, maximum=MAX_WALL_TIME_MS) + _bounded_int(self.cpu_time_ms, maximum=MAX_CPU_TIME_MS) + _bounded_int(self.memory_bytes, maximum=MAX_MEMORY_BYTES) + _bounded_int(self.max_messages, maximum=MAX_MESSAGES) + _bounded_int(self.max_bytes_read, maximum=MAX_COUNTER, allow_zero=True) + _bounded_int(self.max_bytes_written, maximum=MAX_COUNTER, allow_zero=True) + _bounded_int(self.max_console_bytes, maximum=MAX_CONSOLE_BYTES, allow_zero=True) + _bounded_int(self.max_observation_bytes, maximum=MAX_OBSERVATION_BYTES, allow_zero=True) + if self.idle_timeout_ms is not None: + _bounded_int(self.idle_timeout_ms, maximum=MAX_WALL_TIME_MS) + if self.idle_timeout_ms > self.wall_time_ms: + raise ValueError("idle timeout cannot exceed wall budget") + + @classmethod + def scratch_auto_v1(cls) -> "ResourceBudget": + return cls( + profile="scratch.auto.v1", + wall_time_ms=10_000, + cpu_time_ms=5_000, + memory_bytes=256 * 1024 * 1024, + max_messages=256, + max_bytes_read=0, + max_bytes_written=0, + idle_timeout_ms=10_000, + ) + + @classmethod + def artifact_transform_v1(cls) -> "ResourceBudget": + return cls( + profile="artifact.transform.v1", + wall_time_ms=60_000, + cpu_time_ms=30_000, + memory_bytes=512 * 1024 * 1024, + max_messages=16_384, + max_bytes_read=100 * 1024 * 1024, + max_bytes_written=128 * 1024 * 1024, + idle_timeout_ms=60_000, + ) + + def with_watchdog( + self, + *, + wall_time_ms: int | None = None, + idle_timeout_ms: int | None = None, + max_messages: int | None = None, + ) -> "ResourceBudget": + """Return a bounded test/launch override without mutating this budget.""" + + return replace( + self, + wall_time_ms=self.wall_time_ms if wall_time_ms is None else wall_time_ms, + idle_timeout_ms=( + self.idle_timeout_ms + if idle_timeout_ms is None + else idle_timeout_ms + ), + max_messages=self.max_messages if max_messages is None else max_messages, + ) + + +@dataclass(frozen=True, slots=True) +class ResourceSample: + """One cumulative sample from the worker or native Job Object.""" + + cpu_time_ms: int = 0 + peak_memory_bytes: int | None = None + bytes_read: int = 0 + bytes_written: int = 0 + messages: int = 0 + console_bytes: int = 0 + observation_bytes: int = 0 + + def __post_init__(self) -> None: + _bounded_int(self.cpu_time_ms, maximum=MAX_COUNTER, allow_zero=True) + if self.peak_memory_bytes is not None: + _bounded_int(self.peak_memory_bytes, maximum=MAX_MEMORY_BYTES, allow_zero=True) + _bounded_int(self.bytes_read, maximum=MAX_COUNTER, allow_zero=True) + _bounded_int(self.bytes_written, maximum=MAX_COUNTER, allow_zero=True) + _bounded_int(self.messages, maximum=MAX_MESSAGES, allow_zero=True) + _bounded_int(self.console_bytes, maximum=MAX_COUNTER, allow_zero=True) + _bounded_int(self.observation_bytes, maximum=MAX_COUNTER, allow_zero=True) + + +@dataclass(frozen=True, slots=True) +class ResourceUsage: + """Safe cumulative usage returned to the coordinator/UI diagnostics.""" + + wall_time_ms: int = 0 + cpu_time_ms: int = 0 + peak_memory_bytes: int | None = None + bytes_read: int = 0 + bytes_written: int = 0 + messages: int = 0 + console_bytes: int = 0 + observation_bytes: int = 0 + + def __post_init__(self) -> None: + # A terminal sample can be observed after a deadline has already + # elapsed. Keep that value representable so the governor can return + # the stable ``deadline_exceeded`` category instead of leaking a + # validation exception. + _bounded_int(self.wall_time_ms, maximum=MAX_COUNTER, allow_zero=True) + ResourceSample( + cpu_time_ms=self.cpu_time_ms, + peak_memory_bytes=self.peak_memory_bytes, + bytes_read=self.bytes_read, + bytes_written=self.bytes_written, + messages=self.messages, + console_bytes=self.console_bytes, + observation_bytes=self.observation_bytes, + ) + + @property + def accounting_complete(self) -> bool: + """Memory accounting is required for a release-qualified terminal result.""" + + return self.peak_memory_bytes is not None + + +class MonotonicWatchdog: + """Wall/idle watchdog that rejects clock regressions and invalid samples.""" + + def __init__( + self, + *, + wall_time_ms: int, + idle_timeout_ms: int | None = None, + clock: Callable[[], float] = monotonic, + ) -> None: + _bounded_int(wall_time_ms, maximum=MAX_WALL_TIME_MS) + idle = wall_time_ms if idle_timeout_ms is None else idle_timeout_ms + _bounded_int(idle, maximum=MAX_WALL_TIME_MS) + if idle > wall_time_ms: + raise ValueError("idle timeout cannot exceed wall budget") + if not callable(clock): + raise TypeError("watchdog clock must be callable") + self._clock = clock + self._wall_time_ms = wall_time_ms + self._idle_timeout_ms = idle + now = self._read_clock() + self._started_at = now + self._last_progress_at = now + self._last_now = now + + def _read_clock(self) -> float: + try: + value = float(self._clock()) + except Exception: + raise ResourceAccountingError("watchdog_clock_invalid") from None + if not math.isfinite(value): + raise ResourceAccountingError("watchdog_clock_invalid") + if hasattr(self, "_last_now") and value < self._last_now: + raise ResourceAccountingError("watchdog_clock_invalid") + self._last_now = value + return value + + @staticmethod + def _elapsed(now: float, origin: float) -> float: + return max(0.0, (now - origin) * 1000.0) + + @classmethod + def _elapsed_ms(cls, now: float, origin: float) -> int: + return int(cls._elapsed(now, origin)) + + def elapsed_ms(self) -> int: + now = self._read_clock() + return self._elapsed_ms(now, self._started_at) + + def check(self) -> int: + now = self._read_clock() + elapsed_raw = self._elapsed(now, self._started_at) + idle_raw = self._elapsed(now, self._last_progress_at) + if elapsed_raw > self._wall_time_ms: + raise ResourceAccountingError("deadline_exceeded") + if idle_raw > self._idle_timeout_ms: + raise ResourceAccountingError("watchdog_stalled") + return int(elapsed_raw) + + def progress(self) -> int: + now = self._read_clock() + elapsed_raw = self._elapsed(now, self._started_at) + idle_raw = self._elapsed(now, self._last_progress_at) + if elapsed_raw > self._wall_time_ms: + raise ResourceAccountingError("deadline_exceeded") + if idle_raw > self._idle_timeout_ms: + raise ResourceAccountingError("watchdog_stalled") + self._last_progress_at = now + return int(elapsed_raw) + + +class ResourceGovernor: + """Enforce one immutable budget over monotonic cumulative samples.""" + + def __init__( + self, + budget: ResourceBudget, + *, + clock: Callable[[], float] = monotonic, + ) -> None: + if not isinstance(budget, ResourceBudget): + raise TypeError("resource budget is invalid") + self.budget = budget + self.watchdog = MonotonicWatchdog( + wall_time_ms=budget.wall_time_ms, + idle_timeout_ms=budget.idle_timeout_ms, + clock=clock, + ) + self._usage = ResourceUsage() + self._sample: ResourceSample | None = None + self._terminal_code: str | None = None + + @property + def usage(self) -> ResourceUsage: + return self._usage + + def _fail(self, code: str) -> None: + self._terminal_code = code + raise ResourceAccountingError(code) + + def _validate_monotonic(self, sample: ResourceSample) -> None: + previous = self._sample + if previous is None: + return + for field in ( + "cpu_time_ms", + "bytes_read", + "bytes_written", + "messages", + "console_bytes", + "observation_bytes", + ): + if getattr(sample, field) < getattr(previous, field): + self._fail("accounting_invalid") + if ( + previous.peak_memory_bytes is not None + and sample.peak_memory_bytes is None + ): + self._fail("accounting_invalid") + if ( + previous.peak_memory_bytes is not None + and sample.peak_memory_bytes is not None + and sample.peak_memory_bytes < previous.peak_memory_bytes + ): + self._fail("accounting_invalid") + + def _enforce(self, usage: ResourceUsage) -> None: + if usage.wall_time_ms > self.budget.wall_time_ms: + self._fail("deadline_exceeded") + if usage.cpu_time_ms > self.budget.cpu_time_ms: + self._fail("cpu_exhausted") + if ( + usage.peak_memory_bytes is not None + and usage.peak_memory_bytes > self.budget.memory_bytes + ): + self._fail("memory_exhausted") + if usage.bytes_read > self.budget.max_bytes_read: + self._fail("input_limit") + if usage.bytes_written > self.budget.max_bytes_written: + self._fail("output_limit") + if usage.console_bytes > self.budget.max_console_bytes: + self._fail("console_limit") + if usage.observation_bytes > self.budget.max_observation_bytes: + self._fail("observation_limit") + if usage.messages > self.budget.max_messages: + self._fail("message_budget_exhausted") + + def observe(self, sample: ResourceSample, *, progress: bool = False) -> ResourceUsage: + if self._terminal_code is not None: + raise ResourceAccountingError(self._terminal_code) + if not isinstance(sample, ResourceSample): + self._fail("accounting_invalid") + elapsed = self.watchdog.progress() if progress else self.watchdog.check() + self._validate_monotonic(sample) + usage = ResourceUsage( + wall_time_ms=elapsed, + cpu_time_ms=sample.cpu_time_ms, + peak_memory_bytes=sample.peak_memory_bytes, + bytes_read=sample.bytes_read, + bytes_written=sample.bytes_written, + messages=sample.messages, + console_bytes=sample.console_bytes, + observation_bytes=sample.observation_bytes, + ) + self._enforce(usage) + self._sample = sample + self._usage = usage + return usage + + def finish(self) -> ResourceUsage: + if self._terminal_code is not None: + raise ResourceAccountingError(self._terminal_code) + self.watchdog.check() + if not self._usage.accounting_complete: + self._fail("accounting_unavailable") + return self._usage + + +__all__ = [ + "MAX_CONSOLE_BYTES", + "MAX_COUNTER", + "MAX_CPU_TIME_MS", + "MAX_MEMORY_BYTES", + "MAX_MESSAGES", + "MAX_OBSERVATION_BYTES", + "MAX_WALL_TIME_MS", + "MonotonicWatchdog", + "ResourceAccountingError", + "ResourceBudget", + "ResourceGovernor", + "ResourceSample", + "ResourceUsage", +] diff --git a/backend/cortex_backend/execution/worker_protocol.py b/backend/cortex_backend/execution/worker_protocol.py index 5aa75cb..f6751b6 100644 --- a/backend/cortex_backend/execution/worker_protocol.py +++ b/backend/cortex_backend/execution/worker_protocol.py @@ -287,6 +287,8 @@ def __init__(self, provider: Any) -> None: self._input = bytearray() self._result: RecipeProviderResult | None = None self._cancelled = False + self._bytes_read = 0 + self._bytes_written = 0 self._lock = RLock() @property @@ -294,6 +296,16 @@ def state(self) -> Literal["idle", "receiving", "running", "complete", "cancelle with self._lock: return self._state # type: ignore[return-value] + @property + def bytes_read(self) -> int: + with self._lock: + return self._bytes_read + + @property + def bytes_written(self) -> int: + with self._lock: + return self._bytes_written + def prepare(self, message: WorkerPrepare) -> None: with self._lock: if self._state != "idle": @@ -302,6 +314,8 @@ def prepare(self, message: WorkerPrepare) -> None: self._input = bytearray() self._result = None self._cancelled = False + self._bytes_read = 0 + self._bytes_written = 0 self._state = "receiving" def input_chunk(self, message: WorkerInputChunk) -> None: @@ -316,6 +330,7 @@ def input_chunk(self, message: WorkerInputChunk) -> None: if len(self._input) + len(content) > self._prepare.input_size: raise WorkerProtocolError("input_size_exceeded") self._input.extend(content) + self._bytes_read += len(content) def cancel(self, message: WorkerCancel) -> None: with self._lock: @@ -396,7 +411,7 @@ def collect(self, message: WorkerCollect) -> WorkerOutputChunk: if not chunk: raise WorkerProtocolError("output_offset_invalid") encoded = base64.urlsafe_b64encode(chunk).decode("ascii").rstrip("=") - return WorkerOutputChunk( + chunk_model = WorkerOutputChunk( schema_version="recipe.worker.output_chunk.v1", request_id=message.request_id, job_id=message.job_id, @@ -405,6 +420,8 @@ def collect(self, message: WorkerCollect) -> WorkerOutputChunk: sha256=hashlib.sha256(chunk).hexdigest(), final=end == len(output), ) + self._bytes_written += len(chunk) + return chunk_model class RecipeWorkerDispatcher: @@ -432,6 +449,14 @@ def state(self) -> str: return self._session.state + @property + def bytes_read(self) -> int: + return self._session.bytes_read + + @property + def bytes_written(self) -> int: + return self._session.bytes_written + 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): diff --git a/backend/cortex_backend/execution/worker_runtime.py b/backend/cortex_backend/execution/worker_runtime.py index 094263b..ee1494d 100644 --- a/backend/cortex_backend/execution/worker_runtime.py +++ b/backend/cortex_backend/execution/worker_runtime.py @@ -13,12 +13,19 @@ from queue import Empty, Queue import re from threading import Event, Thread -from time import monotonic +from time import process_time_ns from typing import Any, Callable, Final, Literal, Protocol from .broker import BrokerMessage from .lifecycle import RuntimeHealth from .recipe_provider import RecipeImageProvider +from .resource_accounting import ( + ResourceAccountingError, + ResourceBudget, + ResourceGovernor, + ResourceSample, + ResourceUsage, +) from .worker_protocol import ( RecipeWorkerDispatcher, RecipeWorkerSession, @@ -83,6 +90,8 @@ class WorkerRuntimeReport: processed_messages: int terminal_state: Literal["complete", "cancelled", "failed", "message_budget"] + resource_usage: ResourceUsage | None = None + resource_error: str | None = None def _validate_principal(value: str) -> str: @@ -117,6 +126,7 @@ def __init__( provider_factory: Callable[[], WorkerProvider] = RecipeImageProvider, max_messages: int = DEFAULT_MAX_MESSAGES, watchdog_timeout_ms: int = DEFAULT_WATCHDOG_TIMEOUT_MS, + resource_sampler: Callable[[], ResourceSample] | None = None, ) -> None: if not all( callable(getattr(connection, method, None)) @@ -132,9 +142,17 @@ def __init__( 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") + if resource_sampler is not None and not callable(resource_sampler): + raise TypeError("worker resource sampler must be callable") self._provider_factory = provider_factory self._max_messages = max_messages self._watchdog_timeout_ms = watchdog_timeout_ms + self._resource_budget = ResourceBudget.artifact_transform_v1().with_watchdog( + wall_time_ms=watchdog_timeout_ms, + idle_timeout_ms=watchdog_timeout_ms, + max_messages=max_messages, + ) + self._resource_sampler = resource_sampler @staticmethod def _envelope_body(message: BrokerMessage) -> Any: @@ -254,10 +272,54 @@ def run(self) -> WorkerRuntimeReport: if not isinstance(health, RuntimeHealth) or not health.available: raise WorkerRuntimeError("provider_unavailable") dispatcher = RecipeWorkerDispatcher(RecipeWorkerSession(provider)) + resource_governor = ResourceGovernor(self._resource_budget) + process_started_ns = process_time_ns() + resource_error_code: str | None = None pending_collect: BrokerMessage | None = None cancel_acknowledged = False active_request: BrokerMessage | None = None - transform_started_at = 0.0 + + def resource_sample() -> ResourceSample: + if self._resource_sampler is not None: + try: + sample = self._resource_sampler() + except ResourceAccountingError: + raise + except Exception: + raise ResourceAccountingError("accounting_invalid") from None + if not isinstance(sample, ResourceSample): + raise ResourceAccountingError("accounting_invalid") + return sample + return ResourceSample( + cpu_time_ms=max(0, (process_time_ns() - process_started_ns) // 1_000_000), + bytes_read=dispatcher.bytes_read, + bytes_written=dispatcher.bytes_written, + messages=processed, + ) + + def observe_resources(*, progress: bool) -> ResourceUsage: + nonlocal resource_error_code + try: + return resource_governor.observe(resource_sample(), progress=progress) + except ResourceAccountingError as error: + resource_error_code = error.code + raise + + def report(state: Literal["complete", "cancelled", "failed", "message_budget"]) -> WorkerRuntimeReport: + nonlocal resource_error_code + if resource_error_code is None: + try: + observe_resources(progress=True) + except ResourceAccountingError: + pass + if resource_error_code is None and not resource_governor.usage.accounting_complete: + resource_error_code = "accounting_unavailable" + return WorkerRuntimeReport( + processed, + state, + resource_usage=resource_governor.usage, + resource_error=resource_error_code, + ) def start_receiver() -> None: nonlocal receiver @@ -271,45 +333,48 @@ def start_receiver() -> None: receiver.start() 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", - ), + try: + observe_resources(progress=active_request is None) + except ResourceAccountingError as error: + if ( + active_request is not None + and completion_thread is not None + and completion_thread.is_alive() + and not cancel_acknowledged + and error.code in {"deadline_exceeded", "watchdog_stalled"} + ): + 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", + }, ) - ) - terminal = "failed" - return WorkerRuntimeReport(processed, terminal) + 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 report(terminal) + raise WorkerRuntimeError(error.code) from None if receiver is None: try: incoming.put(self._connection.receive_message()) @@ -329,7 +394,7 @@ def start_receiver() -> None: if isinstance(error, WorkerProtocolError): if error.code == "cancelled" and cancel_acknowledged: terminal = "cancelled" - return WorkerRuntimeReport(processed, terminal) + return report(terminal) response = WorkerError( schema_version="recipe.worker.error.v1", request_id=completed_request.request_id, @@ -340,7 +405,7 @@ def start_receiver() -> None: self._response(completed_request, response) ) terminal = "cancelled" if error.code == "cancelled" else "failed" - return WorkerRuntimeReport(processed, terminal) + return report(terminal) if response is not None: self._connection.send_message( self._response(completed_request, response) @@ -364,10 +429,10 @@ def start_receiver() -> None: if isinstance(deferred_response, WorkerOutputChunk): if deferred_response.final: terminal = "complete" - return WorkerRuntimeReport(processed, terminal) + return report(terminal) if cancel_acknowledged: terminal = "cancelled" - return WorkerRuntimeReport(processed, terminal) + return report(terminal) continue if isinstance(request, Exception): raise WorkerRuntimeError("broker_receive_failed") from None @@ -406,7 +471,6 @@ def start_receiver() -> None: if request.operation == "input_complete": self._validate_envelope(request) active_request = request - transform_started_at = monotonic() start_receiver() completion_thread = Thread( target=self._dispatch_completion, @@ -435,16 +499,16 @@ def start_receiver() -> None: cancel_acknowledged = True if completion_thread is None: terminal = "cancelled" - return WorkerRuntimeReport(processed, terminal) + return report(terminal) elif isinstance(response, WorkerError): cancel_acknowledged = False if session_state == "failed": terminal = "failed" - return WorkerRuntimeReport(processed, terminal) + return report(terminal) if request.operation == "collect" and isinstance(response, WorkerOutputChunk): if response.final: terminal = "complete" - return WorkerRuntimeReport(processed, terminal) + return report(terminal) terminal = "message_budget" raise WorkerRuntimeError("message_budget_exceeded") diff --git a/docs/adr/0001-capability-tiered-agentic-execution-harness.md b/docs/adr/0001-capability-tiered-agentic-execution-harness.md index 91d52b4..3e1e192 100644 --- a/docs/adr/0001-capability-tiered-agentic-execution-harness.md +++ b/docs/adr/0001-capability-tiered-agentic-execution-harness.md @@ -957,7 +957,10 @@ pass without executing code. also qualifies owner/path binding, active-content rejection, exact claims, quarantine, rollback, and repository integrity without user/model input. Collect opt-in aggregate reliability metrics, never content, only after the remaining - release gates close and the provider is sandbox-qualified. + release gates close and the provider is sandbox-qualified. The deterministic + resource/watchdog corpus now qualifies immutable profile budgets, monotonic + watchdog categories, cumulative accounting rejection, actual Windows Job Object + accounting, and kill-on-close process-tree reaping. **Implementation gate:** parser, artifact-boundary, qualification-provider, worker-provenance, sandbox-qualification, and native-launcher-policy regression @@ -972,8 +975,9 @@ The 2026-07-28 packaged qualification evidence closes the hostile decoder corpus signed provenance, broker identity, and in-flight cancellation portions of this gate. The 2026-07-28 parser qualification also closes the deterministic parser- fuzzing portion. The 2026-07-28 artifact-security qualification closes the -deterministic copy-in/publication review. Resource/watchdog accounting, external -review, and production lifecycle health remain open. +deterministic copy-in/publication review. The 2026-07-29 resource/watchdog +qualification closes the deterministic accounting and watchdog-control portion; +external review and production lifecycle health remain open. ### Phase 3 — `scratch.auto.v1` arbitrary WebAssembly code diff --git a/docs/adr/0001-phase2-evidence.md b/docs/adr/0001-phase2-evidence.md index 5bcbf96..e2c9ade 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, deterministic parser-fuzz qualification, signed-manifest verification, native broker transport, authenticated worker loop, signed bundle installation, trusted artifact boundary, artifact security review, signed worker launch/broker qualification, and packaged transform/hostile-decoder/cancellation qualification complete; resource/watchdog accounting, external review, and lifecycle release gates remain open +- **Status:** Typed contract, deterministic parser-fuzz qualification, signed-manifest verification, native broker transport, authenticated worker loop, signed bundle installation, trusted artifact boundary, artifact security review, signed worker launch/broker qualification, packaged transform/hostile-decoder/cancellation qualification, and resource/watchdog accounting qualification complete; external review and lifecycle 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) @@ -25,10 +25,11 @@ | 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`. | | Deterministic artifact security review | **Complete (qualification-only)** | `artifact_security_review.py --json --strict` passed the fixed 12-case disposable corpus (`artifact-boundary-review.v1`, digest `a748cc9f0a514c8d`): owner/path binding, link/hardlink rejection, active/non-finite content, exact claims/quarantine, rollback, and repository size integrity. Missing link primitives remain blocked. | +| Deterministic resource/watchdog accounting | **Complete (qualification-only)** | `resource_watchdog_qualification.py --json --strict` passed the fixed `resource-watchdog.v1` corpus (digest `5eac03e2b4981543`): immutable ADR budgets, wall/idle watchdogs, clock and cumulative-sample regression, stable CPU/memory/input/output/console/observation/message limit precedence, missing-memory fail-closed behavior, actual Windows Job Object accounting, and kill-on-close process-tree reaping. Provider launch remains disabled pending external review and lifecycle wiring. | | 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 (signed launch/broker/hostile/cancellation)** | `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. The native broker uses bounded availability polling before reads so the cancellation reader remains live while the packaged provider transforms. | +| 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 | **Blocked / release gate** | Signed installation, provenance, AppContainer/job identity, broker handshake, `prepare`, `input_chunk`, `input_complete`, `collect_output`, hostile decoder rejection, in-flight cancellation acknowledgement, and artifact security review are qualified. The remaining gate is resource/watchdog accounting, external review, and production lifecycle wiring. | +| OS sandbox provider and provider-produced image outputs | **Blocked / release gate** | 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. External review and production lifecycle wiring remain required before provider enablement. | ## Security invariants @@ -87,8 +88,8 @@ generation with one exact `image_transform`/`recipe_worker.exe` role and stable byte identity can proceed to a future launcher; no executable is loaded here. 23. The disposable launcher applies all required Job Object policy before resume, - queries the configured limits/accounting, never grants breakaway, and reports - the absent worker/broker gates as blocking. + queries configured limits plus actual CPU, memory, process, and I/O accounting, + never grants breakaway, and reports the absent worker/broker gates as blocking. 24. Release signing reads an external raw private key only for the signing operation, self-verifies the canonical manifest, rejects reparse/hardlink/mutable package inputs, and never treats a generated manifest as launch authorization. @@ -108,6 +109,10 @@ it checks owner/path binding, active-content and numeric safety, link rejection, exact claims, quarantine, rollback, and stored-size integrity. The probe accepts no user/model input and reports an unavailable required link primitive as blocked. +30. Resource budgets are immutable and bounded; watchdog clocks must be finite and + monotonic; cumulative CPU, memory, byte, and message samples cannot regress; + limit precedence is stable; and a terminal result without required accounting + remains unavailable rather than being reported as a green qualification. ## Re-run target @@ -128,6 +133,7 @@ python tools/execution_spikes/recipe_sandbox_qualification.py --json --strict python tools/execution_spikes/recipe_worker_e2e_qualification.py --json --strict python tools/execution_spikes/recipe_parser_fuzz.py --json --strict python tools/execution_spikes/artifact_security_review.py --json --strict +python tools/execution_spikes/resource_watchdog_qualification.py --json --strict python -m compileall -q backend\cortex_backend\execution tests python -m pytest -q python tools/generate_contracts.py @@ -200,5 +206,34 @@ and quarantine, all-or-nothing rollback, and repository stored-size integrity. The review also fixed two fail-closed gaps: exponent-overflow JSON is rejected as non-finite, and oversized JSON integers no longer leak a `ValueError`; UTF-16 active markup is rejected instead of being classified as inert binary. The focused -artifact suite now passes 24 boundary tests plus the reproducibility test. Resource/ -watchdog accounting, external review, and production lifecycle health remain open. +artifact suite now passes 24 boundary tests plus the reproducibility test. At the +time of this artifact run, resource/watchdog accounting, external review, and +production lifecycle health remained open. + +**Resource/watchdog qualification result (2026-07-29):** The deterministic +`resource_watchdog_qualification.py --json --strict` probe passed all 15 cases in +the fixed `resource-watchdog.v1` corpus (digest `5eac03e2b4981543`). The corpus +locks the `scratch.auto.v1` and `artifact.transform.v1` budgets, wall and idle +watchdog categories, clock/sample regression rejection, stable CPU/memory/input/ +output/console/observation/message limit precedence, and missing-memory fail-closed +terminal behavior. +On Windows it additionally queried actual Job Object CPU, memory, process, +page-fault, and I/O accounting for a suspended fixed child and proved kill-on-close +reaping of its descendant tree. The initial native query exposed and fixed an ABI +gap (the Windows accounting struct requires both per-period user and kernel time +fields) and made the Win32 `HRESULT` binding explicit for supported CPython +versions. The worker loop now reports cumulative CPU/byte/message usage, and the +session exposes cumulative input/output byte counters. Resource/watchdog +qualification is complete for the fixed controls; real signed-worker enforcement, +external review, and production lifecycle health remain open. + +**Resource stage verification (2026-07-29):** The full Python suite passed 289 +tests with one expected Windows-platform skip and one pre-existing +`pytest-asyncio` deprecation warning. The resource/watchdog strict probe passed +15/15 cases, the artifact-security strict probe passed 12/12 cases, compileall +and `git diff --check` passed, generated API contracts were unchanged, and the +frontend lint, typecheck, 39 unit tests, and production build all passed. The +updated sandbox harness remained intentionally `qualification_status=blocked` +only because the immutable signed worker/broker production gates are still absent; +its AppContainer, cancellation, launcher-accounting, and resource controls were +green. diff --git a/docs/adr/0001-phase2-native-launcher.md b/docs/adr/0001-phase2-native-launcher.md index 6078088..4edf0c3 100644 --- a/docs/adr/0001-phase2-native-launcher.md +++ b/docs/adr/0001-phase2-native-launcher.md @@ -50,9 +50,12 @@ sequence with a fixed benign executable: 7. close all handles and remove the disposable profile/marker. The lower-level `appcontainer_smoke._run_child()` now exposes bounded optional -resource-limit parameters and returns the queried policy/accounting. Existing -filesystem/network and cancellation probes continue to use the same helper and -remain green after this change. +resource-limit parameters and returns the queried policy plus actual basic CPU, +memory, process, page-fault, and I/O accounting. The deterministic +`resource_watchdog_qualification.py` corpus also exercises monotonic logical +budgets and the kill-on-close process-tree watchdog. Existing filesystem/network +and cancellation probes continue to use the same helper and remain green after +this change. This is qualification evidence for control application, not approval to launch a recipe worker. The fixed probe reports `provider_launch_authorized=false` until @@ -63,18 +66,21 @@ or a weaker sandbox. ## Evidence and limits -On the controlled Windows host (2026-07-22), the resource-policy check passed with: +On the controlled Windows host (2026-07-29), the resource-policy check passed with: - AppContainer token confirmed; - active-process limit `1`; - process memory `64 MiB`, job memory `128 MiB`; - process CPU `2 s`, job CPU `4 s`; +- actual Job Object accounting fields present with non-negative CPU, memory, + process, page-fault, and I/O values; - kill-on-close present; and - breakaway flags absent. -The policy values are queried before the Job Object handle closes. This does not -yet prove enforcement against a real signed recipe worker, resource exhaustion -behavior across supported Windows versions, or external launcher review. +The policy and accounting values are queried before the Job Object handle closes. +The fixed probe and watchdog corpus do not yet prove enforcement against a real +signed recipe worker, resource-exhaustion behavior across supported Windows +versions, or external launcher review. ## Remaining blockers @@ -83,11 +89,12 @@ 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 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. +- The packaged worker loop is now broker-only and fail-closed; the disposable + signed-worker qualification covers authenticated input/output, hostile decoder, + and cancellation cases, but the immutable production generation is not shipped + at the launcher path. +- Real signed-worker resource enforcement, external security review, and lifecycle + health-gated wiring remain separate release gates. If any control is missing or cannot be verified, the provider remains unavailable; there is no host-process or weaker-sandbox fallback. @@ -103,5 +110,6 @@ AppContainer child on Windows, verifies its token, applies/query-verifies Job Ob policy, and closes it without resuming. `tests/test_native_launcher_qualification.py` covers the disposable probe's non-Windows blocking, report fail-closed behavior, and no-breakaway invariant. The -full repository suite and the existing AppContainer/cancellation corpus are -required before this boundary can be merged. +`resource_watchdog_qualification.py --json --strict` corpus additionally requires +actual accounting and full-tree reaping. The full repository suite and the existing +AppContainer/cancellation corpus are required before this boundary can be merged. diff --git a/docs/adr/0001-phase2-sandbox-qualification.md b/docs/adr/0001-phase2-sandbox-qualification.md index f66f369..14a7c98 100644 --- a/docs/adr/0001-phase2-sandbox-qualification.md +++ b/docs/adr/0001-phase2-sandbox-qualification.md @@ -20,8 +20,9 @@ The harness is deliberately fail-closed and has independent checks for: qualification-only Pillow core, while recording `sandboxed=false`; and 4. it requires the future fixed recipe worker package at the repository's fixed packaging location; -5. it reports per-worker CPU/memory/breakaway/accounting controls as blocked until - the native launcher applies and verifies them; and +5. it runs the deterministic resource/watchdog corpus, requiring immutable + budgets, cumulative accounting, actual Job Object CPU/memory/process/I/O + accounting, and kill-on-close tree reaping; 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 packaged worker completes the authenticated client handshake and hostile corpus. @@ -42,8 +43,9 @@ fallback. | Control | Evidence in this stage | Release interpretation | | --- | --- | --- | | AppContainer token and zero-capability denials | `appcontainer_smoke.py`, child report `recipe_appcontainer_control` | Required prerequisite; does not prove LPAC policy or provider launch identity | -| Job Object kill-on-close and tree cancellation | `cancellation_corpus.py`, child report `recipe_cancellation_control` | Required prerequisite; resource limits and accounting remain open | +| Job Object kill-on-close and tree cancellation | `cancellation_corpus.py`, child report `recipe_cancellation_control` | Required prerequisite; the watchdog corpus proves full-tree reaping | | Suspended launch/resource policy | `native_launcher_qualification.py`, child report `recipe_native_launcher_policy` | Policy application/query passes for a fixed benign child; real worker enforcement/review remains open | +| Resource/watchdog accounting | `resource_watchdog_qualification.py`, child report `recipe_resource_controls` | Immutable budgets, actual Job Object accounting, and kill-on-close reaping qualify; signed-worker enforcement and external review remain open | | 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 fixed package precondition | **Storage gate complete; launch remains blocked** until a packaged executable and native launcher exist | | Broker identity and framed IPC | Native broker transport tests and ADR | Must be bound to the actual worker PID/token before launch | @@ -91,7 +93,7 @@ that a blocked worker gate never authorizes provider launch. ## Consequences This stage provides reproducible evidence for the controls that already exist and -prevents accidental false-green qualification. It does not claim decoder isolation, -LPAC capability policy, resource enforcement, signed runtime provenance, native -worker identity, or production readiness. Those remain explicit blockers before -the provider can be wired to lifecycle or exposed to any model/UI route. +prevents accidental false-green qualification. It does not claim signed-worker +resource enforcement, signed runtime provenance, native worker identity, or +production readiness. Those remain explicit blockers before the provider can be +wired to lifecycle or exposed to any model/UI route. diff --git a/docs/adr/0001-phase2-worker-protocol.md b/docs/adr/0001-phase2-worker-protocol.md index e4fa407..6732952 100644 --- a/docs/adr/0001-phase2-worker-protocol.md +++ b/docs/adr/0001-phase2-worker-protocol.md @@ -69,9 +69,17 @@ root. `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. `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. +(10 tests) covers authenticated envelope binding, redacted repairable failures, +provider failure, message budgets, terminal cleanup, watchdog expiry, cumulative +resource reporting, fixed-entrypoint parsing, and cancellation delivered while a +transform is running. `RecipeWorkerSession` exposes cumulative input/output byte +counters for the coordinator's accounting sample. The deterministic +`resource_watchdog_qualification.py --json --strict` corpus covers immutable +budgets, monotonic watchdog categories, sample regression, stable limit +precedence, missing-memory fail-closed behavior, actual Job Object accounting, +and kill-on-close tree reaping. A terminal runtime report marks +`resource_error=accounting_unavailable` when no peak-memory sampler is supplied; +it cannot be mistaken for release-qualified accounting. 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. @@ -82,5 +90,6 @@ This ADR does not authorize provider execution. The remaining stage must install 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. +packaged process with the qualified resource/watchdog and artifact-boundary +controls. Lifecycle/UI enablement remains behind those gates and external security +review. diff --git a/tests/test_native_launcher_qualification.py b/tests/test_native_launcher_qualification.py index 59413f3..ea4fc62 100644 --- a/tests/test_native_launcher_qualification.py +++ b/tests/test_native_launcher_qualification.py @@ -16,6 +16,10 @@ def test_non_windows_resource_probe_fails_closed(monkeypatch): assert result["status"] == "blocked" +def test_native_api_hresult_type_is_available_across_supported_python_versions(): + assert native._HRESULT is not None + + def test_launcher_report_never_authorizes_when_worker_or_broker_is_blocked(monkeypatch): monkeypatch.setattr( qualification, diff --git a/tests/test_phase2_resource_accounting.py b/tests/test_phase2_resource_accounting.py new file mode 100644 index 0000000..8342917 --- /dev/null +++ b/tests/test_phase2_resource_accounting.py @@ -0,0 +1,120 @@ +"""Deterministic logical resource/watchdog accounting tests.""" + +from __future__ import annotations + +import math + +import pytest + +from cortex_backend.execution.resource_accounting import ( + ResourceAccountingError, + ResourceBudget, + ResourceGovernor, + ResourceSample, +) + + +def _clock(values: list[float]): + iterator = iter(values) + return lambda: next(iterator) + + +def _budget(**overrides): + values = { + "profile": "test.v1", + "wall_time_ms": 100, + "cpu_time_ms": 100, + "memory_bytes": 1024, + "max_messages": 8, + "max_bytes_read": 8, + "max_bytes_written": 8, + "idle_timeout_ms": 100, + } + values.update(overrides) + return ResourceBudget(**values) + + +def test_adr_profiles_are_bounded_and_immutable(): + scratch = ResourceBudget.scratch_auto_v1() + artifact = ResourceBudget.artifact_transform_v1() + + assert scratch.wall_time_ms == 10_000 + assert scratch.cpu_time_ms == 5_000 + assert scratch.memory_bytes == 256 * 1024 * 1024 + assert artifact.wall_time_ms == 60_000 + assert artifact.cpu_time_ms == 30_000 + assert artifact.memory_bytes == 512 * 1024 * 1024 + assert artifact.with_watchdog(wall_time_ms=60_000) == artifact + with pytest.raises(ValueError): + artifact.with_watchdog(wall_time_ms=600_001) + + +@pytest.mark.parametrize( + ("values", "expected"), + [ + ([0.0, 0.100001], "deadline_exceeded"), + ([0.0, 0.010001], "watchdog_stalled"), + ([1.0, 0.5], "watchdog_clock_invalid"), + ], +) +def test_watchdog_categories_are_deterministic(values, expected): + idle = 100 if expected == "deadline_exceeded" else 10 + governor = ResourceGovernor(_budget(idle_timeout_ms=idle), clock=_clock(values)) + with pytest.raises(ResourceAccountingError) as error: + governor.observe(ResourceSample(peak_memory_bytes=1)) + assert error.value.code == expected + + +@pytest.mark.parametrize( + ("sample", "expected"), + [ + (ResourceSample(cpu_time_ms=101, peak_memory_bytes=1), "cpu_exhausted"), + (ResourceSample(peak_memory_bytes=1025), "memory_exhausted"), + (ResourceSample(bytes_read=9, peak_memory_bytes=1), "input_limit"), + (ResourceSample(bytes_written=9, peak_memory_bytes=1), "output_limit"), + (ResourceSample(console_bytes=1_048_577, peak_memory_bytes=1), "console_limit"), + ( + ResourceSample(observation_bytes=65_537, peak_memory_bytes=1), + "observation_limit", + ), + (ResourceSample(messages=9, peak_memory_bytes=1), "message_budget_exhausted"), + ], +) +def test_governor_enforces_stable_limit_precedence(sample, expected): + governor = ResourceGovernor(_budget(), clock=lambda: 0.0) + with pytest.raises(ResourceAccountingError) as error: + governor.observe(sample) + assert error.value.code == expected + + +def test_governor_rejects_cumulative_regression_and_missing_terminal_memory(): + governor = ResourceGovernor(_budget(), clock=lambda: 0.0) + governor.observe(ResourceSample(peak_memory_bytes=4, messages=1)) + with pytest.raises(ResourceAccountingError) as regression: + governor.observe(ResourceSample(peak_memory_bytes=None, messages=1)) + assert regression.value.code == "accounting_invalid" + + incomplete = ResourceGovernor(_budget(), clock=lambda: 0.0) + incomplete.observe(ResourceSample()) + with pytest.raises(ResourceAccountingError) as missing: + incomplete.finish() + assert missing.value.code == "accounting_unavailable" + + +def test_governor_returns_redacted_cumulative_usage(): + governor = ResourceGovernor(_budget(), clock=lambda: 0.0) + usage = governor.observe( + ResourceSample( + cpu_time_ms=4, + peak_memory_bytes=128, + bytes_read=3, + bytes_written=2, + messages=1, + ) + ) + assert usage.accounting_complete + assert usage.cpu_time_ms == 4 + assert usage.peak_memory_bytes == 128 + assert governor.finish() == usage + assert not hasattr(usage, "path") + assert math.isfinite(usage.wall_time_ms) diff --git a/tests/test_phase2_resource_watchdog_qualification.py b/tests/test_phase2_resource_watchdog_qualification.py new file mode 100644 index 0000000..8d8d58d --- /dev/null +++ b/tests/test_phase2_resource_watchdog_qualification.py @@ -0,0 +1,39 @@ +"""Regression tests for the redacted resource/watchdog qualification report.""" + +from __future__ import annotations + +import tools.execution_spikes.resource_watchdog_qualification as qualification + + +def test_resource_watchdog_report_is_reproducible_without_native_details(monkeypatch): + green = lambda: {"name": "native", "status": "pass"} + monkeypatch.setattr(qualification, "_native_job_accounting", green) + monkeypatch.setattr(qualification, "_native_tree_reaping", green) + + first = qualification.run_qualification() + second = qualification.run_qualification() + + assert first == second + assert first["status"] == "pass" + assert first["qualification_status"] == "pass" + assert first["corpus"] == "resource-watchdog.v1" + assert len(first["checks"]) == 15 + assert all("path" not in str(check) for check in first["checks"]) + + +def test_resource_watchdog_report_never_authorizes_provider(monkeypatch): + monkeypatch.setattr( + qualification, + "_native_job_accounting", + lambda: {"name": "native", "status": "blocked"}, + ) + monkeypatch.setattr( + qualification, + "_native_tree_reaping", + lambda: {"name": "native", "status": "blocked"}, + ) + + report = qualification.run_qualification() + + assert report["provider_launch_authorized"] is False + assert report["status"] == "blocked" diff --git a/tests/test_phase2_worker_protocol.py b/tests/test_phase2_worker_protocol.py index f08bec7..a3436e5 100644 --- a/tests/test_phase2_worker_protocol.py +++ b/tests/test_phase2_worker_protocol.py @@ -83,6 +83,7 @@ def test_worker_session_requires_in_order_hashed_input_and_collects_output(): session.prepare(prepare) session.input_chunk(_chunk(content[:8])) session.input_chunk(_chunk(content[8:], offset=8)) + assert session.bytes_read == len(content) result = session.complete_input( WorkerInputComplete( @@ -107,6 +108,7 @@ def test_worker_session_requires_in_order_hashed_input_and_collects_output(): ) assert output.decoded() assert output.final + assert session.bytes_written == len(output.decoded()) def test_dispatcher_rejects_unknown_or_malformed_operations(): diff --git a/tests/test_phase2_worker_runtime.py b/tests/test_phase2_worker_runtime.py index d109c36..93c471d 100644 --- a/tests/test_phase2_worker_runtime.py +++ b/tests/test_phase2_worker_runtime.py @@ -19,6 +19,7 @@ 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.resource_accounting import ResourceSample from cortex_backend.execution.worker_protocol import ( WorkerCancel, WorkerCollect, @@ -156,6 +157,11 @@ def provider_factory() -> RecipeImageProvider: assert report.terminal_state == "complete" assert report.processed_messages == 4 + assert report.resource_usage is not None + assert report.resource_usage.bytes_read == len(_image_bytes()) + assert report.resource_usage.bytes_written > 0 + assert report.resource_usage.messages == 4 + assert report.resource_error == "accounting_unavailable" assert connection.closed assert providers[0].health_snapshot.code == "recipe_provider_stopped" assert [message.operation for message in connection.sent] == [ @@ -199,6 +205,30 @@ def test_runtime_redacts_malformed_body_and_allows_bounded_repair(): assert "path" not in str(connection.sent[0].body) +def test_runtime_accepts_cumulative_native_accounting_sample(): + content = _image_bytes() + connection = _FakeConnection(_successful_messages(content)) + + report = RecipeWorkerBrokerRuntime( + connection, + expected_principal_id=PRINCIPAL, + job_id=JOB_ID, + resource_sampler=lambda: ResourceSample( + cpu_time_ms=1, + peak_memory_bytes=1024, + bytes_read=len(content), + bytes_written=1, + messages=4, + ), + ).run() + + assert report.terminal_state == "complete" + assert report.resource_error is None + assert report.resource_usage is not None + assert report.resource_usage.accounting_complete + assert report.resource_usage.peak_memory_bytes == 1024 + + def test_runtime_cancellation_is_terminal_and_acknowledged(): content = _image_bytes() prepare = _prepare(content) @@ -358,6 +388,7 @@ def test_runtime_watchdog_captures_stalled_transform_and_closes(): ).run() assert report.terminal_state == "failed" + assert report.resource_error in {"deadline_exceeded", "watchdog_stalled"} assert connection.sent[-1].body["schema_version"] == "recipe.worker.error.v1" assert connection.sent[-1].body["code"] == "timeout" assert connection.closed diff --git a/tools/execution_spikes/README.md b/tools/execution_spikes/README.md index 4575ba7..9e891d8 100644 --- a/tools/execution_spikes/README.md +++ b/tools/execution_spikes/README.md @@ -41,7 +41,7 @@ The expected result at this stage is `qualification_status=blocked`: the native AppContainer and Job Object controls may pass, but the signed `recipe_worker.exe` bundle is not shipped yet. A blocked worker-provenance check is intentional and must remain blocking until trust-root verification, native worker identity, and -resource/watchdog enforcement are implemented. +real-worker lifecycle enforcement are implemented. The native launcher qualification prints a passing resource-policy subcheck when the fixed suspended child receives and reports Job Object CPU/memory/active-process @@ -89,9 +89,30 @@ handshake, normal `collect_output`, hostile decoder rejection for truncated PNG and active SVG bytes, and an in-flight `cancel_ack` after `input_complete`. The native read path polls `PeekNamedPipe` with a bounded 5 ms wait before `ReadFile`, keeping the worker's cancellation reader live without starving the provider -transform. Resource/watchdog accounting, external review, and production-lifecycle -release gates remain open; parser-fuzz and artifact-security evidence are qualified -below. +transform. Resource/watchdog accounting is qualified by the fixed corpus below; +external review and production-lifecycle release gates remain open. Parser-fuzz +and artifact-security evidence are qualified below. + +## Run the resource/watchdog accounting qualification + +This deterministic corpus exercises the immutable ADR budgets and the worker's +monotonic accounting contract without model or user input. On Windows it also +runs the fixed Job Object policy probe and kill-on-close descendant corpus; the +report records only stable status/evidence categories and never authorizes a +provider launch: + +```powershell +python tools/execution_spikes/resource_watchdog_qualification.py --json +python tools/execution_spikes/resource_watchdog_qualification.py --json --strict +``` + +The current `resource-watchdog.v1` evidence passes all 15 cases (digest +`5eac03e2b4981543`): budget matrix, wall/idle watchdogs, clock and sample +regression, stable CPU/memory/input/output/console/observation/message limits, +missing-memory fail-closed behavior, actual Job Object accounting, and full-tree +reaping. This is +qualification-only evidence; signed-worker enforcement, external review, and +production lifecycle wiring remain blocked. ## Run the typed parser fuzz qualification @@ -158,8 +179,11 @@ unavailable required link primitive; a blocked result is not a release pass. controls with the qualification-only decoder corpus and a mandatory signed worker provenance gate; it never authorizes a host-process fallback. - `native_launcher_qualification`: creates only a fixed suspended `findstr.exe` - child, applies and queries Job Object resource policy before resume, and reports + child, applies and queries Job Object resource policy/accounting before resume, and reports the signed-worker and broker-binding blockers without launching either. +- `resource_watchdog_qualification`: runs the fixed logical budget/watchdog corpus, + requires actual native Job Object accounting and kill-on-close tree reaping on + Windows, and never enables a provider. - `recipe_worker_e2e_qualification`: signs and installs a disposable worker generation, proves the live native broker identity boundary, and runs the fixed packaged-worker protocol corpus; any provider or cleanup timeout remains @@ -182,8 +206,7 @@ unavailable required link primitive; a blocked result is not a release pass. An API export check is not proof of AppContainer process isolation. The native helper is evidence for this disposable smoke corpus only; it is not authorized to launch guest runtimes or model-generated code. LPAC policy qualification, -resource limits, cancellation corpus, and the security review remain separate -Phase 0 gates. +real-worker enforcement, and the security review remain separate release gates. ## Safety rules diff --git a/tools/execution_spikes/appcontainer_smoke.py b/tools/execution_spikes/appcontainer_smoke.py index fd1fbd8..5a0027a 100644 --- a/tools/execution_spikes/appcontainer_smoke.py +++ b/tools/execution_spikes/appcontainer_smoke.py @@ -43,9 +43,11 @@ _WAIT_OBJECT_0 = 0 _WAIT_TIMEOUT = 0x102 _INFINITE = 0xFFFFFFFF +_JOB_OBJECT_BASIC_ACCOUNTING_INFORMATION = 1 _JOB_OBJECT_BASIC_PROCESS_ID_LIST = 3 _PROCESS_QUERY_LIMITED_INFORMATION = 0x1000 _SYNCHRONIZE = 0x00100000 +_HRESULT = getattr(wintypes, "HRESULT", ctypes.c_long) class _LargeInteger(ctypes.Union): @@ -63,6 +65,19 @@ class _IoCounters(ctypes.Structure): ] +class _BasicAccountingInformation(ctypes.Structure): + _fields_ = [ + ("total_user_time", _LargeInteger), + ("total_kernel_time", _LargeInteger), + ("this_period_total_user_time", _LargeInteger), + ("this_period_total_kernel_time", _LargeInteger), + ("total_page_fault_count", wintypes.DWORD), + ("total_processes", wintypes.DWORD), + ("active_processes", wintypes.DWORD), + ("total_terminated_processes", wintypes.DWORD), + ] + + class _BasicLimitInformation(ctypes.Structure): _fields_ = [ ("per_process_user_time", _LargeInteger), @@ -160,9 +175,10 @@ def _configure_apis() -> tuple[Any, Any, Any, Any, Any, Any, Any]: wintypes.DWORD, ctypes.POINTER(wintypes.LPVOID), ] - userenv.CreateAppContainerProfile.restype = wintypes.HRESULT + # HRESULT is not exposed by every supported CPython wintypes module. + userenv.CreateAppContainerProfile.restype = _HRESULT userenv.DeleteAppContainerProfile.argtypes = [wintypes.LPCWSTR] - userenv.DeleteAppContainerProfile.restype = wintypes.HRESULT + userenv.DeleteAppContainerProfile.restype = _HRESULT advapi32.FreeSid.argtypes = [wintypes.LPVOID] advapi32.FreeSid.restype = wintypes.LPVOID advapi32.OpenProcessToken.argtypes = [ @@ -481,6 +497,15 @@ def _job_extended_info(kernel32: Any, job: wintypes.HANDLE) -> dict[str, int]: ctypes.byref(returned), ): raise ctypes.WinError(ctypes.get_last_error()) + accounting = _BasicAccountingInformation() + if not query( + job, + _JOB_OBJECT_BASIC_ACCOUNTING_INFORMATION, + ctypes.byref(accounting), + ctypes.sizeof(accounting), + ctypes.byref(returned), + ): + raise ctypes.WinError(ctypes.get_last_error()) return { "limit_flags": int(info.basic_limit_information.limit_flags), "active_process_limit": int(info.basic_limit_information.active_process_limit), @@ -494,6 +519,24 @@ def _job_extended_info(kernel32: Any, job: wintypes.HANDLE) -> dict[str, int]: ), "peak_process_memory_used": int(info.peak_process_memory_used), "peak_job_memory_used": int(info.peak_job_memory_used), + "total_user_time_100ns": int(accounting.total_user_time.quad_part), + "total_kernel_time_100ns": int(accounting.total_kernel_time.quad_part), + "this_period_user_time_100ns": int( + accounting.this_period_total_user_time.quad_part + ), + "this_period_kernel_time_100ns": int( + accounting.this_period_total_kernel_time.quad_part + ), + "total_page_fault_count": int(accounting.total_page_fault_count), + "total_processes": int(accounting.total_processes), + "active_processes": int(accounting.active_processes), + "total_terminated_processes": int(accounting.total_terminated_processes), + "read_operations": int(info.io_info.read_operations), + "write_operations": int(info.io_info.write_operations), + "other_operations": int(info.io_info.other_operations), + "read_transfers": int(info.io_info.read_transfers), + "write_transfers": int(info.io_info.write_transfers), + "other_transfers": int(info.io_info.other_transfers), } diff --git a/tools/execution_spikes/native_launcher_qualification.py b/tools/execution_spikes/native_launcher_qualification.py index c581970..eb4fbef 100644 --- a/tools/execution_spikes/native_launcher_qualification.py +++ b/tools/execution_spikes/native_launcher_qualification.py @@ -116,6 +116,28 @@ def _probe_resource_policy() -> dict[str, Any]: | _JOB_OBJECT_LIMIT_PROCESS_TIME | _JOB_OBJECT_LIMIT_JOB_TIME ) + accounting_fields = ( + "total_user_time_100ns", + "total_kernel_time_100ns", + "this_period_user_time_100ns", + "this_period_kernel_time_100ns", + "total_page_fault_count", + "total_processes", + "active_processes", + "total_terminated_processes", + "read_operations", + "write_operations", + "other_operations", + "read_transfers", + "write_transfers", + "other_transfers", + ) + accounting_present = all( + field in info and type(info[field]) is int for field in accounting_fields + ) + total_cpu_100ns = int(info.get("total_user_time_100ns", -1)) + int( + info.get("total_kernel_time_100ns", -1) + ) checks = { "appcontainer_token": details.get("is_appcontainer") is True, "suspended_policy_applied_before_resume": True, @@ -129,6 +151,13 @@ def _probe_resource_policy() -> dict[str, Any]: "breakaway_not_granted": flags & (_JOB_OBJECT_LIMIT_BREAKAWAY_OK | _JOB_OBJECT_LIMIT_SILENT_BREAKAWAY_OK) == 0, + "accounting_fields_present": accounting_present, + "accounting_values_nonnegative": accounting_present + and all(info[field] >= 0 for field in accounting_fields), + "accounting_process_observed": info.get("total_processes", 0) >= 1, + "accounting_memory_within_limit": info.get("peak_job_memory_used", -1) + <= 128 * 1024 * 1024, + "accounting_cpu_within_limit": 0 <= total_cpu_100ns <= 40_000_000, } status = PASS if all(checks.values()) else FAIL return _result( diff --git a/tools/execution_spikes/recipe_sandbox_qualification.py b/tools/execution_spikes/recipe_sandbox_qualification.py index 9612e6c..42b2709 100644 --- a/tools/execution_spikes/recipe_sandbox_qualification.py +++ b/tools/execution_spikes/recipe_sandbox_qualification.py @@ -289,7 +289,7 @@ def _probe_signed_worker_precondition() -> dict[str, Any]: def _probe_future_worker_controls() -> list[dict[str, Any]]: - """Run the fixed launcher policy probe and expose remaining blockers.""" + """Run fixed launcher/resource probes and expose remaining blockers.""" helper = ROOT / "tools" / "execution_spikes" / "native_launcher_qualification.py" launcher = _run_fixed_helper( @@ -298,15 +298,17 @@ def _probe_future_worker_controls() -> list[dict[str, Any]]: 30, ) launcher["name"] = "recipe_native_launcher_policy" + resource_helper = ROOT / "tools" / "execution_spikes" / "resource_watchdog_qualification.py" + resource = _run_fixed_helper( + resource_helper, + "cortex-resource-watchdog-qualification", + 30, + ) + resource["name"] = "recipe_resource_controls" return [ launcher, - _result( - "recipe_resource_controls", - BLOCKED, - "The fixed resource-policy spike reports its configured/queryable limits, but release still requires a real worker launch and external enforcement review.", - launch_refused=True, - ), + resource, _result( "recipe_broker_identity", BLOCKED, diff --git a/tools/execution_spikes/resource_watchdog_qualification.py b/tools/execution_spikes/resource_watchdog_qualification.py new file mode 100644 index 0000000..6efc1ce --- /dev/null +++ b/tools/execution_spikes/resource_watchdog_qualification.py @@ -0,0 +1,392 @@ +"""Deterministic resource/watchdog accounting qualification corpus. + +This helper is a release-gate probe, not an execution fallback. The pure cases +exercise the immutable Phase 2 budgets, monotonic watchdog, cumulative sample +validation, failure precedence, and missing-accounting behavior without running +model or user input. On Windows it also invokes the existing fixed native +Job Object policy and kill-on-close process-tree corpus, but records only their +stable status/evidence categories. It never authorizes a provider launch. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +from pathlib import Path +import sys +from typing import Any, Callable + +ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(ROOT / "backend")) +sys.path.insert(0, str(ROOT / "tools" / "execution_spikes")) + +from cortex_backend.execution.resource_accounting import ( + ResourceAccountingError, + ResourceBudget, + ResourceGovernor, + ResourceSample, +) + + +PASS = "pass" +BLOCKED = "blocked" +FAIL = "fail" +CORPUS = "resource-watchdog.v1" +CASE_NAMES = ( + "budget_matrix", + "wall_deadline", + "idle_watchdog", + "clock_regression", + "sample_regression", + "cpu_limit", + "memory_limit", + "input_limit", + "output_limit", + "console_limit", + "observation_limit", + "message_limit", + "missing_memory_accounting", + "native_job_accounting", + "native_tree_reaping", +) + + +def _result(name: str, status: str, evidence: str, **details: Any) -> dict[str, Any]: + payload: dict[str, Any] = {"name": name, "status": status, "evidence": evidence} + if details: + payload["details"] = details + return payload + + +def _expect_code(action: Callable[[], Any], expected: str) -> bool: + try: + action() + except ResourceAccountingError as error: + return error.code == expected + except Exception: + return False + return False + + +def _constant_clock() -> Callable[[], float]: + return lambda: 0.0 + + +def _budget_matrix() -> dict[str, Any]: + scratch = ResourceBudget.scratch_auto_v1() + artifact = ResourceBudget.artifact_transform_v1() + expected = { + "scratch.auto.v1": { + "wall_time_ms": 10_000, + "cpu_time_ms": 5_000, + "memory_bytes": 256 * 1024 * 1024, + "max_messages": 256, + "max_bytes_read": 0, + "max_bytes_written": 0, + "max_console_bytes": 1 * 1024 * 1024, + "max_observation_bytes": 64 * 1024, + "idle_timeout_ms": 10_000, + }, + "artifact.transform.v1": { + "wall_time_ms": 60_000, + "cpu_time_ms": 30_000, + "memory_bytes": 512 * 1024 * 1024, + "max_messages": 16_384, + "max_bytes_read": 100 * 1024 * 1024, + "max_bytes_written": 128 * 1024 * 1024, + "max_console_bytes": 1 * 1024 * 1024, + "max_observation_bytes": 64 * 1024, + "idle_timeout_ms": 60_000, + }, + } + actual = { + budget.profile: { + key: getattr(budget, key) + for key in expected[budget.profile] + } + for budget in (scratch, artifact) + } + checks = { + "profiles_match_adr": actual == expected, + "hard_caps_immutable": artifact.with_watchdog( + wall_time_ms=60_000, + idle_timeout_ms=60_000, + max_messages=16_384, + ) + == artifact, + } + return _result( + "resource_budget_matrix", + PASS if all(checks.values()) else FAIL, + "The two ADR execution profiles resolve to immutable bounded values.", + checks=checks, + ) + + +def _watchdog_case(name: str, clock_values: list[float], expected: str, *, idle: int = 10) -> dict[str, Any]: + values = iter(clock_values) + + def clock() -> float: + return next(values) + + budget = ResourceBudget( + profile="qualification.v1", + wall_time_ms=100, + cpu_time_ms=100, + memory_bytes=1024, + max_messages=8, + max_bytes_read=8, + max_bytes_written=8, + idle_timeout_ms=idle, + ) + governor = ResourceGovernor(budget, clock=clock) + passed = _expect_code( + lambda: governor.observe( + ResourceSample(peak_memory_bytes=1), + progress=False, + ), + expected, + ) + return _result( + name, + PASS if passed else FAIL, + "A fixed monotonic clock produced the expected fail-closed watchdog category.", + expected_code=expected, + ) + + +def _sample_regression() -> dict[str, Any]: + governor = ResourceGovernor( + ResourceBudget( + profile="qualification.v1", + wall_time_ms=100, + cpu_time_ms=100, + memory_bytes=1024, + max_messages=8, + max_bytes_read=8, + max_bytes_written=8, + ), + clock=_constant_clock(), + ) + governor.observe(ResourceSample(peak_memory_bytes=4, messages=1)) + passed = _expect_code( + lambda: governor.observe(ResourceSample(peak_memory_bytes=None, messages=1)), + "accounting_invalid", + ) + return _result( + "sample_regression", + PASS if passed else FAIL, + "A missing or regressing cumulative memory sample is rejected after a value was observed.", + expected_code="accounting_invalid", + ) + + +def _limit_case(name: str, budget: ResourceBudget, sample: ResourceSample, expected: str) -> dict[str, Any]: + governor = ResourceGovernor(budget, clock=_constant_clock()) + passed = _expect_code(lambda: governor.observe(sample), expected) + return _result( + name, + PASS if passed else FAIL, + "Resource failure precedence is stable and reports only the safe category.", + expected_code=expected, + ) + + +def _native_job_accounting() -> dict[str, Any]: + try: + from native_launcher_qualification import _probe_resource_policy + + result = _probe_resource_policy() + except Exception as exc: + return _result( + "native_job_accounting", + FAIL, + "The fixed native Job Object accounting probe failed closed.", + error_type=type(exc).__name__, + ) + status = result.get("status") + if status == PASS: + return _result( + "native_job_accounting", + PASS, + "A fixed suspended AppContainer child returned configured limits and actual Job Object accounting.", + ) + if status == BLOCKED: + return _result( + "native_job_accounting", + BLOCKED, + "Windows Job Object accounting qualification is unavailable on this host.", + ) + return _result( + "native_job_accounting", + FAIL, + "The fixed native Job Object accounting probe did not pass every check.", + ) + + +def _native_tree_reaping() -> dict[str, Any]: + try: + from cancellation_corpus import run + + result = run() + except Exception as exc: + return _result( + "native_tree_reaping", + FAIL, + "The fixed kill-on-close watchdog corpus failed closed.", + error_type=type(exc).__name__, + ) + status = result.get("status") + if status == PASS: + return _result( + "native_tree_reaping", + PASS, + "A fixed AppContainer descendant tree was reaped by the kill-on-close watchdog path.", + ) + if status == BLOCKED: + return _result( + "native_tree_reaping", + BLOCKED, + "The Windows kill-on-close watchdog corpus is unavailable on this host.", + ) + return _result( + "native_tree_reaping", + FAIL, + "The fixed kill-on-close watchdog corpus did not prove full-tree reaping.", + ) + + +def run_qualification() -> dict[str, Any]: + """Return stable redacted evidence for the resource/watchdog corpus.""" + + small_budget = ResourceBudget( + profile="qualification.v1", + wall_time_ms=100, + cpu_time_ms=1, + memory_bytes=8, + max_messages=1, + max_bytes_read=1, + max_bytes_written=1, + max_console_bytes=1, + max_observation_bytes=1, + idle_timeout_ms=100, + ) + checks = [ + _budget_matrix(), + _watchdog_case("wall_deadline", [0.0, 0.100001], "deadline_exceeded", idle=100), + _watchdog_case("idle_watchdog", [0.0, 0.010001], "watchdog_stalled"), + _watchdog_case("clock_regression", [1.0, 0.5], "watchdog_clock_invalid"), + _sample_regression(), + _limit_case( + "cpu_limit", + small_budget, + ResourceSample(cpu_time_ms=2, peak_memory_bytes=1), + "cpu_exhausted", + ), + _limit_case( + "memory_limit", + small_budget, + ResourceSample(peak_memory_bytes=9), + "memory_exhausted", + ), + _limit_case( + "input_limit", + small_budget, + ResourceSample(peak_memory_bytes=1, bytes_read=2), + "input_limit", + ), + _limit_case( + "output_limit", + small_budget, + ResourceSample(peak_memory_bytes=1, bytes_written=2), + "output_limit", + ), + _limit_case( + "console_limit", + small_budget, + ResourceSample(peak_memory_bytes=1, console_bytes=2), + "console_limit", + ), + _limit_case( + "observation_limit", + small_budget, + ResourceSample(peak_memory_bytes=1, observation_bytes=2), + "observation_limit", + ), + _limit_case( + "message_limit", + small_budget, + ResourceSample(peak_memory_bytes=1, messages=2), + "message_budget_exhausted", + ), + ] + missing = ResourceGovernor( + ResourceBudget( + profile="qualification.v1", + wall_time_ms=100, + cpu_time_ms=100, + memory_bytes=1024, + max_messages=8, + max_bytes_read=8, + max_bytes_written=8, + ), + clock=_constant_clock(), + ) + missing.observe(ResourceSample()) + checks.append( + _result( + "missing_memory_accounting", + PASS + if _expect_code(missing.finish, "accounting_unavailable") + else FAIL, + "A terminal result without peak-memory accounting remains fail-closed.", + expected_code="accounting_unavailable", + ) + ) + checks.extend((_native_job_accounting(), _native_tree_reaping())) + + statuses = [check["status"] for check in checks] + if any(status == FAIL for status in statuses): + status = FAIL + elif any(status == BLOCKED for status in statuses): + status = BLOCKED + else: + status = PASS + digest = hashlib.sha256( + (CORPUS + "\n" + "\n".join(CASE_NAMES)).encode("ascii") + ).hexdigest()[:16] + return { + "name": "cortex-resource-watchdog-qualification", + "probe": "cortex-resource-watchdog-qualification", + "schema_version": 1, + "corpus": CORPUS, + "corpus_digest": digest, + "checks": checks, + "provider_launch_authorized": False, + "status": status, + "qualification_status": status, + } + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--json", action="store_true", help="Emit compact JSON only.") + parser.add_argument( + "--strict", + action="store_true", + help="Exit 2 unless every resource/watchdog case is green.", + ) + args = parser.parse_args() + report = run_qualification() + if args.json: + print(json.dumps(report, separators=(",", ":"), sort_keys=True)) + else: + print(json.dumps(report, indent=2, sort_keys=True)) + if args.strict and report["qualification_status"] != PASS: + return 2 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())