Enforce canonical setup result contract in CLI and GUI

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.

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.

src/cmdforge/_setup_contract/__init__.py

@@ -0,0 +1 @@
+"""Vendored setup-result contract; see README.md."""

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}"]

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))

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

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 {})

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

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