from __future__ import annotations import hashlib import json import math import re import shutil import socket import subprocess import sys import threading import time import uuid from collections.abc import Mapping from contextlib import suppress from dataclasses import dataclass from datetime import datetime, timedelta, timezone from email.utils import parsedate_to_datetime from http.client import HTTPException from pathlib import Path from typing import Any, Callable, NamedTuple from urllib.error import HTTPError, URLError from urllib.parse import quote, urlsplit from urllib.request import HTTPRedirectHandler, Request, build_opener, urlopen SEARCH_AUDIENCE = "https://search.azure.com" MANAGEMENT_AUDIENCE = "https://management.azure.com" SEARCH_HOST = re.compile( r"^[a-z0-9](?:[a-z0-9-]{0,58}[a-z0-9])?\.search\.windows\.net$" ) SECRET_FIELDS = { "accesskey", "accesstoken", "accountkey", "apikey", "applicationsecret", "authorization", "clientsecret", "credential", "connectionsecret", "key", "password", "privatekey", "refreshtoken", "sas", "sastoken", "secret", "sharedaccesskey", "storageaccountkey", "token", } RESOURCE_ID_CONNECTION = re.compile(r"^ResourceId=/[^;\r\n]+;?$") def normalize_azure_location(value: Any) -> str | None: """Normalize ASCII case/whitespace only; this does not validate region availability.""" if not isinstance(value, str) or len(value) > 128 or not value.isascii(): return None compact = re.sub(r"\s+", "", value, flags=re.ASCII).lower() return compact if re.fullmatch(r"[a-z][a-z0-9]{1,40}", compact) else None def _normalize_response_headers(headers: Mapping[str, str]) -> dict[str, str]: normalized: dict[str, str] = {} for name, value in headers.items(): key = name.lower() if key in {"x-ms-request-id", "request-id"} and normalized.get(key): continue if key == "retry-after" and key in normalized: normalized[key] = "" continue normalized[key] = value return normalized def _request_id(headers: Mapping[str, str]) -> str | None: normalized = _normalize_response_headers(headers) return normalized.get("x-ms-request-id") or normalized.get("request-id") or None @dataclass(frozen=True) class HttpResult: status: int body: Any headers: dict[str, str] etag_values: tuple[str, ...] | None = None recovery_deadline: float | None = None ack_failure: HelperFailure | None = None @property def request_id(self) -> str | None: return _request_id(self.headers) @property def retry_after(self) -> RetryAfter: return retry_after_metadata(self.headers) class RetryAfter(NamedTuple): kind: str value: float = 0 class RetryAfterTiming(NamedTuple): received_at_utc: float | None not_before_utc: float | None server_delay_seconds: int | None = None def valid_utc_timestamp(value): return type(value) in (int, float) and -62135596800 <= value < 253402300800 and math.isfinite(value) def retry_after_not_before(metadata, received_at): """Resolve typed metadata against UTC, without shortening a server interval.""" if not valid_utc_timestamp(received_at) or not isinstance(metadata, RetryAfter): return None value = metadata.value if not valid_utc_timestamp(value): return None if metadata.kind == "seconds": value = received_at + value if value >= 0 else math.inf elif metadata.kind == "date-rfc850": try: epoch = datetime(1970, 1, 1, tzinfo=timezone.utc) parsed = epoch + timedelta(seconds=value) current_year = (epoch + timedelta(seconds=received_at)).year year = current_year // 100 * 100 + parsed.year % 100 if year > current_year + 50: year -= 100 value = parsed.replace(year=year).timestamp() except (ValueError, OverflowError): return None elif metadata.kind in {"missing", "invalid"}: value = received_at + 1 elif metadata.kind != "date": return None return value if valid_utc_timestamp(value) else None def retry_after_timing(headers, *, received_at=None): """Portable timing alongside the unchanged, capped RetryAfter protocol.""" received_at = time.time() if received_at is None else received_at if not valid_utc_timestamp(received_at): return RetryAfterTiming(None, None) metadata = retry_after_metadata(headers) deadline = retry_after_not_before(metadata, received_at) server_delay = metadata.value if metadata.kind == "seconds" else None if metadata.kind == "overlong": values = [value for name, value in headers.items() if isinstance(name, str) and name.lower() == "retry-after"] if (len(values) == 1 and isinstance(values[0], str) and len(values[0]) <= 128 and re.fullmatch(r"[0-9]+", values[0].strip(" \t"))): seconds = int(values[0].strip(" \t")) if seconds < 253402300800 - received_at: deadline = received_at + seconds server_delay = seconds return RetryAfterTiming(received_at, deadline, server_delay) def retry_after_metadata(headers: Mapping[str, str]) -> RetryAfter: """Retain only a bounded delay/date, never raw server header text.""" values = [] for name, value in headers.items(): if isinstance(name, str) and name.lower() == "retry-after": values.append(value) if len(values) > 1: return RetryAfter("invalid") if not values: return RetryAfter("missing") value = values[0] if not isinstance(value, str): return RetryAfter("invalid") # An unbounded field cannot safely authorize an early request. if len(value) > 128: return RetryAfter("overlong") if not value.isascii(): return RetryAfter("invalid") value = value.strip(" \t") if re.fullmatch(r"[0-9]+", value): seconds = int(value) return RetryAfter("seconds", seconds) if seconds <= 30 else RetryAfter("overlong") # HTTP-date includes obsolete RFC850/asctime forms, but not arbitrary email dates. if not re.fullmatch( r"(?:[A-Z][a-z]{2}, [0-9]{2} [A-Z][a-z]{2} [0-9]{4} [0-9:]{8} GMT" r"|[A-Z][a-z]+, [0-9]{2}-[A-Z][a-z]{2}-[0-9]{2} [0-9:]{8} GMT" r"|[A-Z][a-z]{2} [A-Z][a-z]{2} [ 0-9][0-9] [0-9:]{8} [0-9]{4})", value ): return RetryAfter("invalid") try: parsed = parsedate_to_datetime(value) stamp = parsed.replace(tzinfo=timezone.utc).timestamp() return RetryAfter("date-rfc850" if "-" in value else "date", stamp) except (TypeError, ValueError, OverflowError): return RetryAfter("invalid") class HelperFailure(RuntimeError): def __init__( self, code: str, message: str, *, blocked_at: str, writes: list[dict[str, Any]] | None = None, resources_remaining: list[dict[str, Any]] | None = None, resources_reused: list[dict[str, Any]] | None = None, resources_unverified: list[dict[str, Any]] | None = None, request_id: str | None = None, status: int | None = None, partial: bool = False, warnings: list[str] | None = None, retry_after: RetryAfter | None = None, recovery_deadline: float | None = None, retry_after_timing: RetryAfterTiming | None = None, ) -> None: super().__init__(message) self.code = code self.message = message self.blocked_at = blocked_at self.writes = writes or [] self.resources_remaining = resources_remaining or [] self.resources_reused = resources_reused or [] self.resources_unverified = resources_unverified or [] self.request_id = request_id self.http_status = status self.partial = partial self.warnings = warnings or [] self.retry_after = retry_after or RetryAfter("missing") self.recovery_deadline = recovery_deadline self.retry_after_timing = retry_after_timing self.response_close_failed = False self.file_batch = None class ReadRecovery: """One 429 delay opportunity and one deadline across an explicit read sequence.""" def __init__(self, *, monotonic=None, wall_clock=None, sleeper=None, deadline=None, on_wait=None): self.monotonic = monotonic or time.monotonic self.wall_clock = wall_clock or time.time self.sleeper = sleeper or time.sleep self.deadline = min(self.monotonic() + 60, deadline if deadline is not None else math.inf) self.delayed = False self.request_ids: list[str] = [] self.warnings: list[str] = [] self.on_wait = on_wait @staticmethod def safe_id(request_id): return request_id if isinstance(request_id, str) and re.fullmatch( r"[A-Za-z0-9][A-Za-z0-9._:-]{0,127}", request_id ) else "[withheld]" def record(self, request_id): if request_id and len(self.request_ids) < 202: self.request_ids.append(self.safe_id(request_id)) def diagnostics(self): return [*self.warnings, *( ["Read-only request IDs: " + ", ".join(self.request_ids)] if self.request_ids else [] )] def annotate(self, failure): for warning in self.diagnostics(): if warning not in failure.warnings: failure.warnings.append(warning) return failure def _stop(self, code, message): self.warnings.append(message) raise self.annotate(HelperFailure( code, message + " Stop; resume only read-only verification, never replay the write.", blocked_at="verification", request_id=self.request_ids[-1] if self.request_ids else None, )) def delay(self, failure): if failure.blocked_at == "local-persistence": self._stop("read-recovery-persistence-failed", "Required receipt persistence failed; recovery is terminal.") if failure.response_close_failed: self._stop("read-recovery-response-close-failed", "HTTP error response cleanup failed; delayed recovery is blocked.") if failure.http_status != 429: return if failure.recovery_deadline is not None: self.deadline = min(self.deadline, failure.recovery_deadline) if self.delayed: self._stop("read-recovery-exhausted", "The single HTTP 429 delay opportunity is exhausted.") metadata = failure.retry_after delay = metadata.value if metadata.kind in {"date", "date-rfc850"}: now = self.wall_clock() deadline = retry_after_not_before(metadata, now) if deadline is None: self._stop("read-recovery-delay-exceeded", "Retry-After date cannot be represented safely.") delay = max(0, deadline - now) elif metadata.kind in {"missing", "invalid"}: delay = 1 self.warnings.append("Retry-After missing/invalid; using the fixed 1-second fallback.") if metadata.kind == "overlong" or not math.isfinite(delay) or delay > 30: self._stop("read-recovery-delay-exceeded", "Retry-After exceeds the 30-second wait allowance; it was not shortened.") remaining = self.deadline - self.monotonic() if remaining <= delay: self._stop("read-recovery-budget-exhausted", "Insufficient recovery read budget for Retry-After.") self.delayed = True start = self.monotonic() if self.on_wait is not None: self.on_wait(delay) self.sleeper(delay) elapsed = self.monotonic() - start if elapsed < delay: self._stop("read-recovery-wait-incomplete", "The required Retry-After delay did not elapse.") if self.monotonic() >= self.deadline: self._stop("read-recovery-budget-exhausted", "Recovery read deadline elapsed during the wait.") def get(self, url, token, *, transport, max_requests=2, **kwargs): self.deadline = min(self.deadline, kwargs.get("response_deadline", math.inf)) first = None for attempt in range(2): try: remaining = self.deadline - self.monotonic() if remaining <= 0 or attempt >= max_requests: self._stop("read-recovery-budget-exhausted", "The recovery read deadline elapsed.") options = {**kwargs, "follow_redirects": False, "response_deadline": self.deadline, "timeout": min(kwargs.get("timeout", 180), remaining), "max_response_bytes": kwargs.get("max_response_bytes", 1024 * 1024)} result = transport("GET", url, token, **options) if result.recovery_deadline is not None: self.deadline = min(self.deadline, result.recovery_deadline) self.record(result.request_id) if self.monotonic() >= self.deadline: self._stop("read-recovery-budget-exhausted", "The recovery response exceeded its read deadline.") if result.status != 429: return result raise HelperFailure( "azure-http-error", "Azure request failed with HTTP 429.", blocked_at="verification", status=429, request_id=result.request_id, retry_after=result.retry_after, recovery_deadline=result.recovery_deadline, ) except HelperFailure as failure: if not self.request_ids or self.request_ids[-1] != failure.request_id: self.record(failure.request_id) if failure.http_status != 429 or attempt: if first is not None: self.warnings.append( f"Recovery read stopped ({failure.code}); delay opportunity exhausted; " "initial read failure retained. Resume read-only; never replay the write." ) raise self.annotate(first) from failure raise self.annotate(failure) first = failure try: if attempt + 1 >= max_requests: self._stop("read-recovery-exhausted", "No requests remain in the recovery read allowance.") self.delay(failure) except HelperFailure as stopped: self.warnings.append(stopped.message) raise self.annotate(first) from stopped raise AssertionError("unreachable") def canonical_bytes(value: Any) -> bytes: return json.dumps( value, ensure_ascii=True, sort_keys=True, separators=(",", ":"), ).encode("utf-8") def digest(value: Any) -> str: return f"sha256:{hashlib.sha256(canonical_bytes(value)).hexdigest()}" def file_digest(path: Path) -> str: hasher = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): hasher.update(chunk) return f"sha256:{hasher.hexdigest()}" def load_approved_input(path: Path) -> tuple[dict[str, Any], dict[str, Any], str]: try: document = json.loads(path.read_text(encoding="utf-8")) except OSError as exc: raise HelperFailure( "input-unreadable", "Input file cannot be read.", blocked_at="input-resolution", ) from exc except json.JSONDecodeError as exc: raise HelperFailure( "input-invalid-json", f"Input file is not valid JSON: {exc}", blocked_at="input-resolution", ) from exc if not isinstance(document, dict) or document.get("schema_version") != "1.0": raise HelperFailure( "input-schema-invalid", "Input must be an object with schema_version 1.0.", blocked_at="input-resolution", ) reject_secrets(document) require_allowed_fields( document, {"schema_version", "plan", "approval"}, label="input envelope", ) plan = document.get("plan") approval = document.get("approval") if not isinstance(plan, dict) or not isinstance(approval, dict): raise HelperFailure( "input-schema-invalid", "Input must contain plan and approval objects.", blocked_at="input-resolution", ) require_allowed_fields( approval, {"confirmed", "fingerprint"}, label="approval", ) computed = digest(plan) if approval.get("confirmed") is not True: raise HelperFailure( "approval-missing", "The exact plan has not been explicitly approved.", blocked_at="confirmation", ) if approval.get("fingerprint") != computed: raise HelperFailure( "approval-mismatch", "The approved fingerprint does not match the canonical plan.", blocked_at="confirmation", ) return document, plan, computed def reject_secrets(value: Any, *, path: str = "") -> None: if isinstance(value, dict): for key, child in value.items(): normalized = re.sub(r"[^a-z0-9]", "", str(key).casefold()) child_path = f"{path}/{key}" if normalized in SECRET_FIELDS and child not in (None, "", []): raise HelperFailure( "secret-input-forbidden", f"Secret-bearing field is forbidden at {child_path}.", blocked_at="input-resolution", ) if normalized == "connectionstring" and child not in (None, ""): if ( not isinstance(child, str) or RESOURCE_ID_CONNECTION.fullmatch(child) is None ): raise HelperFailure( "secret-input-forbidden", "Only one ResourceId storage connectionString component is allowed.", blocked_at="input-resolution", ) reject_secrets(child, path=child_path) elif isinstance(value, list): for index, child in enumerate(value): reject_secrets(child, path=f"{path}/{index}") def require_allowed_fields( value: dict[str, Any], allowed: set[str], *, label: str, ) -> None: if set(value) - allowed: raise HelperFailure( "input-schema-invalid", f"{label} contains unsupported fields.", blocked_at="input-resolution", ) def is_ambiguous_mutation_failure(failure: HelperFailure) -> bool: return ( failure.partial or failure.code == "azure-response-ambiguous" or failure.http_status in {408, 429} or ( isinstance(failure.http_status, int) and failure.http_status >= 500 ) ) def is_ambiguous_status(status: int) -> bool: return status in {408, 429} or status >= 500 def sdk_error_status(error: Exception) -> int | None: for source in (error, getattr(error, "response", None)): status = getattr(source, "status_code", None) if type(status) is int and 100 <= status <= 599: return status return None def sdk_error_metadata(error: Exception, fallback_code: str | None = None) -> dict[str, Any]: """Read only bounded identifier fields, never exception text or response bodies.""" def identifier(value: Any) -> str | None: if isinstance(value, str) and re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.:-]{0,127}", value): return value return None response = getattr(error, "response", None) raw_headers = getattr(response, "headers", None) headers = {} if isinstance(raw_headers, Mapping): headers = _normalize_response_headers({ key: identifier(value) for key, value in raw_headers.items() if isinstance(key, str) and key.lower() in {"x-ms-request-id", "request-id", "x-ms-error-code"} }) result = {"status": sdk_error_status(error), "request_id": _request_id(headers)} if fallback_code is not None: detail = getattr(error, "error", None) code = detail.get("code") if isinstance(detail, Mapping) else getattr(detail, "code", None) result["code"] = ( identifier(code) or identifier(getattr(error, "code", None)) or headers.get("x-ms-error-code") or fallback_code ) return result def is_ambiguous_sdk_error(error: Exception) -> bool: status = sdk_error_status(error) return status is None or is_ambiguous_status(status) def redact_sensitive(value: Any) -> Any: if isinstance(value, dict): result: dict[str, Any] = {} for key, child in value.items(): normalized = re.sub(r"[^a-z0-9]", "", str(key).casefold()) if normalized in SECRET_FIELDS or normalized == "connectionstring": result[str(key)] = "[REDACTED]" else: result[str(key)] = redact_sensitive(child) return result if isinstance(value, list): return [redact_sensitive(child) for child in value] return value def validate_search_endpoint(endpoint: Any) -> str: if not isinstance(endpoint, str): raise HelperFailure( "endpoint-invalid", "Search endpoint must be a string.", blocked_at="input-resolution", ) try: endpoint.encode("utf-8") parsed = urlsplit(endpoint) port = parsed.port except (ValueError, UnicodeEncodeError) as exc: raise HelperFailure( "endpoint-invalid", "Search endpoint must be a valid UTF-8 HTTPS service root.", blocked_at="input-resolution", ) from exc if ( parsed.scheme != "https" or not parsed.hostname or SEARCH_HOST.fullmatch(parsed.hostname) is None or parsed.path not in {"", "/"} or parsed.query or parsed.fragment or parsed.username or parsed.password or port not in {None, 443} ): raise HelperFailure( "endpoint-invalid", "Search endpoint must be an HTTPS search.windows.net service root.", blocked_at="input-resolution", ) return endpoint.rstrip("/") def odata_name(name: Any) -> str: if not isinstance(name, str) or not name or len(name) > 128: raise HelperFailure( "name-invalid", "Resource name must be a non-empty string no longer than 128 characters.", blocked_at="input-resolution", ) try: return quote(name.replace("'", "''"), safe="") except UnicodeEncodeError as exc: raise HelperFailure( "name-invalid", "Resource name must be valid UTF-8 text.", blocked_at="input-resolution", ) from exc def azure_cli_token(resource: str) -> str: try: executable = shutil.which("az") if executable is None: raise FileNotFoundError("Azure CLI executable was not found on PATH.") completed = subprocess.run( [ executable, "account", "get-access-token", "--resource", resource, "--query", "accessToken", "--output", "tsv", ], check=True, capture_output=True, text=True, timeout=60, ) except FileNotFoundError as exc: raise HelperFailure( "azure-cli-unavailable", "Azure CLI is required for keyless authentication.", blocked_at="execution", ) from exc except subprocess.CalledProcessError as exc: raise HelperFailure( "azure-authentication-failed", "Azure CLI could not acquire the required access token.", blocked_at="execution", ) from exc except subprocess.TimeoutExpired as exc: raise HelperFailure( "azure-authentication-timeout", "Azure CLI did not return an access token within 60 seconds.", blocked_at="execution", ) from exc token = completed.stdout.strip() if not token: raise HelperFailure( "azure-authentication-failed", "Azure CLI returned an empty access token.", blocked_at="execution", ) return token class _NoRedirect(HTTPRedirectHandler): def redirect_request(self, req: Any, fp: Any, code: int, msg: str, headers: Any, newurl: str) -> None: return None def _read_response_with_deadline( response: Any, deadline: float, max_bytes: int, *, method: str ) -> bytes: def timed_out() -> HelperFailure: return HelperFailure( "response-deadline-exceeded", "HTTP response body exceeded its monotonic deadline.", blocked_at="verification", request_id=_request_id(response.headers), status=response.status, partial=method not in {"GET", "HEAD"}, ) remaining = deadline - time.monotonic() if remaining <= 0: raise timed_out() connection = getattr(getattr(getattr(response, "fp", None), "raw", None), "_sock", None) if not isinstance(connection, socket.socket): raise HelperFailure( "response-deadline-unsupported", "Deadline reads require an interruptible urllib HTTP socket.", blocked_at="verification", request_id=_request_id(response.headers), status=response.status, partial=method not in {"GET", "HEAD"}, ) expired = threading.Event() def interrupt() -> None: expired.set() # BufferedReader.close can wait on the active read lock; interrupt its socket instead. with suppress(OSError): connection.shutdown(socket.SHUT_RDWR) with suppress(OSError): connection.close() watchdog = threading.Timer(max(0, deadline - time.monotonic()), interrupt) watchdog.start() try: if expired.is_set() or time.monotonic() >= deadline: raise timed_out() try: payload = response.read(max_bytes + 1) except (OSError, HTTPException, ValueError) as exc: if expired.is_set() or time.monotonic() >= deadline: raise timed_out() from exc raise HelperFailure( "response-read-failed", "HTTP response body could not be read.", blocked_at="verification", request_id=_request_id(response.headers), status=response.status, partial=method not in {"GET", "HEAD"}, ) from exc if expired.is_set() or time.monotonic() >= deadline: raise timed_out() return payload finally: watchdog.cancel() watchdog.join() def http_request( method: str, url: str, token: str, *, body: bytes | None = None, headers: dict[str, str] | None = None, timeout: int = 180, raw_response: bool = False, max_response_bytes: int | None = None, follow_redirects: bool = True, response_deadline: float | None = None, ) -> HttpResult: if response_deadline is not None: if ( not isinstance(response_deadline, (int, float)) or not math.isfinite(response_deadline) or response_deadline - time.monotonic() > threading.TIMEOUT_MAX ): raise HelperFailure( "response-deadline-invalid", "Response deadline must be a supported finite monotonic time.", blocked_at="input-resolution", ) if max_response_bytes is None: max_response_bytes = 1024 * 1024 if not isinstance(max_response_bytes, int) or max_response_bytes < 0: raise HelperFailure( "response-limit-invalid", "Response byte limit must be a nonnegative integer.", blocked_at="input-resolution", ) if response_deadline <= time.monotonic(): raise HelperFailure( "response-deadline-exceeded", "HTTP deadline elapsed before the request.", blocked_at="verification", ) request_headers = { "Accept": "application/json;odata.metadata=minimal", "Authorization": f"Bearer {token}", "x-ms-client-request-id": str(uuid.uuid4()), } request_headers.update(headers or {}) request = Request(url=url, data=body, headers=request_headers, method=method) try: open_request = urlopen if follow_redirects else build_opener(_NoRedirect).open with open_request(request, timeout=timeout) as response: if response_deadline is not None: payload = _read_response_with_deadline( response, response_deadline, max_response_bytes, method=method ) else: payload = ( response.read(max_response_bytes + 1) if max_response_bytes is not None else response.read() ) etag_values = tuple(value for name, value in response.headers.items() if name.lower() == "etag") response_headers = _normalize_response_headers(response.headers) if max_response_bytes is not None and len(payload) > max_response_bytes: raise HelperFailure( "response-too-large", "Azure response exceeded the bounded read limit.", blocked_at="verification", request_id=_request_id(response_headers), status=response.status, partial=method not in {"GET", "HEAD"}, ) if raw_response: parsed: Any = payload elif not payload: parsed = None else: try: parsed = json.loads(payload.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise HelperFailure( "response-invalid-json", "Azure returned a non-JSON response where JSON was required.", blocked_at="verification", request_id=_request_id(response_headers), status=response.status, partial=method not in {"GET", "HEAD"}, ) from exc return HttpResult(response.status, parsed, response_headers, etag_values, response_deadline) except HTTPError as exc: retry_after = retry_after_metadata(exc.headers) timing = retry_after_timing(exc.headers) response_headers = _normalize_response_headers(exc.headers) status = int(exc.code) close_failed = False if status == 429: try: exc.close() except (OSError, HTTPException): close_failed = True elif response_deadline is not None: exc.close() ambiguous = method not in {"GET", "HEAD"} and is_ambiguous_status(status) failure = HelperFailure( "azure-http-error", f"Azure request failed with HTTP {status}.", blocked_at="execution", request_id=_request_id(response_headers), status=status, partial=ambiguous, retry_after=retry_after, recovery_deadline=response_deadline, retry_after_timing=timing, warnings=(["response-close-failed: HTTP 429 response cleanup failed; " "original HTTP failure retained and delayed recovery blocked."] if close_failed else []), ) failure.response_close_failed = close_failed raise failure from exc except (URLError, TimeoutError) as exc: raise HelperFailure( "azure-response-ambiguous", "Azure request did not return an authoritative response.", blocked_at="verification", partial=method not in {"GET", "HEAD"}, ) from exc def blocked_result( failure: HelperFailure, *, outcome: str, fingerprint: str | None, owner: Any = None, ) -> dict[str, Any]: result = _blocked_result(failure, outcome=outcome, fingerprint=fingerprint, owner=owner) if failure.file_batch is not None: result["file_batch"] = failure.file_batch result["safe_next_decision"] = ( "Retain the source and original ACK/journal. Do not rerun creation, replay an uncertain upload, " "or delete/reset resources. Plan upload-only continuation for never-attempted files with " "file_upload.py --plan; missing original evidence blocks continuation, not retention." ) return result def _blocked_result(failure, *, outcome, fingerprint, owner): if failure.partial or failure.writes: return { "status": "partial", "outcome": outcome, "approved_plan": { "fingerprint": fingerprint, "confirmed": fingerprint is not None, }, "first_failure": { "code": failure.code, "operation": failure.blocked_at, "status": failure.http_status, "message": failure.message, "request_id": failure.request_id, }, "completed_writes": failure.writes, "failed_or_unverified_postconditions": [failure.code], "resources_remaining": { "run_owned": failure.resources_remaining, "reused": failure.resources_reused, **({"unverified": failure.resources_unverified} if failure.resources_unverified else {}), }, "rollback": {"possible": False, "exact_plan": []}, "cleanup": {"status": "separate-plan-and-approval-required"}, "owner": owner, "warnings": failure.warnings, } result = { "status": "blocked", "outcome": outcome, "blocked_at": failure.blocked_at, "first_blocker": { "code": failure.code, "message": failure.message, "status": failure.http_status, "request_id": failure.request_id, }, "missing_or_conflicting_input": failure.code, "read_only_evidence": [], "writes_performed": [], "safe_next_decision": "Resolve the first blocker, rebuild the plan, and obtain new approval.", "ownership": {"run_owned": [], "reused_not_owned": []}, "cleanup": "not applicable", } if fingerprint is not None: result["approved_plan"] = { "fingerprint": fingerprint, "confirmed": True, } if failure.warnings: result["warnings"] = failure.warnings return result def emit_result( result: dict[str, Any], *, stream: Any = None, preserve_unapproved_input: bool = False ) -> None: if stream is None: stream = sys.stdout safe = redact_sensitive(result) if preserve_unapproved_input: document = result.get("execution_input") if ( result.get("status") != "planned" or not isinstance(document, dict) or document.get("schema_version") != "1.0" or not isinstance(document.get("plan"), dict) or not isinstance(document.get("approval"), dict) or document["approval"].get("confirmed") is not False or document.get("approval") != {"confirmed": False, "fingerprint": digest(document["plan"])} ): raise HelperFailure( "planning-output-invalid", "Only an exact unapproved execution input can retain ResourceId bindings.", blocked_at="verification", ) reject_secrets(document) require_allowed_fields(document, {"schema_version", "plan", "approval"}, label="planning envelope") safe["execution_input"] = document stream.write( json.dumps( safe, sort_keys=True, separators=(",", ":"), ) + "\n" ) TokenProvider = Callable[[str], str] Transport = Callable[..., HttpResult]