From e042cd752bf4a91daaf5582a170b015a1b751761 Mon Sep 17 00:00:00 2001 From: rob Date: Mon, 14 Sep 2026 22:37:12 -0300 Subject: [PATCH] Enforce canonical setup result contract in CLI and GUI --- src/cmdforge/_setup_contract/LICENSE | 21 + src/cmdforge/_setup_contract/README.md | 6 + src/cmdforge/_setup_contract/__init__.py | 1 + src/cmdforge/_setup_contract/contract.py | 304 ++++++ src/cmdforge/_setup_contract/models.py | 927 ++++++++++++++++++ src/cmdforge/cli/provider_commands.py | 34 +- .../gui/dialogs/provider_install_dialog.py | 28 +- tests/test_gui_provider_install.py | 63 +- tests/test_provider_install.py | 19 +- 9 files changed, 1350 insertions(+), 53 deletions(-) create mode 100644 src/cmdforge/_setup_contract/LICENSE create mode 100644 src/cmdforge/_setup_contract/README.md create mode 100644 src/cmdforge/_setup_contract/__init__.py create mode 100644 src/cmdforge/_setup_contract/contract.py create mode 100644 src/cmdforge/_setup_contract/models.py diff --git a/src/cmdforge/_setup_contract/LICENSE b/src/cmdforge/_setup_contract/LICENSE new file mode 100644 index 0000000..f833b42 --- /dev/null +++ b/src/cmdforge/_setup_contract/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 Rob and contributors + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/src/cmdforge/_setup_contract/README.md b/src/cmdforge/_setup_contract/README.md new file mode 100644 index 0000000..4755fb1 --- /dev/null +++ b/src/cmdforge/_setup_contract/README.md @@ -0,0 +1,6 @@ +# Setup completion contract snapshot + +Source: agent-harness-setup (MIT), core/models.py and contract.py. +Generated snapshot: do not edit independently. The helper's test_client_contract_parity.py +checks both client copies against the canonical source (only the model import differs). +No recipes, installer code, or credentials are included. diff --git a/src/cmdforge/_setup_contract/__init__.py b/src/cmdforge/_setup_contract/__init__.py new file mode 100644 index 0000000..885f537 --- /dev/null +++ b/src/cmdforge/_setup_contract/__init__.py @@ -0,0 +1 @@ +"""Vendored setup-result contract; see README.md.""" diff --git a/src/cmdforge/_setup_contract/contract.py b/src/cmdforge/_setup_contract/contract.py new file mode 100644 index 0000000..7c52016 --- /dev/null +++ b/src/cmdforge/_setup_contract/contract.py @@ -0,0 +1,304 @@ +"""Consumer-side contract for completion results (v1), as settled. + +``agent_harness_setup`` **produces** completion results; this module exists so that a consumer +can decide whether to trust one using the producer's strict models, and so that +the rules live in one reviewable place instead of being re-derived by each consumer. + +The rules here are deliberately conservative: a result is only accepted when the status, the +process exit code, and the verification/cleanup records agree. "Cannot tell" is never promoted to +"success" - that is the specific defect this contract exists to prevent. + +Clients without a runtime helper dependency vendor this module and its model unchanged +(apart from the relative model import). tests/test_client_contract_parity.py checks both +copies and runs the same malformed-result corpus against them. +""" + +from __future__ import annotations + +import os +import stat +import json +from pathlib import Path +from typing import Any, Mapping, Sequence + +#: Schema version this contract describes. Result payloads must say exactly this. +CONTRACT_SCHEMA_VERSION = 1 + +#: Statuses a result may carry (mirrors the producer's ``ResultStatus``). +CONTRACT_STATUSES = frozenset( + { + "success", + "failed", + "cancelled", + "unsupported_platform", + "manual_setup_required", + "dry_run", + "already_installed", + "approval_mismatch", + } +) + +#: Statuses that assert something *completed*. These require a known, matching exit code: an +#: unknown exit code must never be read as success. +COMPLETION_CLAIMING_STATUSES = frozenset({"success", "dry_run", "manual_setup_required"}) + +#: Exit codes that are coherent with each status. Measured against the helper, not assumed: +#: ``already_installed`` -> 7, ``approval_mismatch`` -> 8 and ``dry_run`` -> 0 were observed +#: directly; 3 (unsupported platform), 4 (manual setup) and 10/130/143 (cancellation) are the +#: documented codes, and ``failed`` covers execution failure (1) and cleanup unverified (11). +STATUS_EXIT_CODES: Mapping[str, frozenset[int]] = { + "success": frozenset({0}), + "dry_run": frozenset({0}), + "already_installed": frozenset({7}), + "approval_mismatch": frozenset({8}), + "unsupported_platform": frozenset({3}), + "manual_setup_required": frozenset({4}), + "cancelled": frozenset({10, 130, 143}), + # A failed run reports a non-zero status; 11 specifically means "installed and verified, but + # process cleanup could not be confirmed" (see the runner's cleanup contract). + "failed": frozenset({1, 3, 5, 11}), +} + +#: Exit code that separates "the helper finished" from "we do not know what happened". +CLEANUP_UNVERIFIED_EXIT_CODE = 11 + + +def status_exit_code_problems(status: Any, exit_code: int | None) -> list[str]: + """Return the reasons ``status`` and ``exit_code`` do not agree (empty list when they do).""" + problems: list[str] = [] + if not isinstance(status, str) or status not in CONTRACT_STATUSES: + return [f"unknown status {status!r}"] + allowed = STATUS_EXIT_CODES.get(status, frozenset()) + + if exit_code is None: + if status in COMPLETION_CLAIMING_STATUSES: + problems.append( + f"status {status!r} claims a completed run but the process exit code is unknown; " + "an unknown exit code must never be treated as success" + ) + return problems + + if not isinstance(exit_code, int) or isinstance(exit_code, bool): + return [f"exit code must be an integer or None, got {type(exit_code).__name__}"] + + if exit_code not in allowed: + problems.append( + f"status {status!r} is not coherent with exit code {exit_code} " + f"(expected one of {sorted(allowed)})" + ) + return problems + + +def _validation_error_from_model(payload: Mapping[str, Any]) -> str | None: + """Run mandatory model validation; an unavailable validator fails closed.""" + try: + from .models import CompletionResult + except ImportError: + return "result model validator is unavailable" + if not hasattr(CompletionResult, "from_dict"): # pragma: no cover - defensive + return "result model validator is unavailable" + try: + CompletionResult.from_dict(dict(payload)) + except (TypeError, ValueError) as exc: + return f"payload violates the result contract: {exc}" + return None + + +def validate_completion_result( + payload: Any, + *, + expected_request_id: str | None = None, + exit_code: int | None = None, +) -> list[str]: + """Return every reason ``payload`` must not be trusted (empty list means it may be). + + Checks, in order: JSON shape, contract version, status vocabulary, required evidence for a + completion claim, request correlation, status/exit-code coherence, and the producer's own + model validation when it is importable. + """ + problems: list[str] = [] + + if not isinstance(payload, Mapping): + return [f"result must be a JSON object, got {type(payload).__name__}"] + + schema_version = payload.get("schema_version") + if schema_version != CONTRACT_SCHEMA_VERSION: + problems.append( + f"schema_version must be {CONTRACT_SCHEMA_VERSION}, got {schema_version!r}" + ) + + status = payload.get("status") + if not isinstance(status, str) or status not in CONTRACT_STATUSES: + problems.append(f"unknown status {status!r}") + + if not isinstance(payload.get("request_id"), str) or not payload["request_id"].strip(): + problems.append("request_id must be a non-blank string") + + if expected_request_id is not None and payload.get("request_id") != expected_request_id: + problems.append( + f"request_id {payload.get('request_id')!r} does not match the expected " + f"{expected_request_id!r}" + ) + + # A success claim needs positive evidence on disk, not just a status string. + if status == "success": + verification = payload.get("verification") + if not isinstance(verification, Mapping) or verification.get("verified") is not True: + problems.append("success status requires verification.verified == true") + elif not isinstance(verification.get("executable_path"), str) or not str( + verification.get("executable_path") + ).strip(): + problems.append("success status requires a non-empty verification.executable_path") + if payload.get("cleanup_unverified") is True: + problems.append( + "success status contradicts cleanup_unverified == true (cleanup unknown must " + "not be reported as success)" + ) + if payload.get("cleanup_complete") is False: + problems.append("success status contradicts cleanup_complete == false") + + if payload.get("cleanup_complete") is True and payload.get("cleanup_unverified") is True: + problems.append("cleanup_complete true with cleanup_unverified true is contradictory") + + problems.extend(status_exit_code_problems(status, exit_code)) + + if not problems: + model_error = _validation_error_from_model(payload) + if model_error: + problems.append(model_error) + + return problems + + +def read_completion_result( + result_file: Path | str, + *, + expected_request_id: str | None, + exit_code: int | None, + expected_harness_id: str | None = None, +) -> tuple[str, dict | None, Sequence[str]]: + """Read bounded evidence once from a no-follow descriptor, then validate fully. + + This entry point requires correlation; payload-only validation remains available + for producer tests. No validator import failure is treated as successful validation. + """ + if not isinstance(expected_request_id, str) or not expected_request_id.strip(): + return "unknown", None, ["no correlation request ID was supplied for this run"] + path = Path(result_file) + private, reason = is_private_result_location(path) + if not private: + return "unknown", None, [reason] + try: + fd = os.open(path, os.O_RDONLY | os.O_NONBLOCK | getattr(os, "O_NOFOLLOW", 0)) + with os.fdopen(fd, "rb") as stream: + info = os.fstat(stream.fileno()) + if not stat.S_ISREG(info.st_mode) or info.st_uid != os.geteuid() or info.st_mode & 0o022: + return "unknown", None, ["result descriptor is not a private regular file"] + raw = stream.read(65537) + if len(raw) > 65536: + return "unknown", None, ["result exceeds 64 KiB"] + def unique_object(pairs): + obj = {} + for key, value in pairs: + if key in obj: + raise ValueError("duplicate JSON field") + obj[key] = value + return obj + payload = json.loads(raw.decode("utf-8"), object_pairs_hook=unique_object) + verdict, problems = consumer_verdict( + payload, expected_request_id=expected_request_id, exit_code=exit_code + ) + if expected_harness_id is not None and isinstance(payload, dict) and payload.get("harness_id") != expected_harness_id: + return "unknown", None, ["harness identity does not match this request"] + return verdict, payload if isinstance(payload, dict) else None, problems + except (OSError, ValueError, TypeError, UnicodeError, RecursionError) as exc: + return "unknown", None, [f"invalid completion result: {exc}"] + + +def is_private_result_location(result_file: Path | str) -> tuple[bool, str]: + """Whether a result file sits somewhere only its owner could have written it. + + A result file is the evidence a consumer acts on, so where it lives matters: a path writable + by other users can be replaced by a stale or forged result between the run and the read. + + Checks: the path is a regular file (not a symlink), owned by the current effective user, not + writable by group or other, and its directory is not writable by other (a sticky + world-writable directory such as ``/tmp`` is accepted, since others cannot replace the file + there and the file itself must still be private). ``mtime`` is + deliberately **not** used as freshness evidence: a timestamp is trivially set and says + nothing about who wrote the file - correlation and validation do that job. + """ + path = Path(result_file) + try: + st = os.lstat(path) + except OSError as exc: + return False, f"result file is not readable: {exc}" + + if stat.S_ISLNK(st.st_mode): + return False, "result file is a symlink; refuse to follow it" + if not stat.S_ISREG(st.st_mode): + return False, "result file is not a regular file" + if st.st_uid != os.geteuid(): + return False, f"result file is owned by uid {st.st_uid}, not by this user" + if st.st_mode & 0o022: + return False, "result file is writable by group or other" + + parent = path.parent + try: + parent_st = os.stat(parent) + except OSError as exc: + return False, f"result directory is not readable: {exc}" + # A world-writable directory is only acceptable when it is sticky: with the sticky bit other + # users cannot replace this file, and the file itself is private (checked above). /tmp is the + # familiar case. + if (parent_st.st_mode & 0o022) and not (parent_st.st_mode & 0o1000): + return False, "result directory is group/world-writable and not sticky" + + return True, "result file is owned by this user and not writable by others" + + +def consumer_verdict( + payload: Any, + *, + expected_request_id: str | None = None, + exit_code: int | None = None, + result_file: Path | str | None = None, +) -> tuple[str, Sequence[str]]: + """Summarise a result for a consumer: ``(verdict, problems)``. + + Verdicts: ``success`` (installed, verified, cleanup confirmed), ``installed_unverified`` + (binary is on disk but the run is not a success - exit 11 or a failed status with + verification evidence), ``failed``, ``cancelled``, ``manual_setup_required``, + ``already_installed``, ``unknown``. + """ + problems = list(validate_completion_result(payload, expected_request_id=expected_request_id, exit_code=exit_code)) + if result_file is not None: + private, reason = is_private_result_location(result_file) + if not private: + problems.append(f"result file location is not private: {reason}") + if problems: + return "unknown", problems + + status = payload["status"] + verification = payload.get("verification") or {} + installed = isinstance(verification, Mapping) and verification.get("verified") is True + + if status == "success": + return "success", () + if status == "already_installed": + return "already_installed", () + if status == "cancelled": + return "cancelled", () + if status == "manual_setup_required": + return "manual_setup_required", () + if status == "dry_run": + return "dry_run", () + if status == "approval_mismatch": + return "approval_mismatch", () + if status == "unsupported_platform": + return "unsupported_platform", () + if status == "failed": + if installed and exit_code == CLEANUP_UNVERIFIED_EXIT_CODE: + return "installed_unverified", () + return "failed", () + return "unknown", [f"unhandled status {status!r}"] diff --git a/src/cmdforge/_setup_contract/models.py b/src/cmdforge/_setup_contract/models.py new file mode 100644 index 0000000..9aefc3f --- /dev/null +++ b/src/cmdforge/_setup_contract/models.py @@ -0,0 +1,927 @@ +"""Pure data models with strict validation and bounded serialization. + +Note on schema vs semantic validation boundary: +Standard JSON Schema (draft-07) enforces structural constraints, types, enums, +non-blank strings, collection bounds, and conditional requirements. +However, standard JSON Schema cannot natively enforce certain runtime and +platform semantics, which are strictly validated by the Python data models: +1. POSIX absolute path syntax (Path(p).is_absolute()) for executable paths. +2. Multi-byte UTF-8 byte limits (e.g. max 4096 UTF-8 bytes for error_diagnostics + and version_output, where character count alone would allow overlong payloads). +3. Line count limits (max 20 lines for error_diagnostics). +4. Wire-level serialized UTF-8 byte length (max 65536 bytes / 64 KiB). + +Consumers must execute model semantic validation via CompletionResult.from_dict() +or CompletionResult.from_json() to ensure full boundary conformance. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field, fields +import json +from pathlib import Path +import re +from typing import Any, Literal, Mapping, get_args + +AuthStatus = Literal["unknown", "setup_required"] +ResultStatus = Literal[ + "success", + "failed", + "cancelled", + "unsupported_platform", + "manual_setup_required", + "dry_run", + # The harness was already present, so no installer ran and nothing was changed. Distinct + # from "success" (which would imply this run installed or verified it) and from "failed". + "already_installed", + # A machine caller supplied --approve-plan-digest, but the plan it would authorize is not + # the plan that is now current. Nothing executed; the caller must re-preview. + "approval_mismatch", +] +InstallMethod = Literal["argv", "script", "manual"] + +# Size and collection bounds (Contract §C.5) +MAX_RESULT_SERIALIZED_BYTES = 65536 # 64 KiB UTF-8 +MAX_IDENTIFIER_LENGTH = 128 + + +def _is_env_var_name(value: str) -> bool: + """Whether ``value`` looks like a POSIX environment variable name (UPPER_SNAKE_CASE).""" + return ( + bool(value) + and value[0].isalpha() + and value.upper() == value + and all(ch.isalnum() or ch == "_" for ch in value) + ) +MAX_DESCRIPTION_LENGTH = 1024 +MAX_PATH_LENGTH = 1024 +MAX_DIAGNOSTICS_BYTES = 4096 # 4096 UTF-8 bytes +MAX_DIAGNOSTICS_CHARS = 4096 +MAX_DIAGNOSTICS_LINES = 20 +MAX_COLLECTION_ITEMS = 128 +MAX_ADVISORY_ITEMS = 32 + +_SHA256_RE = re.compile(r"[0-9a-f]{64}") + +ALLOWED_STATUSES = frozenset(get_args(ResultStatus)) +"""Runtime mirror of :data:`ResultStatus`, derived so the two cannot drift.""" + +#: How an execution approval was obtained. "interactive" = a human answered the prompt; +#: "plan_digest" = a machine caller presented the digest of the plan it was shown. +ALLOWED_APPROVAL_SOURCES = frozenset({"interactive", "plan_digest"}) + +# V1 auth status: strictly unknown or setup_required. +# "authenticated" is intentionally rejected without a verified local evidence source. +ALLOWED_AUTH_STATUSES = frozenset({ + "unknown", + "setup_required", +}) + +ALLOWED_INSTALL_METHODS = frozenset({ + "argv", + "script", + "manual", +}) + + +KNOWN_PATH_ADVISORY_FIELDS = frozenset({ + "is_in_path", + "binary_dir", + "export_snippet", + "detected_shell", + "suggested_profile", +}) + +REQUIRED_COMPLETION_RESULT_FIELDS = frozenset({ + "schema_version", + "helper_version", + "request_id", + "timestamp", + "status", + "auth_status", + "path_updates", + "manual_steps_remaining", + "partial_changes", +}) + +KNOWN_PARTIAL_CHANGES_FIELDS = frozenset({ + "observed_created", + "observed_modified", + "unknown_changes", +}) + +KNOWN_VERIFICATION_FIELDS = frozenset({ + "verified", + "executable_path", + "version_output", +}) + + +def _is_exact_int(val: Any) -> bool: + """Check that value is an integer and NOT a boolean.""" + return type(val) is int + + +def _is_exact_bool(val: Any) -> bool: + """Check that value is a boolean.""" + return type(val) is bool + + +def _validate_str_sequence( + items: Any, + field_name: str, + *, + max_items: int, + max_item_length: int = MAX_PATH_LENGTH, + allow_empty_items: bool = False, +) -> tuple[str, ...]: + """Validate and convert a sequence of strings without coercion. + + Rejects mutable/nested invalid types (e.g. ints, None, dicts) without coercion. + Returns an immutable tuple of strings. + """ + if not isinstance(items, (list, tuple)): + raise ValueError(f"{field_name} must be a list or tuple, got {type(items).__name__}") + if len(items) > max_items: + raise ValueError(f"{field_name} exceeds maximum allowed {max_items} items") + + validated: list[str] = [] + for idx, item in enumerate(items): + if not isinstance(item, str): + raise ValueError( + f"{field_name}[{idx}] must be a string, got {type(item).__name__} ({item!r})" + ) + if not allow_empty_items and not item.strip(): + raise ValueError(f"{field_name}[{idx}] must not be empty or blank") + if len(item) > max_item_length: + raise ValueError( + f"{field_name}[{idx}] exceeds maximum allowed {max_item_length} characters" + ) + validated.append(item) + return tuple(validated) + + +@dataclass(frozen=True) +class PrereqDef: + """Prerequisite definition for a harness recipe.""" + + name: str + command: str + check_args: tuple[str, ...] = ("--version",) + description: str = "" + + def __post_init__(self) -> None: + if not isinstance(self.name, str) or not self.name.strip(): + raise ValueError("PrereqDef.name must be a non-empty string") + if not isinstance(self.command, str) or not self.command.strip(): + raise ValueError("PrereqDef.command must be a non-empty string") + val_check_args = _validate_str_sequence( + self.check_args, "PrereqDef.check_args", max_items=32, max_item_length=256, allow_empty_items=True + ) + object.__setattr__(self, "check_args", val_check_args) + if not isinstance(self.description, str): + raise ValueError("PrereqDef.description must be a string") + + def to_dict(self) -> dict[str, Any]: + return { + "name": self.name, + "command": self.command, + "check_args": list(self.check_args), + "description": self.description, + } + + @classmethod + def from_dict(cls, d: Mapping[str, Any]) -> PrereqDef: + if not isinstance(d, Mapping): + raise ValueError(f"PrereqDef payload must be a mapping, got {type(d).__name__}") + if "name" not in d or "command" not in d: + raise ValueError("PrereqDef requires 'name' and 'command'") + return cls( + name=d["name"], + command=d["command"], + check_args=d.get("check_args", ("--version",)), + description=d.get("description", ""), + ) + + +KNOWN_UNINSTALL_FIELDS = frozenset({ + "uninstall_kind", + "uninstall_args", + "config_paths", + "cache_paths", + "model_paths", + "notes", +}) + + +@dataclass(frozen=True) +class UninstallDef: + """Uninstallation and footprint cleanup specification.""" + + uninstall_kind: str + uninstall_args: tuple[str, ...] = () + config_paths: tuple[str, ...] = () + cache_paths: tuple[str, ...] = () + model_paths: tuple[str, ...] = () + notes: str = "" + + def __post_init__(self) -> None: + if not isinstance(self.uninstall_kind, str) or not self.uninstall_kind.strip(): + raise ValueError("uninstall_kind must be a non-empty string") + val_args = _validate_str_sequence(self.uninstall_args, "uninstall_args", max_items=64, max_item_length=MAX_PATH_LENGTH) + object.__setattr__(self, "uninstall_args", val_args) + val_cfg = _validate_str_sequence(self.config_paths, "config_paths", max_items=32, max_item_length=MAX_PATH_LENGTH) + object.__setattr__(self, "config_paths", val_cfg) + val_cache = _validate_str_sequence(self.cache_paths, "cache_paths", max_items=32, max_item_length=MAX_PATH_LENGTH) + object.__setattr__(self, "cache_paths", val_cache) + val_models = _validate_str_sequence(self.model_paths, "model_paths", max_items=32, max_item_length=MAX_PATH_LENGTH) + object.__setattr__(self, "model_paths", val_models) + if not isinstance(self.notes, str): + raise ValueError("notes must be a string") + + def to_dict(self) -> dict[str, Any]: + return { + "uninstall_kind": self.uninstall_kind, + "uninstall_args": list(self.uninstall_args), + "config_paths": list(self.config_paths), + "cache_paths": list(self.cache_paths), + "model_paths": list(self.model_paths), + "notes": self.notes, + } + + @classmethod + def from_dict(cls, d: Mapping[str, Any]) -> UninstallDef: + if not isinstance(d, Mapping): + raise ValueError(f"UninstallDef payload must be a mapping, got {type(d).__name__}") + unknown = set(d.keys()) - KNOWN_UNINSTALL_FIELDS + if unknown: + raise ValueError(f"Unknown field(s) in UninstallDef: {sorted(unknown)}") + if "uninstall_kind" not in d: + raise ValueError("Missing required field 'uninstall_kind' in UninstallDef") + return cls( + uninstall_kind=d["uninstall_kind"], + uninstall_args=d.get("uninstall_args", ()), + config_paths=d.get("config_paths", ()), + cache_paths=d.get("cache_paths", ()), + model_paths=d.get("model_paths", ()), + notes=d.get("notes", ""), + ) + + +@dataclass(frozen=True) +class HarnessRecipe: + """Immutable definition of an agent harness installation recipe.""" + + harness_id: str + display_name: str + description: str + platforms: tuple[str, ...] + prerequisites: tuple[PrereqDef, ...] + install_kind: str + install_args: tuple[str, ...] + target_binaries: tuple[str, ...] + candidate_paths: tuple[str, ...] + verification_args: tuple[str, ...] + auth_kind: str + auth_instructions: str + source_citation: str + verified_date: str + uninstall_def: UninstallDef | None = None + #: Environment variables that authenticate this harness, in the order the doctor should + #: report them. Declared here rather than in a separate map so the tool that installs a + #: harness is also the tool that says which credential it consumes - one record per + #: harness, and no second list to drift out of step with it. + credential_env_vars: tuple[str, ...] = () + #: Loopback endpoint of a local daemon this harness serves, when it has one (Ollama). + #: ``None`` means the harness has no daemon to ping, which is different from "unknown". + local_daemon_endpoint: str | None = None + + def __post_init__(self) -> None: + if not isinstance(self.harness_id, str) or not self.harness_id.strip(): + raise ValueError("harness_id must be a non-empty string") + if len(self.harness_id) > MAX_IDENTIFIER_LENGTH: + raise ValueError(f"harness_id exceeds {MAX_IDENTIFIER_LENGTH} characters") + if not isinstance(self.display_name, str) or not self.display_name.strip(): + raise ValueError("display_name must be a non-empty string") + if not isinstance(self.description, str): + raise ValueError("description must be a string") + + val_platforms = _validate_str_sequence( + self.platforms, "platforms", max_items=16, max_item_length=64 + ) + if not val_platforms: + raise ValueError("platforms must not be empty") + object.__setattr__(self, "platforms", val_platforms) + + if not isinstance(self.prerequisites, (list, tuple)): + raise ValueError("prerequisites must be a list or tuple of PrereqDef") + for idx, p in enumerate(self.prerequisites): + if not isinstance(p, PrereqDef): + raise ValueError(f"prerequisites[{idx}] must be PrereqDef, got {type(p).__name__}") + object.__setattr__(self, "prerequisites", tuple(self.prerequisites)) + + if not isinstance(self.install_kind, str): + raise ValueError(f"install_kind must be a string, got {type(self.install_kind).__name__}") + if self.install_kind not in ALLOWED_INSTALL_METHODS: + raise ValueError(f"install_kind must be one of {sorted(ALLOWED_INSTALL_METHODS)}") + + val_install_args = _validate_str_sequence( + self.install_args, "install_args", max_items=128, max_item_length=1024, allow_empty_items=True + ) + object.__setattr__(self, "install_args", val_install_args) + + val_binaries = _validate_str_sequence( + self.target_binaries, "target_binaries", max_items=32, max_item_length=128 + ) + object.__setattr__(self, "target_binaries", val_binaries) + + val_candidate_paths = _validate_str_sequence( + self.candidate_paths, "candidate_paths", max_items=32, max_item_length=1024 + ) + object.__setattr__(self, "candidate_paths", val_candidate_paths) + + val_ver_args = _validate_str_sequence( + self.verification_args, "verification_args", max_items=32, max_item_length=256, allow_empty_items=True + ) + object.__setattr__(self, "verification_args", val_ver_args) + + if not isinstance(self.auth_kind, str): + raise ValueError(f"auth_kind must be a string, got {type(self.auth_kind).__name__}") + if self.auth_kind not in ("setup_required", "none", "unknown"): + raise ValueError("auth_kind must be 'setup_required', 'none', or 'unknown'") + if not isinstance(self.auth_instructions, str): + raise ValueError("auth_instructions must be a string") + if not isinstance(self.source_citation, str): + raise ValueError("source_citation must be a string") + if not isinstance(self.verified_date, str): + raise ValueError("verified_date must be a string") + if self.uninstall_def is not None and not isinstance(self.uninstall_def, UninstallDef): + raise ValueError(f"uninstall_def must be an UninstallDef or None, got {type(self.uninstall_def).__name__}") + + val_cred_vars = _validate_str_sequence( + self.credential_env_vars, "credential_env_vars", max_items=16, max_item_length=128 + ) + for var in val_cred_vars: + if not _is_env_var_name(var): + raise ValueError( + f"credential_env_vars entry {var!r} is not an environment variable name " + "(expected e.g. 'ANTHROPIC_API_KEY')" + ) + object.__setattr__(self, "credential_env_vars", val_cred_vars) + + if self.local_daemon_endpoint is not None: + endpoint = self.local_daemon_endpoint + if not isinstance(endpoint, str) or not endpoint.startswith(("http://", "https://")): + raise ValueError( + "local_daemon_endpoint must be None or an absolute http(s) URL, got " + f"{endpoint!r}" + ) + + def to_dict(self) -> dict[str, Any]: + return { + "harness_id": self.harness_id, + "display_name": self.display_name, + "description": self.description, + "platforms": list(self.platforms), + "prerequisites": [p.to_dict() for p in self.prerequisites], + "install_kind": self.install_kind, + "install_args": list(self.install_args), + "target_binaries": list(self.target_binaries), + "candidate_paths": list(self.candidate_paths), + "verification_args": list(self.verification_args), + "auth_kind": self.auth_kind, + "auth_instructions": self.auth_instructions, + "source_citation": self.source_citation, + "verified_date": self.verified_date, + "credential_env_vars": list(self.credential_env_vars), + "local_daemon_endpoint": self.local_daemon_endpoint, + "uninstall_def": self.uninstall_def.to_dict() if self.uninstall_def is not None else None, + } + + @classmethod + def from_dict(cls, d: Mapping[str, Any]) -> HarnessRecipe: + if not isinstance(d, Mapping): + raise ValueError(f"HarnessRecipe payload must be a mapping, got {type(d).__name__}") + prereqs_raw = d.get("prerequisites", ()) + if not isinstance(prereqs_raw, (list, tuple)): + raise ValueError("prerequisites must be a list or tuple") + prereqs: list[PrereqDef] = [] + for p in prereqs_raw: + if isinstance(p, PrereqDef): + prereqs.append(p) + elif isinstance(p, Mapping): + prereqs.append(PrereqDef.from_dict(p)) + else: + raise ValueError(f"prerequisite item must be PrereqDef or dict, got {type(p).__name__}") + + u_raw = d.get("uninstall_def") + uninstall_def: UninstallDef | None = None + if u_raw is not None: + if isinstance(u_raw, UninstallDef): + uninstall_def = u_raw + elif isinstance(u_raw, Mapping): + uninstall_def = UninstallDef.from_dict(u_raw) + else: + raise ValueError(f"uninstall_def must be an object or None, got {type(u_raw).__name__}") + + return cls( + harness_id=d["harness_id"], + display_name=d["display_name"], + description=d.get("description", ""), + platforms=d["platforms"], + prerequisites=tuple(prereqs), + install_kind=d["install_kind"], + install_args=d.get("install_args", ()), + target_binaries=d.get("target_binaries", ()), + candidate_paths=d.get("candidate_paths", ()), + verification_args=d.get("verification_args", ()), + auth_kind=d["auth_kind"], + auth_instructions=d.get("auth_instructions", ""), + source_citation=d.get("source_citation", ""), + verified_date=d.get("verified_date", ""), + credential_env_vars=tuple(d.get("credential_env_vars", ())), + local_daemon_endpoint=d.get("local_daemon_endpoint"), + uninstall_def=uninstall_def, + ) + + +@dataclass(frozen=True) +class VerificationResult: + """Post-installation verification result.""" + + verified: bool + executable_path: str | None = None + version_output: str | None = None + + def __post_init__(self) -> None: + if not _is_exact_bool(self.verified): + raise ValueError(f"VerificationResult.verified must be a boolean, got {type(self.verified).__name__}") + + if self.executable_path is not None: + if not isinstance(self.executable_path, str) or not self.executable_path.strip(): + raise ValueError("executable_path must be a non-empty string or None") + if len(self.executable_path) > MAX_PATH_LENGTH: + raise ValueError(f"executable_path exceeds {MAX_PATH_LENGTH} characters") + if not Path(self.executable_path).is_absolute(): + raise ValueError(f"executable_path must be absolute: {self.executable_path}") + + if self.verified: + if self.executable_path is None: + raise ValueError("Verified result must provide a non-empty executable_path") + + if self.version_output is not None: + if not isinstance(self.version_output, str): + raise ValueError(f"version_output must be a string or None, got {type(self.version_output).__name__}") + if len(self.version_output) > MAX_DIAGNOSTICS_CHARS: + raise ValueError(f"version_output exceeds {MAX_DIAGNOSTICS_CHARS} characters") + if len(self.version_output.encode("utf-8")) > MAX_DIAGNOSTICS_BYTES: + raise ValueError(f"version_output exceeds {MAX_DIAGNOSTICS_BYTES} UTF-8 bytes") + + def to_dict(self) -> dict[str, Any]: + return { + "verified": self.verified, + "executable_path": self.executable_path, + "version_output": self.version_output, + } + + @classmethod + def from_dict(cls, d: Mapping[str, Any]) -> VerificationResult: + if not isinstance(d, Mapping): + raise ValueError(f"VerificationResult payload must be a mapping, got {type(d).__name__}") + unknown = set(d.keys()) - KNOWN_VERIFICATION_FIELDS + if unknown: + raise ValueError(f"Unknown field(s) in VerificationResult: {sorted(unknown)}") + if "verified" not in d: + raise ValueError("Missing required field 'verified' in VerificationResult") + return cls( + verified=d["verified"], + executable_path=d.get("executable_path"), + version_output=d.get("version_output"), + ) + + +@dataclass(frozen=True) +class PartialChanges: + """Accounting of observed filesystem changes and unknown additional changes.""" + + observed_created: tuple[str, ...] = () + observed_modified: tuple[str, ...] = () + unknown_changes: bool = False + + def __post_init__(self) -> None: + val_created = _validate_str_sequence( + self.observed_created, "observed_created", max_items=MAX_COLLECTION_ITEMS, max_item_length=MAX_PATH_LENGTH + ) + object.__setattr__(self, "observed_created", val_created) + + val_modified = _validate_str_sequence( + self.observed_modified, "observed_modified", max_items=MAX_COLLECTION_ITEMS, max_item_length=MAX_PATH_LENGTH + ) + object.__setattr__(self, "observed_modified", val_modified) + + if not _is_exact_bool(self.unknown_changes): + raise ValueError(f"unknown_changes must be a boolean, got {type(self.unknown_changes).__name__}") + + def to_dict(self) -> dict[str, Any]: + return { + "observed_created": list(self.observed_created), + "observed_modified": list(self.observed_modified), + "unknown_changes": self.unknown_changes, + } + + @classmethod + def from_dict(cls, d: Mapping[str, Any]) -> PartialChanges: + if not isinstance(d, Mapping): + raise ValueError(f"PartialChanges payload must be a mapping, got {type(d).__name__}") + unknown = set(d.keys()) - KNOWN_PARTIAL_CHANGES_FIELDS + if unknown: + raise ValueError(f"Unknown field(s) in PartialChanges: {sorted(unknown)}") + missing = KNOWN_PARTIAL_CHANGES_FIELDS - set(d.keys()) + if missing: + raise ValueError(f"Missing required field(s) in PartialChanges: {sorted(missing)}") + + return cls( + observed_created=d["observed_created"], + observed_modified=d["observed_modified"], + unknown_changes=d["unknown_changes"], + ) + + +@dataclass(frozen=True) +class PathAdvisory: + """PATH inspection and shell environment configuration advisory.""" + + is_in_path: bool + binary_dir: str + export_snippet: str + detected_shell: str + suggested_profile: str | None = None + + def __post_init__(self) -> None: + if not _is_exact_bool(self.is_in_path): + raise ValueError(f"is_in_path must be a boolean, got {type(self.is_in_path).__name__}") + if not isinstance(self.binary_dir, str) or not self.binary_dir.strip(): + raise ValueError("binary_dir must be a non-empty string") + if not isinstance(self.export_snippet, str) or not self.export_snippet.strip(): + raise ValueError("export_snippet must be a non-empty string") + if not isinstance(self.detected_shell, str) or not self.detected_shell.strip(): + raise ValueError("detected_shell must be a non-empty string") + if self.suggested_profile is not None: + if not isinstance(self.suggested_profile, str) or not self.suggested_profile.strip(): + raise ValueError("suggested_profile must be a non-empty string or None") + + def to_dict(self) -> dict[str, Any]: + return { + "is_in_path": self.is_in_path, + "binary_dir": self.binary_dir, + "export_snippet": self.export_snippet, + "detected_shell": self.detected_shell, + "suggested_profile": self.suggested_profile, + } + + @classmethod + def from_dict(cls, d: Mapping[str, Any]) -> PathAdvisory: + if not isinstance(d, Mapping): + raise ValueError(f"PathAdvisory payload must be a mapping, got {type(d).__name__}") + unknown = set(d.keys()) - KNOWN_PATH_ADVISORY_FIELDS + if unknown: + raise ValueError(f"Unknown field(s) in PathAdvisory: {sorted(unknown)}") + for req in ("is_in_path", "binary_dir", "export_snippet", "detected_shell"): + if req not in d: + raise ValueError(f"Missing required field '{req}' in PathAdvisory") + return cls( + is_in_path=d["is_in_path"], + binary_dir=d["binary_dir"], + export_snippet=d["export_snippet"], + detected_shell=d["detected_shell"], + suggested_profile=d.get("suggested_profile"), + ) + + +@dataclass(frozen=True) +class CompletionResult: + """Versioned, bounded, validated machine-readable completion result.""" + + helper_version: str + request_id: str + timestamp: str + status: str + schema_version: int = 1 + harness_id: str | None = None + recipe_id: str | None = None + recipe_version: str | None = None + install_method: str | None = None + installer_exit_code: int | None = None + verification: VerificationResult | None = None + path_updates: tuple[str, ...] = () + auth_status: str = "unknown" + manual_steps_remaining: tuple[str, ...] = () + partial_changes: PartialChanges = field(default_factory=PartialChanges) + error_diagnostics: str | None = None + path_advisory: PathAdvisory | None = None + #: SHA-256 of the plan this run approved or previewed, so a caller can prove which exact + #: plan was consented to. + plan_digest: str | None = None + #: "interactive" or "plan_digest"; None when the run never had an approved plan. + approval_source: str | None = None + #: Whether the supervised installer process group was confirmed torn down. ``None`` when no + #: installer process was run, so there was nothing to clean up. + cleanup_complete: bool | None = None + #: True when teardown could not be confirmed - inspection unavailable, or members surviving + #: SIGKILL. This never means "nothing survived": processes that detach into their own + #: session, and platforms where the group cannot be inspected, stay outside what this run + #: can observe or terminate. + cleanup_unverified: bool = False + #: What teardown did, and what it could not establish. + cleanup_notes: tuple[str, ...] = () + + def __post_init__(self) -> None: + # Schema version: strict exact int == 1 + if not _is_exact_int(self.schema_version) or self.schema_version != 1: + raise ValueError(f"schema_version must be exact integer 1, got {self.schema_version!r}") + + # Required non-empty string fields + if not isinstance(self.helper_version, str) or not self.helper_version.strip(): + raise ValueError("helper_version must be a non-empty string") + if len(self.helper_version) > MAX_IDENTIFIER_LENGTH: + raise ValueError(f"helper_version exceeds {MAX_IDENTIFIER_LENGTH} chars") + + if not isinstance(self.request_id, str) or not self.request_id.strip(): + raise ValueError("request_id must be a non-empty string") + if len(self.request_id) > MAX_IDENTIFIER_LENGTH: + raise ValueError(f"request_id exceeds {MAX_IDENTIFIER_LENGTH} chars") + + if not isinstance(self.timestamp, str) or not self.timestamp.strip(): + raise ValueError("timestamp must be a non-empty string") + + # Status validation + if not isinstance(self.status, str): + raise ValueError(f"status must be a string, got {type(self.status).__name__}") + if self.status not in ALLOWED_STATUSES: + raise ValueError(f"status must be one of {sorted(ALLOWED_STATUSES)}, got {self.status!r}") + + # Plan digest / approval provenance (optional, but must be well-formed when present) + if self.plan_digest is not None: + if not isinstance(self.plan_digest, str) or not _SHA256_RE.fullmatch(self.plan_digest): + raise ValueError( + "plan_digest must be a 64-character lowercase hex SHA-256 digest or None, " + f"got {self.plan_digest!r}" + ) + if self.approval_source is not None: + if not isinstance(self.approval_source, str) or self.approval_source not in ALLOWED_APPROVAL_SOURCES: + raise ValueError( + f"approval_source must be one of {sorted(ALLOWED_APPROVAL_SOURCES)} or None, " + f"got {self.approval_source!r}" + ) + if self.plan_digest is None: + raise ValueError("approval_source requires plan_digest: an approval always binds a plan") + + # Auth status validation (V1 contract: unknown or setup_required only) + if not isinstance(self.auth_status, str): + raise ValueError(f"auth_status must be a string, got {type(self.auth_status).__name__}") + if self.auth_status not in ALLOWED_AUTH_STATUSES: + if self.auth_status == "authenticated": + raise ValueError( + "In V1, auth_status cannot be 'authenticated' without a verified local evidence source. " + "Allowed values: 'unknown', 'setup_required'." + ) + raise ValueError(f"auth_status must be one of {sorted(ALLOWED_AUTH_STATUSES)}, got {self.auth_status!r}") + + # Optional identity strings + for field_name, val in ( + ("harness_id", self.harness_id), + ("recipe_id", self.recipe_id), + ("recipe_version", self.recipe_version), + ): + if val is not None: + if not isinstance(val, str) or not val.strip(): + raise ValueError(f"{field_name} must be a non-empty string or None") + if len(val) > MAX_IDENTIFIER_LENGTH: + raise ValueError(f"{field_name} exceeds {MAX_IDENTIFIER_LENGTH} chars") + + # Install method validation + if self.install_method is not None: + if not isinstance(self.install_method, str): + raise ValueError(f"install_method must be a string or None, got {type(self.install_method).__name__}") + if self.install_method not in ALLOWED_INSTALL_METHODS: + raise ValueError(f"install_method must be one of {sorted(ALLOWED_INSTALL_METHODS)} or None, got {self.install_method!r}") + + # Installer return code: exact integer or None (never a boolean) + if self.installer_exit_code is not None: + if not _is_exact_int(self.installer_exit_code): + raise ValueError(f"installer_exit_code must be exact integer or None, got {self.installer_exit_code!r}") + + # Verification validation + if self.verification is not None and not isinstance(self.verification, VerificationResult): + raise ValueError(f"verification must be a VerificationResult or None, got {type(self.verification).__name__}") + + # Defensive tuple conversion and validation for collections + val_paths = _validate_str_sequence( + self.path_updates, "path_updates", max_items=MAX_ADVISORY_ITEMS, max_item_length=MAX_PATH_LENGTH + ) + object.__setattr__(self, "path_updates", val_paths) + + val_steps = _validate_str_sequence( + self.manual_steps_remaining, "manual_steps_remaining", max_items=MAX_ADVISORY_ITEMS, max_item_length=MAX_DESCRIPTION_LENGTH + ) + object.__setattr__(self, "manual_steps_remaining", val_steps) + + # Partial changes + if not isinstance(self.partial_changes, PartialChanges): + raise ValueError(f"partial_changes must be a PartialChanges instance, got {type(self.partial_changes).__name__}") + + # Path advisory validation + if self.path_advisory is not None and not isinstance(self.path_advisory, PathAdvisory): + raise ValueError(f"path_advisory must be a PathAdvisory or None, got {type(self.path_advisory).__name__}") + + # Diagnostics string and UTF-8 byte bounds + if self.error_diagnostics is not None: + if not isinstance(self.error_diagnostics, str): + raise ValueError("error_diagnostics must be a string or None") + if len(self.error_diagnostics) > MAX_DIAGNOSTICS_CHARS: + raise ValueError(f"error_diagnostics exceeds {MAX_DIAGNOSTICS_CHARS} characters") + diag_bytes = len(self.error_diagnostics.encode("utf-8")) + if diag_bytes > MAX_DIAGNOSTICS_BYTES: + raise ValueError( + f"error_diagnostics ({diag_bytes} UTF-8 bytes) " + f"exceeds maximum allowed {MAX_DIAGNOSTICS_BYTES} bytes" + ) + if len(self.error_diagnostics.splitlines()) > MAX_DIAGNOSTICS_LINES: + raise ValueError(f"error_diagnostics exceeds {MAX_DIAGNOSTICS_LINES} lines") + + # STRICT SUCCESS INVARIANTS (Contract §C.4) + if self.status == "success": + if self.harness_id is None: + raise ValueError("Successful result requires a non-null harness_id") + if self.recipe_id is None: + raise ValueError("Successful result requires a non-null recipe_id") + if self.recipe_version is None: + raise ValueError("Successful result requires a non-null recipe_version") + if self.install_method is None: + raise ValueError("Successful result requires a non-null install_method") + if self.installer_exit_code != 0: + raise ValueError(f"Successful result requires installer_exit_code == 0, got {self.installer_exit_code}") + if self.verification is None or not self.verification.verified: + raise ValueError("Successful result requires verification.verified == True") + if self.verification.executable_path is None: + raise ValueError("Successful result requires non-null verification.executable_path") + + # CLEANUP REPORTING (must stay honest: unverified is not "nothing survived") + if self.cleanup_complete is not None and not _is_exact_bool(self.cleanup_complete): + raise ValueError( + f"cleanup_complete must be a boolean or None, got {type(self.cleanup_complete).__name__}" + ) + if not _is_exact_bool(self.cleanup_unverified): + raise ValueError( + f"cleanup_unverified must be a boolean, got {type(self.cleanup_unverified).__name__}" + ) + if self.cleanup_complete is True and self.cleanup_unverified: + raise ValueError( + "cleanup_complete=True with cleanup_unverified=True is contradictory: an " + "unverified teardown is never reported as complete" + ) + val_cleanup_notes = _validate_str_sequence( + self.cleanup_notes, + "cleanup_notes", + max_items=128, + max_item_length=1024, + allow_empty_items=False, + ) + object.__setattr__(self, "cleanup_notes", val_cleanup_notes) + + # CANCELLATION INVARIANTS (Contract §C.2) + if self.status == "cancelled": + if self.verification is not None and self.verification.verified: + raise ValueError("Cancelled result cannot report verified == True") + + def to_dict(self) -> dict[str, Any]: + """Convert to JSON-serializable dictionary.""" + d: dict[str, Any] = { + "schema_version": self.schema_version, + "helper_version": self.helper_version, + "request_id": self.request_id, + "timestamp": self.timestamp, + "status": self.status, + "harness_id": self.harness_id, + "recipe_id": self.recipe_id, + "recipe_version": self.recipe_version, + "install_method": self.install_method, + "installer_exit_code": self.installer_exit_code, + "verification": self.verification.to_dict() if self.verification is not None else None, + "path_updates": list(self.path_updates), + "auth_status": self.auth_status, + "manual_steps_remaining": list(self.manual_steps_remaining), + "partial_changes": self.partial_changes.to_dict(), + "error_diagnostics": self.error_diagnostics, + "path_advisory": self.path_advisory.to_dict() if self.path_advisory is not None else None, + "plan_digest": self.plan_digest, + "approval_source": self.approval_source, + "cleanup_complete": self.cleanup_complete, + "cleanup_unverified": self.cleanup_unverified, + "cleanup_notes": list(self.cleanup_notes), + } + return d + + def to_json(self, *, indent: int | None = None) -> str: + """Serialize to JSON string with strict 64 KiB UTF-8 byte enforcement.""" + serialized = json.dumps(self.to_dict(), indent=indent, ensure_ascii=False) + utf8_bytes = serialized.encode("utf-8") + if len(utf8_bytes) > MAX_RESULT_SERIALIZED_BYTES: + raise ValueError( + f"Serialized result size ({len(utf8_bytes)} bytes) exceeds " + f"maximum allowed limit of {MAX_RESULT_SERIALIZED_BYTES} bytes (64 KiB)" + ) + return serialized + + @classmethod + def from_dict(cls, d: Mapping[str, Any]) -> CompletionResult: + """Deserialize and validate from dictionary.""" + if not isinstance(d, Mapping): + raise ValueError(f"CompletionResult payload must be a mapping, got {type(d).__name__}") + + unknown = set(d.keys()) - KNOWN_COMPLETION_RESULT_FIELDS + if unknown: + raise ValueError(f"Unknown field(s) in CompletionResult: {sorted(unknown)}") + + missing = REQUIRED_COMPLETION_RESULT_FIELDS - set(d.keys()) + if missing: + raise ValueError(f"Missing required field(s) in CompletionResult: {sorted(missing)}") + + # Schema version check + schema_version = d["schema_version"] + if not _is_exact_int(schema_version): + raise ValueError(f"schema_version must be exact integer 1, got {type(schema_version).__name__}") + if schema_version != 1: + raise ValueError(f"Unsupported schema_version: {schema_version}") + + # Verification deserialization + ver_raw = d.get("verification") + verification: VerificationResult | None = None + if ver_raw is not None: + if not isinstance(ver_raw, Mapping): + raise ValueError(f"verification must be an object or None, got {type(ver_raw).__name__}") + verification = VerificationResult.from_dict(ver_raw) + + # Partial changes deserialization + pc_raw = d.get("partial_changes") + if pc_raw is None: + raise ValueError("partial_changes field is required") + if not isinstance(pc_raw, Mapping): + raise ValueError(f"partial_changes must be an object, got {type(pc_raw).__name__}") + partial_changes = PartialChanges.from_dict(pc_raw) + + # Path advisory deserialization + path_adv_raw = d.get("path_advisory") + path_advisory: PathAdvisory | None = None + if path_adv_raw is not None: + if isinstance(path_adv_raw, PathAdvisory): + path_advisory = path_adv_raw + elif isinstance(path_adv_raw, Mapping): + path_advisory = PathAdvisory.from_dict(path_adv_raw) + else: + raise ValueError(f"path_advisory must be an object or None, got {type(path_adv_raw).__name__}") + + return cls( + schema_version=schema_version, + helper_version=d["helper_version"], + request_id=d["request_id"], + timestamp=d["timestamp"], + status=d["status"], + harness_id=d.get("harness_id"), + recipe_id=d.get("recipe_id"), + recipe_version=d.get("recipe_version"), + install_method=d.get("install_method"), + installer_exit_code=d.get("installer_exit_code"), + verification=verification, + path_updates=d["path_updates"], # Strictly validated in __post_init__ + auth_status=d.get("auth_status", "unknown"), + manual_steps_remaining=d["manual_steps_remaining"], # Strictly validated in __post_init__ + partial_changes=partial_changes, + error_diagnostics=d.get("error_diagnostics"), + path_advisory=path_advisory, + plan_digest=d.get("plan_digest"), + approval_source=d.get("approval_source"), + cleanup_complete=d.get("cleanup_complete"), + cleanup_unverified=d.get("cleanup_unverified", False), + cleanup_notes=d.get("cleanup_notes", ()), + ) + + @classmethod + def from_json(cls, s: str) -> CompletionResult: + """Parse from JSON string with size validation.""" + utf8_bytes = s.encode("utf-8") + if len(utf8_bytes) > MAX_RESULT_SERIALIZED_BYTES: + raise ValueError( + f"JSON payload size ({len(utf8_bytes)} bytes) exceeds " + f"maximum allowed limit of {MAX_RESULT_SERIALIZED_BYTES} bytes (64 KiB)" + ) + try: + data = json.loads(s) + except json.JSONDecodeError as exc: + raise ValueError(f"Malformed JSON: {exc}") from exc + return cls.from_dict(data) + + +#: Derived from the CompletionResult dataclass itself, so a newly added field can never be +#: rejected as "unknown" by ``from_dict`` (exactly how plan_digest/approval_source broke first). +KNOWN_COMPLETION_RESULT_FIELDS = frozenset(f.name for f in fields(CompletionResult)) diff --git a/src/cmdforge/cli/provider_commands.py b/src/cmdforge/cli/provider_commands.py index da7d447..d83df5a 100644 --- a/src/cmdforge/cli/provider_commands.py +++ b/src/cmdforge/cli/provider_commands.py @@ -244,34 +244,18 @@ def _cmd_providers_install(args): shutil.rmtree(temp_dir, ignore_errors=True) return 1 - # Read and parse completion result - result_data = None - if result_file.is_file(): - try: - result_data = json.loads(result_file.read_text(encoding="utf-8")) - except Exception: - result_data = None - - # Clean up temporary directory + from cmdforge._setup_contract.contract import read_completion_result + verdict, result_data, problems = read_completion_result( + result_file, expected_request_id=req_id, exit_code=helper_exit_code, + expected_harness_id=harness_arg, + ) shutil.rmtree(temp_dir, ignore_errors=True) - - # Validate correlation request ID - if result_data is not None and result_data.get("request_id") != req_id: - print("\nWarning: Correlation request ID mismatch in setup result. Rejecting result.") - return 1 - - # Incoherence handling (Contract §C.5) - if result_data is not None: - status = result_data.get("status") - if helper_exit_code != 0 and status == "success": - print("\nWarning: Helper exited with non-zero code but result reported success. Treating as failure.") - return helper_exit_code - if helper_exit_code == 0 and status in ("failed", "cancelled"): - print("\nWarning: Helper exited with code 0 but result reported failure/cancellation. Treating as failure.") - return 1 + if verdict == "unknown": + print("\nSetup outcome unknown: " + "; ".join(problems)) + return helper_exit_code if helper_exit_code else 1 # Dry-run completion - if getattr(args, "dry_run", False) or (result_data and result_data.get("status") == "dry_run"): + if verdict == "dry_run": return 0 # Cancellation diff --git a/src/cmdforge/gui/dialogs/provider_install_dialog.py b/src/cmdforge/gui/dialogs/provider_install_dialog.py index 9c86a20..bc65ffd 100644 --- a/src/cmdforge/gui/dialogs/provider_install_dialog.py +++ b/src/cmdforge/gui/dialogs/provider_install_dialog.py @@ -163,27 +163,15 @@ class SetupTerminalWorker(QThread): exit_code = self._process.wait() - result_data = None - if self.result_file.is_file(): - try: - result_data = json.loads(self.result_file.read_text(encoding="utf-8")) - except Exception: - result_data = None - - # Validate request_id correlation - if result_data and result_data.get("request_id") != self.request_id: - self.finished.emit(False, "Security error: Request ID mismatch in setup result.", {}) - return - - status = result_data.get("status") if result_data else None - - # Coherence validation (Contract §C.5) - if exit_code != 0 and status == "success": - self.finished.emit(False, f"Helper exited with error code {exit_code} despite reporting success.", result_data or {}) - return - if exit_code == 0 and status in ("failed", "cancelled"): - self.finished.emit(False, "Setup was cancelled or reported failure.", result_data or {}) + from cmdforge._setup_contract.contract import read_completion_result + verdict, result_data, problems = read_completion_result( + self.result_file, expected_request_id=self.request_id, + exit_code=exit_code, expected_harness_id=self.harness_id, + ) + if verdict == "unknown": + self.finished.emit(False, "Setup outcome unknown: " + "; ".join(problems), {}) return + status = result_data["status"] if exit_code == 0 and status == "success": self.finished.emit(True, f"Setup for '{self.harness_id}' verified successfully!", result_data or {}) diff --git a/tests/test_gui_provider_install.py b/tests/test_gui_provider_install.py index 568c757..233f94a 100644 --- a/tests/test_gui_provider_install.py +++ b/tests/test_gui_provider_install.py @@ -29,6 +29,20 @@ from cmdforge.gui.dialogs.provider_install_dialog import ( from cmdforge.providers import Provider, get_provider, load_providers, save_providers +def _write_completion(path, data): + payload = { + "schema_version": 1, "helper_version": "0.1.0", + "timestamp": "2026-09-14T00:00:00Z", "harness_id": "claude-code", + "auth_status": "unknown", "path_updates": [], "manual_steps_remaining": [], + "partial_changes": {"observed_created": [], "observed_modified": [], "unknown_changes": True}, + "installer_exit_code": 0, + "recipe_id": "test-recipe", "recipe_version": "2026-09-10", "install_method": "argv", + } + payload.update(data) + path.write_text(json.dumps(payload), encoding="utf-8") + path.chmod(0o600) + + @pytest.fixture def temp_providers_file(tmp_path): providers_file = tmp_path / ".cmdforge" / "providers.yaml" @@ -141,6 +155,13 @@ def test_setup_terminal_worker_success(tmp_path): result_file = tmp_path / "result.json" req_id = "test-uuid-1234" data = { + "schema_version": 1, + "helper_version": "0.1.0", + "timestamp": "2026-09-14T00:00:00Z", + "auth_status": "unknown", + "manual_steps_remaining": [], + "partial_changes": {"observed_created": [], "observed_modified": [], "unknown_changes": True}, + "verification": {"verified": True, "executable_path": "/custom/bin/claude", "version_output": "test"}, "status": "success", "request_id": req_id, "harness_id": "claude-code", @@ -150,7 +171,7 @@ def test_setup_terminal_worker_success(tmp_path): class MockProcess: def __init__(self, *args, **kwargs): self.stdout = None - result_file.write_text(json.dumps(data), encoding="utf-8") + _write_completion(result_file, data) def wait(self): return 0 @@ -179,6 +200,26 @@ def test_setup_terminal_worker_success(tmp_path): assert res_dict["path_updates"] == ["/custom/bin"] +@pytest.mark.parametrize("overrides", [ + {"schema_version": True}, {"unexpected": 1}, {"status": []}, + {"partial_changes": None}, {"cleanup_complete": "true"}, +]) +def test_worker_rejects_malformed_contract(tmp_path, overrides): + path = tmp_path / "result.json" + data = {"status": "failed", "request_id": "req-1"} + data.update(overrides) + _write_completion(path, data) + process = MagicMock(stdout=None) + process.wait.return_value = 1 + worker = SetupTerminalWorker(["fake-helper"], "claude-code", path, "req-1") + outcomes = [] + worker.finished.connect(lambda *args: outcomes.append(args)) + with patch("subprocess.Popen", return_value=process): + worker.run() + assert len(outcomes) == 1 and outcomes[0][0] is False + assert "unknown" in outcomes[0][1] + + def test_setup_terminal_worker_security_mismatch(tmp_path): """Test SetupTerminalWorker rejects result with request_id mismatch.""" result_file = tmp_path / "result.json" @@ -191,7 +232,7 @@ def test_setup_terminal_worker_security_mismatch(tmp_path): class MockProcess: def __init__(self, *args, **kwargs): self.stdout = None - result_file.write_text(json.dumps(data), encoding="utf-8") + _write_completion(result_file, data) def wait(self): return 0 @@ -212,7 +253,7 @@ def test_setup_terminal_worker_security_mismatch(tmp_path): assert len(results) == 1 success, msg, _ = results[0] assert success is False - assert "Security error" in msg + assert "does not match the expected" in msg def test_setup_terminal_worker_incoherent_exit(tmp_path): @@ -227,7 +268,7 @@ def test_setup_terminal_worker_incoherent_exit(tmp_path): class MockProcess: def __init__(self, *args, **kwargs): self.stdout = None - result_file.write_text(json.dumps(data), encoding="utf-8") + _write_completion(result_file, data) def wait(self): return 1 @@ -248,7 +289,7 @@ def test_setup_terminal_worker_incoherent_exit(tmp_path): assert len(results) == 1 success, msg, _ = results[0] assert success is False - assert "error code 1" in msg + assert "exit code 1" in msg def test_setup_terminal_worker_cancelled(tmp_path): @@ -256,6 +297,14 @@ def test_setup_terminal_worker_cancelled(tmp_path): result_file = tmp_path / "result.json" req_id = "cancel-uuid" data = { + "schema_version": 1, + "helper_version": "0.1.0", + "timestamp": "2026-09-14T00:00:00Z", + "harness_id": "claude-code", + "auth_status": "unknown", + "path_updates": [], + "manual_steps_remaining": [], + "partial_changes": {"observed_created": [], "observed_modified": [], "unknown_changes": True}, "status": "cancelled", "request_id": req_id, } @@ -263,7 +312,7 @@ def test_setup_terminal_worker_cancelled(tmp_path): class MockProcess: def __init__(self, *args, **kwargs): self.stdout = None - result_file.write_text(json.dumps(data), encoding="utf-8") + _write_completion(result_file, data) def wait(self): return 10 @@ -300,7 +349,7 @@ def test_setup_terminal_worker_failure_with_diagnostics(tmp_path): class MockProcess: def __init__(self, *args, **kwargs): self.stdout = None - result_file.write_text(json.dumps(data), encoding="utf-8") + _write_completion(result_file, data) def wait(self): return 1 diff --git a/tests/test_provider_install.py b/tests/test_provider_install.py index d8a93a9..495d1bd 100644 --- a/tests/test_provider_install.py +++ b/tests/test_provider_install.py @@ -56,11 +56,28 @@ def _mock_helper_run(status: str, harness_id: str = "claude-code", exit_code: in "error_diagnostics": diagnostics, } result_file_path.write_text(json.dumps(data), encoding="utf-8") + result_file_path.chmod(0o600) return SimpleNamespace(returncode=exit_code) return _run +@pytest.mark.parametrize("dry_run", [False, True]) +def test_sparse_success_cannot_register_provider(temp_providers_file, capsys, dry_run): + def fake(cmd, *args, **kwargs): + path = Path(cmd[cmd.index("--result-file") + 1]) + request_id = cmd[cmd.index("--request-id") + 1] + path.write_text(json.dumps({"schema_version": 1, "status": "success", "request_id": request_id})) + path.chmod(0o600) + return SimpleNamespace(returncode=0) + argv = ["cmdforge", "providers", "install"] + (["--dry-run"] if dry_run else []) + with patch("cmdforge.cli.provider_commands._find_setup_helper", return_value=["fake"]), patch("subprocess.run", side_effect=fake), patch("sys.argv", argv), patch("builtins.input") as prompt: + assert main() == 1 + prompt.assert_not_called() + assert get_provider("claude") is None + assert "unknown" in capsys.readouterr().out + + def test_helper_discovery_finds_executable(): """Verify _find_setup_helper locates agent-harness-setup.""" with patch("shutil.which", return_value="/usr/local/bin/agent-harness-setup"): @@ -182,7 +199,7 @@ def test_providers_install_incoherent_result_rejected(temp_providers_file, capsy assert exit_code == 1 captured = capsys.readouterr() - assert "Treating as failure" in captured.out + assert "not coherent with exit code 1" in captured.out assert get_provider("claude") is None